Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions engine/hooks/unverified-tag-ledger/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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.
Expand Down
17 changes: 11 additions & 6 deletions engine/hooks/unverified-tag-ledger/claude_stop_check.py
Original file line number Diff line number Diff line change
@@ -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:
Expand All @@ -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__":
Expand Down
37 changes: 32 additions & 5 deletions engine/hooks/unverified-tag-ledger/detect.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
37 changes: 34 additions & 3 deletions engine/hooks/unverified-tag-ledger/tests/test_hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading