Routing guard refusal quotes the words it matched

engine/hooks/agent-routing-guard/detect.py

@@ -48,6 +48,7 @@
 import os
 import re
 import shutil
+from typing import NamedTuple
 
 AGENT_TOOL_NAMES = frozenset({"Agent", "Task"})
 INVOKER_CLI = "invoker-cli"
@@ -102,9 +103,16 @@
     re.compile(r"(?i)\b(?:without|skip|bypass|no)\s+invoker\b"),
 )
 
+
+class PublicationVerbHit(NamedTuple):
+    label: str
+    matched_text: str
+    window: str
+
+
 BLOCK_MESSAGE = (
     "agent-routing-guard: this Agent spawn carries publication work ({verbs}) and "
-    "invoker-cli is on PATH, so the subagent is the wrong vehicle. cat-mode "
+    "invoker-cli is on PATH, so the subagent is the wrong vehicle.\n{hits}\ncat-mode "
     "execution routing rule 3 -- an approved plan or durable/parallel work goes to "
     "Invoker when Invoker is available, not into parallel subagents each landing "
     "its own commit -- decides this one. Follow the installed {skill} skill to "
@@ -116,7 +124,7 @@
 
 UNCHECKED_MESSAGE = (
     "agent-routing-guard: this Agent spawn carries publication work ({verbs}) and "
-    "invoker-cli is on PATH, and the local override could not be checked ({reason}). "
+    "invoker-cli is on PATH, and the local override could not be checked ({reason}).\n{hits}\n"
     "An override that cannot be read is not an override, so cat-mode execution "
     "routing rule 3 stands: an approved plan or durable/parallel work goes to "
     "Invoker, not into parallel publishing subagents. Follow the installed {skill} "
@@ -195,23 +203,28 @@ def _counts_as_action(text: str, start: int, end: int) -> bool:
     return not (_is_hyphen_joined(text, start, end) or _is_negated(text, start))
 
 
-def publication_verbs(prompt: str) -> list[str]:
+def _publication_hit(label: str, text: str, match: re.Match[str]) -> PublicationVerbHit:
+    window_start = max(0, match.start() - 40)
+    window_end = min(len(text), match.end() + 40)
+    return PublicationVerbHit(label, match.group(0), text[window_start:window_end])
+
+
+def publication_verbs(prompt: str) -> list[PublicationVerbHit]:
     """The publication verbs this prompt uses as actions, deduped and ordered."""
     text = prompt or ""
-    found: list[str] = []
+    found: list[PublicationVerbHit] = []
     for label, pattern in ACTION_VERB_RES:
         for match in pattern.finditer(text):
             if _is_noun_use(text, match.start()):
                 continue
             if not _counts_as_action(text, match.start(), match.end()):
                 continue
-            found.append(label)
+            found.append(_publication_hit(label, text, match))
+            break
+    for match in PR_ACTION_RE.finditer(text):
+        if _counts_as_action(text, match.start(), match.end()):
+            found.append(_publication_hit("open a PR", text, match))
             break
-    if any(
-        _counts_as_action(text, match.start(), match.end())
-        for match in PR_ACTION_RE.finditer(text)
-    ):
-        found.append("open a PR")
     return found
 
 
@@ -288,13 +301,21 @@ def override_state(path: str) -> tuple[str, str]:
     return OVERRIDE_ABSENT, ""
 
 
-def block_message(verbs: list[str]) -> str:
-    return BLOCK_MESSAGE.format(verbs=", ".join(verbs), skill=ROUTING_SKILL)
+def _verb_labels(verbs: list[PublicationVerbHit]) -> str:
+    return ", ".join(verb.label for verb in verbs)
+
+
+def _hit_lines(verbs: list[PublicationVerbHit]) -> str:
+    return "\n".join(f'{verb.label}: "{verb.window}"' for verb in verbs)
+
+
+def block_message(verbs: list[PublicationVerbHit]) -> str:
+    return BLOCK_MESSAGE.format(verbs=_verb_labels(verbs), hits=_hit_lines(verbs), skill=ROUTING_SKILL)
 
 
-def unchecked_message(verbs: list[str], reason: str) -> str:
+def unchecked_message(verbs: list[PublicationVerbHit], reason: str) -> str:
     return UNCHECKED_MESSAGE.format(
-        verbs=", ".join(verbs), reason=reason, skill=ROUTING_SKILL
+        verbs=_verb_labels(verbs), hits=_hit_lines(verbs), reason=reason, skill=ROUTING_SKILL
     )
 
 

engine/hooks/agent-routing-guard/tests/test_hooks.py

@@ -43,6 +43,10 @@ def fixture_names() -> list[str]:
     return sorted(n for n in os.listdir(FIXTURE_DIR) if n.endswith(".json"))
 
 
+def publication_labels(prompt: str) -> list[str]:
+    return [hit.label for hit in detect.publication_verbs(prompt)]
+
+
 class Sandbox:
     """A PATH with or without invoker-cli, plus a transcript to read the
     user's current message from."""
@@ -162,7 +166,7 @@ def test_action_verbs_are_detected(self) -> None:
             "land the PRs bottom to top": ["open a PR"],
         }
         for prompt, expected in cases.items():
-            self.assertEqual(detect.publication_verbs(prompt), expected, prompt)
+            self.assertEqual(publication_labels(prompt), expected, prompt)
 
     def test_the_same_words_as_nouns_do_not_count(self) -> None:
         for prompt in (
@@ -173,7 +177,7 @@ def test_the_same_words_as_nouns_do_not_count(self) -> None:
             "which PR introduced this?",
             "report the first push that failed",
         ):
-            self.assertEqual(detect.publication_verbs(prompt), [], prompt)
+            self.assertEqual(publication_labels(prompt), [], prompt)
 
     def test_negated_verbs_do_not_count(self) -> None:
         for prompt in (
@@ -184,7 +188,7 @@ def test_negated_verbs_do_not_count(self) -> None:
             "Neither edit nor merge anything.",
             "Finish without committing, pushing, or merging.",
         ):
-            self.assertEqual(detect.publication_verbs(prompt), [], prompt)
+            self.assertEqual(publication_labels(prompt), [], prompt)
 
     def test_hyphenated_names_do_not_count(self) -> None:
         for prompt in (
@@ -193,7 +197,7 @@ def test_hyphenated_names_do_not_count(self) -> None:
             "Explain how merge-clone works in the repo.",
             "Check whether auto-merge is enabled on the repo settings.",
         ):
-            self.assertEqual(detect.publication_verbs(prompt), [], prompt)
+            self.assertEqual(publication_labels(prompt), [], prompt)
 
     def test_real_publishing_requests_around_negation_still_fire(self) -> None:
         cases = {
@@ -208,7 +212,18 @@ def test_real_publishing_requests_around_negation_still_fire(self) -> None:
             "re-push the branch": ["push"],
         }
         for prompt, expected in cases.items():
-            self.assertEqual(detect.publication_verbs(prompt), expected, prompt)
+            self.assertEqual(publication_labels(prompt), expected, prompt)
+
+    def test_narrated_past_tense_force_push_sentence_keeps_current_verdict(self) -> None:
+        prompt = "Force-push attempts were blocked twice by the stacking tool's hook"
+        self.assertEqual(publication_labels(prompt), ["push"])
+
+    def test_non_git_merge_phrase_keeps_current_verdict(self) -> None:
+        self.assertEqual(publication_labels("merge overlapping findings"), ["merge"])
+
+    def test_clean_read_only_sentence_stays_silent(self) -> None:
+        prompt = "Read the hook and summarize the control flow."
+        self.assertEqual(publication_labels(prompt), [])
 
 
 class OverrideCase(unittest.TestCase):
@@ -325,15 +340,30 @@ def test_read_only_scanner_prompts_are_allowed_through_the_entrypoint(self) -> N
             self.assertEqual(message, "", prompt)
 
     def test_a_publishing_prompt_still_blocks_through_the_entrypoint(self) -> None:
+        prompt = "commit and push the fix, then open a PR"
         payload = {
             "tool_name": "Agent",
             "transcript_path": self.box.transcript("go"),
-            "tool_input": {"prompt": "commit and push the fix, then open a PR"},
+            "tool_input": {"prompt": prompt},
         }
         blocked, message = run_entrypoint(payload, self.box.environ())
         self.assertTrue(blocked)
         self.assertIn("commit, push, open a PR", message)
 
+    def test_each_matched_span_appears_verbatim_in_the_refusal(self) -> None:
+        prompt = "Please commit the guard change, then push it and open a PR."
+        payload = {
+            "tool_name": "Agent",
+            "transcript_path": self.box.transcript("go"),
+            "tool_input": {"prompt": prompt},
+        }
+        hits = detect.publication_verbs(prompt)
+        blocked, message = run_entrypoint(payload, self.box.environ())
+        self.assertTrue(blocked, message)
+        for hit in hits:
+            self.assertIn(hit.matched_text, message)
+            self.assertIn(f'{hit.label}: "{hit.window}"', message)
+
     def test_invoker_absent_allows_the_spawn_without_reading_the_transcript(self) -> None:
         box = Sandbox(invoker_on_path=False)
         try: