diff --git a/tests/agent/tools/test_subagent_tools.py b/tests/agent/tools/test_subagent_tools.py index 35b29e06f..ffcbe4c12 100644 --- a/tests/agent/tools/test_subagent_tools.py +++ b/tests/agent/tools/test_subagent_tools.py @@ -469,102 +469,6 @@ async def test_agent_loop_syncs_updated_max_iterations_before_run(tmp_path): loop.runner.run.assert_awaited_once() -@pytest.mark.asyncio -async def test_drain_pending_waits_only_at_terminal_boundary(tmp_path): - """Ordinary drains stay non-blocking while terminal drains await subagent results.""" - from nanobot.agent.loop import AgentLoop - from nanobot.bus.events import InboundMessage - from nanobot.bus.queue import MessageBus - from nanobot.session.manager import Session - - bus = MessageBus() - provider = MagicMock() - provider.get_default_model.return_value = "test-model" - - loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model") - - pending_queue: asyncio.Queue[InboundMessage] = asyncio.Queue() - session = Session(key="test:drain-block") - injection_callback = None - terminal_injection_callback = None - - # Capture the injection_callback that _run_agent_loop creates - async def fake_runner_run(spec): - nonlocal injection_callback, terminal_injection_callback - injection_callback = spec.injection_callback - terminal_injection_callback = spec.terminal_injection_callback - - return SimpleNamespace( - stop_reason="done", - final_content="done", - error=None, - tool_events=[], - messages=[], - usage=None, - had_injections=False, - tools_used=[], - provider_state=None, - ) - - loop.runner.run = AsyncMock(side_effect=fake_runner_run) - - # Register a running sub-agent in the SubagentManager for this session - async def _hang_forever(): - await asyncio.Event().wait() - - hang_task = asyncio.create_task(_hang_forever()) - loop.subagents._session_tasks.setdefault(session.key, set()).add("sub-drain-1") - loop.subagents._running_tasks["sub-drain-1"] = hang_task - - # Run _run_agent_loop — this defines the _drain_pending closure - await loop._run_agent_loop( - [{"role": "user", "content": "test"}], - runtime=loop.llm_runtime(), - session=session, - channel="test", - chat_id="c1", - pending_queue=pending_queue, - ) - - assert injection_callback is not None - assert terminal_injection_callback is not None - - # Tool-boundary drains must let the runner start its next model iteration. - assert await asyncio.wait_for(injection_callback(), timeout=1.0) == [] - - # Once the runner is ready to exit, it may wait for a background result. - drain_task = asyncio.create_task(terminal_injection_callback()) - - # Let the task enter the blocking queue wait. - await asyncio.sleep(0) - - # Should still be running (blocked on pending_queue.get()) - assert not drain_task.done(), "terminal drain should wait while subagents are running" - - # Now put a message in the queue (simulating sub-agent completion) - await pending_queue.put(InboundMessage( - sender_id="subagent", - channel="test", - chat_id="c1", - content="Sub-agent result", - media=None, - metadata={}, - )) - - # Should unblock and return results - results = await asyncio.wait_for(drain_task, timeout=2.0) - assert len(results) >= 1 - assert results[0]["role"] == "user" - assert "Sub-agent result" in str(results[0]["content"]) - - # Cleanup - hang_task.cancel() - try: - await hang_task - except asyncio.CancelledError: - pass - - @pytest.mark.asyncio async def test_drain_pending_no_block_when_no_subagents(tmp_path): """_drain_pending should not block when no sub-agents are running."""