From 9e8ec1522376adcd8d414d550b7d4b4ccc372bfb Mon Sep 17 00:00:00 2001 From: chengyongru Date: Wed, 1 Jul 2026 16:03:51 +0800 Subject: [PATCH] fix(agent): route direct subagent results in-turn --- nanobot/agent/loop.py | 12 +++++ nanobot/agent/subagent.py | 20 +++++++++ nanobot/agent/tools/long_task.py | 5 ++- nanobot/templates/agent/subagent_announce.md | 5 ++- nanobot/templates/agent/subagent_system.md | 2 +- tests/agent/tools/test_subagent_tools.py | 47 ++++++++++++++++++++ 6 files changed, 88 insertions(+), 3 deletions(-) diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index d2b446993..5242f53d1 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -1852,13 +1852,17 @@ class AgentLoop: ) # Share the dispatch lock so direct calls serialize with bus turns. lock = self._session_locks.setdefault(session_key, asyncio.Lock()) + pending: asyncio.Queue[InboundMessage] = asyncio.Queue(maxsize=20) try: async with lock: + self._pending_queues[session_key] = pending + self.subagents.set_direct_result_queue(session_key, pending) kwargs: dict[str, Any] = { "session_key": session_key, "on_progress": on_progress, "on_stream": on_stream, "on_stream_end": on_stream_end, + "pending_queue": pending, "ephemeral": ephemeral, } if _run_extra_hooks_for_ephemeral: @@ -1872,5 +1876,13 @@ class AgentLoop: **kwargs, ) finally: + self.subagents.clear_direct_result_queue(session_key, pending) + if self._pending_queues.get(session_key) is pending: + self._pending_queues.pop(session_key, None) + while True: + try: + await self.bus.publish_inbound(pending.get_nowait()) + except asyncio.QueueEmpty: + break await self._runtime_events().run_status_changed(msg, session_key, "idle") self._runtime_events().clear_turn(session_key) diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index 85dbce664..487f6b861 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -118,6 +118,22 @@ class SubagentManager: self._running_tasks: dict[str, asyncio.Task[None]] = {} self._task_statuses: dict[str, SubagentStatus] = {} self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...} + self._direct_result_queues: dict[str, asyncio.Queue[InboundMessage]] = {} + + def set_direct_result_queue( + self, + session_key: str, + queue: asyncio.Queue[InboundMessage], + ) -> None: + self._direct_result_queues[session_key] = queue + + def clear_direct_result_queue( + self, + session_key: str, + queue: asyncio.Queue[InboundMessage], + ) -> None: + if self._direct_result_queues.get(session_key) is queue: + self._direct_result_queues.pop(session_key, None) def _subagent_tools_config(self) -> ToolsConfig: """Build a ToolsConfig scoped for subagent use.""" @@ -335,6 +351,10 @@ class SubagentManager: metadata=metadata, ) + if queue := self._direct_result_queues.get(override): + await queue.put(msg) + logger.debug("Subagent [{}] queued result directly for {}", task_id, override) + return await self.bus.publish_inbound(msg) logger.debug("Subagent [{}] announced result to {}:{}", task_id, origin['channel'], origin['chat_id']) diff --git a/nanobot/agent/tools/long_task.py b/nanobot/agent/tools/long_task.py index 12fcec174..9d23542e7 100644 --- a/nanobot/agent/tools/long_task.py +++ b/nanobot/agent/tools/long_task.py @@ -97,7 +97,8 @@ class _GoalToolsMixin(ContextAware): "Sustained objective for this chat thread. First read the built-in **long-goal** skill, " "especially its Start fast section, then call this promptly once the user's intent is clear. " "The goal must still be idempotent, self-contained, bounded, and explicit about done-ness; " - "do not delay this tool call to over-plan, research, or decide execution details.", + "do not delay this tool call to over-plan, research, or decide execution details. " + "Do not use this for a single current-turn answer, including one that uses spawn subagents.", max_length=12_000, ), ui_summary=StringSchema( @@ -139,6 +140,8 @@ class LongTaskTool(Tool, _GoalToolsMixin): def description(self) -> str: return ( "Mark this thread as a sustained long-running task. " + "Use only when the user wants work to persist across future turns or background check-ins; " + "do not use for a single current-turn answer, including one that uses spawn subagents. " "First read the built-in **long-goal** skill, especially its Start fast section; then call this " "as soon as the user's intent is clear. Write a good idempotent goal, but do not delay the tool " "call with long planning, research, or execution-detail thinking. " diff --git a/nanobot/templates/agent/subagent_announce.md b/nanobot/templates/agent/subagent_announce.md index de8fdad39..0a6c3afe1 100644 --- a/nanobot/templates/agent/subagent_announce.md +++ b/nanobot/templates/agent/subagent_announce.md @@ -5,4 +5,7 @@ Task: {{ task }} Result: {{ result }} -Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like "subagent" or task IDs. +Use this result as evidence for the current turn. For MapReduce-style work, +preserve any Summary / Evidence / Open issues structure when reducing multiple +results. Mention gaps or failures if they affect the answer; avoid exposing +internal task IDs unless they are needed for clarity. diff --git a/nanobot/templates/agent/subagent_system.md b/nanobot/templates/agent/subagent_system.md index 6d98d9e1c..838ae193b 100644 --- a/nanobot/templates/agent/subagent_system.md +++ b/nanobot/templates/agent/subagent_system.md @@ -8,7 +8,7 @@ If this task is one slice of a larger MapReduce-style effort, treat yourself as the map step: do only the assigned slice, avoid cross-slice coordination, and leave reduction or final synthesis to the main agent. -End with a compact, mergeable result: +For MapReduce-style slices, end with a compact, mergeable result: - Summary: what you found or changed - Evidence: relevant files, commands, URLs, or observations diff --git a/tests/agent/tools/test_subagent_tools.py b/tests/agent/tools/test_subagent_tools.py index 6a037981d..cbfccf008 100644 --- a/tests/agent/tools/test_subagent_tools.py +++ b/tests/agent/tools/test_subagent_tools.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from nanobot.bus.events import OutboundMessage from nanobot.config.schema import AgentDefaults _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars @@ -482,3 +483,49 @@ async def test_drain_pending_timeout(tmp_path): await hang_task except asyncio.CancelledError: pass + + +@pytest.mark.asyncio +async def test_process_direct_routes_subagent_results_to_pending_queue(tmp_path): + """Single-message CLI mode should consume subagent announcements mid-turn.""" + from nanobot.agent.loop import AgentLoop + from nanobot.bus.events import InboundMessage + from nanobot.bus.queue import MessageBus + + loop = AgentLoop( + bus=MessageBus(), + provider=MagicMock(), + workspace=tmp_path, + model="test-model", + ) + loop._connect_mcp = AsyncMock() # type: ignore[method-assign] + + async def fake_process_message(msg, **kwargs): + pending_queue = kwargs["pending_queue"] + await loop.bus.publish_inbound(InboundMessage( + channel="other", + sender_id="u", + chat_id="room", + content="unrelated", + )) + await loop.subagents._announce_result( + "sub-1", + "label", + "task", + "subagent result", + {"channel": "cli", "chat_id": "direct", "session_key": "cli:direct"}, + "ok", + ) + routed = await asyncio.wait_for(pending_queue.get(), timeout=1) + assert "subagent result" in routed.content + assert routed.metadata["subagent_task_id"] == "sub-1" + return OutboundMessage(channel="cli", chat_id="direct", content="done") + + loop._process_message = fake_process_message # type: ignore[method-assign] + + response = await loop.process_direct("start", session_key="cli:direct") + + assert response is not None + assert response.content == "done" + unrelated = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=1) + assert unrelated.content == "unrelated"