unverified-tag-ledger: gate the next-prompt reminder, default to stale only

engine/hooks/hooks.toml

@@ -225,6 +225,7 @@ summary = "Stops driving the user's keyboard or screen."
 mode = "stop"
 why_mode = "attention"
 summary = "Stops when unchecked claims pile up."
+enabled_by = "CATSTACK_UNVERIFIED_TAG_REMINDER"
 
 [hooks.verdict-flip-watch]
 mode = "warn"

engine/hooks/unverified-tag-ledger/README.md

@@ -40,7 +40,28 @@ fixtures in `tests/test_hooks.py`.
 - **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.
+  behaviour without preventing the turn from ending at all. How much it lists
+  is `CATSTACK_UNVERIFIED_TAG_REMINDER` (see Env).
+
+## Env
+
+| Var | Effect |
+|-----|--------|
+| `CATSTACK_UNVERIFIED_TAG_REMINDER=stale` | Default, and what an unset flag means. Re-inject only claims that have already survived `ESCALATE_AFTER_TURNS` (3) turns. |
+| `CATSTACK_UNVERIFIED_TAG_REMINDER=all` | Re-inject every outstanding claim on every prompt. |
+| `CATSTACK_UNVERIFIED_TAG_REMINDER=off` | No re-injection at all. Rows are still recorded and the Stop refusal still runs. |
+| `CATSTACK_TAG_LEDGER_DIR` | Where the per-session ledger lives (the tests use a tempdir). |
+
+The gate sits on the injection and nowhere else. Gating the tag itself would
+hide the unverified claim rather than stop it, which is the opposite of what
+the ledger is for, and `off` would then also empty the ledger it is mined
+from. Any value other than the three above is named on stderr and falls back
+to `stale`; an `.env` candidate that exists and cannot be read is reported the
+same way rather than passing as "not set".
+
+The hook resolves this flag itself through `engine/hooks/_flags/flags.py`.
+The `enabled_by` line in `engine/hooks/hooks.toml` records which flag the hook
+answers to; nothing reads that field, so it is documentation, not the gate.
 - **Discharge** — a claim is resolved when a later turn runs a verification tool
   (`Bash`, `Read`, `Grep`, `Glob`, `NotebookRead`) and stops re-emitting it.
 - **Where the turn's tool list comes from** — the transcript named by

engine/hooks/unverified-tag-ledger/claude_prompt_reminder.py

@@ -3,13 +3,18 @@
 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.
+
+How much it says is CATSTACK_UNVERIFIED_TAG_REMINDER: off, stale (default), or
+all. The gate is here, on the injection, and nowhere else -- rows keep being
+recorded on every setting, because hiding the tag would hide the unverified
+claim instead of stopping it.
 """
 from __future__ import annotations
 
 import json
 import sys
 
-from detect import reminder
+from detect import reminder, reminder_mode
 
 
 def main() -> None:
@@ -18,8 +23,12 @@ def main() -> None:
     except (json.JSONDecodeError, OSError) as exc:
         sys.stderr.write(f"unverified-tag-ledger: unreadable payload, no reminder: {exc!r}\n")
         return
+    payload = payload if isinstance(payload, dict) else {}
     try:
-        text = reminder(str((payload or {}).get("session_id") or ""))
+        mode, note = reminder_mode(cwd=payload.get("cwd"))
+        if note:
+            sys.stderr.write(note + "\n")
+        text = reminder(str(payload.get("session_id") or ""), mode)
     except Exception as exc:
         sys.stderr.write(f"unverified-tag-ledger: reminder error, continuing: {exc!r}\n")
         return

engine/hooks/unverified-tag-ledger/detect.py

@@ -27,13 +27,20 @@
 
 sys.path.insert(0, os.path.join(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_markers"))
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags"))
 
+import flags  # noqa: E402
 import markers  # noqa: E402
 
 VERIFY_TOOLS = {"Bash", "Read", "Grep", "Glob", "NotebookRead"}
 ESCALATE_AFTER_TURNS = 3
 MAX_LISTED = 5
 
+REMINDER_FLAG = "CATSTACK_UNVERIFIED_TAG_REMINDER"
+REMINDER_MODES = ("off", "stale", "all")
+DEFAULT_REMINDER_MODE = "stale"
+
 DISCHARGE_REFLECT = (
     "unverified-tag-ledger: {count} claim(s) went from unverified to checked this turn: "
     "{claims}. That transition is the whole event: the claim went out first and the check "
@@ -144,9 +151,41 @@ def record_turn(session_id: str, message: str, tools_used: set[str] | None, now=
     return rows
 
 
-def reminder(session_id: str) -> str:
-    """Text for UserPromptSubmit, or empty when nothing is outstanding."""
+def reminder_mode(environ=None, cwd=None, home=None) -> tuple[str, str]:
+    """(mode, note). mode is off, stale, or all; note names what could not be read.
+
+    Three settings, not two, because the complaint is volume and not the
+    ledger. `off` silences the next-prompt reminder and keeps recording rows,
+    so the ledger stays minable either way. `stale` -- the default -- reminds
+    only about claims that have already survived ESCALATE_AFTER_TURNS turns,
+    which is the subset this hook already singles out as a reflect trigger.
+    `all` is the older behaviour, every outstanding claim every prompt.
+
+    Unset means `stale`, deliberately. A flag whose unset value is the old
+    behaviour changes nothing for the person who asked for less.
+    """
+    found = flags.resolve_flag(
+        REMINDER_FLAG, os.environ if environ is None else environ, cwd, home)
+    note = found.unreadable_note(REMINDER_FLAG)
+    raw = (found.value or "").strip().lower()
+    if raw in REMINDER_MODES:
+        return raw, note
+    if raw:
+        extra = (
+            f"unverified-tag-ledger: {REMINDER_FLAG}={found.value!r} is not "
+            f"{', '.join(REMINDER_MODES)}; using {DEFAULT_REMINDER_MODE}.")
+        note = f"{note}\n{extra}" if note else extra
+    return DEFAULT_REMINDER_MODE, note
+
+
+def reminder(session_id: str, mode: str = DEFAULT_REMINDER_MODE) -> str:
+    """Text for UserPromptSubmit, or empty when nothing is due."""
+    if mode == "off":
+        return ""
     open_rows = outstanding(read_ledger(session_id))
+    if mode != "all":
+        open_rows = [row for row in open_rows
+                     if row.get("turns", 0) >= ESCALATE_AFTER_TURNS]
     if not open_rows:
         return ""
     stale = [row for row in open_rows if row.get("turns", 0) >= ESCALATE_AFTER_TURNS]

engine/hooks/unverified-tag-ledger/tests/test_hooks.py

@@ -194,11 +194,63 @@ def test_an_unreadable_turn_is_neither_blocked_nor_called_clean(self) -> None:
 
     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")
+        text = self.detect.reminder("s1", "all")
         self.assertIn("widened scope", text)
         self.assertIn("cat-mode/SKILL.md:269", text)
         self.assertIn("never a place to stop", text)
 
+    def test_reminder_is_silent_on_off(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.assertEqual(self.detect.reminder("s1", "off"), "")
+
+    def test_a_young_claim_is_not_reinjected_by_default(self) -> None:
+        self.detect.record_turn("s1", REAL_TAG_1, set())
+        self.detect.record_turn("s1", REAL_TAG_1, set())
+        rows = self.detect.read_ledger("s1")
+        self.assertEqual(rows[0]["turns"], 1)
+        self.assertEqual(self.detect.reminder("s1"), "")
+
+    def test_a_claim_that_survives_three_turns_is_reinjected_by_default(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())
+        text = self.detect.reminder("s1")
+        self.assertIn("widened scope", text)
+        self.assertIn("reflect trigger", text)
+
+    def test_an_unset_flag_resolves_to_stale_not_to_the_old_behaviour(self) -> None:
+        mode, note = self.detect.reminder_mode(environ={}, cwd=None, home=self.tmp.name)
+        self.assertEqual(mode, "stale")
+        self.assertEqual(note, "")
+
+    def test_each_flag_value_is_honoured(self) -> None:
+        for value in ("off", "stale", "all"):
+            mode, _note = self.detect.reminder_mode(
+                environ={self.detect.REMINDER_FLAG: value}, cwd=None, home=self.tmp.name)
+            self.assertEqual(mode, value)
+
+    def test_a_flag_value_nobody_understands_says_so_and_falls_back(self) -> None:
+        mode, note = self.detect.reminder_mode(
+            environ={self.detect.REMINDER_FLAG: "quiet"}, cwd=None, home=self.tmp.name)
+        self.assertEqual(mode, "stale")
+        self.assertIn("is not off, stale, all", note)
+
+    def test_an_unreadable_env_file_is_reported_as_unchecked(self) -> None:
+        unreadable = os.path.join(self.tmp.name, "env-is-a-directory")
+        os.makedirs(unreadable, exist_ok=True)
+        mode, note = self.detect.reminder_mode(
+            environ={"CATSTACK_ENV_FILE": unreadable}, cwd=None, home=self.tmp.name)
+        self.assertEqual(mode, "stale")
+        self.assertIn("could not read", note)
+
+    def test_recording_keeps_happening_while_the_reminder_is_off(self) -> None:
+        """off is about the injection, never about the ledger."""
+        self.detect.evaluate(self.payload(REAL_TAG_1, tools=True))
+        self.assertEqual(len(self.detect.outstanding(self.detect.read_ledger("s1"))), 1)
+        self.assertEqual(self.detect.reminder("s1", "off"), "")
+
     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())