captain-hook: ⚡️ Coalesce reviews before spawn and skip recorded correction anchors

CHANGELOG.md

@@ -100,6 +100,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   advisory hook fails open and a mandatory hook withholds its completion exactly as for
   transcript evidence. The teammate tool approval excludes native Bash before it reads the
   flag, so a Bash event never pays for the second hook's walk.
+- **Session-reviewer bursts spawn one child, and rescans skip recorded corrections.** The
+  `SessionStart` and `SessionEnd` guard used to detach a Python reviewer child on every event.
+  Most of those children then lost the pending-pass lock after paying for an interpreter, imports,
+  and an origin lookup. The guard now claims that lock before the enrollment probe and passes the locked
+  descriptor to the child, so an event that finds a pass already queued for its transcript
+  directory spawns nothing. A child releases its claim the moment it takes the repo lock, so an
+  event during a running pass still queues exactly one follow-up. A
+  changed transcript now prepares Git correction evidence only for anchors with no row in the
+  corrections ledger, keyed by each anchor's session and event. Recorded anchors no longer start
+  Git subprocesses.
 - **A session may kill or renice a process it started.** The session guard reads the target's
   environment with `ps -E` and allows a literal `kill <pid>` or `renice -p <pid>` when its
   `CLAUDE_CODE_SESSION_ID` names the calling session, including a process reparented to

captain_hook/review/cli.py

@@ -107,13 +107,14 @@ def sweep_hook() -> None:
 @click.option("--transcript", required=True, type=click.Path(path_type=Path), help="The ended session's transcript")
 @click.option("--cwd", "cwd", default=None, help="The session's working directory (default: the process cwd)")
 @click.option("--sweep", is_flag=True, default=False, help="Run the throttled sweep (no PR sync, no brain)")
-def spawn(transcript: Path, cwd: str | None, sweep: bool) -> None:
+@click.option("--pending-fd", type=int, default=None, help="The inherited pending-pass lock descriptor")
+def spawn(transcript: Path, cwd: str | None, sweep: bool, pending_fd: int | None) -> None:
     """Run the detached reviewer pass over one ended session (spawned by ``review run``)."""
     import asyncio
 
     from captain_hook.review.pipeline import spawn_session
 
-    click.echo(asyncio.run(spawn_session(transcript, cwd=cwd or os.getcwd(), sweep=sweep)))
+    click.echo(asyncio.run(spawn_session(transcript, cwd=cwd or os.getcwd(), sweep=sweep, pending_fd=pending_fd)))
 
 
 @review.command()

captain_hook/review/pipeline.py

@@ -147,7 +147,7 @@ def sweep_key(cwd: str) -> str:
     return hashlib.sha256(cwd.encode()).hexdigest()[:12]
 
 
-def spawn_argv(transcript: str, cwd: str | None, *, sweep: bool = False) -> list[str]:
+def spawn_argv(transcript: str, cwd: str | None, *, sweep: bool = False, pending_fd: int | None = None) -> list[str]:
     # -P: the reviewer runs with the session's repo as its cwd, and `-m` would otherwise put that repo
     # at the head of sys.path, where a directory sharing a dependency's name shadows the installed one.
     return [
@@ -161,6 +161,7 @@ def spawn_argv(transcript: str, cwd: str | None, *, sweep: bool = False) -> list
         transcript,
         *(("--cwd", cwd) if cwd else ()),
         *(("--sweep",) if sweep else ()),
+        *(("--pending-fd", str(pending_fd)) if pending_fd is not None else ()),
     ]
 
 
@@ -195,7 +196,7 @@ def payload_transcript(raw: bytes, *, label: str) -> tuple[str, str | None] | No
     return transcript, cwd if isinstance(cwd, str) else None
 
 
-def detach(argv: list[str], *, spawned: str) -> None:
+def detach(argv: list[str], *, spawned: str, pass_fds: tuple[int, ...] = ()) -> None:
     try:
         (log_path := review_log_path()).parent.mkdir(parents=True, exist_ok=True)
         with log_path.open("ab") as log:
@@ -205,6 +206,7 @@ def detach(argv: list[str], *, spawned: str) -> None:
                 stdout=log,
                 stderr=log,
                 start_new_session=True,
+                pass_fds=pass_fds,
                 cwd=reqenv.cwd(),
                 env=reqenv.env_map() | {SPAWNED_ENV: "1"},
             )
@@ -286,10 +288,18 @@ def guard_and_spawn(raw: bytes, *, gate_enrollment: bool = False) -> None:
     the raw CLI entry leaves it off, so the detached child stays the
     authoritative enrollment gate for that path.
 
+    Before the enrollment probe or any spawn, the guard claims the transcript
+    directory's pending-pass lock (:func:`pending_claim`) and hands the locked
+    descriptor to the child, so an event arriving while a pass is already queued
+    for that directory spawns nothing; while a pass runs, the next event queues
+    exactly one follow-up child behind it.
+
     Args:
         raw: The hook's stdin bytes, holding the SessionStart/SessionEnd JSON payload.
         gate_enrollment: Skip the spawn for a non-watched repo (native dispatch only).
     """
+    from captain_hook.review.settings import ReviewPaths
+
     if reqenv.getenv(SPAWNED_ENV):
         breadcrumb("review-run skip: CAPT_HOOK_SPAWNED set")
         return
@@ -300,10 +310,23 @@ def guard_and_spawn(raw: bytes, *, gate_enrollment: bool = False) -> None:
     if (parsed := payload_transcript(raw, label="review-run")) is None:
         return
     transcript, cwd = parsed
-    if gate_enrollment and not enrolled(cwd):
-        breadcrumb("review-run skip: not watching")
-        return
-    detach(spawn_argv(transcript, cwd), spawned=f"spawned {transcript}")
+    with ExitStack() as claim:
+        try:
+            pending_fd = claim.enter_context(pending_claim(ReviewPaths().db_path, Path(transcript)))
+        except OSError:
+            breadcrumb("review-run skip: pending lock OSError")
+            return
+        if pending_fd is None:
+            breadcrumb("review-run skip: a queued pass already covers this transcript directory")
+            return
+        if gate_enrollment and not enrolled(cwd):
+            breadcrumb("review-run skip: not watching")
+            return
+        detach(
+            spawn_argv(transcript, cwd, pending_fd=pending_fd),
+            spawned=f"spawned {transcript}",
+            pass_fds=(pending_fd,),
+        )
 
 
 def guard_and_sweep(raw: bytes, *, gate_enrollment: bool = False) -> None:
@@ -501,13 +524,48 @@ def repo_lock(settings: ReviewSettings, cwd: str, *, wait: bool) -> AbstractCont
     return exclusive_flock(settings.db_path.parent / "locks" / f"repo-{key}.lock", wait=wait)
 
 
-@contextmanager
-def queued_review_lock(settings: ReviewSettings, cwd: str, transcript: Path) -> Generator[bool]:
+def pending_lock_path(db_path: Path, transcript: Path) -> Path:
     key = hashlib.sha256(str(transcript.parent.resolve()).encode()).hexdigest()[:16]
+    return db_path.parent / "locks" / f"pending-{key}.lock"
+
+
+@contextmanager
+def pending_claim(db_path: Path, transcript: Path) -> Iterator[int | None]:
+    """Claims the transcript directory's pending-pass lock for a child about to detach.
+
+    Yields the locked descriptor, or ``None`` when a queued pass already holds the lock. Exit closes
+    this process's descriptor without ``LOCK_UN``: the ``flock`` lives on the open file description,
+    so a child that inherited the descriptor keeps holding it, and with no child it releases. The
+    child releases it with :func:`release_pending` once it holds the repo lock.
+    """
+    (path := pending_lock_path(db_path, transcript)).parent.mkdir(parents=True, exist_ok=True)
+    fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o644)
+    try:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            yield None
+            return
+        yield fd
+    finally:
+        os.close(fd)
+
+
+def release_pending(fd: int) -> None:
+    try:
+        fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        os.close(fd)
+
+
+@contextmanager
+def queued_review_lock(
+    settings: ReviewSettings, cwd: str, transcript: Path, *, pending_fd: int | None = None
+) -> Generator[bool]:
     with ExitStack() as pending, ExitStack() as active:
-        if not pending.enter_context(
-            exclusive_flock(settings.db_path.parent / "locks" / f"pending-{key}.lock", wait=False)
-        ):
+        if pending_fd is not None:
+            pending.callback(release_pending, pending_fd)
+        elif not pending.enter_context(exclusive_flock(pending_lock_path(settings.db_path, transcript), wait=False)):
             yield False
             return
         claimed = active.enter_context(repo_lock(settings, cwd, wait=True))
@@ -642,7 +700,12 @@ async def review_session(transcript: Path, *, cwd: str, settings: ReviewSettings
 
 
 async def spawn_session(
-    transcript: Path, *, cwd: str, settings: ReviewSettings | None = None, sweep: bool = False
+    transcript: Path,
+    *,
+    cwd: str,
+    settings: ReviewSettings | None = None,
+    sweep: bool = False,
+    pending_fd: int | None = None,
 ) -> SpawnReport:
     """Runs :func:`review_session` and records its outcome — the ``review spawn`` entry.
 
@@ -669,6 +732,8 @@ async def spawn_session(
         settings: The reviewer settings; constructed inside the recording
             boundary when omitted, so a settings/env failure still records.
         sweep: Whether to run the throttled sweep (no PR sync, no brain).
+        pending_fd: The pending-pass lock descriptor :func:`guard_and_spawn` claimed and
+            handed down; omitted, the child claims that lock itself.
 
     Returns:
         The recorded :class:`SpawnReport` for this pass.
@@ -682,7 +747,11 @@ async def spawn_session(
     started = datetime.now(UTC)
     try:
         settings = settings or ReviewSettings()
-        lock = repo_lock(settings, cwd, wait=False) if sweep else queued_review_lock(settings, cwd, transcript)
+        lock = (
+            repo_lock(settings, cwd, wait=False)
+            if sweep
+            else queued_review_lock(settings, cwd, transcript, pending_fd=pending_fd)
+        )
         with lock as claimed:
             if not claimed:
                 logger.info(f"review spawn skipped: another pass holds the lock for cwd={cwd}")

captain_hook/review/scan.py

@@ -28,6 +28,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import re
 from dataclasses import asdict, dataclass
 from enum import StrEnum
@@ -73,6 +74,7 @@
     from typing import Any
 
     from cc_transcript.context import ContextWindow
+    from cc_transcript.corrections import CorrectionLog
     from cc_transcript.mining.signals import MiningSignal
     from cc_transcript.models import TranscriptEvent
 
@@ -398,6 +400,7 @@ async def ingest(
     prepared: Mapping[str, Any],
     *,
     corrections: Sequence[Mapping[str, Any]],
+    ledger: CorrectionLedger,
     repo_key: RepoKey | None = None,
 ) -> ScanReport:
     from cc_transcript.mining.store import event_row, now
@@ -437,7 +440,7 @@ async def ingest(
                 session_id=candidate.ref.session_id,
                 occurred_at=candidate.occurred_at,
             )
-    await record_correction_drafts(corrections)
+    await record_correction_drafts(corrections, ledger=ledger)
     return ScanReport(scanned=1, inserted=inserted)
 
 
@@ -470,8 +473,44 @@ def correction_pages(
     return first + second, first_refused + second_refused
 
 
+class CorrectionLedger:
+    def __init__(self) -> None:
+        self.loop = asyncio.get_running_loop()
+        self.log: CorrectionLog | None = None
+
+    async def __aenter__(self) -> CorrectionLedger:
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        if self.log is not None:
+            await self.log.close()
+
+    async def handle(self) -> CorrectionLog:
+        from cc_transcript.corrections import CorrectionLog
+
+        self.log = self.log or await CorrectionLog.open()
+        return self.log
+
+    async def recorded(self, anchors: Sequence[EventRef]) -> set[EventRef]:
+        log = await self.handle()
+        keys = {
+            (row.session_id, row.anchor_uuid)
+            for session_id in {anchor.session_id for anchor in anchors}
+            for row in await log.for_session(session_id)
+        }
+        return {anchor for anchor in anchors if (anchor.session_id, anchor.event_uuid) in keys}
+
+    def recorded_from_thread(self, anchors: Sequence[EventRef]) -> set[EventRef]:
+        return asyncio.run_coroutine_threadsafe(self.recorded(anchors), self.loop).result()
+
+
 def prepare_source(
-    client: Any, path: Path, settings: ReviewSettings, repo_key: RepoKey | None = None
+    client: Any,
+    path: Path,
+    settings: ReviewSettings,
+    repo_key: RepoKey | None = None,
+    *,
+    recorded: Callable[[Sequence[EventRef]], set[EventRef]],
 ) -> tuple[dict[str, Any], list[dict[str, Any]], list[tuple[EventRef, EvidenceIncomplete]]]:
     from captain_hook.decisions import decisions_db_path
     from captain_hook.snapshots.review import REVIEW_POLICY, ProjectionBudget, decode_candidate
@@ -505,6 +544,9 @@ def prepare_source(
         eligible = [
             candidate for raw in candidates if (candidate := decode_candidate(raw)[0]).source_kind != HOOK_COMPLAINT
         ]
+        if eligible:
+            completed = recorded([candidate.ref for candidate in eligible])
+            eligible = [candidate for candidate in eligible if candidate.ref not in completed]
         corrections: list[dict[str, Any]] = []
         refused: list[tuple[EventRef, EvidenceIncomplete]] = []
         for offset in range(0, len(eligible), 256):
@@ -527,8 +569,6 @@ async def scan_transcript(
 async def scan(
     store: ReviewStore, *, settings: ReviewSettings, transcripts: Sequence[Path], repo_key: RepoKey | None = None
 ) -> ScanReport:
-    import asyncio
-
     from loguru import logger
 
     from captain_hook.snapshots.review import review_client
@@ -542,43 +582,46 @@ async def scan(
     next_checkpoint = None
     from captain_hook.snapshots.client import EvidenceIncomplete
 
-    with review_client() as client:
-        try:
-            pages = await asyncio.to_thread(lambda: list(client.pages("discover", roots=roots, checkpoint=checkpoint)))
-        except EvidenceIncomplete as exc:
-            if exc.status != "stale_cursor":
-                raise
-            pages = await asyncio.to_thread(lambda: list(client.pages("discover", roots=roots, checkpoint=None)))
-        for page in pages:
-            next_checkpoint = page["checkpoint"]
-            for entry in page["entries"]:
-                if entry["state"] != "present":
-                    continue
-                path = entry["path"]
-                revision_key = f"snapshot_revision:{dedup_key(path)}"
-                refused_key = f"snapshot_refused:{dedup_key(path)}"
-                if entry["revision"] in {await store.meta(revision_key), await store.meta(refused_key)}:
-                    continue
-                try:
-                    prepared, corrections, anchors_refused = await asyncio.to_thread(
-                        prepare_source, client, Path(path), settings, repo_key
-                    )
-                except EvidenceIncomplete as exc:
-                    if exc.status != "output_limit":
-                        raise
-                    logger.bind(transcript=path).warning(f"transcript evidence refused: {exc.reason}")
-                    await store.set_meta(refused_key, entry["revision"])
-                    refused += 1
-                    continue
-                for anchor, refusal in anchors_refused:
-                    logger.bind(transcript=path, session_id=anchor.session_id, event_uuid=anchor.event_uuid).warning(
-                        f"correction evidence refused: {refusal.reason}"
-                    )
-                report = await ingest(store, prepared, corrections=corrections, repo_key=repo_key)
-                await store.set_meta(revision_key, entry["revision"])
-                scanned += report.scanned
-                inserted += report.inserted
-                refused += len(anchors_refused)
+    async with CorrectionLedger() as ledger:
+        with review_client() as client:
+            try:
+                pages = await asyncio.to_thread(
+                    lambda: list(client.pages("discover", roots=roots, checkpoint=checkpoint))
+                )
+            except EvidenceIncomplete as exc:
+                if exc.status != "stale_cursor":
+                    raise
+                pages = await asyncio.to_thread(lambda: list(client.pages("discover", roots=roots, checkpoint=None)))
+            for page in pages:
+                next_checkpoint = page["checkpoint"]
+                for entry in page["entries"]:
+                    if entry["state"] != "present":
+                        continue
+                    path = entry["path"]
+                    revision_key = f"snapshot_revision:{dedup_key(path)}"
+                    refused_key = f"snapshot_refused:{dedup_key(path)}"
+                    if entry["revision"] in {await store.meta(revision_key), await store.meta(refused_key)}:
+                        continue
+                    try:
+                        prepared, corrections, anchors_refused = await asyncio.to_thread(
+                            prepare_source, client, Path(path), settings, repo_key, recorded=ledger.recorded_from_thread
+                        )
+                    except EvidenceIncomplete as exc:
+                        if exc.status != "output_limit":
+                            raise
+                        logger.bind(transcript=path).warning(f"transcript evidence refused: {exc.reason}")
+                        await store.set_meta(refused_key, entry["revision"])
+                        refused += 1
+                        continue
+                    for anchor, refusal in anchors_refused:
+                        logger.bind(
+                            transcript=path, session_id=anchor.session_id, event_uuid=anchor.event_uuid
+                        ).warning(f"correction evidence refused: {refusal.reason}")
+                    report = await ingest(store, prepared, corrections=corrections, ledger=ledger, repo_key=repo_key)
+                    await store.set_meta(revision_key, entry["revision"])
+                    scanned += report.scanned
+                    inserted += report.inserted
+                    refused += len(anchors_refused)
     if next_checkpoint is not None:
         await store.set_meta(checkpoint_key, next_checkpoint)
     return ScanReport(scanned, inserted, refused)