merge queue: checking #1038 on main (c83ef3a), stacked on #982 and #1000

corpus/skills/cat-mode/SKILL.md

@@ -55,6 +55,7 @@ Once direction is set, act — don't ask permission for each sub-step. One fully
   name every surface tried and grep the artifact already located (`--help`,
   bundle/asar, DB, logs).
 - **A blocked hand-back relays the gate's exit word for word.** When a hook, guard, or check blocks and the user must act, paste its exit message and the exact command, path, or marker it names, unshortened; a summary can drop the one step that gets them out.
+- **Hook noise and crashes are the agent's to notice and fix; never make the user report them.** When a hook error, crash trace, or unexpected hook output shows up in the session, read `~/.cache/catstack-hook-metrics/runs.jsonl` and the hook's own stderr, find the failing hook, and fix it or file the fix in the same turn — never ask the user whether a hook failed or to paste its error. A crash the user had to spot is a detection gap: say what broke and what now reports it. No known prior art.
 - Destructive or hard-to-reverse actions (force-push, bypassing a merge
   queue guard, schema changes) get one stop-and-ask. In the user's own repo,
   "I am in control, just do it" ends the discussion: show the verified list
@@ -103,8 +104,7 @@ The most repeated pattern in this user's history: when a bug, gap, or one-off re
 - **Skills and hooks work the same across every harness and machine.**
 - **Flag an automation candidate after three "check, wait, repeat" cycles.**
 - **Restructure a bloated instruction file rather than appending to it.**
-- **Apply the strongest fix first, not the fastest to write.** An unapplied
-  finding is not a finding.
+- **Apply the strongest fix first, not the fastest to write.** An unapplied finding is not a finding.
 - **Fleet upkeep runs from one script, not a session per machine.** Putting
   every machine on one Invoker release and the current catstack goes through
   `scripts/update_fleet.sh` (dry-run first). A missing step extends that

corpus/skills/cat-mode/tests/fires_asks_user_about_hook_crash.md

@@ -0,0 +1,16 @@
+`disable-model-invocation: true` means the model never reads this
+skill's `description:` to decide whether to apply it. cat-mode applies
+only when the `CATSTACK_CAT_MODE_DEFAULT=on` hook fires, or on an
+explicit `/cat-mode` invocation. Here the hook is on.
+
+The agent finishes a reply. A Stop hook then crashes with a Python
+traceback, and the harness shows `Stop hook error` under the reply. On
+its next turn the agent writes: "I see a hook error at the bottom —
+did a hook fail on your side? Can you paste the full error so I can
+look?" It never opened `runs.jsonl`, never read the hook's stderr, and
+never looked at the hook's source.
+
+This rule fires. The crash was already recorded where the agent could
+read it, so asking the user to report it hands the agent's own job
+back to them. The correct next step is to read the hook log, find the
+failing hook, and fix it.

corpus/skills/cat-mode/tests/stays_silent_agent_fixes_hook_crash.md

@@ -0,0 +1,14 @@
+`disable-model-invocation: true` means the model never reads this
+skill's `description:` to decide whether to apply it. cat-mode applies
+only when the `CATSTACK_CAT_MODE_DEFAULT=on` hook fires, or on an
+explicit `/cat-mode` invocation. Here the hook is on.
+
+The same Stop hook crashes after a reply. Without being asked, the
+agent reads `~/.cache/catstack-hook-metrics/runs.jsonl`, finds the row
+for the failing hook with its non-zero exit and traceback, opens the
+hook's source, reproduces the crash with the recorded input, and fixes
+the crash. Its reply says which hook broke, why, and that the fix is
+in, with the failing and passing runs pasted.
+
+This rule stays silent. The agent noticed the crash and owned it; the
+user never had to report anything.

engine/hooks/llm-judge/judge.py

@@ -554,6 +554,9 @@ def enqueue(job: dict) -> str | None:
     if os.path.basename(job_id) != job_id or job_id.startswith("."):
         raise ValueError(f"job id {job_id!r} is not a plain file name")
     job["id"] = job_id
+    if not str(job.get("transcript") or ""):
+        record_queued(job)
+        return None
     root = state_root()
     job_path = os.path.join(root, "jobs", f"{job_id}.json")
     write_json_atomic(job_path, job)
@@ -586,8 +589,8 @@ def record_queued(job: dict) -> None:
     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",
+            f"catstack-hook-error {job_hook(job)}: judge job {job['id']} has no transcript path, so its "
+            f"verdict could never be delivered; no judge run was started",
             file=sys.stderr,
         )
 

engine/hooks/llm-judge/tests/test_judge.py

@@ -539,6 +539,12 @@ def test_enqueue_with_no_transcript_is_recorded_and_reported_as_an_error(self):
         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_enqueue_with_no_transcript_starts_no_judge_run(self):
+        with patch.object(judge.subprocess, "Popen") as popen, redirect_stderr(io.StringIO()):
+            self.assertIsNone(judge.enqueue(self.job(id="lost-job", transcript="")))
+        popen.assert_not_called()
+        self.assertFalse(os.path.exists(os.path.join(self.state.name, "jobs", "lost-job.json")))
+
     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")

engine/hooks/unverified-tag-ledger/README.md

@@ -60,6 +60,24 @@ tag before refusing, so the ledger stays minable. Any value other than the four
 to `stale`; an `.env` candidate that exists and cannot be read is reported the
 same way rather than passing as "not set".
 
+### Drop counter
+
+Each `do_not_emit` refusal adds to the row's `refusals`. When a refused claim
+leaves the reply, the row closes with `outcome: dropped` (no verification tool
+ran that turn) or `outcome: checked` (one did). A turn whose tool list could
+not be read closes nothing. Totals across every session:
+
+```sh
+python3 engine/hooks/unverified-tag-ledger/detect.py stats
+```
+
+prints `refused_claims`, `dropped`, `checked`, `still_open`, and
+`unreadable_rows`. A rising `dropped` share means the setting is making replies
+say less rather than making claims get checked, which is the known cost of a
+drop policy: instruction tuning on abstention-aware data "can lead to
+over-abstention" (Wen et al., "Know Your Limits: A Survey of Abstention in
+Large Language Models", 2024, https://arxiv.org/abs/2407.18418).
+
 The hook resolves this flag itself through `engine/hooks/_flags/flags.py`.
 The `enabled_by` line in `engine/hooks/hooks.toml` records which flag the hook
 answers to; nothing reads that field, so it is documentation, not the gate.

engine/hooks/unverified-tag-ledger/detect.py

@@ -137,6 +137,10 @@ def record_turn(session_id: str, message: str, tools_used: set[str] | None, now=
             continue
         if row["claim"] in present:
             row["turns"] = row.get("turns", 0) + 1
+        elif row.get("refusals") and tools_used is not None:
+            row["resolved"] = True
+            row["resolved_at"] = stamp
+            row["outcome"] = "checked" if verified else "dropped"
         elif verified:
             row["resolved"] = True
             row["resolved_at"] = stamp
@@ -255,13 +259,19 @@ def evaluate(payload: dict, mode: str | None = None) -> dict:
     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)
+        row["claim"] for row in rows
+        if row.get("resolved") and row["claim"] in was_open and row.get("outcome") != "dropped")
     if discharged:
         notes.append(DISCHARGE_REFLECT.format(
             count=len(discharged), claims="; ".join(discharged[:MAX_LISTED])))
 
     emitted = emitted_tags(message)
     if mode == "do_not_emit" and emitted:
+        refused = {tag["claim"] for tag in parse_tags("\n".join(emitted))}
+        for row in rows:
+            if row["claim"] in refused and not row.get("resolved"):
+                row["refusals"] = row.get("refusals", 0) + 1
+        write_ledger(session_id, rows)
         return {"note": "", "block": (
             f"unverified-tag-ledger: {BEHAVIOR_FLAG}=do_not_emit, and this reply carries "
             f"{len(emitted)} CAT-UNVERIFIED tag(s): {'; '.join(emitted[:MAX_LISTED])}. "
@@ -304,6 +314,49 @@ def decide_stop(payload: dict) -> str:
     return evaluate(payload)["note"]
 
 
+def drop_stats() -> dict:
+    """Totals over every session ledger: how do_not_emit refusals ended.
+
+    `dropped` is a refused claim that left the reply in a turn that ran no
+    verification tool; `checked` is one that left after a check ran. A rising
+    dropped share is the over-abstention signal: the setting is making replies
+    say less instead of making claims get checked. A row that cannot be read
+    is counted in `unreadable_rows`, never skipped as if it were clean.
+    """
+    stats = {"refused_claims": 0, "dropped": 0, "checked": 0, "still_open": 0,
+             "unreadable_rows": 0}
+    base = ledger_dir()
+    if not os.path.isdir(base):
+        return stats
+    for name in sorted(os.listdir(base)):
+        if not name.endswith(".jsonl"):
+            continue
+        path = os.path.join(base, name)
+        try:
+            with open(path, encoding="utf-8") as handle:
+                lines = [line.strip() for line in handle if line.strip()]
+        except (OSError, UnicodeError) as exc:
+            sys.stderr.write(f"unverified-tag-ledger: cannot read {path}: {exc!r}\n")
+            stats["unreadable_rows"] += 1
+            continue
+        for line in lines:
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as exc:
+                sys.stderr.write(f"unverified-tag-ledger: {path} has a non-JSON row: {exc}\n")
+                stats["unreadable_rows"] += 1
+                continue
+            if not row.get("refusals"):
+                continue
+            stats["refused_claims"] += 1
+            outcome = row.get("outcome")
+            if outcome in ("dropped", "checked"):
+                stats[outcome] += 1
+            else:
+                stats["still_open"] += 1
+    return stats
+
+
 def _last_assistant_text(payload: dict) -> str:
     """The reply this Stop event is about.
 
@@ -438,3 +491,11 @@ def tools_used_this_turn(payload: dict) -> set[str] | None:
             f"tool list is unchecked: {exc!r}\n")
         return None
     return names
+
+
+if __name__ == "__main__":
+    if sys.argv[1:] == ["stats"]:
+        print(json.dumps(drop_stats(), indent=2))
+    else:
+        sys.stderr.write("usage: detect.py stats\n")
+        sys.exit(2)

engine/hooks/unverified-tag-ledger/tests/test_hooks.py

@@ -363,6 +363,60 @@ def test_evaluate_reads_the_flag_when_no_mode_is_passed(self) -> None:
             os.environ.pop(self.detect.BEHAVIOR_FLAG, None)
         self.assertIn("do_not_emit", verdict["block"])
 
+    def _refuse_then(self, next_message: str, *, tools: bool) -> dict:
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True), mode="do_not_emit")
+        self.detect.evaluate(self.payload(next_message, tools=tools), mode="do_not_emit")
+        return self.detect.read_ledger("s1")[0]
+
+    def test_a_do_not_emit_refusal_is_counted_on_the_row(self) -> None:
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True), mode="do_not_emit")
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True), mode="do_not_emit")
+        self.assertEqual(self.detect.read_ledger("s1")[0]["refusals"], 2)
+
+    def test_a_refused_claim_removed_with_no_check_is_counted_as_dropped(self) -> None:
+        row = self._refuse_then("Rewrote it without that sentence.", tools=False)
+        self.assertEqual(row["outcome"], "dropped")
+        self.assertTrue(row["resolved"])
+
+    def test_a_refused_claim_removed_after_a_check_is_counted_as_checked(self) -> None:
+        row = self._refuse_then("Ran it; output pasted above.", tools=True)
+        self.assertEqual(row["outcome"], "checked")
+
+    def test_a_claim_that_was_never_refused_gets_no_drop_outcome(self) -> None:
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True), mode="stale")
+        self.detect.evaluate(self.payload("Moved on.", tools=False), mode="stale")
+        self.assertNotIn("outcome", self.detect.read_ledger("s1")[0])
+
+    def test_stats_totals_refusals_and_outcomes_across_sessions(self) -> None:
+        self._refuse_then("Dropped it.", tools=False)
+        self.detect.evaluate(
+            self.payload(REAL_TAG_2, tools=True, session_id="s2"), mode="do_not_emit")
+        self.detect.evaluate(
+            self.payload("Checked.", tools=True, session_id="s2"), mode="do_not_emit")
+        self.detect.evaluate(
+            self.payload(REAL_TAG_1, tools=True, session_id="s3"), mode="do_not_emit")
+        stats = self.detect.drop_stats()
+        self.assertEqual(stats["refused_claims"], 3)
+        self.assertEqual(stats["dropped"], 1)
+        self.assertEqual(stats["checked"], 1)
+        self.assertEqual(stats["still_open"], 1)
+        self.assertEqual(stats["unreadable_rows"], 0)
+
+    def test_stats_counts_an_unreadable_row_instead_of_skipping_it_silently(self) -> None:
+        path = self.detect.ledger_path("bad")
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write("{not json}\n")
+        with redirect_stderr(io.StringIO()):
+            stats = self.detect.drop_stats()
+        self.assertEqual(stats["unreadable_rows"], 1)
+
+    def test_a_dropped_claim_is_not_announced_as_checked(self) -> None:
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True), mode="do_not_emit")
+        verdict = self.detect.evaluate(
+            self.payload("Rewrote it without that claim.", tools=False), mode="do_not_emit")
+        self.assertNotIn("went from unverified to checked", verdict["note"])
+
     def test_corrupt_ledger_row_is_reported_not_swallowed(self) -> None:
         path = self.detect.ledger_path("s3")
         os.makedirs(os.path.dirname(path), exist_ok=True)

engine/hooks/wrong-check-reflect/detect.py

@@ -349,8 +349,9 @@ def enqueue_judge(payload: dict, harness: str = "unknown") -> str | None:
     if not enforcement_gate("wrong-check-reflect", payload.get("cwd")):
         return _skipped(payload, harness, "", "gate_off")
     path = resolve_transcript(payload)
-    if not path and isinstance(payload.get("agent_transcript_path"), str):
-        return _skipped(payload, harness, "", "agent_transcript_missing")
+    named = payload.get("agent_transcript_path") or payload.get("transcript_path") or payload.get("transcriptPath")
+    if isinstance(named, str) and named and not path:
+        return _skipped(payload, harness, "", "transcript_missing")
     text = last_assistant_text(payload, path)
     key = reply_key(path, text)
     if not text.strip():

engine/hooks/wrong-check-reflect/tests/test_hooks.py

@@ -592,6 +592,25 @@ def test_every_skip_records_its_reason(self):
         )
         self.assertEqual(self.jobs(), [])
 
+    def test_subagent_stop_naming_a_missing_agent_transcript_queues_nothing(self):
+        self.use_runners(SLOW_CLEAN)
+        main = self.write_transcript(("assistant", HIT_TEXT))
+        missing = os.path.join(os.path.dirname(main), "subagents", "agent-gone.jsonl")
+        _, err = run_claude({"hook_event_name": "SubagentStop", "session_id": "s-1", "agent_id": "a1",
+                             "transcript_path": main, "agent_transcript_path": missing, "last_assistant_message": HIT_TEXT})
+        self.assertEqual(err, "")
+        self.assertEqual(self.jobs(), [])
+        self.assertIn(("judge_skipped", "transcript_missing"), [(r["action"], r["reason"]) for r in self.stage_rows()])
+
+    def test_stop_naming_a_missing_transcript_queues_nothing(self):
+        self.use_runners(SLOW_CLEAN)
+        missing = os.path.join(self.reflect_state.name, "gone.jsonl")
+        _, err = run_claude({"hook_event_name": "Stop", "session_id": "s-1", "transcript_path": missing,
+                             "last_assistant_message": HIT_TEXT})
+        self.assertEqual(err, "")
+        self.assertEqual(self.jobs(), [])
+        self.assertIn(("judge_skipped", "transcript_missing"), [(r["action"], r["reason"]) for r in self.stage_rows()])
+
     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))

tests/test_cat_mode.py

@@ -1770,6 +1770,45 @@ def test_both_copies_of_the_rule_still_show_the_tag_at_all(self):
                 self.assertTrue(self.markers().well_formed_tags(self.read(path)))
 
 
+class TestCatModeAgentOwnsHookFailures(unittest.TestCase):
+    """Hook noise and crashes are the agent's to notice and fix. The user
+    kept pasting hook errors back into the chat; the rule makes the agent
+    read the hook log itself instead of asking."""
+
+    FIXTURES_DIR = os.path.join(REPO_ROOT, "corpus", "skills", "cat-mode", "tests")
+
+    def fixture(self, name):
+        path = os.path.join(self.FIXTURES_DIR, name)
+        self.assertTrue(os.path.isfile(path), path)
+        with open(path, encoding="utf-8") as handle:
+            return re.sub(r"\s+", " ", handle.read())
+
+    def test_skill_states_the_rule(self):
+        text = normalized_skill_text()
+        self.assertIn(
+            "**Hook noise and crashes are the agent's to notice and fix; never make the user report them.**",
+            text,
+        )
+        self.assertIn("`~/.cache/catstack-hook-metrics/runs.jsonl`", text)
+
+    def test_rule_names_its_prior_art_status(self):
+        text = normalized_skill_text()
+        start = text.index("**Hook noise and crashes are the agent's")
+        bullet = text[start:text.index(" - ", start)]
+        self.assertIn("No known prior art.", bullet)
+
+    def test_positive_fixture_asks_the_user_about_a_hook(self):
+        text = self.fixture("fires_asks_user_about_hook_crash.md")
+        self.assertIn("did a hook fail", text)
+        self.assertIn("never opened `runs.jsonl`", text)
+
+    def test_negative_fixture_reads_the_log_and_fixes_the_crash(self):
+        text = self.fixture("stays_silent_agent_fixes_hook_crash.md")
+        self.assertIn("reads `~/.cache/catstack-hook-metrics/runs.jsonl`", text)
+        self.assertIn("fixes the crash", text)
+        self.assertNotIn("did a hook fail", text)
+
+
 class TestCatModeShapeAdmissionAlertFanout(unittest.TestCase):
     """Each rule is one line in SKILL.md with its full text in references/.
     Pinning both halves keeps a trim of SKILL.md from silently dropping the