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.
This commit is contained in:
STiFLeR7 2026-09-11 10:38:10 +05:30
parent 289e82f64e
commit ee31878773
No known key found for this signature in database
3 changed files with 75 additions and 18 deletions

View file

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

View file

@ -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

View file

@ -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