diff --git a/bench/contradiction_longitudinal.py b/bench/contradiction_longitudinal.py
index 5bac9d8..3b23101 100644
--- a/bench/contradiction_longitudinal.py
+++ b/bench/contradiction_longitudinal.py
@@ -812,9 +812,9 @@ def write_outputs(
 
     env_with_duration = {**env, "wall_clock_duration_seconds": round(duration_seconds, 2)}
     json_blob = {"env": env_with_duration, "summary": summary}
-    json_path.write_text(json.dumps(json_blob, indent=2, default=str))
+    json_path.write_text(json.dumps(json_blob, indent=2, default=str), encoding="utf-8")
 
-    with csv_path.open("w", newline="") as fh:
+    with csv_path.open("w", newline="", encoding="utf-8") as fh:
         w = csv.writer(fh)
         w.writerow([
             "probe_id", "seed", "n_slice", "condition", "topic",
@@ -897,7 +897,7 @@ def write_outputs(
         "- **Metric B-contract (s4_contradiction hint OR anti_hits ≥80%)** tests what the system actually promises (REQUIREMENTS.md MEM-08, MCP-01 dual-route). Cosine cannot do either; pipeline either signals contradictions or it doesn't.",
         "",
     ]
-    md_path.write_text("\n".join(lines))
+    md_path.write_text("\n".join(lines), encoding="utf-8")
 
     print(f"[bench] Wrote: {md_path}")
     print(f"[bench] Wrote: {json_path}")
diff --git a/mcp-wrapper/src/bridge.ts b/mcp-wrapper/src/bridge.ts
index 07db282..5f41fbf 100644
--- a/mcp-wrapper/src/bridge.ts
+++ b/mcp-wrapper/src/bridge.ts
@@ -3,6 +3,7 @@ import * as crypto from "node:crypto";
 import * as net from "node:net";
 import {
   type ConnectTarget,
+  authenticateConnection,
   createDaemonConnection,
   daemonUnreachableHint,
   getDaemonConnectTarget,
@@ -74,13 +75,15 @@ export class PythonCoreBridge {
       throw new DaemonUnreachableError(daemonUnreachableHint());
     }
 
-    let sock: net.Socket;
+    let sock: net.Socket | null = null;
     try {
       sock = await this.connectWithTimeout(
         target,
         SOCKET_CONNECT_TIMEOUT_MS,
       );
+      authenticateConnection(sock, target);
     } catch (e) {
+      try { sock?.destroy(); } catch {  }
       throw new DaemonUnreachableError(daemonUnreachableHint());
     }
     this.sock = sock;
@@ -198,10 +201,12 @@ export class PythonCoreBridge {
         if (target === null) {
           return;
         }
-        this.sock = await this.connectWithTimeout(
+        const sock = await this.connectWithTimeout(
           target,
           SOCKET_CONNECT_TIMEOUT_MS,
         );
+        authenticateConnection(sock, target);
+        this.sock = sock;
         this.attachSocketHandlers();
       } catch {
       } finally {
@@ -279,6 +284,12 @@ export function emitSessionOpen(sessionId: string): Promise<void> {
         return;
       }
       const sock = createDaemonConnection(target, () => {
+        try {
+          authenticateConnection(sock, target);
+        } catch {
+          try { sock.destroy(); } catch {  }
+          return;
+        }
         const msg =
           JSON.stringify({
             type: "session_open",
diff --git a/mcp-wrapper/src/ipc.ts b/mcp-wrapper/src/ipc.ts
index b361983..261972c 100644
--- a/mcp-wrapper/src/ipc.ts
+++ b/mcp-wrapper/src/ipc.ts
@@ -40,6 +40,28 @@ export function readDaemonPort(): number | null {
   }
 }
 
+/**
+ * Windows-only auth token file, mirroring `_ipc.TOKEN_FILE` on the Python
+ * side. The daemon requires every TCP client to send this token as the
+ * first line on a fresh connection (see `_make_authenticated_handler` /
+ * `_send_token_sync` in `_ipc.py`) — a plain connect is not enough.
+ * IAI_DAEMON_TOKEN_PATH overrides for tests.
+ */
+export function daemonTokenPath(): string {
+  const env = process.env.IAI_DAEMON_TOKEN_PATH;
+  if (env) return env;
+  return path.join(daemonBaseDir(), ".daemon.token");
+}
+
+export function readDaemonToken(): string | null {
+  try {
+    const txt = fs.readFileSync(daemonTokenPath(), "utf-8").trim();
+    return txt.length > 0 ? txt : null;
+  } catch {
+    return null;
+  }
+}
+
 /**
  * Resolve the daemon IPC endpoint.
  *   POSIX   -> Unix-domain socket path (string)
@@ -50,6 +72,16 @@ export function readDaemonPort(): number | null {
 export function getDaemonConnectTarget(): ConnectTarget | null {
   const env = process.env.IAI_DAEMON_SOCKET_PATH;
   if (env) return env;
+  // Test-only escape hatch so the TCP+auth-token path can be exercised
+  // without depending on the real process.platform (IS_WINDOWS is a
+  // module-load-time constant and can't be swapped mid-test-run).
+  const tcpPort = process.env.IAI_DAEMON_TCP_PORT;
+  if (tcpPort) {
+    const port = Number.parseInt(tcpPort, 10);
+    if (Number.isFinite(port) && port > 0) {
+      return { host: process.env.IAI_DAEMON_TCP_HOST ?? "127.0.0.1", port };
+    }
+  }
   if (IS_WINDOWS) {
     const port = readDaemonPort();
     return port === null ? null : { host: "127.0.0.1", port };
@@ -95,3 +127,24 @@ export function createDaemonConnection(
     ? net.createConnection(target.port, target.host, connectListener)
     : net.createConnection(target.port, target.host);
 }
+
+/**
+ * Perform the auth-token handshake a TCP connection requires before the
+ * daemon will process any RPC traffic on it. Unix-socket targets (POSIX)
+ * need no handshake and are a no-op here.
+ *
+ * Must be called synchronously right after connect, before anything else
+ * is written to the socket — the daemon's `_make_authenticated_handler`
+ * requires the token to be the very first line, and closes the connection
+ * outright (no error response) if it isn't.
+ */
+export function authenticateConnection(sock: net.Socket, target: ConnectTarget): void {
+  if (typeof target === "string") return;
+  const token = readDaemonToken();
+  if (token === null) {
+    throw new Error(
+      `Daemon auth token not found: ${daemonTokenPath()} missing. The daemon may need a restart to regenerate it.`,
+    );
+  }
+  sock.write(token + "\n");
+}
diff --git a/mcp-wrapper/test/bridge.test.ts b/mcp-wrapper/test/bridge.test.ts
index fae275e..722061d 100644
--- a/mcp-wrapper/test/bridge.test.ts
+++ b/mcp-wrapper/test/bridge.test.ts
@@ -1,7 +1,7 @@
 
 import { describe, it, afterEach } from "node:test";
 import { strict as assert } from "node:assert";
-import { mkdtemp, rm } from "node:fs/promises";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import * as net from "node:net";
@@ -97,6 +97,9 @@ afterEach(async () => {
   tmpDirs = [];
   delete process.env.IAI_DAEMON_SOCKET_PATH;
   delete process.env.IAI_MCP_RECONNECT_TEST_DELAY_MS;
+  delete process.env.IAI_DAEMON_TCP_PORT;
+  delete process.env.IAI_DAEMON_TCP_HOST;
+  delete process.env.IAI_DAEMON_TOKEN_PATH;
 });
 
 async function setup(): Promise<{ daemon: MockDaemon; bridge: PythonCoreBridge }> {
@@ -419,3 +422,142 @@ describe("PythonCoreBridge: rapid connection flaps", () => {
     assert.equal(bridge.isConnected(), true);
   });
 });
+
+
+// Simulates the Windows daemon's `_make_authenticated_handler`: the first
+// line on every connection must equal `expectedToken`, or the connection is
+// closed immediately with nothing written back — mirroring _ipc.py exactly,
+// including the "silently closed, no error frame" behavior that produced the
+// `daemon_unreachable: socket closed` symptom this test guards against.
+class MockAuthDaemon {
+  private server: net.Server;
+  private connections: net.Socket[] = [];
+  private readonly expectedToken: string;
+  public requestHandler: (req: { id: number; method: string; params: unknown }) => unknown;
+
+  constructor(expectedToken: string) {
+    this.expectedToken = expectedToken;
+    this.requestHandler = (req) => ({ ok: true, method: req.method });
+    this.server = net.createServer((conn) => {
+      this.connections.push(conn);
+      let authenticated = false;
+      let buf = "";
+      conn.on("data", (chunk) => {
+        buf += chunk.toString("utf-8");
+        let nl: number;
+        while ((nl = buf.indexOf("\n")) >= 0) {
+          const line = buf.slice(0, nl).trim();
+          buf = buf.slice(nl + 1);
+          if (!authenticated) {
+            authenticated = true;
+            if (line !== this.expectedToken) {
+              conn.destroy();
+              return;
+            }
+            continue;
+          }
+          if (!line) continue;
+          try {
+            const req = JSON.parse(line);
+            const result = this.requestHandler(req);
+            conn.write(JSON.stringify({ jsonrpc: "2.0", id: req.id, result }) + "\n");
+          } catch (e) {
+            const req = JSON.parse(line);
+            conn.write(
+              JSON.stringify({ jsonrpc: "2.0", id: req.id, error: { code: -32603, message: String(e) } }) + "\n",
+            );
+          }
+        }
+      });
+    });
+  }
+
+  async listen(): Promise<number> {
+    return new Promise((resolve) => {
+      this.server.listen(0, "127.0.0.1", () => {
+        resolve((this.server.address() as net.AddressInfo).port);
+      });
+    });
+  }
+
+  async close(): Promise<void> {
+    for (const c of this.connections) { try { c.destroy(); } catch {  } }
+    this.connections = [];
+    return new Promise((resolve) => { this.server.close(() => resolve()); });
+  }
+}
+
+describe("PythonCoreBridge: TCP auth-token handshake (Windows transport)", () => {
+  it("authenticates with the token before any RPC traffic and completes calls normally", async () => {
+    const dir = await makeTmpDir();
+    tmpDirs.push(dir);
+    const tokenPath = join(dir, ".daemon.token");
+    const token = "test-token-abc123";
+    await writeFile(tokenPath, token, "utf-8");
+
+    const daemon = new MockAuthDaemon(token);
+    const port = await daemon.listen();
+
+    process.env.IAI_DAEMON_TCP_PORT = String(port);
+    process.env.IAI_DAEMON_TOKEN_PATH = tokenPath;
+    const bridge = new PythonCoreBridge();
+    bridges.push(bridge);
+
+    await bridge.start();
+    assert.equal(bridge.isConnected(), true);
+
+    const res = await bridge.call<{ echo: string }>("ping");
+    assert.equal(res.method, "ping");
+
+    await daemon.close();
+  });
+
+  it("fails as daemon_unreachable when the token file is missing", async () => {
+    const dir = await makeTmpDir();
+    tmpDirs.push(dir);
+    const missingTokenPath = join(dir, ".daemon.token"); // never written
+
+    const daemon = new MockAuthDaemon("irrelevant-since-no-token-sent");
+    const port = await daemon.listen();
+
+    process.env.IAI_DAEMON_TCP_PORT = String(port);
+    process.env.IAI_DAEMON_TOKEN_PATH = missingTokenPath;
+    const bridge = new PythonCoreBridge();
+    bridges.push(bridge);
+
+    await assert.rejects(
+      () => bridge.start(),
+      (err: Error) => err instanceof DaemonUnreachableError,
+    );
+    assert.equal(bridge.isConnected(), false);
+
+    await daemon.close();
+  });
+
+  it("gets disconnected by the daemon when the token is wrong (pre-fix regression case)", async () => {
+    const dir = await makeTmpDir();
+    tmpDirs.push(dir);
+    const tokenPath = join(dir, ".daemon.token");
+    await writeFile(tokenPath, "wrong-token", "utf-8");
+
+    const daemon = new MockAuthDaemon("actual-expected-token");
+    const port = await daemon.listen();
+
+    process.env.IAI_DAEMON_TCP_PORT = String(port);
+    process.env.IAI_DAEMON_TOKEN_PATH = tokenPath;
+    const bridge = new PythonCoreBridge();
+    bridges.push(bridge);
+
+    // The mismatched-token socket is accepted at the TCP level and only
+    // closed after the daemon reads the (wrong) first line, so start()
+    // itself resolves; the failure shows up as the daemon severing the
+    // connection right after, exactly like the bug this test targets.
+    await bridge.start();
+    await assert.rejects(
+      () => bridge.call("ping"),
+      (err: Error) => err.message.includes("daemon_unreachable"),
+    );
+
+    await daemon.close();
+  });
+});
diff --git a/src/iai_mcp/_ipc.py b/src/iai_mcp/_ipc.py
index 48e7cd2..361db12 100644
--- a/src/iai_mcp/_ipc.py
+++ b/src/iai_mcp/_ipc.py
@@ -35,6 +35,13 @@
 
 _TOKEN_BYTES = 32  # 256-bit random token → 64 hex chars on the wire
 
+# The auth token the *current process* generated for its listening server, held
+# in memory for the daemon's whole lifetime. Populated by _generate_token (only
+# the daemon calls that). Lets the daemon re-assert a vanished token file via
+# reassert_token_if_missing() without re-reading it from disk. Stays None in
+# client processes, which must never re-create the daemon's token.
+_CURRENT_TOKEN: str | None = None
+
 
 # ---------------------------------------------------------------------------
 # Port file helpers (Windows only)
@@ -105,14 +112,45 @@ def _token_file_path() -> Path:
 
 def _generate_token() -> str:
     """Generate a fresh 32-byte random token and persist it to the token file."""
+    global _CURRENT_TOKEN
     token = secrets.token_hex(_TOKEN_BYTES)
     path = _token_file_path()
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(token, encoding="utf-8")
     _restrict_token_file(path)
+    _CURRENT_TOKEN = token
     return token
 
 
+def reassert_token_if_missing() -> bool:
+    """Re-write the Windows auth-token file from the in-memory token if it has
+    gone missing while the daemon is running.
+
+    The token file is a single point of failure: it is written once at startup,
+    but if it later disappears for any reason (external cleanup, AV quarantine,
+    a stale-file sweep) the TCP port stays up while EVERY new client fails the
+    auth handshake — the daemon looks alive but new sessions and ``daemon
+    status`` all report "not running". The daemon still holds the token in
+    memory, so we can restore the file cheaply. Called from the daemon's
+    periodic tick.
+
+    Returns True iff it just re-created the file. No-op (returns False) on POSIX,
+    in client processes (``_CURRENT_TOKEN is None``), or when the file is present.
+    """
+    if not IS_WINDOWS or _CURRENT_TOKEN is None:
+        return False
+    path = _token_file_path()
+    if path.exists():
+        return False
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(_CURRENT_TOKEN, encoding="utf-8")
+        _restrict_token_file(path)
+    except OSError:
+        return False
+    return True
+
+
 def _read_token() -> str | None:
     try:
         return _token_file_path().read_text(encoding="utf-8").strip()
@@ -121,6 +159,11 @@ def _read_token() -> str | None:
 
 
 def _remove_token_file() -> None:
+    # Clear the in-memory copy first so reassert_token_if_missing() cannot
+    # resurrect a token the daemon is deliberately tearing down (clean shutdown,
+    # in-process restart, tests).
+    global _CURRENT_TOKEN
+    _CURRENT_TOKEN = None
     try:
         _token_file_path().unlink()
     except (FileNotFoundError, OSError):
diff --git a/src/iai_mcp/cli/__init__.py b/src/iai_mcp/cli/__init__.py
index cf4dca2..d21efc2 100644
--- a/src/iai_mcp/cli/__init__.py
+++ b/src/iai_mcp/cli/__init__.py
@@ -250,6 +250,10 @@ def _maintenance_compact_metrics(
     cmd_build_native,
     cmd_migrate,
     cmd_deferred_unlock_dead_pids,
+    cmd_episodes_recent,
+    cmd_curiosity_pending,
+    cmd_events_query,
+    cmd_schema_list,
 )
 
 from ._maintenance import (
@@ -1064,6 +1068,53 @@ def _cmd_doctor_lazy(args: argparse.Namespace) -> int:
     )
     dpf.set_defaults(func=cmd_drain_permanent_failed)
 
+    er = sub.add_parser(
+        "episodes-recent",
+        help=(
+            "recent captured user turns. Routes through daemon socket when "
+            "daemon is running; direct-open fallback when it is not."
+        ),
+    )
+    er.add_argument("--n", type=int, default=10, help="how many turns (default 10, max 1000)")
+    er.add_argument("--session-id", default=None, help="filter to a specific session UUID")
+    er.set_defaults(func=cmd_episodes_recent)
+
+    cp = sub.add_parser(
+        "curiosity-pending",
+        help=(
+            "pending curiosity questions queued by the sleep daemon. "
+            "Routes through daemon socket when running; direct-open "
+            "fallback when it is not."
+        ),
+    )
+    cp.add_argument("--session-id", default=None, help="filter to a specific session UUID")
+    cp.set_defaults(func=cmd_curiosity_pending)
+
+    eq = sub.add_parser(
+        "events-query",
+        help=(
+            "user-visible event log (contradictions, trajectory metrics, "
+            "etc). Routes through daemon socket when running; direct-open "
+            "fallback when it is not."
+        ),
+    )
+    eq.add_argument("--kind", required=True, help="event kind, e.g. s4_contradiction, trajectory_metric")
+    eq.add_argument("--since", default=None, help="ISO-8601 timestamp; only events at or after this")
+    eq.add_argument("--severity", default=None, choices=["info", "warning", "critical"])
+    eq.add_argument("--limit", type=int, default=100, help="max events (default 100, capped at 1000)")
+    eq.set_defaults(func=cmd_events_query)
+
+    sl = sub.add_parser(
+        "schema-list",
+        help=(
+            "induced Tier-0/Tier-1 schemas. Routes through daemon socket "
+            "when running; direct-open fallback when it is not."
+        ),
+    )
+    sl.add_argument("--domain", default=None, help="only schemas tagged with this domain")
+    sl.add_argument("--confidence-min", type=float, default=0.0, help="minimum confidence (0.0-1.0)")
+    sl.set_defaults(func=cmd_schema_list)
+
     return parser
 
 
diff --git a/src/iai_mcp/cli/_analytics.py b/src/iai_mcp/cli/_analytics.py
index e7fdadf..0604c35 100644
--- a/src/iai_mcp/cli/_analytics.py
+++ b/src/iai_mcp/cli/_analytics.py
@@ -267,6 +267,142 @@ def cmd_bank_recall(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_episodes_recent(args: argparse.Namespace) -> int:
+    """Recent captured user turns. Routes through the daemon socket when the
+    daemon is reachable; direct-open fallback (same pattern as bank-recall /
+    drain-permanent-failed) when it is not — so callers like the nightly
+    digest never depend on an MCP wrapper being up for a plain store read."""
+    import json as _json
+
+    from iai_mcp import cli as _cli
+
+    params: dict = {"n": args.n}
+    if args.session_id:
+        params["session_id"] = args.session_id
+
+    resp = _cli._send_jsonrpc_request("episodes_recent", params)
+    if isinstance(resp, dict) and "result" in resp:
+        result = resp["result"]
+        if isinstance(result, dict):
+            print(_json.dumps(result, ensure_ascii=False))
+            return 0
+
+    from iai_mcp.core._query_dispatch import _episodes_recent_dispatch
+    from iai_mcp.hippo import HippoLockHeldError
+    from iai_mcp.store import MemoryStore
+
+    try:
+        store = MemoryStore()
+        result = _episodes_recent_dispatch(store, params)
+    except HippoLockHeldError:
+        print("daemon holds store lock; retry when daemon is idle")
+        return 0
+
+    print(_json.dumps(result, ensure_ascii=False))
+    return 0
+
+
+def cmd_curiosity_pending(args: argparse.Namespace) -> int:
+    """Pending curiosity questions queued by the sleep daemon. Same
+    daemon-first / direct-open-fallback shape as cmd_episodes_recent."""
+    import json as _json
+
+    from iai_mcp import cli as _cli
+
+    params: dict = {}
+    if args.session_id:
+        params["session_id"] = args.session_id
+
+    resp = _cli._send_jsonrpc_request("curiosity_pending", params)
+    if isinstance(resp, dict) and "result" in resp:
+        result = resp["result"]
+        if isinstance(result, dict):
+            print(_json.dumps(result, ensure_ascii=False))
+            return 0
+
+    from iai_mcp.core._query_dispatch import _curiosity_pending_dispatch
+    from iai_mcp.hippo import HippoLockHeldError
+    from iai_mcp.store import MemoryStore
+
+    try:
+        store = MemoryStore()
+        result = _curiosity_pending_dispatch(store, params)
+    except HippoLockHeldError:
+        print("daemon holds store lock; retry when daemon is idle")
+        return 0
+
+    print(_json.dumps(result, ensure_ascii=False))
+    return 0
+
+
+def cmd_events_query(args: argparse.Namespace) -> int:
+    """User-visible event log (contradictions, trajectory metrics, etc).
+    Same daemon-first / direct-open-fallback shape as cmd_episodes_recent."""
+    import json as _json
+
+    from iai_mcp import cli as _cli
+
+    params: dict = {"kind": args.kind, "limit": args.limit}
+    if args.since:
+        params["since"] = args.since
+    if args.severity:
+        params["severity"] = args.severity
+
+    resp = _cli._send_jsonrpc_request("events_query", params)
+    if isinstance(resp, dict) and "result" in resp:
+        result = resp["result"]
+        if isinstance(result, dict):
+            print(_json.dumps(result, ensure_ascii=False))
+            return 0
+
+    from iai_mcp.core._query_dispatch import _events_query_dispatch
+    from iai_mcp.hippo import HippoLockHeldError
+    from iai_mcp.store import MemoryStore
+
+    try:
+        store = MemoryStore()
+        result = _events_query_dispatch(store, params)
+    except HippoLockHeldError:
+        print("daemon holds store lock; retry when daemon is idle")
+        return 0
+
+    print(_json.dumps(result, ensure_ascii=False))
+    return 0
+
+
+def cmd_schema_list(args: argparse.Namespace) -> int:
+    """Induced Tier-0/Tier-1 schemas. Same daemon-first / direct-open-
+    fallback shape as cmd_episodes_recent."""
+    import json as _json
+
+    from iai_mcp import cli as _cli
+
+    params: dict = {"confidence_min": args.confidence_min}
+    if args.domain:
+        params["domain"] = args.domain
+
+    resp = _cli._send_jsonrpc_request("schema_list", params)
+    if isinstance(resp, dict) and "result" in resp:
+        result = resp["result"]
+        if isinstance(result, dict):
+            print(_json.dumps(result, ensure_ascii=False))
+            return 0
+
+    from iai_mcp.core._query_dispatch import _schema_list_dispatch
+    from iai_mcp.hippo import HippoLockHeldError
+    from iai_mcp.store import MemoryStore
+
+    try:
+        store = MemoryStore()
+        result = _schema_list_dispatch(store, params)
+    except HippoLockHeldError:
+        print("daemon holds store lock; retry when daemon is idle")
+        return 0
+
+    print(_json.dumps(result, ensure_ascii=False))
+    return 0
+
+
 def cmd_topology(args: argparse.Namespace) -> int:  # noqa: ARG001 -- argparse contract
     from iai_mcp import cli as _cli
 
diff --git a/src/iai_mcp/cli/_maintenance.py b/src/iai_mcp/cli/_maintenance.py
index d177c66..0d178fb 100644
--- a/src/iai_mcp/cli/_maintenance.py
+++ b/src/iai_mcp/cli/_maintenance.py
@@ -69,11 +69,21 @@ def _maintenance_compact_preflight_daemon_alive() -> str | None:
     pid = state.get("daemon_pid")
     if not isinstance(pid, int) or pid <= 0:
         return None
+    # NOT os.kill(pid, 0) on Windows: signal 0 raises OSError [WinError 87]
+    # (invalid parameter) even for a live PID, which would make this guard
+    # conclude "no daemon running" and let maintenance proceed against a store
+    # the live daemon still holds. psutil.pid_exists is correct cross-platform;
+    # the psutil block below then confirms the PID is actually the daemon.
     try:
-        _os.kill(pid, 0)
-    except (ProcessLookupError, PermissionError):
-        return None
-    except OSError:
+        import psutil
+        pid_alive = psutil.pid_exists(pid)
+    except Exception:  # noqa: BLE001 -- fall back to POSIX probe
+        try:
+            _os.kill(pid, 0)
+            pid_alive = True
+        except (ProcessLookupError, PermissionError, OSError):
+            pid_alive = False
+    if not pid_alive:
         return None
     try:
         import psutil
diff --git a/src/iai_mcp/core/__init__.py b/src/iai_mcp/core/__init__.py
index 8362f46..06bc21d 100644
--- a/src/iai_mcp/core/__init__.py
+++ b/src/iai_mcp/core/__init__.py
@@ -684,22 +684,7 @@ def dispatch(store: MemoryStore, method: str, params: dict) -> dict:
         }
 
     if method == "curiosity_pending":
-        from iai_mcp.curiosity import pending_questions
-
-        qs = pending_questions(store, params.get("session_id"))
-        return {
-            "questions": [
-                {
-                    "id": str(q.id),
-                    "text": q.text,
-                    "tier": q.tier,
-                    "entropy": q.entropy,
-                    "triggered_by_record_ids": [str(t) for t in q.triggered_by_record_ids],
-                }
-                for q in qs
-            ],
-            "count": len(qs),
-        }
+        return _curiosity_pending_dispatch(store, params)
 
     if method == "trajectory_record":
         from iai_mcp.trajectory import record_session_metrics
@@ -971,32 +956,7 @@ def _norm(ts: str) -> str:
         return {"rendered": rendered, "new_max_ts": new_max_ts}
 
     if method == "episodes_recent":
-        from iai_mcp.capture import read_pending_live_events
-        n = max(0, min(int(params.get("n", 10)), 1000))
-        session_id = params.get("session_id")
-        pending = read_pending_live_events(session_id=session_id)
-        records = store.recent_user_turns(n, session_id=session_id, pending_live_events=pending)
-        turns = []
-        for r in records:
-            if r.id is None:
-                su = getattr(r, "_pending_source_uuid", None)
-                idem = getattr(r, "_pending_idem_tag", "")
-                if su:
-                    rid = f"pending:{su}"
-                else:
-                    idem_hex = idem[5:] if idem.startswith("idem:") else idem
-                    rid = f"pending:{idem_hex}" if idem_hex else f"pending:unknown"
-            else:
-                rid = str(r.id)
-            turns.append({
-                "record_id": rid,
-                "literal_surface": r.literal_surface,
-                "session_id": (r.provenance or [{}])[0].get("session_id"),
-                "captured_at": (
-                    r.created_at.isoformat() if r.created_at else None
-                ),
-            })
-        return {"turns": turns, "count": len(turns)}
+        return _episodes_recent_dispatch(store, params)
 
     if method == "drain_permanent_failed":
         from iai_mcp.capture import drain_permanent_failed_files
@@ -1281,6 +1241,8 @@ def main() -> None:
 from iai_mcp.core._query_dispatch import (  # noqa: E402
     _schema_list_dispatch,
     _events_query_dispatch,
+    _episodes_recent_dispatch,
+    _curiosity_pending_dispatch,
     EVENTS_QUERY_WHITELIST,
 )
 from iai_mcp.core._identity import (  # noqa: E402
@@ -1316,6 +1278,8 @@ def main() -> None:
     "_payload_to_json",
     "_schema_list_dispatch",
     "_events_query_dispatch",
+    "_episodes_recent_dispatch",
+    "_curiosity_pending_dispatch",
     "_DEFAULT_L0_SEED",
 ]
 
diff --git a/src/iai_mcp/core/_query_dispatch.py b/src/iai_mcp/core/_query_dispatch.py
index 2f843cc..a75e57f 100644
--- a/src/iai_mcp/core/_query_dispatch.py
+++ b/src/iai_mcp/core/_query_dispatch.py
@@ -151,3 +151,56 @@ def _events_query_dispatch(store: MemoryStore, params: dict) -> dict:
             "source_ids": e.get("source_ids", []),
         })
     return {"events": out_events, "count": len(out_events)}
+
+
+def _episodes_recent_dispatch(store: MemoryStore, params: dict) -> dict:
+    """Shared by the `episodes_recent` RPC method and the `episodes-recent`
+    CLI direct-open fallback, so the two can never drift."""
+    from iai_mcp.capture import read_pending_live_events
+
+    n = max(0, min(int(params.get("n", 10)), 1000))
+    session_id = params.get("session_id")
+    pending = read_pending_live_events(session_id=session_id)
+    records = store.recent_user_turns(n, session_id=session_id, pending_live_events=pending)
+    turns: list[dict] = []
+    for r in records:
+        if r.id is None:
+            su = getattr(r, "_pending_source_uuid", None)
+            idem = getattr(r, "_pending_idem_tag", "")
+            if su:
+                rid = f"pending:{su}"
+            else:
+                idem_hex = idem[5:] if idem.startswith("idem:") else idem
+                rid = f"pending:{idem_hex}" if idem_hex else "pending:unknown"
+        else:
+            rid = str(r.id)
+        turns.append({
+            "record_id": rid,
+            "literal_surface": r.literal_surface,
+            "session_id": (r.provenance or [{}])[0].get("session_id"),
+            "captured_at": (
+                r.created_at.isoformat() if r.created_at else None
+            ),
+        })
+    return {"turns": turns, "count": len(turns)}
+
+
+def _curiosity_pending_dispatch(store: MemoryStore, params: dict) -> dict:
+    """Shared by the `curiosity_pending` RPC method and the
+    `curiosity-pending` CLI direct-open fallback."""
+    from iai_mcp.curiosity import pending_questions
+
+    qs = pending_questions(store, params.get("session_id"))
+    return {
+        "questions": [
+            {
+                "id": str(q.id),
+                "text": q.text,
+                "tier": q.tier,
+                "entropy": q.entropy,
+                "triggered_by_record_ids": [str(t) for t in q.triggered_by_record_ids],
+            }
+            for q in qs
+        ],
+        "count": len(qs),
+    }
diff --git a/src/iai_mcp/crypto.py b/src/iai_mcp/crypto.py
index 00dc5a4..9bb7826 100644
--- a/src/iai_mcp/crypto.py
+++ b/src/iai_mcp/crypto.py
@@ -158,7 +158,14 @@ def _try_file_set(self, key: bytes) -> None:
             except OSError:
                 pass
         tmp = final.parent / f"{final.name}.tmp.{os.getpid()}"
-        fd = os.open(str(tmp), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        # O_BINARY: on Windows os.open defaults to text mode, which translates
+        # 0x0A bytes in the raw key to 0x0D0A on write and silently corrupts it
+        # (a random 32-byte key almost always contains a 0x0A). No-op on POSIX.
+        fd = os.open(
+            str(tmp),
+            os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0),
+            0o600,
+        )
         try:
             if hasattr(os, "fchmod"):
                 os.fchmod(fd, 0o600)
diff --git a/src/iai_mcp/daemon/__init__.py b/src/iai_mcp/daemon/__init__.py
index ffc5990..50d1775 100644
--- a/src/iai_mcp/daemon/__init__.py
+++ b/src/iai_mcp/daemon/__init__.py
@@ -892,6 +892,28 @@ async def _tick_body(
     capture_queue=None,
     capture_handler: Callable[[dict], None] | None = None,
 ) -> None:
+    # Step 0: re-assert the Windows auth-token file if it vanished while the
+    # daemon was running. Cheap stat; no-op on POSIX. Without this a token file
+    # deleted out-of-band silently bricks every new client connection (the port
+    # stays up, so the daemon looks alive) until a manual restart.
+    try:
+        from iai_mcp._ipc import reassert_token_if_missing
+
+        if reassert_token_if_missing():
+            log.warning("auth token file was missing — re-asserted from memory")
+            try:
+                await asyncio.to_thread(
+                    write_event,
+                    store,
+                    "auth_token_reasserted",
+                    {"phase": "tick"},
+                    severity="warning",
+                )
+            except (OSError, RuntimeError) as exc:  # noqa: BLE001 -- event write non-critical
+                log.debug("auth_token_reasserted event write failed: %s", exc)
+    except Exception:  # noqa: BLE001 -- tick step MUST NOT crash
+        log.debug("tick step 0 (reassert auth token) failed", exc_info=True)
+
     try:
         from iai_mcp.daemon_state import (
             FIRST_TURN_PENDING_TTL_SEC_DEFAULT,
diff --git a/src/iai_mcp/doctor/_lifecycle_checks.py b/src/iai_mcp/doctor/_lifecycle_checks.py
index 7a1c46e..ab4a4cb 100644
--- a/src/iai_mcp/doctor/_lifecycle_checks.py
+++ b/src/iai_mcp/doctor/_lifecycle_checks.py
@@ -65,26 +65,33 @@ def check_a_daemon_alive() -> CheckResult:
             f"daemon_pid={pid!r} is not a valid PID (corrupt state?)",
         )
 
-    try:
-        os.kill(pid, 0)
-    except ProcessLookupError:
-        return CheckResult(
-            "(a) daemon process alive",
-            False,
-            f"PID {pid} in state but no process found",
-        )
-    except PermissionError:
-        return CheckResult(
-            "(a) daemon process alive",
-            False,
-            f"PID {pid} exists but is not owned by this user",
-        )
-    except OSError as e:
-        return CheckResult(
-            "(a) daemon process alive",
-            False,
-            f"liveness probe failed: {type(e).__name__}: {e}",
-        )
+    # os.kill(pid, 0) is the POSIX liveness idiom, but on Windows os.kill
+    # rejects signal 0 with OSError [WinError 87] (invalid parameter) even for
+    # a live PID — which reported a healthy, ticking daemon as dead. Skip the
+    # probe there and rely on the psutil refinement below, which both confirms
+    # the PID exists and that it is an iai_mcp.daemon. Mirrors
+    # lifecycle_lock._is_pid_alive.
+    if platform.system() != "Windows":
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            return CheckResult(
+                "(a) daemon process alive",
+                False,
+                f"PID {pid} in state but no process found",
+            )
+        except PermissionError:
+            return CheckResult(
+                "(a) daemon process alive",
+                False,
+                f"PID {pid} exists but is not owned by this user",
+            )
+        except OSError as e:
+            return CheckResult(
+                "(a) daemon process alive",
+                False,
+                f"liveness probe failed: {type(e).__name__}: {e}",
+            )
 
     try:
         import psutil
@@ -116,8 +123,12 @@ async def _socket_connect_probe(socket_path: Path, timeout: float) -> str | None
     from iai_mcp._ipc import open_ipc_connection
     try:
         reader, writer = await open_ipc_connection(str(socket_path), timeout=timeout)
-    except FileNotFoundError:
-        return "FileNotFoundError"
+    except FileNotFoundError as e:
+        # Surface the actual message: on Windows open_ipc_connection raises
+        # FileNotFoundError for BOTH a missing port file ("Daemon not running")
+        # and a missing auth token ("Daemon auth token not found") — very
+        # different diagnoses, so don't collapse them to a bare class name.
+        return f"FileNotFoundError: {e}"
     except ConnectionRefusedError:
         return "ConnectionRefusedError"
     except asyncio.TimeoutError:
@@ -137,7 +148,7 @@ def check_b_socket_fresh() -> CheckResult:
     # Windows binds TCP loopback and records the port in a sidecar file — there
     # is no AF_UNIX socket file — so check whichever endpoint actually exists
     # for this platform. (The connect probe below is already cross-platform.)
-    from iai_mcp._ipc import IS_WINDOWS, _port_file_path
+    from iai_mcp._ipc import IS_WINDOWS, _port_file_path, _read_port, _token_file_path
     endpoint = _port_file_path() if IS_WINDOWS else socket_path
     if not endpoint.exists():
         return CheckResult(
@@ -146,6 +157,29 @@ def check_b_socket_fresh() -> CheckResult:
             f"{endpoint} does not exist",
         )
 
+    # A human-readable label for the actual endpoint. On Windows this is the
+    # loopback host:port, NOT the (nonexistent) .daemon.sock path — reporting
+    # the .sock path here was actively misleading.
+    if IS_WINDOWS:
+        port = _read_port()
+        endpoint_label = f"127.0.0.1:{port}" if port is not None else str(endpoint)
+        # The daemon holds its auth token only in memory and mirrors it to
+        # .daemon.token for clients. If that file goes missing while the daemon
+        # runs, the TCP port stays up but EVERY new client fails the auth
+        # handshake — existing long-lived connections survive, so the daemon
+        # looks alive while new sessions / `daemon status` all report "not
+        # running". Call this out specifically; it is not an unreachable daemon.
+        if not _token_file_path().exists():
+            return CheckResult(
+                "(b) socket file fresh",
+                False,
+                f"{endpoint_label} listening but auth token file "
+                f"{_token_file_path()} is missing — new clients cannot "
+                f"authenticate (restart the daemon to regenerate it)",
+            )
+    else:
+        endpoint_label = str(socket_path)
+
     t0 = time.monotonic()
     try:
         err = asyncio.run(_socket_connect_probe(socket_path, timeout=1.0))
@@ -161,12 +195,12 @@ def check_b_socket_fresh() -> CheckResult:
         return CheckResult(
             "(b) socket file fresh",
             False,
-            f"{socket_path} present but unreachable: {err}",
+            f"{endpoint_label} present but unreachable: {err}",
         )
     return CheckResult(
         "(b) socket file fresh",
         True,
-        f"{socket_path} connected in {elapsed_ms} ms",
+        f"{endpoint_label} connected in {elapsed_ms} ms",
     )
 
 
diff --git a/src/iai_mcp/heartbeat_scanner.py b/src/iai_mcp/heartbeat_scanner.py
index aaa2910..45087d6 100644
--- a/src/iai_mcp/heartbeat_scanner.py
+++ b/src/iai_mcp/heartbeat_scanner.py
@@ -35,12 +35,24 @@ class HeartbeatEntry:
 def _is_pid_alive(pid: int) -> bool:
     if pid <= 0:
         return False
+    # NOT os.kill(pid, 0) first: on Windows os.kill rejects signal 0 with
+    # OSError [WinError 87] (invalid parameter) even for a live PID, which made
+    # this scanner (and doctor check (m)) fail against a healthy daemon.
+    # psutil.pid_exists is correct and cross-platform (psutil is a hard dep).
+    try:
+        import psutil
+
+        return psutil.pid_exists(pid)
+    except Exception:
+        pass
     try:
         os.kill(pid, 0)
     except ProcessLookupError:
         return False
     except PermissionError:
         return True
+    except OSError:
+        return False
     return True
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 850ba24..93b995a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -33,6 +33,53 @@
 _TEST_PASSPHRASE = "iai-mcp-test-passphrase-2026-04-30"
 
 
+@pytest.fixture(autouse=True)
+def _home_honors_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Make ``Path.home()`` honor ``$HOME`` on every platform.
+
+    On Windows ``pathlib.Path.home()`` resolves via ``USERPROFILE`` and ignores
+    ``HOME``, but the hermetic test suite (57 fixtures) redirects the operator
+    home by setting ``HOME`` only. Without this, product code that calls
+    ``Path.home()`` at runtime (e.g. ``memory_bank``, the maintenance CLI)
+    reads the real profile dir on Windows and the redirect silently leaks.
+
+    The override reads ``os.environ`` at call time so per-test
+    ``monkeypatch.setenv("HOME", ...)`` overrides are picked up automatically.
+    No-op behaviour on POSIX, where ``Path.home()`` already honors ``HOME``.
+
+    On Windows we additionally mirror any ``HOME`` write into ``USERPROFILE``.
+    In-process code goes through the patched ``Path.home()`` above, but child
+    processes spawned by subprocess-based tests inherit ``os.environ`` and
+    resolve ``Path.home()`` via ``USERPROFILE`` — so the redirect must live in
+    the environment too. Wrapping ``monkeypatch.setenv`` keeps ``USERPROFILE``
+    in sync with whatever ``HOME`` each of the ~57 home-setting fixtures picks.
+    """
+    import pathlib
+
+    _orig_home = pathlib.Path.home.__func__
+
+    def _home(cls):
+        env_home = os.environ.get("HOME")
+        if env_home:
+            return pathlib.Path(env_home)
+        return _orig_home(cls)
+
+    monkeypatch.setattr(pathlib.Path, "home", classmethod(_home))
+
+    if os.name == "nt":
+        _orig_setenv = monkeypatch.setenv
+
+        def _setenv_mirror(name, value, prepend=None):
+            _orig_setenv(name, value, prepend=prepend)
+            if name == "HOME":
+                _orig_setenv("USERPROFILE", value)
+
+        monkeypatch.setenv = _setenv_mirror  # type: ignore[method-assign]
+        # Mirror any HOME already present before the first wrapped setenv call.
+        if os.environ.get("HOME"):
+            _orig_setenv("USERPROFILE", os.environ["HOME"])
+
+
 @pytest.fixture(autouse=True)
 def _hermetic_default_paths(tmp_path_factory, monkeypatch: pytest.MonkeyPatch):
     base = tmp_path_factory.mktemp("iai-hermetic")
diff --git a/tests/test_archive_stuck_backups.py b/tests/test_archive_stuck_backups.py
index 2851549..1c54e47 100644
--- a/tests/test_archive_stuck_backups.py
+++ b/tests/test_archive_stuck_backups.py
@@ -27,8 +27,11 @@ def test_archive_moves_bak_file(tmp_path):
 
     archive_dir = state_dir / "archive"
     assert archive_dir.is_dir()
-    archive_mode = stat.S_IMODE(archive_dir.stat().st_mode)
-    assert archive_mode == 0o700, oct(archive_mode)
+    if os.name != "nt":
+        # POSIX mode bits are meaningless on Windows (chmod only toggles the
+        # read-only bit); S_IMODE reports 0o777 there.
+        archive_mode = stat.S_IMODE(archive_dir.stat().st_mode)
+        assert archive_mode == 0o700, oct(archive_mode)
 
     expected_name = "lifecycle_state.json.HIBERNATION-stuck.bak-20260513T120000Z.bak"
     expected = archive_dir / expected_name
diff --git a/tests/test_bench_worktree_resolution.py b/tests/test_bench_worktree_resolution.py
index cf87bcf..c1f41de 100644
--- a/tests/test_bench_worktree_resolution.py
+++ b/tests/test_bench_worktree_resolution.py
@@ -106,7 +106,7 @@ def _simulate_shim(fake_sys_path: list[str], src_path: str) -> list[str]:
 def test_bench_script_has_shim_before_iai_import(script: str) -> None:
     path = BENCH_DIR / script
     assert path.exists(), f"bench script missing: {path}"
-    tree = ast.parse(path.read_text())
+    tree = ast.parse(path.read_text(encoding="utf-8"))
 
     src_path_idx: int | None = None
     guard_if_idx: int | None = None
@@ -148,7 +148,7 @@ def test_bench_script_has_shim_before_iai_import(script: str) -> None:
 def test_skip_list_bench_scripts_have_no_shim(script: str) -> None:
     path = BENCH_DIR / script
     assert path.exists(), f"skip-list bench script missing: {path}"
-    src = path.read_text()
+    src = path.read_text(encoding="utf-8")
     assert "_SRC_PATH not in sys.path" not in src, (
         f"{script}: skip-list script unexpectedly carries the shim - "
         f"either drop the shim or move the script to BENCH_SCRIPTS_NEEDING_SHIM"
diff --git a/tests/test_bridge_socket_first.py b/tests/test_bridge_socket_first.py
index 8dc1161..89dc6eb 100644
--- a/tests/test_bridge_socket_first.py
+++ b/tests/test_bridge_socket_first.py
@@ -15,6 +15,14 @@
 REPO = Path(__file__).resolve().parent.parent
 WRAPPER = REPO / "mcp-wrapper"
 
+# The mcp-wrapper build path assumes POSIX bash + npm tooling; the CI Windows
+# runner has no Node setup and `subprocess(["npm", ...])` cannot resolve the
+# npm.cmd shim without a shell. Matches test_bridge_no_spawn_path.py.
+pytestmark = pytest.mark.skipif(
+    os.name == "nt",
+    reason="bash + npm tooling assumed POSIX (mcp-wrapper build path)",
+)
+
 
 @pytest.fixture(scope="module")
 def built_wrapper() -> Path:
diff --git a/tests/test_capture_hooks_install_registers_session_recall.py b/tests/test_capture_hooks_install_registers_session_recall.py
index d25365a..7473d0f 100644
--- a/tests/test_capture_hooks_install_registers_session_recall.py
+++ b/tests/test_capture_hooks_install_registers_session_recall.py
@@ -2,11 +2,18 @@
 
 import argparse
 import json
+import os
 import re
 from pathlib import Path
 
 import pytest
 
+# The installer emits PowerShell hooks on Windows and bash hooks on POSIX.
+_HOOK_EXT = "ps1" if os.name == "nt" else "sh"
+_HOOK_INTERP = "powershell" if os.name == "nt" else "bash"
+_RECALL_HOOK = f"iai-mcp-session-recall.{_HOOK_EXT}"
+_CAPTURE_HOOK = f"iai-mcp-session-capture.{_HOOK_EXT}"
+
 
 @pytest.fixture
 def fake_home(tmp_path, monkeypatch):
@@ -30,26 +37,28 @@ def test_install_adds_sessionstart_entry_idempotent(fake_home):
     rc2 = cli_mod.cmd_capture_hooks_install(argparse.Namespace())
     assert rc2 == 0
 
-    data = json.loads(_settings_path(fake_home).read_text())
+    data = json.loads(_settings_path(fake_home).read_text(encoding="utf-8"))
     ss_entries = data.get("hooks", {}).get("SessionStart", [])
     matching = [
         e for e in ss_entries
-        if any("iai-mcp-session-recall.sh" in (h.get("command") or "")
+        if any(_RECALL_HOOK in (h.get("command") or "")
                for h in (e.get("hooks") or []))
     ]
     assert len(matching) == 1, ss_entries
     entry = matching[0]
     assert entry.get("matcher") == "startup|resume|clear|compact", entry
     cmd = entry["hooks"][0]["command"]
-    assert re.search(r"bash .*iai-mcp-session-recall\.sh", cmd), cmd
+    assert re.search(
+        rf"{_HOOK_INTERP}.*iai-mcp-session-recall\.{_HOOK_EXT}", cmd
+    ), cmd
 
     stop_entries = data.get("hooks", {}).get("Stop", [])
     assert any(
-        "iai-mcp-session-capture.sh" in (h.get("command") or "")
+        _CAPTURE_HOOK in (h.get("command") or "")
         for e in stop_entries for h in (e.get("hooks") or [])
     ), stop_entries
 
-    assert (fake_home / ".claude" / "hooks" / "iai-mcp-session-recall.sh").exists()
+    assert (fake_home / ".claude" / "hooks" / _RECALL_HOOK).exists()
 
 
 def test_uninstall_removes_sessionstart_entry_and_script(fake_home):
@@ -57,12 +66,12 @@ def test_uninstall_removes_sessionstart_entry_and_script(fake_home):
 
     cli_mod.cmd_capture_hooks_install(argparse.Namespace())
 
-    recall_dst = fake_home / ".claude" / "hooks" / "iai-mcp-session-recall.sh"
+    recall_dst = fake_home / ".claude" / "hooks" / _RECALL_HOOK
     assert recall_dst.exists(), "install did not copy recall hook"
-    data = json.loads(_settings_path(fake_home).read_text())
+    data = json.loads(_settings_path(fake_home).read_text(encoding="utf-8"))
     ss = data.get("hooks", {}).get("SessionStart", [])
     assert any(
-        "iai-mcp-session-recall.sh" in (h.get("command") or "")
+        _RECALL_HOOK in (h.get("command") or "")
         for e in ss for h in (e.get("hooks") or [])
     ), ss
 
@@ -71,11 +80,11 @@ def test_uninstall_removes_sessionstart_entry_and_script(fake_home):
 
     settings = _settings_path(fake_home)
     if settings.exists():
-        data = json.loads(settings.read_text())
+        data = json.loads(settings.read_text(encoding="utf-8"))
         ss = data.get("hooks", {}).get("SessionStart", [])
         for e in ss:
             for h in (e.get("hooks") or []):
-                assert "iai-mcp-session-recall.sh" not in (h.get("command") or "")
+                assert _RECALL_HOOK not in (h.get("command") or "")
 
     assert not recall_dst.exists()
 
@@ -87,5 +96,5 @@ def test_status_reports_session_recall_alongside_stop(fake_home, capsys):
     capsys.readouterr()
     cli_mod.cmd_capture_hooks_status(argparse.Namespace())
     out = capsys.readouterr().out
-    assert "iai-mcp-session-recall.sh" in out, out
-    assert "iai-mcp-session-capture.sh" in out, out
+    assert _RECALL_HOOK in out, out
+    assert _CAPTURE_HOOK in out, out
diff --git a/tests/test_centrality_approx_bench.py b/tests/test_centrality_approx_bench.py
index f255fac..dbedc40 100644
--- a/tests/test_centrality_approx_bench.py
+++ b/tests/test_centrality_approx_bench.py
@@ -23,7 +23,7 @@
 
 import math
 import time
-from uuid import uuid4
+from uuid import UUID
 
 import numpy as np
 import pytest
@@ -60,7 +60,11 @@ def _scale_free_community_graph(n: int, seed: int) -> MemoryGraph:
     the seed set must recover. Fully deterministic from ``seed``.
     """
     rng = np.random.default_rng(seed)
-    ids = [uuid4() for _ in range(n)]
+    # Derived from ``rng`` (not ``uuid4()``) so node identity is as reproducible
+    # as the edge structure: with an unseeded uuid4, near-tied scores at the
+    # top-K cutoff could swap which node wins the str(id) tie-break between
+    # runs, making the fidelity gate flaky independent of the algorithm.
+    ids = [UUID(bytes=rng.bytes(16)) for _ in range(n)]
     g = MemoryGraph()
     for uid in ids:
         g.add_node(uid, community_id=None, embedding=[0.0] * 8)
diff --git a/tests/test_constitutional_guards.py b/tests/test_constitutional_guards.py
index 2041a49..441bc3b 100644
--- a/tests/test_constitutional_guards.py
+++ b/tests/test_constitutional_guards.py
@@ -47,7 +47,7 @@ def test_no_api_key_in_daemon():
     subprocess with the user's subscription instead."""
     offenders: list[str] = []
     for f in _existing_daemon_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         if "ANTHROPIC_API_KEY" in text:
             offenders.append(f.name)
     assert not offenders, f"violation: ANTHROPIC_API_KEY found in {offenders}"
@@ -74,7 +74,7 @@ def test_no_api_key_anywhere_in_src():
     probe has been removed."""
     offenders: list[str] = []
     for f in _all_iai_mcp_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         if "ANTHROPIC_API_KEY" in text:
             offenders.append(str(f.relative_to(SRC.parent.parent)))
     assert not offenders, (
@@ -97,7 +97,7 @@ def test_no_anthropic_sdk_import_anywhere_in_src():
     import_pattern = re.compile(r"^(?:from anthropic\b|import anthropic\b)", re.MULTILINE)
     offenders: list[tuple[str, list[str]]] = []
     for f in _all_iai_mcp_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         matches = import_pattern.findall(text)
         if matches:
             offenders.append((str(f.relative_to(SRC.parent.parent)), matches))
@@ -113,7 +113,7 @@ def test_no_anthropic_client_construction_anywhere_in_src():
     paths are removed."""
     offenders: list[tuple[str, str]] = []
     for f in _all_iai_mcp_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         if "anthropic.Anthropic(" in text:
             # Surface the surrounding line for diagnostic clarity.
             for line in text.splitlines():
@@ -144,7 +144,7 @@ def test_no_anthropic_messages_sdk_calls_anywhere_in_src():
     )
     offenders: list[tuple[str, str]] = []
     for f in _all_iai_mcp_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         for pat in forbidden_patterns:
             if pat in text:
                 for line in text.splitlines():
@@ -172,7 +172,7 @@ def test_reconsolidation_critic_does_not_modify_literal_surface():
     """
     f = SRC / "reconsolidation_critic.py"
     assert f.exists(), "reconsolidation_critic.py missing"
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     forbidden = (
         re.compile(r"\.literal_surface\s*="),
         re.compile(r"store\.insert\b"),
@@ -207,7 +207,7 @@ def test_no_anthropic_dependency_in_pyproject():
     pyproject.toml; this guard prevents an accidental re-pin."""
     pyproject = SRC.parent.parent / "pyproject.toml"
     assert pyproject.exists(), "pyproject.toml missing"
-    text = pyproject.read_text()
+    text = pyproject.read_text(encoding="utf-8")
     # Block actual dependency lines like `"anthropic>=0.40.0",`, but allow
     # comments mentioning the SDK.
     dep_pattern = re.compile(r'^\s*"anthropic[>=<~!]', re.MULTILINE)
@@ -229,7 +229,7 @@ def test_no_lockf_anywhere():
     modules -- mixing the two is also a bug."""
     offenders: list[str] = []
     for f in SRC.glob("*.py"):
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         if "fcntl.lockf" in text:
             offenders.append(f.name)
     assert not offenders, f"fcntl.lockf forbidden, found in {offenders}"
@@ -246,7 +246,7 @@ def test_no_literal_surface_mutation_in_daemon():
     pattern = re.compile(r"\.literal_surface\s*=")
     offenders: list[tuple[str, list[str]]] = []
     for f in _existing_daemon_files():
-        text = f.read_text()
+        text = f.read_text(encoding="utf-8")
         matches = pattern.findall(text)
         if matches:
             offenders.append((f.name, matches))
@@ -263,7 +263,7 @@ def test_no_hardcoded_clock_time_in_quiet_window():
     f = SRC / "quiet_window.py"
     if not f.exists():
         return  # module not yet created
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     # Look for common patterns that would indicate clock-based decisions.
     forbidden = [
         r"\b22:00\b",
@@ -368,7 +368,7 @@ def test_no_cache_control_in_session_assembler():
     """
     f = SRC / "session.py"
     assert f.exists(), "session.py missing"
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     # Comments that mention "cache_control" are fine (they document the
     # pitfall). We only guard against actual code references like setattr/
     # cache_control=... — scan for the pattern with an equals sign.
@@ -390,7 +390,7 @@ def test_no_api_key_in_response_decorator():
     """response_decorator.py stays local-only."""
     f = SRC / "response_decorator.py"
     assert f.exists(), "response_decorator.py missing"
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     lower = text.lower()
     assert "anthropic" not in lower, (
         "violation: response_decorator references 'anthropic'"
@@ -414,7 +414,7 @@ def test_identity_audit_has_no_lock_import():
     f = SRC / "identity_audit.py"
     if not f.exists():
         return
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     # No import of iai_mcp.concurrency, no `ProcessLock` symbol reference.
     assert "iai_mcp.concurrency" not in text, (
         "violation: identity_audit.py imports iai_mcp.concurrency"
@@ -438,7 +438,7 @@ def test_no_api_key_in_hippea_cascade():
     f = SRC / "hippea_cascade.py"
     if not f.exists():
         return  # module not yet created
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     assert "ANTHROPIC_API_KEY" not in text, (
         "violation: ANTHROPIC_API_KEY in hippea_cascade.py"
     )
@@ -460,7 +460,7 @@ def test_hippea_cascade_is_read_only_against_store():
     f = SRC / "hippea_cascade.py"
     if not f.exists():
         return
-    text = f.read_text()
+    text = f.read_text(encoding="utf-8")
     forbidden_calls = [
         "store.insert(",
         "store.append_provenance(",
diff --git a/tests/test_contradiction_longitudinal_route_columns.py b/tests/test_contradiction_longitudinal_route_columns.py
index 17d324d..ee78832 100644
--- a/tests/test_contradiction_longitudinal_route_columns.py
+++ b/tests/test_contradiction_longitudinal_route_columns.py
@@ -137,7 +137,7 @@ def test_csv_writer_emits_route_and_cue_hash_columns(tmp_path: Path) -> None:
     csv_path = tmp_path / f"contradiction_longitudinal_{run_id}.csv"
     assert csv_path.exists(), f"CSV not written at {csv_path}"
 
-    with csv_path.open(newline="") as fh:
+    with csv_path.open(newline="", encoding="utf-8") as fh:
         reader = csv.DictReader(fh)
         fieldnames = reader.fieldnames or []
         assert "route" in fieldnames, (
diff --git a/tests/test_cpu_features.py b/tests/test_cpu_features.py
index 2ab5cf7..4f066a0 100644
--- a/tests/test_cpu_features.py
+++ b/tests/test_cpu_features.py
@@ -36,12 +36,12 @@ def test_linux_proc_cpuinfo_with_avx2_returns_true(
     monkeypatch.setattr(cf.platform, "machine", lambda: "x86_64")
 
     def _fake_read_text(self, *a, **kw):
-        if str(self) == "/proc/cpuinfo":
+        if self.as_posix() == "/proc/cpuinfo":
             return _LINUX_CPUINFO_WITH_AVX2
         raise FileNotFoundError(str(self))
 
     def _fake_exists(self):
-        return str(self) == "/proc/cpuinfo"
+        return self.as_posix() == "/proc/cpuinfo"
 
     monkeypatch.setattr(Path, "read_text", _fake_read_text)
     monkeypatch.setattr(Path, "exists", _fake_exists)
@@ -61,12 +61,12 @@ def test_linux_proc_cpuinfo_without_avx2_returns_false(
     monkeypatch.setattr(cf.platform, "machine", lambda: "x86_64")
 
     def _fake_read_text(self, *a, **kw):
-        if str(self) == "/proc/cpuinfo":
+        if self.as_posix() == "/proc/cpuinfo":
             return _LINUX_CPUINFO_WITHOUT_AVX2
         raise FileNotFoundError(str(self))
 
     def _fake_exists(self):
-        return str(self) == "/proc/cpuinfo"
+        return self.as_posix() == "/proc/cpuinfo"
 
     monkeypatch.setattr(Path, "read_text", _fake_read_text)
     monkeypatch.setattr(Path, "exists", _fake_exists)
diff --git a/tests/test_crypto_file_backend.py b/tests/test_crypto_file_backend.py
index 91f2b97..3551d56 100644
--- a/tests/test_crypto_file_backend.py
+++ b/tests/test_crypto_file_backend.py
@@ -25,6 +25,11 @@ def test_try_file_get_returns_bytes_on_valid_0o600_file(tmp_path: Path) -> None:
     assert len(got) == 32
 
 
+@pytest.mark.skipif(
+    not hasattr(os, "geteuid"),
+    reason="POSIX permission-bit ownership check is skipped on Windows "
+    "(crypto._try_file_get guards it behind hasattr(os, 'geteuid'))",
+)
 def test_try_file_get_rejects_world_or_group_bits(tmp_path: Path) -> None:
     from iai_mcp.crypto import CryptoKey, CryptoKeyError
 
@@ -51,6 +56,10 @@ def test_try_file_get_rejects_wrong_length(tmp_path: Path) -> None:
     assert "wrong length" in str(exc_info.value).lower()
 
 
+@pytest.mark.skipif(
+    not hasattr(os, "geteuid"),
+    reason="uid ownership check is POSIX-only; os.geteuid() is absent on Windows",
+)
 def test_try_file_get_rejects_foreign_uid(tmp_path: Path, monkeypatch) -> None:
     from iai_mcp.crypto import CryptoKey, CryptoKeyError
 
diff --git a/tests/test_daemon_kill_mid_consolidation_lossless.py b/tests/test_daemon_kill_mid_consolidation_lossless.py
index 8d3842b..be1e641 100644
--- a/tests/test_daemon_kill_mid_consolidation_lossless.py
+++ b/tests/test_daemon_kill_mid_consolidation_lossless.py
@@ -10,6 +10,7 @@
 from pathlib import Path
 from uuid import UUID, uuid4
 
+from iai_mcp.hippo import HippoLockHeldError
 from iai_mcp.store import MemoryStore
 from iai_mcp.types import EMBED_DIM, MemoryRecord
 
@@ -170,13 +171,29 @@ def test_sigkill_mid_write_leaves_store_lossless():
                 raise AssertionError("writer child never signalled readiness")
             time.sleep(0.01)
 
-        os.kill(child.pid, signal.SIGKILL)
+        # Windows has no SIGKILL; os.kill(pid, SIGTERM) maps to TerminateProcess,
+        # which is the same abrupt no-cleanup kill this test needs to prove the
+        # store stays lossless after a hard kill mid-write.
+        os.kill(child.pid, getattr(signal, "SIGKILL", signal.SIGTERM))
         child.wait(timeout=30)
         child = None
 
         churn_vec = [0.0] * K_SEEDS + [1.0] + [0.0] * (EMBED_DIM - K_SEEDS - 1)
 
-        reopened = MemoryStore(path=tmp_root)
+        # Windows TerminateProcess (os.kill SIGTERM) releases the killed child's
+        # byte-range lock on the hippo .lock asynchronously during process
+        # teardown, so the reopen can briefly race the lock release. Poll until
+        # the exclusive lock is free. First attempt succeeds on POSIX.
+        reopened = None
+        for _ in range(50):
+            try:
+                reopened = MemoryStore(path=tmp_root)
+                break
+            except HippoLockHeldError:
+                time.sleep(0.1)
+        assert reopened is not None, (
+            "hippo lock not released within 5s after killing the writer child"
+        )
         try:
 
             for i, sid in enumerate(seed_ids):
diff --git a/tests/test_daemon_launchd_reenable.py b/tests/test_daemon_launchd_reenable.py
index 5cd5a70..f824650 100644
--- a/tests/test_daemon_launchd_reenable.py
+++ b/tests/test_daemon_launchd_reenable.py
@@ -105,6 +105,11 @@ def test_rendered_plist_watchdog_values_match_code_defaults() -> None:
     assert daemon_mod.WATCHDOG_COLD_START_GRACE_SEC == 600.0
 
 
+@pytest.mark.skipif(
+    not hasattr(__import__("os"), "getuid"),
+    reason="launchd gui/<uid> domain target is macOS-only; os.getuid() is "
+    "absent on Windows",
+)
 def test_reenable_emits_bootout_bootstrap_kickstart_in_order(
     fake_state_dir: Path,
     captured_launchctl: list[list[str]],
diff --git a/tests/test_ipc_token_reassert.py b/tests/test_ipc_token_reassert.py
new file mode 100644
index 0000000..dccdc94
--- /dev/null
+++ b/tests/test_ipc_token_reassert.py
@@ -0,0 +1,81 @@
+"""Unit tests for the Windows auth-token self-heal (iai_mcp._ipc).
+
+Regression coverage for the incident where .daemon.token vanished out-of-band
+while the daemon kept running: the TCP port stayed up but every NEW client
+failed the auth handshake (looked like "daemon not running") until a manual
+restart. reassert_token_if_missing() lets the daemon restore the file from its
+in-memory token on the next tick.
+"""
+from __future__ import annotations
+
+import pytest
+
+from iai_mcp import _ipc
+
+
+@pytest.fixture
+def token_env(tmp_path, monkeypatch):
+    """Point the token file at a hermetic tmp path and neutralize the Windows
+    ACL call (icacls) so the logic is testable on any platform."""
+    endpoint = tmp_path / "daemon"
+    monkeypatch.setenv("IAI_DAEMON_SOCKET_PATH", str(endpoint))
+    monkeypatch.setattr(_ipc, "_restrict_token_file", lambda _p: None)
+    return _ipc._token_file_path()
+
+
+def test_reassert_recreates_missing_token(token_env, monkeypatch):
+    monkeypatch.setattr(_ipc, "IS_WINDOWS", True)
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", "deadbeef" * 8)
+    assert not token_env.exists()
+
+    assert _ipc.reassert_token_if_missing() is True
+    assert token_env.exists()
+    assert token_env.read_text(encoding="utf-8").strip() == "deadbeef" * 8
+
+
+def test_reassert_is_noop_when_token_present(token_env, monkeypatch):
+    monkeypatch.setattr(_ipc, "IS_WINDOWS", True)
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", "abc123" * 8)
+    token_env.write_text("existing-token", encoding="utf-8")
+
+    assert _ipc.reassert_token_if_missing() is False
+    # Must not clobber a token already on disk.
+    assert token_env.read_text(encoding="utf-8") == "existing-token"
+
+
+def test_reassert_noop_without_in_memory_token(token_env, monkeypatch):
+    # A client process never generated a token: it must NOT create one.
+    monkeypatch.setattr(_ipc, "IS_WINDOWS", True)
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", None)
+
+    assert _ipc.reassert_token_if_missing() is False
+    assert not token_env.exists()
+
+
+def test_reassert_noop_on_posix(token_env, monkeypatch):
+    monkeypatch.setattr(_ipc, "IS_WINDOWS", False)
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", "ffff" * 16)
+
+    assert _ipc.reassert_token_if_missing() is False
+    assert not token_env.exists()
+
+
+def test_generate_token_populates_in_memory_copy(token_env, monkeypatch):
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", None)
+    token = _ipc._generate_token()
+    assert _ipc._CURRENT_TOKEN == token
+    assert token_env.read_text(encoding="utf-8").strip() == token
+
+
+def test_remove_token_clears_in_memory_copy(token_env, monkeypatch):
+    monkeypatch.setattr(_ipc, "IS_WINDOWS", True)
+    monkeypatch.setattr(_ipc, "_CURRENT_TOKEN", "cafe" * 16)
+    token_env.write_text("cafe" * 16, encoding="utf-8")
+
+    _ipc._remove_token_file()
+
+    # In-memory copy cleared so a later reassert cannot resurrect a token the
+    # daemon deliberately tore down.
+    assert _ipc._CURRENT_TOKEN is None
+    assert _ipc.reassert_token_if_missing() is False
+    assert not token_env.exists()