test(subagent): verify cascade exec termination on /stop

- terminate_by_owner kills matching sessions, skips others, handles
  empty owner case
- cancel_by_session calls terminate_by_owner on the session key

tests/agent/test_task_cancel.py

@@ -274,6 +274,27 @@ class TestSubagentCancellation:
         )
         assert await mgr.cancel_by_session("nonexistent") == 0
 
+    @pytest.mark.asyncio
+    async def test_cancel_by_session_terminates_exec_sessions(self):
+        from nanobot.agent.subagent import SubagentManager
+        from nanobot.agent.tools.exec_session import ExecSessionManager
+        from nanobot.bus.queue import MessageBus
+
+        bus = MessageBus()
+        mgr = SubagentManager(
+            workspace=MagicMock(),
+            bus=bus,
+            max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        )
+        # Replace the real exec session manager with a mock
+        mock_exec_mgr = AsyncMock(spec=ExecSessionManager)
+        mock_exec_mgr.terminate_by_owner = AsyncMock(return_value=0)
+        mgr._exec_session_manager = mock_exec_mgr
+
+        await mgr.cancel_by_session("test:c1")
+
+        mock_exec_mgr.terminate_by_owner.assert_awaited_once_with("test:c1")
+
     @pytest.mark.asyncio
     async def test_subagent_preserves_reasoning_fields_in_tool_turn(self, monkeypatch, tmp_path):
         from nanobot.agent.subagent import SubagentManager

tests/tools/test_exec_session_tools.py

@@ -622,6 +622,88 @@ def test_agent_loop_shutdown_attempts_all_cleanup_after_errors(monkeypatch):
     asyncio.run(run())
 
 
+def test_terminate_by_owner_kills_matching_sessions(tmp_path):
+    async def run() -> None:
+        manager = ExecSessionManager()
+        tool = ExecTool(working_dir=str(tmp_path), timeout=30, session_manager=manager)
+
+        token_a = bind_request_context(
+            RequestContext(channel="cli", chat_id="a", session_key="cli:a")
+        )
+        try:
+            initial_a = await tool.execute(
+                command=_waiting_shell_command("a_ready"),
+                yield_time_ms=100,
+            )
+        finally:
+            reset_request_context(token_a)
+        sid_a = _session_id(initial_a)
+
+        token_b = bind_request_context(
+            RequestContext(channel="cli", chat_id="b", session_key="cli:b")
+        )
+        try:
+            initial_b = await tool.execute(
+                command=_waiting_shell_command("b_ready"),
+                yield_time_ms=100,
+            )
+        finally:
+            reset_request_context(token_b)
+        sid_b = _session_id(initial_b)
+
+        proc_a = manager._sessions[sid_a].process
+        proc_b = manager._sessions[sid_b].process
+        assert proc_a.returncode is None
+        assert proc_b.returncode is None
+
+        killed = await manager.terminate_by_owner("cli:a")
+
+        assert killed == 1
+        assert proc_a.returncode is not None
+        assert proc_b.returncode is None
+        assert sid_a not in manager._sessions
+        assert sid_b in manager._sessions
+
+        await manager.close_all()
+
+    asyncio.run(run())
+
+
+def test_terminate_by_owner_returns_zero_for_no_match(tmp_path):
+    async def run() -> None:
+        manager = ExecSessionManager()
+        killed = await manager.terminate_by_owner("nonexistent")
+        assert killed == 0
+        assert manager._sessions == {}
+
+    asyncio.run(run())
+
+
+def test_terminate_by_owner_skips_sessions_without_owner_key(tmp_path):
+    async def run() -> None:
+        manager = ExecSessionManager()
+        tool = ExecTool(working_dir=str(tmp_path), timeout=30, session_manager=manager)
+
+        # Spawn without owner (no request context)
+        initial = await tool.execute(
+            command=_waiting_shell_command("ready"),
+            yield_time_ms=100,
+        )
+        sid = _session_id(initial)
+        proc = manager._sessions[sid].process
+        assert proc.returncode is None
+
+        killed = await manager.terminate_by_owner("cli:a")
+
+        assert killed == 0
+        assert proc.returncode is None
+        assert sid in manager._sessions
+
+        await manager.close_all()
+
+    asyncio.run(run())
+
+
 def test_agent_loop_shutdown_preserves_single_cleanup_error(monkeypatch):
     async def run() -> None:
         loop = object.__new__(AgentLoop)