fix: restore mailbox on consume_pending session write failure

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, []

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