From b9e7c7f6fe67253fecc387df49604ae40e4bc6b4 Mon Sep 17 00:00:00 2001 From: chengyongru <61816729+chengyongru@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:53:28 +0800 Subject: [PATCH] fix: queue concurrent subagents (#5566) * fix: queue concurrent subagents * chore: keep spawn schema concise --- docs/configuration.md | 4 +- nanobot/agent/subagent.py | 33 ++++- nanobot/agent/tools/spawn.py | 13 +- nanobot/config/schema.py | 2 +- tests/agent/test_subagent_lifecycle.py | 29 +++++ tests/agent/tools/test_subagent_tools.py | 154 +++++++++++++++++++---- 6 files changed, 197 insertions(+), 38 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 0cb779006..f687db940 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2213,7 +2213,7 @@ The notification gate runs on a built-in system prompt. Advanced users can overr ## Subagent Concurrency -By default, nanobot only allows one spawned subagent at a time. When the limit is reached, the `spawn` tool returns an error so the agent can decide to wait or rearrange its work. This protects local LLM servers from loading multiple KV caches at once. If your provider can handle more parallel work, raise the limit: +By default, nanobot allows four subagents to run at the same time. Additional subagents wait for capacity instead of being rejected. Lower the limit if a local model server cannot hold multiple KV caches, or raise it when the provider can handle more parallel work: ```json { @@ -2229,7 +2229,7 @@ The deprecated `agents.defaults.failOnToolError` field is silently ignored when | Option | Default | Description | |--------|---------|-------------| -| `agents.defaults.maxConcurrentSubagents` | `1` | Maximum number of spawned subagents that may run at the same time. Attempts to spawn beyond this limit return an error. | +| `agents.defaults.maxConcurrentSubagents` | `4` | Maximum number of subagents that may run at the same time. Additional tasks wait for capacity. | ## Auto Compact diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index 7a3053948..1f2b75cf8 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -55,7 +55,8 @@ class SubagentStatus: label: str task_description: str started_at: float # time.monotonic() - phase: str = "initializing" # initializing | awaiting_tools | tools_completed | final_response | done | error + # queued | initializing | awaiting_tools | tools_completed | final_response | done | error + phase: str = "initializing" iteration: int = 0 tool_events: list[dict[str, str]] = field(default_factory=list) usage: LLMUsage | None = None @@ -147,6 +148,7 @@ class SubagentManager: if max_concurrent_subagents is not None else defaults.max_concurrent_subagents ) + self._run_slots = asyncio.Semaphore(self.max_concurrent_subagents) self.runner = AgentRunner() self._exec_session_manager = ExecSessionManager() self._llm_wall_timeout_for_session = llm_wall_timeout_for_session @@ -363,6 +365,35 @@ class SubagentManager: workspace_scope: WorkspaceScope | None = None, *, announce: bool = True, + ) -> str: + """Wait for capacity, then execute one subagent task.""" + status.phase = "queued" + async with self._run_slots: + status.phase = "initializing" + return await self._run_admitted_subagent( + task_id, + task, + label, + origin, + status, + runtime, + origin_message_id, + workspace_scope, + announce=announce, + ) + + async def _run_admitted_subagent( + self, + task_id: str, + task: str, + label: str, + origin: _SubagentOrigin, + status: SubagentStatus, + runtime: LLMRuntime, + origin_message_id: str | None = None, + workspace_scope: WorkspaceScope | None = None, + *, + announce: bool = True, ) -> str: """Execute the subagent task and announce the result.""" logger.info("Subagent [{}] starting task: {}", task_id, label) diff --git a/nanobot/agent/tools/spawn.py b/nanobot/agent/tools/spawn.py index 1936434e1..8d5581c6a 100644 --- a/nanobot/agent/tools/spawn.py +++ b/nanobot/agent/tools/spawn.py @@ -73,6 +73,11 @@ class SpawnTool(Tool): "and use a dedicated subdirectory when helpful." ) + @property + def concurrency_safe(self) -> bool: + """Each call owns its task state; the manager serializes capacity admission.""" + return True + async def execute( self, task: str, @@ -82,14 +87,6 @@ class SpawnTool(Tool): **kwargs: Any, ) -> str: """Spawn a subagent to execute the given task.""" - running = self._manager.get_running_count() - limit = self._manager.max_concurrent_subagents - if running >= limit: - return ( - f"Cannot spawn subagent: concurrency limit reached " - f"({running}/{limit} running). Wait for a running subagent " - f"to complete before spawning a new one." - ) request_ctx = current_request_context() if request_ctx is None or request_ctx.runtime is None: return ToolResult.error("Error: spawn requires an active model runtime") diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index 25b997b78..946c71c78 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -128,7 +128,7 @@ class AgentDefaults(Base): temperature: float = 0.1 fallback_models: list[FallbackCandidate] = Field(default_factory=list) max_tool_iterations: int = 200 - max_concurrent_subagents: int = Field(default=1, ge=1) + max_concurrent_subagents: int = Field(default=4, ge=1) max_tool_result_chars: int = 16_000 provider_retry_mode: Literal["standard", "persistent"] = "standard" tool_hint_max_length: int = Field( diff --git a/tests/agent/test_subagent_lifecycle.py b/tests/agent/test_subagent_lifecycle.py index 35ed5aa49..a082b427d 100644 --- a/tests/agent/test_subagent_lifecycle.py +++ b/tests/agent/test_subagent_lifecycle.py @@ -520,6 +520,35 @@ class TestCancelBySession: count = await sm.cancel_by_session("nonexistent") assert count == 0 + @pytest.mark.asyncio + async def test_cancels_active_and_queued_tasks(self, tmp_path): + sm = _manager(tmp_path, max_concurrent_subagents=1) + active_entered = asyncio.Event() + queued_entered = asyncio.Event() + + async def _blocked_run(spec): + task = spec.initial_messages[-1]["content"] + if task == "active": + active_entered.set() + else: + queued_entered.set() + await asyncio.Event().wait() + + sm.runner.run = _blocked_run + runtime = _runtime() + await sm.spawn("active", runtime=runtime, session_key="s1") + await asyncio.wait_for(active_entered.wait(), timeout=1.0) + await sm.spawn("queued", runtime=runtime, session_key="s1") + await asyncio.sleep(0) + + assert not queued_entered.is_set() + assert await sm.cancel_by_session("s1") == 2 + await asyncio.sleep(0) + + assert not queued_entered.is_set() + assert sm._running_tasks == {} + assert sm._session_tasks == {} + @pytest.mark.asyncio async def test_already_done_not_counted(self, tmp_path): sm = _manager(tmp_path) diff --git a/tests/agent/tools/test_subagent_tools.py b/tests/agent/tools/test_subagent_tools.py index ddba42271..92ecc451b 100644 --- a/tests/agent/tools/test_subagent_tools.py +++ b/tests/agent/tools/test_subagent_tools.py @@ -211,8 +211,8 @@ async def test_spawn_forwards_temperature_to_run_spec(tmp_path): @pytest.mark.asyncio -async def test_spawn_tool_rejects_when_at_concurrency_limit(tmp_path): - """SpawnTool should return an error string when the concurrency limit is reached.""" +async def test_background_spawn_waits_for_concurrency_capacity(tmp_path): + """Background tasks should be accepted and start when capacity becomes available.""" from nanobot.agent.subagent import SubagentManager from nanobot.agent.tools.spawn import SpawnTool from nanobot.bus.queue import MessageBus @@ -224,14 +224,23 @@ async def test_spawn_tool_rejects_when_at_concurrency_limit(tmp_path): workspace=tmp_path, bus=bus, max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + max_concurrent_subagents=1, ) mgr._announce_result = AsyncMock() - # Block the first subagent so it stays "running" - release = asyncio.Event() + first_entered = asyncio.Event() + second_entered = asyncio.Event() + release_first = asyncio.Event() + release_second = asyncio.Event() async def fake_run(spec): - await release.wait() + task = spec.initial_messages[-1]["content"] + if task == "first task": + first_entered.set() + await release_first.wait() + else: + second_entered.set() + await release_second.wait() return SimpleNamespace( stop_reason="done", final_content="done", @@ -250,19 +259,24 @@ async def test_spawn_tool_rejects_when_at_concurrency_limit(tmp_path): session_key="test:c1", runtime=_runtime(provider), )): - # First spawn succeeds - result = await tool.execute(task="first task") - assert "started" in result + first_result = await tool.execute(task="first task") + assert "started" in first_result + await asyncio.wait_for(first_entered.wait(), timeout=1.0) - # Second spawn should be rejected (default limit is 1) - result = await tool.execute(task="second task") - assert "Cannot spawn subagent" in result - assert "concurrency limit reached" in result + second_result = await tool.execute(task="second task") + assert "started" in second_result + tasks = list(mgr._running_tasks.values()) + await asyncio.sleep(0) + assert not second_entered.is_set() + phases = {status.task_description: status.phase for status in mgr._task_statuses.values()} + assert phases == {"first task": "initializing", "second task": "queued"} - # Release the first subagent - release.set() - # Allow cleanup - await asyncio.gather(*mgr._running_tasks.values(), return_exceptions=True) + release_first.set() + await asyncio.wait_for(second_entered.wait(), timeout=1.0) + release_second.set() + await asyncio.gather(*tasks, return_exceptions=True) + await asyncio.sleep(0) + assert mgr._running_tasks == {} @pytest.mark.asyncio @@ -300,7 +314,7 @@ async def test_spawn_tool_waits_for_inline_result(): @pytest.mark.asyncio -async def test_inline_spawn_counts_toward_concurrency_limit(tmp_path): +async def test_inline_spawn_waits_for_concurrency_capacity(tmp_path): from nanobot.agent.subagent import SubagentManager from nanobot.agent.tools.context import RequestContext, request_context from nanobot.agent.tools.spawn import SpawnTool @@ -312,12 +326,19 @@ async def test_inline_spawn_counts_toward_concurrency_limit(tmp_path): max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, max_concurrent_subagents=1, ) - release = asyncio.Event() - entered = asyncio.Event() + first_entered = asyncio.Event() + second_entered = asyncio.Event() + release_first = asyncio.Event() + release_second = asyncio.Event() async def fake_run(spec): - entered.set() - await release.wait() + task = spec.initial_messages[-1]["content"] + if task == "first": + first_entered.set() + await release_first.wait() + else: + second_entered.set() + await release_second.wait() return SimpleNamespace( stop_reason="done", final_content="done", @@ -334,19 +355,99 @@ async def test_inline_spawn_counts_toward_concurrency_limit(tmp_path): runtime=_runtime(MagicMock()), )): first = asyncio.create_task(tool.execute(task="first", wait=True)) - await asyncio.wait_for(entered.wait(), timeout=1.0) + await asyncio.wait_for(first_entered.wait(), timeout=1.0) - second = await tool.execute(task="second", wait=True) + second = asyncio.create_task(tool.execute(task="second", wait=True)) + await asyncio.sleep(0) - assert "concurrency limit reached" in second - assert manager.get_running_count() == 1 - release.set() + assert not second.done() + assert not second_entered.is_set() + assert manager.get_running_count() == 2 + release_first.set() assert await first == "done" + await asyncio.wait_for(second_entered.wait(), timeout=1.0) + release_second.set() + assert await second == "done" assert manager.get_running_count() == 0 assert manager._session_tasks == {} +@pytest.mark.asyncio +async def test_runner_executes_inline_spawn_batch_concurrently(tmp_path): + """Adjacent blocking consultations should share one concurrent tool batch.""" + from nanobot.agent.runner import AgentRunner, AgentRunSpec + from nanobot.agent.subagent import SubagentManager + from nanobot.agent.tools.context import RequestContext, request_context + from nanobot.agent.tools.registry import ToolRegistry + from nanobot.agent.tools.spawn import SpawnTool + from nanobot.bus.queue import MessageBus + from nanobot.providers.base import ToolCallRequest + + manager = SubagentManager( + workspace=tmp_path, + bus=MessageBus(), + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + max_concurrent_subagents=2, + ) + both_entered = asyncio.Event() + release = asyncio.Event() + entered: list[str] = [] + + async def fake_run(spec): + entered.append(spec.initial_messages[-1]["content"]) + if len(entered) == 2: + both_entered.set() + await release.wait() + return SimpleNamespace( + stop_reason="done", + final_content=spec.initial_messages[-1]["content"], + error=None, + tool_events=[], + ) + + manager.runner.run = AsyncMock(side_effect=fake_run) + tools = ToolRegistry() + tools.register(SpawnTool(manager)) + runtime = _runtime(MagicMock()) + spec = AgentRunSpec( + initial_messages=[], + tools=tools, + runtime=runtime, + max_iterations=1, + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + concurrent_tools=True, + ) + calls = [ + ToolCallRequest( + id="spawn-1", + name="spawn", + arguments={"task": "first", "wait": True}, + ), + ToolCallRequest( + id="spawn-2", + name="spawn", + arguments={"task": "second", "wait": True}, + ), + ] + + with request_context(RequestContext( + channel="test", + chat_id="c1", + session_key="test:c1", + runtime=runtime, + )): + execution = asyncio.create_task(AgentRunner()._execute_tools(spec, calls, {}, {})) + await asyncio.wait_for(both_entered.wait(), timeout=1.0) + release.set() + results, events = await execution + + assert set(entered) == {"first", "second"} + assert results == ["first", "second"] + assert [event["status"] for event in events] == ["ok", "ok"] + assert manager._running_tasks == {} + + @pytest.mark.asyncio async def test_cancel_by_session_cancels_inline_subagent(tmp_path): from nanobot.agent.subagent import SubagentManager @@ -391,6 +492,7 @@ def test_subagent_default_max_concurrent_matches_agent_defaults(tmp_path): max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, ) + assert AgentDefaults().max_concurrent_subagents == 4 assert mgr.max_concurrent_subagents == AgentDefaults().max_concurrent_subagents