fix: end /next drains that re-offer an already-skipped message

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,14 +1268,23 @@ 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",
                         self.room_id,
                         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)

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
 
 

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