From 96ff8871a52cd0601af08efcf3c2c084e0d6e68e Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Wed, 23 Sep 2026 21:59:28 +0800 Subject: [PATCH] llm-judge: record every judge stage, and report jobs that never reach the person The runner's per-run table counted wrong-check-reflect as "silent 2,582, spoke 0" on Claude. That number cannot tell a mute hook from a working one: the Stop hook never prints, it queues a background judge job, and the verdict arrives on a later PostToolUse. The live cache showed the real losses were downstream: 39 hit verdicts for this hook sat undelivered, 25 of them on jobs queued with an empty transcript path, which no drain can ever find. Each stage now writes one catstack.hook_event.v1 row with a reason: - judge_skipped (wrong-check-reflect): stop_hook_active, gate_off, empty_reply, already_prompted, user_asked_reflect, judge_child, bad_payload - judge_queued (llm-judge, every hook): transcript or no_transcript - judge_finished (llm-judge): hit, clean or unchecked Delivery was already recorded by drain. Rows join on the job id. A job queued with no transcript also prints catstack-hook-error, so the runner classifies the run as caught_error and hook-health surfaces it on the next prompt instead of the verdict vanishing. report.py --judge prints the per-hook funnel plus leak counts (no_transcript, stuck, undelivered, undelivered_hits) older than --grace (default 1h); --check exits 1 on any leak. Stage rows carry no rule_id, so the existing rule table ignores them. Co-Authored-By: Claude Opus 5.5 (1M context) Change-Id: Ic6eeb8c30dabc90c4d11270e98ef4d1a2eecddfd --- engine/hooks/_runner/report.py | 114 ++++++++++++++++++ engine/hooks/_runner/tests/test_report.py | 57 +++++++++ engine/hooks/_sdk/events.py | 15 +++ engine/hooks/_sdk/tests/test_events.py | 18 ++- engine/hooks/llm-judge/judge.py | 41 ++++++- engine/hooks/llm-judge/tests/test_judge.py | 33 +++++ .../wrong-check-reflect/claude_stop_check.py | 2 +- .../hooks/wrong-check-reflect/codex_notify.py | 2 +- .../wrong-check-reflect/cursor_session.py | 2 +- engine/hooks/wrong-check-reflect/detect.py | 44 +++++-- .../wrong-check-reflect/tests/test_hooks.py | 40 ++++++ 11 files changed, 352 insertions(+), 16 deletions(-) diff --git a/engine/hooks/_runner/report.py b/engine/hooks/_runner/report.py index afc5504f..71c99f33 100644 --- a/engine/hooks/_runner/report.py +++ b/engine/hooks/_runner/report.py @@ -438,6 +438,101 @@ def format_event_table(report: dict[str, Any]) -> str: return "\n".join(lines) + "\n" +JUDGE_STAGES = ("judge_skipped", "judge_queued", "judge_finished") +VERDICTS = ("hit", "clean", "unchecked") +LEAKS = ("no_transcript", "stuck", "undelivered", "undelivered_hits") + + +def is_delivery(row: dict[str, Any]) -> bool: + return row.get("harness") == "judge" and row.get("mode_source") == "judge" and row.get("action") in VERDICTS + + +def _judge_summary(hook: str) -> dict[str, Any]: + return { + "hook": hook, + "skipped": {}, + "queued": 0, + "finished": {verdict: 0 for verdict in VERDICTS}, + "delivered": {verdict: 0 for verdict in VERDICTS}, + **{leak: 0 for leak in LEAKS}, + } + + +def build_judge_report( + rows: list[dict[str, Any]], + malformed: int, + warnings: list[str], + now: datetime, + grace: timedelta, +) -> dict[str, Any]: + summaries: dict[str, dict[str, Any]] = {} + jobs: dict[str, dict[str, Any]] = {} + for row in rows: + action = row.get("action") + if action not in JUDGE_STAGES and not is_delivery(row): + continue + hook = str(row.get("hook") or "") + summary = summaries.setdefault(hook, _judge_summary(hook)) + reason = str(row.get("reason") or "") + ts = parse_ts(row.get("ts")) + job = jobs.setdefault(str(row.get("finding_id") or ""), {"hook": hook}) + if action == "judge_skipped": + summary["skipped"][reason] = summary["skipped"].get(reason, 0) + 1 + elif action == "judge_queued": + summary["queued"] += 1 + if reason == "no_transcript": + summary["no_transcript"] += 1 + job["queued"] = ts + elif action == "judge_finished": + verdict = reason if reason in VERDICTS else "unchecked" + summary["finished"][verdict] += 1 + job["finished"] = ts + job["verdict"] = verdict + else: + summary["delivered"][str(action)] += 1 + job["delivered"] = ts + for job in jobs.values(): + summary = summaries[job["hook"]] + queued, finished = job.get("queued"), job.get("finished") + if queued is not None and finished is None and now - queued > grace: + summary["stuck"] += 1 + if finished is not None and "delivered" not in job and now - finished > grace: + summary["undelivered"] += 1 + if job.get("verdict") == "hit": + summary["undelivered_hits"] += 1 + ordered = [summaries[hook] for hook in sorted(summaries)] + return { + "window_rows": len(rows), + "malformed_rows": malformed, + "warnings": warnings, + "grace_seconds": int(grace.total_seconds()), + "hooks": ordered, + "leaks": sum(summary[leak] for summary in ordered for leak in ("no_transcript", "stuck", "undelivered")), + } + + +def _counts(values: dict[str, int]) -> str: + return ",".join(f"{name}={count}" for name, count in sorted(values.items())) or "-" + + +def format_judge_table(report: dict[str, Any]) -> str: + lines = list(report["warnings"]) + if report["malformed_rows"]: + lines.append(f"skipped {report['malformed_rows']} malformed event row(s)") + lines.append("hook skipped queued finished delivered " + " ".join(LEAKS)) + for row in report["hooks"]: + lines.append( + f"{row['hook']} {_counts(row['skipped'])} {row['queued']} {_counts(row['finished'])} " + f"{_counts(row['delivered'])} " + " ".join(str(row[leak]) for leak in LEAKS) + ) + if report["leaks"]: + lines.append( + f"LEAK: {report['leaks']} judge job(s) queued with no transcript, stuck, or never delivered " + f"after {report['grace_seconds']}s" + ) + return "\n".join(lines) + "\n" + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--since", default="7d") @@ -445,12 +540,31 @@ def main(argv: list[str] | None = None) -> int: mode = parser.add_mutually_exclusive_group() mode.add_argument("--events", action="store_true", default=True) mode.add_argument("--runs", action="store_true") + mode.add_argument("--judge", action="store_true") + parser.add_argument("--grace", default="1h") + parser.add_argument("--check", action="store_true") args = parser.parse_args(argv) try: since = parse_since(args.since) + grace = parse_since(args.grace) except ValueError as exc: print(str(exc), file=sys.stderr) return 2 + if args.judge: + now = datetime.now(timezone.utc) + rows, malformed, warnings = read_event_rows(metrics_dir(), now - since) + if rows is None: + for warning in warnings: + print(warning) + return 2 + report = build_judge_report(rows, malformed, warnings, now, grace) + if args.json: + print(json.dumps(report, sort_keys=True)) + else: + print(format_judge_table(report), end="") + if warnings: + return 2 + return 1 if args.check and report["leaks"] else 0 if not args.runs: try: registry_data = registry.load_registry() diff --git a/engine/hooks/_runner/tests/test_report.py b/engine/hooks/_runner/tests/test_report.py index 71964d0b..43baa123 100644 --- a/engine/hooks/_runner/tests/test_report.py +++ b/engine/hooks/_runner/tests/test_report.py @@ -186,6 +186,63 @@ def write_events(self, rows: list[dict[str, object]], malformed: bool = False) - handle.write("{bad\n") return path + def stage(self, action: str, reason: str, job: str, hours_ago: float = 3, hook: str = "wrong-check-reflect") -> dict[str, object]: + ts = datetime.now(timezone.utc) - timedelta(hours=hours_ago) + return {"schema": "catstack.hook_event.v1", "ts": ts.isoformat(), "harness": "claude", "hook": hook, + "rule_id": "", "mode_source": "stage", "action": action, "reason": reason, "finding_id": job} + + def delivered(self, verdict: str, job: str, hook: str = "wrong-check-reflect") -> dict[str, object]: + return {"schema": "catstack.hook_event.v1", "ts": datetime.now(timezone.utc).isoformat(), "harness": "judge", + "hook": hook, "rule_id": hook, "mode_source": "judge", "action": verdict, "finding_id": job} + + def test_judge_report_counts_every_stage_and_each_leak(self) -> None: + self.write_events( + [ + self.stage("judge_skipped", "already_prompted", "s1"), + self.stage("judge_skipped", "already_prompted", "s2"), + self.stage("judge_skipped", "gate_off", "s3"), + self.stage("judge_queued", "transcript", "delivered-hit"), + self.stage("judge_finished", "hit", "delivered-hit"), + self.delivered("hit", "delivered-hit"), + self.stage("judge_queued", "transcript", "lost-hit"), + self.stage("judge_finished", "hit", "lost-hit"), + self.stage("judge_queued", "no_transcript", "no-path"), + self.stage("judge_finished", "clean", "no-path"), + self.stage("judge_queued", "transcript", "hung"), + self.stage("judge_queued", "transcript", "fresh", hours_ago=0.1), + ] + ) + result = self.run_report("--judge", "--json", "--since", "1d") + self.assertEqual(result.returncode, 0, result.stderr) + [row] = json.loads(result.stdout)["hooks"] + self.assertEqual(row["skipped"], {"already_prompted": 2, "gate_off": 1}) + self.assertEqual(row["queued"], 5) + self.assertEqual(row["finished"], {"hit": 2, "clean": 1, "unchecked": 0}) + self.assertEqual(row["delivered"], {"hit": 1, "clean": 0, "unchecked": 0}) + self.assertEqual( + {key: row[key] for key in ("no_transcript", "stuck", "undelivered", "undelivered_hits")}, + {"no_transcript": 1, "stuck": 1, "undelivered": 2, "undelivered_hits": 1}, + ) + + def test_judge_check_fails_on_a_leak_and_passes_when_every_verdict_is_delivered(self) -> None: + self.write_events([self.stage("judge_queued", "transcript", "j"), self.stage("judge_finished", "hit", "j")]) + leaked = self.run_report("--judge", "--check", "--since", "1d") + self.assertEqual(leaked.returncode, 1, leaked.stdout + leaked.stderr) + self.assertIn("LEAK: 1 judge job(s)", leaked.stdout) + self.write_events( + [self.stage("judge_queued", "transcript", "j"), self.stage("judge_finished", "hit", "j"), self.delivered("hit", "j")] + ) + clean = self.run_report("--judge", "--check", "--since", "1d") + self.assertEqual(clean.returncode, 0, clean.stdout + clean.stderr) + self.assertNotIn("LEAK", clean.stdout) + + def test_judge_rows_stay_out_of_the_rule_table(self) -> None: + self.seed_configs() + self.write_events([self.stage("judge_skipped", "gate_off", "s1")]) + result = self.run_report("--since", "1d") + self.assertNotIn("judge_skipped", result.stdout) + self.assertNotIn("wrong-check-reflect", result.stdout) + def test_seeded_rows_include_no_record_and_unregistered(self) -> None: self.seed_configs() self.write_rows( diff --git a/engine/hooks/_sdk/events.py b/engine/hooks/_sdk/events.py index 5e88bf00..8ee1a717 100644 --- a/engine/hooks/_sdk/events.py +++ b/engine/hooks/_sdk/events.py @@ -65,6 +65,21 @@ def once_per_session_or_compaction(hook: str, event: dict[str, object]) -> bool: return True +def write_stage_event( + hook: str, + harness: str, + session_id: str, + action: str, + reason: str, + finding_id: str | None = None, + stderr: TextIO | None = None, +) -> bool: + err = stderr if stderr is not None else sys.stderr + row = _row(hook, harness, {"session_id": session_id}, None, "", "stage", action, 0, finding_id) + row["reason"] = reason + return _append_rows(hook, [row], err) + + def prune_old_event_files(days: int = 30, stderr: TextIO | None = None) -> None: err = stderr if stderr is not None else sys.stderr root = _metrics_dir() diff --git a/engine/hooks/_sdk/tests/test_events.py b/engine/hooks/_sdk/tests/test_events.py index 38254f62..a994d5ec 100644 --- a/engine/hooks/_sdk/tests/test_events.py +++ b/engine/hooks/_sdk/tests/test_events.py @@ -14,7 +14,7 @@ sys.path.insert(0, str(SDK_DIR)) import runtime -from events import is_human_prompt, once_per_session_or_compaction, write_events +from events import is_human_prompt, once_per_session_or_compaction, write_events, write_stage_event from finding import Finding from modes import effective_mode @@ -177,6 +177,22 @@ def _stdio(self, stdin_text: str): sys.stdout = old_stdout sys.stderr = old_stderr + def test_stage_event_carries_its_reason_and_no_rule_id(self) -> None: + with tempfile.TemporaryDirectory() as tmp, mock.patch.dict( + os.environ, {"CATSTACK_HOOK_METRICS_DIR": tmp}, clear=False + ): + self.assertTrue( + write_stage_event("demo", "claude", "session-1", "judge_skipped", "already_prompted", "job-1") + ) + rows = self._rows(tmp) + + self.assertEqual(1, len(rows)) + self.assertEqual("judge_skipped", rows[0]["action"]) + self.assertEqual("already_prompted", rows[0]["reason"]) + self.assertEqual("stage", rows[0]["mode_source"]) + self.assertEqual("job-1", rows[0]["finding_id"]) + self.assertEqual("", rows[0]["rule_id"]) + def _rows(self, directory: str) -> list[dict[str, object]]: files = list(Path(directory).glob("events-*.jsonl")) self.assertEqual(1, len(files)) diff --git a/engine/hooks/llm-judge/judge.py b/engine/hooks/llm-judge/judge.py index de894513..1a50cc74 100644 --- a/engine/hooks/llm-judge/judge.py +++ b/engine/hooks/llm-judge/judge.py @@ -17,7 +17,7 @@ SDK_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk") sys.path.insert(0, SDK_DIR) -from events import write_events # noqa: E402 +from events import write_events, write_stage_event # noqa: E402 from finding import Finding # noqa: E402 TIMEOUT_SECONDS = 60 @@ -345,9 +345,47 @@ def enqueue(job: dict) -> str | None: stderr=log_handle, cwd=root, ) + record_queued(job) return job_id +def job_harness(job: dict) -> str: + harness = job.get("harness") + return harness if isinstance(harness, str) and harness else "judge" + + +def job_hook(job: dict) -> str: + hook = job.get("hook") + return hook if isinstance(hook, str) and hook else "llm-judge" + + +def record_queued(job: dict) -> None: + transcript = str(job.get("transcript") or "") + reason = "transcript" if transcript else "no_transcript" + write_stage_event(job_hook(job), job_harness(job), transcript, "judge_queued", reason, job["id"]) + if not transcript: + print( + f"catstack-hook-error {job_hook(job)}: judge job {job['id']} was queued with no transcript " + f"path, so its verdict can never be delivered", + file=sys.stderr, + ) + + +def record_finished(job: dict, result: dict) -> None: + errors = io.StringIO() + written = write_stage_event( + job_hook(job), + job_harness(job), + str(job.get("transcript") or ""), + "judge_finished", + str(result.get("outcome") or "unchecked"), + str(job.get("id") or "unknown"), + stderr=errors, + ) + if not written: + log(f"job {job.get('id')}: finished event not written: {errors.getvalue().strip()}") + + def run_job(path: str) -> dict: stem = os.path.splitext(os.path.basename(path))[0] job: dict = {"id": stem} @@ -365,6 +403,7 @@ def run_job(path: str) -> dict: result = verdict(job, {"outcome": "unchecked", "attempts": []}) result["reason"] = clip(f"judge error {type(exc).__name__}", str(exc)) write_json_atomic(os.path.join(verdict_dir(str(job.get("transcript") or "")), f"{stem}.json"), result) + record_finished(job, result) try: os.remove(path) except OSError as exc: diff --git a/engine/hooks/llm-judge/tests/test_judge.py b/engine/hooks/llm-judge/tests/test_judge.py index 33cce814..5d083d3d 100644 --- a/engine/hooks/llm-judge/tests/test_judge.py +++ b/engine/hooks/llm-judge/tests/test_judge.py @@ -3,6 +3,7 @@ Run: python3 -m unittest discover -s engine/hooks/llm-judge/tests -v """ +import io import json import os import sys @@ -10,6 +11,7 @@ import time import unittest import warnings +from contextlib import redirect_stderr from unittest.mock import patch LIB_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -407,6 +409,37 @@ def test_run_job_writes_verdict_under_transcript_hash_and_deletes_job(self): with open(written, encoding="utf-8") as handle: self.assertEqual(json.load(handle)["outcome"], "hit") + def stage_rows(self): + folder = os.path.join(self.state.name, "metrics") + rows = [] + for name in sorted(os.listdir(folder)) if os.path.isdir(folder) else []: + if name.startswith("events-"): + with open(os.path.join(folder, name), encoding="utf-8") as handle: + rows.extend(json.loads(line) for line in handle) + return [row for row in rows if row.get("mode_source") == "stage"] + + def test_enqueue_records_a_queued_stage_event_keyed_by_job_id(self): + with patch.object(judge.subprocess, "Popen"): + judge.enqueue(self.job(id="queued-job", harness="claude")) + rows = self.stage_rows() + self.assertEqual([(r["action"], r["reason"], r["finding_id"], r["harness"]) for r in rows], + [("judge_queued", "transcript", "queued-job", "claude")]) + + def test_enqueue_with_no_transcript_is_recorded_and_reported_as_an_error(self): + err = io.StringIO() + with patch.object(judge.subprocess, "Popen"), redirect_stderr(err): + judge.enqueue(self.job(id="lost-job", transcript="")) + self.assertEqual([(r["action"], r["reason"]) for r in self.stage_rows()], [("judge_queued", "no_transcript")]) + self.assertTrue(err.getvalue().startswith("catstack-hook-error demo-hook: judge job lost-job")) + + def test_run_job_records_a_finished_stage_event_with_the_verdict(self): + self.use_runners(ANSWER_MATCH) + path = os.path.join(self.state.name, "jobs", "job-1.json") + judge.write_json_atomic(path, self.job()) + judge.run_job(path) + self.assertEqual([(r["action"], r["reason"], r["finding_id"]) for r in self.stage_rows()], + [("judge_finished", "hit", "job-1")]) + def test_run_job_error_is_logged_and_still_writes_unchecked_verdict(self): os.environ[judge.RUNNERS_ENV] = "not json" path = os.path.join(self.state.name, "jobs", "broken-job.json") diff --git a/engine/hooks/wrong-check-reflect/claude_stop_check.py b/engine/hooks/wrong-check-reflect/claude_stop_check.py index f83368bb..925f7c2a 100644 --- a/engine/hooks/wrong-check-reflect/claude_stop_check.py +++ b/engine/hooks/wrong-check-reflect/claude_stop_check.py @@ -14,7 +14,7 @@ def main() -> None: except (json.JSONDecodeError, OSError): return payload = payload if isinstance(payload, dict) else {} - try_enqueue_judge(payload) + try_enqueue_judge(payload, "claude") if __name__ == "__main__": diff --git a/engine/hooks/wrong-check-reflect/codex_notify.py b/engine/hooks/wrong-check-reflect/codex_notify.py index 366b53c7..04950760 100644 --- a/engine/hooks/wrong-check-reflect/codex_notify.py +++ b/engine/hooks/wrong-check-reflect/codex_notify.py @@ -29,7 +29,7 @@ def main() -> None: if payload.get("type") != "agent-turn-complete": return - try_enqueue_judge(payload) + try_enqueue_judge(payload, "codex") if __name__ == "__main__": diff --git a/engine/hooks/wrong-check-reflect/cursor_session.py b/engine/hooks/wrong-check-reflect/cursor_session.py index 8c57541c..9a2f5136 100644 --- a/engine/hooks/wrong-check-reflect/cursor_session.py +++ b/engine/hooks/wrong-check-reflect/cursor_session.py @@ -15,7 +15,7 @@ def main() -> None: print(json.dumps({"followup_message": ""})) return payload = payload if isinstance(payload, dict) else {} - try_enqueue_judge(payload) + try_enqueue_judge(payload, "cursor") print(json.dumps({"followup_message": ""})) diff --git a/engine/hooks/wrong-check-reflect/detect.py b/engine/hooks/wrong-check-reflect/detect.py index af7d394a..0bb9339b 100644 --- a/engine/hooks/wrong-check-reflect/detect.py +++ b/engine/hooks/wrong-check-reflect/detect.py @@ -11,7 +11,10 @@ sys.path.insert(0, os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) +from events import write_stage_event # noqa: E402 from flags import enforcement_gate # noqa: E402 HOOKS_DIR = os.path.dirname(os.path.abspath(__file__)) @@ -322,30 +325,49 @@ def _phrases(): return module -def enqueue_judge(payload: dict) -> str | None: - if not isinstance(payload, dict) or payload.get("stop_hook_active"): - return None +def _session(payload: dict, transcript: str) -> str: + for key in ("session_id", "conversation_id", "conversationId"): + value = payload.get(key) + if isinstance(value, str) and value: + return value + return transcript + + +def _skipped(payload: dict, harness: str, transcript: str, reason: str) -> None: + write_stage_event("wrong-check-reflect", harness, _session(payload, transcript), "judge_skipped", reason) + return None + + +def enqueue_judge(payload: dict, harness: str = "unknown") -> str | None: + if not isinstance(payload, dict): + return _skipped({}, harness, "", "bad_payload") + if payload.get("stop_hook_active"): + return _skipped(payload, harness, "", "stop_hook_active") if not enforcement_gate("wrong-check-reflect", payload.get("cwd")): - return None + return _skipped(payload, harness, "", "gate_off") path = resolve_transcript(payload) text = last_assistant_text(payload, path) key = reply_key(path, text) - if not text.strip() or already_prompted(key): - return None + if not text.strip(): + return _skipped(payload, harness, path, "empty_reply") + if already_prompted(key): + return _skipped(payload, harness, path, "already_prompted") if path and user_already_asked_reflect(path): - return None + return _skipped(payload, harness, path, "user_asked_reflect") dictionary = _phrases().load("wrong-check-reflect") job = _phrases().job(dictionary, path, text) job["id"] = uuid.uuid4().hex + job["harness"] = harness job_id = _judge().enqueue(job) - if job_id is not None: - mark_prompted(key) + if job_id is None: + return _skipped(payload, harness, path, "judge_child") + mark_prompted(key) return job_id -def try_enqueue_judge(payload: dict) -> None: +def try_enqueue_judge(payload: dict, harness: str = "unknown") -> None: try: - enqueue_judge(payload) + enqueue_judge(payload, harness) except Exception as exc: print(f"catstack-hook-error wrong-check-reflect: {type(exc).__name__}: {exc}", file=sys.stderr) return diff --git a/engine/hooks/wrong-check-reflect/tests/test_hooks.py b/engine/hooks/wrong-check-reflect/tests/test_hooks.py index a76867a8..27a62d0c 100644 --- a/engine/hooks/wrong-check-reflect/tests/test_hooks.py +++ b/engine/hooks/wrong-check-reflect/tests/test_hooks.py @@ -505,6 +505,46 @@ def test_unreadable_transcript_reports_unchecked_instead_of_going_quiet(self): self.assertIsNone(rows) self.assertIn("unchecked", err.getvalue()) + def stage_rows(self): + folder = os.path.join(self.state.name, "metrics") + rows = [] + for name in sorted(os.listdir(folder)) if os.path.isdir(folder) else []: + if name.startswith("events-"): + with open(os.path.join(folder, name), encoding="utf-8") as handle: + rows.extend(json.loads(line) for line in handle) + return [row for row in rows if row.get("mode_source") == "stage"] + + def test_every_skip_records_its_reason(self): + path = self.write_transcript(("user", REFLECT_COMMAND), ("assistant", HIT_TEXT)) + detect.enqueue_judge({"transcript_path": path, "stop_hook_active": True}, "claude") + detect.enqueue_judge({"transcript_path": path}, "claude") + other = self.write_transcript(("assistant", HIT_TEXT), name="other.jsonl") + detect.mark_prompted(detect.reply_key(other, HIT_TEXT)) + detect.enqueue_judge({"transcript_path": other}, "cursor") + with patch.dict(os.environ, {flags.REFLECT_ENFORCEMENT: "0"}): + detect.enqueue_judge({"transcript_path": other}, "codex") + detect.enqueue_judge({"last_assistant_message": " "}, "claude") + self.assertEqual( + [(r["action"], r["reason"], r["harness"]) for r in self.stage_rows()], + [ + ("judge_skipped", "stop_hook_active", "claude"), + ("judge_skipped", "user_asked_reflect", "claude"), + ("judge_skipped", "already_prompted", "cursor"), + ("judge_skipped", "gate_off", "codex"), + ("judge_skipped", "empty_reply", "claude"), + ], + ) + self.assertEqual(self.jobs(), []) + + def test_claude_stop_records_the_queued_job_under_the_claude_harness(self): + self.use_runners(SLOW_CLEAN) + path = self.write_transcript(("assistant", HIT_TEXT)) + run_claude({"transcript_path": path, "session_id": "s-1"}) + queued = [r for r in self.stage_rows() if r["action"] == "judge_queued"] + self.assertEqual([(r["hook"], r["harness"], r["reason"]) for r in queued], + [("wrong-check-reflect", "claude", "transcript")]) + self.assertEqual(queued[0]["finding_id"], self.wait_for_jobs(1)[0][: -len(".json")]) + def test_claude_malformed_stdin_fail_open(self): err = io.StringIO() with patch.object(sys, "stdin", io.StringIO("not-json")):