feat(daemon): cross-platform IPC transport for Windows support

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

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()

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:

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

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:

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,

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.

src/leapflow/tools/terminal_session.py

@@ -207,14 +207,19 @@ 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,
             stdout=subprocess.PIPE,
             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"}

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