Rebase onto upstream (a4d95fd)
#12
@@ -16,6 +16,7 @@ from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFile
|
||||
from nanobot.agent.tools.shell import ExecTool
|
||||
from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
|
||||
from nanobot.agent.tools.spawn import SpawnTool
|
||||
from nanobot.agent.tools.wait import WaitForSubagentsTool
|
||||
|
||||
|
||||
class SubagentManager:
|
||||
@@ -45,6 +46,7 @@ class SubagentManager:
|
||||
self.restrict_to_workspace = restrict_to_workspace
|
||||
self._running_tasks: dict[str, asyncio.Task[None]] = {}
|
||||
self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...}
|
||||
self._task_results: dict[str, str] = {}
|
||||
|
||||
async def spawn(
|
||||
self,
|
||||
@@ -76,8 +78,8 @@ class SubagentManager:
|
||||
|
||||
bg_task.add_done_callback(_cleanup)
|
||||
|
||||
logger.info("Spawned subagent [{}]: {}", task_id, display_label)
|
||||
return f"Subagent [{display_label}] started (id: {task_id}). I'll notify you when it completes."
|
||||
logger.info(f"Spawned subagent [{task_id}]: {display_label}")
|
||||
return f"Subagent [{display_label}] started. Task ID: {task_id}"
|
||||
|
||||
async def _run_subagent(
|
||||
self,
|
||||
@@ -107,8 +109,9 @@ class SubagentManager:
|
||||
tools.register(WebSearchTool(api_key=self.brave_api_key))
|
||||
tools.register(WebFetchTool())
|
||||
spawn_tool = SpawnTool(manager=self)
|
||||
spawn_tool.set_context(origin["channel"], origin["chat_id"])
|
||||
spawn_tool.set_context("subagent", origin["chat_id"])
|
||||
tools.register(spawn_tool)
|
||||
tools.register(WaitForSubagentsTool(manager=self))
|
||||
|
||||
# Build messages with subagent-specific prompt
|
||||
system_prompt = self._build_subagent_prompt(task)
|
||||
@@ -190,6 +193,13 @@ class SubagentManager:
|
||||
"""Announce the subagent result to the main agent via the message bus."""
|
||||
status_text = "completed successfully" if status == "ok" else "failed"
|
||||
|
||||
# Child subagents (spawned by other subagents) store results silently.
|
||||
# The parent orchestrator collects them via wait_for_subagents.
|
||||
if origin["channel"] == "subagent":
|
||||
self._task_results[task_id] = result
|
||||
logger.debug(f"Subagent [{task_id}] stored result silently (child subagent)")
|
||||
return
|
||||
|
||||
announce_content = f"""[Subagent '{label}' {status_text}]
|
||||
|
||||
Task: {task}
|
||||
@@ -246,6 +256,7 @@ Skills are available at: {self.workspace}/skills/ (read SKILL.md files as needed
|
||||
|
||||
When you have completed the task, provide a clear summary of your findings or actions."""
|
||||
|
||||
<<<<<<< HEAD
|
||||
async def cancel_by_session(self, session_key: str) -> int:
|
||||
"""Cancel all subagents for the given session. Returns count cancelled."""
|
||||
tasks = [self._running_tasks[tid] for tid in self._session_tasks.get(session_key, [])
|
||||
@@ -255,6 +266,26 @@ When you have completed the task, provide a clear summary of your findings or ac
|
||||
if tasks:
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
return len(tasks)
|
||||
=======
|
||||
async def wait_for(self, task_ids: list[str]) -> str:
|
||||
"""Wait for specified child subagents to complete and return their results."""
|
||||
tasks_to_wait = [
|
||||
self._running_tasks[tid]
|
||||
for tid in task_ids
|
||||
if tid in self._running_tasks
|
||||
]
|
||||
if tasks_to_wait:
|
||||
await asyncio.gather(*tasks_to_wait, return_exceptions=True)
|
||||
|
||||
results = []
|
||||
for tid in task_ids:
|
||||
result = self._task_results.get(tid)
|
||||
if result is not None:
|
||||
results.append(f"[{tid}]:\n{result}")
|
||||
else:
|
||||
results.append(f"[{tid}]: No result found (invalid ID or task failed before storing)")
|
||||
return "\n\n---\n\n".join(results)
|
||||
>>>>>>> c606ab9 (Add wait_for_subagents tool and silence child subagent announcements)
|
||||
|
||||
def get_running_count(self) -> int:
|
||||
"""Return the number of currently running subagents."""
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Wait-for-subagents tool for orchestrator subagents."""
|
||||
|
||||
from typing import Any, TYPE_CHECKING
|
||||
|
||||
from nanobot.agent.tools.base import Tool
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from nanobot.agent.subagent import SubagentManager
|
||||
|
||||
|
||||
class WaitForSubagentsTool(Tool):
|
||||
"""
|
||||
Tool to wait for child subagents to complete and collect their results.
|
||||
|
||||
Use this after spawning multiple subagents to wait for all of them
|
||||
and get their results for synthesis.
|
||||
"""
|
||||
|
||||
def __init__(self, manager: "SubagentManager"):
|
||||
self._manager = manager
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "wait_for_subagents"
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return (
|
||||
"Wait for one or more child subagents to complete and return their results. "
|
||||
"Use this after spawning subagents to collect all results before synthesizing. "
|
||||
"Blocks until all specified subagents finish."
|
||||
)
|
||||
|
||||
@property
|
||||
def parameters(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_ids": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of task IDs to wait for (from spawn tool responses)",
|
||||
},
|
||||
},
|
||||
"required": ["task_ids"],
|
||||
}
|
||||
|
||||
async def execute(self, task_ids: list[str], **kwargs: Any) -> str:
|
||||
"""Wait for the specified subagents and return their results."""
|
||||
return await self._manager.wait_for(task_ids)
|
||||
Reference in New Issue
Block a user