mirror of
https://github.com/usestrix/strix.git
synced 2026-10-09 03:18:31 +00:00
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.
This commit is contained in:
parent
f4ef8867f6
commit
df9c214a6f
2 changed files with 46 additions and 1 deletions
|
|
@ -350,10 +350,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, []
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue