Block a CAT-UNVERIFIED tag that hands a read-only check back as 'say X'

engine/hooks/frustration-watchdog/claude_stop_check.py

@@ -86,6 +86,11 @@
     r"i(?:'ll| will) (?:handle|do|take|run|fix|keep|watch))\b"
 )
 
+CHECK_IN_FLIGHT_RE = re.compile(
+    r"(?i)\bi(?:'m| am) (?:running|checking|testing|probing|reading|querying|fetching) "
+    r"[^.\n]{1,100}?\bnow\b"
+)
+
 HOOK_REFUSAL_RE = re.compile(r"\s*PreToolUse:\S+ hook error\b")
 """How Claude Code records a tool call a PreToolUse hook refused: an is_error
 tool_result whose text opens "PreToolUse:Bash hook error: [<command>]: ..."."""
@@ -209,7 +214,9 @@ def ends_the_wait(message):
     stripped = message.rstrip()
     if stripped.endswith("?"):
         return True  # a direct question is a handoff
-    return bool(NEXT_STEP_RE.search(message) or ETA_RE.search(message))
+    return bool(
+        NEXT_STEP_RE.search(message) or ETA_RE.search(message) or CHECK_IN_FLIGHT_RE.search(message)
+    )
 
 
 def detect(data):

engine/hooks/frustration-watchdog/tests/test_hooks.py

@@ -72,6 +72,23 @@ def test_waiting_user_with_concrete_action_passes(self):
         finally:
             os.unlink(path)
 
+    def test_check_in_flight_counts_as_next_step(self):
+        path = transcript_with([("2026-08-18T02:13:30Z", "i am waiting for you to do something")])
+        try:
+            blocked, _ = run_hook(path, "I am running the GraphQL probe now. Result in this turn.")
+            self.assertFalse(blocked)
+        finally:
+            os.unlink(path)
+
+    def test_narrated_past_work_is_not_a_check_in_flight(self):
+        path = transcript_with([("2026-08-18T02:13:30Z", "i am waiting for you to do something")])
+        try:
+            blocked, err = run_hook(path, "I ran the GraphQL probe earlier and it looked fine.")
+            self.assertTrue(blocked)
+            self.assertIn("impatience-shaped", err)
+        finally:
+            os.unlink(path)
+
     def test_direct_question_counts_as_handoff(self):
         path = transcript_with([("2026-08-18T02:13:30Z", "WHY ARE WE NOT LAUNCHING A ZOOM MEETING RIGHT NOW")])
         try:

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

@@ -66,7 +66,20 @@
     re.IGNORECASE | re.DOTALL,
 )
 
+HANDBACK_RE = re.compile(
+    r"\b(?:say|reply|type|tell me to)\s+(?:the word\s+)?(?:[\"'\u2018\u201c][^\"'\u2019\u201d]{1,60}[\"'\u2019\u201d]"
+    r"|(?:go|test|test it|run it|check it|yes|ok|okay|probe it)\b)",
+    re.IGNORECASE,
+)
+WRITE_ACTION_RE = re.compile(
+    r"\b(?:push|deploy|delete|merge|commit|post|send|apply|write|create|publish|"
+    r"overwrite|force|restart|kill|install|submit|approve|release)\w*\b",
+    re.IGNORECASE,
+)
+SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+|\n+")
+
 RULE_NO_VERIFICATION_TOOL = "unverified-tag-ledger.no-verification-tool"
+RULE_HANDBACK = "unverified-tag-ledger.handback"
 RULE_DO_NOT_EMIT = "unverified-tag-ledger.do-not-emit"
 RULE_PROMPT_REMINDER = "unverified-tag-ledger.prompt-reminder"
 
@@ -242,6 +255,22 @@ def emitted_tags(message: str) -> list[str]:
     return [match.group(0) for match in markers.TAG_RE.finditer(prose)]
 
 
+def handback_offers(message: str) -> list[str]:
+    """Sentences outside code that hand a read-only check back to the user.
+
+    "say 'test it'" and "tell me to test it" make the user the trigger for a
+    check the agent can run itself. An offer whose sentence names a write
+    (push, deploy, delete, ...) is a real approval gate and is left alone.
+    """
+    prose = INLINE_CODE_RE.sub("", FENCED_RE.sub("", message or ""))
+    offers = []
+    for sentence in SENTENCE_SPLIT_RE.split(prose):
+        sentence = " ".join(sentence.split())
+        if HANDBACK_RE.search(sentence) and not WRITE_ACTION_RE.search(sentence):
+            offers.append(sentence)
+    return offers
+
+
 def evaluate(payload: dict, mode: str | None = None) -> dict:
     """Record the turn, then decide whether it may end.
 
@@ -290,6 +319,14 @@ def evaluate(payload: dict, mode: str | None = None) -> dict:
     if not new_claims:
         return {"note": "\n".join(notes), "block": ""}
 
+    offers = handback_offers(message)
+    if offers:
+        return {"note": "", "block": (
+            "unverified-tag-ledger: this reply tags a claim as unverified and hands the check back "
+            f"to the user: {'; '.join(offers[:MAX_LISTED])}. The check is read-only, so the user "
+            "does not need to trigger it. Run the check now (Bash/Read/Grep/WebFetch) and paste its "
+            "output. This block is not released by stop_hook_active: deleting the offer always ends the loop.")}
+
     if tools is None:
         notes.append(
             f"unverified-tag-ledger: logged {len(new_claims)} CAT-UNVERIFIED claim(s), but this "
@@ -339,6 +376,16 @@ def _stop_findings(event: dict[str, object]) -> list[Finding]:
     if not block:
         return []
     message = _last_assistant_text(payload)
+    if "hands the check back" in block:
+        return [
+            Finding(
+                rule_id=RULE_HANDBACK,
+                subject=offer,
+                message=block,
+                evidence=offer,
+            )
+            for offer in handback_offers(message)
+        ]
     if behavior == "do_not_emit":
         tags = emitted_tags(message)
         return [

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

@@ -179,6 +179,34 @@ def test_rewrite_turn_is_released_so_the_block_cannot_loop(self) -> None:
             self.payload(REAL_TAG_1, tools=False, stop_hook_active=True))
         self.assertEqual(verdict["block"], "")
 
+    def test_tag_plus_say_offer_is_blocked_even_after_a_read(self) -> None:
+        reply = (
+            f"{REAL_TAG_1}\n\nIf you want me to settle it, say 'test it' and I will run the probe.")
+        verdict = self.detect.evaluate(self.payload(reply, tools=True))
+        self.assertIn("hands the check back", verdict["block"])
+        self.assertIn("run the check now", verdict["block"].lower())
+
+    def test_tag_plus_tell_me_to_test_offer_is_blocked(self) -> None:
+        reply = f"{REAL_TAG_1}\n\nTell me to test it and I will."
+        verdict = self.detect.evaluate(self.payload(reply, tools=True))
+        self.assertIn("hands the check back", verdict["block"])
+
+    def test_say_offer_block_is_not_released_by_stop_hook_active(self) -> None:
+        reply = f"{REAL_TAG_1}\n\nSay 'go' and I will run it."
+        verdict = self.detect.evaluate(
+            self.payload(reply, tools=True, stop_hook_active=True))
+        self.assertIn("hands the check back", verdict["block"])
+
+    def test_tag_plus_say_offer_for_a_write_is_allowed(self) -> None:
+        reply = f"{REAL_TAG_1}\n\nSay 'push it' and I will push the branch to origin."
+        verdict = self.detect.evaluate(self.payload(reply, tools=True))
+        self.assertEqual(verdict["block"], "")
+
+    def test_say_offer_without_a_tag_is_allowed(self) -> None:
+        verdict = self.detect.evaluate(
+            self.payload("Ran the probe: 3 passed. Say 'next' for the follow-up.", tools=True))
+        self.assertEqual(verdict["block"], "")
+
     def test_untagged_turn_with_no_tools_is_never_blocked(self) -> None:
         verdict = self.detect.evaluate(
             self.payload("Short answer, nothing claimed.", tools=False))