Compare commits

...
7 changed files with 99 additions and 2 deletions
+12
View File
@@ -1852,13 +1852,17 @@ class AgentLoop:
) )
# Share the dispatch lock so direct calls serialize with bus turns. # Share the dispatch lock so direct calls serialize with bus turns.
lock = self._session_locks.setdefault(session_key, asyncio.Lock()) lock = self._session_locks.setdefault(session_key, asyncio.Lock())
pending: asyncio.Queue[InboundMessage] = asyncio.Queue(maxsize=20)
try: try:
async with lock: async with lock:
self._pending_queues[session_key] = pending
self.subagents.set_direct_result_queue(session_key, pending)
kwargs: dict[str, Any] = { kwargs: dict[str, Any] = {
"session_key": session_key, "session_key": session_key,
"on_progress": on_progress, "on_progress": on_progress,
"on_stream": on_stream, "on_stream": on_stream,
"on_stream_end": on_stream_end, "on_stream_end": on_stream_end,
"pending_queue": pending,
"ephemeral": ephemeral, "ephemeral": ephemeral,
} }
if _run_extra_hooks_for_ephemeral: if _run_extra_hooks_for_ephemeral:
@@ -1872,5 +1876,13 @@ class AgentLoop:
**kwargs, **kwargs,
) )
finally: 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") await self._runtime_events().run_status_changed(msg, session_key, "idle")
self._runtime_events().clear_turn(session_key) self._runtime_events().clear_turn(session_key)
+20
View File
@@ -118,6 +118,22 @@ class SubagentManager:
self._running_tasks: dict[str, asyncio.Task[None]] = {} self._running_tasks: dict[str, asyncio.Task[None]] = {}
self._task_statuses: dict[str, SubagentStatus] = {} self._task_statuses: dict[str, SubagentStatus] = {}
self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...} 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: def _subagent_tools_config(self) -> ToolsConfig:
"""Build a ToolsConfig scoped for subagent use.""" """Build a ToolsConfig scoped for subagent use."""
@@ -335,6 +351,10 @@ class SubagentManager:
metadata=metadata, 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) await self.bus.publish_inbound(msg)
logger.debug("Subagent [{}] announced result to {}:{}", task_id, origin['channel'], origin['chat_id']) logger.debug("Subagent [{}] announced result to {}:{}", task_id, origin['channel'], origin['chat_id'])
+4 -1
View File
@@ -97,7 +97,8 @@ class _GoalToolsMixin(ContextAware):
"Sustained objective for this chat thread. First read the built-in **long-goal** skill, " "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. " "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; " "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, max_length=12_000,
), ),
ui_summary=StringSchema( ui_summary=StringSchema(
@@ -139,6 +140,8 @@ class LongTaskTool(Tool, _GoalToolsMixin):
def description(self) -> str: def description(self) -> str:
return ( return (
"Mark this thread as a sustained long-running task. " "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 " "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 " "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. " "call with long planning, research, or execution-detail thinking. "
+3
View File
@@ -63,6 +63,9 @@ class SpawnTool(Tool, ContextAware):
return ( return (
"Spawn a subagent to handle a task in the background. " "Spawn a subagent to handle a task in the background. "
"Use this for complex or time-consuming tasks that can run independently. " "Use this for complex or time-consuming tasks that can run independently. "
"For MapReduce-style work, spawn only independent map slices with clear "
"boundaries; keep reduction, conflict resolution, and final user-facing "
"synthesis in the main agent. "
"The subagent will complete the task and report back when done. " "The subagent will complete the task and report back when done. "
"For deliverables or existing projects, inspect the workspace first " "For deliverables or existing projects, inspect the workspace first "
"and use a dedicated subdirectory when helpful." "and use a dedicated subdirectory when helpful."
+4 -1
View File
@@ -5,4 +5,7 @@ Task: {{ task }}
Result: Result:
{{ 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.
@@ -4,6 +4,15 @@
You are a subagent spawned by the main agent to complete a specific task. You are a subagent spawned by the main agent to complete a specific task.
Stay focused on the assigned task. Your final response will be reported back to the main agent. Stay focused on the assigned task. Your final response will be reported back to the main agent.
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.
For MapReduce-style slices, end with a compact, mergeable result:
- Summary: what you found or changed
- Evidence: relevant files, commands, URLs, or observations
- Open issues: blockers, failures, or "none"
{% include 'agent/_snippets/untrusted_content.md' %} {% include 'agent/_snippets/untrusted_content.md' %}
+47
View File
@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
from nanobot.bus.events import OutboundMessage
from nanobot.config.schema import AgentDefaults from nanobot.config.schema import AgentDefaults
_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
@@ -482,3 +483,49 @@ async def test_drain_pending_timeout(tmp_path):
await hang_task await hang_task
except asyncio.CancelledError: except asyncio.CancelledError:
pass 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"