From df9c214a6fa199db0123f00b4b6b0225a172490e Mon Sep 17 00:00:00 2001 From: STiFLeR7 Date: Tue, 25 Aug 2026 12:16:01 +0530 Subject: [PATCH] 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 c96204dfe..2b673ca05 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -350,10 +350,18 @@ async def consume_pending( 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 d389bde3f..08093815f 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -657,6 +657,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()