diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a3ffe59cf..af82a5fb9 100644
--- a/.github/workflows/ci.yml
+++ b/.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
 
diff --git a/docs/adr/ADR-1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md b/docs/adr/ADR-1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md
new file mode 100644
index 000000000..a1eb5de97
--- /dev/null
+++ b/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).
diff --git a/docs/provenance/capture-worker-design.md b/docs/provenance/capture-worker-design.md
index 5082ab3a1..6447d9f2b 100644
--- a/docs/provenance/capture-worker-design.md
+++ b/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
 
diff --git a/mcp_server/hooks/capture_dispatch.py b/mcp_server/hooks/capture_dispatch.py
index 324f5d274..584b51beb 100644
--- a/mcp_server/hooks/capture_dispatch.py
+++ b/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(
diff --git a/mcp_server/hooks/capture_drain.py b/mcp_server/hooks/capture_drain.py
new file mode 100644
index 000000000..205de09c8
--- /dev/null
+++ b/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()
diff --git a/mcp_server/hooks/capture_store.py b/mcp_server/hooks/capture_store.py
new file mode 100644
index 000000000..ac2164ccb
--- /dev/null
+++ b/mcp_server/hooks/capture_store.py
@@ -0,0 +1,48 @@
+"""The one store path for an admitted capture payload, wherever it runs.
+
+The resident worker awaits ``store`` (through ``remember``) on its persistent
+loop. A platform the worker does not support (``capture_peer.is_supported()``
+is false, e.g. Windows: no ``AF_UNIX``, no ``fcntl``, no ``geteuid``) awaits the
+same function from the spool drainer (``capture_drain``). Same validation, same
+handler, same payload: only the hosting process differs.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from mcp_server.hooks.capture_worker_policy import validate_payload
+
+logger = logging.getLogger(__name__)
+
+
+async def store(payload: dict[str, object]) -> dict[str, object]:
+    """precondition: payload is the hook's unchanged ``remember`` payload.
+    postcondition: payload was validated and handed to the remember handler;
+    its result is returned. A refused payload or missing dependency raises."""
+    validate_payload(payload)
+    from mcp_server.hooks.post_tool_capture import _load_remember  # noqa: PLC0415 — reuse the existing composition loader only when a store is needed
+
+    try:
+        _, handler = _load_remember()
+    except SystemExit as exc:
+        raise RuntimeError(
+            "capture dependencies unavailable; verify installation"
+        ) from exc
+    result = await handler(payload)
+    if result.get("stored"):
+        logger.info(
+            "captured %s memory_id=%s", payload["origin_tool"], result.get("memory_id")
+        )
+    else:
+        logger.info(
+            "gated %s: %s",
+            payload["origin_tool"],
+            result.get("reason", "below_threshold"),
+        )
+    return result
+
+
+async def remember(payload: dict[str, object]) -> None:
+    """The resident worker's callback: store and discard the handler result."""
+    await store(payload)
diff --git a/mcp_server/hooks/capture_worker.py b/mcp_server/hooks/capture_worker.py
index c5b46b1a2..8ff5b8b61 100644
--- a/mcp_server/hooks/capture_worker.py
+++ b/mcp_server/hooks/capture_worker.py
@@ -13,7 +13,8 @@
 import time
 from pathlib import Path
 
-from mcp_server.hooks.capture_worker_policy import limits, validate_payload
+from mcp_server.hooks.capture_store import remember
+from mcp_server.hooks.capture_worker_policy import limits
 from mcp_server.hooks.capture_worker_logging import configure
 from mcp_server.hooks._store_lifecycle import close_shared_store_on_exit
 from mcp_server.infrastructure.capture_server import CaptureServer, ServerPolicy
@@ -22,29 +23,6 @@
 logger = logging.getLogger(__name__)
 
 
-async def remember(payload: dict[str, object]) -> None:
-    validate_payload(payload)
-    from mcp_server.hooks.post_tool_capture import _load_remember  # noqa: PLC0415 — reuse the existing composition loader only inside the worker
-
-    try:
-        _, handler = _load_remember()
-    except SystemExit as exc:
-        raise RuntimeError(
-            "resident capture dependencies unavailable; verify installation"
-        ) from exc
-    result = await handler(payload)
-    if result.get("stored"):
-        logger.info(
-            "captured %s memory_id=%s", payload["origin_tool"], result.get("memory_id")
-        )
-    else:
-        logger.info(
-            "gated %s: %s",
-            payload["origin_tool"],
-            result.get("reason", "below_threshold"),
-        )
-
-
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--listener-fd", type=int, required=True)
diff --git a/mcp_server/hooks/post_tool_capture.py b/mcp_server/hooks/post_tool_capture.py
index 36b68ecf2..5ff3ae8c4 100644
--- a/mcp_server/hooks/post_tool_capture.py
+++ b/mcp_server/hooks/post_tool_capture.py
@@ -269,7 +269,7 @@ def _store_memory(tool_name: str, content: str, tags: list[str], cwd: str) -> No
         "force": False,
     }
     if dispatch(payload):
-        _log(f"queued {tool_name} for resident capture (persistence pending)")
+        _log(f"queued {tool_name} for capture (persistence pending)")
 
 
 # ── Periodic cascade advancement ──────────────────────────────────────
diff --git a/mcp_server/infrastructure/capture_peer.py b/mcp_server/infrastructure/capture_peer.py
index ded0d837d..928b0cb2d 100644
--- a/mcp_server/infrastructure/capture_peer.py
+++ b/mcp_server/infrastructure/capture_peer.py
@@ -13,8 +13,13 @@
 import sys
 
 
+def is_supported() -> bool:
+    """Capability test: AF_UNIX and kernel peer identity exist on this platform."""
+    return sys.platform in {"linux", "darwin"} and hasattr(socket, "AF_UNIX")
+
+
 def supported() -> None:
-    if sys.platform not in {"linux", "darwin"} or not hasattr(socket, "AF_UNIX"):
+    if not is_supported():
         raise OSError(
             "capture worker requires Linux/macOS Unix sockets; no TCP fallback"
         )
diff --git a/mcp_server/infrastructure/capture_spool.py b/mcp_server/infrastructure/capture_spool.py
new file mode 100644
index 000000000..418f9ed26
--- /dev/null
+++ b/mcp_server/infrastructure/capture_spool.py
@@ -0,0 +1,186 @@
+"""Durable file spool for capture payloads on platforms without the resident worker.
+
+The hook only writes a file here and starts a detached drainer; the drainer
+(``mcp_server.hooks.capture_drain``) stores the files. Nothing in this module
+needs ``AF_UNIX``, ``fcntl`` or ``geteuid``, so it runs on Windows. Its lock
+is the kernel's (``msvcrt.locking`` / ``fcntl.flock``): a crashed drainer
+releases it with its process, so no stale-lock recovery exists or is needed.
+
+source: ADR-1094
+"""
+
+from __future__ import annotations
+
+import errno
+import json
+import os
+import sys
+import time
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import TextIO
+
+from mcp_server.shared.log_rotation import open_rotating_log
+
+if sys.platform == "win32":
+    import msvcrt
+else:
+    import fcntl
+
+# source: ADR-1084 (private directory under the configured Cortex root, mode 0700
+# where the platform has modes; on Windows it inherits the user-profile ACL)
+PRIVATE_DIR = 0o700
+# source: ADR-0488 (the worker's log is 0o600)
+PRIVATE_FILE = 0o600
+# source: ADR-0486 (the worker's own private ``.capture-worker`` directory)
+SPOOL_ROOT = ".capture-worker"
+REJECTED_SUFFIX = ".rejected"
+_PARTIAL_SUFFIX = ".partial"
+# source: ADR-1094 (measured 2026-10-07: a cold capture takes 10.5 s; the bound is
+# about ten times that, so only a hang reaches it, and then the file is rejected
+# and reported instead of holding the lock for good)
+STORE_SECONDS = 120.0
+# source: ADR-1094 (a backlog older than this means the drainer is not making
+# progress, it spends at most STORE_SECONDS per file; the hook reports it)
+STALL_SECONDS = 600.0
+# source: ADR-1094 (a ``.partial`` is a hook that died between write and rename, a
+# write takes milliseconds, so an hour is long past any live writer)
+PARTIAL_KEEP_SECONDS = 3600.0
+# source: ADR-1094 (refused payloads are kept a week for inspection, then removed)
+REJECTED_KEEP_SECONDS = 7 * 86400.0
+_PAYLOAD_SUFFIX = ".json"
+# source: https://learn.microsoft.com/en-us/cpp/c-runtime-library/reference/locking
+# (``msvcrt.locking`` locks ``nbytes`` from the current position; one byte is
+# the minimum that carries a lock)
+_LOCK_BYTES = 1
+
+
+def spool_directory(root: Path) -> Path:
+    """postcondition: ``<root>/.capture-worker/spool`` exists; its path is returned."""
+    root = root.expanduser().absolute()
+    spool = root / SPOOL_ROOT / "spool"
+    # one level at a time: ``mkdir(parents=True)`` applies ``mode`` to the leaf only
+    for directory in (root, spool.parent, spool):
+        directory.mkdir(mode=PRIVATE_DIR, exist_ok=True)
+    return spool
+
+
+def log_path(spool: Path) -> Path:
+    return spool.parent / "drain.log"
+
+
+@contextmanager
+def open_log(spool: Path) -> Iterator[TextIO]:
+    """The drainer's single log, private and size-rotated, for its stdout/stderr.
+
+    The drainer (stdio, logging, tracebacks, library progress bars) writes only
+    here, so one file records everything, including a crash before its own
+    logging exists. Rotation happens at each open (``open_rotating_log``, ADR-0655:
+    one ``.1`` generation), so the file stays bounded. The Popen must run inside
+    this context. source: ADR-0488 (private diagnostics, owner-only mode)"""
+    with open_rotating_log(log_path(spool)) as stream:
+        # on the descriptor that is actually written, after any rotation
+        # created a fresh file with the umask's mode
+        if sys.platform != "win32":
+            os.fchmod(stream.fileno(), PRIVATE_FILE)
+        yield stream
+
+
+def write(spool: Path, payload: dict[str, object]) -> Path:
+    """precondition: ``payload`` is JSON-serialisable.
+    postcondition: one complete ``*.json`` file holds ``payload``; a reader never
+    sees a partial one (written under another suffix, then ``os.replace``d)."""
+    name = f"{time.time_ns():020d}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
+    partial = spool / f".{name}{_PARTIAL_SUFFIX}"
+    target = spool / f"{name}{_PAYLOAD_SUFFIX}"
+    partial.write_text(json.dumps(payload), encoding="utf-8")
+    os.replace(partial, target)
+    return target
+
+
+def pending(spool: Path) -> list[Path]:
+    """Complete payload files in name order. The order is not a contract: captures
+    are independent and the name's clock prefix is only a hint (ADR-1094), so
+    nothing may infer age or sequence from it; use the files' mtime."""
+    return sorted(spool.glob(f"*{_PAYLOAD_SUFFIX}"))
+
+
+def oldest_pending_age(spool: Path, now: float) -> float | None:
+    """Age in seconds of the oldest complete payload file, or None when none exists.
+    A file the drainer removes while we look is skipped, not an error."""
+    mtimes = []
+    for path in pending(spool):
+        try:
+            mtimes.append(path.stat().st_mtime)
+        except FileNotFoundError:
+            continue
+    return now - min(mtimes) if mtimes else None
+
+
+def sweep(spool: Path, now: float) -> int:
+    """Remove orphaned ``.partial`` files and expired ``.rejected`` files.
+
+    postcondition: returns how many were removed; a file that vanished meanwhile
+    counts as removed by someone else."""
+    removed = 0
+    for suffix, keep in (
+        (_PARTIAL_SUFFIX, PARTIAL_KEEP_SECONDS),
+        (REJECTED_SUFFIX, REJECTED_KEEP_SECONDS),
+    ):
+        for path in spool.glob(f"*{suffix}"):
+            try:
+                if now - path.stat().st_mtime > keep:
+                    path.unlink()
+                    removed += 1
+            except FileNotFoundError:
+                continue
+    return removed
+
+
+def read(path: Path) -> dict[str, object]:
+    """Raises ``OSError`` or ``ValueError`` on an unreadable or non-object file."""
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError(f"capture spool file is not a JSON object: {path.name}")
+    return value
+
+
+def reject(path: Path) -> Path:
+    """Keep a refused file for inspection; it no longer matches ``pending``."""
+    target = path.with_name(path.name + REJECTED_SUFFIX)
+    os.replace(path, target)
+    return target
+
+
+@contextmanager
+def drain_lock(spool: Path) -> Iterator[bool]:
+    """Non-blocking exclusive drainer lock; yields whether this process holds it.
+
+    postcondition: at most one process yields True at a time; a lost race
+    yields False at once (never waits), any other ``OSError`` propagates."""
+    descriptor = os.open(spool.parent / "drain.lock", os.O_RDWR | os.O_CREAT, 0o600)
+    held = False
+    try:
+        held = _try_lock(descriptor)
+        yield held
+    finally:
+        if held and sys.platform == "win32":
+            # Windows frees the locks of a closed handle asynchronously; unlock
+            # explicitly so the next acquisition (the drainer's rescan) sees it free.
+            msvcrt.locking(descriptor, msvcrt.LK_UNLCK, _LOCK_BYTES)
+        os.close(descriptor)
+
+
+def _try_lock(descriptor: int) -> bool:
+    try:
+        if sys.platform == "win32":
+            msvcrt.locking(descriptor, msvcrt.LK_NBLCK, _LOCK_BYTES)
+        else:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError as exc:
+        if exc.errno not in {errno.EAGAIN, errno.EACCES}:
+            raise
+        return False
+    return True
diff --git a/tests_py/hooks/test_capture_drain.py b/tests_py/hooks/test_capture_drain.py
new file mode 100644
index 000000000..d8993e867
--- /dev/null
+++ b/tests_py/hooks/test_capture_drain.py
@@ -0,0 +1,221 @@
+"""The spool drainer: one handler load per burst, one drainer at a time, nothing lost.
+
+``test_drainer_outlives_its_parent_and_stores_the_spooled_line`` starts the real
+drainer from a parent that exits at once, with the real platform's detach options
+(Windows flags on the Windows CI leg), and synchronises on pipe EOF, never on a delay.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from mcp_server.hooks import capture_dispatch, capture_drain
+from mcp_server.infrastructure import capture_spool
+from tests_py.hooks.test_capture_unsupported_platform import EXPECTED
+from tests_py.hooks.test_capture_worker_policy import payload
+
+STORED = {"stored": True, "memory_id": 1}
+
+
+def _stored_rows() -> int:
+    from mcp_server.infrastructure.memory_config import get_memory_settings
+    from mcp_server.infrastructure.memory_store import get_shared_store
+
+    settings = get_memory_settings()
+    store = get_shared_store(settings.DB_PATH, settings.EMBEDDING_DIM)
+    return store.count_memories()["total"]
+
+
+class _Spool(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.spool = capture_spool.spool_directory(Path(self.directory.name))
+        self.report = mock.patch.object(capture_drain, "report_failure").start()
+        self.addCleanup(mock.patch.stopall)
+
+
+class TestDrain(_Spool):
+    def test_stores_every_pending_file_through_the_real_handler(self):
+        capture_spool.write(self.spool, EXPECTED)
+        before = _stored_rows()
+        capture_drain.drain(self.spool)
+        self.assertEqual(_stored_rows(), before + 1)
+        self.assertEqual(capture_spool.pending(self.spool), [])
+        self.report.assert_not_called()
+
+    def test_a_held_lock_exits_without_touching_the_handler(self):
+        capture_spool.write(self.spool, EXPECTED)
+        with (
+            capture_spool.drain_lock(self.spool) as outer,
+            mock.patch.object(
+                capture_drain, "store", side_effect=AssertionError("handler used")
+            ),
+        ):
+            self.assertTrue(outer)
+            capture_drain.drain(self.spool)
+        self.assertEqual(len(capture_spool.pending(self.spool)), 1)
+
+    def test_refused_payload_is_rejected_and_reported_and_does_not_block_the_rest(self):
+        capture_spool.write(self.spool, {**payload(), "origin_tool": "Unknown"})
+        good = capture_spool.write(self.spool, EXPECTED)
+        calls = []
+
+        async def store(value):
+            calls.append(value)
+            if value["origin_tool"] == "Unknown":
+                raise ValueError("unknown producing capture tool")
+            return STORED
+
+        with mock.patch.object(capture_drain, "store", store):
+            capture_drain.drain(self.spool)
+        self.assertEqual(len(calls), 2)
+        self.assertFalse(good.exists())
+        rejected = sorted(self.spool.glob("*" + capture_spool.REJECTED_SUFFIX))
+        self.assertEqual(len(rejected), 1)
+        self.assertIn("unknown producing capture tool", self.report.call_args.args[0])
+
+    def test_handler_failure_is_rejected_and_reported_not_swallowed(self):
+        capture_spool.write(self.spool, EXPECTED)
+
+        async def store(value):
+            raise RuntimeError("database is down")
+
+        with mock.patch.object(capture_drain, "store", store):
+            capture_drain.drain(self.spool)
+        self.assertEqual(capture_spool.pending(self.spool), [])
+        self.assertEqual(len(list(self.spool.glob("*.rejected"))), 1)
+        self.assertIn("database is down", self.report.call_args.args[0])
+
+    def test_unreadable_file_is_rejected_and_reported(self):
+        (self.spool / "00000000000000000001-1-aaaaaaaa.json").write_text("{not json")
+        capture_drain.drain(self.spool)
+        self.assertEqual(capture_spool.pending(self.spool), [])
+        self.report.assert_called_once()
+
+    def test_file_written_during_a_pass_is_stored_by_the_rescan(self):
+        capture_spool.write(self.spool, EXPECTED)
+        seen = []
+
+        async def store(value):
+            seen.append(value)
+            if len(seen) == 1:
+                capture_spool.write(self.spool, EXPECTED)
+            return STORED
+
+        with mock.patch.object(capture_drain, "store", store):
+            capture_drain.drain(self.spool)
+        self.assertEqual(len(seen), 2)
+        self.assertEqual(capture_spool.pending(self.spool), [])
+
+    def test_a_hung_store_is_cancelled_rejected_and_reported_not_held_forever(self):
+        hung = capture_spool.write(self.spool, EXPECTED)
+        good = capture_spool.write(self.spool, {**EXPECTED, "content": "second"})
+        seen = []
+
+        async def store(value):
+            seen.append(value["content"])
+            if value["content"] != "second":
+                await asyncio.Event().wait()  # never set: only the deadline ends it
+            return STORED
+
+        with (
+            mock.patch.object(capture_drain, "store", store),
+            mock.patch.object(capture_spool, "STORE_SECONDS", 0.05),
+        ):
+            capture_drain.drain(self.spool)
+        self.assertEqual(len(seen), 2)  # the file behind the hung one was stored
+        self.assertFalse(good.exists())
+        self.assertFalse(hung.exists())
+        self.assertEqual(len(list(self.spool.glob("*.rejected"))), 1)
+        self.assertIn("TimeoutError", self.report.call_args.args[0])
+
+    def test_reject_failure_deletes_the_file_and_reports_once_without_crashing(self):
+        capture_spool.write(self.spool, EXPECTED)
+
+        async def store(value):
+            raise RuntimeError("database is down")
+
+        with (
+            mock.patch.object(capture_drain, "store", store),
+            mock.patch.object(
+                capture_spool, "reject", side_effect=OSError("read-only filesystem")
+            ),
+        ):
+            capture_drain.drain(self.spool)
+        self.assertEqual(capture_spool.pending(self.spool), [])
+        self.report.assert_called_once()
+        message = self.report.call_args.args[0]
+        self.assertIn("deleted, rename failed: read-only filesystem", message)
+        self.assertIn("database is down", message)
+
+    def test_a_file_that_can_neither_be_renamed_nor_deleted_is_reported_once_per_run(
+        self,
+    ):
+        stuck = capture_spool.write(self.spool, EXPECTED)
+
+        async def store(value):
+            raise RuntimeError("database is down")
+
+        with (
+            mock.patch.object(capture_drain, "store", store),
+            mock.patch.object(
+                capture_spool, "reject", side_effect=OSError("rename denied")
+            ),
+            mock.patch.object(Path, "unlink", side_effect=OSError("delete denied")),
+        ):
+            capture_drain.drain(self.spool)  # must terminate
+        self.assertEqual(capture_spool.pending(self.spool), [stuck])
+        self.report.assert_called_once()
+        self.assertIn("left in place", self.report.call_args.args[0])
+
+    def test_a_drain_sweeps_orphaned_partial_files(self):
+        orphan = self.spool / ".dead.partial"
+        orphan.write_text("{")
+        long_ago = time.time() - capture_spool.PARTIAL_KEEP_SECONDS - 10
+        os.utime(orphan, (long_ago, long_ago))
+        capture_drain.drain(self.spool)
+        self.assertFalse(orphan.exists())
+
+
+# The intermediate process plays the hook: it starts the drainer through the
+# production ``spawn_drainer`` (real platform options: Windows creationflags on the
+# Windows leg), handing it its own stderr pipe, and exits at once.
+HOOK = (
+    "import sys\n"
+    "from mcp_server.hooks.capture_dispatch import spawn_drainer\n"
+    "spawn_drainer(sys.stderr.buffer)\n"
+)
+
+
+class TestDetachedDrainer(unittest.TestCase):
+    def test_drainer_outlives_its_parent_and_stores_the_spooled_line(self):
+        spool = capture_spool.spool_directory(capture_dispatch.CLAUDE_DIR)
+        self.assertEqual(capture_spool.pending(spool), [])
+        capture_spool.write(spool, EXPECTED)
+        before = _stored_rows()
+        hook = subprocess.Popen(
+            [sys.executable, "-c", HOOK],
+            stderr=subprocess.PIPE,
+            cwd=Path(capture_dispatch.__file__).resolve().parents[2],
+        )
+        self.assertEqual(hook.wait(), 0)  # the parent is gone ...
+        assert hook.stderr is not None
+        # ... EOF on the pipe arrives only when the drainer, the last holder of its
+        # write end, has exited: no delay decides anything.
+        diagnostics = hook.stderr.read()
+        hook.stderr.close()
+        self.assertEqual(capture_spool.pending(spool), [], diagnostics)
+        self.assertEqual(_stored_rows(), before + 1, diagnostics)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests_py/hooks/test_capture_unsupported_platform.py b/tests_py/hooks/test_capture_unsupported_platform.py
new file mode 100644
index 000000000..12fe67d4d
--- /dev/null
+++ b/tests_py/hooks/test_capture_unsupported_platform.py
@@ -0,0 +1,213 @@
+"""Issue #659: a platform without the resident worker spools and detaches a drainer.
+
+The branch is chosen by the capability test, never by catching a transport error,
+and the hook never loads the handler (the import that costs seconds against the
+hook's 10 s budget): it writes a file and starts a detached process.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from contextlib import contextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+from mcp_server.hooks import capture_dispatch, post_tool_capture as hook
+from mcp_server.infrastructure import capture_peer, capture_spool
+from tests_py.hooks.test_capture_worker_policy import payload
+
+CONTENT = (
+    "Bash: pytest tests_py/hooks -q\n"
+    "3 failed, 41 passed: the capture worker refused a Windows interpreter "
+    "and nothing was stored for two weeks of editing sessions."
+)
+STREAM = mock.sentinel.stream  # popen_options only forwards it
+EXPECTED = {
+    "content": CONTENT,
+    "tags": ["auto-captured"],
+    "directory": "/fixture",
+    "source": "post_tool_capture",
+    "origin_tool": "Bash",
+    "write_class": "auto",
+    "force": False,
+}
+
+
+@contextmanager
+def _windows():
+    # Scoped to the two modules that read the platform: a process-wide
+    # sys.platform would also redirect unrelated lazy imports (msvcrt) that
+    # this POSIX test host cannot load.
+    fake = SimpleNamespace(platform="win32", executable=sys.executable)
+    with (
+        mock.patch.object(capture_dispatch, "sys", fake),
+        mock.patch.object(capture_peer, "sys", fake),
+    ):
+        yield
+
+
+class TestCapability(unittest.TestCase):
+    def test_windows_is_reported_unsupported_without_raising(self):
+        with _windows():
+            self.assertFalse(capture_peer.is_supported())
+
+    def test_missing_af_unix_is_unsupported_even_on_a_posix_name(self):
+        with mock.patch.object(capture_peer, "socket", mock.Mock(spec=[])):
+            self.assertFalse(capture_peer.is_supported())
+
+    def test_linux_and_macos_keep_the_resident_worker(self):
+        for name in ("linux", "darwin"):
+            with mock.patch.object(capture_peer, "sys", SimpleNamespace(platform=name)):
+                self.assertEqual(
+                    capture_peer.is_supported(), hasattr(socket, "AF_UNIX")
+                )
+
+    @unittest.skipUnless(sys.platform == "win32", "native Windows only")
+    def test_native_windows_is_unsupported(self):
+        self.assertFalse(capture_peer.is_supported())
+
+    def test_unsupported_platform_still_refuses_the_worker_transport(self):
+        with _windows(), self.assertRaisesRegex(OSError, "requires Linux/macOS"):
+            capture_peer.supported()
+
+
+class TestPopenOptions(unittest.TestCase):
+    def test_windows_detaches_through_creationflags_only(self):
+        options = capture_dispatch.popen_options("win32", STREAM)
+        # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP (Win32 process-creation flags)
+        self.assertEqual(options["creationflags"], 0x00000008 | 0x00000200)
+        for ignored_or_unsupported in ("start_new_session", "pass_fds"):
+            self.assertNotIn(ignored_or_unsupported, options)
+
+    def test_every_platform_closes_stdin_logs_output_and_inherits_the_environment(self):
+        for name in ("win32", "linux", "darwin", "freebsd14"):
+            options = capture_dispatch.popen_options(name, STREAM)
+            self.assertTrue(options["close_fds"])
+            self.assertIs(options["stdin"], subprocess.DEVNULL)
+            self.assertIs(options["stdout"], STREAM)
+            self.assertIs(options["stderr"], STREAM)
+            self.assertEqual(options["env"], dict(os.environ))
+
+    def test_posix_detaches_through_a_new_session(self):
+        for name in ("linux", "darwin", "freebsd14"):
+            options = capture_dispatch.popen_options(name, STREAM)
+            self.assertIs(options["start_new_session"], True)
+            self.assertNotIn("creationflags", options)
+
+
+class TestHookOnUnsupportedPlatform(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.root = Path(self.directory.name)
+        mock.patch.object(capture_dispatch, "CLAUDE_DIR", self.root).start()
+        self.addCleanup(mock.patch.stopall)
+
+    def _store(self):
+        hook._store_memory("Bash", CONTENT, ["auto-captured"], "/fixture")
+
+    def test_hook_spools_and_detaches_without_loading_the_handler(self):
+        with (
+            _windows(),
+            mock.patch.object(
+                hook, "_load_remember", side_effect=AssertionError("handler loaded")
+            ),
+            mock.patch.object(capture_dispatch, "deliver") as deliver,
+            mock.patch.object(capture_dispatch.subprocess, "Popen") as popen,
+            mock.patch.object(hook, "_log"),
+        ):
+            self._store()
+        deliver.assert_not_called()
+        spool = capture_spool.spool_directory(self.root)
+        files = capture_spool.pending(spool)
+        self.assertEqual([capture_spool.read(f) for f in files], [EXPECTED])
+        popen.assert_called_once()
+        command = popen.call_args.args[0]
+        self.assertEqual(command[1:], ["-m", "mcp_server.hooks.capture_drain"])
+        stream = popen.call_args.kwargs["stdout"]
+        self.assertEqual(
+            popen.call_args.kwargs, capture_dispatch.popen_options("win32", stream)
+        )
+        self.assertTrue(stream.closed)
+        self.assertTrue((self.root / ".capture-worker" / "drain.log").exists())
+
+    def test_a_stalled_backlog_is_reported_and_the_capture_still_queues(self):
+        spool = capture_spool.spool_directory(self.root)
+        old = capture_spool.write(spool, EXPECTED)
+        long_ago = time.time() - capture_spool.STALL_SECONDS - 10
+        os.utime(old, (long_ago, long_ago))
+        with (
+            _windows(),
+            mock.patch.object(capture_dispatch.subprocess, "Popen"),
+            mock.patch.object(capture_dispatch, "report_failure") as report,
+            mock.patch.object(hook, "_log"),
+        ):
+            self._store()
+        report.assert_called_once()
+        self.assertEqual(report.call_args.kwargs["operation"], "capture_spool_stalled")
+        self.assertEqual(len(capture_spool.pending(spool)), 2)
+
+    def test_a_fresh_backlog_is_not_reported_as_stalled(self):
+        spool = capture_spool.spool_directory(self.root)
+        capture_spool.write(spool, EXPECTED)
+        with (
+            _windows(),
+            mock.patch.object(capture_dispatch.subprocess, "Popen"),
+            mock.patch.object(capture_dispatch, "report_failure") as report,
+            mock.patch.object(hook, "_log"),
+        ):
+            self._store()
+        report.assert_not_called()
+
+    def test_spawn_failure_keeps_the_file_and_is_reported(self):
+        with (
+            _windows(),
+            mock.patch.object(
+                capture_dispatch.subprocess, "Popen", side_effect=OSError("no spawn")
+            ),
+            mock.patch.object(capture_dispatch, "report_failure") as report,
+            mock.patch.object(hook, "_log"),
+        ):
+            self._store()
+        self.assertEqual(
+            len(capture_spool.pending(self.root / ".capture-worker" / "spool")), 1
+        )
+        self.assertIn("no spawn", report.call_args.args[0])
+
+    def test_refused_payload_is_reported_and_never_spooled(self):
+        with (
+            _windows(),
+            mock.patch.object(capture_dispatch.subprocess, "Popen") as popen,
+            mock.patch.object(capture_dispatch, "report_failure") as report,
+        ):
+            self.assertFalse(
+                capture_dispatch.dispatch({**payload(), "origin_tool": "Unknown"})
+            )
+        popen.assert_not_called()
+        self.assertEqual(
+            capture_spool.pending(capture_spool.spool_directory(self.root)), []
+        )
+        self.assertIn("unknown producing capture tool", report.call_args.args[0])
+
+    def test_supported_platform_uses_the_worker_not_the_spool(self):
+        with (
+            mock.patch.object(capture_dispatch, "is_supported", return_value=True),
+            mock.patch.object(capture_dispatch, "deliver") as deliver,
+            mock.patch.object(capture_dispatch.subprocess, "Popen") as popen,
+            mock.patch.object(hook, "_log"),
+        ):
+            self._store()
+        deliver.assert_called_once()
+        popen.assert_not_called()
+        self.assertFalse((self.root / ".capture-worker").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests_py/infrastructure/test_capture_spool.py b/tests_py/infrastructure/test_capture_spool.py
new file mode 100644
index 000000000..e4ab92340
--- /dev/null
+++ b/tests_py/infrastructure/test_capture_spool.py
@@ -0,0 +1,149 @@
+"""File spool primitives: atomic publication, ordering, rejection, non-blocking lock."""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from mcp_server.infrastructure import capture_spool
+from mcp_server.shared import log_rotation
+
+
+class TestCaptureSpool(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.spool = capture_spool.spool_directory(Path(self.directory.name))
+
+    def test_spool_lives_under_the_private_capture_directory(self):
+        self.assertEqual(
+            self.spool,
+            Path(self.directory.name).absolute() / ".capture-worker" / "spool",
+        )
+
+    @unittest.skipIf(sys.platform == "win32", "Windows has no POSIX modes")
+    def test_every_created_level_is_owner_only(self):
+        capture = self.spool.parent
+        for directory in (capture, self.spool):
+            self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o700)
+
+    def test_log_appends_across_openings(self):
+        for text in ("one\n", "two\n"):
+            with capture_spool.open_log(self.spool) as stream:
+                stream.write(text)
+        self.assertEqual(capture_spool.log_path(self.spool).read_text(), "one\ntwo\n")
+
+    @unittest.skipIf(sys.platform == "win32", "Windows has no POSIX modes")
+    def test_log_is_owner_only_even_when_it_pre_exists_with_a_wider_mode(self):
+        path = capture_spool.log_path(self.spool)
+        with capture_spool.open_log(self.spool):
+            pass
+        self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+        path.chmod(0o644)
+        with capture_spool.open_log(self.spool):
+            pass
+        self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+
+    def test_log_is_rotated_so_it_stays_bounded(self):
+        path = capture_spool.log_path(self.spool)
+        path.write_bytes(b"x" * log_rotation.MAX_LOG_BYTES)
+        with capture_spool.open_log(self.spool) as stream:
+            stream.write("fresh\n")
+        self.assertEqual(path.read_text(), "fresh\n")
+        self.assertEqual(
+            path.with_name(path.name + ".1").stat().st_size, log_rotation.MAX_LOG_BYTES
+        )
+
+    def test_oldest_pending_age_is_none_when_empty_else_the_oldest_files_age(self):
+        self.assertIsNone(capture_spool.oldest_pending_age(self.spool, 1000.0))
+        files = [capture_spool.write(self.spool, {"n": n}) for n in range(5)]
+        # mtimes set explicitly, in an order unrelated to the names, so neither the
+        # clock tick nor the name order can decide the result
+        mtimes = (700.0, 100.0, 900.0, 300.0, 500.0)
+        for path, mtime in zip(files, mtimes, strict=True):
+            os.utime(path, (mtime, mtime))
+        self.assertEqual(capture_spool.oldest_pending_age(self.spool, 1000.0), 900.0)
+
+    @unittest.skipIf(sys.platform == "win32", "Windows has no POSIX modes")
+    def test_log_is_owner_only_after_a_real_rotation(self):
+        path = capture_spool.log_path(self.spool)
+        path.write_bytes(b"x" * log_rotation.MAX_LOG_BYTES)
+        path.chmod(0o644)
+        old_umask = os.umask(0o022)
+        self.addCleanup(os.umask, old_umask)
+        with capture_spool.open_log(self.spool):
+            pass
+        previous = path.with_name(path.name + ".1")
+        self.assertEqual(previous.stat().st_size, log_rotation.MAX_LOG_BYTES)
+        self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+
+    def test_sweep_removes_only_expired_partial_and_rejected_files(self):
+        now = 10_000_000.0
+        stale_partial = self.spool / ".a.partial"
+        fresh_partial = self.spool / ".b.partial"
+        stale_rejected = capture_spool.reject(capture_spool.write(self.spool, {"n": 1}))
+        fresh_rejected = capture_spool.reject(capture_spool.write(self.spool, {"n": 2}))
+        live = capture_spool.write(self.spool, {"n": 3})
+        for path in (stale_partial, fresh_partial):
+            path.write_text("{")
+        age = {
+            stale_partial: capture_spool.PARTIAL_KEEP_SECONDS + 1,
+            fresh_partial: 1,
+            stale_rejected: capture_spool.REJECTED_KEEP_SECONDS + 1,
+            fresh_rejected: capture_spool.PARTIAL_KEEP_SECONDS + 1,
+            live: capture_spool.REJECTED_KEEP_SECONDS + 1,
+        }
+        for path, seconds in age.items():
+            os.utime(path, (now - seconds, now - seconds))
+        self.assertEqual(capture_spool.sweep(self.spool, now), 2)
+        self.assertEqual(
+            {p for p in self.spool.iterdir()}, {fresh_partial, fresh_rejected, live}
+        )
+
+    def test_a_written_payload_round_trips_and_leaves_no_partial_file(self):
+        path = capture_spool.write(self.spool, {"content": "é☃"})
+        self.assertEqual(capture_spool.read(path), {"content": "é☃"})
+        self.assertEqual(sorted(p.name for p in self.spool.iterdir()), [path.name])
+
+    def test_pending_lists_complete_files_and_ignores_partial_and_rejected_ones(self):
+        first = capture_spool.write(self.spool, {"n": 1})
+        second = capture_spool.write(self.spool, {"n": 2})
+        (self.spool / ".x.partial").write_text("{")
+        capture_spool.reject(capture_spool.write(self.spool, {"n": 3}))
+        self.assertEqual(set(capture_spool.pending(self.spool)), {first, second})
+
+    def test_reject_keeps_the_file_out_of_pending(self):
+        path = capture_spool.write(self.spool, {"n": 1})
+        kept = capture_spool.reject(path)
+        self.assertTrue(kept.exists())
+        self.assertFalse(path.exists())
+        self.assertEqual(capture_spool.pending(self.spool), [])
+
+    def test_non_object_json_is_refused(self):
+        path = self.spool / "1.json"
+        path.write_text("[1]")
+        with self.assertRaisesRegex(ValueError, "not a JSON object"):
+            capture_spool.read(path)
+
+    def test_second_holder_loses_without_waiting_and_lock_is_reusable(self):
+        with capture_spool.drain_lock(self.spool) as first:
+            with capture_spool.drain_lock(self.spool) as second:
+                self.assertEqual((first, second), (True, False))
+        with capture_spool.drain_lock(self.spool) as again:
+            self.assertTrue(again)
+
+    def test_an_unexpected_lock_error_propagates(self):
+        failure = OSError(5, "I/O error")
+        with mock.patch.object(capture_spool, "_try_lock", side_effect=failure):
+            with self.assertRaises(OSError):
+                with capture_spool.drain_lock(self.spool):
+                    pass
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/wiki/adr/cortex/1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md b/wiki/adr/cortex/1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md
new file mode 100644
index 000000000..6bd33ce57
--- /dev/null
+++ b/wiki/adr/cortex/1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md
@@ -0,0 +1,103 @@
+---
+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).
diff --git a/wiki/manifest.json b/wiki/manifest.json
index c2aa13ab0..655e1ddb1 100644
--- a/wiki/manifest.json
+++ b/wiki/manifest.json
@@ -3543,6 +3543,10 @@
     "ADR-1093": {
       "mirror": "ADR-1093-publish-cross-parameter-constraints-in-the-tool-input-schema.md",
       "path": "adr/cortex/1093-publish-cross-parameter-constraints-in-the-tool-input-schema.md"
+    },
+    "ADR-1094": {
+      "mirror": "ADR-1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md",
+      "path": "adr/cortex/1094-spool-captures-for-a-detached-drainer-where-the-resident-worker-cannot-run.md"
     }
   },
   "project": "cortex",