From 289e82f64e789ab4cc5a6a7538339837750f4c1a Mon Sep 17 00:00:00 2001 From: STiFLeR7 Date: Tue, 25 Aug 2026 12:16:01 +0530 Subject: [PATCH 1/3] fix: restore mailbox on consume_pending session write failure consume_pending() drains the agent's mailbox and zeroes pending_counts under the lock, then - outside the lock - tries to persist the drained items via session.add_items(). On failure it only logged the exception and fell through to returning the original (now-inflated) count as if the write had succeeded. Callers with include_items=False (execution.py after a wait, respond/tool.py) rely entirely on the session for the next turn, so a transient write failure (e.g. a locked SQLite session) silently dropped the user's message: not in the mailbox (already cleared), not in the session (write failed), and reported as delivered to the caller. On write failure, restore the drained messages to the front of the mailbox and add their count back to pending_counts, wake the runtime so a blocked wait_for_message() notices the pending work again, and return (0, []) so this attempt is honestly reported as not having delivered anything. This makes the next consume_pending() call retry the same messages instead of losing them. Fixes #1107. --- strix/core/agents.py | 10 +++++++++- tests/test_execution.py | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/strix/core/agents.py b/strix/core/agents.py index 7e738be1..f709304f 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -507,10 +507,18 @@ class AgentCoordinator: await session.add_items(items) except Exception: logger.exception( - "failed to append %d queued messages to the session of %s", + "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_execution.py b/tests/test_execution.py index 0daf0da2..40968fef 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -723,6 +723,43 @@ 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_error_parked_agent_only_released_by_user_message(tmp_path: Any) -> None: coordinator = AgentCoordinator() From ee31878773761f73b563a3e452faf81769b04890 Mon Sep 17 00:00:00 2001 From: STiFLeR7 Date: Fri, 11 Sep 2026 10:38:10 +0530 Subject: [PATCH 2/3] 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 From 7bb32c29be9b2bcf974d27f7fa0037c1af45f981 Mon Sep 17 00:00:00 2001 From: STiFLeR7 Date: Fri, 9 Oct 2026 09:37:27 +0530 Subject: [PATCH 3/3] fix(tests): give the root agent a real session in the pending-report wait test test_wait_delivers_a_pending_report_before_checking_liveness asserted message_arrived against a root agent with no attached session, which only passed because consume_pending used to misreport delivery on a failed write. Now that it correctly returns (0, []) when there's no session to persist to, give the test a real SQLiteSession, matching every other agent-with-a-session test in this suite. Co-Authored-By: Claude Sonnet 5 --- tests/test_agent_graph_coordination.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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")