From ee31878773761f73b563a3e452faf81769b04890 Mon Sep 17 00:00:00 2001 From: STiFLeR7 Date: Fri, 11 Sep 2026 10:38:10 +0530 Subject: [PATCH] fix: report zero delivered when consume_pending has no session to write to The no-session branch already logged that queued messages were not persisted, but still returned a positive count/items, reporting them as delivered to every caller. Return 0, [] instead, matching what the log already says happened. Flagged in review on this PR (itzzdev09): https://github.com/usestrix/strix/pull/1161 Also adds the regression test that review flagged as worth keeping: mailbox[0:0] = queued (not append) is what keeps a message that arrives during a failed write's restore sorted after the one being retried. Two existing tests relied on root having no session attached while still expecting a positive pending count from consume_pending; that only worked because of the bug above. Gave both a real session, matching how every other agent in this test suite is set up. --- strix/core/agents.py | 34 ++++++++++++------------ tests/test_execution.py | 53 +++++++++++++++++++++++++++++++++++++ tests/test_wait_for_user.py | 6 ++++- 3 files changed, 75 insertions(+), 18 deletions(-) diff --git a/strix/core/agents.py b/strix/core/agents.py index f709304f..a1c82e75 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -501,24 +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; " - "restoring them to the mailbox for retry", - 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 ) - 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, [] + runtime.wake.set() + return 0, [] await self._maybe_snapshot() if not include_items: return count, [] diff --git a/tests/test_execution.py b/tests/test_execution.py index 40968fef..1913ad25 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -760,6 +760,55 @@ async def test_consume_pending_restores_mailbox_on_session_write_failure( 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() @@ -1136,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. @@ -1147,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, @@ -1162,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