diff --git a/engine/hooks/unverified-tag-ledger/README.md b/engine/hooks/unverified-tag-ledger/README.md index d6625f6..37dd7eb 100644 --- a/engine/hooks/unverified-tag-ledger/README.md +++ b/engine/hooks/unverified-tag-ledger/README.md @@ -31,9 +31,12 @@ 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. + 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 @@ -58,7 +61,7 @@ is not JSON is reported on stderr and skipped, never silently dropped. 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 +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. diff --git a/engine/hooks/unverified-tag-ledger/claude_stop_check.py b/engine/hooks/unverified-tag-ledger/claude_stop_check.py index 8438323..b5a4e13 100644 --- a/engine/hooks/unverified-tag-ledger/claude_stop_check.py +++ b/engine/hooks/unverified-tag-ledger/claude_stop_check.py @@ -1,14 +1,16 @@ #!/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. +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 decide_stop +from detect import evaluate def main() -> None: @@ -18,12 +20,15 @@ def main() -> None: sys.stderr.write(f"unverified-tag-ledger: unreadable payload, allowing: {exc!r}\n") return try: - note = decide_stop(payload if isinstance(payload, dict) else {}) + 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 note: - sys.stderr.write(note + "\n") + if verdict["block"]: + sys.stderr.write(verdict["block"] + "\n") + sys.exit(2) + if verdict["note"]: + sys.stderr.write(verdict["note"] + "\n") if __name__ == "__main__": diff --git a/engine/hooks/unverified-tag-ledger/detect.py b/engine/hooks/unverified-tag-ledger/detect.py index abef373..c5b70f9 100644 --- a/engine/hooks/unverified-tag-ledger/detect.py +++ b/engine/hooks/unverified-tag-ledger/detect.py @@ -156,22 +156,49 @@ def reminder(session_id: str) -> str: return "\n".join(lines) -def decide_stop(payload: dict) -> str: +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 "" + 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 "" - return ( + 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).") + "(cat-mode/SKILL.md:269)."), "block": ""} + + +def decide_stop(payload: dict) -> str: + return evaluate(payload)["note"] def _last_assistant_text(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 9fd0dfe..88f0383 100644 --- a/engine/hooks/unverified-tag-ledger/tests/test_hooks.py +++ b/engine/hooks/unverified-tag-ledger/tests/test_hooks.py @@ -52,11 +52,42 @@ def test_message_with_no_tag_leaves_ledger_empty(self) -> None: 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) + 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")