diff --git a/src/leapflow/cli/commands/host.py b/src/leapflow/cli/commands/host.py index 5e20964..f1382b3 100644 --- a/src/leapflow/cli/commands/host.py +++ b/src/leapflow/cli/commands/host.py @@ -266,18 +266,27 @@ async def _cmd_start() -> int: "def _signal_handler(*a): stop_event.set(); " "signal.signal(signal.SIGTERM, _signal_handler); " "signal.signal(signal.SIGINT, _signal_handler); " + "(signal.signal(signal.SIGBREAK, _signal_handler) " + "if hasattr(signal, 'SIGBREAK') else None); " "loop.run_until_complete(stop_event.wait()); " "loop.run_until_complete(daemon.stop()); " "print('ObservationDaemon stopped', flush=True)" ) + creationflags = 0 + start_new_session = True + if os.name == "nt": # pragma: no cover - platform specific + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + start_new_session = False + with open(log_file, "a") as lf: proc = subprocess.Popen( [sys.executable, "-c", daemon_script], stdout=lf, stderr=lf, stdin=subprocess.DEVNULL, - start_new_session=True, + start_new_session=start_new_session, + creationflags=creationflags, ) # Wait briefly to confirm startup diff --git a/src/leapflow/daemon/_transport.py b/src/leapflow/daemon/_transport.py new file mode 100644 index 0000000..30510f0 --- /dev/null +++ b/src/leapflow/daemon/_transport.py @@ -0,0 +1,130 @@ +"""Cross-platform daemon IPC transport. + +Unix (macOS/Linux): Unix Domain Socket via asyncio.start_unix_server / open_unix_connection +Windows: TCP loopback (127.0.0.1:dynamic-port) via asyncio.start_server / open_connection +""" +from __future__ import annotations + +import abc +import asyncio +import socket +import sys +from collections.abc import Callable +from pathlib import Path +from typing import Any, Tuple + + +class DaemonTransport(abc.ABC): + """Abstract base for daemon IPC transport.""" + + @abc.abstractmethod + async def start_server( + self, + client_connected_cb: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Any], + runtime_dir: Path, + ) -> asyncio.AbstractServer: + """Start listening for client connections. Returns the server instance.""" + + @abc.abstractmethod + async def connect(self, runtime_dir: Path) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]: + """Open a client connection to the daemon.""" + + @abc.abstractmethod + def probe_healthy(self, runtime_dir: Path) -> bool: + """Synchronous quick health check — returns True if the daemon is reachable.""" + + @abc.abstractmethod + def cleanup(self, runtime_dir: Path) -> None: + """Remove transport artifacts (socket file, port file, etc.).""" + + +class UnixSocketTransport(DaemonTransport): + """Unix Domain Socket transport (macOS/Linux).""" + + def _sock_path(self, runtime_dir: Path) -> Path: + return runtime_dir / "leapd.sock" + + async def start_server( + self, + client_connected_cb: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Any], + runtime_dir: Path, + ) -> asyncio.AbstractServer: + sock_path = self._sock_path(runtime_dir) + return await asyncio.start_unix_server( + client_connected_cb, + path=str(sock_path), + ) + + async def connect(self, runtime_dir: Path) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]: + sock_path = self._sock_path(runtime_dir) + return await asyncio.open_unix_connection(str(sock_path)) + + def probe_healthy(self, runtime_dir: Path) -> bool: + sock_path = self._sock_path(runtime_dir) + try: + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(1.0) + s.connect(str(sock_path)) + s.close() + return True + except (OSError, socket.timeout): + return False + + def cleanup(self, runtime_dir: Path) -> None: + self._sock_path(runtime_dir).unlink(missing_ok=True) + + +class TcpLoopbackTransport(DaemonTransport): + """TCP loopback transport for Windows (127.0.0.1:dynamic-port).""" + + def _port_path(self, runtime_dir: Path) -> Path: + return runtime_dir / "leapd.port" + + def _read_port(self, runtime_dir: Path) -> int: + port_path = self._port_path(runtime_dir) + return int(port_path.read_text().strip()) + + async def start_server( + self, + client_connected_cb: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Any], + runtime_dir: Path, + ) -> asyncio.AbstractServer: + server = await asyncio.start_server( + client_connected_cb, + host="127.0.0.1", + port=0, + ) + # Extract the assigned port from the server socket and persist it. + addr = server.sockets[0].getsockname() + port = addr[1] + port_path = self._port_path(runtime_dir) + port_path.write_text(str(port)) + return server + + async def connect(self, runtime_dir: Path) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]: + port = self._read_port(runtime_dir) + return await asyncio.open_connection("127.0.0.1", port) + + def probe_healthy(self, runtime_dir: Path) -> bool: + try: + port = self._read_port(runtime_dir) + except (OSError, ValueError): + return False + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(1.0) + s.connect(("127.0.0.1", port)) + s.close() + return True + except (OSError, socket.timeout): + return False + + def cleanup(self, runtime_dir: Path) -> None: + self._port_path(runtime_dir).unlink(missing_ok=True) + + +def get_transport() -> DaemonTransport: + """Return the platform-appropriate daemon transport.""" + if sys.platform == "win32": + return TcpLoopbackTransport() + return UnixSocketTransport() diff --git a/src/leapflow/daemon/client.py b/src/leapflow/daemon/client.py index 4142a00..40cb9be 100644 --- a/src/leapflow/daemon/client.py +++ b/src/leapflow/daemon/client.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import Any +from leapflow.daemon._transport import get_transport from leapflow.daemon.lifecycle import ( DaemonInfo, DaemonLock, @@ -288,7 +289,7 @@ async def shutdown(self) -> None: async def _open(self) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: try: return await asyncio.wait_for( - asyncio.open_unix_connection(str(self._sock_path)), + get_transport().connect(self._sock_path.parent), timeout=self._timeout_s, ) except (TimeoutError, OSError) as exc: diff --git a/src/leapflow/daemon/lease.py b/src/leapflow/daemon/lease.py index 2b393a7..f94b897 100644 --- a/src/leapflow/daemon/lease.py +++ b/src/leapflow/daemon/lease.py @@ -9,6 +9,8 @@ from dataclasses import dataclass from pathlib import Path +from leapflow.daemon.lifecycle import _process_alive + _CLIENTS_DIR = "clients" _DEFAULT_LEASE_TTL_S = 120.0 _DEFAULT_TOUCH_INTERVAL_S = 30.0 @@ -174,9 +176,3 @@ def _read_lease(path: Path) -> ClientLeaseSnapshot | None: return None -def _process_alive(pid: int) -> bool: - try: - os.kill(pid, 0) - return True - except OSError: - return False diff --git a/src/leapflow/daemon/lifecycle.py b/src/leapflow/daemon/lifecycle.py index c4addf4..85ce0eb 100644 --- a/src/leapflow/daemon/lifecycle.py +++ b/src/leapflow/daemon/lifecycle.py @@ -22,7 +22,6 @@ import logging import os import signal -import socket import subprocess import sys import time @@ -172,8 +171,8 @@ def write_pid_file(runtime_dir: Path, pid: Optional[int] = None) -> None: def cleanup_runtime_dir(runtime_dir: Path) -> None: - """Remove daemon runtime files (PID, socket, metadata).""" - for name in ("leapd.pid", "leapd.json", "leapd.sock"): + """Remove daemon runtime files (PID, socket, metadata, port).""" + for name in ("leapd.pid", "leapd.json", "leapd.sock", "leapd.port"): path = runtime_dir / name path.unlink(missing_ok=True) logger.info("daemon: cleaned up %s", runtime_dir) @@ -294,14 +293,27 @@ def spawn_daemon(settings: object, *, mock_host: bool = False) -> subprocess.Pop command.extend(["daemon", "serve", "--internal"]) log_file = open(log_path, "ab") try: - proc = subprocess.Popen( - command, - stdin=subprocess.DEVNULL, - stdout=log_file, - stderr=subprocess.STDOUT, - start_new_session=True, - close_fds=True, - ) + if sys.platform == "win32": # pragma: no cover - platform specific + creationflags = ( + subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS + ) + proc = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=log_file, + stderr=subprocess.STDOUT, + start_new_session=False, + creationflags=creationflags, + ) + else: + proc = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=log_file, + stderr=subprocess.STDOUT, + start_new_session=True, + close_fds=True, + ) finally: log_file.close() logger.info("daemon: spawned pid=%s log=%s", proc.pid, log_path) @@ -402,15 +414,11 @@ def _process_alive(pid: int) -> bool: def _sock_healthy(sock_path: Path) -> bool: - """Quick health check by connecting to the Unix socket.""" - try: - s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - s.settimeout(1.0) - s.connect(str(sock_path)) - s.close() - return True - except (OSError, socket.timeout): - return False + """Quick health check by connecting to the daemon transport.""" + from leapflow.daemon._transport import get_transport + + transport = get_transport() + return transport.probe_healthy(sock_path.parent) def _format_duration(seconds: Optional[float]) -> str: diff --git a/src/leapflow/daemon/server.py b/src/leapflow/daemon/server.py index 52f910a..639cb74 100644 --- a/src/leapflow/daemon/server.py +++ b/src/leapflow/daemon/server.py @@ -11,6 +11,7 @@ from pathlib import Path from typing import Any +from leapflow.daemon._transport import get_transport from leapflow.daemon.lease import default_lease_ttl_s, read_active_client_leases from leapflow.daemon.lifecycle import cleanup_runtime_dir, write_pid_file from leapflow.daemon.protocol import ErrorCode, METHOD_REGISTRY, RpcRequest, RpcResponse, StreamChunk @@ -88,10 +89,11 @@ def has_keepalive_work(self) -> bool: async def serve_forever(self) -> None: """Start listening and serve until cancelled.""" self._runtime_dir.mkdir(parents=True, exist_ok=True) - self._sock_path.unlink(missing_ok=True) - self._server = await asyncio.start_unix_server( + transport = get_transport() + transport.cleanup(self._runtime_dir) + self._server = await transport.start_server( self._handle_client, - path=str(self._sock_path), + self._runtime_dir, ) write_pid_file(self._runtime_dir) try: @@ -100,14 +102,14 @@ async def serve_forever(self) -> None: except asyncio.CancelledError: raise finally: - self._sock_path.unlink(missing_ok=True) + transport.cleanup(self._runtime_dir) async def stop(self) -> None: """Stop accepting clients and close the listening socket.""" if self._server is not None: self._server.close() await self._server.wait_closed() - self._sock_path.unlink(missing_ok=True) + get_transport().cleanup(self._runtime_dir) async def _handle_client( self, diff --git a/src/leapflow/tools/shell_tools.py b/src/leapflow/tools/shell_tools.py index b063e07..6ef64c6 100644 --- a/src/leapflow/tools/shell_tools.py +++ b/src/leapflow/tools/shell_tools.py @@ -14,6 +14,8 @@ import os import re import shlex +import subprocess +import sys from pathlib import Path from typing import Any, Dict, FrozenSet, Optional, Protocol, runtime_checkable @@ -264,12 +266,17 @@ async def shell_run(params: Dict[str, Any]) -> Dict[str, Any]: return {"ok": False, "error": message} try: + _popen_kwargs: Dict[str, Any] = {} + if sys.platform == "win32": # pragma: no cover - platform specific + _popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + _popen_kwargs["start_new_session"] = True proc = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd, - start_new_session=True, + **_popen_kwargs, ) # Attach immediately so the shell's descendants inherit group membership # and a timeout can kill the whole tree, not just the shell itself. diff --git a/src/leapflow/tools/terminal_session.py b/src/leapflow/tools/terminal_session.py index 6d5f087..e946401 100644 --- a/src/leapflow/tools/terminal_session.py +++ b/src/leapflow/tools/terminal_session.py @@ -207,6 +207,11 @@ async def terminal_open(params: Dict[str, Any]) -> Dict[str, Any]: return {"ok": False, "error": "Initial command blocked by safety policy.", "failure_code": "blocked"} try: + _popen_kwargs: Dict[str, Any] = {} + if sys.platform == "win32": # pragma: no cover - platform specific + _popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + _popen_kwargs["start_new_session"] = True proc = subprocess.Popen( [shell], stdin=subprocess.PIPE, @@ -214,7 +219,7 @@ async def terminal_open(params: Dict[str, Any]) -> Dict[str, Any]: stderr=subprocess.STDOUT, cwd=cwd, bufsize=0, - start_new_session=True, + **_popen_kwargs, ) except (OSError, ValueError) as exc: return {"ok": False, "error": f"Failed to start shell: {exc}", "failure_code": "spawn_failed"} diff --git a/src/leapflow/utils/process_group.py b/src/leapflow/utils/process_group.py index d2c5639..69c97cd 100644 --- a/src/leapflow/utils/process_group.py +++ b/src/leapflow/utils/process_group.py @@ -90,6 +90,7 @@ def terminate(self, sig: int = signal.SIGTERM) -> bool: return False try: os.killpg(self._pgid, sig) + self._pgid = None return True except OSError: return False