fix: snapshot queued plan acceptance identity

openkyrozen/interfaces/tui/commands.py

@@ -276,10 +276,7 @@ def _command(self, name: str, args: Any, request_id: str) -> None:
             self.emit("response", request_id, text="Interaction mode set to plan.")
             self.interaction(request_id)
         elif action == "accept":
-            plan = runtime._interaction_controller.state().get("pending_plan")
-            acceptance = ({"plan_id": plan["plan_id"], "version": plan["version"]}
-                          if plan else None)
-            self.submit("accept plan", request_id, plan_acceptance=acceptance)
+            self.submit("accept plan", request_id)
         elif action == "cancel":
             try:
                 runtime.cancel_interaction_plan()

openkyrozen/interfaces/tui/turns.py

@@ -166,6 +166,11 @@ def callback(event: dict[str, Any] | RuntimeEvent) -> None:
 
 
 def submit(self, text: str, request_id: str, *, plan_acceptance: dict[str, Any] | None = None) -> None:
+    if plan_acceptance is None and self.agent.is_plan_acceptance(text):
+        state = self.agent._interaction_controller.state()
+        plan = state.get("pending_plan") or state.get("executing_plan")
+        plan_acceptance = ({"plan_id": plan["plan_id"], "version": plan["version"]}
+                           if plan else {"plan_id": "", "version": 0})
     with self._state_lock:
         if self._busy:
             self.emit("error", request_id, code="busy", error="A turn is already running.")

openkyrozen/interfaces/web/chat.py

@@ -287,6 +287,19 @@ def _apply_chat_controls(self, session: dict[str, Any], body: dict[str, Any], me
     return message, None
 
 
+def _capture_plan_acceptance(self, session: dict[str, Any], body: dict[str, Any],
+                             message: str) -> dict[str, Any] | None:
+    plan_action = body.get("plan_action")
+    if isinstance(plan_action, dict) and str(plan_action.get("action") or "").lower() == "accept":
+        return {"plan_id": plan_action.get("plan_id"), "version": plan_action.get("version")}
+    if self._agent.is_plan_acceptance(message):
+        state = self._interaction_for_session(session).state()
+        plan = state.get("pending_plan") or state.get("executing_plan")
+        return ({"plan_id": plan["plan_id"], "version": plan["version"]}
+                if plan else {"plan_id": "", "version": 0})
+    return None
+
+
 def _run_session_chat(self, session: dict[str, Any], message: str, *, on_runtime_event=None,
                       plan_acceptance: dict[str, Any] | None = None) -> str:
     """Run a session without swapping process-global agent state."""
@@ -400,13 +413,10 @@ async def api_chat(self, request: Request):
     session["profile"] = service._normalise_profile(body.get("profile", session.get("profile", "auto")))
     service._set_memory_context(session, body)
     msg = service._validate_message(service._sanitize_api_message(str(body.get("message", "")).strip()))
+    plan_acceptance = service._capture_plan_acceptance(session, body, msg)
     msg, immediate_reply = (await asyncio.to_thread(service._apply_chat_controls, session, body, msg)
                             if "fast_backend" in body or "system_one_backend" in body or "decision_assist_backend" in body
                             else service._apply_chat_controls(session, body, msg))
-    plan_action = body.get("plan_action")
-    plan_acceptance = ({"plan_id": plan_action["plan_id"], "version": plan_action["version"]}
-                       if isinstance(plan_action, dict)
-                       and str(plan_action.get("action") or "").lower() == "accept" else None)
     if not msg and immediate_reply is None:
         raise HTTPException(400, "Empty message")
     service._audit("CHAT", f"user={session['user_id']} msg={msg[:80]}", session["user_id"])
@@ -451,13 +461,10 @@ async def api_chat_stream(self, request: Request):
     session["profile"] = service._normalise_profile(body.get("profile", session.get("profile", "auto")))
     service._set_memory_context(session, body)
     msg = service._validate_message(service._sanitize_api_message(str(body.get("message", "")).strip()))
+    plan_acceptance = service._capture_plan_acceptance(session, body, msg)
     msg, immediate_reply = (await asyncio.to_thread(service._apply_chat_controls, session, body, msg)
                             if "fast_backend" in body or "system_one_backend" in body or "decision_assist_backend" in body
                             else service._apply_chat_controls(session, body, msg))
-    plan_action = body.get("plan_action")
-    plan_acceptance = ({"plan_id": plan_action["plan_id"], "version": plan_action["version"]}
-                       if isinstance(plan_action, dict)
-                       and str(plan_action.get("action") or "").lower() == "accept" else None)
     if not msg and immediate_reply is None:
         raise HTTPException(400, "Empty message")
     service._audit("CHAT_STREAM", f"user={session['user_id']} msg={msg[:80]}", session["user_id"])

openkyrozen/interfaces/web/service.py

@@ -148,7 +148,7 @@ def CHAT_HTML(self):
 
     from openkyrozen.interfaces.web.tasks import (_execute_durable_task, _task_manager, _task_worker, _task_scopes, _recover_task_scopes, _run_task_worker_cycle, _run_scheduled_job, api_v2_tasks, api_v2_create_task, api_v2_resume_task, api_v2_schedules, api_v2_create_schedule, api_v2_disable_schedule)
 
-    from openkyrozen.interfaces.web.chat import (_normalise_session_id, _normalise_profile, _normalise_memory_actor, _normalise_memory_context, _initialise_session_defaults, _set_memory_context, _actor_for_request, _get_or_create_session, _interaction_for_session, _history_for_session, _ensure_history_baseline, _public_history_node, _apply_chat_controls, _run_session_chat, _validate_message, _server_capabilities, _allowed_server_tools, _cost_summary, _cost_report, _json_object, api_chat, api_chat_stream, api_cancel_chat)
+    from openkyrozen.interfaces.web.chat import (_normalise_session_id, _normalise_profile, _normalise_memory_actor, _normalise_memory_context, _initialise_session_defaults, _set_memory_context, _actor_for_request, _get_or_create_session, _interaction_for_session, _history_for_session, _ensure_history_baseline, _public_history_node, _apply_chat_controls, _capture_plan_acceptance, _run_session_chat, _validate_message, _server_capabilities, _allowed_server_tools, _cost_summary, _cost_report, _json_object, api_chat, api_chat_stream, api_cancel_chat)
 
     from openkyrozen.interfaces.web.pages import (chat_page, fast_diagnostics, decision_assist_status, pwa_manifest)
 

tests/test_interaction.py

@@ -1391,6 +1391,20 @@ def test_queued_plan_acceptance_rejects_a_newer_proposal(self):
         self.assertEqual(session.interaction.state()["pending_plan"]["version"], newer["version"])
         self.assertEqual(session.interaction.state()["executing_plan"]["version"], first["version"])
 
+    def test_queued_acceptance_without_a_snapshot_cannot_take_a_later_plan(self):
+        session = self.agent.open_session("queued-empty-acceptance")
+        proposal = session.interaction.propose_plan({
+            "title": "Later proposal", "summary": "Remain pending until reviewed.",
+            "steps": [{"id": "step", "title": "Verify", "description": "Wait for explicit review.",
+                       "acceptance": ["The plan remains pending"]}],
+        })
+
+        reply = self.agent.chat(session, "accept plan", plan_acceptance={"plan_id": "", "version": 0})
+
+        self.assertIn("changed before acceptance", reply)
+        self.assertEqual(session.interaction.state()["pending_plan"]["plan_id"], proposal["plan_id"])
+        self.assertIsNone(session.interaction.state()["executing_plan"])
+
     def test_restore_history_refreshes_active_session_transcript(self):
         from openkyrozen.agent.conversation import ConversationHistory
         from types import SimpleNamespace

tests/test_server.py

@@ -273,6 +273,22 @@ def test_chat_interaction_controls_persist_and_restore_without_new_endpoint(self
                     self.assertEqual(run.call_args.kwargs["plan_acceptance"], {
                         "plan_id": plan["plan_id"], "version": plan["version"],
                     })
+
+                    typed_plan = controller.propose_plan({
+                        "title": "Typed acceptance", "summary": "Keep the reviewed identity.",
+                        "assumptions": [], "steps": [{
+                            "id": "typed-step", "title": "Verify", "description": "Verify the queue boundary.",
+                            "acceptance": ["The queued identity is preserved"],
+                        }],
+                    })
+                    with patch.object(server, "_run_session_chat", return_value="executed") as run:
+                        accepted = client.post("/api/chat", json={
+                            "session_id": "interaction-session", "message": "accept plan",
+                        })
+                    self.assertEqual(accepted.status_code, 200, accepted.text)
+                    self.assertEqual(run.call_args.kwargs["plan_acceptance"], {
+                        "plan_id": typed_plan["plan_id"], "version": typed_plan["version"],
+                    })
                     restored = client.get("/api/v2/sessions/interaction-session")
                     self.assertEqual(restored.status_code, 200, restored.text)
                     self.assertIn("interaction", restored.json())
@@ -290,6 +306,84 @@ def test_chat_interaction_controls_persist_and_restore_without_new_endpoint(self
             finally:
                 server._sessions = original_sessions
 
+    def test_streamed_typed_plan_acceptance_preserves_reviewed_identity(self):
+        client = TestClient(server.app)
+        with tempfile.TemporaryDirectory() as directory:
+            memory = MemoryBank(Path(directory) / "state.sqlite3", workspace_id="interaction-stream")
+            original_sessions = server._sessions
+            server._sessions = {}
+            try:
+                with patch.object(server._agent.current_session, "memory", memory), \
+                        patch.object(server, "_emit_chat_completed"):
+                    session = server._get_or_create_session("streamed-acceptance-session")
+                    controller = server._interaction_for_session(session)
+                    plan = controller.propose_plan({
+                        "title": "Streamed acceptance", "summary": "Keep the request identity.",
+                        "assumptions": [], "steps": [{
+                            "id": "streamed-step", "title": "Verify", "description": "Preserve this review.",
+                            "acceptance": ["The reviewed plan reaches execution"],
+                        }],
+                    })
+                    with patch.object(server, "_run_session_chat", return_value="executed") as run:
+                        response = client.post("/api/chat/stream", json={
+                            "session_id": "streamed-acceptance-session", "message": "accept plan",
+                        })
+                    self.assertEqual(response.status_code, 200, response.text)
+                    self.assertIn('"status": "completed"', response.text)
+                    self.assertEqual(run.call_args.kwargs["plan_acceptance"], {
+                        "plan_id": plan["plan_id"], "version": plan["version"],
+                    })
+            finally:
+                server._sessions = original_sessions
+
+    def test_textual_plan_acceptance_keeps_identity_across_awaited_controls(self):
+        with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {
+            "KYROZEN_DB_PATH": str(Path(directory) / "bootstrap.sqlite3"),
+            "KYROZEN_DISABLE_VECTOR_INDEX": "1",
+        }):
+            service = WebService(None)
+            client = TestClient(service.app)
+            memory = MemoryBank(Path(directory) / "state.sqlite3", workspace_id="interaction-await")
+            original_sessions = service._sessions
+            service._sessions = {}
+            try:
+                with patch.object(service._agent.current_session, "memory", memory), \
+                        patch.object(service, "_emit_chat_completed"):
+                    session = service._get_or_create_session("awaited-acceptance-session")
+                    controller = service._interaction_for_session(session)
+                    reviewed = controller.propose_plan({
+                        "title": "Reviewed plan", "summary": "Keep this identity across the await.",
+                        "assumptions": [], "steps": [{
+                            "id": "reviewed-step", "title": "Verify", "description": "Preserve this proposal.",
+                            "acceptance": ["The reviewed identity is passed through"],
+                        }],
+                    })
+
+                    def replace_proposal(_session, _body, message):
+                        controller.propose_plan({
+                            "title": "Newer plan", "summary": "This arrives while controls yield.",
+                            "assumptions": [], "steps": [{
+                                "id": "newer-step", "title": "Verify", "description": "Do not accept implicitly.",
+                                "acceptance": ["The newer proposal remains pending"],
+                            }],
+                        })
+                        return message, None
+
+                    with patch.object(service, "_apply_chat_controls", side_effect=replace_proposal), \
+                            patch.object(service, "_run_session_chat", return_value="executed") as run:
+                        response = client.post("/api/chat", json={
+                            "session_id": "awaited-acceptance-session", "message": "accept plan",
+                            "decision_assist_backend": "off",
+                        })
+
+                    self.assertEqual(response.status_code, 200, response.text)
+                    self.assertEqual(run.call_args.kwargs["plan_acceptance"], {
+                        "plan_id": reviewed["plan_id"], "version": reviewed["version"],
+                    })
+                    self.assertEqual(controller.state()["pending_plan"]["title"], "Newer plan")
+            finally:
+                service._sessions = original_sessions
+
     def test_json_post_boundaries_reject_non_objects_and_malformed_payloads(self):
         client = TestClient(server.app)
         endpoints = (

tests/test_tui_protocol.py

@@ -84,6 +84,39 @@ def test_plan_action_carries_the_reviewed_identity_into_the_worker(self):
 
         submit.assert_called_once_with("accept plan", "accept-1", plan_acceptance=plan)
 
+    def test_typed_plan_acceptance_snapshots_identity_before_worker_queue(self):
+        plan = {"plan_id": "plan-queued", "version": 7}
+        with patch.object(self.backend.agent, "is_plan_acceptance", return_value=True), \
+                patch.object(self.backend.agent._interaction_controller, "state", return_value={
+                    "pending_plan": plan,
+                }), patch("openkyrozen.interfaces.tui.turns.threading.Thread") as thread:
+            self.backend.submit("accept plan", "accept-queued")
+
+        self.assertEqual(thread.call_args.kwargs["args"], ("accept plan", "accept-queued", plan))
+
+    def test_typed_acceptance_without_a_plan_cannot_accept_a_later_proposal(self):
+        with patch.object(self.backend.agent, "is_plan_acceptance", return_value=True), \
+                patch.object(self.backend.agent._interaction_controller, "state", return_value={
+                    "pending_plan": None, "executing_plan": None,
+                }), patch("openkyrozen.interfaces.tui.turns.threading.Thread") as thread:
+            self.backend.submit("accept plan", "accept-empty")
+
+        self.assertEqual(thread.call_args.kwargs["args"], (
+            "accept plan", "accept-empty", {"plan_id": "", "version": 0},
+        ))
+
+    def test_plan_command_snapshots_once_before_a_later_proposal_can_appear(self):
+        plan = {"plan_id": "later-plan", "version": 2}
+        with patch.object(self.backend.agent._interaction_controller, "state", side_effect=[
+            {"pending_plan": None, "executing_plan": None}, {"pending_plan": plan},
+        ]) as state, patch("openkyrozen.interfaces.tui.turns.threading.Thread") as thread:
+            self.backend._command("plan", "accept", "accept-empty")
+
+        state.assert_called_once()
+        self.assertEqual(thread.call_args.kwargs["args"], (
+            "accept plan", "accept-empty", {"plan_id": "", "version": 0},
+        ))
+
     def test_permissions_jev_alias_is_normalized_before_persistence(self):
         with patch.object(self.backend.agent.fast_mode, "jev_key", return_value="configured"), \
                 patch.object(self.backend.agent, "set_permission_mode", return_value={