hook: backstop layer catches a stale verdict without needing the model to notice

engine/hooks/verdict-flip-watch/README.md

@@ -0,0 +1,60 @@
+# verdict-flip-watch
+
+Third and last layer of the self-correction guard. Advisory Stop hook.
+
+A verifier that printed `ok` earlier in the session and failed later means any
+status reported off the earlier run is stale — as a matter of record, not
+interpretation. This hook reads the transcript's own Bash commands and their
+`tool_result` output, pairs them, and notes the first target whose verdict
+flipped from pass to fail. It stays silent when the outgoing message already
+mentions it (`wrong`, `vacuous`, `stale`, `retract`, `now fails`, …).
+
+## Why a third layer
+
+The two layers above it both depend on the model:
+
+1. `wrong-check-reflect` matches the shape of a retraction in the outgoing
+   text — so it only helps once the model has decided to admit something.
+2. `principle-flag-your-own-corrections` carries the judgment no regex can
+   enumerate — but still needs the model to notice the claim went stale.
+
+This hook needs neither. It catches the silent switch to the corrected value,
+which is the failure `principle-flag-your-own-corrections` names: the user
+cannot tell a silent correction from consistency.
+
+The live case it was written for: `check_skill_test_coverage.py` printed `ok`
+for a stacked slice it never compared, that result was reported as "fully
+green", and the same script failed once it was given the slice refs. Nothing
+mechanical connected the two runs.
+
+## Scope
+
+Only verifier-shaped commands count — `check_*`, `test_*`, `run_*`, pytest,
+unittest, npm/pnpm test, cargo/go test, make test/check, jest, vitest,
+preflight. Tracking every `ls` would make a flip meaningless.
+
+`fail` wins over `pass` when both appear in one output, because a run can
+print `ok` lines for early gates and still fail overall.
+
+Advisory on purpose: exit 0, stderr. A gate can legitimately start failing
+because the turn broke it deliberately, and the hook cannot know intent. Once
+per transcript per target. Fail-open on any parse or IO error.
+
+## Files
+
+- `detect.py` — command/result pairing, verdict classification, `decide()`
+- `claude_stop_check.py` — Claude `Stop` entrypoint (stderr, exit 0)
+- `claude.hook.json` / `install_claude_hook.py` — settings.json merge (idempotent)
+- `tests/fixtures/` — transcripts that fire and stay silent
+- `tests/test_hooks.py`
+
+## Install
+
+`./install.sh` from the repo root. Restart the harness.
+
+## Tests
+
+```sh
+python3 -m unittest discover -s engine/hooks/verdict-flip-watch/tests -v
+python3 scripts/check_hook_test_coverage.py engine/hooks/verdict-flip-watch
+```

engine/hooks/verdict-flip-watch/claude.hook.json

@@ -0,0 +1,16 @@
+{
+  "hooks": {
+    "Stop": [
+      {
+        "matcher": "*",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 $HOME/.claude/hooks/verdict-flip-watch/claude_stop_check.py",
+            "timeout": 10
+          }
+        ]
+      }
+    ]
+  }
+}

engine/hooks/verdict-flip-watch/claude_stop_check.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+"""Claude Stop hook: note a verifier that passed earlier and failed later.
+
+Advisory — stderr plus exit 0 — because a gate can legitimately start failing
+when the turn broke it on purpose. Fail-open on any read/parse error.
+"""
+from __future__ import annotations
+
+import json
+import sys
+
+from detect import decide
+
+
+def main() -> None:
+    try:
+        payload = json.load(sys.stdin)
+    except (json.JSONDecodeError, OSError):
+        return
+    try:
+        message = decide(payload if isinstance(payload, dict) else {})
+    except Exception:
+        return
+    if message:
+        sys.stderr.write(message + "\n")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/verdict-flip-watch/detect.py

@@ -0,0 +1,209 @@
+"""Backstop layer: a check that passed earlier and failed later makes any claim
+made off the earlier run stale -- whether or not the model noticed.
+
+Three layers guard self-correction, weakest dependency last:
+
+1. `wrong-check-reflect` matches the SHAPE of a retraction in the outgoing
+   text. It only fires once the model has already decided to admit something.
+2. `principle-flag-your-own-corrections` carries the judgment no regex can
+   enumerate. It still requires the model to realise a claim went stale.
+3. This hook requires neither. It reads the transcript's own command results.
+   If `scripts/x.py` printed ok at one point and failed later, the earlier
+   "green" was wrong as a matter of record, and the only question is whether
+   this message says so.
+
+That third property is the point: it catches the silent switch to the
+corrected value, which `principle-flag-your-own-corrections` names as the real
+failure -- the user cannot tell a silent correction from consistency.
+
+Advisory: stderr plus exit 0. A gate can legitimately start failing because
+the turn broke something on purpose, so this informs rather than blocks.
+Fail-open on any parse or IO error.
+
+VERIFIER_RE is what counts as a command worth tracking: a checker, test runner,
+or build. Tracking every `ls` and `git status` would make a flip meaningless.
+
+ACKNOWLEDGED_RE is the outgoing message already owning the flip: a correction
+marker, or the word stale/vacuous near the verdict. It reuses the
+wrong-check-reflect vocabulary.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+
+STATE_DIR = os.environ.get(
+    "VERDICT_FLIP_WATCH_STATE_DIR",
+    os.path.join(os.path.expanduser("~"), ".cache", "catstack-verdict-flip-watch"),
+)
+
+VERIFIER_RE = re.compile(
+    r"(?:^|[\s/])(?:check_|test_|run_)|"
+    r"\b(?:pytest|unittest|npm\s+(?:test|run\s+\w*test\w*)|pnpm\s+(?:test|run)|"
+    r"cargo\s+test|go\s+test|make\s+(?:test|check)|tox|jest|vitest|preflight)\b",
+    re.IGNORECASE,
+)
+SCRIPTISH_RE = re.compile(r"[\w./-]+\.(?:py|sh|mjs|js|ts)\b")
+
+FAIL_RE = re.compile(
+    r"(?:^|\n)\s*(?:fail|FAILED|ERROR)\b|\bTraceback\b|\bFAILED\s*\(|"
+    r"\bexit(?:\s*code)?[\s=:]+[1-9]\b|\bassertionerror\b",
+    re.IGNORECASE,
+)
+PASS_RE = re.compile(
+    r"(?:^|\n)\s*ok\b|\bOK\s*$|\bpassed\b|\ball\s+\d+\s+\w+\s+(?:passed|ok)\b|"
+    r"\bexit(?:\s*code)?[\s=:]+0\b",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+ACKNOWLEDGED_RE = re.compile(
+    r"(?i)\b(?:wrong|incorrect|vacuous|stale|retract(?:ing|ed)?|mistaken|"
+    r"misread|premature|no\s+longer\s+(?:true|holds)|"
+    r"does(?:n'?t|\s+not)\s+hold|now\s+fails|started\s+failing|"
+    r"earlier\s+(?:run|pass|result|claim))\b"
+)
+
+MESSAGE = (
+    "verdict-flip-watch: `{target}` passed earlier in this session and failed "
+    "later, and this message does not mention it. Any status reported off the "
+    "earlier run is stale. Say which claim is affected and what changed, then "
+    "follow principle-flag-your-own-corrections (the admission is a reflect "
+    "trigger, not just a sentence). If the flip is expected because this turn "
+    "broke it on purpose, say that instead."
+)
+
+
+def _blocks(data: dict) -> list:
+    message = data.get("message")
+    content = message.get("content") if isinstance(message, dict) else None
+    return content if isinstance(content, list) else []
+
+
+def normalize_target(command: str) -> str | None:
+    """The thing being verified, so two runs of it can be compared."""
+    if not VERIFIER_RE.search(command or ""):
+        return None
+    script = SCRIPTISH_RE.search(command)
+    if script:
+        return script.group(0)
+    words = (command or "").split()
+    return " ".join(words[:2]) if words else None
+
+
+def classify(output: str) -> str:
+    """'fail' wins over 'pass': a run can print ok lines and still fail."""
+    if FAIL_RE.search(output or ""):
+        return "fail"
+    if PASS_RE.search(output or ""):
+        return "pass"
+    return "unknown"
+
+
+def result_text(data: dict) -> str:
+    out: list[str] = []
+    for block in _blocks(data):
+        if not isinstance(block, dict) or block.get("type") != "tool_result":
+            continue
+        content = block.get("content")
+        if isinstance(content, str):
+            out.append(content)
+        elif isinstance(content, list):
+            for part in content:
+                if isinstance(part, dict) and isinstance(part.get("text"), str):
+                    out.append(part["text"])
+    return "\n".join(out)
+
+
+def verdicts(transcript_path: str) -> list[tuple[str, str]]:
+    """(target, 'pass'|'fail') in transcript order. [] when unreadable."""
+    try:
+        with open(transcript_path, encoding="utf-8") as handle:
+            lines = handle.readlines()
+    except OSError:
+        return []
+    pending: list[str] = []
+    found: list[tuple[str, str]] = []
+    for raw in lines:
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(data, dict):
+            continue
+        if data.get("type") == "assistant":
+            for block in _blocks(data):
+                if (
+                    isinstance(block, dict)
+                    and block.get("type") == "tool_use"
+                    and block.get("name") == "Bash"
+                ):
+                    command = (block.get("input") or {}).get("command")
+                    target = normalize_target(command) if isinstance(command, str) else None
+                    if target:
+                        pending.append(target)
+            continue
+        text = result_text(data)
+        if not text or not pending:
+            continue
+        verdict = classify(text)
+        target = pending.pop(0)
+        if verdict != "unknown":
+            found.append((target, verdict))
+    return found
+
+
+def find_flip(transcript_path: str) -> str | None:
+    """A target that passed and then later failed, else None."""
+    passed: set[str] = set()
+    for target, verdict in verdicts(transcript_path):
+        if verdict == "pass":
+            passed.add(target)
+        elif verdict == "fail" and target in passed:
+            return target
+    return None
+
+
+def _state_file(transcript_path: str, target: str) -> str:
+    key = f"{os.path.abspath(transcript_path or 'none')}::{target}"
+    digest = hashlib.sha1(key.encode()).hexdigest()[:16]
+    return os.path.join(STATE_DIR, f"{digest}.noted")
+
+
+def already_noted(transcript_path: str, target: str) -> bool:
+    return os.path.isfile(_state_file(transcript_path, target))
+
+
+def mark_noted(transcript_path: str, target: str) -> None:
+    path = _state_file(transcript_path, target)
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write(target + "\n")
+    except OSError:
+        pass
+
+
+def decide(payload: dict) -> str | None:
+    if payload.get("stop_hook_active"):
+        return None
+    message = payload.get("last_assistant_message") or ""
+    if ACKNOWLEDGED_RE.search(message):
+        return None
+    transcript_path = (
+        payload.get("agent_transcript_path")
+        or payload.get("transcript_path")
+        or payload.get("transcriptPath")
+        or ""
+    )
+    if not transcript_path:
+        return None
+    try:
+        target = find_flip(transcript_path)
+    except Exception:
+        return None
+    if not target or already_noted(transcript_path, target):
+        return None
+    mark_noted(transcript_path, target)
+    return MESSAGE.format(target=target)

engine/hooks/verdict-flip-watch/install_claude_hook.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""Merge verdict-flip-watch into ~/.claude/settings.json Stop hooks. Idempotent."""
+from __future__ import annotations
+
+import json
+import os
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+SETTINGS_PATH = os.path.expanduser("~/.claude/settings.json")
+FRAGMENT_PATH = os.path.join(HERE, "claude.hook.json")
+MARKER = "verdict-flip-watch/claude_stop_check.py"
+EVENT = "Stop"
+
+
+def _is_ours(entry: dict) -> bool:
+    return any(MARKER in h.get("command", "") for h in entry.get("hooks", []))
+
+
+def merge_hook(settings: dict, fragment: dict) -> bool:
+    entry_list = settings.setdefault("hooks", {}).setdefault(EVENT, [])
+    new_entries = fragment.get("hooks", {}).get(EVENT, [])
+    before = json.dumps(entry_list, sort_keys=True)
+    kept = [e for e in entry_list if not _is_ours(e)]
+    entry_list[:] = kept + new_entries
+    return json.dumps(entry_list, sort_keys=True) != before
+
+
+def main() -> None:
+    settings: dict = {}
+    if os.path.exists(SETTINGS_PATH):
+        with open(SETTINGS_PATH) as handle:
+            settings = json.load(handle)
+    with open(FRAGMENT_PATH) as handle:
+        fragment = json.load(handle)
+    if not merge_hook(settings, fragment):
+        print("ok      claude Stop verdict-flip-watch already up to date")
+        return
+    os.makedirs(os.path.dirname(SETTINGS_PATH), exist_ok=True)
+    with open(SETTINGS_PATH, "w") as handle:
+        json.dump(settings, handle, indent=2)
+        handle.write("\n")
+    print("added   claude Stop verdict-flip-watch")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/verdict-flip-watch/tests/fixtures/fixed.jsonl

@@ -0,0 +1,5 @@
+{"type": "user", "message": {"role": "user", "content": "fix the gate"}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 -m unittest discover -s engine/hooks/x/tests"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "FAILED (failures=1)"}]}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 -m unittest discover -s engine/hooks/x/tests"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "Ran 12 tests in 0.2s\n\nOK"}]}}

engine/hooks/verdict-flip-watch/tests/fixtures/flip.jsonl

@@ -0,0 +1,6 @@
+{"type": "user", "message": {"role": "user", "content": "verify the stacked slice"}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 scripts/check_skill_test_coverage.py"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "ok      skill test coverage"}]}}
+{"type": "user", "message": {"role": "user", "content": "run it against the slice refs"}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 scripts/check_skill_test_coverage.py --base skill-scenarios --head HEAD"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "fail  engine/skills/make-pr: changed without a corresponding test change"}]}}

engine/hooks/verdict-flip-watch/tests/fixtures/noise.jsonl

@@ -0,0 +1,5 @@
+{"type": "user", "message": {"role": "user", "content": "look around"}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "ls engine/hooks"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "ok"}]}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "ls engine/hooks/missing"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "ls: cannot access: No such file or directory\nexit=2"}]}}

engine/hooks/verdict-flip-watch/tests/fixtures/stable.jsonl

@@ -0,0 +1,5 @@
+{"type": "user", "message": {"role": "user", "content": "verify the slice"}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 scripts/check_skill_test_coverage.py --base a --head b"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "ok      skill test coverage"}]}}
+{"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {"command": "python3 scripts/check_skill_test_coverage.py --base a --head b"}}]}}
+{"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "content": "ok      skill test coverage"}]}}