A scope correction needs evidence the agent actually did unrequested work

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: <requested outcome and explicit non-goals>`.
 2. The contract releases the first tool gate but stays in session state.
    Apologies and unmarked restatements never clear it.

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

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,18 +132,32 @@ 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",
                 "message": {"role": "user", "content": text},
             }) + "\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,60 +172,66 @@ 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",
         )
 
     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",
         )
 
     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.</result>\n"
             "</task-notification>"
         )
-        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)