diff --git a/strix/core/agents.py b/strix/core/agents.py index 7e738be1..a1c82e75 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -501,16 +501,24 @@ class AgentCoordinator: agent_id, len(items), ) - else: - try: - async with session_write_lock(session): - await session.add_items(items) - except Exception: - logger.exception( - "failed to append %d queued messages to the session of %s", - len(items), - agent_id, + return 0, [] + try: + async with session_write_lock(session): + await session.add_items(items) + except Exception: + logger.exception( + "failed to append %d queued messages to the session of %s; " + "restoring them to the mailbox for retry", + len(items), + agent_id, + ) + async with self._lock: + runtime.mailbox[0:0] = queued + self.pending_counts[agent_id] = self.pending_counts.get(agent_id, 0) + len( + queued ) + runtime.wake.set() + return 0, [] await self._maybe_snapshot() if not include_items: return count, [] diff --git a/tests/test_agent_graph_coordination.py b/tests/test_agent_graph_coordination.py index 9694656a..ef3e8dc9 100644 --- a/tests/test_agent_graph_coordination.py +++ b/tests/test_agent_graph_coordination.py @@ -13,6 +13,7 @@ import json from typing import TYPE_CHECKING, Any, cast import pytest +from agents.memory import SQLiteSession from agents.tool_context import ToolContext from strix.core.agents import AgentCoordinator @@ -162,8 +163,12 @@ async def test_wait_returns_at_once_when_no_child_can_answer() -> None: @pytest.mark.asyncio -async def test_wait_delivers_a_pending_report_before_checking_liveness() -> None: +async def test_wait_delivers_a_pending_report_before_checking_liveness( + tmp_path: Path, +) -> None: coordinator = await _graph(interactive=False) + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) await coordinator.send("root", {"from": "child", "type": "completion", "content": "done"}) await coordinator.set_status("child", "completed") diff --git a/tests/test_execution.py b/tests/test_execution.py index 0daf0da2..1913ad25 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -723,6 +723,92 @@ async def test_send_queues_without_session_and_drains_on_consume(tmp_path: Any) session.close() +@pytest.mark.asyncio +async def test_consume_pending_restores_mailbox_on_session_write_failure( + tmp_path: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + Regression test for https://github.com/usestrix/strix/issues/1107 + + If session.add_items fails, the drained message must not be lost: it should + be restored to the mailbox (so a retry can pick it up) rather than reported + as delivered via a positive count. + """ + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) + + assert await coordinator.send("root", {"from": "user", "content": "hello"}) is True + + async def _boom(*_args: Any, **_kwargs: Any) -> None: + raise RuntimeError("db is locked") + + monkeypatch.setattr(session, "add_items", _boom) + + count, items = await coordinator.consume_pending("root", include_items=True) + assert (count, items) == (0, []) + + runtime = coordinator.runtimes["root"] + assert runtime.mailbox == [{"from": "user", "content": "hello"}] + assert coordinator.pending_counts["root"] == 1 + + monkeypatch.undo() + count, items = await coordinator.consume_pending("root", include_items=True) + assert count == 1 + assert items[0]["content"] == "hello" + session.close() + + +@pytest.mark.asyncio +async def test_consume_pending_restore_preserves_arrival_order( + tmp_path: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + A message that arrives while the failed write is being restored must stay ordered + after the message that was already draining, not before it. Regression for + https://github.com/usestrix/strix/pull/1161#pullrequestreview (itzzdev09's flagged + scenario 3): restoring via mailbox[0:0] = queued, not append, is what preserves this. + """ + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) + + assert await coordinator.send("root", {"from": "user", "content": "first"}) is True + + async def _boom(*_args: Any, **_kwargs: Any) -> None: + raise RuntimeError("db is locked") + + monkeypatch.setattr(session, "add_items", _boom) + await coordinator.consume_pending("root", include_items=True) + monkeypatch.undo() + + assert await coordinator.send("root", {"from": "user", "content": "second"}) is True + + count, items = await coordinator.consume_pending("root", include_items=True) + assert count == 2 + assert [item["content"] for item in items] == ["first", "second"] + session.close() + + +@pytest.mark.asyncio +async def test_consume_pending_reports_zero_when_no_session_attached() -> None: + """ + Regression for a gap flagged alongside #1107: an agent with no SDK session attached + already logs that queued messages were not persisted, but used to still report them + as delivered via a positive count. The messages are genuinely dropped here (there is + no session to retry against), so the return value must say so. + """ + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + assert await coordinator.send("root", {"from": "user", "content": "hello"}) is True + + count, items = await coordinator.consume_pending("root", include_items=True) + assert (count, items) == (0, []) + + @pytest.mark.asyncio async def test_error_parked_agent_only_released_by_user_message(tmp_path: Any) -> None: coordinator = AgentCoordinator() @@ -1099,6 +1185,7 @@ async def test_interactive_recovery_exhaustion_parks_instead_of_crashing( @pytest.mark.asyncio async def test_interactive_subagent_exhaustion_tells_its_parent( + tmp_path: Any, monkeypatch: pytest.MonkeyPatch, ) -> None: """A parked child must report up so its parent stops waiting on it. @@ -1110,6 +1197,8 @@ async def test_interactive_subagent_exhaustion_tells_its_parent( coordinator = AgentCoordinator() await coordinator.register("root", "strix", parent_id=None) await coordinator.register("child", "recon", parent_id="root") + root_session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=root_session) calls: list[Any] = [] monkeypatch.setattr( execution, @@ -1125,6 +1214,7 @@ async def test_interactive_subagent_exhaustion_tells_its_parent( notice = str(items[0]) assert "child" in notice assert "parked" in notice + root_session.close() @pytest.mark.asyncio diff --git a/tests/test_wait_for_user.py b/tests/test_wait_for_user.py index 29881f57..90b1082b 100644 --- a/tests/test_wait_for_user.py +++ b/tests/test_wait_for_user.py @@ -6,6 +6,7 @@ import json from typing import Any import pytest +from agents.memory import SQLiteSession from agents.tool_context import ToolContext from strix.core.agents import AgentCoordinator @@ -63,9 +64,11 @@ async def test_rejected_in_an_autonomous_run() -> None: @pytest.mark.asyncio -async def test_a_message_that_already_arrived_is_taken_instead_of_parking() -> None: +async def test_a_message_that_already_arrived_is_taken_instead_of_parking(tmp_path: Any) -> None: context = await _context(interactive=True) coordinator = context["coordinator"] + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) await coordinator.send("root", {"from": "user", "content": "wait, one more thing"}) result = await _call(context) @@ -73,6 +76,7 @@ async def test_a_message_that_already_arrived_is_taken_instead_of_parking() -> N assert result["wait_outcome"] == "message_arrived" assert result["pending_messages"] == 1 assert coordinator.statuses["root"] == "running" + session.close() @pytest.mark.asyncio