diff --git a/src/leapflow/cli/commands/slash_handlers.py b/src/leapflow/cli/commands/slash_handlers.py
index cca129e..d0ce849 100644
--- a/src/leapflow/cli/commands/slash_handlers.py
+++ b/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 ""
diff --git a/src/leapflow/daemon/_transport.py b/src/leapflow/daemon/_transport.py
index c4aa7dd..3dca2bd 100644
--- a/src/leapflow/daemon/_transport.py
+++ b/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)
 
diff --git a/src/leapflow/daemon/client.py b/src/leapflow/daemon/client.py
index 40cb9be..c1d471e 100644
--- a/src/leapflow/daemon/client.py
+++ b/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}).")
diff --git a/src/leapflow/daemon/lifecycle.py b/src/leapflow/daemon/lifecycle.py
index ca3d04d..f94794f 100644
--- a/src/leapflow/daemon/lifecycle.py
+++ b/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:
diff --git a/src/leapflow/daemon/server.py b/src/leapflow/daemon/server.py
index 639cb74..0a5bfd9 100644
--- a/src/leapflow/daemon/server.py
+++ b/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()
diff --git a/src/leapflow/daemon/session_registry.py b/src/leapflow/daemon/session_registry.py
index 2bc22aa..1c71ee8 100644
--- a/src/leapflow/daemon/session_registry.py
+++ b/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),
+        )
diff --git a/src/leapflow/dashboard/intent.py b/src/leapflow/dashboard/intent.py
index 44c6eb3..11fa12f 100644
--- a/src/leapflow/dashboard/intent.py
+++ b/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 "")
diff --git a/src/leapflow/storage/duckdb_connect.py b/src/leapflow/storage/duckdb_connect.py
index 3d34a46..c61cd57 100644
--- a/src/leapflow/storage/duckdb_connect.py
+++ b/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"))
diff --git a/src/leapflow/tools/shell_tools.py b/src/leapflow/tools/shell_tools.py
index 6ef64c6..807cf6a 100644
--- a/src/leapflow/tools/shell_tools.py
+++ b/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
diff --git a/src/leapflow/utils/shell_lex.py b/src/leapflow/utils/shell_lex.py
new file mode 100644
index 0000000..d8b0dc7
--- /dev/null
+++ b/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)
diff --git a/tests/_harness/cassette.py b/tests/_harness/cassette.py
index 3419b87..8569f64 100644
--- a/tests/_harness/cassette.py
+++ b/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>"),
 )
diff --git a/tests/_harness/leapd.py b/tests/_harness/leapd.py
index bcc1345..b9c1479 100644
--- a/tests/_harness/leapd.py
+++ b/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 "
diff --git a/tests/conftest.py b/tests/conftest.py
index 956f86f..cfdccf3 100644
--- a/tests/conftest.py
+++ b/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
 # ════════════════════════════════════════════════════════════════
diff --git a/tests/journeys/test_r6_lifecycle.py b/tests/journeys/test_r6_lifecycle.py
index cfeae7b..410dcb4 100644
--- a/tests/journeys/test_r6_lifecycle.py
+++ b/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"
diff --git a/tests/regression/test_test_layer_contracts.py b/tests/regression/test_test_layer_contracts.py
index 126375f..66ab553 100644
--- a/tests/regression/test_test_layer_contracts.py
+++ b/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]:
diff --git a/tests/test_cli_entrypoint.py b/tests/test_cli_entrypoint.py
index c65b1e6..46098e5 100644
--- a/tests/test_cli_entrypoint.py
+++ b/tests/test_cli_entrypoint.py
@@ -941,12 +941,13 @@ def stop(self) -> None:
 async def test_host_status_reports_daemon_host_backend(monkeypatch, tmp_path, capsys) -> None:
     from conftest import make_settings
     from leapflow.cli.commands import host as host_module
+    from leapflow.daemon._transport import get_transport
 
     class Info:
         pid = 123
         is_healthy = True
         is_running = True
-        sock_path = tmp_path / "leapd.sock"
+        sock_path = get_transport().readiness_path(tmp_path)
 
     settings = replace(make_settings(str(tmp_path)), use_cua_driver=True)
 
diff --git a/tests/test_code_tools.py b/tests/test_code_tools.py
index b454da2..6b753a6 100644
--- a/tests/test_code_tools.py
+++ b/tests/test_code_tools.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import asyncio
+from pathlib import PurePath
 
 from leapflow.tools import file_operations as fo
 from leapflow.tools.file_operations import code_search, edit_file, file_find, file_write
@@ -27,7 +28,7 @@ def test_code_search_finds_matches_with_location(tmp_path) -> None:
 
     assert result["ok"] is True
     assert result["match_count"] >= 2
-    hits = {(m["path"].split("/")[-1], m["line"]) for m in result["matches"]}
+    hits = {(PurePath(m["path"]).name, m["line"]) for m in result["matches"]}
     assert ("a.py", 2) in hits and ("b.py", 1) in hits
     assert all("text" in m and m["line"] for m in result["matches"])
 
@@ -165,7 +166,7 @@ def test_file_find_recursive_glob(tmp_path) -> None:
     result = _run(file_find({"glob": "*.py", "path": str(tmp_path)}))
 
     assert result["ok"] is True
-    found = {p.split("/")[-1] for p in result["files"]}
+    found = {PurePath(p).name for p in result["files"]}
     assert found == {"mod.py", "top.py"}
 
 
@@ -176,7 +177,7 @@ def test_file_find_skips_dep_dirs(tmp_path) -> None:
 
     result = _run(file_find({"glob": "*.py", "path": str(tmp_path)}))
 
-    assert [p.split("/")[-1] for p in result["files"]] == ["keep.py"]
+    assert [PurePath(p).name for p in result["files"]] == ["keep.py"]
 
 
 def test_file_find_truncates_at_max_results(tmp_path) -> None:
diff --git a/tests/test_daemon_event_loop_blocking.py b/tests/test_daemon_event_loop_blocking.py
index 531b89a..cc1f797 100644
--- a/tests/test_daemon_event_loop_blocking.py
+++ b/tests/test_daemon_event_loop_blocking.py
@@ -263,7 +263,9 @@ async def _ticker() -> None:
             result = await ctx._run_deferred_db(lambda: (time.sleep(0.3), 42)[1])
             elapsed = time.monotonic() - start
             assert result == 42
-            assert elapsed >= 0.3
+            # Windows' monotonic clock quantizes to ~15.6ms, so the measured
+            # interval can read a few ms short of the blocking sleep itself.
+            assert elapsed >= 0.3 - 0.03
         finally:
             stop.set()
             await ticker_task
diff --git a/tests/test_daemon_rpc.py b/tests/test_daemon_rpc.py
index 2e4cda6..2eb97a1 100644
--- a/tests/test_daemon_rpc.py
+++ b/tests/test_daemon_rpc.py
@@ -11,6 +11,8 @@
 
 import pytest
 
+import leapflow
+from leapflow.daemon._transport import get_transport
 from leapflow.daemon.client import DaemonClient, DaemonUnavailableError, ensure_daemon_client
 from leapflow.daemon.lifecycle import DaemonInfo
 from leapflow.daemon.protocol import StreamChunk
@@ -18,7 +20,7 @@
 
 
 def _short_tempdir() -> str:
-    """Base directory for temp runtime dirs holding ``leapd.sock``.
+    """Base directory for temp runtime dirs holding ``leapd.sock`` or ``leapd.port``.
 
     AF_UNIX socket paths cap near 108 chars, and macOS's default TMPDIR
     (/var/folders/...) is too long, so POSIX keeps /tmp; Windows has no
@@ -74,8 +76,12 @@ async def engine_chat(self, message: str, **kwargs: Any) -> AsyncIterator[Stream
 
 
 class _SlowFirstChunkService(_FakeService):
+    def __init__(self, first_chunk_delay: float = 0.16) -> None:
+        super().__init__()
+        self._first_chunk_delay = first_chunk_delay
+
     async def engine_chat(self, message: str, **kwargs: Any) -> AsyncIterator[StreamChunk]:
-        await asyncio.sleep(0.08)
+        await asyncio.sleep(self._first_chunk_delay)
         yield StreamChunk(request_id="", content=f"slow {message}", event_type="chunk")
         yield StreamChunk(request_id="", content="done", event_type="final")
 
@@ -123,13 +129,14 @@ async def engine_chat(self, message: str, **kwargs: Any) -> AsyncIterator[Stream
 async def _start_server(runtime_dir: Path, service=None, *, stream_heartbeat_s: float | None = None):
     server = UnixRpcServer(
         service or _FakeService(),
-        sock_path=runtime_dir / "leapd.sock",
+        sock_path=get_transport().readiness_path(runtime_dir),
         runtime_dir=runtime_dir,
         stream_heartbeat_s=stream_heartbeat_s,
     )
     task = asyncio.create_task(server.serve_forever())
+    readiness = get_transport().readiness_path(runtime_dir)
     for _ in range(50):
-        if (runtime_dir / "leapd.sock").exists():
+        if readiness.exists():
             return server, task, runtime_dir
         await asyncio.sleep(0.02)
     task.cancel()
@@ -140,7 +147,8 @@ async def _start_server(runtime_dir: Path, service=None, *, stream_heartbeat_s:
 async def test_daemon_client_receives_stream_events() -> None:
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime")
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
 
         try:
             events = [event async for event in client.engine_chat("world")]
@@ -164,7 +172,8 @@ async def test_daemon_server_injects_rpc_request_id_into_engine_chat() -> None:
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         service = _RequestIdCaptureService()
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=service)
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
 
         try:
             events = [event async for event in client.engine_chat("world")]
@@ -590,7 +599,8 @@ async def test_daemon_client_can_cancel_engine_turn() -> None:
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         service = _FakeService()
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=service)
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
 
         try:
             cancelled = await client.engine_cancel()
@@ -611,10 +621,14 @@ async def test_daemon_client_stream_heartbeat_prevents_idle_timeout() -> None:
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(
             Path(root) / "runtime",
-            service=_SlowFirstChunkService(),
-            stream_heartbeat_s=0.01,
+            service=_SlowFirstChunkService(first_chunk_delay=0.6),
+            stream_heartbeat_s=0.1,
         )
-        client = DaemonClient(runtime_dir / "leapd.sock", timeout_s=0.03)
+        socket_path = get_transport().readiness_path(runtime_dir)
+        # The read timeout must stay smaller than the first-chunk delay (so an
+        # idle stream would die) but large enough that slow CI dispatch latency
+        # cannot starve the first heartbeat; 0.3s keeps both invariants.
+        client = DaemonClient(socket_path, timeout_s=0.3)
 
         try:
             events = [event async for event in client.engine_chat("world")]
@@ -644,7 +658,8 @@ async def test_dispatch_stream_keeps_contextvar_token_valid_across_chunks() -> N
         server, task, runtime_dir = await _start_server(
             Path(root) / "runtime", service=_ContextVarStreamingService(),
         )
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         events = []
         try:
             async for event in client.engine_chat("hi"):
@@ -1000,7 +1015,8 @@ def reload_runtime_config_if_changed(self) -> bool:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=service)
 
         async def consume(session_id: str, decision: str) -> list[str]:
-            client = DaemonClient(runtime_dir / "leapd.sock")
+            socket_path = get_transport().readiness_path(runtime_dir)
+            client = DaemonClient(socket_path)
             contents: list[str] = []
             async for event in client.engine_chat("hi", session_id=session_id):
                 if event.type == "approval_request":
@@ -1085,7 +1101,8 @@ def reload_runtime_config_if_changed(self) -> bool:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=service)
 
         async def consume(session_id: str) -> list[tuple[str, str, dict[str, Any]]]:
-            client = DaemonClient(runtime_dir / "leapd.sock")
+            socket_path = get_transport().readiness_path(runtime_dir)
+            client = DaemonClient(socket_path)
             events: list[tuple[str, str, dict[str, Any]]] = []
             async for event in client.engine_chat("hi", session_id=session_id):
                 metadata = dict(event.metadata or {})
@@ -1232,19 +1249,21 @@ async def shutdown(self) -> None:
         shutdown_event = asyncio.Event()
         server = UnixRpcServer(
             service,
-            sock_path=runtime_dir / "leapd.sock",
+            sock_path=get_transport().readiness_path(runtime_dir),
             runtime_dir=runtime_dir,
             on_shutdown=shutdown_event.set,
         )
         task = asyncio.create_task(server.serve_forever())
+        readiness = get_transport().readiness_path(runtime_dir)
         for _ in range(50):
-            if (runtime_dir / "leapd.sock").exists():
+            if readiness.exists():
                 break
             await asyncio.sleep(0.02)
         else:
             task.cancel()
             raise AssertionError("server did not start")
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         try:
             await client.shutdown()
             await asyncio.wait_for(shutdown_event.wait(), timeout=1.0)
@@ -1307,7 +1326,8 @@ async def host_restart(self) -> dict[str, Any]:
 
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=HostRpcService())
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         try:
             status = await client.host_status()
             started = await client.host_start()
@@ -1365,7 +1385,8 @@ async def app_command(self, args: str = "") -> dict[str, Any]:
 
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=SlashMetadataService())
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         try:
             tools = await client.tools_list()
             usage = await client.usage_summary()
@@ -1516,7 +1537,8 @@ async def approval_cancel(self, pending_id: str, reason: str = "cancelled") -> d
 
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime", service=ApprovalRpcService())
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         try:
             result = await client.approval_resolve("p1", "allow_once", reason="user")
             status = await client.approval_status()
@@ -1538,7 +1560,8 @@ async def approval_cancel(self, pending_id: str, reason: str = "cancelled") -> d
 async def test_daemon_client_reports_unknown_method() -> None:
     with tempfile.TemporaryDirectory(prefix="lfd-", dir=_short_tempdir()) as root:
         server, task, runtime_dir = await _start_server(Path(root) / "runtime")
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
 
         try:
             with pytest.raises(DaemonUnavailableError, match="Unknown method"):
@@ -1563,7 +1586,7 @@ async def test_ensure_daemon_client_reuses_healthy_daemon() -> None:
 
         try:
             client = await ensure_daemon_client(settings)
-            assert client.sock_path == runtime_dir / "leapd.sock"
+            assert client.sock_path == get_transport().readiness_path(runtime_dir)
             status = await client.status()
         finally:
             task.cancel()
@@ -1583,7 +1606,8 @@ async def test_daemon_client_surfaces_stream_errors() -> None:
             Path(root) / "runtime",
             service=_FailingStreamService(),
         )
-        client = DaemonClient(runtime_dir / "leapd.sock")
+        socket_path = get_transport().readiness_path(runtime_dir)
+        client = DaemonClient(socket_path)
         events = []
 
         try:
@@ -1665,7 +1689,7 @@ async def test_runtime_service_hot_reloads_config_before_daemon_chat(
         assert status["runtime_dir"] == str(settings.runtime_dir)
         assert status["llm_context_length"] == 700_000
         assert status["context_used"] == 0
-        assert status["runtime_source"].endswith("leapflow/__init__.py")
+        assert status["runtime_source"] == str(leapflow.__file__)
         assert status["runtime_executable"]
         assert status["runtime_version"]
     finally:
@@ -1779,7 +1803,7 @@ async def test_ensure_daemon_client_does_not_spawn_when_daemon_unhealthy(
     settings = make_settings(str(tmp_path))
     unhealthy = DaemonInfo(
         pid=4321,
-        sock_path=settings.runtime_dir / "leapd.sock",
+        sock_path=get_transport().readiness_path(settings.runtime_dir),
         start_time=1.0,
         is_running=True,
         is_healthy=False,
@@ -1806,7 +1830,7 @@ async def test_recover_daemon_client_probes_rpc_and_force_restarts(
     from leapflow.daemon.lifecycle import StopDaemonResult
 
     settings = make_settings(str(tmp_path))
-    sock_path = settings.runtime_dir / "leapd.sock"
+    sock_path = get_transport().readiness_path(settings.runtime_dir)
     attempts = 0
 
     async def fake_ensure(*args, **kwargs):
@@ -1911,7 +1935,7 @@ def test_stop_daemon_sends_sigterm_waits_and_cleans(monkeypatch, tmp_path) -> No
 
     running = lifecycle_module.DaemonInfo(
         pid=1234,
-        sock_path=tmp_path / "runtime" / "leapd.sock",
+        sock_path=get_transport().readiness_path(tmp_path / "runtime"),
         start_time=None,
         is_running=True,
         is_healthy=True,
@@ -1950,7 +1974,7 @@ def test_stop_daemon_force_escalates_after_timeout(monkeypatch, tmp_path) -> Non
 
     running = lifecycle_module.DaemonInfo(
         pid=1234,
-        sock_path=tmp_path / "runtime" / "leapd.sock",
+        sock_path=get_transport().readiness_path(tmp_path / "runtime"),
         start_time=None,
         is_running=True,
         is_healthy=False,
@@ -1979,7 +2003,7 @@ def test_stop_daemon_narrates_progress(monkeypatch, tmp_path) -> None:
 
     running = lifecycle_module.DaemonInfo(
         pid=4885,
-        sock_path=tmp_path / "runtime" / "leapd.sock",
+        sock_path=get_transport().readiness_path(tmp_path / "runtime"),
         start_time=None,
         is_running=True,
         is_healthy=False,
diff --git a/tests/test_dev_terminal_tools.py b/tests/test_dev_terminal_tools.py
index ad0a122..64a0e95 100644
--- a/tests/test_dev_terminal_tools.py
+++ b/tests/test_dev_terminal_tools.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 
 from leapflow.tools import dev_tools
 from leapflow.tools import terminal_session as ts
@@ -113,18 +114,22 @@ def test_terminal_disabled_by_default() -> None:
 
 def test_terminal_session_lifecycle() -> None:
     ts.set_terminal_sessions_enabled(True)
+    sid = ""
     try:
         opened = _run(terminal_open({}))
-        assert opened["ok"] is True
+        assert opened["ok"] is True, opened
         sid = opened["session_id"]
 
-        sent = _run(terminal_send({"session_id": sid, "input": "echo hello-term", "wait": 0.6}))
+        # PowerShell cold-starts slowly (banner + profile + first prompt),
+        # especially under parallel xdist load; widen the capture budget there.
+        send_wait, read_wait, reads = (2.0, 1.0, 8) if sys.platform == "win32" else (0.6, 0.4, 5)
+        sent = _run(terminal_send({"session_id": sid, "input": "echo hello-term", "wait": send_wait}))
         assert sent["ok"] is True
         output = sent["output"]
-        for _ in range(5):
+        for _ in range(reads):
             if "hello-term" in output:
                 break
-            output += _run(terminal_read({"session_id": sid, "wait": 0.4}))["output"]
+            output += _run(terminal_read({"session_id": sid, "wait": read_wait}))["output"]
         assert "hello-term" in output
 
         listed = _run(terminal_list({}))
@@ -132,10 +137,13 @@ def test_terminal_session_lifecycle() -> None:
 
         closed = _run(terminal_close({"session_id": sid}))
         assert closed["ok"] is True and closed["closed"] is True
+        sid = ""
 
         after = _run(terminal_send({"session_id": sid, "input": "x"}))
         assert after["ok"] is False and after["failure_code"] == "session_not_found"
     finally:
+        if sid:
+            _run(terminal_close({"session_id": sid}))
         ts.set_terminal_sessions_enabled(False)
 
 
@@ -153,9 +161,9 @@ def test_terminal_max_sessions(monkeypatch) -> None:
     monkeypatch.setattr(ts, "_MAX_SESSIONS", 1)
     opened = _run(terminal_open({}))
     try:
-        assert opened["ok"] is True
+        assert opened["ok"] is True, opened
         second = _run(terminal_open({}))
         assert second["ok"] is False and second["failure_code"] == "too_many_sessions"
     finally:
-        _run(terminal_close({"session_id": opened["session_id"]}))
+        _run(terminal_close({"session_id": opened.get("session_id", "")}))
         ts.set_terminal_sessions_enabled(False)
diff --git a/tests/test_process_group.py b/tests/test_process_group.py
index 0def35c..fb8611e 100644
--- a/tests/test_process_group.py
+++ b/tests/test_process_group.py
@@ -70,6 +70,7 @@ def test_second_terminate_is_false() -> None:
     group = ProcessGroup()
     group.attach(proc.pid)
     assert group.terminate() is True
+    proc.wait(timeout=5)  # reap on POSIX: a zombie keeps the group alive
     assert group.terminate() is False  # job handle released / pgid already dead
 
 
diff --git a/tests/test_tool_call_hardening.py b/tests/test_tool_call_hardening.py
index c03dbe1..6b93866 100644
--- a/tests/test_tool_call_hardening.py
+++ b/tests/test_tool_call_hardening.py
@@ -9,6 +9,7 @@
 
 import asyncio
 import json
+import os
 
 from leapflow.engine.engine import (
     _head_tail_truncate,
@@ -170,8 +171,12 @@ def test_compact_error_preserves_stderr_without_error_field() -> None:
 
 
 def test_shell_run_populates_error_on_failure() -> None:
+    import os
+
     from leapflow.tools.shell_tools import shell_run
-    result = asyncio.run(shell_run({"command": "echo BOOM_ERR 1>&2; exit 2"}))
+    # cmd.exe (the Windows shell backend) has no ';' separator; '&&' works on both.
+    joiner = "&&" if os.name == "nt" else ";"
+    result = asyncio.run(shell_run({"command": f"echo BOOM_ERR 1>&2 {joiner} exit 2"}))
     assert result["ok"] is False and result["returncode"] == 2
     assert "BOOM_ERR" in result["error"] and "BOOM_ERR" in result["stderr"]
 
@@ -220,8 +225,9 @@ def test_workspace_context_resolves_relative_paths_and_blocks_cross_workspace(tm
         assert repo_result["ok"] is True
         assert repo_result["root"] == str(workspace.resolve())
 
-        shell_result = asyncio.run(shell_run({"command": "pwd"}))
-        assert shell_result["ok"] is True
+        pwd_command = "echo %CD%" if os.name == "nt" else "pwd"
+        shell_result = asyncio.run(shell_run({"command": pwd_command}))
+        assert shell_result["ok"] is True, shell_result
         assert shell_result["cwd"] == str(workspace.resolve())
 
         blocked = asyncio.run(file_read({"path": str(other / "secret.txt")}))
@@ -251,6 +257,8 @@ def test_shell_gate_blocks_expanded_and_relative_escapes(tmp_path, monkeypatch)
     )
     from leapflow.tools.shell_tools import shell_run
 
+    import os
+
     workspace = tmp_path / "work"
     other = tmp_path / "other"
     workspace.mkdir()
@@ -277,8 +285,16 @@ def test_shell_gate_blocks_expanded_and_relative_escapes(tmp_path, monkeypatch)
             assert result["resolved_path"].startswith(str(other.resolve())), command
 
         # Traversal that stays inside must still run: the gate judges the resolved
-        # target, so `sub/../alpha.txt` is an ordinary in-workspace read.
-        inside = asyncio.run(shell_run({"command": "cat sub/../alpha.txt"}))
+        # target, so `sub/../alpha.txt` is an ordinary in-workspace read, and the
+        # content must come back. On Windows no cmd-era builtin reads a '..'
+        # operand (Git's MSYS cat falls back to stdin and hangs, cmd's `type`
+        # rejects the path), so the read goes through PowerShell's Get-Content —
+        # as a command argument; the shell backend stays cmd.
+        if os.name == "nt":
+            command = 'powershell -NoProfile -Command "Get-Content sub/../alpha.txt"'
+        else:
+            command = "cat sub/../alpha.txt"
+        inside = asyncio.run(shell_run({"command": command}))
         assert inside["ok"] is True
         assert inside["stdout"].strip() == "inside"
     finally:
diff --git a/tools/impact.py b/tools/impact.py
index 51fc76d..237aee1 100644
--- a/tools/impact.py
+++ b/tools/impact.py
@@ -193,7 +193,7 @@ def test_modules_importing(modules: set[str]) -> set[str]:
                 for name in names
                 for module in modules
             ):
-                selected.add(str(path.relative_to(REPO_ROOT)))
+                selected.add(path.relative_to(REPO_ROOT).as_posix())
                 break
     return selected
 
@@ -338,7 +338,7 @@ def journey_metadata() -> list[JourneyMeta]:
                 continue
         found.append(
             JourneyMeta(
-                path=str(path.relative_to(REPO_ROOT)),
+                path=path.relative_to(REPO_ROOT).as_posix(),
                 subject_paths=subjects,
                 live_signal=live,
             )
@@ -451,7 +451,7 @@ def build_map() -> int:
         if not test_file.endswith(".py"):
             continue
         try:
-            relative = str(Path(file_path).resolve().relative_to(REPO_ROOT))
+            relative = Path(file_path).resolve().relative_to(REPO_ROOT).as_posix()
         except ValueError:
             continue
         mapping.setdefault(test_file, set()).add(relative)