fix(agents): collapse repeated waits queued inside one model turn

tests/test_wait_dedupe.py

@@ -0,0 +1,126 @@
+"""Tests for collapsing repeated waits queued inside one model turn.
+
+An orchestrator that writes out its whole poll loop ahead of time queues
+many ``wait_for_agents`` calls in a single response. Each one parks for its
+full timeout, so the agent stops reacting for hours while its children run
+unsupervised. Only the first wait of a turn parks; the rest return at once.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+from agents import RunContextWrapper
+from agents.tool_context import ToolContext
+
+from strix.core.agents import AgentCoordinator
+from strix.core.hooks import LLM_TURN_KEY, ReportUsageHooks
+from strix.tools.agents_graph.tools import wait_for_agents
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+_WAIT_SECONDS = 2
+
+
+@pytest.fixture
+def _fast_wait(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    # The real ceiling is 300s per wait; the shape of the bug is the same.
+    monkeypatch.setattr(
+        "strix.tools.agents_graph.tools._WAIT_DEFAULT_TIMEOUT_S", _WAIT_SECONDS, raising=True
+    )
+    yield
+
+
+async def _context() -> dict[str, Any]:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    return {"agent_id": "root", "coordinator": coordinator}
+
+
+async def _wait(inner: dict[str, Any]) -> dict[str, Any]:
+    ctx = ToolContext(
+        context=inner,
+        tool_name="wait_for_agents",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+    raw: str = await wait_for_agents.on_invoke_tool(
+        ctx, json.dumps({"reason": "waiting for wave 1", "timeout_seconds": _WAIT_SECONDS})
+    )
+    return cast("dict[str, Any]", json.loads(raw))
+
+
+@pytest.mark.asyncio
+async def test_waits_queued_in_one_turn_each_park_without_the_guard(_fast_wait: None) -> None:
+    # Repro: no turn marker in context (as before the fix) — every queued wait
+    # parks for its full timeout, so N waits cost N x timeout.
+    inner = await _context()
+
+    started = time.monotonic()
+    outcomes = [(await _wait(inner))["wait_outcome"] for _ in range(3)]
+    elapsed = time.monotonic() - started
+
+    assert outcomes == ["timeout", "timeout", "timeout"]
+    assert elapsed >= 3 * _WAIT_SECONDS
+
+
+@pytest.mark.asyncio
+async def test_repeated_waits_in_one_turn_are_collapsed(_fast_wait: None) -> None:
+    inner = await _context()
+    inner[LLM_TURN_KEY] = 1
+
+    started = time.monotonic()
+    outcomes = [(await _wait(inner))["wait_outcome"] for _ in range(3)]
+    elapsed = time.monotonic() - started
+
+    assert outcomes == ["timeout", "already_waited", "already_waited"]
+    assert elapsed < 2 * _WAIT_SECONDS
+
+
+@pytest.mark.asyncio
+async def test_a_wait_in_the_next_turn_still_parks(_fast_wait: None) -> None:
+    inner = await _context()
+    inner[LLM_TURN_KEY] = 1
+    assert (await _wait(inner))["wait_outcome"] == "timeout"
+    assert (await _wait(inner))["wait_outcome"] == "already_waited"
+
+    inner[LLM_TURN_KEY] = 2
+
+    assert (await _wait(inner))["wait_outcome"] == "timeout"
+
+
+@pytest.mark.asyncio
+async def test_each_model_turn_bumps_the_turn_marker() -> None:
+    hooks = ReportUsageHooks(model="gw-model")
+    context: RunContextWrapper[dict[str, Any]] = RunContextWrapper(context={})
+    agent = cast("Any", None)
+
+    await hooks.on_llm_start(context, agent, None, [])
+    await hooks.on_llm_start(context, agent, None, [])
+
+    assert context.context[LLM_TURN_KEY] == 2
+
+
+@pytest.mark.asyncio
+async def test_a_collapsed_wait_still_reports_arriving_messages(_fast_wait: None) -> None:
+    inner = await _context()
+    inner[LLM_TURN_KEY] = 1
+    coordinator = cast("AgentCoordinator", inner["coordinator"])
+
+    async def _send() -> None:
+        await asyncio.sleep(0.1)
+        await coordinator.send("root", {"type": "information", "content": "child done"})
+
+    task = asyncio.create_task(_send())
+    first = await _wait(inner)
+    await task
+
+    assert first["wait_outcome"] == "message_arrived"
+    assert (await _wait(inner))["wait_outcome"] == "already_waited"