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()