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"