From e6a8ae7927c2a005a30d12ab51166850c6cb1c3c Mon Sep 17 00:00:00 2001 From: Yasyf Mohamedali Date: Fri, 2 Oct 2026 16:14:22 -0700 Subject: [PATCH 1/2] =?UTF-8?q?captain-hook:=20=E2=9A=A1=EF=B8=8F=20Coales?= =?UTF-8?q?ce=20reviews=20before=20spawn=20and=20skip=20recorded=20correct?= =?UTF-8?q?ion=20anchors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session-Id: 95aa46a7-b92a-4c7f-80dc-775333ed92a9 --- CHANGELOG.md | 10 ++ captain_hook/review/cli.py | 5 +- captain_hook/review/pipeline.py | 95 ++++++++++++++--- captain_hook/review/scan.py | 125 +++++++++++++++------- captain_hook/review/settings.py | 15 ++- captain_hook/snapshots/review.py | 38 +++---- tests/snapshot_review_helpers.py | 1 + tests/test_review_pipeline.py | 91 +++++++++++++++- tests/test_snapshot_review_consumers.py | 135 ++++++++++++++++++++++-- 9 files changed, 428 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 541c2296..a53abc30 100644 --- a/CHANGELOG.md +++ b/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 ` or `renice -p ` when its `CLAUDE_CODE_SESSION_ID` names the calling session, including a process reparented to diff --git a/captain_hook/review/cli.py b/captain_hook/review/cli.py index 36503114..15f73360 100644 --- a/captain_hook/review/cli.py +++ b/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() diff --git a/captain_hook/review/pipeline.py b/captain_hook/review/pipeline.py index c1a734e1..14be2a87 100644 --- a/captain_hook/review/pipeline.py +++ b/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}") diff --git a/captain_hook/review/scan.py b/captain_hook/review/scan.py index 1c19d94e..95b5f956 100644 --- a/captain_hook/review/scan.py +++ b/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) diff --git a/captain_hook/review/settings.py b/captain_hook/review/settings.py index bd8ba2c6..bb9a15e3 100644 --- a/captain_hook/review/settings.py +++ b/captain_hook/review/settings.py @@ -6,7 +6,7 @@ from cc_transcript.mining.confidence import MEDIUM, VERY_HIGH, Confidence from pydantic import Field -from pydantic_settings import SettingsConfigDict +from pydantic_settings import BaseSettings, SettingsConfigDict from spawnllm import TModel from captain_hook.settings import HooksSettings, resolve_state_dir @@ -16,6 +16,19 @@ def resolve_review_db_path() -> Path: return resolve_state_dir() / "review" / "review-v1.db" +class ReviewPaths(BaseSettings): + """The reviewer's database location alone, read from ``HOOKS_REVIEW_DB_PATH``. + + The hook guard resolves its admission locks from this without validating the full + :class:`ReviewSettings`, so a malformed reviewer knob still reaches the detached child, + which records the failure. + """ + + model_config = SettingsConfigDict(env_prefix="HOOKS_REVIEW_") + + db_path: Path = Field(default_factory=resolve_review_db_path) + + class ReviewSettings(HooksSettings): """Session-reviewer settings, backed by environment variables with ``HOOKS_REVIEW_`` prefix. diff --git a/captain_hook/snapshots/review.py b/captain_hook/snapshots/review.py index 7b66a498..0f3b0539 100644 --- a/captain_hook/snapshots/review.py +++ b/captain_hook/snapshots/review.py @@ -10,7 +10,7 @@ from dataclasses import asdict, dataclass, replace from datetime import datetime from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any from cc_transcript.context import ContextWindow from cc_transcript.corrections import Correction @@ -22,6 +22,9 @@ from captain_hook.snapshots.client import MAX_RESULT_BYTES, EvidenceIncomplete from captain_hook.snapshots.client import client_scope as review_client +if TYPE_CHECKING: + from captain_hook.review.scan import CorrectionLedger + REVIEW_POLICY = {"id": "captain-review", "version": "1"} @@ -382,30 +385,29 @@ async def prepare_corrections(snapshot: Any, request: Mapping[str, Any]) -> dict return {"kind": "corrections", "corrections": drafts} -async def record_correction_drafts(drafts: Sequence[Mapping[str, Any]]) -> None: - from cc_transcript.corrections import CorrectionLog +async def record_correction_drafts(drafts: Sequence[Mapping[str, Any]], *, ledger: CorrectionLedger) -> None: from cc_transcript.extract.correct import CorrectionPick, usable_backend from spawnllm import extract if not drafts: return backend = usable_backend() - async with await CorrectionLog.open() as log: - for draft in drafts: - anchor = draft["anchor"] - if await log.for_anchor(anchor["session_id"], anchor["event_uuid"]): - continue - choices = draft["choices"] - if not choices: + log = await ledger.handle() + for draft in drafts: + anchor = draft["anchor"] + if await log.for_anchor(anchor["session_id"], anchor["event_uuid"]): + continue + choices = draft["choices"] + if not choices: + continue + if backend is None: + chosen = max(choices, key=lambda choice: choice["overlap"]) + else: + pick = await extract(draft["prompt"], CorrectionPick, backend=backend, model="medium") + if pick.candidate is None or not 1 <= pick.candidate <= len(choices): continue - if backend is None: - chosen = max(choices, key=lambda choice: choice["overlap"]) - else: - pick = await extract(draft["prompt"], CorrectionPick, backend=backend, model="medium") - if pick.candidate is None or not 1 <= pick.candidate <= len(choices): - continue - chosen = choices[pick.candidate - 1] - await log.append(Correction(**json.loads(chosen["correction_json"]))) + chosen = choices[pick.candidate - 1] + await log.append(Correction(**json.loads(chosen["correction_json"]))) @dataclass(frozen=True, slots=True) diff --git a/tests/snapshot_review_helpers.py b/tests/snapshot_review_helpers.py index 843a0c56..923bd3e7 100644 --- a/tests/snapshot_review_helpers.py +++ b/tests/snapshot_review_helpers.py @@ -204,6 +204,7 @@ def forbidden_bridge(*args, **kwargs): lambda paths: sorted({str(Path(path).parent) for path in paths if Path(path).is_absolute()}), ) monkeypatch.setattr("captain_hook.snapshots.review.record_correction_drafts", AsyncMock()) + monkeypatch.setattr("captain_hook.review.scan.CorrectionLedger.recorded", AsyncMock(return_value=set())) try: yield owner finally: diff --git a/tests/test_review_pipeline.py b/tests/test_review_pipeline.py index 49d74b1c..89f14bc8 100644 --- a/tests/test_review_pipeline.py +++ b/tests/test_review_pipeline.py @@ -52,6 +52,7 @@ guard_and_spawn, guard_and_sweep, judge_lock, + pending_claim, queued_review_lock, repo_lock, review_log_path, @@ -305,7 +306,8 @@ def test_spawns_detached_child_with_log_and_marker_env( json.dumps({"transcript_path": str(transcript), "cwd": str(tmp_path), "reason": "other"}).encode() ) [(argv, kwargs)] = popen_calls - assert argv == spawn_argv(str(transcript), str(tmp_path)) + [pending_fd] = kwargs["pass_fds"] + assert argv == spawn_argv(str(transcript), str(tmp_path), pending_fd=pending_fd) assert argv[:6] == [sys.executable, "-P", "-m", "captain_hook", "review", "spawn"] assert kwargs["start_new_session"] is True assert kwargs["env"][SPAWNED_ENV] == "1" @@ -639,6 +641,93 @@ def test_gated_sweep_skip_does_not_burn_the_stamp( assert len(popen_calls) == 1 +class TestPendingAdmission: + def payload(self, tmp_path: Path) -> dict[str, object]: + transcript = write_transcript(tmp_path / "s.jsonl", correction_entries()) + return {"transcript_path": str(transcript), "cwd": str(tmp_path), "hook_event_name": "SessionStart"} + + def test_session_starts_with_claimed_pending_work_spawn_one_child( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + from captain_hook.review import pipeline + + children: list[int] = [] + probes: list[str | None] = [] + + def spawn(argv: list[str], **kwargs: Any) -> SimpleNamespace: + [pending_fd] = kwargs["pass_fds"] + assert argv[-2:] == ["--pending-fd", str(pending_fd)] + children.append(os.dup(pending_fd)) + return SimpleNamespace(pid=4242) + + def probe(cwd: str | None) -> bool: + probes.append(cwd) + return True + + monkeypatch.setattr(pipeline.subprocess, "Popen", spawn) + monkeypatch.setattr(pipeline, "enrolled", probe) + payload = self.payload(tmp_path) + try: + dispatch_review("SessionStart", payload) + dispatch_review("SessionStart", payload) + assert len(children) == 1 + assert probes == [str(tmp_path)] + assert "review-run skip: a queued pass already covers" in review_log_path().read_text() + os.close(children[0]) + dispatch_review("SessionStart", payload) + dispatch_review("SessionEnd", payload) + assert len(children) == 2 + assert probes == [str(tmp_path), str(tmp_path)] + finally: + for child in children[1:]: + os.close(child) + + def test_a_failed_detach_releases_the_pending_claim(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from captain_hook.review import pipeline + + attempts: list[list[str]] = [] + + def refuse(argv: list[str], **kwargs: Any) -> None: + attempts.append(argv) + raise OSError("fork refused") + + monkeypatch.setattr(pipeline.subprocess, "Popen", refuse) + payload = self.payload(tmp_path) + dispatch_review("SessionStart", payload) + dispatch_review("SessionStart", payload) + assert len(attempts) == 2 + + def test_child_holds_the_handed_down_claim_until_it_takes_the_repo_lock(self, tmp_path: Path) -> None: + settings = ReviewSettings(db_path=tmp_path / "review.db") + transcript = tmp_path / "projects" / "s1.jsonl" + with pending_claim(settings.db_path, transcript) as parent_fd: + assert parent_fd is not None + child_fd = os.dup(parent_fd) + with pending_claim(settings.db_path, transcript) as rival: + assert rival is None + with queued_review_lock(settings, str(tmp_path), transcript, pending_fd=child_fd) as claimed: + assert claimed + with pending_claim(settings.db_path, transcript) as follow_up: + assert follow_up is not None + + def test_an_active_child_admits_the_next_event_while_the_parent_still_holds_its_copy(self, tmp_path: Path) -> None: + settings = ReviewSettings(db_path=tmp_path / "review.db") + transcript = tmp_path / "projects" / "s1.jsonl" + with pending_claim(settings.db_path, transcript) as parent_fd: + assert parent_fd is not None + with queued_review_lock(settings, str(tmp_path), transcript, pending_fd=os.dup(parent_fd)) as claimed: + assert claimed + with pending_claim(settings.db_path, transcript) as follow_up: + assert follow_up is not None + + def test_a_malformed_reviewer_knob_still_spawns_the_recording_child( + self, popen_calls: list[tuple[list[str], dict[str, Any]]], tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("HOOKS_REVIEW_JUDGE_CONCURRENCY", "not-a-number") + guard_and_spawn(json.dumps(self.payload(tmp_path)).encode()) + assert len(popen_calls) == 1 + + class TestDispatchReview: def test_routes_stop_to_sweep(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[tuple[str, bool]] = [] diff --git a/tests/test_snapshot_review_consumers.py b/tests/test_snapshot_review_consumers.py index 97b65c33..b52542a3 100644 --- a/tests/test_snapshot_review_consumers.py +++ b/tests/test_snapshot_review_consumers.py @@ -17,7 +17,7 @@ from cc_transcript.mining.candidates import dedup_key from captain_hook.review.judge import CONTEXT_BUDGET, TRIGGER_BUDGET, prompt_builder -from captain_hook.review.scan import ScanReport, ingest, prepare_source, scan +from captain_hook.review.scan import CorrectionLedger, ScanReport, ingest, prepare_source, scan from captain_hook.snapshots.client import CURRENT_CLIENT, EvidenceIncomplete, SnapshotClient from captain_hook.snapshots.review import ( REVIEW_POLICY, @@ -44,6 +44,15 @@ HOOK_COMPLAINT_TEXT = "**Note**: The task tracker reminder re-fired on a sequence I already completed - ignoring it." +def nothing_recorded(anchors) -> set[EventRef]: + return set() + + +async def recorded_in_ledger(anchors) -> set[EventRef]: + async with CorrectionLedger() as ledger: + return await ledger.recorded(anchors) + + def window(session: str = "session") -> ContextWindow: return ContextWindow( anchor=EventRef(SessionId(session), EventUuid("anchor")), @@ -329,7 +338,7 @@ def pages(*args, **kwargs): client = SimpleNamespace(acquire=lambda path: session, pages=pages) with pytest.raises(EvidenceIncomplete): - prepare_source(client, Path("/unused"), settings) + prepare_source(client, Path("/unused"), settings, recorded=nothing_recorded) assert released == [True] @@ -348,7 +357,7 @@ def pages(self, operation, **kwargs): monkeypatch.setattr( scan_module, "prepare_source", - lambda *args: (_ for _ in ()).throw(EvidenceIncomplete("deadline", "bounded")), + lambda *args, **kwargs: (_ for _ in ()).throw(EvidenceIncomplete("deadline", "bounded")), ) try: with pytest.raises(EvidenceIncomplete): @@ -367,7 +376,8 @@ async def test_ingest_records_empty_source_atomically(store): "repo_key": REPO, "candidates_json": [], } - assert await ingest(store, prepared, corrections=[]) == ScanReport(1, 0) + async with CorrectionLedger() as ledger: + assert await ingest(store, prepared, corrections=[], ledger=ledger) == ScanReport(1, 0) assert await store.file_mtimes() == {"/fixture/source.jsonl": 1.23} @@ -385,7 +395,7 @@ async def test_ingest_failure_rolls_back_source_and_feedback(store, monkeypatch) } monkeypatch.setattr(store, "record_observation", AsyncMock(side_effect=RuntimeError("write failed"))) with pytest.raises(RuntimeError, match="write failed"): - await ingest(store, prepared, corrections=[]) + await ingest(store, prepared, corrections=[], ledger=CorrectionLedger()) assert await store.file_mtimes() == {} assert await store.db.sql("SELECT dedup_key FROM feedback_events") == [] @@ -443,10 +453,11 @@ async def choose(prompt, schema, **kwargs): assert CORRECTION in prompt return CorrectionPick(candidate=1, note="fixture") - monkeypatch.setattr("cc_transcript.corrections.CorrectionLog.open", AsyncMock(return_value=Log())) monkeypatch.setattr("cc_transcript.extract.correct.usable_backend", lambda: object()) monkeypatch.setattr("spawnllm.extract", choose) - await record_correction_drafts(prepared["corrections"]) + await record_correction_drafts( + prepared["corrections"], ledger=SimpleNamespace(handle=AsyncMock(return_value=Log())) + ) assert len(recorded) == 1 assert recorded[0].incorrect_old == old assert recorded[0].incorrect_new == new @@ -608,7 +619,9 @@ def review_comments(count: int) -> list[dict]: def prepared_anchors(path: Path, settings) -> tuple[Counter[str], Counter[str], list[tuple[EventRef, Any]]]: - prepared, corrections, refused = prepare_source(CURRENT_CLIENT.get(), path, settings, REPO) + prepared, corrections, refused = prepare_source( + CURRENT_CLIENT.get(), path, settings, REPO, recorded=nothing_recorded + ) eligible = Counter( candidate.ref.event_uuid for raw in prepared["candidates_json"] @@ -617,6 +630,98 @@ def prepared_anchors(path: Path, settings) -> tuple[Counter[str], Counter[str], return eligible, Counter(draft["anchor"]["event_uuid"] for draft in corrections), refused +@pytest.fixture +def git_calls(monkeypatch) -> list[tuple[str, ...]]: + calls: list[tuple[str, ...]] = [] + + class CountingGit: + def __init__(self, snapshot, *, max_bytes, max_processes=84): + self.snapshot = snapshot + + def __call__(self, repo, *arguments): + calls.append(arguments) + + monkeypatch.setattr("captain_hook.snapshots.review.BoundedGit", CountingGit) + return calls + + +@pytest.fixture +def prepared_batches(monkeypatch) -> list[list[str]]: + from captain_hook.review import scan as scan_module + + batches: list[list[str]] = [] + split = scan_module.correction_pages + + def counted(client, session, batch, repo): + batches.append([candidate.ref.event_uuid for candidate in batch]) + return split(client, session, batch, repo) + + monkeypatch.setattr(scan_module, "correction_pages", counted) + return batches + + +def anchor_uuid(edit: list[dict]) -> str: + return edit[-1]["uuid"] + + +@pytest.mark.usefixtures(native_review_owner.__name__) +async def test_rescan_prepares_only_anchors_without_a_recorded_correction( + tmp_path, store, settings, prepared_batches, git_calls +): + edits = [corrected_edit(f"module{index}") for index in range(10)] + path = write_transcript(tmp_path / "s.jsonl", [entry for edit in edits[:8] for entry in edit]) + await scan(store, settings=settings, transcripts=[tmp_path], repo_key=REPO) + assert sorted(uuid for batch in prepared_batches for uuid in batch) == sorted(map(anchor_uuid, edits[:8])) + prepared_batches.clear() + write_transcript(path, [entry for edit in edits for entry in edit]) + assert (await scan(store, settings=settings, transcripts=[tmp_path], repo_key=REPO)).scanned == 1 + assert prepared_batches == [[anchor_uuid(edit) for edit in edits[8:]]] + + +@pytest.mark.usefixtures(native_review_owner.__name__) +async def test_rescan_of_an_unchanged_transcript_prepares_nothing_and_runs_no_git( + tmp_path, store, settings, prepared_batches, git_calls +): + write_transcript(tmp_path / "s.jsonl", [entry for label in ("parser", "lexer") for entry in corrected_edit(label)]) + await scan(store, settings=settings, transcripts=[tmp_path], repo_key=REPO) + assert len(prepared_batches) == 1 + prepared_batches.clear() + git_before = len(git_calls) + assert await scan(store, settings=settings, transcripts=[tmp_path], repo_key=REPO) == ScanReport(0, 0, 0) + assert (prepared_batches, len(git_calls)) == ([], git_before) + + +@pytest.mark.usefixtures(native_review_owner.__name__) +def test_a_correction_recorded_under_another_anchor_ref_leaves_the_anchor_prepared( + tmp_path, settings, prepared_batches +): + import asyncio + from dataclasses import replace + + from cc_transcript.corrections import Correction, CorrectionLog + + parser, lexer = corrected_edit("parser"), corrected_edit("lexer") + path = write_transcript(tmp_path / "s.jsonl", [*parser, *lexer]) + _, corrections, _ = prepare_source(CURRENT_CLIENT.get(), path, settings, REPO, recorded=nothing_recorded) + rows = { + draft["anchor"]["event_uuid"]: Correction(**json.loads(draft["choices"][0]["correction_json"])) + for draft in corrections + } + + async def record() -> None: + async with await CorrectionLog.open() as log: + await log.append(rows[anchor_uuid(parser)]) + await log.append(replace(rows[anchor_uuid(lexer)], anchor_uuid=EventUuid("superseded-anchor"))) + + asyncio.run(record()) + prepared_batches.clear() + _, corrections, _ = prepare_source( + CURRENT_CLIENT.get(), path, settings, REPO, recorded=lambda anchors: asyncio.run(recorded_in_ledger(anchors)) + ) + assert prepared_batches == [[anchor_uuid(lexer)]] + assert [draft["anchor"]["event_uuid"] for draft in corrections] == [anchor_uuid(lexer)] + + @pytest.fixture def event_reads(monkeypatch) -> Counter[int]: from cc_transcript.snapshots import _Events @@ -750,7 +855,7 @@ def test_review_preparation_walks_a_hook_complaint_session_under_the_output_boun assistant_text(HOOK_COMPLAINT_TEXT), ] prepared, corrections, refused = prepare_source( - CURRENT_CLIENT.get(), write_transcript(tmp_path / "s.jsonl", entries), settings, REPO + CURRENT_CLIENT.get(), write_transcript(tmp_path / "s.jsonl", entries), settings, REPO, recorded=nothing_recorded ) assert prepared["disposition"] == "eligible" assert (corrections, refused) == ([], []) @@ -760,7 +865,11 @@ def test_review_preparation_walks_a_hook_complaint_session_under_the_output_boun @pytest.mark.usefixtures(native_review_owner.__name__) def test_review_preparation_reads_a_multi_comment_message_once(tmp_path, settings, event_reads): prepared, _, refused = prepare_source( - CURRENT_CLIENT.get(), write_transcript(tmp_path / "s.jsonl", review_comments(60)), settings, REPO + CURRENT_CLIENT.get(), + write_transcript(tmp_path / "s.jsonl", review_comments(60)), + settings, + REPO, + recorded=nothing_recorded, ) kinds = Counter(decode_candidate(raw)[0].source_kind for raw in prepared["candidates_json"]) assert (kinds["review_comment"], refused) == (60, []) @@ -771,7 +880,11 @@ def test_review_preparation_reads_a_multi_comment_message_once(tmp_path, setting def test_review_preparation_refuses_a_candidate_over_the_owned_page_by_size(tmp_path, settings): with pytest.raises(EvidenceIncomplete) as refused: prepare_source( - CURRENT_CLIENT.get(), write_transcript(tmp_path / "s.jsonl", review_comments(120)), settings, REPO + CURRENT_CLIENT.get(), + write_transcript(tmp_path / "s.jsonl", review_comments(120)), + settings, + REPO, + recorded=nothing_recorded, ) assert refused.value.status == "output_limit" assert re.fullmatch( From 2e1d2a3b93edbd0740814aab09abc52c717e85b0 Mon Sep 17 00:00:00 2001 From: Yasyf Mohamedali Date: Fri, 2 Oct 2026 20:06:33 -0700 Subject: [PATCH 2/2] =?UTF-8?q?captain-hook:=20=F0=9F=A7=AA=20Isolate=20de?= =?UTF-8?q?tach=20failure=20test=20from=20enrollment=20probes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Context: The detach admission regression patched process-global Popen, so enrollment consumed the stub before either detach. Summary: Freeze enrollment and assert both attempts are detach calls with failure breadcrumbs. Motivation: Verify claim release through the intended failure path across the Linux and macOS CI shards. Details: Add an enrolled stub, check pending-fd argv, and count the two detach-failed log entries. --- tests/test_review_pipeline.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_review_pipeline.py b/tests/test_review_pipeline.py index 89f14bc8..1a32e6c9 100644 --- a/tests/test_review_pipeline.py +++ b/tests/test_review_pipeline.py @@ -692,10 +692,12 @@ def refuse(argv: list[str], **kwargs: Any) -> None: raise OSError("fork refused") monkeypatch.setattr(pipeline.subprocess, "Popen", refuse) + monkeypatch.setattr(pipeline, "enrolled", lambda cwd: True) payload = self.payload(tmp_path) dispatch_review("SessionStart", payload) dispatch_review("SessionStart", payload) - assert len(attempts) == 2 + assert [argv[-2] for argv in attempts] == ["--pending-fd", "--pending-fd"] + assert review_log_path().read_text().count("detach failed") == 2 def test_child_holds_the_handed_down_claim_until_it_takes_the_repo_lock(self, tmp_path: Path) -> None: settings = ReviewSettings(db_path=tmp_path / "review.db")