diff --git a/docs/rest-client.md b/docs/rest-client.md index bbee8bc70..ea670582d 100644 --- a/docs/rest-client.md +++ b/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. diff --git a/docs/websocket-events.md b/docs/websocket-events.md index f233da3fb..b81e2f135 100644 --- a/docs/websocket-events.md +++ b/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 diff --git a/src/band/platform/link.py b/src/band/platform/link.py index 524c3cd07..b82e7b7d2 100644 --- a/src/band/platform/link.py +++ b/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: diff --git a/src/band/platform/message_lifecycle.py b/src/band/platform/message_lifecycle.py index 2c6d1c8df..b5c34fa40 100644 --- a/src/band/platform/message_lifecycle.py +++ b/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: diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 6948d1133..e102e964e 100644 --- a/src/band/runtime/execution.py +++ b/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( diff --git a/tests/runtime/agentruntime/conftest.py b/tests/runtime/agentruntime/conftest.py index dc9c4aeba..936cc70ab 100644 --- a/tests/runtime/agentruntime/conftest.py +++ b/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() diff --git a/tests/runtime/conftest.py b/tests/runtime/conftest.py index 46f8a06ed..8ebd6dc5e 100644 --- a/tests/runtime/conftest.py +++ b/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 diff --git a/tests/runtime/executioninterrupt/conftest.py b/tests/runtime/executioninterrupt/conftest.py index ca766fca6..d1e952605 100644 --- a/tests/runtime/executioninterrupt/conftest.py +++ b/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 diff --git a/tests/runtime/executioninterrupt/test_cycle_watchdog.py b/tests/runtime/executioninterrupt/test_cycle_watchdog.py index 24d0eeb75..03085aabb 100644 --- a/tests/runtime/executioninterrupt/test_cycle_watchdog.py +++ b/tests/runtime/executioninterrupt/test_cycle_watchdog.py @@ -5,6 +5,8 @@ import asyncio from unittest.mock import patch +import pytest + from band.runtime.execution import ExecutionContext, TurnOutcome from band.runtime.sessionconfig import SessionConfig from tests.conftest import BlockingHandler, make_message_event @@ -187,6 +189,7 @@ async def finish_at_deadline(ctx, event): mock_link.mark_processed.assert_awaited_once_with("room-123", "boundary") mock_link.mark_failed.assert_not_awaited() + @pytest.mark.looptime async def test_watchdog_does_not_wait_for_stuck_cancellation_cleanup( self, mock_link ): @@ -211,17 +214,19 @@ async def blocks_during_cleanup(ctx, event): config=SessionConfig(max_cycle_seconds=_WATCHDOG_TEST_DEADLINE), ) - with patch("band.core.turn_lifecycle.CYCLE_CANCEL_GRACE_SECONDS", 0.01): - result = await asyncio.wait_for( - ctx._process_event(make_message_event(msg_id="stuck-cleanup")), - timeout=0.2, - ) - - assert result is TurnOutcome.ADVANCED - assert cleanup_started.is_set() - mock_link.mark_processed.assert_not_awaited() - mock_link.mark_failed.assert_awaited_once() - assert "max_cycle_seconds" in mock_link.mark_failed.await_args.args[2] - - release_cleanup.set() - await asyncio.wait_for(cleanup_finished.wait(), timeout=0.2) + try: + with patch("band.core.turn_lifecycle.CYCLE_CANCEL_GRACE_SECONDS", 0.01): + result = await asyncio.wait_for( + ctx._process_event(make_message_event(msg_id="stuck-cleanup")), + timeout=0.2, + ) + + assert result is TurnOutcome.ADVANCED + assert cleanup_started.is_set() + assert not cleanup_finished.is_set() + mock_link.mark_processed.assert_not_awaited() + mock_link.mark_failed.assert_awaited_once() + assert "max_cycle_seconds" in mock_link.mark_failed.await_args.args[2] + finally: + release_cleanup.set() + await asyncio.wait_for(cleanup_finished.wait(), timeout=0.2) diff --git a/tests/runtime/resync/conftest.py b/tests/runtime/resync/conftest.py index d0bb19cc0..a57503cb3 100644 --- a/tests/runtime/resync/conftest.py +++ b/tests/runtime/resync/conftest.py @@ -19,7 +19,7 @@ ROOM_ID, rest_client_over, ) -from tests.runtime.resync.kit import ReconciledRoom, ReconciliationPeer +from tests.runtime.resync.kit import BacklogApi, ReconciledRoom, ReconciliationPeer @pytest.fixture @@ -55,6 +55,8 @@ def mock_link(): link.mark_failed = AsyncMock() 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) async def empty_aiter(): return @@ -101,3 +103,12 @@ async def handler(ctx: ExecutionContext, event: Any) -> None: yield ctx, peer, invoked finally: await ctx.stop() + + +@pytest_asyncio.fixture +async def backlog() -> AsyncIterator[tuple[BandLink, BacklogApi]]: + peer = BacklogApi() + async with rest_client_over(peer.respond) as rest: + link = BandLink(agent_id=AGENT_ID, api_key="test-key") + link.rest = rest + yield link, peer diff --git a/tests/runtime/resync/kit.py b/tests/runtime/resync/kit.py index 5d3be76db..856d08e24 100644 --- a/tests/runtime/resync/kit.py +++ b/tests/runtime/resync/kit.py @@ -3,13 +3,17 @@ from __future__ import annotations import asyncio -from datetime import UTC, datetime +import json +from datetime import UTC, datetime, timedelta +from typing import Any import httpx from band.runtime.execution import ExecutionContext from band.runtime.messages import PlatformMessage from tests.runtime.helpers import ( + AGENT_ID, + ROOM_ID, LifecyclePlatform, ) @@ -50,3 +54,151 @@ def make_platform_message( metadata={}, created_at=datetime(2024, 1, 1, tzinfo=UTC), ) + + +class BacklogApi: + """Stateful HTTP peer for ordered recovery through the real REST client.""" + + def __init__(self) -> None: + self.messages: dict[str, dict[str, Any]] = {} + self.statuses: dict[str, str] = {} + self.transitions: list[tuple[str, str]] = [] + self.next_calls = 0 + self.list_calls = 0 + self.listing_queries: list[dict[str, str]] = [] + self.fixed_head: str | None = None + self.listing_error: tuple[str, int] | None = None + self.refuse_processed: set[str] = set() + self.processing_response: httpx.Response | None = None + self.pagination = "page" + self.cursor_behavior = "normal" + self.reject_page_parameters = False + self.stopped = False + + def add(self, message_id: str, status: str, *, self_authored: bool = False) -> None: + created_at = datetime(2026, 1, 1, tzinfo=UTC) + timedelta( + seconds=len(self.messages) + ) + self.messages[message_id] = { + "id": message_id, + "chat_room_id": ROOM_ID, + "content": message_id, + "sender_id": AGENT_ID if self_authored else "user-1", + "sender_type": "Agent" if self_authored else "User", + "sender_name": "Agent" if self_authored else "User One", + "message_type": "text", + "attachments": [], + "inserted_at": created_at.isoformat(), + "updated_at": created_at.isoformat(), + } + self.statuses[message_id] = status + + def message(self, message_id: str) -> dict[str, Any]: + return { + **self.messages[message_id], + "metadata": { + "mentions": [], + "delivery_status": {AGENT_ID: {"status": self.statuses[message_id]}}, + }, + } + + def respond(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/participants"): + return httpx.Response(200, json={"data": []}) + if path.endswith("/activity"): + working = json.loads(request.content)["working"] + return httpx.Response( + 200, json={"data": {"working": working and not self.stopped}} + ) + if path.endswith("/messages/next"): + self.next_calls += 1 + if self.next_calls > 50: + raise AssertionError("Repeated /next head did not make progress") + if self.stopped: + return httpx.Response(204) + message_id = self.fixed_head or next( + (mid for mid, status in self.statuses.items() if status != "processed"), + None, + ) + if message_id is None: + return httpx.Response(204) + return httpx.Response(200, json={"data": self.message(message_id)}) + + if path.endswith("/messages") and request.method == "GET": + if self.reject_page_parameters and "page" in request.url.params: + return httpx.Response(400, json={"error": "page pagination retired"}) + self.list_calls += 1 + self.listing_queries.append(dict(request.url.params)) + if self.list_calls > 50: + raise AssertionError("Cursor listing did not make progress") + status = request.url.params.get("status", "") + page = int( + request.url.params.get("cursor") or request.url.params.get("page", "1") + ) + if self.listing_error == (status, page): + return httpx.Response(400, json={"error": "listing unavailable"}) + eligible = ( + {"new", "pending", "delivered", "failed", "processing"} + if not status + else {"pending", "delivered", "failed"} + if status == "pending" + else {status} + ) + message_ids = [ + mid for mid, current in self.statuses.items() if current in eligible + ] + total_pages = max(1, len(message_ids)) + next_cursor = None + if self.pagination in {"cursor", "mixed"} and page < total_pages: + next_cursor = str(page + 1) + if self.cursor_behavior == "missing": + next_cursor = None + elif self.cursor_behavior == "repeated": + next_cursor = "1" + return httpx.Response( + 200, + json={ + "data": [self.message(mid) for mid in message_ids[page - 1 : page]], + "metadata": { + "page": page if self.pagination == "page" else None, + "page_size": 1, + "total_count": len(message_ids), + "total_pages": ( + total_pages + if self.pagination in {"page", "mixed"} + else None + ), + "limit": 1, + "has_more": page < total_pages, + "next_cursor": next_cursor, + }, + }, + ) + + message_id, action = path.rsplit("/", 2)[-2:] + if request.method == "POST" and action in { + "processing", + "processed", + "failed", + }: + if self.stopped: + return httpx.Response(204) + if action == "processing" and self.processing_response is not None: + return self.processing_response + if action == "processed" and message_id in self.refuse_processed: + return httpx.Response(400, json={"error": "ack unavailable"}) + self.statuses[message_id] = action + self.transitions.append((message_id, action)) + return httpx.Response( + 200, + json={ + "data": { + "id": message_id, + "status": action, + "attempt_number": 1, + "success": True, + } + }, + ) + raise AssertionError(f"Unexpected HTTP request: {request.method} {path}") diff --git a/tests/runtime/resync/test_actionable_recovery.py b/tests/runtime/resync/test_actionable_recovery.py new file mode 100644 index 000000000..b976a6883 --- /dev/null +++ b/tests/runtime/resync/test_actionable_recovery.py @@ -0,0 +1,926 @@ +"""Complete actionable snapshots behind retained heads, with FIFO admission.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any + +import httpx +import pytest + +from band.client.streaming import ControlMode +from band.core.exceptions import InputNotAccepted +from band.platform.link import BandLink +from band.runtime.execution import ExecutionContext, ResyncOutcome +from band.runtime.sessionconfig import SessionConfig +from tests.conftest import make_message_event, make_participant_added_event +from tests.runtime.helpers import AGENT_ID, ROOM_ID, rest_client_over +from tests.runtime.resync.kit import BacklogApi + + +class TestActionableRecovery: + """A retained actionable head must not hide or overtake eligible work.""" + + @staticmethod + def context( + link: BandLink, handler: Callable[[ExecutionContext, Any], Awaitable[None]] + ) -> ExecutionContext: + return ExecutionContext( + ROOM_ID, + link, + handler, + agent_id=AGENT_ID, + config=SessionConfig( + enable_context_hydration=False, + enable_working_state=False, + report_turn_failures_to_room=False, + max_message_retries=2, + ), + ) + + @staticmethod + async def recover(ctx: ExecutionContext, phase: str) -> bool | ResyncOutcome: + async with asyncio.timeout(1): + if phase == "bootstrap": + return await ctx._synchronize_with_next() + return await ctx._resync_pending_messages() + + @pytest.mark.parametrize("trigger", ["exhausted", "stale"]) + async def test_snapshot_hands_off_at_websocket_marker( + self, backlog: tuple[BandLink, BacklogApi], trigger: str + ) -> None: + link, peer = backlog + if trigger == "exhausted": + peer.add("exhausted", "failed") + peer.add("A", "processing" if trigger == "stale" else "pending") + peer.add("B", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="A")) + await ctx.on_event( + make_participant_added_event(room_id=ROOM_ID, participant_id="new-user") + ) + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="B")) + + assert await self.recover(ctx, "bootstrap") is True + while not ctx.queue.empty(): + await ctx._process_event(ctx.queue.get_nowait()) + + assert seen == ["A", "new-user", "B"] + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("mode", list(ControlMode)) + @pytest.mark.parametrize("snapshot", ["rows", "empty"]) + async def test_control_during_snapshot_invalidates_even_empty_response( + self, + backlog: tuple[BandLink, BacklogApi], + phase: str, + mode: ControlMode, + snapshot: str, + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("older", "pending") + peer.add("stale", "processing") + entered = asyncio.Event() + release = asyncio.Event() + seen: list[str] = [] + + async def delayed_listing(request: httpx.Request) -> httpx.Response: + response = peer.respond(request) + if request.url.path.endswith("/messages"): + if snapshot == "empty": + response = httpx.Response( + 200, + json={ + "data": [], + "metadata": {"has_more": False, "limit": 100}, + }, + ) + entered.set() + await release.wait() + return response + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + async with rest_client_over(delayed_listing) as rest: + link.rest = rest + recovery = asyncio.create_task(self.recover(ctx, phase)) + try: + async with asyncio.timeout(1): + await entered.wait() + if mode is ControlMode.PLAY: + await ctx.resume_room() + elif mode is ControlMode.STOP: + ctx.stop_room() + else: + ctx.interrupt(kind=mode) + release.set() + result = await recovery + assert result is ( + False if phase == "bootstrap" else ResyncOutcome.BLOCKED + ) + assert seen == [] + assert peer.transitions == [] + finally: + release.set() + recovery.cancel() + await asyncio.gather(recovery, return_exceptions=True) + + async with rest_client_over(peer.respond) as rest: + link.rest = rest + if ctx.is_stopped: + await ctx.resume_room() + assert await self.recover(ctx, "idle") is ResyncOutcome.WORK + assert seen == ["older", "stale"] + assert peer.statuses == { + "exhausted": "failed", + "older": "processed", + "stale": "processed", + } + + async def test_server_order_survives_missing_timestamp( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("older", "pending") + peer.add("newer", "pending") + peer.messages["older"]["inserted_at"] = None + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + + assert await self.recover(ctx, "bootstrap") is True + assert seen == ["older", "newer"] + + async def test_delivery_status_change_cannot_hide_older_work( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("older", "processing") + peer.add("newer", "pending") + seen: list[str] = [] + + def changing_listing(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/messages") and "page" in request.url.params: + peer.statuses["older"] = "failed" + return peer.respond(request) + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + async with rest_client_over(changing_listing) as rest: + link.rest = rest + assert await self.recover(ctx, "bootstrap") is True + assert seen == ["older", "newer"] + assert peer.statuses == { + "exhausted": "failed", + "older": "processed", + "newer": "processed", + } + assert all("status" not in query for query in peer.listing_queries) + + async def test_completed_bootstrap_does_not_disable_later_synchronization( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + assert await self.recover(ctx, "bootstrap") is True + peer.add("missed-during-disconnect", "processing") + assert await self.recover(ctx, "bootstrap") is True + assert seen == ["missed-during-disconnect"] + assert peer.statuses["missed-during-disconnect"] == "processed" + assert peer.listing_queries == [] + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("pagination", ["page", "cursor", "retired-page"]) + async def test_exhausted_head_preserves_failure_and_recovers_paginated_fifo( + self, backlog: tuple[BandLink, BacklogApi], phase: str, pagination: str + ) -> None: + link, peer = backlog + peer.pagination = "mixed" if pagination == "retired-page" else pagination + peer.reject_page_parameters = pagination == "retired-page" + peer.add("exhausted", "failed") + peer.add("old-pending", "pending") + peer.add("retryable-failed", "failed") + peer.add("stale-processing", "processing") + peer.add("new-pending", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == [ + "old-pending", + "retryable-failed", + "stale-processing", + "new-pending", + ] + assert peer.statuses["exhausted"] == "failed" + assert all(mid != "exhausted" for mid, _ in peer.transitions) + assert ctx._retry_tracker.is_permanently_failed("exhausted") + assert not ctx.claims.is_completed(ctx.room_id, "exhausted") + assert all("status" not in query for query in peer.listing_queries) + assert all(query["sort_order"] == "asc" for query in peer.listing_queries) + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.BLOCKED + ) + await ctx._process_event( + make_message_event(room_id=ROOM_ID, msg_id="new-pending") + ) + assert len(seen) == 4 + assert peer.next_calls == 2 + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + async def test_head_exhausted_during_recovery_does_not_hide_later_message( + self, backlog: tuple[BandLink, BacklogApi], phase: str + ) -> None: + link, peer = backlog + peer.add("failing-head", "pending") + peer.add("later", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + if event.payload.id == "failing-head": + raise RuntimeError("permanent handler failure") + + ctx = self.context(link, handle) + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["failing-head", "failing-head", "later"] + assert peer.statuses == {"failing-head": "failed", "later": "processed"} + assert ctx._retry_tracker.is_permanently_failed("failing-head") + assert ("failing-head", "processed") not in peer.transitions + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("skip", ["completed", "self-authored", "server-processed"]) + async def test_locally_skipped_repeated_head_never_reexecutes( + self, backlog: tuple[BandLink, BacklogApi], phase: str, skip: str + ) -> None: + link, peer = backlog + peer.add( + "skipped", + "processed" if skip == "server-processed" else "pending", + self_authored=skip == "self-authored", + ) + peer.add("later", "pending") + peer.fixed_head = "skipped" + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + if skip == "completed": + ctx.claims.remember_completed(ctx.room_id, "skipped") + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["later"] + assert all(mid != "skipped" for mid, _ in peer.transitions) + assert peer.next_calls == 1 + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize( + "skip", ["completed", "self-authored", "exhausted", "server-processed"] + ) + async def test_skipped_only_head_finishes_bounded_pass( + self, backlog: tuple[BandLink, BacklogApi], phase: str, skip: str + ) -> None: + link, peer = backlog + peer.add( + "skipped", + "processed" if skip == "server-processed" else "pending", + self_authored=skip == "self-authored", + ) + peer.fixed_head = "skipped" + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + if skip == "completed": + ctx.claims.remember_completed(ctx.room_id, "skipped") + elif skip == "exhausted": + ctx._retry_tracker.mark_permanently_failed("skipped") + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.BLOCKED + ) + assert seen == [] + assert peer.transitions == [] + assert peer.next_calls == 1 + assert peer.list_calls == 1 + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + async def test_inflight_eligible_snapshot_head_blocks_newer_work( + self, backlog: tuple[BandLink, BacklogApi], phase: str + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("owned", "pending") + peer.add("newer", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + assert ctx.claims.try_claim(ctx.room_id, "owned") + try: + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.RETRY + ) + assert seen == [] + assert peer.transitions == [] + assert ctx._sync_complete is False + finally: + ctx.claims.release(ctx.room_id, "owned") + + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["owned", "newer"] + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("failed_page", [1, 2]) + async def test_incomplete_snapshot_never_executes_partial_rows( + self, backlog: tuple[BandLink, BacklogApi], phase: str, failed_page: int + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("older", "pending") + peer.add("newer", "pending") + peer.listing_error = ("", failed_page) + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.RETRY + ) + assert seen == [] + assert peer.transitions == [] + assert peer.statuses["exhausted"] == "failed" + + peer.listing_error = None + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["older", "newer"] + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("cursor_behavior", ["missing", "repeated"]) + async def test_unusable_cursor_continuation_is_not_drained( + self, backlog: tuple[BandLink, BacklogApi], phase: str, cursor_behavior: str + ) -> None: + link, peer = backlog + peer.pagination = "cursor" + peer.cursor_behavior = cursor_behavior + peer.add("exhausted", "failed") + peer.add("older", "pending") + peer.add("newer", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.RETRY + ) + assert ctx._sync_complete is False + assert peer.transitions == [] + assert seen == [] + assert peer.list_calls <= 2 + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + async def test_ack_pending_snapshot_head_never_replays_or_overtakes( + self, backlog: tuple[BandLink, BacklogApi], phase: str + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("unacked", "pending") + peer.add("newer", "pending") + peer.refuse_processed.add("unacked") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + ctx.claims.remember_ack_pending(ctx.room_id, "unacked") + + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.RETRY + ) + assert ctx.claims.is_ack_pending(ctx.room_id, "unacked") + assert seen == [] + assert peer.transitions == [] + + peer.refuse_processed.clear() + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["newer"] + assert peer.statuses["unacked"] == "processed" + assert ("unacked", "processing") not in peer.transitions + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + async def test_retryable_snapshot_failure_blocks_newer_until_exhausted( + self, backlog: tuple[BandLink, BacklogApi], phase: str + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("retryable", "failed") + peer.add("newer", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + if event.payload.id == "retryable": + raise RuntimeError("retryable failure") + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + + for _ in range(2): + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.RETRY + ) + assert peer.statuses["newer"] == "pending" + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["retryable", "retryable", "newer"] + assert peer.statuses["retryable"] == "failed" + assert ctx._retry_tracker.is_permanently_failed("retryable") + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + async def test_unaccepted_snapshot_head_retains_fifo_and_failure_budget( + self, backlog: tuple[BandLink, BacklogApi], phase: str + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("unaccepted", "pending") + peer.add("newer", "pending") + accepted = False + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if not accepted: + raise InputNotAccepted("provider is busy") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + for _ in range(3): + assert await self.recover(ctx, phase) is ( + False if phase == "bootstrap" else ResyncOutcome.BLOCKED + ) + assert seen == [] + assert peer.statuses["newer"] == "pending" + assert not ctx._retry_tracker.is_permanently_failed("unaccepted") + assert ctx._deferred_message_id == "unaccepted" + assert all(action != "failed" for _, action in peer.transitions) + + accepted = True + assert await self.recover(ctx, phase) is ( + True if phase == "bootstrap" else ResyncOutcome.WORK + ) + assert seen == ["unaccepted", "newer"] + assert ctx._deferred_message_id is None + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("refusal", ["stopped", "empty-success", "unsuccessful"]) + async def test_snapshot_requires_accepted_claim_before_provider_effects( + self, backlog: tuple[BandLink, BacklogApi], phase: str, refusal: str + ) -> None: + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("candidate", "pending") + peer.add("newer", "pending") + if refusal == "stopped": + peer.processing_response = httpx.Response(204) + elif refusal == "empty-success": + peer.processing_response = httpx.Response(200) + else: + peer.processing_response = httpx.Response( + 200, + json={ + "data": { + "id": "candidate", + "status": "processing", + "attempt_number": 1, + "success": False, + } + }, + ) + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + await self.recover(ctx, phase) + assert seen == [] + assert peer.statuses["newer"] == "pending" + assert all(action != "processed" for _, action in peer.transitions) + + peer.processing_response = None + if ctx.is_stopped: + await ctx.resume_room() + await self.recover(ctx, phase) + assert seen == ["candidate", "newer"] + + async def test_overlapping_pages_are_deduplicated_before_any_claim( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + peer.pagination = "cursor" + peer.add("exhausted", "failed") + peer.add("older", "pending") + peer.add("newer", "pending") + seen: list[str] = [] + + def overlapping_page(request: httpx.Request) -> httpx.Response: + response = peer.respond(request) + if request.url.path.endswith("/messages"): + assert peer.transitions == [] + if request.url.params.get("cursor") == "3": + payload = response.json() + payload["data"].insert(0, peer.message("older")) + return httpx.Response(200, json=payload) + return response + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + async with rest_client_over(overlapping_page) as rest: + link.rest = rest + assert await self.recover(ctx, "bootstrap") is True + assert seen == ["older", "newer"] + assert peer.transitions == [ + ("older", "processing"), + ("older", "processed"), + ("newer", "processing"), + ("newer", "processed"), + ] + + @pytest.mark.parametrize("skip", ["completed", "exhausted", "server-processed"]) + async def test_skipped_websocket_marker_reads_snapshot_before_queue_handoff( + self, backlog: tuple[BandLink, BacklogApi], skip: str + ) -> None: + link, peer = backlog + peer.add("marker", "processed" if skip == "server-processed" else "pending") + peer.add("later", "pending") + peer.fixed_head = "marker" + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + seen.append(event.payload.id) + + ctx = self.context(link, handle) + if skip == "completed": + ctx.claims.remember_completed(ctx.room_id, "marker") + elif skip == "exhausted": + ctx._retry_tracker.mark_permanently_failed("marker") + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="marker")) + await ctx.on_event( + make_participant_added_event(room_id=ROOM_ID, participant_id="new-user") + ) + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="later")) + assert await self.recover(ctx, "bootstrap") is True + assert peer.list_calls == (1 if skip == "server-processed" else 2) + assert seen == [] + assert peer.transitions == [] + while not ctx.queue.empty(): + await ctx._process_event(ctx.queue.get_nowait()) + assert seen == ["new-user", "later"] + + async def test_retryable_websocket_marker_is_not_removed_as_terminal( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + peer.add("marker", "pending") + failed = True + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if failed: + raise RuntimeError("transient failure") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="marker")) + assert await self.recover(ctx, "bootstrap") is False + assert ctx._first_ws_msg_id == "marker" + assert ctx.queue.qsize() == 1 + assert ctx._sync_complete is False + assert ctx._deferred_message_id == "marker" + failed = False + assert await self.recover(ctx, "bootstrap") is True + assert seen == ["marker"] + assert ctx.queue.empty() + assert ctx._first_ws_msg_id is None + + async def test_terminal_deferred_head_preserves_synthetic_fifo_before_snapshot_rows( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + link, peer = backlog + peer.add("unaccepted", "pending") + peer.add("later", "pending") + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if event.payload.id == "unaccepted": + raise InputNotAccepted("provider is busy") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + assert await self.recover(ctx, "idle") is ResyncOutcome.BLOCKED + await ctx._retain_or_handle_control( + make_message_event( + room_id=ROOM_ID, + msg_id="local-contact", + sender_id="contact-events", + sender_type="System", + ) + ) + peer.statuses["unaccepted"] = "processed" + peer.fixed_head = "unaccepted" + assert await self.recover(ctx, "idle") is ResyncOutcome.BLOCKED + assert peer.list_calls == 1 + assert seen == [] + assert peer.statuses["later"] == "pending" + assert ctx._deferred_message_id is None + assert await self.recover(ctx, "idle") is ResyncOutcome.WORK + assert seen == ["local-contact", "later"] + assert peer.statuses["later"] == "processed" + + async def test_bootstrap_snapshot_completes_before_queued_synthetic_work( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + """A synthetic turn queued while bootstrap is deferred must not stall the + snapshot: bootstrap finishes the backlog, then the queue drains.""" + link, peer = backlog + peer.add("first", "pending") + peer.add("exhausted", "failed") + peer.add("later", "pending") + accepted = False + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if not accepted: + raise InputNotAccepted("provider is busy") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + assert await self.recover(ctx, "bootstrap") is False + await ctx._retain_or_handle_control( + make_message_event( + room_id=ROOM_ID, + msg_id="local-contact", + sender_id="contact-events", + sender_type="System", + ) + ) + accepted = True + for _ in range(3): + async with asyncio.timeout(1): + await ctx._reconcile() + + assert seen == ["first", "later", "local-contact"] + assert ctx._sync_complete + assert not ctx._pending_synthetic + assert peer.statuses["later"] == "processed" + + async def test_deferred_row_behind_terminal_head_settles_before_synthetic_work( + self, backlog: tuple[BandLink, BacklogApi] + ) -> None: + """With an older real message deferred behind a terminal /next head, a + queued synthetic turn waits for that message, then runs before later rows.""" + link, peer = backlog + peer.add("exhausted", "failed") + peer.add("deferred", "pending") + peer.add("later", "pending") + accepted = False + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if not accepted: + raise InputNotAccepted("provider is busy") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + assert await self.recover(ctx, "idle") is ResyncOutcome.BLOCKED + assert ctx._deferred_message_id == "deferred" + await ctx._retain_or_handle_control( + make_message_event( + room_id=ROOM_ID, + msg_id="local-contact", + sender_id="contact-events", + sender_type="System", + ) + ) + accepted = True + for _ in range(3): + await self.recover(ctx, "idle") + + assert seen == ["deferred", "local-contact", "later"] + assert ctx._deferred_message_id is None + assert peer.statuses["later"] == "processed" + + @staticmethod + async def retain_synthetic(ctx: ExecutionContext, *message_ids: str) -> None: + for message_id in message_ids: + await ctx._retain_or_handle_control( + make_message_event( + room_id=ROOM_ID, + msg_id=message_id, + sender_id="contact-events", + sender_type="System", + ) + ) + + @staticmethod + async def control_while_first_settles( + ctx: ExecutionContext, + peer: BacklogApi, + control: ControlMode, + entered: asyncio.Event, + release: asyncio.Event, + ) -> None: + """Land ``control`` while the deferred row ``first`` is mid-turn, then + run reconciliation until the room has had every chance to drain.""" + recovering = asyncio.create_task(ctx._reconcile()) + try: + async with asyncio.timeout(2): + await entered.wait() + if control is ControlMode.STOP: + peer.stopped = True + ctx.stop_room() + elif control is ControlMode.PLAY: + await ctx.resume_room() + else: + ctx.interrupt(kind=control) + release.set() + await recovering + if control is ControlMode.STOP: + peer.stopped = False + await ctx.resume_room() + async with asyncio.timeout(2): + for _ in range(4): + await ctx._reconcile() + finally: + release.set() + recovering.cancel() + await asyncio.gather(recovering, return_exceptions=True) + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("control", list(ControlMode)) + async def test_control_while_deferred_row_settles_never_strands_synthetic_work( + self, backlog: tuple[BandLink, BacklogApi], phase: str, control: ControlMode + ) -> None: + """A control landing while a deferred row finishes must still release that + row's deferral, though a terminal head keeps /next from ever going empty; + otherwise queued synthetic turns wait on a row that is already settled.""" + link, peer = backlog + if phase == "bootstrap": + peer.add("exhausted", "failed") + peer.add("first", "pending") + if phase == "idle": + peer.add("exhausted", "failed") + peer.add("later", "pending") + entered, release = asyncio.Event(), asyncio.Event() + accepted, gated = False, False + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + nonlocal gated + if not accepted: + raise InputNotAccepted("provider is busy") + if event.payload.id == "first" and not gated: + gated = True + entered.set() + await release.wait() + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + if phase == "idle": + ctx._sync_complete = True + assert await self.recover(ctx, phase) in {False, ResyncOutcome.BLOCKED} + assert ctx._deferred_message_id == "first" + await self.retain_synthetic(ctx, "local-A", "local-B") + accepted = True + try: + await self.control_while_first_settles(ctx, peer, control, entered, release) + finally: + await ctx.stop() + + replayed = [] if control is ControlMode.INTERRUPT else ["first"] + if phase == "bootstrap": + # Bootstrap finishes the backlog before the queue, as /next sync does. + assert seen == [*replayed, "later", "local-A", "local-B"] + else: + assert seen == [*replayed, "local-A", "local-B", "later"] + assert ctx._deferred_message_id is None + assert not ctx._pending_synthetic + + @pytest.mark.parametrize("phase", ["bootstrap", "idle"]) + @pytest.mark.parametrize("head", ["next", "terminal"]) + async def test_deferral_held_while_ack_pending_and_released_once_acked( + self, backlog: tuple[BandLink, BacklogApi], phase: str, head: str + ) -> None: + """A deferred row whose processed ack fails keeps its deferral, so nothing + overtakes it; once the ack retry lands, queued synthetic turns run next.""" + link, peer = backlog + if head == "terminal": + peer.add("exhausted", "failed") + peer.add("first", "pending") + peer.add("later", "pending") + accepted = False + seen: list[str] = [] + + async def handle(ctx: ExecutionContext, event: Any) -> None: + if not accepted: + raise InputNotAccepted("provider is busy") + seen.append(event.payload.id) + + ctx = self.context(link, handle) + ctx._retry_tracker.mark_permanently_failed("exhausted") + if phase == "idle": + ctx._sync_complete = True + try: + async with asyncio.timeout(3): + await self.recover(ctx, phase) + assert ctx._deferred_message_id == "first" + await self.retain_synthetic(ctx, "local-A", "local-B") + accepted = True + peer.refuse_processed.add("first") + await ctx._reconcile() + assert ctx.claims.is_ack_pending(ROOM_ID, "first") + assert ctx._deferred_message_id == "first" + assert seen == ["first"] + assert peer.statuses["later"] == "pending" + + peer.refuse_processed.clear() + for _ in range(4): + await ctx._reconcile() + finally: + await ctx.stop() + + if phase == "bootstrap": + assert seen == ["first", "later", "local-A", "local-B"] + else: + assert seen == ["first", "local-A", "local-B", "later"] + assert ctx._deferred_message_id is None + assert not ctx._pending_synthetic diff --git a/tests/runtime/resync/test_resync_passes.py b/tests/runtime/resync/test_resync_passes.py index 4163a1147..7e7c1ca2d 100644 --- a/tests/runtime/resync/test_resync_passes.py +++ b/tests/runtime/resync/test_resync_passes.py @@ -167,6 +167,7 @@ async def test_unchanged_completed_head_defers_without_spinning( ctx.claims.remember_completed(ROOM_ID, "stuck") peer.add_message("stuck") assert await ctx._resync_pending_messages() is ResyncOutcome.BLOCKED - assert len(peer.next_reads) == 2 + assert len(peer.next_reads) == 1 + assert peer.actionable_cursors == [None] assert peer.requested("processed") == [] assert invoked == [] diff --git a/tests/runtime/resync/test_stopped_room_recovery.py b/tests/runtime/resync/test_stopped_room_recovery.py index e91e79e1f..e358ae4de 100644 --- a/tests/runtime/resync/test_stopped_room_recovery.py +++ b/tests/runtime/resync/test_stopped_room_recovery.py @@ -3,12 +3,14 @@ from __future__ import annotations import asyncio +from typing import Any import pytest from band.client.streaming import ControlMode from band.runtime.cycle import TurnScope -from band.runtime.execution import ResyncOutcome +from band.runtime.execution import ExecutionContext, ResyncOutcome +from tests.conftest import make_message_event from tests.runtime.conftest import wait_for_condition from tests.runtime.helpers import ( ROOM_ID, @@ -39,6 +41,45 @@ async def test_known_stop_recovers_after_missed_play( assert invoked == ["resumed"] assert peer.marked("processed") == ["resumed"] + @pytest.mark.looptime + async def test_deferred_local_event_recovers_missed_play_before_newer_delivery( + self, reconciled_room: ReconciledRoom + ) -> None: + ctx, peer, invoked = reconciled_room + entered = asyncio.Event() + first_attempt = True + + async def handler(ctx: ExecutionContext, event: Any) -> None: + nonlocal first_attempt + if event.payload.id == "local-contact" and first_attempt: + first_attempt = False + entered.set() + await asyncio.Event().wait() + invoked.append(event.payload.id) + + ctx._on_execute = handler + await ctx.start() + await wait_for_condition(lambda: ctx._sync_complete) + await ctx.on_event( + make_message_event( + room_id=ROOM_ID, + msg_id="local-contact", + sender_id="contact-events", + sender_type="System", + ) + ) + await entered.wait() + ctx.stop_room() + peer.stopped = True + peer.add_message("after-local") + await ctx.on_event(make_message_event(room_id=ROOM_ID, msg_id="after-local")) + await asyncio.sleep(61) + assert invoked == [] + peer.stopped = False + await asyncio.sleep(60) + assert invoked == ["local-contact", "after-local"] + assert peer.marked("processed") == ["after-local"] + @pytest.mark.looptime @pytest.mark.parametrize("mode", list(ControlMode)) async def test_control_during_stopped_probe_invalidates_response( diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index 8fdb098d5..3934be986 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -82,6 +82,7 @@ def mock_link(): link.mark_failed = AsyncMock(return_value=True) link.get_next_message = AsyncMock(return_value=None) # No backlog by default link.get_stale_processing_messages = AsyncMock(return_value=[]) + link.get_actionable_messages = AsyncMock(return_value=[]) link.report_activity = AsyncMock(return_value=True) return link @@ -658,6 +659,7 @@ def mock_link_with_next(self): link.mark_failed = AsyncMock() link.get_next_message = AsyncMock(return_value=None) # No backlog by default link.get_stale_processing_messages = AsyncMock(return_value=[]) # No stale msgs + link.get_actionable_messages = AsyncMock(return_value=[]) link.report_activity = AsyncMock(return_value=True) return link diff --git a/tests/runtime/test_execution_working_state.py b/tests/runtime/test_execution_working_state.py index fa39effdf..21cd23352 100644 --- a/tests/runtime/test_execution_working_state.py +++ b/tests/runtime/test_execution_working_state.py @@ -41,6 +41,7 @@ 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 diff --git a/tests/test_session_sync.py b/tests/test_session_sync.py index 5f6d3328f..369e88f58 100644 --- a/tests/test_session_sync.py +++ b/tests/test_session_sync.py @@ -56,6 +56,7 @@ def mock_link(): link.mark_processed = AsyncMock() link.mark_failed = AsyncMock() link.get_stale_processing_messages = AsyncMock(return_value=[]) + link.get_actionable_messages = AsyncMock(return_value=[]) return link