feat(agent): support inline subagent consultation
This commit is contained in:
Xubin Ren
+80
-14
@@ -146,7 +146,7 @@ class SubagentManager:
|
||||
self.runner = AgentRunner()
|
||||
self._exec_session_manager = ExecSessionManager()
|
||||
self._llm_wall_timeout_for_session = llm_wall_timeout_for_session
|
||||
self._running_tasks: dict[str, asyncio.Task[None]] = {}
|
||||
self._running_tasks: dict[str, asyncio.Task[str]] = {}
|
||||
self._task_statuses: dict[str, SubagentStatus] = {}
|
||||
self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...}
|
||||
|
||||
@@ -275,6 +275,54 @@ class SubagentManager:
|
||||
logger.info("Spawned subagent [{}]: {}", task_id, display_label)
|
||||
return f"Subagent [{display_label}] started (id: {task_id}). I'll notify you when it completes."
|
||||
|
||||
async def run_inline(
|
||||
self,
|
||||
task: str,
|
||||
label: str | None = None,
|
||||
origin_channel: str = "cli",
|
||||
origin_chat_id: str = "direct",
|
||||
session_key: str | None = None,
|
||||
origin_message_id: str | None = None,
|
||||
temperature: float | None = None,
|
||||
workspace_scope: WorkspaceScope | None = None,
|
||||
*,
|
||||
runtime: LLMRuntime | None = None,
|
||||
) -> str:
|
||||
"""Run a subagent synchronously and return its result to the caller."""
|
||||
if runtime is None:
|
||||
runtime = self._compat_spawn_runtime()
|
||||
if temperature is not None:
|
||||
runtime = runtime.with_generation_overrides(temperature=temperature)
|
||||
task_id = str(uuid.uuid4())[:8]
|
||||
display_label = label or task[:30] + ("..." if len(task) > 30 else "")
|
||||
origin = {
|
||||
"channel": origin_channel,
|
||||
"chat_id": origin_chat_id,
|
||||
"session_key": session_key,
|
||||
}
|
||||
status = SubagentStatus(
|
||||
task_id=task_id,
|
||||
label=display_label,
|
||||
task_description=task,
|
||||
started_at=time.monotonic(),
|
||||
)
|
||||
self._task_statuses[task_id] = status
|
||||
logger.info("Running inline subagent [{}]: {}", task_id, display_label)
|
||||
try:
|
||||
return await self._run_subagent(
|
||||
task_id,
|
||||
task,
|
||||
display_label,
|
||||
origin,
|
||||
status,
|
||||
runtime,
|
||||
origin_message_id,
|
||||
workspace_scope,
|
||||
announce=False,
|
||||
)
|
||||
finally:
|
||||
self._task_statuses.pop(task_id, None)
|
||||
|
||||
async def _run_subagent(
|
||||
self,
|
||||
task_id: str,
|
||||
@@ -285,7 +333,9 @@ class SubagentManager:
|
||||
runtime: LLMRuntime,
|
||||
origin_message_id: str | None = None,
|
||||
workspace_scope: WorkspaceScope | None = None,
|
||||
) -> None:
|
||||
*,
|
||||
announce: bool = True,
|
||||
) -> str:
|
||||
"""Execute the subagent task and announce the result."""
|
||||
logger.info("Subagent [{}] starting task: {}", task_id, label)
|
||||
|
||||
@@ -347,27 +397,43 @@ class SubagentManager:
|
||||
|
||||
if result.stop_reason == "tool_error":
|
||||
status.tool_events = list(result.tool_events)
|
||||
await self._announce_result(
|
||||
task_id, label, task,
|
||||
self._format_partial_progress(result),
|
||||
origin, "error", origin_message_id,
|
||||
)
|
||||
final_result = self._format_partial_progress(result)
|
||||
final_status = "error"
|
||||
elif result.stop_reason == "error":
|
||||
await self._announce_result(
|
||||
task_id, label, task,
|
||||
result.error or "Error: subagent execution failed.",
|
||||
origin, "error", origin_message_id,
|
||||
)
|
||||
final_result = result.error or "Error: subagent execution failed."
|
||||
final_status = "error"
|
||||
else:
|
||||
final_result = result.final_content or "Task completed but no final response was generated."
|
||||
final_status = "ok"
|
||||
logger.info("Subagent [{}] completed successfully", task_id)
|
||||
await self._announce_result(task_id, label, task, final_result, origin, "ok", origin_message_id)
|
||||
if announce:
|
||||
await self._announce_result(
|
||||
task_id,
|
||||
label,
|
||||
task,
|
||||
final_result,
|
||||
origin,
|
||||
final_status,
|
||||
origin_message_id,
|
||||
)
|
||||
return final_result
|
||||
|
||||
except Exception as e:
|
||||
status.phase = "error"
|
||||
status.error = str(e)
|
||||
logger.exception("Subagent [{}] failed", task_id)
|
||||
await self._announce_result(task_id, label, task, f"Error: {e}", origin, "error", origin_message_id)
|
||||
final_result = f"Error: {e}"
|
||||
if announce:
|
||||
await self._announce_result(
|
||||
task_id,
|
||||
label,
|
||||
task,
|
||||
final_result,
|
||||
origin,
|
||||
"error",
|
||||
origin_message_id,
|
||||
)
|
||||
return final_result
|
||||
|
||||
async def _announce_result(
|
||||
self,
|
||||
|
||||
@@ -6,7 +6,12 @@ from typing import TYPE_CHECKING, Any
|
||||
|
||||
from nanobot.agent.tools.base import Tool, ToolResult, tool_parameters
|
||||
from nanobot.agent.tools.context import current_request_context
|
||||
from nanobot.agent.tools.schema import NumberSchema, StringSchema, tool_parameters_schema
|
||||
from nanobot.agent.tools.schema import (
|
||||
BooleanSchema,
|
||||
NumberSchema,
|
||||
StringSchema,
|
||||
tool_parameters_schema,
|
||||
)
|
||||
from nanobot.security.workspace_access import current_workspace_scope
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -26,6 +31,14 @@ if TYPE_CHECKING:
|
||||
minimum=0.0,
|
||||
maximum=2.0,
|
||||
),
|
||||
wait=BooleanSchema(
|
||||
description=(
|
||||
"Wait for the subagent and return its result directly. Use this for a "
|
||||
"blocking consultation that must inform the current turn. Defaults to "
|
||||
"false for background execution."
|
||||
),
|
||||
default=False,
|
||||
),
|
||||
required=["task"],
|
||||
)
|
||||
)
|
||||
@@ -48,6 +61,7 @@ class SpawnTool(Tool):
|
||||
return (
|
||||
"Spawn a subagent to handle a task in the background. "
|
||||
"Use this for complex or time-consuming tasks that can run independently. "
|
||||
"Set wait=true for a consultation whose result must inform the current turn. "
|
||||
"The subagent will complete the task and report back when done. "
|
||||
"For deliverables or existing projects, inspect the workspace first "
|
||||
"and use a dedicated subdirectory when helpful."
|
||||
@@ -58,6 +72,7 @@ class SpawnTool(Tool):
|
||||
task: str,
|
||||
label: str | None = None,
|
||||
temperature: float | None = None,
|
||||
wait: bool = False,
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
"""Spawn a subagent to execute the given task."""
|
||||
@@ -75,7 +90,8 @@ class SpawnTool(Tool):
|
||||
origin_channel = request_ctx.channel
|
||||
origin_chat_id = request_ctx.chat_id
|
||||
session_key = request_ctx.session_key or f"{origin_channel}:{origin_chat_id}"
|
||||
return await self._manager.spawn(
|
||||
method = self._manager.run_inline if wait else self._manager.spawn
|
||||
return await method(
|
||||
task=task,
|
||||
runtime=request_ctx.runtime,
|
||||
label=label,
|
||||
|
||||
@@ -19,6 +19,37 @@ def _runtime(provider: MagicMock, model: str = "test-model") -> LLMRuntime:
|
||||
return LLMRuntime.capture(provider, model, context_window_tokens=128_000)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_inline_returns_result_without_announcement(tmp_path):
|
||||
"""Inline subagents return directly instead of injecting a follow-up."""
|
||||
from nanobot.agent.subagent import SubagentManager
|
||||
from nanobot.bus.queue import MessageBus
|
||||
|
||||
provider = MagicMock()
|
||||
manager = SubagentManager(
|
||||
workspace=tmp_path,
|
||||
bus=MessageBus(),
|
||||
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
|
||||
)
|
||||
manager.runner.run = AsyncMock(return_value=SimpleNamespace(
|
||||
stop_reason="done",
|
||||
final_content="review result",
|
||||
error=None,
|
||||
tool_events=[],
|
||||
))
|
||||
manager._announce_result = AsyncMock()
|
||||
|
||||
result = await manager.run_inline(
|
||||
task="review this",
|
||||
session_key="test:c1",
|
||||
runtime=_runtime(provider),
|
||||
)
|
||||
|
||||
assert result == "review result"
|
||||
manager._announce_result.assert_not_awaited()
|
||||
assert manager._task_statuses == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subagent_exec_tool_receives_allowed_env_keys(tmp_path):
|
||||
"""allowed_env_keys from ExecToolConfig must be forwarded to the subagent's ExecTool."""
|
||||
@@ -200,6 +231,40 @@ async def test_spawn_tool_rejects_when_at_concurrency_limit(tmp_path):
|
||||
await asyncio.gather(*mgr._running_tasks.values(), return_exceptions=True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spawn_tool_waits_for_inline_result():
|
||||
from nanobot.agent.tools.context import RequestContext, request_context
|
||||
from nanobot.agent.tools.spawn import SpawnTool
|
||||
|
||||
class Manager:
|
||||
max_concurrent_subagents = 1
|
||||
|
||||
def __init__(self):
|
||||
self.inline = AsyncMock(return_value="review result")
|
||||
self.spawn = AsyncMock(return_value="started")
|
||||
|
||||
def get_running_count(self):
|
||||
return 0
|
||||
|
||||
async def run_inline(self, **kwargs):
|
||||
return await self.inline(**kwargs)
|
||||
|
||||
manager = Manager()
|
||||
tool = SpawnTool(manager)
|
||||
runtime = _runtime(MagicMock())
|
||||
with request_context(RequestContext(
|
||||
channel="test",
|
||||
chat_id="c1",
|
||||
session_key="test:c1",
|
||||
runtime=runtime,
|
||||
)):
|
||||
result = await tool.execute(task="review this", wait=True)
|
||||
|
||||
assert result == "review result"
|
||||
manager.inline.assert_awaited_once()
|
||||
manager.spawn.assert_not_awaited()
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user