diff --git a/openkyrozen/interfaces/tui/commands.py b/openkyrozen/interfaces/tui/commands.py index a56053b4..456c26b3 100644 --- a/openkyrozen/interfaces/tui/commands.py +++ b/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() diff --git a/openkyrozen/interfaces/tui/turns.py b/openkyrozen/interfaces/tui/turns.py index 4e804ed2..24fefa82 100644 --- a/openkyrozen/interfaces/tui/turns.py +++ b/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.") diff --git a/openkyrozen/interfaces/web/chat.py b/openkyrozen/interfaces/web/chat.py index 4a38b123..73c9e4fc 100644 --- a/openkyrozen/interfaces/web/chat.py +++ b/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"]) diff --git a/openkyrozen/interfaces/web/service.py b/openkyrozen/interfaces/web/service.py index ff34ee69..d78856b2 100644 --- a/openkyrozen/interfaces/web/service.py +++ b/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) diff --git a/tests/test_interaction.py b/tests/test_interaction.py index 9a118787..b656d891 100644 --- a/tests/test_interaction.py +++ b/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 diff --git a/tests/test_server.py b/tests/test_server.py index a841c260..efabb939 100644 --- a/tests/test_server.py +++ b/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 = ( diff --git a/tests/test_tui_protocol.py b/tests/test_tui_protocol.py index 32cbebf8..2ba01ed5 100644 --- a/tests/test_tui_protocol.py +++ b/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={