Merge 7bb32c29be into e21f9ce4cc

strix/core/agents.py

@@ -501,16 +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",
-                        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
                     )
+                    runtime.wake.set()
+                return 0, []
         await self._maybe_snapshot()
         if not include_items:
             return count, []

tests/test_agent_graph_coordination.py

@@ -13,6 +13,7 @@ import json
 from typing import TYPE_CHECKING, Any, cast
 
 import pytest
+from agents.memory import SQLiteSession
 from agents.tool_context import ToolContext
 
 from strix.core.agents import AgentCoordinator
@@ -162,8 +163,12 @@ async def test_wait_returns_at_once_when_no_child_can_answer() -> None:
 
 
 @pytest.mark.asyncio
-async def test_wait_delivers_a_pending_report_before_checking_liveness() -> None:
+async def test_wait_delivers_a_pending_report_before_checking_liveness(
+    tmp_path: Path,
+) -> None:
     coordinator = await _graph(interactive=False)
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
     await coordinator.send("root", {"from": "child", "type": "completion", "content": "done"})
     await coordinator.set_status("child", "completed")
 

tests/test_execution.py

@@ -723,6 +723,92 @@ 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_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()
@@ -1099,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.
@@ -1110,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,
@@ -1125,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

tests/test_wait_for_user.py

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