fix(agent): track inline subagent lifecycle

This commit is contained in:
Xubin Ren 2026-07-24 17:16:15 +08:00
parent 3a400e0207
commit 7e15c4c447
2 changed files with 130 additions and 2 deletions

View File

@ -13,6 +13,7 @@ from loguru import logger
from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.hook import AgentHook, AgentHookContext
from nanobot.agent.runner import AgentRunner, AgentRunSpec from nanobot.agent.runner import AgentRunner, AgentRunSpec
from nanobot.agent.tools.base import ToolResult
from nanobot.agent.tools.context import ( from nanobot.agent.tools.context import (
RequestContext, RequestContext,
ToolContext, ToolContext,
@ -308,8 +309,8 @@ class SubagentManager:
) )
self._task_statuses[task_id] = status self._task_statuses[task_id] = status
logger.info("Running inline subagent [{}]: {}", task_id, display_label) logger.info("Running inline subagent [{}]: {}", task_id, display_label)
try: inline_task = asyncio.create_task(
return await self._run_subagent( self._run_subagent(
task_id, task_id,
task, task,
display_label, display_label,
@ -320,8 +321,22 @@ class SubagentManager:
workspace_scope, workspace_scope,
announce=False, 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: finally:
self._running_tasks.pop(task_id, None)
self._task_statuses.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( async def _run_subagent(
self, self,

View File

@ -47,7 +47,40 @@ async def test_run_inline_returns_result_without_announcement(tmp_path):
assert result == "review result" assert result == "review result"
manager._announce_result.assert_not_awaited() manager._announce_result.assert_not_awaited()
assert manager._running_tasks == {}
assert manager._task_statuses == {} 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 @pytest.mark.asyncio
@ -265,6 +298,86 @@ async def test_spawn_tool_waits_for_inline_result():
manager.spawn.assert_not_awaited() 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): def test_subagent_default_max_concurrent_matches_agent_defaults(tmp_path):
"""Direct SubagentManager construction should use the agent default concurrency limit.""" """Direct SubagentManager construction should use the agent default concurrency limit."""
from nanobot.agent.subagent import SubagentManager from nanobot.agent.subagent import SubagentManager