fix: bind mutations to accepted plan steps

Author: EvanProgramming

main.py

@@ -3879,6 +3879,17 @@ def _execute_turn_action(action: str, args: Any, *, operation_scope: str,
         )
     args = str(args)
     operation_id = _operation_id(operation_scope, canonical, args)
+    if (_is_state_changing_action(canonical, args)
+            and _interaction_controller.state().get("executing_plan")):
+        allowed, reason = tasks.mutation_matches_current_task(canonical, args)
+        if not allowed:
+            result = f"Error: action does not match the accepted plan; {reason}."
+            _notify_tool_execute(canonical, args, result)
+            return _make_execution_receipt(
+                action=canonical, args=args, authorized=False, started_at=started_at,
+                success=False, result=result, operation_scope=operation_scope,
+                failure="plan_action_mismatch",
+            )
     if _is_state_changing_action(canonical, args) and operation_id in successful_operations:
         result = "Error: duplicate successful state-changing action refused for this turn."
         _notify_tool_execute(canonical, args, result)

task_engine.py

@@ -317,6 +317,25 @@ def _match_receipt_task(self, action: str, args: str) -> dict[str, Any] | None:
                 return previous
         return None
 
+    def mutation_matches_current_task(self, action: str, args: str) -> tuple[bool, str]:
+        """Authorize a mutation only when it belongs to the next ordered task."""
+        current = next((
+            task for task in self.tasks
+            if canonical_status(task.get("status", "pending")) in {"pending", "running"}
+        ), None)
+        if current is None:
+            return False, "the accepted plan has no pending task"
+        if self._receipt_match_score(current, action, args) > 0:
+            return True, ""
+        if _ordered_plan_info(current) is None or not _ordered_action_compatible(
+                current.get("description", ""), action):
+            return False, f"the next accepted step is: {current.get('description', '')}"
+        if _receipt_action(action) == "write_file":
+            target = str(args).split("|", 1)[0].strip().replace("\\", "/").rsplit("/", 1)[-1]
+            if not target or target.lower() not in str(current.get("description", "")).lower():
+                return False, f"the next accepted step does not name {target or 'that file'}"
+        return True, ""
+
     def record_evidence(self, *, task_id: str | None = None, action: str, result: str, success: bool,
                         acceptance: str | None = None, args: str | None = None,
                         receipt_id: str | None = None) -> dict[str, Any]:

tests/test_interaction.py

@@ -169,6 +169,56 @@ def test_read_only_modes_deny_mutating_and_dynamic_tools(self):
                 main._execution_capability_token = previous_token
                 main._set_workspace_root(previous_root)
 
+    def test_accepted_plan_rejects_unrelated_mutation_before_execution(self):
+        previous = (
+            main.tasks, main._interaction_controller, main._execution_capability_token,
+            main._get_workspace_root(),
+        )
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = EventStore(root / "state.sqlite3")
+            manager = TaskManager(store, workspace_id="bound", session_id="surface:tui")
+            controller = InteractionController(
+                store, workspace_id="bound", session_id="surface:tui",
+            )
+            proposal = controller.propose_plan({
+                "title": "Create page", "summary": "Create only the accepted page.",
+                "assumptions": [], "steps": [{
+                    "id": "page", "title": "Create index.html",
+                    "description": "Write index.html with the requested markup.",
+                    "acceptance": ["index.html exists"],
+                }],
+            })
+            controller.accept_plan(
+                manager, plan_id=proposal["plan_id"], version=proposal["version"],
+            )
+            main.tasks = manager
+            main._interaction_controller = controller
+            main._set_workspace_root(root)
+            main._execution_capability_token = issue_capability_token(
+                "test:accepted-plan", frozenset({"write"}),
+            )
+            try:
+                operations: set[str] = set()
+                rejected = main._execute_turn_action(
+                    "write_file", "README.md|wrong", operation_scope="bound",
+                    successful_operations=operations,
+                )
+                self.assertFalse(rejected.success)
+                self.assertEqual(rejected.failure, "plan_action_mismatch")
+                self.assertFalse((root / "README.md").exists())
+
+                accepted = main._execute_turn_action(
+                    "write_file", "index.html|ready", operation_scope="bound",
+                    successful_operations=operations,
+                )
+                self.assertTrue(accepted.success, accepted.result)
+                self.assertEqual((root / "index.html").read_text(encoding="utf-8"), "ready")
+            finally:
+                (main.tasks, main._interaction_controller, main._execution_capability_token,
+                 previous_root) = previous
+                main._set_workspace_root(previous_root)
+
     def test_mocked_provider_flow_questions_revision_acceptance_and_agent_receipt(self):
         class LearningStub:
             def feedback_signal(self, _text): return None