fix: keep accepted plans authoritative

Author: EvanProgramming

main.py

@@ -4862,7 +4862,8 @@ def _observe_model_response(text: str) -> dict[str, Any]:
     parsed["define_tool_registered"] = define_tool_registered
     if (parsed["raw"] and not has_control and not parsed["protocol_error"]
             and _active_interaction_mode.get() == "agent"):
-        tasks.from_llm_block(parsed["raw"])
+        if not _interaction_controller.state().get("executing_plan"):
+            tasks.from_llm_block(parsed["raw"])
         tasks.mark_done_from_text(parsed["raw"])
     return parsed
 

tests/test_interaction.py

@@ -133,28 +133,46 @@ def test_plan_revision_acceptance_is_versioned_and_idempotent(self):
 
     def test_provider_plan_text_cannot_duplicate_an_accepted_checklist(self):
         previous_tasks = main.tasks
+        previous_controller = main._interaction_controller
         manager = TaskManager(self.store, workspace_id="workspace", session_id="session")
         proposal = self.controller.propose_plan({
             "title": "Eight steps", "summary": "Execute once.", "assumptions": [],
             "steps": [
                 {"id": f"step-{index}", "title": f"Step {index}",
-                 "description": f"Perform step {index}.", "acceptance": [f"Step {index} done"]}
+                 "description": (
+                     "Create index.html with the requested content."
+                     if index == 0 else f"Perform step {index}."
+                 ), "acceptance": [f"Step {index} done"]}
                 for index in range(8)
             ],
         })
         self.controller.accept_plan(
             manager, plan_id=proposal["plan_id"], version=proposal["version"],
         )
         main.tasks = manager
+        main._interaction_controller = self.controller
+        mode_token = main._active_interaction_mode.set("agent")
         try:
             repeated = "Plan:\n" + "\n".join(
                 f"{index}. Perform step {index}." for index in range(1, 9)
             )
             main._tasks_from_plan(repeated)
             main._tasks_from_plan(repeated)
+            main._observe_model_response(
+                "TaskList:\n```json\n"
+                + json.dumps([f"Use run_cmd to perform step {index}" for index in range(1, 9)])
+                + "\n```"
+            )
             self.assertEqual(len(manager.tasks), 8)
+            evidence = manager.record_evidence(
+                action="write_file", args="index.html|ready", result="Wrote index.html", success=True,
+            )
+            self.assertEqual(evidence["task_id"], manager.tasks[0]["id"])
+            self.assertEqual(manager.tasks[0]["status"], "succeeded")
         finally:
             main.tasks = previous_tasks
+            main._interaction_controller = previous_controller
+            main._active_interaction_mode.reset(mode_token)
 
     def test_exact_acceptance_phrases_and_control_parser(self):
         self.assertTrue(is_plan_acceptance("Execute plan."))