diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index db3c5125..b4f4cece 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -13,6 +13,7 @@ from loguru import logger from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.runner import AgentRunner, AgentRunSpec +from nanobot.agent.tools.base import ToolResult from nanobot.agent.tools.context import ( RequestContext, ToolContext, @@ -308,8 +309,8 @@ class SubagentManager: ) self._task_statuses[task_id] = status logger.info("Running inline subagent [{}]: {}", task_id, display_label) - try: - return await self._run_subagent( + inline_task = asyncio.create_task( + self._run_subagent( task_id, task, display_label, @@ -320,8 +321,22 @@ class SubagentManager: workspace_scope, announce=False, ) + ) + self._running_tasks[task_id] = inline_task + if session_key: + self._session_tasks.setdefault(session_key, set()).add(task_id) + try: + result = await inline_task + if status.phase == "error" or status.stop_reason in {"error", "tool_error"}: + return ToolResult.error(result) + return result finally: + self._running_tasks.pop(task_id, None) self._task_statuses.pop(task_id, None) + if session_key and (ids := self._session_tasks.get(session_key)): + ids.discard(task_id) + if not ids: + del self._session_tasks[session_key] async def _run_subagent( self, diff --git a/tests/agent/tools/test_subagent_tools.py b/tests/agent/tools/test_subagent_tools.py index befed85c..ec496325 100644 --- a/tests/agent/tools/test_subagent_tools.py +++ b/tests/agent/tools/test_subagent_tools.py @@ -47,7 +47,40 @@ async def test_run_inline_returns_result_without_announcement(tmp_path): assert result == "review result" manager._announce_result.assert_not_awaited() + assert manager._running_tasks == {} assert manager._task_statuses == {} + assert manager._session_tasks == {} + + +@pytest.mark.asyncio +async def test_run_inline_returns_structured_error(tmp_path): + """Inline subagent failures remain tool errors for the parent runner.""" + from nanobot.agent.subagent import SubagentManager + from nanobot.agent.tools.registry import is_tool_error_result + from nanobot.bus.queue import MessageBus + + manager = SubagentManager( + workspace=tmp_path, + bus=MessageBus(), + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + ) + manager.runner.run = AsyncMock(return_value=SimpleNamespace( + stop_reason="error", + final_content=None, + error="subagent failed", + tool_events=[], + )) + + result = await manager.run_inline( + task="review this", + session_key="test:c1", + runtime=_runtime(MagicMock()), + ) + + assert result == "subagent failed" + assert is_tool_error_result("spawn", result) + assert manager._running_tasks == {} + assert manager._session_tasks == {} @pytest.mark.asyncio @@ -265,6 +298,86 @@ async def test_spawn_tool_waits_for_inline_result(): manager.spawn.assert_not_awaited() +@pytest.mark.asyncio +async def test_inline_spawn_counts_toward_concurrency_limit(tmp_path): + from nanobot.agent.subagent import SubagentManager + from nanobot.agent.tools.context import RequestContext, request_context + from nanobot.agent.tools.spawn import SpawnTool + from nanobot.bus.queue import MessageBus + + manager = SubagentManager( + workspace=tmp_path, + bus=MessageBus(), + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + max_concurrent_subagents=1, + ) + release = asyncio.Event() + entered = asyncio.Event() + + async def fake_run(spec): + entered.set() + await release.wait() + return SimpleNamespace( + stop_reason="done", + final_content="done", + error=None, + tool_events=[], + ) + + manager.runner.run = AsyncMock(side_effect=fake_run) + tool = SpawnTool(manager) + with request_context(RequestContext( + channel="test", + chat_id="c1", + session_key="test:c1", + runtime=_runtime(MagicMock()), + )): + first = asyncio.create_task(tool.execute(task="first", wait=True)) + await asyncio.wait_for(entered.wait(), timeout=1.0) + + second = await tool.execute(task="second", wait=True) + + assert "concurrency limit reached" in second + assert manager.get_running_count() == 1 + release.set() + assert await first == "done" + + assert manager.get_running_count() == 0 + assert manager._session_tasks == {} + + +@pytest.mark.asyncio +async def test_cancel_by_session_cancels_inline_subagent(tmp_path): + from nanobot.agent.subagent import SubagentManager + from nanobot.bus.queue import MessageBus + + manager = SubagentManager( + workspace=tmp_path, + bus=MessageBus(), + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + ) + entered = asyncio.Event() + + async def fake_run(spec): + entered.set() + await asyncio.Event().wait() + + manager.runner.run = AsyncMock(side_effect=fake_run) + inline = asyncio.create_task(manager.run_inline( + task="wait", + session_key="test:c1", + runtime=_runtime(MagicMock()), + )) + await asyncio.wait_for(entered.wait(), timeout=1.0) + + assert await manager.cancel_by_session("test:c1") == 1 + with pytest.raises(asyncio.CancelledError): + await inline + assert manager._running_tasks == {} + assert manager._task_statuses == {} + assert manager._session_tasks == {} + + def test_subagent_default_max_concurrent_matches_agent_defaults(tmp_path): """Direct SubagentManager construction should use the agent default concurrency limit.""" from nanobot.agent.subagent import SubagentManager