diff --git a/engine/hooks/agent-routing-guard/detect.py b/engine/hooks/agent-routing-guard/detect.py index c8fc9317..1987fb3e 100644 --- a/engine/hooks/agent-routing-guard/detect.py +++ b/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 ) diff --git a/engine/hooks/agent-routing-guard/tests/test_hooks.py b/engine/hooks/agent-routing-guard/tests/test_hooks.py index f53b313f..e3504a7e 100644 --- a/engine/hooks/agent-routing-guard/tests/test_hooks.py +++ b/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: