From 44d7c232e14988ca3cba35c5819c1b88d3ca7a21 Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Fri, 11 Sep 2026 18:50:14 -0700 Subject: [PATCH] feat(hooks): log CAT-UNVERIFIED tags instead of treating them as evidence cat-mode/SKILL.md:269 says a hedge is "a trigger to verify, never a place to stop". Every evidence hook implemented the opposite: markers.well_formed_tags() is consumed as equivalent to evidence (prove-it-ship-gate/detect.py), so a correctly-formed tag silenced the whole stack, was counted nowhere, and was revisited never. Only malformed and legacy tag shapes ever fired. The prose forbade stopping while the tooling rewarded it. unverified-tag-ledger records each well-formed tag against the session and re-surfaces outstanding claims on the next prompt. It deliberately does not block the emitting turn: the tag exists for checks that cannot run in that turn, so blocking there deadlocks. A claim is discharged when a later turn runs a verification tool and stops re-emitting it; one outstanding three turns or more is reported as a reflect trigger. Observed 2026-09-11 in a NiceSpeak streaming session: two well-formed tags were emitted, each ended its turn, neither left a trace. Both are the positive fixtures in tests/test_hooks.py. Fail-before/pass-after, same payload: diu-stop, hedge-runs-prove-it and prove-it-ship-gate all exit 0 silent with zero ledger rows; the new Stop hook logs the claim and the UserPromptSubmit hook names it on the next prompt. 12 unit tests, check_hook_test_coverage.py and check_no_comments pass. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018u8S5ct3kFhosinSbybc7W Change-Id: I1846b813ee3eeb6a1984eda8cbb8d39399c16d29 --- engine/hooks/unverified-tag-ledger/README.md | 78 +++++++ .../unverified-tag-ledger/claude.hook.json | 16 ++ .../claude.prompt.hook.json | 15 ++ .../claude_prompt_reminder.py | 31 +++ .../claude_stop_check.py | 30 +++ engine/hooks/unverified-tag-ledger/detect.py | 198 ++++++++++++++++++ .../unverified-tag-ledger/tests/test_hooks.py | 113 ++++++++++ install.sh | 1 + 8 files changed, 482 insertions(+) create mode 100644 engine/hooks/unverified-tag-ledger/README.md create mode 100644 engine/hooks/unverified-tag-ledger/claude.hook.json create mode 100644 engine/hooks/unverified-tag-ledger/claude.prompt.hook.json create mode 100644 engine/hooks/unverified-tag-ledger/claude_prompt_reminder.py create mode 100644 engine/hooks/unverified-tag-ledger/claude_stop_check.py create mode 100644 engine/hooks/unverified-tag-ledger/detect.py create mode 100644 engine/hooks/unverified-tag-ledger/tests/test_hooks.py diff --git a/engine/hooks/unverified-tag-ledger/README.md b/engine/hooks/unverified-tag-ledger/README.md new file mode 100644 index 0000000..d6625f6 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/README.md @@ -0,0 +1,78 @@ +# unverified-tag-ledger + +A well-formed `{{CAT-UNVERIFIED: -- cannot verify: }}` tag is a +**deferral**, not a discharge. This hook makes that true mechanically. + +## The gap it closes + +`cat-mode/SKILL.md:269` says: + +> Any hedge auto-runs prove-it in the same turn — a hedge is a trigger to +> verify, never a place to stop. + +Every evidence hook implemented the opposite. `_markers/markers.py` defines +`excuses_paragraph()`, and consumers treat a well-formed tag as equivalent to +evidence — for example `prove-it-ship-gate/detect.py`: + +```python +if markers.well_formed_tags(message) or has_evidence(message): +``` + +Only the *broken* tag shapes ever fired: `malformed_tags` (names no blocker) +and `has_legacy_marker` (bare `UNVERIFIED:`), both in +`diu-stop/claude_stop_check.py`. A correctly-formed tag produced silence from +the entire stack, was counted nowhere, and was revisited never. The written +rule said "never a place to stop" while the tooling rewarded stopping. + +Observed 2026-09-11 (NiceSpeak streaming session): two well-formed tags were +emitted, each ended its turn, and neither left a trace. Both are the positive +fixtures in `tests/test_hooks.py`. + +## What it does + +- **Stop** (`claude_stop_check.py`) — parses well-formed tags out of the reply + and appends them to a per-session ledger. It does **not** block. Blocking the + turn that emits a tag deadlocks, because the tag exists precisely for checks + that cannot run in that turn. +- **UserPromptSubmit** (`claude_prompt_reminder.py`) — lists outstanding claims + on the next prompt, quoting the rule and naming each claim plus what it is + blocked on. The next prompt is the earliest point a reminder can change + behaviour without preventing the turn from ending at all. +- **Discharge** — a claim is resolved when a later turn runs a verification tool + (`Bash`, `Read`, `Grep`, `Glob`, `NotebookRead`) and stops re-emitting it. +- **Escalation** — a claim outstanding `ESCALATE_AFTER_TURNS` (3) turns or more + is reported as a reflect trigger rather than accumulating quietly. + +Malformed tags are deliberately ignored here; `diu-stop` already rejects those. + +## Ledger + +`~/.cache/catstack-unverified-ledger/.jsonl`, one JSON row per +claim (`claim`, `reason`, `first_seen`, `turns`, `resolved`). Override the +directory with `CATSTACK_TAG_LEDGER_DIR` (the tests use a tempdir). A row that +is not JSON is reported on stderr and skipped, never silently dropped. + +## Tests + +``` +cd engine/hooks/unverified-tag-ledger && python3 -m unittest discover -s tests +``` + +12 tests: both real tags as positive fixtures, a no-tag negative control, the +malformed-tag negative, discharge-on-verify, stays-outstanding-without-verify, +no duplicate on re-emit, stale escalation, per-session isolation, and the +corrupt-row report. + +## Prior art + +The shape is a **defect-tracking rule**: a known-unresolved item is recorded +and re-surfaced rather than left to memory. Nancy G. Leveson, *CAST Handbook: +How to Learn More from Incidents and Accidents*, 2019 +(https://psas.scripts.mit.edu/home/get_file4.php?name=CAST_Handbook.pdf) names +the failure this prevents — "fixing the symptoms of problems but not tackling +the systemic causes" — by requiring the count be published before any single +item is called fixed. Saltzer and Schroeder, "Basic Principles of Information +Protection", 1975 +(https://web.mit.edu/Saltzer/www/publications/protection/Basic.html) supplies +the default: base the decision on explicit permission, so absence of a check is +never read as a pass. diff --git a/engine/hooks/unverified-tag-ledger/claude.hook.json b/engine/hooks/unverified-tag-ledger/claude.hook.json new file mode 100644 index 0000000..fc42fd5 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude.hook.json @@ -0,0 +1,16 @@ +{ + "hooks": { + "Stop": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/unverified-tag-ledger/claude_stop_check.py", + "timeout": 10 + } + ] + } + ] + } +} diff --git a/engine/hooks/unverified-tag-ledger/claude.prompt.hook.json b/engine/hooks/unverified-tag-ledger/claude.prompt.hook.json new file mode 100644 index 0000000..eb324de --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude.prompt.hook.json @@ -0,0 +1,15 @@ +{ + "hooks": { + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/unverified-tag-ledger/claude_prompt_reminder.py", + "timeout": 10 + } + ] + } + ] + } +} diff --git a/engine/hooks/unverified-tag-ledger/claude_prompt_reminder.py b/engine/hooks/unverified-tag-ledger/claude_prompt_reminder.py new file mode 100644 index 0000000..6e95a10 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude_prompt_reminder.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""Claude Code UserPromptSubmit hook: surface CAT-UNVERIFIED claims that +earlier turns deferred and never settled. This is where cat-mode/SKILL.md:269 +gets teeth -- the Stop hook cannot block the turn that emits a tag without +deadlocking, so the reminder lands on the next prompt instead. +""" +from __future__ import annotations + +import json +import sys + +from detect import reminder + + +def main() -> None: + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, OSError) as exc: + sys.stderr.write(f"unverified-tag-ledger: unreadable payload, no reminder: {exc!r}\n") + return + try: + text = reminder(str((payload or {}).get("session_id") or "")) + except Exception as exc: + sys.stderr.write(f"unverified-tag-ledger: reminder error, continuing: {exc!r}\n") + return + if text: + print(text) + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/unverified-tag-ledger/claude_stop_check.py b/engine/hooks/unverified-tag-ledger/claude_stop_check.py new file mode 100644 index 0000000..8438323 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude_stop_check.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +"""Claude Code Stop hook: record well-formed CAT-UNVERIFIED tags against the +session. Never blocks -- the tag exists for checks that cannot run now, so +blocking here would deadlock the turn. Fails open on read or parse errors. +""" +from __future__ import annotations + +import json +import sys + +from detect import decide_stop + + +def main() -> None: + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, OSError) as exc: + sys.stderr.write(f"unverified-tag-ledger: unreadable payload, allowing: {exc!r}\n") + return + try: + note = decide_stop(payload if isinstance(payload, dict) else {}) + except Exception as exc: + sys.stderr.write(f"unverified-tag-ledger: detector error, allowing this reply: {exc!r}\n") + return + if note: + sys.stderr.write(note + "\n") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/unverified-tag-ledger/detect.py b/engine/hooks/unverified-tag-ledger/detect.py new file mode 100644 index 0000000..abef373 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/detect.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""A well-formed CAT-UNVERIFIED tag is a deferral, not a discharge. + +Every other evidence hook treats `markers.well_formed_tags(message)` as +equivalent to evidence and goes silent (see prove-it-ship-gate/detect.py). +cat-mode/SKILL.md:269 says the opposite: "Any hedge auto-runs prove-it in the +same turn -- a hedge is a trigger to verify, never a place to stop." Nothing +reconciled the two, so a correctly-formed tag was a free, unlogged exit. + +This module does not re-block the turn that emits a tag; blocking there +deadlocks, because the tag exists precisely for checks that cannot run now. +It records the tag against the session and surfaces it on the next prompt, +which is the earliest point a reminder can change behaviour without +preventing the turn from ending at all. + +A tag is discharged when a later turn runs a verification tool and stops +re-emitting it. Turn count since first sight is kept so a tag that survives +many turns can be escalated rather than quietly accumulating. +""" +from __future__ import annotations + +import json +import os +import re +import sys +import time + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_markers")) + +import markers # noqa: E402 + +VERIFY_TOOLS = {"Bash", "Read", "Grep", "Glob", "NotebookRead"} +ESCALATE_AFTER_TURNS = 3 +MAX_LISTED = 5 + +CLAIM_RE = re.compile( + r"\{\{\s*CAT-UNVERIFIED\s*:?\s*(?P.*?)(?:--|—)\s*cannot\s+verify\s*:\s*(?P[^}]*)\}\}", + re.IGNORECASE | re.DOTALL, +) + + +def ledger_dir() -> str: + base = os.environ.get("CATSTACK_TAG_LEDGER_DIR") + if base: + return base + return os.path.join(os.path.expanduser("~"), ".cache", "catstack-unverified-ledger") + + +def ledger_path(session_id: str) -> str: + safe = re.sub(r"[^A-Za-z0-9_.-]", "_", session_id or "unknown") + return os.path.join(ledger_dir(), f"{safe}.jsonl") + + +def parse_tags(message: str) -> list[dict]: + """Well-formed tags only, split into claim and reason.""" + out = [] + for raw in markers.well_formed_tags(message or ""): + match = CLAIM_RE.search(raw) + if not match: + continue + claim = " ".join(match.group("claim").split()) + reason = " ".join(match.group("reason").split()) + if claim and reason: + out.append({"claim": claim, "reason": reason}) + return out + + +def read_ledger(session_id: str) -> list[dict]: + path = ledger_path(session_id) + if not os.path.exists(path): + return [] + rows = [] + with open(path, encoding="utf-8") as handle: + for number, line in enumerate(handle, start=1): + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError as exc: + sys.stderr.write( + f"unverified-tag-ledger: {path}:{number} is not JSON, skipping row: {exc}\n") + return rows + + +def write_ledger(session_id: str, rows: list[dict]) -> None: + path = ledger_path(session_id) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, sort_keys=True) + "\n") + + +def outstanding(rows: list[dict]) -> list[dict]: + return [row for row in rows if not row.get("resolved")] + + +def record_turn(session_id: str, message: str, tools_used: set[str], now=None) -> list[dict]: + """Log new tags, discharge ones this turn verified and dropped.""" + stamp = now() if now else time.time() + rows = read_ledger(session_id) + present = {tag["claim"] for tag in parse_tags(message)} + verified = bool(tools_used & VERIFY_TOOLS) + + for row in rows: + if row.get("resolved"): + continue + if row["claim"] in present: + row["turns"] = row.get("turns", 0) + 1 + elif verified: + row["resolved"] = True + row["resolved_at"] = stamp + else: + row["turns"] = row.get("turns", 0) + 1 + + known = {row["claim"] for row in rows} + for tag in parse_tags(message): + if tag["claim"] in known: + continue + rows.append({ + "claim": tag["claim"], + "reason": tag["reason"], + "first_seen": stamp, + "turns": 0, + "resolved": False, + }) + + write_ledger(session_id, rows) + return rows + + +def reminder(session_id: str) -> str: + """Text for UserPromptSubmit, or empty when nothing is outstanding.""" + open_rows = outstanding(read_ledger(session_id)) + if not open_rows: + return "" + stale = [row for row in open_rows if row.get("turns", 0) >= ESCALATE_AFTER_TURNS] + lines = [ + "unverified-tag-ledger: " + f"{len(open_rows)} CAT-UNVERIFIED claim(s) from earlier turns are still unverified.", + "cat-mode/SKILL.md:269 -- a hedge is a trigger to verify, never a place to stop. " + "The tag deferred these; it did not settle them.", + ] + for row in open_rows[:MAX_LISTED]: + lines.append(f" - {row['claim']} (blocked on: {row['reason']}; {row.get('turns', 0)} turn(s) old)") + if len(open_rows) > MAX_LISTED: + lines.append(f" ... and {len(open_rows) - MAX_LISTED} more") + lines.append( + "For each: run the check now and paste its output, or say plainly that it is still blocked and why. " + "The user can retire one by saying to drop it.") + if stale: + lines.append( + f"{len(stale)} of these are {ESCALATE_AFTER_TURNS}+ turns old -- that is a reflect trigger, " + "not a backlog item.") + return "\n".join(lines) + + +def decide_stop(payload: dict) -> str: + session_id = str(payload.get("session_id") or "") + message = _last_assistant_text(payload) + tools = _tools_used(payload) + rows = record_turn(session_id, message, tools) + new_claims = {tag["claim"] for tag in parse_tags(message)} + if not new_claims: + return "" + 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 "" + return ( + 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).") + + +def _last_assistant_text(payload: dict) -> str: + for key in ("last_assistant_message", "assistant_message", "message"): + value = payload.get(key) + if isinstance(value, str) and value.strip(): + return value + transcript = payload.get("transcript") or [] + if isinstance(transcript, list): + for entry in reversed(transcript): + if isinstance(entry, dict) and entry.get("role") == "assistant": + content = entry.get("content") + if isinstance(content, str): + return content + return "" + + +def _tools_used(payload: dict) -> set[str]: + raw = payload.get("tools_used") or payload.get("tool_names") or [] + if isinstance(raw, str): + return {raw} + if isinstance(raw, list): + return {str(item) for item in raw} + return set() diff --git a/engine/hooks/unverified-tag-ledger/tests/test_hooks.py b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py new file mode 100644 index 0000000..9fd0dfe --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""The positive fixtures are the two real tags from the session that motivated +this hook (2026-09-11, NiceSpeak streaming): both were emitted, both ended the +turn, neither left a trace anywhere. +""" +from __future__ import annotations + +import os +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +HOOK = os.path.dirname(HERE) +sys.path.insert(0, HOOK) + +REAL_TAG_1 = ( + "{{CAT-UNVERIFIED: that it widened scope past the one session I gave it " + "-- cannot verify: it never answered when asked twice; its " + '"21,298,308 tokens / three sessions" line is the only signal}}') +REAL_TAG_2 = ( + "{{CAT-UNVERIFIED: that I told you to plug in the phone because I trusted the status string " + "-- cannot verify: my own reasoning isn't observable by any command}}") +MALFORMED = "{{CAT-UNVERIFIED: something I did not check}}" + + +class LedgerTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + os.environ["CATSTACK_TAG_LEDGER_DIR"] = self.tmp.name + for module in ("detect", "markers"): + sys.modules.pop(module, None) + import detect + self.detect = detect + + def tearDown(self) -> None: + os.environ.pop("CATSTACK_TAG_LEDGER_DIR", None) + self.tmp.cleanup() + + def test_real_tag_is_parsed_into_claim_and_reason(self) -> None: + tags = self.detect.parse_tags(f"Some prose.\n\n{REAL_TAG_1}") + self.assertEqual(len(tags), 1) + self.assertIn("widened scope", tags[0]["claim"]) + self.assertIn("never answered", tags[0]["reason"]) + + def test_malformed_tag_is_not_logged(self) -> None: + self.detect.record_turn("s1", MALFORMED, set()) + self.assertEqual(self.detect.read_ledger("s1"), []) + + def test_message_with_no_tag_leaves_ledger_empty(self) -> None: + self.detect.record_turn("s1", "Ran the tests, 59/59 pass.", {"Bash"}) + self.assertEqual(self.detect.read_ledger("s1"), []) + self.assertEqual(self.detect.reminder("s1"), "") + + def test_emitting_a_tag_logs_it_without_blocking(self) -> None: + note = self.detect.decide_stop({"session_id": "s1", "message": REAL_TAG_1}) + self.assertIn("deferred, not discharged", note) + self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 1) + + def test_reminder_names_the_claim_and_cites_the_rule(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + text = self.detect.reminder("s1") + self.assertIn("widened scope", text) + self.assertIn("cat-mode/SKILL.md:269", text) + self.assertIn("never a place to stop", text) + + def test_two_tags_in_one_session_both_tracked(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.detect.record_turn("s1", REAL_TAG_2, set()) + self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 2) + + def test_verified_and_dropped_tag_is_discharged(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.detect.record_turn("s1", "Here is the pasted output proving it.", {"Bash"}) + self.assertEqual(self.detect.outstanding(self.detect.read_ledger("s1")), []) + self.assertEqual(self.detect.reminder("s1"), "") + + def test_redropping_without_verifying_keeps_it_outstanding(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.detect.record_turn("s1", "Moving on to something else.", set()) + self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 1) + + def test_reemitting_the_same_tag_does_not_duplicate_it(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.assertEqual(len(self.detect.read_ledger("s1")), 1) + + def test_stale_tag_escalates_to_reflect(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + for _ in range(self.detect.ESCALATE_AFTER_TURNS): + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.assertIn("reflect trigger", self.detect.reminder("s1")) + + def test_sessions_do_not_leak_into_each_other(self) -> None: + self.detect.record_turn("s1", REAL_TAG_1, set()) + self.assertEqual(self.detect.reminder("s2"), "") + + 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) + with open(path, "w", encoding="utf-8") as handle: + handle.write("{not json}\n") + import io + from contextlib import redirect_stderr + buffer = io.StringIO() + with redirect_stderr(buffer): + rows = self.detect.read_ledger("s3") + self.assertEqual(rows, []) + self.assertIn("is not JSON", buffer.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/install.sh b/install.sh index 146dddc..c1360a0 100755 --- a/install.sh +++ b/install.sh @@ -248,6 +248,7 @@ echo "--- claude hooks: wait / hedge / callout stack ---" link_item "wait-needs-wakeup" "$REPO_DIR/engine/hooks/wait-needs-wakeup" "$HOME/.claude/hooks/wait-needs-wakeup" link_item "hedge-runs-prove-it" "$REPO_DIR/engine/hooks/hedge-runs-prove-it" "$HOME/.claude/hooks/hedge-runs-prove-it" link_item "gate-blame-needs-evidence" "$REPO_DIR/engine/hooks/gate-blame-needs-evidence" "$HOME/.claude/hooks/gate-blame-needs-evidence" +link_item "unverified-tag-ledger" "$REPO_DIR/engine/hooks/unverified-tag-ledger" "$HOME/.claude/hooks/unverified-tag-ledger" link_item "incidence-needs-repetition" "$REPO_DIR/engine/hooks/incidence-needs-repetition" "$HOME/.claude/hooks/incidence-needs-repetition" link_item "verdict-flip-watch" "$REPO_DIR/engine/hooks/verdict-flip-watch" "$HOME/.claude/hooks/verdict-flip-watch" link_item "new-file-callout" "$REPO_DIR/engine/hooks/new-file-callout" "$HOME/.claude/hooks/new-file-callout"