[Fix] Fix leapd bugs and unit tests on windows

src/leapflow/cli/commands/slash_handlers.py

@@ -180,7 +180,9 @@ def build_config_payload(ctx: "Context", args: str = "") -> dict[str, Any]:
 
     service = ConfigService(ctx.settings)
     try:
-        tokens = shlex.split(args)
+        from leapflow.utils.shell_lex import split_args
+
+        tokens = split_args(args)
     except ValueError as exc:
         return {"ok": False, "message": f"Invalid /config syntax: {exc}"}
     if not tokens:
@@ -702,7 +704,9 @@ def _parse_app_options(tokens: list[str]) -> tuple[dict[str, str], str]:
 
 def _parse_app_params(args: str) -> dict[str, Any]:
     try:
-        tokens = shlex.split(args)
+        from leapflow.utils.shell_lex import split_args
+
+        tokens = split_args(args)
     except ValueError as exc:
         return {"ok": False, "error": f"Invalid /app arguments: {exc}"}
 
@@ -1218,7 +1222,9 @@ async def _execute_dashboard(
     sub = name.split(" ", 1)[1].strip() if " " in name else ""
     rest = (sub + ((" " + args) if args else "")).strip() if sub else args.strip()
     try:
-        tokens = shlex.split(rest) if rest else []
+        from leapflow.utils.shell_lex import split_args
+
+        tokens = split_args(rest) if rest else []
     except ValueError:
         tokens = rest.split()
     verb = tokens[0].lower() if tokens else ""

src/leapflow/daemon/_transport.py

@@ -33,6 +33,10 @@ async def connect(self, runtime_dir: Path) -> Tuple[asyncio.StreamReader, asynci
     def probe_healthy(self, runtime_dir: Path) -> bool:
         """Synchronous quick health check — returns True if the daemon is reachable."""
 
+    @abc.abstractmethod
+    def readiness_path(self, runtime_dir: Path) -> Path:
+        """Artifact file the daemon publishes when it is ready to serve."""
+
     @abc.abstractmethod
     def cleanup(self, runtime_dir: Path) -> None:
         """Remove transport artifacts (socket file, port file, etc.)."""
@@ -70,6 +74,9 @@ def probe_healthy(self, runtime_dir: Path) -> bool:
         except (OSError, socket.timeout):
             return False
 
+    def readiness_path(self, runtime_dir: Path) -> Path:
+        return self._sock_path(runtime_dir)
+
     def cleanup(self, runtime_dir: Path) -> None:
         self._sock_path(runtime_dir).unlink(missing_ok=True)
 
@@ -119,6 +126,9 @@ def probe_healthy(self, runtime_dir: Path) -> bool:
         except (OSError, socket.timeout):
             return False
 
+    def readiness_path(self, runtime_dir: Path) -> Path:
+        return self._port_path(runtime_dir)
+
     def cleanup(self, runtime_dir: Path) -> None:
         self._port_path(runtime_dir).unlink(missing_ok=True)
 

src/leapflow/daemon/client.py

@@ -288,9 +288,12 @@ async def shutdown(self) -> None:
 
     async def _open(self) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
         try:
+            # _timeout_s is the per-request read budget (heartbeat tests set it
+            # to 0.1s); connection establishment needs its own, larger floor or
+            # every RPC flakes under load when the handshake exceeds it.
             return await asyncio.wait_for(
                 get_transport().connect(self._sock_path.parent),
-                timeout=self._timeout_s,
+                timeout=max(self._timeout_s, 5.0),
             )
         except (TimeoutError, OSError) as exc:
             raise DaemonUnavailableError(
@@ -321,7 +324,7 @@ async def ensure_daemon_client(
 ) -> DaemonClient:
     """Return a client connected to a healthy daemon, starting one if needed."""
     runtime_dir = settings.runtime_dir
-    sock_path = runtime_dir / "leapd.sock"
+    sock_path = get_transport().readiness_path(runtime_dir)
     info = DaemonInfo.discover(runtime_dir)
     if info.is_healthy:
         _emit(status_callback, f"Connected to leapd (pid={info.pid}).")

src/leapflow/daemon/lifecycle.py

@@ -48,12 +48,14 @@ class DaemonInfo:
     def discover(cls, runtime_dir: Path) -> DaemonInfo:
         """Probe the runtime directory to determine daemon state."""
         pid = _read_pid(runtime_dir / "leapd.pid")
-        sock_path = runtime_dir / "leapd.sock"
+        from leapflow.daemon._transport import get_transport
+
+        sock_path = get_transport().readiness_path(runtime_dir)
         meta = _read_meta(runtime_dir / "leapd.json")
         start_time = meta.get("start_time") if meta else None
 
         is_running = pid is not None and _process_alive(pid)
-        is_healthy = is_running and sock_path.exists() and _sock_healthy(sock_path)
+        is_healthy = is_running and _sock_healthy(runtime_dir)
 
         return cls(
             pid=pid,
@@ -439,12 +441,11 @@ def _process_alive(pid: int) -> bool:
             kernel32.CloseHandle(handle)
 
 
-def _sock_healthy(sock_path: Path) -> bool:
+def _sock_healthy(runtime_dir: Path) -> bool:
     """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)
+    return get_transport().probe_healthy(runtime_dir)
 
 
 def _format_duration(seconds: Optional[float]) -> str:

src/leapflow/daemon/server.py

@@ -289,7 +289,7 @@ async def serve_daemon(settings: Any, *, mock_host: bool = False) -> int:
     from leapflow.daemon.service import RuntimeLeapService
 
     runtime_dir = settings.runtime_dir
-    sock_path = runtime_dir / "leapd.sock"
+    sock_path = get_transport().readiness_path(runtime_dir)
     service = RuntimeLeapService(settings, mock_host=mock_host)
     await service.start()
     loop = asyncio.get_running_loop()

src/leapflow/daemon/session_registry.py

@@ -49,15 +49,25 @@ def __init__(self, session_id: str, expected: Path, requested: Path) -> None:
 class SessionExecutionContext:
     """Per-session execution state: an engine + workspace + turn lock."""
 
+    # Coarse clocks quantize: on Windows, time.monotonic resolves to ~15.6ms,
+    # so two activities can share a timestamp. This counter stamps every
+    # activity strictly after the previous one, keeping "most recent" and
+    # "oldest" orderings deterministic.
+    _activity_counter = 0
+
     def __init__(self, session_id: str, engine: Any, workspace_root: Path) -> None:
         self.session_id = session_id
         self.engine = engine
         self.workspace_root = workspace_root
         self.lock = asyncio.Lock()  # serialize this session's own turns
         self.last_active = time.monotonic()
+        SessionExecutionContext._activity_counter += 1
+        self.activity_seq = SessionExecutionContext._activity_counter
 
     def touch(self) -> None:
         self.last_active = time.monotonic()
+        SessionExecutionContext._activity_counter += 1
+        self.activity_seq = SessionExecutionContext._activity_counter
 
 
 class SessionRegistry:
@@ -150,7 +160,7 @@ def _evict_oldest(self) -> None:
         ]
         if not candidates:
             return
-        oldest = min(candidates, key=lambda c: c.last_active)
+        oldest = min(candidates, key=lambda c: (c.last_active, c.activity_seq))
         del self._contexts[oldest.session_id]
 
     def active_count(self) -> int:
@@ -178,4 +188,7 @@ def most_recent_any_client(self) -> Optional[SessionExecutionContext]:
         """
         if not self._contexts:
             return None
-        return max(self._contexts.values(), key=lambda ctx: ctx.last_active)
+        return max(
+            self._contexts.values(),
+            key=lambda ctx: (ctx.last_active, ctx.activity_seq),
+        )

src/leapflow/dashboard/intent.py

@@ -9,10 +9,11 @@
 
 from __future__ import annotations
 
-import shlex
 from dataclasses import dataclass
 from typing import Any, Mapping
 
+from leapflow.utils.shell_lex import split_args
+
 
 @dataclass(frozen=True)
 class DashboardIntent:
@@ -33,7 +34,7 @@ def from_params(cls, data: Mapping[str, Any]) -> "DashboardIntent":
     def from_args(cls, args: str) -> "DashboardIntent":
         """Parse a slash argument string; the first token is the template name."""
         try:
-            tokens = shlex.split(args or "")
+            tokens = split_args(args or "")
         except ValueError:
             tokens = (args or "").split()
         return cls(template=tokens[0].strip() if tokens else "")

src/leapflow/storage/duckdb_connect.py

@@ -24,7 +24,18 @@
 
 logger = logging.getLogger(__name__)
 
-LOCK_KEYWORDS = frozenset({"lock", "locked", "exclusive", "cannot set lock"})
+LOCK_KEYWORDS = frozenset({
+    "lock",
+    "locked",
+    "exclusive",
+    "cannot set lock",
+    # Windows "file held by another process" forms — DuckDB raises these
+    # instead of a lock error when the DB file is open elsewhere (English
+    # and Chinese system locales). Must classify as locked, not corrupted.
+    "cannot open file",
+    "being used by another process",
+    "另一个程序正在使用",
+})
 
 _CONNECT_RETRIES = int(os.getenv("LEAPFLOW_DB_CONNECT_RETRIES", "5" if sys.platform == "win32" else "3"))
 _CONNECT_BACKOFF_S = float(os.getenv("LEAPFLOW_DB_CONNECT_BACKOFF_S", "1.0" if sys.platform == "win32" else "0.5"))

src/leapflow/tools/shell_tools.py

@@ -13,7 +13,6 @@
 import logging
 import os
 import re
-import shlex
 import subprocess
 import sys
 from pathlib import Path
@@ -27,6 +26,7 @@
     workspace_scope_error,
 )
 from leapflow.utils.process_group import ProcessGroup
+from leapflow.utils.shell_lex import split_args
 
 logger = logging.getLogger(__name__)
 
@@ -159,13 +159,16 @@ def _expand_operand(token: str) -> str:
     Expansions that are not a single filesystem operand are discarded: a variable
     holding a search list (``$PATH``) expands to ``os.pathsep``-joined entries
     that begin with ``/`` but name no file, so treating it as a path would block
-    ordinary commands like ``echo $PATH`` or ``PATH=$PATH:./bin npm test``.
+    ordinary commands like ``echo $PATH`` or ``PATH=$PATH:./bin npm test``. On
+    Windows, mixed-shell environments (MSYS/git-bash) join the same list with
+    ``:``, so both separators disqualify — except a drive-letter colon.
     """
     candidate = token.split("=", 1)[-1]
     if "$" not in candidate:
         return candidate
     expanded = os.path.expandvars(candidate)
-    if os.pathsep in expanded or any(char.isspace() for char in expanded):
+    body = expanded[2:] if len(expanded) >= 2 and expanded[1] == ":" else expanded
+    if os.pathsep in body or ":" in body or any(char.isspace() for char in expanded):
         return ""
     return expanded
 
@@ -175,6 +178,11 @@ def _has_parent_traversal(operand: str) -> bool:
     return ".." in Path(operand).parts
 
 
+# Drive-letter operands (``C:\...`` / ``C:/...``) address files exactly like
+# POSIX ``/...`` ones; without this the gate never inspects them on Windows.
+_WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
+
+
 def _command_workspace_escape(command: str, cwd: Path | None = None) -> dict[str, Any] | None:
     """Reject path operands that resolve outside the active workspace.
 
@@ -195,7 +203,7 @@ def _command_workspace_escape(command: str, cwd: Path | None = None) -> dict[str
         return None
     base = cwd if cwd is not None else ctx.workspace_root
     try:
-        tokens = shlex.split(command)
+        tokens = split_args(command)
     except ValueError:
         tokens = command.split()
     for token in tokens:
@@ -204,7 +212,7 @@ def _command_workspace_escape(command: str, cwd: Path | None = None) -> dict[str
         operand = _expand_operand(token)
         if not operand:
             continue
-        if operand.startswith(("/", "~")):
+        if operand.startswith(("/", "~")) or _WINDOWS_ABSOLUTE.match(operand):
             path = Path(operand)
         elif _has_parent_traversal(operand):
             path = base / operand

src/leapflow/utils/shell_lex.py

@@ -0,0 +1,18 @@
+"""Shell-style argument tokenization that survives Windows paths."""
+
+from __future__ import annotations
+
+import os
+import shlex
+
+
+def split_args(text: str) -> list[str]:
+    """Tokenize a command/argument string like ``shlex.split``, keeping Windows paths.
+
+    POSIX shlex treats ``\\`` as an escape and silently strips the separators
+    out of ``C:\\a\\b.yaml``; on Windows ``\\`` is a path separator, so double
+    it before lexing. Quote semantics are otherwise unchanged. Raises
+    ``ValueError`` on unbalanced quoting, exactly like ``shlex.split``.
+    """
+    lexable = text.replace("\\", "\\\\") if os.name == "nt" else text
+    return shlex.split(lexable)

tests/_harness/cassette.py

@@ -46,6 +46,9 @@
     (re.compile(r"\b(?:sess|ws|req|traj|ep|skill|watch|call)-[0-9a-zA-Z]{6,}\b"), "<ID>"),
     (re.compile(r"\bcall_[0-9a-zA-Z]{6,}\b"), "<ID>"),
     (re.compile(r"/(?:private/)?(?:var|tmp)/[^\s\"',)\]]*"), "<TMP>"),
+    # Windows drive-letter paths: journeys scratch under %TEMP% and prompts
+    # plus tool results embed those paths, so normalize like POSIX /tmp.
+    (re.compile(r"[A-Za-z]:\\[^\s\"',)\]]*"), "<TMP>"),
     (re.compile(r"127\.0\.0\.1:\d+"), "127.0.0.1:<PORT>"),
     (re.compile(r"\b1[0-9]{9}(?:\.[0-9]+)?\b"), "<TS>"),
 )

tests/_harness/leapd.py

@@ -22,6 +22,7 @@
 
 from leapflow.daemon.client import DaemonClient
 from leapflow.daemon.lifecycle import DaemonInfo, cleanup_stale, wait_ready
+from leapflow.daemon._transport import get_transport
 from leapflow.layout import build_layout
 
 READY_TIMEOUT_S = 60.0
@@ -152,7 +153,7 @@ class Leapd:
     @property
     def sock_path(self) -> Path:
         """Unix socket the daemon listens on."""
-        return self.runtime_dir / "leapd.sock"
+        return get_transport().readiness_path(self.runtime_dir)
 
     @property
     def log_path(self) -> Path:
@@ -248,7 +249,7 @@ def start_leapd(
     runtime_dir.mkdir(parents=True, exist_ok=True)
     cleanup_stale(runtime_dir)
 
-    sock_path = runtime_dir / "leapd.sock"
+    sock_path = get_transport().readiness_path(runtime_dir)
     if len(str(sock_path)) > MAX_SOCKET_PATH_LEN:
         raise LeapdStartupError(
             f"daemon socket path is {len(str(sock_path))} bytes, over the "

tests/conftest.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import sys
 import time
 from pathlib import Path
 from typing import Any, AsyncIterator, List, Optional
@@ -23,6 +24,30 @@
 from leapflow.storage.trajectory_store import TrajectoryStore
 
 
+# ════════════════════════════════════════════════════════════════
+# Headless prompt_toolkit output
+# ════════════════════════════════════════════════════════════════
+
+
+@pytest.fixture(autouse=(sys.platform == "win32"))
+def _headless_prompt_toolkit_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    """On Windows, tests never attach to a real console; render through a DummyOutput.
+
+    prompt_toolkit's win32 default output probes the console screen buffer at
+    construction and raises NoConsoleScreenBufferError anywhere a real conhost
+    is absent (git-bash, CI, pytest capture). POSIX Vt100 output has no such
+    probe, so the fixture only arms on Windows. The import below is the exact
+    symbol Application resolves lazily, so replacing it covers every app the
+    tests build.
+    """
+    from prompt_toolkit.output import DummyOutput
+    from prompt_toolkit.output import defaults as _output_defaults
+
+    monkeypatch.setattr(
+        _output_defaults, "create_output", lambda *args, **kwargs: DummyOutput()
+    )
+
+
 # ════════════════════════════════════════════════════════════════
 # Layer markers — applied by path so existing files need no edit
 # ════════════════════════════════════════════════════════════════

tests/journeys/test_r6_lifecycle.py

@@ -22,6 +22,7 @@
 
 from leapflow.daemon.client import DaemonUnavailableError
 from leapflow.daemon.lifecycle import DaemonInfo, cleanup_stale
+from leapflow.daemon._transport import get_transport
 from tests._harness.cassette_proxy import answer, scripted
 from tests._harness.journey import Journey, JourneyFactory
 from tests._harness.leapd import await_for, start_leapd
@@ -107,7 +108,8 @@ async def test_r6_daemon_lifecycle(journeys: JourneyFactory) -> None:
         # Simulate the crash case: runtime files present, no process behind them.
         journey.daemon.runtime_dir.mkdir(parents=True, exist_ok=True)
         (journey.daemon.runtime_dir / "leapd.pid").write_text("999999", encoding="utf-8")
-        (journey.daemon.runtime_dir / "leapd.sock").touch(exist_ok=True)
+        sock_path = get_transport().readiness_path(journey.daemon.runtime_dir)
+        sock_path.touch(exist_ok=True)
 
         stale = DaemonInfo.discover(journey.daemon.runtime_dir)
         assert not stale.is_healthy, "a stale socket was reported as healthy"

tests/regression/test_test_layer_contracts.py

@@ -93,7 +93,7 @@ def _test_modules() -> list[pathlib.Path]:
 
 
 def _relative(path: pathlib.Path) -> str:
-    return str(path.relative_to(TESTS_ROOT))
+    return path.relative_to(TESTS_ROOT).as_posix()
 
 
 def _harness_modules() -> list[pathlib.Path]: