From c410ea444c75d2161eb19edb1c4d001082897a88 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Sat, 8 Aug 2026 22:55:09 +0900 Subject: [PATCH] fix(agent): stop session-owned exec processes --- nanobot/agent/loop.py | 7 ++++--- tests/agent/test_loop_session_policy.py | 9 ++++++++- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 6f73f0534..46a33b53d 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -788,9 +788,9 @@ class AgentLoop: logger.warning("Command '{}' matched but dispatch returned None", raw) async def _cancel_active_tasks(self, key: str) -> int: - """Cancel and await all active tasks and subagents for *key*. + """Cancel and await all active work for *key*. - Returns the total number of cancelled tasks + subagents. + Returns the total number of cancelled tasks, subagents, and exec sessions. """ tasks = tuple(self._active_tasks.pop(key, set())) cancelled = sum(1 for t in tasks if not t.done() and t.cancel()) @@ -798,7 +798,8 @@ class AgentLoop: with suppress(asyncio.CancelledError, Exception): await t sub_cancelled = await self.subagents.cancel_by_session(key) - return cancelled + sub_cancelled + exec_cancelled = await self._exec_session_manager.terminate_by_owner(key) + return cancelled + sub_cancelled + exec_cancelled async def discard_session(self, key: str) -> None: """Stop active work for *key* and forget its cached session.""" diff --git a/tests/agent/test_loop_session_policy.py b/tests/agent/test_loop_session_policy.py index a74337905..88dae71e3 100644 --- a/tests/agent/test_loop_session_policy.py +++ b/tests/agent/test_loop_session_policy.py @@ -119,12 +119,18 @@ async def test_session_discard_control_cancels_active_turn(tmp_path, monkeypatch loop = _loop(tmp_path, []) async def wait_for_discard(key: str) -> None: - while loop.sessions.get_cached(key) is not None: + while loop.sessions.get_cached(key) is not None or key in loop._discarding_sessions: await asyncio.sleep(0) loop.provider.chat_with_retry = AsyncMock(side_effect=block_provider) monkeypatch.setattr(loop, "_connect_mcp", AsyncMock()) monkeypatch.setattr(loop, "close_mcp", AsyncMock()) + terminate_exec_sessions = AsyncMock(return_value=1) + monkeypatch.setattr( + loop._exec_session_manager, + "terminate_by_owner", + terminate_exec_sessions, + ) key = "websocket:transient-cancelled" loop.sessions.get_or_create_transient( key, @@ -152,6 +158,7 @@ async def test_session_discard_control_cancels_active_turn(tmp_path, monkeypatch await asyncio.wait_for(active_task, timeout=2) await asyncio.wait_for(wait_for_discard(key), timeout=2) assert loop.sessions.get_cached(key) is None + terminate_exec_sessions.assert_awaited_once_with(key) loop.stop() await loop.bus.publish_inbound(_message(key, "wake"))