fix(sdk): recover backlog past exhausted failed messages

docs/rest-client.md

@@ -63,3 +63,42 @@ The generated client's empty-body shortcut runs before status classification in
 pinned `band-client-rest` 0.0.41 and remains in released 0.0.46. Public raw response
 status preserves the distinction without accessing generated private fields or
 adding another HTTP client. Real-client HTTP transport tests pin this boundary.
+
+## Retained failed messages and recovery
+
+The platform's `/next` endpoint selects the oldest delivery that is not processed,
+including failed and processing deliveries. Exhausting the SDK's local retry budget
+does not remove that message from the platform. The SDK preserves its failed status
+and diagnostics rather than acknowledging unsuccessful work as processed.
+
+When a locally completed, exhausted, self-authored or server-processed head prevents
+`/next` from advancing, startup and idle recovery use
+`BandLink.get_actionable_messages(room_id)` to collect one oldest-first snapshot.
+Pagination finishes before any delivery status changes. Message IDs are deduplicated
+without reordering the server's stream, including messages without timestamps.
+The listing is unfiltered: separate pending and processing scans could miss an older
+delivery whose status changes between requests. Cursor continuation takes precedence
+over legacy page counts; legacy page metadata is used only before cursor mode starts.
+An incomplete listing, transport failure or unusable cursor is a recovery failure,
+not proof that the room is drained.
+
+Recovery retains the current `/next` selection boundary. A 204 response never falls
+back to a processing-status listing, which cannot establish whether the room is
+stopped. Normal `/next` recovery already includes stale processing work in platform
+order. Snapshot recovery likewise includes pending, retryable failed and processing
+deliveries together, without a preliminary filtered sweep.
+
+An eligible delivery that is in flight, refused by the platform, awaiting its
+processed acknowledgement or still retryable retains its FIFO position and blocks
+newer work. `InputNotAccepted` also retains that position without consuming the
+failure budget. A successful local execution whose acknowledgement needs retrying
+is never executed again.
+
+Startup hands control back to the WebSocket queue at its synchronization marker only
+once that delivery is terminal. A newer STOP, PLAY or INTERRUPT invalidates a snapshot
+read across the control boundary, including an empty snapshot; no stale row is claimed
+or executed. Claims and context hydration keep the runtime's existing control-revision
+fences. Completing startup does not disable later reconnect synchronization.
+
+An idle snapshot with no new work behind a retained head reports a blocked pass, not
+a confirmed empty room, so it cannot grow the empty-room polling backoff.

docs/websocket-events.md

@@ -45,6 +45,11 @@ Newer STOP, PLAY or INTERRUPT controls invalidate obsolete recovery results. A 2
 response preserves the local pause because it cannot distinguish stopped from
 empty rooms.
 
+Local events have no REST delivery to replay. If a local turn is deferred, its
+queue position is retained while due stopped-room probes can discover a missed
+PLAY. These probes do not dispatch newer REST work before the local event is
+accepted; newer queued events remain behind it.
+
 While working-state reporting is active, its existing activity keep-alive also
 recovers a missed STOP: the platform answers a `working: true` report with
 `working: false` when the room execution is stopped. The runtime cancels the

src/band/platform/link.py

@@ -605,6 +605,9 @@ async def get_actionable_page(
             self.rest, room_id, cursor=cursor
         )
 
+    async def get_actionable_messages(self, room_id: str) -> list[PlatformMessage]:
+        return await self._messages.get_actionable_messages(self.rest, room_id)
+
     async def get_actionable_message(
         self, room_id: str, message_id: str
     ) -> PlatformMessage | None:

src/band/platform/message_lifecycle.py

@@ -273,6 +273,47 @@ async def get_actionable_page(
             [_platform_message(item, room_id) for item in response.data], next_cursor
         )
 
+    async def get_actionable_messages(
+        self, rest: AsyncRestClient, room_id: str
+    ) -> list[PlatformMessage]:
+        """Collect a complete, deduplicated snapshot before delivery statuses change."""
+        messages: dict[str, PlatformMessage] = {}
+        page = 1
+        pagination: dict[str, Any] = {}
+        seen_cursors: set[str] = set()
+        while True:
+            response = await rest.agent_api_messages.list_agent_messages(
+                chat_id=room_id,
+                sort_order="asc",
+                limit=ACTIONABLE_PAGE_SIZE,
+                request_options=DEFAULT_REQUEST_OPTIONS,
+                **pagination,
+            )
+            for item in response.data:
+                messages.setdefault(item.id, _platform_message(item, room_id))
+
+            cursor = response.metadata.next_cursor
+            if response.metadata.has_more is True and cursor:
+                if cursor in seen_cursors:
+                    raise ValueError("Message listing has no usable pagination cursor")
+                seen_cursors.add(cursor)
+                pagination = {"cursor": cursor}
+                continue
+
+            total_pages = response.metadata.total_pages
+            if (
+                "cursor" not in pagination
+                and total_pages is not None
+                and page < total_pages
+            ):
+                page += 1
+                pagination = {"page": page}
+                continue
+
+            if response.metadata.has_more is True:
+                raise ValueError("Message listing has no usable pagination cursor")
+            return list(messages.values())
+
     async def get_actionable_message(
         self, rest: AsyncRestClient, room_id: str, message_id: str
     ) -> PlatformMessage | None:

src/band/runtime/execution.py

@@ -1266,6 +1266,22 @@ def _set_deferred_head(self, message_id: str | None) -> None:
             self._deferred_message_id = message_id
             self._control_cursor = None
 
+    def _release_settled_deferral(self) -> None:
+        """Drop the deferral once its row is settled locally.
+
+        A row settles in more places than the recovery loops see: a stop or
+        control fence can return right after its turn, and an ack retry can
+        complete it before recovery starts. A settled row is not listed again,
+        so a deferral left on it would hold queued synthetic turns forever;
+        every pass and every processed row re-check it here.
+        """
+        deferred = self._deferred_message_id
+        if deferred is not None and (
+            self.claims.is_completed(self.room_id, deferred)
+            or self._retry_tracker.is_permanently_failed(deferred)
+        ):
+            self._set_deferred_head(None)
+
     def _defer_unaccepted_input(self, message_id: str | None) -> None:
         if message_id:
             self._retry_tracker.discard_attempt(message_id)
@@ -1457,28 +1473,20 @@ async def _synchronize_with_next(self) -> bool:
         """
         Synchronize backlog via /next API until caught up with WebSocket.
 
-        /next includes unfinished processing messages and gates stopped rooms.
-        Process in platform order until empty or the first queued WebSocket
-        message is reached; remove its queued duplicate before continuing.
+        /next selects unfinished work and gates stopped rooms. A locally
+        skipped head hands recovery to one complete chronological snapshot.
+        Stop at the first queued WebSocket message once its delivery is terminal.
         """
         logger.debug(
             "ExecutionContext %s: Starting /next synchronization", self.room_id
         )
 
-        previous_id: str | None = None
         try:
             while True:
                 revision = self._control_revision
                 next_msg = await self._get_next_message()
                 if revision != self._control_revision:
                     return self._stopped
-                if (
-                    next_msg is not None
-                    and next_msg.id == previous_id
-                    and self._backlog_is_nonrunnable(next_msg)
-                ):
-                    break
-                previous_id = next_msg.id if next_msg is not None else None
 
                 if next_msg is None:
                     logger.debug(
@@ -1488,13 +1496,17 @@ async def _synchronize_with_next(self) -> bool:
                     self._sync_complete = True
                     return True
 
-                if self._retry_tracker.is_permanently_failed(next_msg.id):
-                    logger.warning(
-                        "ExecutionContext %s: Skipping permanently failed message %s",
-                        self.room_id,
-                        next_msg.id,
+                if self._backlog_is_nonrunnable(next_msg):
+                    await self._process_backlog_message(next_msg, revision=revision)
+                    if revision != self._control_revision:
+                        return self._stopped
+                    outcome = await self._recover_actionable_backlog(
+                        next_msg, sync_to_websocket=True
                     )
-                    break
+                    synchronized = outcome in {ResyncOutcome.EMPTY, ResyncOutcome.WORK}
+                    if synchronized:
+                        self._sync_complete = True
+                    return synchronized
 
                 at_sync_point = next_msg.id == self._first_ws_msg_id
                 level = logging.INFO if at_sync_point else logging.DEBUG
@@ -1507,15 +1519,21 @@ async def _synchronize_with_next(self) -> bool:
                 result = await self._process_backlog_message(
                     next_msg, revision=revision
                 )
+                self._release_settled_deferral()
                 if result in DEFERRED_OUTCOMES:
                     self._set_deferred_head(next_msg.id)
                     return False
-                if at_sync_point:
+                if revision != self._control_revision:
+                    return self._stopped
+                if at_sync_point and self._backlog_is_nonrunnable(next_msg):
                     # Preserve other queued events when removing the sync point.
                     self._drain_duplicate_from_queue(next_msg.id)
                     self._first_ws_msg_id = None
                     self._sync_complete = True
                     return True
+                if at_sync_point:
+                    self._set_deferred_head(next_msg.id)
+                    return False
 
                 if self._stopped:
                     # A stop control signal landed mid-cycle: the message was
@@ -1531,14 +1549,6 @@ async def _synchronize_with_next(self) -> bool:
                     )
                     break
 
-                if self._retry_tracker.is_permanently_failed(next_msg.id):
-                    logger.warning(
-                        "ExecutionContext %s: Message %s permanently failed",
-                        self.room_id,
-                        next_msg.id,
-                    )
-                    break
-
         except Exception:
             logger.exception("ExecutionContext %s: Sync error", self.room_id)
             return False
@@ -1547,6 +1557,69 @@ async def _synchronize_with_next(self) -> bool:
         self._sync_complete = True
         return True
 
+    async def _recover_actionable_backlog(
+        self, head: PlatformMessage, *, sync_to_websocket: bool = False
+    ) -> ResyncOutcome:
+        """Drain one complete FIFO snapshot without re-polling a retained head."""
+        revision = self._control_revision
+        messages = await self.link.get_actionable_messages(self.room_id)
+        if revision != self._control_revision:
+            return ResyncOutcome.BLOCKED
+        if head.id == self._deferred_message_id:
+            self._set_deferred_head(None)
+        self._release_settled_deferral()
+        if (
+            not sync_to_websocket
+            and self._deferred_message_id is None
+            and self._pending_synthetic
+        ):
+            return ResyncOutcome.BLOCKED
+        if sync_to_websocket and head.id == self._first_ws_msg_id:
+            # /next selected this terminal marker before the snapshot; later
+            # rows must not overtake queued events even if it is no longer listed.
+            self._drain_duplicate_from_queue(head.id)
+            self._first_ws_msg_id = None
+            self._sync_complete = True
+            return ResyncOutcome.EMPTY
+        for msg in messages:
+            if self._stopped:
+                return ResyncOutcome.STOPPED
+            if revision != self._control_revision:
+                return ResyncOutcome.BLOCKED
+            try:
+                result = await self._process_backlog_message(msg, revision=revision)
+            except Exception:
+                self._set_deferred_head(msg.id)
+                raise
+            self._release_settled_deferral()
+            if self._stopped:
+                return ResyncOutcome.STOPPED
+            if revision != self._control_revision:
+                return ResyncOutcome.BLOCKED
+            if result is TurnOutcome.NOT_ACCEPTED:
+                self._set_deferred_head(msg.id)
+                return ResyncOutcome.BLOCKED
+            if result is TurnOutcome.RETRY_LATER or not self._backlog_is_nonrunnable(
+                msg
+            ):
+                self._set_deferred_head(msg.id)
+                return ResyncOutcome.RETRY
+            if sync_to_websocket and msg.id == self._first_ws_msg_id:
+                self._drain_duplicate_from_queue(msg.id)
+                self._first_ws_msg_id = None
+                self._sync_complete = True
+                break
+            # Same handoff rule as on entry: synthetic turns wait behind an
+            # older deferred message, and bootstrap drains them only once
+            # synchronized, since its caller retries sync, not the queue.
+            if (
+                not sync_to_websocket
+                and self._deferred_message_id is None
+                and self._pending_synthetic
+            ):
+                return ResyncOutcome.BLOCKED
+        return ResyncOutcome.WORK if self._resync_handled_work else ResyncOutcome.EMPTY
+
     async def _get_next_message(self) -> PlatformMessage | None:
         """
         Get next unprocessed message from REST API.
@@ -1561,12 +1634,12 @@ async def _resync_pending_messages(
     ) -> ResyncOutcome:
         """Drain actionable work, distinguishing empty from blocked recovery."""
         self._resync_handled_work = False
+        self._release_settled_deferral()
         if (
             self._deferred_message_id is None
             and (outcome := await self._drain_synthetic()) is not None
         ):
             return outcome
-        previous_id: str | None = None
         try:
             while True:
                 revision = self._control_revision
@@ -1585,15 +1658,18 @@ async def _resync_pending_messages(
                         if self._resync_handled_work
                         else ResyncOutcome.EMPTY
                     )
-                if (
-                    next_msg.id == previous_id
-                    and self._backlog_is_nonrunnable(next_msg)
-                ) or self._retry_tracker.is_permanently_failed(next_msg.id):
-                    return ResyncOutcome.BLOCKED
-                previous_id = next_msg.id
+                if self._backlog_is_nonrunnable(next_msg):
+                    outcome = await self._recover_actionable_backlog(next_msg)
+                    # A retained /next head is not positive evidence of an empty room.
+                    return (
+                        ResyncOutcome.BLOCKED
+                        if outcome is ResyncOutcome.EMPTY
+                        else outcome
+                    )
                 result = await self._process_backlog_message(
                     next_msg, revision=revision
                 )
+                self._release_settled_deferral()
                 if self._stopped:
                     return ResyncOutcome.STOPPED
                 if revision != self._control_revision:
@@ -1620,6 +1696,7 @@ def _backlog_is_nonrunnable(self, msg: PlatformMessage) -> bool:
         return (
             self.claims.is_completed(self.room_id, msg.id)
             or self._retry_tracker.is_permanently_failed(msg.id)
+            or self._delivery_status_for_agent(msg.metadata) == DeliveryStatus.PROCESSED
             or bool(
                 self._agent_id
                 and is_self_echo(

tests/runtime/agentruntime/conftest.py

@@ -35,6 +35,7 @@ def mock_link():
     # Message lifecycle methods
     link.get_next_message = AsyncMock(return_value=None)
     link.get_stale_processing_messages = AsyncMock(return_value=[])
+    link.get_actionable_messages = AsyncMock(return_value=[])
     link.mark_processing = AsyncMock()
     link.mark_processed = AsyncMock()
     link.mark_failed = AsyncMock()

tests/runtime/conftest.py

@@ -257,5 +257,6 @@ def link(mock_rest_client: MagicMock) -> MagicMock:
     link.mark_failed = AsyncMock(return_value=True)
     link.get_next_message = AsyncMock(return_value=None)
     link.get_stale_processing_messages = AsyncMock(return_value=[])
+    link.get_actionable_messages = AsyncMock(return_value=[])
     link.report_activity = AsyncMock(return_value=True)
     return link

tests/runtime/executioninterrupt/conftest.py

@@ -25,5 +25,6 @@ def mock_link():
     link.mark_failed = AsyncMock(return_value=True)
     link.get_next_message = AsyncMock(return_value=None)
     link.get_stale_processing_messages = AsyncMock(return_value=[])
+    link.get_actionable_messages = AsyncMock(return_value=[])
     link.report_activity = AsyncMock(return_value=True)
     return link