diff --git a/corpus/skills/cat-mode/SKILL.md b/corpus/skills/cat-mode/SKILL.md index 97a3caba2..c826356b2 100644 --- a/corpus/skills/cat-mode/SKILL.md +++ b/corpus/skills/cat-mode/SKILL.md @@ -138,6 +138,7 @@ isolated subagents and report back async rather than blocking on each one. durable artifact.** Separable and parallel is not authorization to fan out; a fan-out default cannot hand a subagent publishing authority the routing table never granted. Route that work through Execution routing. +- **Many PR stacks: one parallel unit per stack, never serial** (Invoker, else a worktree subagent each). - **A fork/subagent told to touch files must run in its own worktree, not the live checkout** — even when told "read-only." Scope wording is not filesystem isolation. diff --git a/corpus/skills/cat-mode/references/execution-routing.md b/corpus/skills/cat-mode/references/execution-routing.md index b59942609..dd105c7c6 100644 --- a/corpus/skills/cat-mode/references/execution-routing.md +++ b/corpus/skills/cat-mode/references/execution-routing.md @@ -6,6 +6,7 @@ Catstack owns judgment and local fallback. Invoker owns durable plan submission, ## Decision +0. **More than one independent publishing unit** (several PR stacks to land or repair, several workflows): never serial in the parent chat. Invoker first, one workflow per unit; `subagent_worktree_per_unit` when Invoker is unavailable or the user directs subagents. `route_execution(units=N)` returns it; see [subagents.md](subagents.md). 1. **Invoker unavailable** (no `invoker_prepare_plan_review` / `invoker_submit_plan` tools): stay local — subagents, `loop-generator`, `land-stack`, current chat execution. 2. **Small local work** (one-file fix, short edit, read-only question): stay local even if Invoker is installed. Post-land wait until `MERGED`, merge-queue babysit, and already-named execution Backlog are **not** this bucket — they are `durable_parallel`. 3. **Approved plan or durable/parallel work** and Invoker MCP is available: delegate. If Invoker is missing, use a separate git worktree + PR stack. Do not park that work in the parent chat. diff --git a/corpus/skills/cat-mode/references/subagents.md b/corpus/skills/cat-mode/references/subagents.md index 8180b19c7..5ed7b83d3 100644 --- a/corpus/skills/cat-mode/references/subagents.md +++ b/corpus/skills/cat-mode/references/subagents.md @@ -36,6 +36,29 @@ definition whatever `produces` claims; and an empty or unrecognized `produces` raises rather than falling through to fan-out, because an output nobody declared is unchecked, not clean. +## Many stacks: parallel per unit, never serial + +Routing picks *who* runs publishing work; it never licenses running several +independent units one after another in the parent thread. When the work is N +independent PR stacks (landing, conflict repair, review fixes), each stack is +its own unit with its own worktree, and the units run in parallel: an Invoker +workflow per stack first, and one worktree-isolated subagent per stack as the +fallback when Invoker is unavailable or the user directs it +(`subagent_worktree_per_unit`). The per-unit subagent still inherits only the +scope the parent names, and its transcript is still grepped for writes. + +The failure shape: asked to land dozens of admin-bypass PRs across two repos, +the parent recommended working the ~20 rebases and review fixes "one at a +time" and started serially in one worktree, until the user asked for a +worktree subagent per stack. The routing table allowed it: `route_execution` +returned `local` for publishing work without Invoker at any unit count. + +Prior art: Amdahl's law — Gene M. Amdahl, "Validity of the single processor +approach to achieving large scale computing capabilities", AFIPS 1967, +https://doi.org/10.1145/1465482.1465560 — the serial fraction bounds the +whole job, so independent units forced through one thread set the finish +time. + ## Defer to the harness's routing skill The precedence above is catstack's fallback, not the owner. When a harness diff --git a/corpus/skills/cat-mode/scripts/route_execution.py b/corpus/skills/cat-mode/scripts/route_execution.py index 4fbf70287..2ad1bc086 100644 --- a/corpus/skills/cat-mode/scripts/route_execution.py +++ b/corpus/skills/cat-mode/scripts/route_execution.py @@ -11,8 +11,8 @@ from typing import Literal WorkKind = Literal["readonly", "small_local", "approved_plan", "durable_parallel"] -Route = Literal["local", "delegate_invoker"] -Delegation = Literal["local", "delegate_invoker", "subagent_fanout"] +Route = Literal["local", "delegate_invoker", "subagent_worktree_per_unit"] +Delegation = Literal["local", "delegate_invoker", "subagent_fanout", "subagent_worktree_per_unit"] DURABLE_ALIASES = frozenset({"post_land_babysit", "named_execution_backlog"}) @@ -36,6 +36,13 @@ "invoker_wait_for_workflow_or_status", ) +SUBAGENT_PER_UNIT_STEPS = ( + "one_worktree_per_unit", + "spawn_one_subagent_per_unit_in_parallel", + "collect_reports_async", + "grep_transcripts_for_writes", +) + SUBAGENT_FANOUT_STEPS = ( "spawn_worktree_isolated_subagents", "collect_reports_async", @@ -72,14 +79,32 @@ def normalize_work_kind(work_kind: str) -> WorkKind: raise ValueError(f"unknown work_kind: {work_kind!r}") -def route_execution(*, tools: set[str] | frozenset[str] | list[str], work_kind: str) -> Route: +def route_execution( + *, + tools: set[str] | frozenset[str] | list[str], + work_kind: str, + units: int = 1, + user_directed_subagents: bool = False, +) -> Route: """Return where execution should run for this request. - 1. Invoker MCP missing → local - 2. Small / read-only work → local even if Invoker exists - 3. Approved plan or durable/parallel → delegate_invoker + `units` counts independent publishing units (PR stacks, workflows). + More than one never runs serially in the parent thread: + Invoker first, else one worktree-isolated subagent per unit. + + 1. units > 1 → delegate_invoker, or subagent_worktree_per_unit when + Invoker is missing or the user directed subagents + 2. Invoker MCP missing → local + 3. Small / read-only work → local even if Invoker exists + 4. Approved plan or durable/parallel → delegate_invoker """ kind = normalize_work_kind(work_kind) + if units < 1: + raise ValueError(f"units must be >= 1, got {units!r}") + if units > 1 and kind != "readonly": + if invoker_mcp_available(tools) and not user_directed_subagents: + return "delegate_invoker" + return "subagent_worktree_per_unit" if not invoker_mcp_available(tools): return "local" if kind in ("readonly", "small_local"): @@ -112,6 +137,8 @@ def route_delegation( tools: set[str] | frozenset[str] | list[str], work_kind: str, produces: set[str] | frozenset[str] | list[str] | tuple[str, ...], + units: int = 1, + user_directed_subagents: bool = False, ) -> Delegation: """Resolve the Subagents default against the execution-routing table. @@ -122,7 +149,9 @@ def route_delegation( """ normalize_work_kind(work_kind) if publishes(work_kind=work_kind, produces=produces): - return route_execution(tools=tools, work_kind=work_kind) + return route_execution( + tools=tools, work_kind=work_kind, units=units, user_directed_subagents=user_directed_subagents, + ) return "subagent_fanout" @@ -131,6 +160,8 @@ def handoff_steps_for(route: Route | Delegation) -> tuple[str, ...]: return ("stay_local",) if route == "subagent_fanout": return SUBAGENT_FANOUT_STEPS + if route == "subagent_worktree_per_unit": + return SUBAGENT_PER_UNIT_STEPS return DELEGATE_HANDOFF_STEPS @@ -142,9 +173,15 @@ def handoff_steps_for(route: Route | Delegation) -> tuple[str, ...]: tools = payload.get("tools", []) work_kind = payload.get("work_kind", "small_local") produces = payload.get("produces") + units = int(payload.get("units", 1)) + directed = bool(payload.get("user_directed_subagents", False)) if produces is None: - route: Route | Delegation = route_execution(tools=tools, work_kind=work_kind) + route: Route | Delegation = route_execution( + tools=tools, work_kind=work_kind, units=units, user_directed_subagents=directed, + ) else: - route = route_delegation(tools=tools, work_kind=work_kind, produces=produces) + route = route_delegation( + tools=tools, work_kind=work_kind, produces=produces, units=units, user_directed_subagents=directed, + ) defer_to = installed_harness_routing_skill(payload.get("home")) print(json.dumps({"route": route, "steps": list(handoff_steps_for(route)), "defer_to": defer_to})) diff --git a/corpus/skills/principle-guard-the-context-window/scripts/ci_logs.py b/corpus/skills/principle-guard-the-context-window/scripts/ci_logs.py index cc2a69e3a..69aece91f 100755 --- a/corpus/skills/principle-guard-the-context-window/scripts/ci_logs.py +++ b/corpus/skills/principle-guard-the-context-window/scripts/ci_logs.py @@ -342,7 +342,7 @@ def capture_github(args: argparse.Namespace, root: Path) -> dict[str, Any]: "notes": notes, } write_manifest(directory, manifest) - return artifact_response(manifest, cached=False, downloaded=True, notes=notes) + return artifact_response(manifest, cached=False, downloaded=True, notes=[]) def judge_completeness(producer_status: int, job_status: str, log_bytes: int) -> tuple[str, str]: diff --git a/corpus/skills/principle-guard-the-context-window/tests/test_ci_logs.py b/corpus/skills/principle-guard-the-context-window/tests/test_ci_logs.py index 1c402bbca..905781a50 100644 --- a/corpus/skills/principle-guard-the-context-window/tests/test_ci_logs.py +++ b/corpus/skills/principle-guard-the-context-window/tests/test_ci_logs.py @@ -377,6 +377,26 @@ def test_in_progress_job_is_never_marked_complete(self): self.assertFalse(payload["artifact"]["complete"]) + def test_first_capture_lists_each_note_once(self): + env = self.gh_env(failing_log(), mode="in_progress") + payload, _ = self.run_helper( + "capture", + "--artifact-root", + str(self.artifacts), + "--repo", + "owner/name", + "--run", + "12345", + "--job", + "41", + "--gh-path", + str(self.gh), + env=env, + ) + notes = payload["notes"] + self.assertTrue(notes, payload) + self.assertEqual(len(notes), len(set(notes)), notes) + class TestSnippet(CiLogsCase): def test_structural_blocks_carry_the_failure_and_the_final_status(self): payload, _ = self.import_log(failing_log(blocks=1)) diff --git a/engine/hooks/_runner/report.py b/engine/hooks/_runner/report.py index afc5504fe..71c99f338 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 71964d0bc..43baa1235 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 5e88bf00e..8ee1a7179 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 38254f627..a994d5ec9 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 de8945139..1a50cc74e 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/phrases/wrong-check-reflect.json b/engine/hooks/llm-judge/phrases/wrong-check-reflect.json index 173dbcbef..cf121c913 100644 --- a/engine/hooks/llm-judge/phrases/wrong-check-reflect.json +++ b/engine/hooks/llm-judge/phrases/wrong-check-reflect.json @@ -1,7 +1,7 @@ { "checker": "wrong-check-reflect", "reads": "reply", - "meaning": "The latest assistant reply admits that something it told the user earlier was wrong, misread, or answered the wrong question.", + "meaning": "The latest assistant reply admits that something it told the user earlier was wrong, misread, or answered the wrong question, OR that an earlier claim of its own went out before the check backing it had run. A reply saying the claim turned out correct but was stated before it was checked IS a match: the failure is the order, not the answer, so no word in the reply has to say anything was wrong.", "match": [ "my earlier check was wrong", "You're right, I misread that", @@ -10,7 +10,9 @@ "I was wrong about the path", "Correction: the file I pointed you to earlier is not the one in use; the real one is src/b.py.", "Good catch. The earlier number was off; the real count is 12.", - "My earlier read of the config was wrong \u2014 the default is 4, not 8." + "My earlier read of the config was wrong \u2014 the default is 4, not 8.", + "Correcting one claim and arming the check I implied:", + "I was right - but I said it a turn before I checked it" ], "not_match": [ "You're right. Let's go with option B.", diff --git a/engine/hooks/llm-judge/tests/test_judge.py b/engine/hooks/llm-judge/tests/test_judge.py index 33cce814c..5d083d3d8 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/unverified-tag-ledger/README.md b/engine/hooks/unverified-tag-ledger/README.md index ccfa1d8cf..121cd94ad 100644 --- a/engine/hooks/unverified-tag-ledger/README.md +++ b/engine/hooks/unverified-tag-ledger/README.md @@ -53,6 +53,12 @@ fixtures in `tests/test_hooks.py`. reason is written to stderr. - **Escalation** — a claim outstanding `ESCALATE_AFTER_TURNS` (3) turns or more is reported as a reflect trigger rather than accumulating quietly. +- **Discharge is itself a reflect trigger** — a row going outstanding -> + discharged is the record of a claim that went out first and was checked + after. That is an evidence-order miss, and it carries no wrongness word, so + the phrase scanners (`engine/skills/reflect/scripts/self_retraction_scan.py`, + and the `wrong-check-reflect` dictionary) cannot see it from the text. This + hook sees it from state instead, and says so on the Stop that discharges. Malformed tags are deliberately ignored here; `diu-stop` already rejects those. diff --git a/engine/hooks/unverified-tag-ledger/detect.py b/engine/hooks/unverified-tag-ledger/detect.py index 8a47c2eff..1f54902da 100644 --- a/engine/hooks/unverified-tag-ledger/detect.py +++ b/engine/hooks/unverified-tag-ledger/detect.py @@ -34,6 +34,15 @@ ESCALATE_AFTER_TURNS = 3 MAX_LISTED = 5 +DISCHARGE_REFLECT = ( + "unverified-tag-ledger: {count} claim(s) went from unverified to checked this turn: " + "{claims}. That transition is the whole event: the claim went out first and the check " + "ran after. No wording has to admit anything for this to be true, which is why the " + "phrase scanners miss it -- an evidence-order miss carries no wrongness word. " + "Treat it as a reflect trigger, not a milestone: run reflect on this transcript, or " + "say plainly why this one does not need it." +) + CLAIM_RE = re.compile( r"\{\{\s*CAT-UNVERIFIED\s*:?\s*(?P.*?)(?:--|—)\s*cannot\s+verify\s*:\s*(?P[^}]*)\}\}", re.IGNORECASE | re.DOTALL, @@ -177,18 +186,26 @@ def evaluate(payload: dict) -> dict: session_id = str(payload.get("session_id") or "") message = _last_assistant_text(payload) tools = tools_used_this_turn(payload) + was_open = {row["claim"] for row in outstanding(read_ledger(session_id))} rows = record_turn(session_id, message, tools) + notes = [] + discharged = sorted( + row["claim"] for row in rows if row.get("resolved") and row["claim"] in was_open) + if discharged: + notes.append(DISCHARGE_REFLECT.format( + count=len(discharged), claims="; ".join(discharged[:MAX_LISTED]))) new_claims = {tag["claim"] for tag in parse_tags(message)} if not new_claims: - return {"note": "", "block": ""} + return {"note": "\n".join(notes), "block": ""} if tools is None: - return {"note": ( + notes.append( f"unverified-tag-ledger: logged {len(new_claims)} CAT-UNVERIFIED claim(s), but this " "turn's tool calls could not be read from transcript_path (see the line above), so " "whether a check was attempted is UNCHECKED, not clean. Nothing was discharged and " - "the turn was not refused."), "block": ""} + "the turn was not refused.") + return {"note": "\n".join(notes), "block": ""} if not tools & VERIFY_TOOLS and not payload.get("stop_hook_active"): claims = "; ".join(sorted(new_claims)[:MAX_LISTED]) @@ -201,12 +218,12 @@ def evaluate(payload: dict) -> dict: fresh = [row for row in rows if row["claim"] in new_claims and not row.get("resolved") and row.get("turns", 0) == 0] - if not fresh: - return {"note": "", "block": ""} - return {"note": ( - f"unverified-tag-ledger: logged {len(fresh)} CAT-UNVERIFIED claim(s) against this session. " - "They are deferred, not discharged, and will be raised again next turn " - "(cat-mode/SKILL.md:269)."), "block": ""} + if fresh: + notes.append( + f"unverified-tag-ledger: logged {len(fresh)} CAT-UNVERIFIED claim(s) against this " + "session. They are deferred, not discharged, and will be raised again next turn " + "(cat-mode/SKILL.md:269).") + return {"note": "\n".join(notes), "block": ""} def decide_stop(payload: dict) -> str: diff --git a/engine/hooks/unverified-tag-ledger/tests/test_hooks.py b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py index 12ab99310..251cede62 100644 --- a/engine/hooks/unverified-tag-ledger/tests/test_hooks.py +++ b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py @@ -38,6 +38,9 @@ "-- cannot verify: my own reasoning isn't observable by any command}}") MALFORMED = "{{CAT-UNVERIFIED: something I did not check}}" +EVIDENCE_ORDER_1 = "Correcting one claim and arming the check I implied:" +EVIDENCE_ORDER_2 = "I was right - but I said it a turn before I checked it" + def _real_transcript_lines() -> list[str]: with open(REAL_TRANSCRIPT, encoding="utf-8") as handle: @@ -209,6 +212,33 @@ def test_verified_and_dropped_tag_is_discharged_through_the_real_payload(self) - self.assertEqual(self.detect.outstanding(self.detect.read_ledger("s1")), []) self.assertEqual(self.detect.reminder("s1"), "") + def test_a_discharged_claim_fires_the_reflect_trigger(self) -> None: + self.detect.evaluate(self.payload(REAL_TAG_1, tools=True)) + verdict = self.detect.evaluate( + self.payload("Here is the pasted output proving it.", tools=True)) + self.assertIn("reflect trigger", verdict["note"]) + self.assertIn("widened scope", verdict["note"]) + + def test_an_evidence_order_correction_triggers_with_no_wrongness_word(self) -> None: + """The transition fires; the reply's wording is not consulted at all.""" + for index, reply in enumerate((EVIDENCE_ORDER_1, EVIDENCE_ORDER_2)): + session = f"evidence-order-{index}" + self.detect.evaluate(self.payload(REAL_TAG_1, tools=True, session_id=session)) + verdict = self.detect.evaluate( + self.payload(reply, tools=True, session_id=session)) + self.assertIn("reflect trigger", verdict["note"]) + self.assertIn("carries no wrongness word", verdict["note"]) + + def test_a_turn_that_discharges_nothing_stays_silent_about_reflect(self) -> None: + verdict = self.detect.evaluate( + self.payload("Ran the tests, all green.", tools=True)) + self.assertEqual(verdict["note"], "") + + def test_a_reemitted_tag_is_not_reported_as_discharged(self) -> None: + self.detect.evaluate(self.payload(REAL_TAG_1, tools=True)) + verdict = self.detect.evaluate(self.payload(REAL_TAG_1, tools=True)) + self.assertNotIn("reflect trigger", verdict["note"]) + def test_unchecked_turn_does_not_discharge_a_row(self) -> None: self.detect.evaluate(self.payload(REAL_TAG_1, tools=True)) self.detect.evaluate({ diff --git a/engine/hooks/wrong-check-reflect/claude_stop_check.py b/engine/hooks/wrong-check-reflect/claude_stop_check.py index f83368bbe..925f7c2ae 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 366b53c7b..04950760d 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 8c57541c9..9a2f5136a 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 af7d394af..0bb9339b3 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/eval_dictionary.py b/engine/hooks/wrong-check-reflect/eval_dictionary.py index 258dabb2f..9a9e89273 100644 --- a/engine/hooks/wrong-check-reflect/eval_dictionary.py +++ b/engine/hooks/wrong-check-reflect/eval_dictionary.py @@ -17,6 +17,9 @@ (HIT_TEXT, True), ("You're right. Let's go with option B.", False), ("I double-checked my earlier count and it holds; nothing in it was wrong.", False), + ("Correcting one claim and arming the check I implied:", True), + ("I was right - but I said it a turn before I checked it", True), + ("I ran the check first and then said it, so the order was right.", False), ) diff --git a/engine/hooks/wrong-check-reflect/tests/test_hooks.py b/engine/hooks/wrong-check-reflect/tests/test_hooks.py index a76867a8d..27a62d0c6 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")): diff --git a/engine/skills/reflect/scripts/tests/test_self_retraction_scan.py b/engine/skills/reflect/scripts/tests/test_self_retraction_scan.py index 72266f872..852582772 100644 --- a/engine/skills/reflect/scripts/tests/test_self_retraction_scan.py +++ b/engine/skills/reflect/scripts/tests/test_self_retraction_scan.py @@ -74,6 +74,30 @@ def test_third_party_blame_inside_a_reported_clause_stays_clean(self): self.assertIsNone(self_retraction_scan.find_admission(text)) +class TestEvidenceOrderIsOutOfReach(unittest.TestCase): + """Two real corrections this scan cannot see, and the reason it cannot. + + Both are corrections about evidence ORDER: the claim was true, and it was + asserted before the check ran. Nothing in either sentence says anything was + wrong, so every pattern here misses them by construction. Pinned so the + next author widens the regex knowingly rather than by accident: the catch + for this class is the unverified-tag-ledger discharge transition, which + reads state rather than wording. + """ + + def test_arming_the_implied_check_is_not_reachable_by_wording(self): + text = "Correcting one claim and arming the check I implied:" + self.assertIsNone(self_retraction_scan.find_admission(text)) + + def test_right_but_asserted_early_is_not_reachable_by_wording(self): + text = "I was right - but I said it a turn before I checked it" + self.assertIsNone(self_retraction_scan.find_admission(text)) + + def test_the_same_sentence_with_a_wrongness_word_does_fire(self): + text = "I was wrong about the path; I said it a turn before I checked it." + self.assertIsNotNone(self_retraction_scan.find_admission(text)) + + class TestScanAssistantTexts(unittest.TestCase): def test_collects_one_hit_per_admission(self): hits = self_retraction_scan.scan_assistant_texts( diff --git a/engine/skills/reflect/scripts/tests/test_token_audit.py b/engine/skills/reflect/scripts/tests/test_token_audit.py index 5f15c1fb2..ef1799286 100644 --- a/engine/skills/reflect/scripts/tests/test_token_audit.py +++ b/engine/skills/reflect/scripts/tests/test_token_audit.py @@ -1248,6 +1248,54 @@ def test_redundant_reads_flag_yes(self): os.unlink(path) os.unlink(out.name) + def test_bash_only_session_reports_tool_shape_detectors_unchecked(self): + u = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0} + lines = [ + claude_assistant_line("m1", "u1", [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "sed -n 1,50p /a.py"}}], u), + claude_assistant_line("m2", "u2", [{"type": "tool_use", "id": "t2", "name": "Bash", "input": {"command": "sed -n 1,50p /a.py"}}], u), + claude_assistant_line("m3", "u3", [{"type": "tool_use", "id": "t3", "name": "Bash", "input": {"command": "sed -i s/a/b/ /a.py"}}], u), + ] + path = write_jsonl(lines) + out = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) + out.close() + try: + with redirect_stdout(io.StringIO()): + token_audit.audit_claude(path, out_path=out.name) + with open(out.name) as f: + report = json.load(f) + for name in ("model-tier-candidates", "redundant-reads", "no-verify-edit-streak"): + fl = self._flag_by_name(report, name) + self.assertEqual(fl["value"], "unchecked", name) + self.assertIsNone(fl["count"], name) + self.assertIn("went through Bash", fl["rationale"]) + finally: + os.unlink(path) + os.unlink(out.name) + + def test_structured_tool_session_still_measures_tool_shape_detectors(self): + u = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0} + lines = [ + claude_assistant_line("m1", "u1", [{"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "/a.py", "offset": 1, "limit": 10}}], u), + claude_assistant_line("m2", "u2", [{"type": "tool_use", "id": "t2", "name": "Read", "input": {"file_path": "/a.py", "offset": 1, "limit": 10}}], u), + claude_assistant_line("m3", "u3", [{"type": "tool_use", "id": "t3", "name": "Bash", "input": {"command": "echo hi"}}], u), + ] + path = write_jsonl(lines) + out = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) + out.close() + try: + with redirect_stdout(io.StringIO()): + token_audit.audit_claude(path, out_path=out.name) + with open(out.name) as f: + report = json.load(f) + for name in ("model-tier-candidates", "redundant-reads", "no-verify-edit-streak"): + fl = self._flag_by_name(report, name) + self.assertIn(fl["value"], ("yes", "no"), name) + self.assertIsInstance(fl["count"], int, name) + self.assertEqual(self._flag_by_name(report, "redundant-reads")["count"], 1) + finally: + os.unlink(path) + os.unlink(out.name) + def test_recurring_failure_signatures_flag_yes(self): u = {"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0} lines = [] diff --git a/engine/skills/reflect/scripts/token_audit.py b/engine/skills/reflect/scripts/token_audit.py index 474f9c0fb..406205643 100644 --- a/engine/skills/reflect/scripts/token_audit.py +++ b/engine/skills/reflect/scripts/token_audit.py @@ -792,7 +792,7 @@ def audit_subagents(path, audit_started_at=None): fired = [name for name in SUBAGENT_THRASH_FLAGS if flags.get(name, {}).get("value") == "yes"] if fired: thrash["by_agent"][fname] = fired - thrash["redundant_reads"] += flags["redundant-reads"]["count"] + thrash["redundant_reads"] += flags["redundant-reads"]["count"] or 0 for fp in stats.get("redundant_read_files", []): thrash["redundant_read_files"].append(f"{fname}:{fp}") thrash["tool_errors"] += stats["n_errors"] @@ -959,6 +959,15 @@ def audit_claude(path, out_path=None, include_subagents=True): simple_turns += 1 simple_turn_output_tokens += msg_usage[mid]["output_tokens"] + FILE_TOOLS = set(LOOKUP_TOOLS) | {"Edit", "Write"} + used_tools = {n for names in msg_tool_names.values() for n in names if isinstance(n, str)} + bash_only = "Bash" in used_tools and not (used_tools & FILE_TOOLS) + lookup_blind = read_blind = edit_blind = bash_only + BLIND_RATIONALE = ( + "all file access in this session went through Bash; this detector keys on the " + "{} tool name(s) and cannot observe it" + ) + grand = total_input + total_output + total_cache_read + total_cache_creation from_model = models.most_common(1)[0][0] if models else "claude-sonnet-5" tier_backtest = None @@ -1047,8 +1056,9 @@ def audit_claude(path, out_path=None, include_subagents=True): flags = [ _flag( "model-tier-candidates", - "yes" if simple_turns else "no", - simple_turns, + "unchecked" if lookup_blind else ("yes" if simple_turns else "no"), + None if lookup_blind else simple_turns, + BLIND_RATIONALE.format("Read/Grep/Glob") if lookup_blind else ( f"{simple_turns}/{n_assistant} turns called only Read/Grep/Glob " f"({simple_turn_output_tokens:,} output tokens on those turns)" @@ -1062,8 +1072,9 @@ def audit_claude(path, out_path=None, include_subagents=True): ), _flag( "redundant-reads", - "yes" if redundant else "no", - len(redundant), + "unchecked" if read_blind else ("yes" if redundant else "no"), + None if read_blind else len(redundant), + BLIND_RATIONALE.format("Read") if read_blind else f"{len(redundant)} redundant re-read(s) of an identical file+offset/limit window with no edit in between", ), _flag( @@ -1074,8 +1085,9 @@ def audit_claude(path, out_path=None, include_subagents=True): ), _flag( "no-verify-edit-streak", - "yes" if flagged_files or global_streak_max >= THRESH else "no", - global_streak_max, + "unchecked" if edit_blind else ("yes" if flagged_files or global_streak_max >= THRESH else "no"), + None if edit_blind else global_streak_max, + BLIND_RATIONALE.format("Edit/Write") if edit_blind else ( f"longest edit streak with zero verification: {global_streak_max}; " f"{len(flagged_files)} file(s) at or above threshold {THRESH}; " diff --git a/tests/test_execution_routing.py b/tests/test_execution_routing.py index 49b3a80fd..8fb7bddba 100644 --- a/tests/test_execution_routing.py +++ b/tests/test_execution_routing.py @@ -40,6 +40,47 @@ def test_partial_tools_still_local(self): ) self.assertEqual(route, "local") + def test_many_publishing_units_never_run_serially(self): + tools = list(self.router.INVOKER_REQUIRED_TOOLS) + for kind in ("durable_parallel", "approved_plan", "post_land_babysit", "small_local"): + for available in (tools, []): + with self.subTest(kind=kind, invoker=bool(available)): + route = self.router.route_execution(tools=available, work_kind=kind, units=13) + self.assertNotEqual(route, "local") + + def test_many_units_prefer_invoker_then_worktree_subagent_per_unit(self): + tools = list(self.router.INVOKER_REQUIRED_TOOLS) + self.assertEqual( + self.router.route_execution(tools=tools, work_kind="post_land_babysit", units=13), + "delegate_invoker", + ) + self.assertEqual( + self.router.route_execution(tools=[], work_kind="post_land_babysit", units=13), + "subagent_worktree_per_unit", + ) + self.assertEqual( + self.router.route_execution( + tools=tools, work_kind="post_land_babysit", units=13, user_directed_subagents=True, + ), + "subagent_worktree_per_unit", + ) + steps = self.router.handoff_steps_for("subagent_worktree_per_unit") + self.assertEqual(steps[0], "one_worktree_per_unit") + self.assertIn("grep_transcripts_for_writes", steps) + + def test_many_units_route_delegation_reaches_per_unit_route(self): + route = self.router.route_delegation( + tools=[], work_kind="durable_parallel", produces=["pull_request"], units=4, + ) + self.assertEqual(route, "subagent_worktree_per_unit") + + def test_units_must_be_positive(self): + with self.assertRaises(ValueError): + self.router.route_execution(tools=[], work_kind="durable_parallel", units=0) + + def test_single_unit_keeps_existing_routes(self): + self.assertEqual(self.router.route_execution(tools=[], work_kind="approved_plan", units=1), "local") + def test_small_local_stays_local_even_with_invoker(self): tools = list(self.router.INVOKER_REQUIRED_TOOLS) for kind in ("small_local", "readonly"):