diff --git a/docs/ecosystem.md b/docs/ecosystem.md index be12148..4dc59e7 100644 --- a/docs/ecosystem.md +++ b/docs/ecosystem.md @@ -187,8 +187,10 @@ A subagent launched through the Agent tool runs under the same which mirrors each `engine/hooks//claude*.hook.json` `Stop` entry, and a hook opts out only in its own manifest with `"subagent_stop": {"inherit": false, "reason": "..."}` (today: -`frustration-watchdog`, which reads the human's last message, and `auto-pr`, -whose PR instruction is for the session owner). Under `SubagentStop`, +`frustration-watchdog`, which reads the human's last message, `auto-pr`, +whose PR instruction is for the session owner, and `unverified-tag-ledger`, +whose ledger is keyed by session id and whose reminder needs a next user +prompt). Under `SubagentStop`, `transcript_path` is the parent's transcript and `agent_transcript_path` is the subagent's own, so transcript-reading hooks prefer the latter. `UserPromptSubmit` hooks never reach a subagent, because its prompt arrives diff --git a/engine/hooks/unverified-tag-ledger/README.md b/engine/hooks/unverified-tag-ledger/README.md new file mode 100644 index 0000000..3bd5600 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/README.md @@ -0,0 +1,81 @@ +# 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` (the retired bare `UNVERIFIED:` form), 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, then **refuses the turn** (exit 2) + if it tagged a claim without running any verification tool. A tag earns its + place only after an attempt: requiring an attempt is not requiring success, so + run the check and tag it only when the check cannot settle the claim. + `stop_hook_active` releases the refusal, or the rewrite turn — which has no + tool call of its own — would loop forever. +- **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 +``` + +17 tests: both real tags as positive fixtures, the refusal on an untried tag, the `stop_hook_active` release that prevents a refusal loop, 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..800014f --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude.hook.json @@ -0,0 +1,20 @@ +{ + "hooks": { + "Stop": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/unverified-tag-ledger/claude_stop_check.py", + "timeout": 10 + } + ] + } + ] + }, + "subagent_stop": { + "inherit": false, + "reason": "the ledger and its next-prompt reminder belong to the session owner; a subagent shares the parent's session id and would log its own tags against the parent's ledger, and it has no next user prompt to be reminded at" + } +} 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..b5a4e13 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/claude_stop_check.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Claude Code Stop hook: record well-formed CAT-UNVERIFIED tags against the +session, and refuse a turn that tags a claim without having run any +verification tool (cat-mode/SKILL.md:269 -- a hedge is a trigger to verify). +`stop_hook_active` releases the refusal so the rewrite turn can finish. Fails +open on read or parse errors. +""" +from __future__ import annotations + +import json +import sys + +from detect import evaluate + + +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: + verdict = evaluate(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 verdict["block"]: + sys.stderr.write(verdict["block"] + "\n") + sys.exit(2) + if verdict["note"]: + sys.stderr.write(verdict["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..c5b70f9 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/detect.py @@ -0,0 +1,225 @@ +#!/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 evaluate(payload: dict) -> dict: + """Record the turn, then decide whether it may end. + + A tag earns its place only after an attempt. cat-mode/SKILL.md:269 asks for + a verify in the SAME turn, so a tag emitted by a turn that ran no + verification tool is a claim nobody tried to check, and that turn is + refused. Requiring an attempt is not requiring success: run the check, and + if it cannot run or comes back inconclusive, the tag is then honest. + + `stop_hook_active` releases the block so the rewrite turn can finish -- + without it the refusal loops forever, because a reply being rewritten to + satisfy this hook has no tool call of its own either. + """ + 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 {"note": "", "block": ""} + + if not tools & VERIFY_TOOLS and not payload.get("stop_hook_active"): + claims = "; ".join(sorted(new_claims)[:MAX_LISTED]) + return {"note": "", "block": ( + "unverified-tag-ledger: this turn tags a claim as unverified but ran no " + f"verification tool. Untried claim(s): {claims}. " + "cat-mode/SKILL.md:269 -- a hedge is a trigger to verify, never a place to stop. " + "Run the check now (Bash/Read/Grep/Glob) and paste its output. The tag is for a " + "check that was attempted and could not settle the claim, not for one nobody ran.")} + + 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": ""} + + +def decide_stop(payload: dict) -> str: + return evaluate(payload)["note"] + + +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/install_claude_hook.py b/engine/hooks/unverified-tag-ledger/install_claude_hook.py new file mode 100644 index 0000000..a55049b --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/install_claude_hook.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""Idempotently merge unverified-tag-ledger's Claude Code hooks into +~/.claude/settings.json: the Stop hook (records tags and refuses a turn that +tagged without checking) and the UserPromptSubmit hook (re-surfaces claims +earlier turns deferred and never settled). + +Each entry in HOOK_SPECS is (hook type in settings.json, marker identifying this +hook's own entry, fragment file). The marker is directory-qualified: diu-stop's +scripts carry the same basenames, so a bare "claude_stop_check.py" marker would +match diu-stop's entries and delete them instead of this hook's. + +Safe to rerun on every `install.sh`: each hook type is identified by whether +any of its entries' "command" mentions that hook's own marker script, +replaces just that hook type's unverified-tag-ledger entry with the current fragment +file's content, and leaves every other key in settings.json (model, theme, +other hooks, other hook types, ...) untouched. That means editing either +fragment file and rerunning install.sh converges cleanly instead of +appending a duplicate entry each time. +""" +import json +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +SETTINGS_PATH = os.path.expanduser("~/.claude/settings.json") + +HOOK_SPECS = [ + ("Stop", "unverified-tag-ledger/claude_stop_check.py", os.path.join(HERE, "claude.hook.json")), + ("UserPromptSubmit", "unverified-tag-ledger/claude_prompt_reminder.py", os.path.join(HERE, "claude.prompt.hook.json")), +] + + +def _is_ours(entry, marker): + return any(marker in h.get("command", "") for h in entry.get("hooks", [])) + + +def merge_hook(settings, hook_type, marker, fragment): + """Pure: returns (new_settings, changed). Replaces any existing + unverified-tag-ledger entry of this hook type with fragment's, appends if none + existed yet.""" + settings = json.loads(json.dumps(settings)) + entry_list = settings.setdefault("hooks", {}).setdefault(hook_type, []) + new_entries = fragment["hooks"][hook_type] + + before = json.dumps(entry_list, sort_keys=True) + entry_list[:] = [e for e in entry_list if not _is_ours(e, marker)] + new_entries + changed = json.dumps(entry_list, sort_keys=True) != before + return settings, changed + + +def main(): + settings = {} + if os.path.exists(SETTINGS_PATH): + with open(SETTINGS_PATH) as f: + settings = json.load(f) + + any_changed = False + for hook_type, marker, fragment_path in HOOK_SPECS: + with open(fragment_path) as f: + fragment = json.load(f) + settings, changed = merge_hook(settings, hook_type, marker, fragment) + if changed: + any_changed = True + print(f"link claude {hook_type} hook merged into settings.json") + else: + print(f"ok claude {hook_type} hook already up to date") + + if not any_changed: + return + + os.makedirs(os.path.dirname(SETTINGS_PATH), exist_ok=True) + with open(SETTINGS_PATH, "w") as f: + json.dump(settings, f, indent=2) + f.write("\n") + print(" (restart Claude Code to pick up the change)") + + +if __name__ == "__main__": + main() 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..88f0383 --- /dev/null +++ b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py @@ -0,0 +1,144 @@ +#!/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_tag_after_a_real_attempt_is_logged_and_allowed(self) -> None: + verdict = self.detect.evaluate( + {"session_id": "s1", "message": REAL_TAG_1, "tools_used": ["Bash"]}) + self.assertEqual(verdict["block"], "") + self.assertIn("deferred, not discharged", verdict["note"]) + self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 1) + + def test_tag_with_no_attempt_is_blocked(self) -> None: + verdict = self.detect.evaluate( + {"session_id": "s1", "message": REAL_TAG_1, "tools_used": []}) + self.assertIn("ran no verification tool", verdict["block"]) + self.assertIn("cat-mode/SKILL.md:269", verdict["block"]) + self.assertIn("widened scope", verdict["block"]) + + def test_blocked_turn_is_still_recorded(self) -> None: + self.detect.evaluate({"session_id": "s1", "message": REAL_TAG_1, "tools_used": []}) + self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 1) + + def test_rewrite_turn_is_released_so_the_block_cannot_loop(self) -> None: + verdict = self.detect.evaluate({ + "session_id": "s1", "message": REAL_TAG_1, + "tools_used": [], "stop_hook_active": True}) + self.assertEqual(verdict["block"], "") + + def test_untagged_turn_with_no_tools_is_never_blocked(self) -> None: + verdict = self.detect.evaluate( + {"session_id": "s1", "message": "Short answer, nothing claimed.", "tools_used": []}) + self.assertEqual(verdict["block"], "") + self.assertEqual(verdict["note"], "") + + def test_both_real_session_turns_would_have_been_blocked(self) -> None: + for tag in (REAL_TAG_1, REAL_TAG_2): + verdict = self.detect.evaluate( + {"session_id": "replay", "message": f"prose\n\n{tag}", "tools_used": []}) + self.assertIn("ran no verification tool", verdict["block"]) + + 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..a8acfac 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" @@ -334,6 +335,7 @@ fi # either file. See each script's docstring for exactly what it does. echo "--- claude Stop + UserPromptSubmit hooks (\$HOME/.claude/settings.json) ---" python3 "$REPO_DIR/engine/hooks/diu-stop/install_claude_hook.py" +python3 "$REPO_DIR/engine/hooks/unverified-tag-ledger/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/bug-complaint-leak/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/reflect-on-thrash/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/scope-lock/install_claude_hook.py" diff --git a/tests/test_mirror_subagent_stop.py b/tests/test_mirror_subagent_stop.py index f1e4a5f..fa4cec2 100644 --- a/tests/test_mirror_subagent_stop.py +++ b/tests/test_mirror_subagent_stop.py @@ -16,7 +16,7 @@ import mirror_stop_hooks_to_subagent_stop as mod # noqa: E402 -OPTED_OUT_ON_MAIN = {"frustration-watchdog", "auto-pr"} +OPTED_OUT_ON_MAIN = {"frustration-watchdog", "auto-pr", "unverified-tag-ledger"} def stop_entry(name: str, script: str = "claude_stop_check.py") -> dict: