diff --git a/engine/hooks/scope-lock/README.md b/engine/hooks/scope-lock/README.md index b4cc0fc5..5f6fb6f2 100644 --- a/engine/hooks/scope-lock/README.md +++ b/engine/hooks/scope-lock/README.md @@ -16,6 +16,14 @@ the product: | Proposal | `should we do this the same way we did in invoker?` | | Substitution | `surprised we elected to use subagents instead of invoker` | +A recognised shape records a correction only when the transcript also shows a +non-read-only tool call in the agent turn opened by the user's previous +message. Read-only lookups do not corroborate a correction. Missing, +unreadable, or unparseable transcript evidence is a distinct unknown result, +not zero work; an unknown does not record a correction. This avoids imposing a +lock when the check could not run, at the accepted cost that drift which exists +only in a stated plan, before any mutating call, is no longer caught. + The last three shapes are the ones a user reaches for first, before they get blunt, so leaving them out costs the whole early warning. Each one needs three things in the same sentence before it counts: `you` or `we` as the actor doing @@ -28,9 +36,9 @@ session. The state machine is per harness session: -1. First same-class correction records a persistent lock. Local read-only - tools remain available, but mutating, shell, delegated, and external tools - are blocked until the transcript contains one standalone line: +1. First corroborated same-class correction records a persistent lock. Local + read-only tools remain available, but mutating, shell, delegated, and + external tools are blocked until the transcript contains one standalone line: `SCOPE CONTRACT: `. 2. The contract releases the first tool gate but stays in session state. Apologies and unmarked restatements never clear it. diff --git a/engine/hooks/scope-lock/detect.py b/engine/hooks/scope-lock/detect.py index 92311128..c49aff32 100644 --- a/engine/hooks/scope-lock/detect.py +++ b/engine/hooks/scope-lock/detect.py @@ -149,9 +149,17 @@ def extract_prompt_text(payload: dict[str, Any]) -> str: return "" -def correction_class(text: str) -> str | None: - """Return the stable correction class, excluding explicit expansions.""" - if not text or AUTOMATED_NOTIFICATION_RE.match(text): +def _normalized_tool_name(name: Any) -> str: + return re.sub(r"[^a-z_]", "", str(name or "").lower()) + + +def _is_local_read_only_tool(name: Any) -> bool: + return _normalized_tool_name(name) in LOCAL_READ_ONLY_TOOLS + + +def correction_class(text: str, mutating_work: bool | None) -> str | None: + """Return the stable correction class when transcript evidence corroborates it.""" + if mutating_work is not True or not text or AUTOMATED_NOTIFICATION_RE.match(text): return None window = text[-CORRECTION_SCAN_TAIL_CHARS:] if EXPANSION_RE.search(window): @@ -240,6 +248,90 @@ def _transcript_path(payload: dict[str, Any]) -> str: return value if isinstance(value, str) else "" +def _transcript_payload(data: dict[str, Any]) -> dict[str, Any]: + nested = data.get("payload") + if data.get("type") == "response_item" and isinstance(nested, dict): + return nested + return data + + +def _message_parts(data: dict[str, Any]) -> tuple[str, Any]: + entry = _transcript_payload(data) + message = entry.get("message") + if isinstance(message, dict): + return str(message.get("role") or ""), message.get("content") + return str(entry.get("role") or ""), entry.get("content") + + +def _user_text(data: dict[str, Any]) -> str | None: + entry = _transcript_payload(data) + role, content = _message_parts(data) + if entry.get("type") != "user" and role != "user": + return None + if isinstance(content, str): + return content if content.strip() else None + if not isinstance(content, list): + return None + text = "\n".join( + str(block.get("text") or "") + for block in content + if isinstance(block, dict) and block.get("type") in {"text", "input_text"} + ) + return text if text.strip() else None + + +def _tool_names(data: dict[str, Any]) -> list[str]: + entry = _transcript_payload(data) + if entry.get("type") in {"custom_tool_call", "function_call"}: + return [str(entry.get("name") or "")] + + role, content = _message_parts(data) + if entry.get("type") != "assistant" and role != "assistant": + return [] + if not isinstance(content, list): + return [] + return [ + str(block.get("name") or "") + for block in content + if isinstance(block, dict) + and block.get("type") in {"tool_use", "toolCall", "custom_tool_call", "function_call"} + ] + + +def mutating_work_after_previous_user( + payload: dict[str, Any], current_prompt: str +) -> bool | None: + """Return mutating-work evidence for the prior user turn, or None if unavailable.""" + path = _transcript_path(payload) + if not path: + return None + + turns: list[tuple[str, list[str]]] = [] + try: + with open(path, encoding="utf-8") as handle: + for line in handle: + try: + data = json.loads(line) + except json.JSONDecodeError: + return None + if not isinstance(data, dict): + return None + text = _user_text(data) + if text is not None: + turns.append((text, [])) + continue + if turns: + turns[-1][1].extend(_tool_names(data)) + except (OSError, UnicodeError): + return None + + if not turns or turns[-1][0].strip() != current_prompt.strip(): + return None + if len(turns) < 2: + return False + return any(not _is_local_read_only_tool(name) for name in turns[-2][1]) + + def _line_count(path: str) -> int: try: with open(path, encoding="utf-8") as handle: @@ -323,7 +415,8 @@ def process_prompt(payload: dict[str, Any]) -> dict[str, Any]: save_state(payload, state) return state - correction = correction_class(prompt) + mutating_work = mutating_work_after_previous_user(payload, prompt) + correction = correction_class(prompt, mutating_work) if not correction: return state @@ -387,7 +480,6 @@ def tool_block_reason(payload: dict[str, Any]) -> tuple[bool, str]: save_state(payload, state) return False, "" - tool = re.sub(r"[^a-z_]", "", _tool_name(payload).lower()) - if tool in LOCAL_READ_ONLY_TOOLS: + if _is_local_read_only_tool(_tool_name(payload)): return False, "" return True, FIRST_GATE diff --git a/engine/hooks/scope-lock/tests/test_hooks.py b/engine/hooks/scope-lock/tests/test_hooks.py index 2d4fcfdb..7a37b4ad 100644 --- a/engine/hooks/scope-lock/tests/test_hooks.py +++ b/engine/hooks/scope-lock/tests/test_hooks.py @@ -102,6 +102,28 @@ def test_missing_transcript_counts_zero_lines_rather_than_guessing(self): payload = {"session_id": "session-3", "transcript_path": "/nonexistent/session.jsonl"} self.assertEqual(detect._line_count(detect._transcript_path(payload)), 0) + def test_unreadable_transcript_evidence_records_no_correction(self): + transcript = os.path.join(self.tmp.name, "unreadable.jsonl") + payload = { + "session_id": "session-unreadable-transcript", + "transcript_path": transcript, + "prompt": "what are you doing", + } + real_open = open + + def open_except_transcript(path, *args, **kwargs): + if path == transcript: + raise PermissionError("transcript is unreadable") + return real_open(path, *args, **kwargs) + + with patch("builtins.open", side_effect=open_except_transcript): + evidence = detect.mutating_work_after_previous_user(payload, payload["prompt"]) + result = detect.process_prompt(payload) + + self.assertIsNone(evidence) + self.assertNotIn("phase", result) + self.assertNotIn("correction_counts", result) + class ScopeLockCase(unittest.TestCase): def setUp(self) -> None: @@ -110,11 +132,15 @@ def setUp(self) -> None: self.transcript = os.path.join(self.tmp.name, "session.jsonl") open(self.transcript, "w", encoding="utf-8").close() self.base = {"session_id": "session-1", "transcript_path": self.transcript} + self.append_user("Please complete the requested work.") + self.append_tool("Write") def tearDown(self) -> None: self.tmp.cleanup() - def prompt(self, text: str) -> dict: + def prompt(self, text: str, *, mutating_work: bool = True) -> dict: + if mutating_work: + self.append_tool("Write") with open(self.transcript, "a", encoding="utf-8") as handle: handle.write(json.dumps({ "type": "user", @@ -122,6 +148,16 @@ def prompt(self, text: str) -> dict: }) + "\n") return detect.process_prompt({**self.base, "prompt": text}) + def append_tool(self, name: str) -> None: + with open(self.transcript, "a", encoding="utf-8") as handle: + handle.write(json.dumps({ + "type": "assistant", + "message": { + "role": "assistant", + "content": [{"type": "tool_use", "name": name, "input": {}}], + }, + }) + "\n") + def append_user(self, text: str) -> None: with open(self.transcript, "a", encoding="utf-8") as handle: handle.write(json.dumps({ @@ -136,44 +172,46 @@ def tool(self, name: str = "Bash") -> tuple[bool, str]: class TestDetection(ScopeLockCase): def test_repeated_drift_fixture_detects_same_class(self): messages = fixture_messages("repeated_drift.jsonl") - self.assertEqual([detect.correction_class(m) for m in messages], ["scope", "scope"]) + self.assertEqual([detect.correction_class(m, True) for m in messages], ["scope", "scope"]) def test_ordinary_product_confusion_does_not_trigger(self): [message] = fixture_messages("ordinary_product_confusion.jsonl") - self.assertIsNone(detect.correction_class(message)) + self.assertIsNone(detect.correction_class(message, True)) def test_explicit_scope_expansion_does_not_trigger(self): [message] = fixture_messages("explicit_scope_expansion.jsonl") - self.assertIsNone(detect.correction_class(message)) + self.assertIsNone(detect.correction_class(message, True)) def test_execution_routing_fixture_detects_every_real_correction(self): messages = fixture_messages("execution_routing_correction.jsonl") self.assertEqual(len(messages), 4) self.assertEqual( - [detect.correction_class(m) for m in messages], + [detect.correction_class(m, True) for m in messages], ["scope", "scope", "scope", "scope"], ) def test_interrogative_correction_triggers(self): self.assertEqual( - detect.correction_class("wait why are you running this locally and not in invoker?"), + detect.correction_class("wait why are you running this locally and not in invoker?", True), "scope", ) self.assertEqual( - detect.correction_class("why did we do this with subagents rather than the queue?"), + detect.correction_class("why did we do this with subagents rather than the queue?", True), "scope", ) def test_proposal_shaped_correction_triggers(self): self.assertEqual( detect.correction_class( - "if we are backtesting this, should we doing this the same way we did in invoker?" + "if we are backtesting this, should we doing this the same way we did in invoker?", + True, ), "scope", ) self.assertEqual( detect.correction_class( - "we should parallelize these with invoker instead. we shouldn't do this locally." + "we should parallelize these with invoker instead. we shouldn't do this locally.", + True, ), "scope", ) @@ -181,7 +219,8 @@ def test_proposal_shaped_correction_triggers(self): def test_substitution_correction_triggers(self): self.assertEqual( detect.correction_class( - "also im a bit surprised we elected to use subagents instead of invoker execution. why?" + "also im a bit surprised we elected to use subagents instead of invoker execution. why?", + True, ), "scope", ) @@ -189,7 +228,10 @@ def test_substitution_correction_triggers(self): def test_genuine_question_fixture_does_not_trigger(self): messages = fixture_messages("genuine_question.jsonl") self.assertEqual(len(messages), 4) - self.assertEqual([detect.correction_class(m) for m in messages], [None, None, None, None]) + self.assertEqual( + [detect.correction_class(m, True) for m in messages], + [None, None, None, None], + ) def test_question_about_an_artifact_rather_than_the_agent_does_not_trigger(self): for message in ( @@ -198,7 +240,7 @@ def test_question_about_an_artifact_rather_than_the_agent_does_not_trigger(self) "can you explain why the contract has to land in a prior turn?", "does invoker support this, or do we need to run it locally first?", ): - self.assertIsNone(detect.correction_class(message), message) + self.assertIsNone(detect.correction_class(message, True), message) def test_pasted_transcript_trigger_phrase_far_from_end_does_not_trigger(self): # Mirrors a real session: a pasted terminal transcript quoting a @@ -211,11 +253,11 @@ def test_pasted_transcript_trigger_phrase_far_from_end_does_not_trigger(self): + ' the model wrote "do not use Invoker" in its own gate text ' + filler ) - self.assertIsNone(detect.correction_class(text)) + self.assertIsNone(detect.correction_class(text, True)) def test_trigger_phrase_within_tail_window_still_triggers(self): text = "x" * 200 + " ok whatever, just do it locally" - self.assertEqual(detect.correction_class(text), "scope") + self.assertEqual(detect.correction_class(text, True), "scope") def test_automated_task_notification_never_triggers_correction(self): text = ( @@ -224,7 +266,7 @@ def test_automated_task_notification_never_triggers_correction(self): "\"just do it locally\" in the quoted transcript.\n" "" ) - self.assertIsNone(detect.correction_class(text)) + self.assertIsNone(detect.correction_class(text, True)) def test_automated_task_notification_never_satisfies_reflection_check(self): text = ( @@ -242,6 +284,33 @@ def test_bare_reflect_without_leading_slash_still_satisfies_check(self): class TestStateMachine(ScopeLockCase): + def test_correction_wording_with_mutating_work_records_correction(self): + self.append_user("Please delegate the requested implementation.") + self.append_tool("Agent") + + result = self.prompt("what are you doing", mutating_work=False) + + self.assertEqual(result["phase"], "contract_required") + self.assertEqual(result["correction_counts"], {"scope": 1}) + + def test_correction_wording_without_mutating_work_records_nothing(self): + self.append_user("Please inspect the current status.") + self.append_tool("Read") + + result = self.prompt("what are you doing", mutating_work=False) + + self.assertNotIn("phase", result) + self.assertNotIn("correction_counts", result) + + def test_explicit_expansion_with_mutating_work_stays_excluded(self): + result = self.prompt( + "what are you doing? Also include the deployment scripts.", + mutating_work=True, + ) + + self.assertNotIn("phase", result) + self.assertNotIn("correction_counts", result) + def test_real_conversation_contract_then_ok_do_it_allows_write(self): rows = fixture_rows("contract_continuation.jsonl") request = rows[0]["message"]["content"] @@ -485,7 +554,9 @@ def test_first_gate_instructs_ending_the_turn(self): class TestHarnessWrappers(ScopeLockCase): def test_claude_prompt_injects_scope_contract_instruction(self): - payload = {**self.base, "prompt": "wtf are you doing? Just fix it locally."} + prompt = "wtf are you doing? Just fix it locally." + self.append_user(prompt) + payload = {**self.base, "prompt": prompt} code, out, _ = run_main(claude_prompt_scope.main, payload) self.assertEqual(code, 0) body = json.loads(out) @@ -498,7 +569,9 @@ def test_claude_pretool_blocks_with_exit_two(self): self.assertIn("SCOPE CONTRACT:", err) def test_cursor_before_submit_records_lock(self): - payload = {**self.base, "prompt": "wtf are you doing? Just fix it locally."} + prompt = "wtf are you doing? Just fix it locally." + self.append_user(prompt) + payload = {**self.base, "prompt": prompt} code, out, _ = run_main(cursor_before_submit.main, payload) self.assertEqual(code, 0) self.assertEqual(json.loads(out), {"continue": True}) @@ -513,7 +586,9 @@ def test_cursor_pretool_blocks(self): self.assertIn("SCOPE CONTRACT:", body["user_message"]) def test_codex_prompt_injects_scope_contract_instruction(self): - payload = {**self.base, "prompt": "wtf are you doing? Just fix it locally."} + prompt = "wtf are you doing? Just fix it locally." + self.append_user(prompt) + payload = {**self.base, "prompt": prompt} code, out, _ = run_main(codex_prompt_scope.main, payload) self.assertEqual(code, 0) body = json.loads(out)