fix(hooks): spool captures for a detached drainer where the resident worker cannot run (#659)

.github/workflows/ci.yml

@@ -492,6 +492,10 @@ jobs:
       # test_semantic_fallback_169.py and test_doctor.py cover #634's
       # sqlite-vec install-and-backfill fix; test_wiki_maintenance.py and
       # test_wiki_backlog_pass.py cover #636's consolidate-on-SQLite fix.
+      # The three capture files join per issue #659: here, and only here,
+      # capture_peer.is_supported() is false natively, msvcrt takes the real
+      # drainer lock and the detached child runs with the real Windows
+      # creationflags. source: ADR-1094
       - name: Run portability + backend tests
         env:
           HF_HUB_OFFLINE: "1"
@@ -510,6 +514,9 @@ jobs:
           tests_py/test_doctor.py
           tests_py/handlers/consolidation/test_wiki_maintenance.py
           tests_py/handlers/consolidation/test_wiki_backlog_pass.py
+          tests_py/hooks/test_capture_unsupported_platform.py
+          tests_py/hooks/test_capture_drain.py
+          tests_py/infrastructure/test_capture_spool.py
 
   # source: ADR-0802
 

docs/adr/ADR-1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md

@@ -0,0 +1,105 @@
+<!-- Generated by Cortex wiki_reindex from wiki/adr/cortex/1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md; ADR-1094. Read-only mirror: edit the canonical wiki page. -->
+
+---
+created: 2026-10-07T00:00:00Z
+kind: adr
+number: 1094
+status: accepted
+tags: [capture, windows, hooks, issue-659]
+title: Spool captures for a detached drainer where the resident worker cannot run
+---
+# ADR-1094: Spool captures for a detached drainer where the resident worker cannot run
+
+## Status
+
+accepted
+
+## Context
+
+Issue #659: since 17370a61 every PostToolUse capture goes through a resident
+worker that needs `AF_UNIX`, `fcntl`, `geteuid` and `Popen(pass_fds)`. CPython on
+Windows has none of them, `capture_peer.supported()` raised, and the hook
+reported `capture_skipped` for every event: 1,730 attempts, zero memories, with
+no visible signal (reporter's `telemetry.jsonl`, Windows 11, 4.23.1).
+
+The first fix (PR #663 at baa8ece9) stored in the hook process on such platforms.
+Measured on 2026-10-07 (macOS, isolated SQLite store, warm): `handlers/remember.py`
+imports `sentence_transformers` in 3.61 s (`torch` 0.72 s, weights 0.10 s, one
+encode 0.01 s); the whole in-process capture took 4.4 s warm and 10.5 s cold. The
+hook has `timeout: 10` (`.claude-plugin/plugin.json`), and the host aborts a hook
+that exceeds it ("PostToolUse hook timed out (per-hook abort)", Claude Code
+2.1.292). The resident worker (ADR-0486, ADR-0508) exists precisely to pay that
+import once; the in-process path put it back on the synchronous path of every
+event, so the capture would be killed at the limit instead of refused.
+
+## Decision
+
+Where `capture_peer.is_supported()` is false the hook does not load the handler.
+It validates the payload (`validate_payload`), writes it unchanged and atomically
+to `<CLAUDE_DIR>/.capture-worker/spool/` (`infrastructure/capture_spool.py`) and
+starts `python -m mcp_server.hooks.capture_drain` without waiting. POSIX is
+unchanged (the resident worker).
+
+- The spawn options are a pure function, `capture_dispatch.popen_options(platform, stream)`:
+  Windows `creationflags = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP`,
+  `close_fds`, stdin to `DEVNULL`, stdout and stderr to `stream`, the drainer's one
+  log `drain.log` (owner-only mode, size-rotated at each spawn by
+  `open_rotating_log`, ADR-0655; the drainer's logging goes to its stderr, so a
+  crash before logging exists and library progress bars land in the same bounded
+  file), the launcher's environment; no `pass_fds` and no `start_new_session`
+  (CPython ignores it on Windows). Other platforms use `start_new_session`.
+- The drainer takes a non-blocking kernel lock (`msvcrt.locking` `LK_NBLCK` /
+  `flock` `LOCK_NB`; the pattern of the Codex SessionEnd queue, ADR-1084). A loser
+  exits at once without importing the handler. The winner stores every pending
+  file through the same `capture_store.store` the worker awaits, on one event loop
+  (one model load per burst), then releases the lock and scans once more. A file
+  written between its last scan and the release has a drainer that lost the lock
+  and exited; the rescan stores it.
+- Delivery is at-least-once: a file is deleted after `store` returned; a replay after
+  a crash is absorbed by the write gate. A refused, unreadable or failing file is
+  renamed `*.rejected` and reported through `report_failure("capture_skipped")`;
+  nothing is swallowed. If the rename itself fails the file is deleted after the
+  report (a payload already refused is not replayed by every later drainer); if
+  that fails too it is skipped for the run.
+- Bounded and observable. Each store is cancelled after `STORE_SECONDS` (120 s,
+  about ten times the measured cold capture) and the file refused like any
+  failure, so a hung handler cannot hold the lock for good. A hang that does not
+  yield to cancellation (a blocking call) is covered by the hook: when the oldest
+  pending file is older than `STALL_SECONDS` (600 s) it reports
+  `capture_spool_stalled` through telemetry on every capture, and still queues the
+  new payload. The drainer removes `.partial` files older than an hour (a hook that
+  died between write and rename) and `.rejected` files older than a week.
+- Order is not a contract. Captures are independent writes to a write-gated store, so
+  the drainer processes files in name order only for determinism; the name's clock
+  prefix can tie or invert (Windows clock resolution is about 15.6 ms). Age is
+  therefore taken from the files' own `mtime` (the oldest of all pending files),
+  never from the name.
+- The directory is private under the configured Cortex root (mode 0700 where the
+  platform has modes; on Windows it inherits the user-profile ACL).
+
+## Rejected
+
+- The PR as first written (store in the hook): 4.4 to 10.5 s against a 10 s budget.
+- One process per event with no spool: N imports of `torch`.
+- Algorithmic embedding re-embedded at `consolidate`: `sqlite_store_search.py`
+  isolates embedding spaces and the write gate is blind to neural rows.
+- `async: true` on the hook: every platform, still one `torch` import per event.
+- Named pipe with a current-user DACL: a second transport, lease and spawn stack.
+- Spool on every platform: it would change the POSIX capture path (ADR-0486,
+  ADR-0508), outside this issue.
+
+## Consequences
+
+Positive: the hook returns in milliseconds on Windows (0.06 s measured for the hook
+body on macOS with the Windows branch simulated); one model load serves a burst;
+`ci.yml`'s Windows leg runs the capture test files, so `is_supported()` false
+natively, the real `msvcrt` lock and the real detached spawn are exercised.
+
+Negative: a capture is persisted after the hook returns, not before; a drainer
+killed by the host's process-tree cleanup leaves the file for the next event's
+drainer. `CREATE_BREAKAWAY_FROM_JOB` is not requested: it fails when the host's job
+object forbids breakaway. Whether the host destroys the tree on abort is unobserved,
+since the hook no longer approaches its limit. The reporter's check, a
+`telemetry.jsonl` with `remember` samples and no `capture_skipped`, is the
+signal that closes #659's capture part. `check_setup` answering `ready: true` while
+capture is broken is a separate defect (#660).

docs/provenance/capture-worker-design.md

@@ -44,15 +44,33 @@ lose accepted work. There is no automatic replay after uncertain delivery, which
 could duplicate a write. Durable delivery is outside this change.
 
 Absent/refused socket triggers one launch attempt. Spawn, transport, malformed
-frame, unsupported platform and admission-deadline failures produce a diagnostic and a
-`capture_skipped` telemetry sample; inference never falls back into the hook.
+frame and admission-deadline failures produce a diagnostic and a
+`capture_skipped` telemetry sample; a failed worker never falls back into the hook.
 Dispatch, receive and processing errors report their measured monotonic duration.
 Listener/cleanup failures use a distinct `capture_worker_lifecycle` sample with
 their own measured duration. Neither event is recorded as `remember`, so these
 diagnostics do not enter the handler's `remember` latency series.
-Worker handler failure has the same observable error boundary. Windows is
-explicitly unsupported because this implementation requires Unix credentials,
-Unix socket paths and `flock`; no TCP or abstract-socket fallback exists.
+Worker handler failure has the same observable error boundary. A platform the worker
+cannot run on (Windows: CPython has no `AF_UNIX`, no `fcntl`, no `geteuid`, and
+`Popen` has no `pass_fds`) is decided by the capability test
+`capture_peer.is_supported()`, before any I/O, not by catching an error. There the
+hook never loads the handler: importing `sentence_transformers` alone measured
+3.61 s on 2026-10-07 (macOS, isolated SQLite store, `torch` 0.72 s, warm), against a
+hook `timeout` of 10 s. It validates the payload, writes it atomically to
+`.capture-worker/spool/<time>-<pid>-<id>.json` and starts `capture_drain` detached
+(`capture_dispatch.popen_options`: `DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP` on
+Windows, `start_new_session` elsewhere, no inherited descriptors; stderr to
+`drain.log`), without waiting.
+The drainer takes a non-blocking kernel lock (`msvcrt.locking` / `flock`); a drainer
+that loses it exits, the winner stores every pending file through the same
+`capture_store.store` the worker awaits (one model load per burst), deletes each file
+only after it was stored, releases the lock and scans once more, so a file written
+during the release is never stranded. Delivery is at-least-once; a replay is absorbed
+by the write gate. A refused or failing file is renamed `*.rejected` and reported as
+`capture_skipped`. No TCP, abstract-socket or named-pipe transport is built: the spool
+needs none, and a named pipe with a current-user DACL would add a second transport,
+lease and spawn implementation (issue #659, ADR-1094). `ci.yml`'s Windows leg runs the
+capture test files, so the native Windows behaviour is exercised there.
 
 ## Protocol and trust boundary
 

mcp_server/hooks/capture_dispatch.py

@@ -9,9 +9,12 @@
 import sys
 import time
 from pathlib import Path
+from typing import IO, Any
 
 from mcp_server.hooks.capture_worker_policy import limits, validate_payload
+from mcp_server.infrastructure import capture_spool
 from mcp_server.infrastructure.capture_client import deliver
+from mcp_server.infrastructure.capture_peer import is_supported
 from mcp_server.infrastructure.capture_transport import CaptureTransportError
 from mcp_server.infrastructure.config import CLAUDE_DIR
 
@@ -28,6 +31,63 @@ def report_failure(
     telemetry.record(operation, latency_ms=elapsed * 1000.0, ok=False)
 
 
+# source: https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags
+# (DETACHED_PROCESS: no console of the parent; CREATE_NEW_PROCESS_GROUP: not
+# signalled with the parent's group). The ``subprocess.*`` names of these flags
+# exist only on Windows, so the documented values are spelled out to keep
+# ``popen_options`` a pure function that macOS and Linux can test.
+_WINDOWS_DETACHED = 0x00000008 | 0x00000200
+
+
+def popen_options(platform: str, stream: IO[Any]) -> dict[str, Any]:
+    """Options that start the drainer fully detached from the hook process.
+
+    postcondition: stdin is closed, stdout and stderr go to ``stream``
+    (the drainer's one log, so a crash before its logging exists is not lost), and
+    the child does not die with the hook. Windows has no ``start_new_session``
+    (``subprocess`` ignores it there) and no ``pass_fds``; it detaches through
+    ``creationflags``."""
+    options: dict[str, Any] = {
+        "stdin": subprocess.DEVNULL,
+        "stdout": stream,
+        "stderr": stream,
+        "close_fds": True,
+        "cwd": Path(__file__).resolve().parents[2],
+        "env": dict(os.environ),
+    }
+    if platform == "win32":
+        options["creationflags"] = _WINDOWS_DETACHED
+    else:
+        options["start_new_session"] = True
+    return options
+
+
+def spawn_drainer(stream: IO[Any]) -> subprocess.Popen[Any]:
+    """Start ``capture_drain`` and do not wait for it. source: ADR-1094"""
+    command = [sys.executable, "-m", "mcp_server.hooks.capture_drain"]
+    return subprocess.Popen(command, **popen_options(sys.platform, stream))
+
+
+def _spool(payload: dict[str, object]) -> None:
+    """Platform without the resident worker: durable file, then a detached drainer.
+
+    The file is written before the drainer starts, so a spawn failure loses
+    nothing: the next capture's drainer stores it."""
+    spool = capture_spool.spool_directory(CLAUDE_DIR)
+    age = capture_spool.oldest_pending_age(spool, time.time())
+    if age is not None and age > capture_spool.STALL_SECONDS:
+        # a drainer that is hung or cannot start leaves a growing backlog and, until
+        # now, said nothing: surface it through telemetry on every capture (ADR-1094)
+        report_failure(
+            f"capture spool stalled: oldest pending file is {age:.0f}s old",
+            age,
+            operation="capture_spool_stalled",
+        )
+    capture_spool.write(spool, payload)
+    with capture_spool.open_log(spool) as stream:
+        spawn_drainer(stream)
+
+
 def _spawn(listener: socket.socket, lease: int) -> None:
     command = [
         sys.executable,
@@ -55,7 +115,10 @@ def dispatch(payload: dict[str, object]) -> bool:
     started = time.monotonic()
     try:
         validate_payload(payload)
-        deliver(CLAUDE_DIR, payload, limits(), _spawn)
+        if is_supported():
+            deliver(CLAUDE_DIR, payload, limits(), _spawn)
+        else:
+            _spool(payload)
     except (OSError, ValueError, CaptureTransportError) as exc:
         elapsed = time.monotonic() - started
         report_failure(

mcp_server/hooks/capture_drain.py

@@ -0,0 +1,107 @@
+"""Detached drainer of the capture spool: the store path where no worker can run.
+
+``capture_dispatch`` writes each admitted payload to the spool and starts this
+process without waiting. Whichever drainer wins the non-blocking lock stores
+every pending file through ``capture_store.store`` (the handler, imported and
+loaded once for the whole burst); the others exit at once. After releasing the
+lock the winner scans once more, so a file written between its last scan and the
+release is never stranded (its own drainer lost the lock and exited).
+
+Delivery is at-least-once: a file is deleted only after ``store`` returned, and a
+replay after a crash is absorbed by the write gate's duplicate detection.
+A refused or failing payload is renamed ``*.rejected`` (kept for inspection) and
+reported as ``capture_skipped``; nothing is swallowed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from pathlib import Path
+
+from mcp_server.hooks._store_lifecycle import close_shared_store_on_exit
+from mcp_server.hooks.capture_dispatch import report_failure
+from mcp_server.hooks.capture_store import store
+from mcp_server.infrastructure import capture_spool
+from mcp_server.infrastructure.config import CLAUDE_DIR
+
+
+def _refuse(path: Path, exc: BaseException, elapsed: float) -> bool:
+    """Retire a file that could not be stored and report it once.
+
+    postcondition: returns True when the file no longer matches ``pending``.
+    Normally it is renamed ``*.rejected``. If even that fails (a broken
+    filesystem), the file is deleted so it cannot be replayed by every later
+    drainer; if that fails too it stays and the caller skips it for this run.
+    Every outcome is part of the one ``capture_skipped`` report."""
+    detail = f"{type(exc).__name__}: {exc}"
+    resolved = True
+    try:
+        fate = capture_spool.reject(path).name
+    except OSError as rename_error:
+        try:
+            path.unlink(missing_ok=True)
+            fate = f"deleted, rename failed: {rename_error}"
+        except OSError as delete_error:
+            resolved = False
+            fate = f"left in place, rename and delete failed: {delete_error}"
+    report_failure(f"capture rejected after {elapsed:.3f}s ({fate}): {detail}", elapsed)
+    return resolved
+
+
+async def drain_pending(spool: Path, skip: set[Path]) -> int:
+    """precondition: the caller holds ``capture_spool.drain_lock``.
+    postcondition: every pending file not in ``skip`` is stored and deleted, or
+    retired by ``_refuse`` (added to ``skip`` when it could not be); a store that
+    exceeds ``capture_spool.STORE_SECONDS`` is cancelled and refused like any
+    failure. Returns the number stored."""
+    stored = 0
+    for path in capture_spool.pending(spool):
+        if path in skip:
+            continue
+        started = time.monotonic()
+        try:
+            await asyncio.wait_for(
+                store(capture_spool.read(path)), capture_spool.STORE_SECONDS
+            )
+        except Exception as exc:  # noqa: BLE001 — per-file boundary: a refused, failing or hung payload must not strand the files behind it; it is reported and retired
+            if not _refuse(path, exc, time.monotonic() - started):
+                skip.add(path)
+            continue
+        path.unlink()
+        stored += 1
+    return stored
+
+
+def drain(spool: Path) -> None:
+    """Drain until a scan taken after releasing the lock finds nothing new.
+
+    invariant: each pass stores, retires or skips every file it sees, so the
+    loop ends; a lost lock returns immediately."""
+    skip: set[Path] = set()
+    while True:
+        with capture_spool.drain_lock(spool) as held:
+            if not held:
+                return
+            asyncio.run(drain_pending(spool, skip))
+            capture_spool.sweep(spool, time.time())
+        if not [p for p in capture_spool.pending(spool) if p not in skip]:
+            return
+
+
+def main() -> None:
+    spool = capture_spool.spool_directory(CLAUDE_DIR)
+    # the single log is this process's stderr (see ``capture_spool.open_log``)
+    logging.basicConfig(
+        level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
+    )
+    with close_shared_store_on_exit():
+        drain(spool)
+
+
+if __name__ == "__main__":
+    from mcp_server.hooks.wiring import wire_composition_root  # noqa: PLC0415 — source: issue #560
+
+    wire_composition_root()
+    main()