diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index 0015b712f3b..a9569647bab 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -97,6 +97,7 @@ def __init__( webui_runtime_model_name: Callable[[], str | None] | None = None, webui_cron_pending_job_ids: Callable[[str], set[str]] | None = None, webui_local_trigger_pending_ids: Callable[[str], set[str]] | None = None, + webui_session_discard: Callable[[str], Awaitable[None]] | None = None, webui_static_dist: bool = True, webui_runtime_surface: str = "browser", webui_runtime_capabilities: dict[str, Any] | None = None, @@ -118,6 +119,7 @@ def __init__( self._webui_runtime_model_name = webui_runtime_model_name self._webui_cron_pending_job_ids = webui_cron_pending_job_ids self._webui_local_trigger_pending_ids = webui_local_trigger_pending_ids + self._webui_session_discard = webui_session_discard self._webui_static_dist = webui_static_dist self._webui_runtime_surface = webui_runtime_surface self._webui_runtime_capabilities = dict(webui_runtime_capabilities or {}) @@ -189,6 +191,7 @@ def _build_channel( local_trigger_store=self._local_trigger_store, cron_pending_job_ids=self._webui_cron_pending_job_ids, local_trigger_pending_ids=self._webui_local_trigger_pending_ids, + session_discard=self._webui_session_discard, channel_feature_action=self.apply_channel_feature_action, channel_runtime_status=self.get_status, mcp_runtime_status=self._webui_mcp_runtime_status, diff --git a/nanobot/channels/websocket/tests/test_websocket_http_routes.py b/nanobot/channels/websocket/tests/test_websocket_http_routes.py index 4f77b455f90..c7df13c2b78 100644 --- a/nanobot/channels/websocket/tests/test_websocket_http_routes.py +++ b/nanobot/channels/websocket/tests/test_websocket_http_routes.py @@ -73,6 +73,7 @@ def _make_handler( local_trigger_store: LocalTriggerStore | None = None, cron_pending_job_ids: Any | None = None, local_trigger_pending_ids: Any | None = None, + session_discard: Any | None = None, channel_feature_action: Any | None = None, channel_runtime_status: Any | None = None, mcp_reload: Any | None = None, @@ -93,6 +94,7 @@ def _make_handler( local_trigger_store=local_trigger_store, cron_pending_job_ids=cron_pending_job_ids, local_trigger_pending_ids=local_trigger_pending_ids, + session_discard=session_discard, channel_feature_action=channel_feature_action, channel_runtime_status=channel_runtime_status, mcp_reload=mcp_reload, @@ -111,6 +113,7 @@ def _ch( local_trigger_store: LocalTriggerStore | None = None, cron_pending_job_ids: Any | None = None, local_trigger_pending_ids: Any | None = None, + session_discard: Any | None = None, channel_feature_action: Any | None = None, channel_runtime_status: Any | None = None, mcp_reload: Any | None = None, @@ -135,6 +138,7 @@ def _ch( local_trigger_store=local_trigger_store, cron_pending_job_ids=cron_pending_job_ids, local_trigger_pending_ids=local_trigger_pending_ids, + session_discard=session_discard, channel_feature_action=channel_feature_action, channel_runtime_status=channel_runtime_status, mcp_reload=mcp_reload, @@ -2245,18 +2249,39 @@ async def test_session_delete_removes_file( from nanobot.webui.transcript import append_transcript_object append_transcript_object("websocket:doomed", {"event": "user", "chat_id": "doomed", "text": "x"}) - channel = _ch(bus, session_manager=sm, port=29903) + discard_started = asyncio.Event() + allow_discard = asyncio.Event() + + async def discard_session(key: str) -> None: + assert key == "websocket:doomed" + discard_started.set() + await allow_discard.wait() + + channel = _ch( + bus, + session_manager=sm, + session_discard=discard_session, + port=29903, + ) server_task = asyncio.create_task(channel.start()) try: path = sm._get_session_path("websocket:doomed") assert path.exists() webui_path = tmp_path / "webui" / f"{SessionManager.safe_key('websocket:doomed')}.jsonl" assert webui_path.is_file() - resp = await _webui_mutate( - channel, - "session.delete", - {"key": "websocket:doomed"}, - ) + delete_task = asyncio.create_task( + _webui_mutate( + channel, + "session.delete", + {"key": "websocket:doomed"}, + ) + ) + await asyncio.wait_for(discard_started.wait(), timeout=2) + assert path.exists() + assert webui_path.is_file() + + allow_discard.set() + resp = await delete_task assert resp.status_code == 200 assert resp.json()["deleted"] is True assert not path.exists() @@ -2266,6 +2291,28 @@ async def test_session_delete_removes_file( await server_task +@pytest.mark.asyncio +async def test_session_delete_requires_lifecycle_control( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_session(tmp_path, key="websocket:doomed") + channel = _ch(bus, session_manager=sm, port=_free_port()) + server_task = asyncio.create_task(channel.start()) + try: + path = sm._get_session_path("websocket:doomed") + response = await _webui_mutate( + channel, + "session.delete", + {"key": "websocket:doomed"}, + ) + + assert response.status_code == 503 + assert path.exists() + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions( bus: MagicMock, tmp_path: Path @@ -2649,6 +2696,7 @@ async def test_session_delete_blocks_and_cascades_local_triggers( bus, session_manager=sm, local_trigger_store=trigger_store, + session_discard=AsyncMock(), port=port, ) server_task = asyncio.create_task(channel.start()) @@ -2697,7 +2745,13 @@ async def test_session_delete_can_cascade_bound_automations( channel="websocket", to="doomed", ) - channel = _ch(bus, session_manager=sm, cron_service=cron, port=29916) + channel = _ch( + bus, + session_manager=sm, + cron_service=cron, + session_discard=AsyncMock(), + port=29916, + ) server_task = asyncio.create_task(channel.start()) try: path = sm._get_session_path("websocket:doomed") @@ -2766,7 +2820,12 @@ async def test_session_delete_action_accepts_websocket_keys( bus: MagicMock, tmp_path: Path ) -> None: sm = _seed_session(tmp_path, key="websocket:encoded-key") - channel = _ch(bus, session_manager=sm, port=29910) + channel = _ch( + bus, + session_manager=sm, + session_discard=AsyncMock(), + port=29910, + ) server_task = asyncio.create_task(channel.start()) try: path = sm._get_session_path("websocket:encoded-key") diff --git a/nanobot/cli/gateway_runtime.py b/nanobot/cli/gateway_runtime.py index f31cb0e90ee..7da97b3b69b 100644 --- a/nanobot/cli/gateway_runtime.py +++ b/nanobot/cli/gateway_runtime.py @@ -673,6 +673,7 @@ def _webui_skill_state_action(disabled_skills: set[str]) -> None: webui_runtime_model_name=_webui_runtime_model_name, webui_cron_pending_job_ids=agent.pending_cron_job_ids_for_session, webui_local_trigger_pending_ids=agent.pending_local_trigger_ids_for_session, + webui_session_discard=agent.discard_session, webui_static_dist=webui_static_dist, webui_runtime_surface=webui_runtime_surface, webui_runtime_capabilities=webui_runtime_capabilities, diff --git a/nanobot/webui/gateway_services.py b/nanobot/webui/gateway_services.py index 9820fdbb73b..43a9e5ec4c6 100644 --- a/nanobot/webui/gateway_services.py +++ b/nanobot/webui/gateway_services.py @@ -63,6 +63,7 @@ def build_gateway_services( local_trigger_store: LocalTriggerStore | None = None, cron_pending_job_ids: Callable[[str], set[str]] | None = None, local_trigger_pending_ids: Callable[[str], set[str]] | None = None, + session_discard: Callable[[str], Awaitable[None]] | None = None, channel_feature_action: Callable[..., Any] | None = None, channel_runtime_status: Callable[[], dict[str, Any]] | None = None, mcp_runtime_status: Callable[[], Mapping[str, str]] | None = None, @@ -117,6 +118,7 @@ def build_gateway_services( local_trigger_store=local_trigger_store, cron_pending_job_ids=cron_pending_job_ids, local_trigger_pending_ids=local_trigger_pending_ids, + session_discard=session_discard, channel_feature_action=channel_feature_action, channel_runtime_status=channel_runtime_status, mcp_runtime_status=mcp_runtime_status, diff --git a/nanobot/webui/ws_http.py b/nanobot/webui/ws_http.py index 363db6d3f0a..b643852441b 100644 --- a/nanobot/webui/ws_http.py +++ b/nanobot/webui/ws_http.py @@ -305,6 +305,7 @@ def __init__( local_trigger_store: LocalTriggerStore | None = None, cron_pending_job_ids: Callable[[str], set[str]] | None = None, local_trigger_pending_ids: Callable[[str], set[str]] | None = None, + session_discard: Callable[[str], Awaitable[None]] | None = None, channel_feature_action: Callable[..., Any] | None = None, channel_runtime_status: Callable[[], dict[str, Any]] | None = None, mcp_runtime_status: Callable[[], Mapping[str, str]] | None = None, @@ -332,6 +333,7 @@ def __init__( self.local_trigger_store = local_trigger_store self.cron_pending_job_ids = cron_pending_job_ids self.local_trigger_pending_ids = local_trigger_pending_ids + self.session_discard = session_discard self._log = log self._runtime_surface = runtime_surface @@ -652,7 +654,7 @@ async def _dispatch_session_routes(self, request: WsRequest, got: str) -> Respon m = re.match(r"^/api/sessions/([^/]+)/delete$", got) if m: - return self._handle_session_delete(request, m.group(1)) + return await self._handle_session_delete(request, m.group(1)) return None @@ -814,7 +816,7 @@ def _handle_session_automations(self, request: WsRequest, key: str) -> Response: ) ) - def _handle_session_delete(self, request: WsRequest, key: str) -> Response: + async def _handle_session_delete(self, request: WsRequest, key: str) -> Response: if not self.check_api_token(request): return _http_error(401, "Unauthorized") if self.session_manager is None: @@ -839,6 +841,9 @@ def _handle_session_delete(self, request: WsRequest, key: str) -> Response: "automations": serialize_automation_jobs(automation_jobs), } ) + if self.session_discard is None: + return _http_error(503, "session lifecycle unavailable") + await self.session_discard(decoded_key) if automation_jobs: for job in automation_jobs: if isinstance(job, LocalTrigger): diff --git a/tests/agent/test_loop_session_policy.py b/tests/agent/test_loop_session_policy.py index 3a2a065f505..b09925c114c 100644 --- a/tests/agent/test_loop_session_policy.py +++ b/tests/agent/test_loop_session_policy.py @@ -107,6 +107,38 @@ async def test_missing_required_session_cannot_fall_back_to_disk(tmp_path) -> No assert loop.sessions.read_session_file(key) is None +@pytest.mark.asyncio +async def test_persistent_session_discard_cancels_without_archiving(tmp_path) -> None: + provider_started = asyncio.Event() + + async def block_provider(**_kwargs: object) -> LLMResponse: + provider_started.set() + await asyncio.Event().wait() + raise AssertionError("provider blocker unexpectedly released") + + loop = _loop(tmp_path, []) + loop.provider.chat_with_retry = AsyncMock(side_effect=block_provider) + loop.consolidator.maybe_consolidate_by_tokens = AsyncMock() + key = "websocket:persistent-cancelled" + session = loop.sessions.get_or_create(key) + session.add_message("user", "existing history") + loop.sessions.save(session) + + active_task = asyncio.create_task(loop._dispatch(_message(key, "discard this turn"))) + loop._active_tasks[key] = {active_task} + await asyncio.wait_for(provider_started.wait(), timeout=2) + + await loop.discard_session(key) + + assert active_task.cancelled() + assert loop.sessions.get_cached(key) is None + assert loop.context.memory.read_unprocessed_history(since_cursor=0) == [] + + assert loop.sessions.delete_session(key) is True + await asyncio.sleep(0) + assert loop.sessions.read_session_file(key) is None + + @pytest.mark.asyncio async def test_session_discard_control_cancels_active_turn(tmp_path, monkeypatch) -> None: provider_started = asyncio.Event() diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index b55bb2fd6f9..166a2252472 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -96,6 +96,9 @@ def pending_cron_job_ids_for_session(_session_key: str) -> set[str]: def pending_local_trigger_ids_for_session(_session_key: str) -> set[str]: return set() + async def discard_session(self, _session_key: str) -> None: + return None + async def submit_local_trigger_turn( self, _msg: InboundMessage,