From 6f5df16f56cd2305d76fc2eaed0e98fde8404a52 Mon Sep 17 00:00:00 2001 From: Natan Zalkin Date: Tue, 6 Oct 2026 20:05:39 +0300 Subject: [PATCH] fix: end /next drains that re-offer an already-skipped message The /next catch-up loops polled until the endpoint yielded nothing. A message skipped from local knowledge alone -- completed, permanently failed, or the agent's own -- never changes its platform state, so once the processed-ack retry budget exhausted and recorded local completion without a durable mark, /next kept returning that same oldest message: the drain spun on it and newer REST-only backlog starved behind it. Classify those skips as BacklogProcessResult.NO_PROGRESS and end a drain that re-receives a message it already skipped in the same pass, the way the stop and permanently-failed guards already end a drain. Retryable outcomes -- a handler failure with attempt budget left, a pending processed ack, an unacquired claim -- keep their in-drain retry semantics unchanged. --- src/band/runtime/execution.py | 43 ++++++++++++++++++-- tests/runtime/test_execution.py | 4 +- tests/runtime/test_resync.py | 70 +++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 6 deletions(-) diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 8359e7260..d6998eef0 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -102,6 +102,11 @@ class ResyncRequest: class BacklogProcessResult(Enum): ADVANCED = "advanced" RETRY_LATER = "retry_later" + # Skipped from local knowledge (completed, permanently failed, or the + # agent's own message); the platform's copy of the message is unchanged, + # so /next will offer it again while it stays the oldest not-processed + # one. + NO_PROGRESS = "no_progress" class ExecutionState(StrEnum): @@ -1240,6 +1245,10 @@ async def _synchronize_with_next(self) -> bool: ) try: + # A message skipped from local knowledge alone changes nothing + # server-side; if /next offers it again the drain has stopped + # making durable progress and must not spin on it. + skipped: set[str] = set() while True: # Cancellation handles exit next_msg = await self._get_next_message() @@ -1259,6 +1268,15 @@ async def _synchronize_with_next(self) -> bool: ) break + if next_msg.id in skipped: + logger.warning( + "ExecutionContext %s: /next re-returned message %s " + "already skipped in this drain; ending sync", + self.room_id, + next_msg.id, + ) + break + if next_msg.id == self._first_ws_msg_id: logger.info( "ExecutionContext %s: Sync point reached at message %s", @@ -1266,7 +1284,7 @@ async def _synchronize_with_next(self) -> bool: next_msg.id, ) result = await self._process_backlog_message(next_msg) - if result == BacklogProcessResult.ADVANCED: + if result != BacklogProcessResult.RETRY_LATER: # Remove all WS copies of the sync-point message while # preserving the relative order of other queued events. self._drain_duplicate_from_queue(next_msg.id) @@ -1283,6 +1301,8 @@ async def _synchronize_with_next(self) -> bool: result = await self._process_backlog_message(next_msg) if result == BacklogProcessResult.RETRY_LATER: return False + if result == BacklogProcessResult.NO_PROGRESS: + skipped.add(next_msg.id) if self._stopped: # A stop control signal landed mid-cycle: the message was @@ -1336,6 +1356,7 @@ async def _resync_pending_messages(self) -> bool: "ExecutionContext %s: Re-polling /next for missed messages", self.room_id ) caught_up = 0 + skipped: set[str] = set() try: while True: next_msg = await self._get_next_message() @@ -1350,6 +1371,18 @@ async def _resync_pending_messages(self) -> bool: ) break + if next_msg.id in skipped: + # The no-spin invariant of _synchronize_with_next: /next + # keeps offering a message skipped from local knowledge + # alone until the platform retires it. + logger.warning( + "ExecutionContext %s: /next re-returned message %s " + "already skipped in this drain; ending resync", + self.room_id, + next_msg.id, + ) + break + logger.info( "ExecutionContext %s: Catching up missed message %s via /next resync", self.room_id, @@ -1358,6 +1391,8 @@ async def _resync_pending_messages(self) -> bool: result = await self._process_backlog_message(next_msg) if result == BacklogProcessResult.RETRY_LATER: return False + if result == BacklogProcessResult.NO_PROGRESS: + skipped.add(next_msg.id) caught_up += 1 @@ -1429,17 +1464,17 @@ async def _process_backlog_message( agent_id=self._agent_id, ): logger.debug("Skipping self-message %s", msg_id) - return BacklogProcessResult.ADVANCED + return BacklogProcessResult.NO_PROGRESS # Skip permanently failed messages if self._retry_tracker.is_permanently_failed(msg_id): logger.debug("Skipping permanently failed message %s", msg_id) - return BacklogProcessResult.ADVANCED + return BacklogProcessResult.NO_PROGRESS # Skip if already processed (dedupe) if self.claims.is_completed(self.room_id, msg_id): logger.debug("Skipping duplicate backlog message: %s", msg_id) - return BacklogProcessResult.ADVANCED + return BacklogProcessResult.NO_PROGRESS if self.claims.is_ack_pending(self.room_id, msg_id): logger.debug("Retrying processed ack for backlog message: %s", msg_id) diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index ad05652c7..a6de17f60 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -985,7 +985,7 @@ async def test_backlog_skips_self_authored_message( result = await ctx._process_backlog_message(msg) - assert result == BacklogProcessResult.ADVANCED + assert result == BacklogProcessResult.NO_PROGRESS mock_handler.assert_not_awaited() mock_link_with_next.mark_processing.assert_not_awaited() mock_link_with_next.mark_processed.assert_not_awaited() @@ -1881,7 +1881,7 @@ async def test_retry_saturation_skips_handler_on_next_delivery( # A further delivery of the same message must not invoke the handler. result = await ctx._process_backlog_message(msg) - assert result == BacklogProcessResult.ADVANCED + assert result == BacklogProcessResult.NO_PROGRESS assert failing_handler.await_count == 1 diff --git a/tests/runtime/test_resync.py b/tests/runtime/test_resync.py index 4f2f7c180..20aa44105 100644 --- a/tests/runtime/test_resync.py +++ b/tests/runtime/test_resync.py @@ -253,6 +253,76 @@ async def test_exception_does_not_propagate(self, mock_link, mock_handler): await ctx._resync_pending_messages() +# --------------------------------------------------------------------------- +# TestDrainProgressGuard +# --------------------------------------------------------------------------- + + +class RepeatingNextMessage: + """A ``/next`` stub that keeps returning the same message — what the + endpoint does while that message stays the oldest not-processed one.""" + + def __init__(self, msg: PlatformMessage) -> None: + self.msg = msg + self.calls = 0 + + async def __call__(self, room_id: str) -> PlatformMessage: + self.calls += 1 + if self.calls > 10: + raise RuntimeError("/next drain did not terminate") + return self.msg + + +class TestDrainProgressGuard: + """Tests for /next drains refusing to spin on a message they cannot advance.""" + + async def test_resync_ends_when_next_repeats_a_skipped_message( + self, mock_link, mock_handler + ): + msg = make_platform_message(msg_id="wedged-1", room_id="room-1") + next_message = RepeatingNextMessage(msg) + mock_link.get_next_message.side_effect = next_message.__call__ + + ctx = ExecutionContext("room-1", mock_link, mock_handler) + ctx.claims.remember_completed(ctx.room_id, "wedged-1") + + assert await ctx._resync_pending_messages() is True + assert next_message.calls == 2 + + async def test_startup_sync_ends_when_next_repeats_a_skipped_message( + self, mock_link, mock_handler + ): + msg = make_platform_message(msg_id="wedged-1", room_id="room-1") + next_message = RepeatingNextMessage(msg) + mock_link.get_next_message.side_effect = next_message.__call__ + + ctx = ExecutionContext("room-1", mock_link, mock_handler) + ctx.claims.remember_completed(ctx.room_id, "wedged-1") + + assert await ctx._synchronize_with_next() is True + assert next_message.calls == 2 + + async def test_startup_sync_retries_failed_message_in_same_drain(self, mock_link): + # A turn that fails with retry budget left stays actionable, so /next + # offers it again in the same drain and the retry must run there. + calls = [] + + async def handler(_ctx: object, _event: object) -> None: + calls.append(1) + if len(calls) == 1: + raise RuntimeError("transient turn failure") + + msg = make_platform_message(msg_id="retry-1", room_id="room-1") + mock_link.get_next_message.side_effect = [msg, msg, None] + + ctx = ExecutionContext( + "room-1", mock_link, handler, config=SessionConfig(max_message_retries=2) + ) + + assert await ctx._synchronize_with_next() is True + assert len(calls) == 2 + + # --------------------------------------------------------------------------- # TestAgentRuntimeOnReconnected # ---------------------------------------------------------------------------