From 4eb7fff583a0caa5396fadda46582063a9f492f3 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 10:24:46 +0300 Subject: [PATCH 1/8] test(sdk): add message-claim regression tests for freshly-joined peers Green regressions locking current single-context guarantees: - The gateway sequence (message posted before the peer's context is live, delivered via /next with the WebSocket copy arriving mid-execution) executes the handler exactly once. - Cancellation mid-processing releases the local in-flight claim. Red targets defining the desired lifecycle, to turn green with the fix: - Two live contexts for one room execute a shared message once. - ACK_PENDING state survives LRU pressure instead of being evicted into a side-effect replay on redelivery. Co-Authored-By: Claude Fable 5 --- tests/runtime/test_execution.py | 138 +++++++++++++++++++++++++++++++- 1 file changed, 137 insertions(+), 1 deletion(-) diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index cb59f9f48..82db894df 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -813,6 +813,110 @@ async def delayed_mark_processing(room_id: str, message_id: str) -> bool: assert mock_handler.await_count == 1 assert ctx._inflight_message_ids == set() + async def test_first_message_to_fresh_room_executes_once(self, mock_link_with_next): + """A message posted before the room's context was live executes once. + + The gateway sequence: create room, add peer, post immediately. The + peer's startup sync receives the message from /next, and the + WebSocket copy arrives while that execution is still in flight. The + second delivery must be deduplicated, not re-executed. + """ + from band.runtime.types import PlatformMessage + + handler_started = asyncio.Event() + release_handler = asyncio.Event() + handled_message_ids: list[str] = [] + + async def blocking_handler(_context: ExecutionContext, event) -> None: + handled_message_ids.append(event.payload.id) + handler_started.set() + await release_handler.wait() + + first_message = PlatformMessage( + id="msg-first", + room_id="room-123", + content="posted before the peer subscribed", + sender_id="user-1", + sender_type="User", + sender_name="User One", + message_type="text", + metadata={}, + created_at=datetime.now(timezone.utc), + ) + mock_link_with_next.get_next_message = AsyncMock( + side_effect=[first_message, None] + ) + + ctx = ExecutionContext( + "room-123", + mock_link_with_next, + blocking_handler, + agent_id="agent-123", + config=SessionConfig(enable_context_hydration=False), + ) + + await ctx.start() + await asyncio.wait_for(handler_started.wait(), timeout=1.0) + + # WebSocket copy of the same message arrives mid-execution. + await ctx.on_event(make_message_event(room_id="room-123", msg_id="msg-first")) + + release_handler.set() + await asyncio.sleep(0.2) # Let sync finish and Phase 2 drain the WS copy + + assert handled_message_ids == ["msg-first"] + assert mock_link_with_next.mark_processing.await_count == 1 + assert mock_link_with_next.mark_processed.await_count == 1 + + await ctx.stop() + + async def test_fresh_contexts_do_not_execute_the_same_message_twice( + self, mock_link_with_next + ): + """Two live contexts for one room must execute a shared message once.""" + handler_started = asyncio.Event() + release_handler = asyncio.Event() + handler_calls = 0 + + async def blocking_handler(_context: ExecutionContext, _event: object) -> None: + nonlocal handler_calls + handler_calls += 1 + handler_started.set() + await release_handler.wait() + + first_context = ExecutionContext( + "room-123", + mock_link_with_next, + blocking_handler, + agent_id="agent-123", + config=SessionConfig( + enable_context_hydration=False, + enable_working_state=False, + ), + ) + second_context = ExecutionContext( + "room-123", + mock_link_with_next, + blocking_handler, + agent_id="agent-123", + config=SessionConfig( + enable_context_hydration=False, + enable_working_state=False, + ), + ) + event = make_message_event(room_id="room-123", msg_id="msg-fresh-peer") + + first_task = asyncio.create_task(first_context._process_event(event)) + await asyncio.wait_for(handler_started.wait(), timeout=1.0) + second_task = asyncio.create_task(second_context._process_event(event)) + + await asyncio.sleep(0) + release_handler.set() + await asyncio.gather(first_task, second_task) + + assert handler_calls == 1 + assert mock_link_with_next.mark_processing.await_count == 1 + async def test_mark_processing_failure_does_not_execute_message( self, mock_link_with_next, mock_handler ): @@ -991,6 +1095,35 @@ async def test_websocket_processed_ack_failure_retries_ack_without_handler_repla assert "msg-ws-ack-retry" in ctx._processed_ids assert "msg-ws-ack-retry" not in ctx._processed_ack_pending_ids + async def test_ack_pending_message_survives_lru_pressure_without_replay( + self, mock_link_with_next, mock_handler + ): + """Cache pressure must never evict ACK_PENDING into a side-effect replay. + + A message whose handler completed but whose durable processed ack + failed may only retry the ack on redelivery. Filling the cache past + capacity must not erase that state and re-execute the handler. + """ + ctx = ExecutionContext( + "room-123", + mock_link_with_next, + mock_handler, + agent_id="agent-123", + config=SessionConfig(enable_context_hydration=False), + ) + + ctx._remember_processed_ack_pending("msg-unacked") + for i in range(ctx._max_processed_ids): + ctx._remember_processed_ack_pending(f"msg-filler-{i}") + + event = make_message_event(room_id="room-123", msg_id="msg-unacked") + await ctx._process_event(event) + + mock_handler.assert_not_awaited() + mock_link_with_next.mark_processed.assert_awaited_once_with( + "room-123", "msg-unacked" + ) + async def test_websocket_processed_ack_failure_retries_ack_before_newer_queue( self, mock_link_with_next, mock_handler ): @@ -1362,7 +1495,7 @@ class TestCancellationDuringProcessing: """Tests for cancellation during message processing.""" async def test_stop_cancels_slow_processing(self, mock_link): - """stop() should cancel message processing.""" + """stop() should cancel processing and release the in-flight claim.""" async def slow_handler(ctx, event): await asyncio.sleep(10) # Would take 10 seconds @@ -1393,6 +1526,9 @@ async def slow_handler(ctx, event): # Should complete quickly (not wait 10 seconds for handler) assert elapsed < 1.0, f"stop() took {elapsed}s - should cancel processing" + # Cancellation must release the local in-flight claim + assert ctx._inflight_message_ids == set() + class TestContextHydrationConfig: """Test context hydration behavior with config.""" From aae6f5b933b91e1b1e0d7e03b1d04955965c1d7f Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 10:28:59 +0300 Subject: [PATCH 2/8] test(sdk): drive the ack-pending LRU regression through real deliveries Arrange ACK_PENDING and the cache overflow via _process_event with a failing durable ack instead of poking the bookkeeping directly, and give the retry budget headroom so eviction surfaces as the actual hazard: a handler replay on redelivery. Co-Authored-By: Claude Fable 5 --- tests/runtime/test_execution.py | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index 82db894df..82935ba8c 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -1101,25 +1101,42 @@ async def test_ack_pending_message_survives_lru_pressure_without_replay( """Cache pressure must never evict ACK_PENDING into a side-effect replay. A message whose handler completed but whose durable processed ack - failed may only retry the ack on redelivery. Filling the cache past - capacity must not erase that state and re-execute the handler. + failed may only retry the ack on redelivery — even after enough later + completions to overflow the dedupe cache. """ + mock_link_with_next.mark_processing = AsyncMock(return_value=True) + mock_link_with_next.mark_processed = AsyncMock(return_value=False) ctx = ExecutionContext( "room-123", mock_link_with_next, mock_handler, agent_id="agent-123", - config=SessionConfig(enable_context_hydration=False), + config=SessionConfig( + enable_context_hydration=False, + enable_working_state=False, + # Retry headroom so eviction shows up as a handler replay, + # not as the retry budget silently dropping the redelivery. + max_message_retries=5, + ), ) - ctx._remember_processed_ack_pending("msg-unacked") + async def deliver(message_id: str) -> None: + await ctx._process_event( + make_message_event(room_id="room-123", msg_id=message_id) + ) + + # Handler completes but the durable ack fails → ACK_PENDING. + await deliver("msg-unacked") + # Enough later completions to overflow the dedupe cache capacity. for i in range(ctx._max_processed_ids): - ctx._remember_processed_ack_pending(f"msg-filler-{i}") + await deliver(f"msg-filler-{i}") - event = make_message_event(room_id="room-123", msg_id="msg-unacked") - await ctx._process_event(event) + executions_before_redelivery = mock_handler.await_count + mock_link_with_next.mark_processed = AsyncMock(return_value=True) + + await deliver("msg-unacked") - mock_handler.assert_not_awaited() + assert mock_handler.await_count == executions_before_redelivery mock_link_with_next.mark_processed.assert_awaited_once_with( "room-123", "msg-unacked" ) From b335be4e3db337be99b23299068a7bc8cd879153 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 10:40:12 +0300 Subject: [PATCH 3/8] fix(sdk): never evict ack-pending message state under cache pressure Evicting a completed-but-unacked message from the ack-pending map turned redelivery into a full handler replay of already-executed side effects. Ack-pending entries have a bounded exit of their own (each retry either succeeds or the retry budget promotes the message to completed), so the LRU cap on this map was never needed for growth control. Co-Authored-By: Claude Fable 5 --- src/band/runtime/execution.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index c825d4655..49288fe8e 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -384,12 +384,15 @@ def _remember_processed_message(self, message_id: str) -> None: self._processed_ids.popitem(last=False) def _remember_processed_ack_pending(self, message_id: str) -> None: - """Track local completion while waiting for durable processed ack.""" + """Track local completion while waiting for durable processed ack. + + Never evicted under cache pressure: losing this state would replay + the handler's side effects on redelivery. Entries drain through the + ack retry budget instead (success or promotion to completed). + """ self._processed_ack_pending_ids[message_id] = True self._processed_ack_pending_ids.move_to_end(message_id) self._processed_ack_retry_counts.setdefault(message_id, 0) - if len(self._processed_ack_pending_ids) > self._max_processed_ids: - self._processed_ack_pending_ids.popitem(last=False) async def _retry_processed_ack(self, message_id: str) -> bool: """Retry durable processed ack for a locally completed message.""" From 5c1325ee2e625aa90880f352426d13aeec7489ad Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 11:23:26 +0300 Subject: [PATCH 4/8] fix(runtime): scope message claims by room --- .../integrations/claude_sdk/dedup_tools.py | 6 +- src/band/runtime/claims.py | 108 +++++++++++++++++ src/band/runtime/execution.py | 110 ++++++------------ src/band/runtime/runtime.py | 6 + tests/runtime/test_execution.py | 96 +++++++-------- tests/runtime/test_resync.py | 4 +- tests/runtime/test_runtime.py | 34 ++++++ tests/test_session_sync.py | 22 ++-- 8 files changed, 252 insertions(+), 134 deletions(-) create mode 100644 src/band/runtime/claims.py diff --git a/src/band/integrations/claude_sdk/dedup_tools.py b/src/band/integrations/claude_sdk/dedup_tools.py index 59d2c89ea..ca069dd78 100644 --- a/src/band/integrations/claude_sdk/dedup_tools.py +++ b/src/band/integrations/claude_sdk/dedup_tools.py @@ -13,9 +13,9 @@ chat). The platform happily accepts every POST, so the duplicate becomes a visible -chat message and a charged LLM call. ``ExecutionContext._processed_ids`` only -dedupes *inbound* user messages from the platform; it has no view of outbound -tool calls produced by Claude. +chat message and a charged LLM call. The runtime's ``MessageClaimRegistry`` +only dedupes *inbound* user messages from the platform; it has no view of +outbound tool calls produced by Claude. This wrapper sits in front of the per-room ``AgentToolsProtocol`` that the ``ClaudeSDKAdapter`` registers in ``_room_tools[room_id]``. It dedupes diff --git a/src/band/runtime/claims.py b/src/band/runtime/claims.py new file mode 100644 index 000000000..ab17700a3 --- /dev/null +++ b/src/band/runtime/claims.py @@ -0,0 +1,108 @@ +"""Shared message-claim ledger for the inbound delivery lifecycle.""" + +from __future__ import annotations + +from collections import OrderedDict +from typing import TypeAlias + +DEFAULT_COMPLETED_CACHE_SIZE = 500 +ClaimKey: TypeAlias = tuple[str, str] + + +class MessageClaimRegistry: + """Ownership ledger ensuring one execution per inbound message ID. + + Lifecycle states: + + - **in flight** — claimed by exactly one execution via ``try_claim``; + released on failure or cancellation, never evicted. + - **ack pending** — the handler completed but the durable processed ack + failed; redelivery retries only the ack, never the handler. Never + evicted (losing it would replay side effects); drains through the ack + retry budget instead. + - **completed** — durably processed; kept in a bounded LRU for dedup. + + ``AgentRuntime`` owns one registry and passes it to every context it + creates, so recreated contexts retain ownership state. Claims are keyed by + room and message ID, preventing work in one room from affecting another. + A context constructed standalone gets a private registry, preserving the + previous per-context behavior. + + ``try_claim`` checks and inserts without an ``await``, so event-loop + scheduling makes claims atomic within one runtime. + + Scope, honestly stated: this coordinates executions within one runtime + (one event loop). It cannot coordinate separate processes, containers, + or hosts — the platform re-serves in-flight messages to fresh actors and + ``mark_processing`` is not an exclusive claim, so deployments that shard + one agent id across such boundaries need a platform-level claim (see + ``band.runtime.single_instance`` for the same boundary statement). + """ + + def __init__(self, max_completed: int = DEFAULT_COMPLETED_CACHE_SIZE) -> None: + self.max_completed = max_completed + self._inflight: set[ClaimKey] = set() + self._ack_pending: OrderedDict[ClaimKey, bool] = OrderedDict() + self._ack_retries: dict[ClaimKey, int] = {} + self._completed: OrderedDict[ClaimKey, bool] = OrderedDict() + + def try_claim(self, room_id: str, message_id: str) -> bool: + """Claim a message for one execution; False if another owner holds it.""" + key = (room_id, message_id) + if key in self._inflight: + return False + self._inflight.add(key) + return True + + def release(self, room_id: str, message_id: str) -> None: + """Release an in-flight claim (failure, cancellation, or completion).""" + self._inflight.discard((room_id, message_id)) + + @property + def inflight_ids(self) -> set[str]: + """Currently claimed message IDs (copy).""" + return set(self._inflight) + + def is_completed(self, room_id: str, message_id: str) -> bool: + """Whether the message already completed; a hit refreshes LRU recency.""" + key = (room_id, message_id) + if key not in self._completed: + return False + self._completed.move_to_end(key) + return True + + @property + def completed_ids(self) -> list[str]: + """Completed message IDs, oldest first (copy, no recency side effect).""" + return [message_id for _, message_id in self._completed] + + def remember_completed(self, room_id: str, message_id: str) -> None: + """Record durable completion and clear any pending-ack state.""" + key = (room_id, message_id) + self._completed[key] = True + self._completed.move_to_end(key) + self._ack_pending.pop(key, None) + self._ack_retries.pop(key, None) + if len(self._completed) > self.max_completed: + self._completed.popitem(last=False) + + def is_ack_pending(self, room_id: str, message_id: str) -> bool: + """Whether the message completed locally but lacks a durable ack.""" + return (room_id, message_id) in self._ack_pending + + def remember_ack_pending(self, room_id: str, message_id: str) -> None: + """Record local completion awaiting a durable processed ack.""" + key = (room_id, message_id) + self._ack_pending[key] = True + self._ack_retries.setdefault(key, 0) + + def pending_ack_ids(self, room_id: str) -> list[str]: + """Message IDs awaiting a durable processed ack, oldest first.""" + return [message_id for room, message_id in self._ack_pending if room == room_id] + + def record_ack_retry(self, room_id: str, message_id: str) -> int: + """Count a failed ack retry; returns the total for budget checks.""" + key = (room_id, message_id) + retries = self._ack_retries.get(key, 0) + 1 + self._ack_retries[key] = retries + return retries diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 49288fe8e..3cd8fbd20 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -16,7 +16,6 @@ import asyncio import logging -from collections import OrderedDict from datetime import datetime, timezone from enum import Enum from typing import ( @@ -48,6 +47,7 @@ SYNTHETIC_SENDER_TYPE, SYNTHETIC_CONTACT_EVENTS_SENDER_ID, ) +from .claims import MessageClaimRegistry from .retry_tracker import MessageRetryTracker from band.runtime.context_serialization import context_item_to_dict from .working_state import WorkingStateReporter @@ -189,6 +189,7 @@ def __init__( on_participant_removed: ParticipantRemovedCallback | None = None, *, hub_room_id: str | None = None, + claim_registry: MessageClaimRegistry | None = None, ): """ Initialize execution context for a specific room. @@ -204,6 +205,10 @@ def __init__( hub_room_id: Optional hub-room ID. Forwarded to AgentTools so the schema methods can auto-enable contact tools when this context belongs to the hub room. + claim_registry: Optional shared message-claim registry. AgentRuntime + passes one registry to all contexts it creates so a message ID + executes at most once per runtime. Defaults to a private + instance for standalone contexts. """ self.room_id = room_id self.link = link @@ -242,19 +247,10 @@ def __init__( # LLM context tracking self._llm_initialized = False - # Dedupe cache (LRU for detecting duplicates during sync) - self._processed_ids: OrderedDict[str, bool] = OrderedDict() - self._max_processed_ids: int = 500 - - # Local in-flight guard shared by /next and WebSocket processing. - # This prevents the same message ID from executing twice before the - # durable processed status becomes visible to both paths. - self._inflight_message_ids: set[str] = set() - - # Messages whose local handler completed but whose durable processed ack - # failed. Redelivery should retry only the ack, not the side effects. - self._processed_ack_pending_ids: OrderedDict[str, bool] = OrderedDict() - self._processed_ack_retry_counts: dict[str, int] = {} + # Message ownership ledger (in-flight claims, completed LRU, pending + # acks) shared by /next and WebSocket processing. Runtime-provided so + # all contexts of one agent coordinate; private otherwise. + self.claims = claim_registry or MessageClaimRegistry() # Crash recovery: sync point marker and retry tracking self._first_ws_msg_id: str | None = None # First WS message = sync point @@ -374,40 +370,18 @@ def _message_processed_for_agent(self, message_id: str, metadata: Any) -> bool: == DeliveryStatus.PROCESSED ) - def _remember_processed_message(self, message_id: str) -> None: - """Track a processed message ID in the local LRU dedupe cache.""" - self._processed_ids[message_id] = True - self._processed_ids.move_to_end(message_id) - self._processed_ack_pending_ids.pop(message_id, None) - self._processed_ack_retry_counts.pop(message_id, None) - if len(self._processed_ids) > self._max_processed_ids: - self._processed_ids.popitem(last=False) - - def _remember_processed_ack_pending(self, message_id: str) -> None: - """Track local completion while waiting for durable processed ack. - - Never evicted under cache pressure: losing this state would replay - the handler's side effects on redelivery. Entries drain through the - ack retry budget instead (success or promotion to completed). - """ - self._processed_ack_pending_ids[message_id] = True - self._processed_ack_pending_ids.move_to_end(message_id) - self._processed_ack_retry_counts.setdefault(message_id, 0) - async def _retry_processed_ack(self, message_id: str) -> bool: """Retry durable processed ack for a locally completed message.""" - if message_id not in self._processed_ack_pending_ids: + if not self.claims.is_ack_pending(self.room_id, message_id): return False - self._processed_ack_pending_ids.move_to_end(message_id) durable_processed = await self.link.mark_processed(self.room_id, message_id) if durable_processed: self._retry_tracker.mark_success(message_id) - self._remember_processed_message(message_id) + self.claims.remember_completed(self.room_id, message_id) return True - retries = self._processed_ack_retry_counts.get(message_id, 0) + 1 - self._processed_ack_retry_counts[message_id] = retries + retries = self.claims.record_ack_retry(self.room_id, message_id) if retries >= self._retry_tracker.max_retries: logger.warning( "ExecutionContext %s: processed ack retry budget exhausted for message %s; keeping local completion marker", @@ -415,22 +389,11 @@ async def _retry_processed_ack(self, message_id: str) -> bool: message_id, ) self._retry_tracker.mark_success(message_id) - self._remember_processed_message(message_id) + self.claims.remember_completed(self.room_id, message_id) return True return False - def _try_claim_local_message(self, message_id: str) -> bool: - """Claim a message ID for local processing before hydration/status writes.""" - if message_id in self._inflight_message_ids: - return False - self._inflight_message_ids.add(message_id) - return True - - def _release_local_message(self, message_id: str) -> None: - """Release a local in-flight message claim.""" - self._inflight_message_ids.discard(message_id) - # --- Execution protocol implementation --- async def start(self) -> None: @@ -955,7 +918,7 @@ async def _process_loop(self) -> None: async def _retry_pending_processed_acks(self) -> bool: """Retry durable processed acks for locally completed messages.""" - for msg_id in list(self._processed_ack_pending_ids): + for msg_id in self.claims.pending_ack_ids(self.room_id): if not await self._retry_processed_ack(msg_id): return False return True @@ -1203,18 +1166,17 @@ async def _process_backlog_message( return _BacklogProcessResult.ADVANCED # Skip if already processed (dedupe) - if msg_id in self._processed_ids: - self._processed_ids.move_to_end(msg_id) + if self.claims.is_completed(self.room_id, msg_id): logger.debug("Skipping duplicate backlog message: %s", msg_id) return _BacklogProcessResult.ADVANCED - if msg_id in self._processed_ack_pending_ids: + if self.claims.is_ack_pending(self.room_id, msg_id): logger.debug("Retrying processed ack for backlog message: %s", msg_id) if await self._retry_processed_ack(msg_id): return _BacklogProcessResult.ADVANCED return _BacklogProcessResult.RETRY_LATER - if not self._try_claim_local_message(msg_id): + if not self.claims.try_claim(self.room_id, msg_id): logger.debug("Skipping already in-flight backlog message: %s", msg_id) return _BacklogProcessResult.RETRY_LATER @@ -1231,7 +1193,7 @@ async def _process_backlog_message( msg_id, self.room_id, ) - self._remember_processed_message(msg_id) + self.claims.remember_completed(self.room_id, msg_id) return _BacklogProcessResult.ADVANCED # Track attempts - check if exceeded BEFORE processing @@ -1311,9 +1273,9 @@ async def _process_backlog_message( durable_processed = await self.link.mark_processed(self.room_id, msg_id) if durable_processed: self._retry_tracker.mark_success(msg_id) - self._remember_processed_message(msg_id) + self.claims.remember_completed(self.room_id, msg_id) else: - self._remember_processed_ack_pending(msg_id) + self.claims.remember_ack_pending(self.room_id, msg_id) logger.warning( "ExecutionContext %s: Local execution completed but durable processed mark failed for backlog message %s", self.room_id, @@ -1338,7 +1300,7 @@ async def _process_backlog_message( return _BacklogProcessResult.ADVANCED finally: - self._release_local_message(msg_id) + self.claims.release(self.room_id, msg_id) self._set_state("idle") def _drain_duplicate_from_queue(self, msg_id: str) -> None: @@ -1449,20 +1411,24 @@ async def _process_event(self, event: PlatformEvent) -> bool: return True # Skip duplicates - if msg_id in self._processed_ids: - self._processed_ids.move_to_end(msg_id) + if self.claims.is_completed(self.room_id, msg_id): logger.debug("Skipping duplicate message %s", msg_id) return True - if msg_id in self._processed_ack_pending_ids: + if self.claims.is_ack_pending(self.room_id, msg_id): logger.debug("Retrying processed ack for message %s", msg_id) if await self._retry_processed_ack(msg_id): return True return False - if not self._try_claim_local_message(msg_id): - logger.debug("Skipping already in-flight message %s", msg_id) - return True + if not self.claims.try_claim(self.room_id, msg_id): + # Another execution owns this message. Defer rather than + # treat it as handled: the resync safety net re-checks it, + # so an owner failure never silently loses the message. + logger.debug( + "Message %s owned by another execution; deferring", msg_id + ) + return False claimed_msg_id = msg_id if self._message_processed_for_agent(msg_id, payload.metadata): @@ -1471,8 +1437,8 @@ async def _process_event(self, event: PlatformEvent) -> bool: msg_id, self.room_id, ) - self._remember_processed_message(msg_id) - self._release_local_message(msg_id) + self.claims.remember_completed(self.room_id, msg_id) + self.claims.release(self.room_id, msg_id) return True self._set_state("processing") @@ -1491,7 +1457,7 @@ async def _process_event(self, event: PlatformEvent) -> bool: msg_id, self.room_id, ) - self._remember_processed_message(msg_id) + self.claims.remember_completed(self.room_id, msg_id) return True # Track attempts @@ -1537,9 +1503,9 @@ async def _process_event(self, event: PlatformEvent) -> bool: durable_processed = await self.link.mark_processed(self.room_id, msg_id) if durable_processed: self._retry_tracker.mark_success(msg_id) - self._remember_processed_message(msg_id) + self.claims.remember_completed(self.room_id, msg_id) else: - self._remember_processed_ack_pending(msg_id) + self.claims.remember_ack_pending(self.room_id, msg_id) logger.warning( "ExecutionContext %s: Local execution completed but durable processed mark failed for message %s", self.room_id, @@ -1566,5 +1532,5 @@ async def _process_event(self, event: PlatformEvent) -> bool: finally: if claimed_msg_id: - self._release_local_message(claimed_msg_id) + self.claims.release(self.room_id, claimed_msg_id) self._set_state("idle") diff --git a/src/band/runtime/runtime.py b/src/band/runtime/runtime.py index 04c5a6383..f7babad54 100644 --- a/src/band/runtime/runtime.py +++ b/src/band/runtime/runtime.py @@ -12,6 +12,7 @@ from band.platform.event import PlatformEvent +from .claims import MessageClaimRegistry from .execution import Execution, ExecutionContext, ExecutionHandler from .presence import RoomPresence from .types import ( @@ -131,6 +132,10 @@ def __init__( # Per-room executions self.executions: dict[str, Execution] = {} + # Shared by all contexts this runtime creates, so a message ID + # executes at most once per runtime even across context recreation. + self._claim_registry = MessageClaimRegistry() + # Set up presence callbacks self.presence.on_room_joined = self._on_room_joined self.presence.on_room_left = self._on_room_left @@ -273,6 +278,7 @@ async def _create_execution(self, room_id: str) -> Execution: on_participant_added=self._on_participant_added, on_participant_removed=self._on_participant_removed, hub_room_id=self._hub_room_id, + claim_registry=self._claim_registry, ) self.executions[room_id] = execution diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index 82935ba8c..ff4b3a478 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -8,6 +8,7 @@ import pytest +from band.runtime.claims import MessageClaimRegistry from band.runtime.execution import ( Execution, ExecutionContext, @@ -553,7 +554,7 @@ async def test_sync_point_clears_marker_and_keeps_dedupe_cache( # Marker should be cleared assert ctx._first_ws_msg_id is None # Dedupe cache should keep processed sync id to avoid WS reprocessing - assert "msg-sync-001" in ctx._processed_ids + assert "msg-sync-001" in ctx.claims.completed_ids await ctx.stop() @@ -661,7 +662,7 @@ async def test_ws_replay_with_processed_metadata_is_not_reopened( mock_handler.assert_not_called() mock_link_with_next.mark_processing.assert_not_called() - assert "msg-processed-replay" in ctx._processed_ids + assert "msg-processed-replay" in ctx.claims.completed_ids await ctx.stop() @@ -700,7 +701,7 @@ async def test_ws_replay_uses_hydrated_context_delivery_status( mock_handler.assert_not_called() mock_link_with_next.mark_processing.assert_not_called() - assert "msg-stale-replay" in ctx._processed_ids + assert "msg-stale-replay" in ctx.claims.completed_ids await ctx.stop() @@ -811,7 +812,7 @@ async def delayed_mark_processing(room_id: str, message_id: str) -> bool: await backlog_task assert mock_handler.await_count == 1 - assert ctx._inflight_message_ids == set() + assert ctx.claims.inflight_ids == set() async def test_first_message_to_fresh_room_executes_once(self, mock_link_with_next): """A message posted before the room's context was live executes once. @@ -873,7 +874,11 @@ async def blocking_handler(_context: ExecutionContext, event) -> None: async def test_fresh_contexts_do_not_execute_the_same_message_twice( self, mock_link_with_next ): - """Two live contexts for one room must execute a shared message once.""" + """Contexts sharing a runtime's registry execute a shared message once. + + The losing delivery is deferred (not treated as durably handled), so + the resync safety net re-checks it if the owner later fails. + """ handler_started = asyncio.Event() release_handler = asyncio.Event() handler_calls = 0 @@ -884,26 +889,23 @@ async def blocking_handler(_context: ExecutionContext, _event: object) -> None: handler_started.set() await release_handler.wait() - first_context = ExecutionContext( - "room-123", - mock_link_with_next, - blocking_handler, - agent_id="agent-123", - config=SessionConfig( - enable_context_hydration=False, - enable_working_state=False, - ), - ) - second_context = ExecutionContext( - "room-123", - mock_link_with_next, - blocking_handler, - agent_id="agent-123", - config=SessionConfig( - enable_context_hydration=False, - enable_working_state=False, - ), - ) + registry = MessageClaimRegistry() + + def fresh_context() -> ExecutionContext: + return ExecutionContext( + "room-123", + mock_link_with_next, + blocking_handler, + agent_id="agent-123", + config=SessionConfig( + enable_context_hydration=False, + enable_working_state=False, + ), + claim_registry=registry, + ) + + first_context = fresh_context() + second_context = fresh_context() event = make_message_event(room_id="room-123", msg_id="msg-fresh-peer") first_task = asyncio.create_task(first_context._process_event(event)) @@ -912,10 +914,12 @@ async def blocking_handler(_context: ExecutionContext, _event: object) -> None: await asyncio.sleep(0) release_handler.set() - await asyncio.gather(first_task, second_task) + owner_handled, duplicate_handled = await asyncio.gather(first_task, second_task) assert handler_calls == 1 assert mock_link_with_next.mark_processing.await_count == 1 + assert owner_handled is True + assert duplicate_handled is False async def test_mark_processing_failure_does_not_execute_message( self, mock_link_with_next, mock_handler @@ -937,7 +941,7 @@ async def test_mark_processing_failure_does_not_execute_message( ) mock_handler.assert_not_called() mock_link_with_next.mark_processed.assert_not_called() - assert ctx._inflight_message_ids == set() + assert ctx.claims.inflight_ids == set() async def test_backlog_processed_ack_failure_is_not_remembered( self, mock_link_with_next, mock_handler @@ -972,8 +976,8 @@ async def test_backlog_processed_ack_failure_is_not_remembered( mock_link_with_next.mark_processed.assert_awaited_once_with( "room-123", "msg-ack-fails" ) - assert "msg-ack-fails" not in ctx._processed_ids - assert "msg-ack-fails" in ctx._processed_ack_pending_ids + assert "msg-ack-fails" not in ctx.claims.completed_ids + assert ctx.claims.is_ack_pending(ctx.room_id, "msg-ack-fails") async def test_websocket_processed_ack_failure_is_not_remembered( self, mock_link_with_next, mock_handler @@ -995,8 +999,8 @@ async def test_websocket_processed_ack_failure_is_not_remembered( mock_link_with_next.mark_processed.assert_awaited_once_with( "room-123", "msg-ws-ack-fails" ) - assert "msg-ws-ack-fails" not in ctx._processed_ids - assert "msg-ws-ack-fails" in ctx._processed_ack_pending_ids + assert "msg-ws-ack-fails" not in ctx.claims.completed_ids + assert ctx.claims.is_ack_pending(ctx.room_id, "msg-ws-ack-fails") async def test_backlog_processed_ack_failure_retries_ack_without_handler_replay( self, mock_link_with_next, mock_handler @@ -1031,8 +1035,8 @@ async def test_backlog_processed_ack_failure_retries_ack_without_handler_replay( mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-backlog-ack-retry" in ctx._processed_ids - assert "msg-backlog-ack-retry" not in ctx._processed_ack_pending_ids + assert "msg-backlog-ack-retry" in ctx.claims.completed_ids + assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-backlog-ack-retry") async def test_processed_ack_retry_budget_exhaustion_keeps_local_completion( self, mock_link_with_next, mock_handler @@ -1070,8 +1074,8 @@ async def test_processed_ack_retry_budget_exhaustion_keeps_local_completion( mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-ack-budget" in ctx._processed_ids - assert "msg-ack-budget" not in ctx._processed_ack_pending_ids + assert "msg-ack-budget" in ctx.claims.completed_ids + assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-ack-budget") async def test_websocket_processed_ack_failure_retries_ack_without_handler_replay( self, mock_link_with_next, mock_handler @@ -1092,8 +1096,8 @@ async def test_websocket_processed_ack_failure_retries_ack_without_handler_repla mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-ws-ack-retry" in ctx._processed_ids - assert "msg-ws-ack-retry" not in ctx._processed_ack_pending_ids + assert "msg-ws-ack-retry" in ctx.claims.completed_ids + assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-ws-ack-retry") async def test_ack_pending_message_survives_lru_pressure_without_replay( self, mock_link_with_next, mock_handler @@ -1128,7 +1132,7 @@ async def deliver(message_id: str) -> None: # Handler completes but the durable ack fails → ACK_PENDING. await deliver("msg-unacked") # Enough later completions to overflow the dedupe cache capacity. - for i in range(ctx._max_processed_ids): + for i in range(ctx.claims.max_completed): await deliver(f"msg-filler-{i}") executions_before_redelivery = mock_handler.await_count @@ -1165,9 +1169,9 @@ async def test_websocket_processed_ack_failure_retries_ack_before_newer_queue( "msg-new-ws", ] assert mock_link_with_next.mark_processed.await_count == 3 - assert "msg-old-ws" in ctx._processed_ids - assert "msg-new-ws" in ctx._processed_ids - assert ctx._processed_ack_pending_ids == {} + assert "msg-old-ws" in ctx.claims.completed_ids + assert "msg-new-ws" in ctx.claims.completed_ids + assert ctx.claims.pending_ack_ids(ctx.room_id) == [] await ctx.stop() @@ -1205,7 +1209,7 @@ async def test_sync_point_claim_failure_does_not_clear_marker( assert ctx._first_ws_msg_id == "msg-sync-claim-fails" mock_handler.assert_not_called() - assert "msg-sync-claim-fails" not in ctx._processed_ids + assert "msg-sync-claim-fails" not in ctx.claims.completed_ids await ctx.stop() @@ -1244,7 +1248,7 @@ async def test_startup_backlog_claim_failure_does_not_spin( "room-123", "msg-startup-claim-fails" ) mock_handler.assert_not_called() - assert "msg-startup-claim-fails" not in ctx._processed_ids + assert "msg-startup-claim-fails" not in ctx.claims.completed_ids async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler @@ -1324,7 +1328,7 @@ async def test_resync_claim_failure_does_not_spin( "room-123", "msg-resync-claim-fails" ) mock_handler.assert_not_called() - assert "msg-resync-claim-fails" not in ctx._processed_ids + assert "msg-resync-claim-fails" not in ctx.claims.completed_ids async def test_resync_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler @@ -1367,7 +1371,7 @@ async def test_resync_claim_failure_does_not_process_newer_ws_event( "room-123", "msg-resync-older-claim-fails" ) mock_handler.assert_not_called() - assert "msg-resync-newer-ws" not in ctx._processed_ids + assert "msg-resync-newer-ws" not in ctx.claims.completed_ids await ctx.stop() @@ -1544,7 +1548,7 @@ async def slow_handler(ctx, event): assert elapsed < 1.0, f"stop() took {elapsed}s - should cancel processing" # Cancellation must release the local in-flight claim - assert ctx._inflight_message_ids == set() + assert ctx.claims.inflight_ids == set() class TestContextHydrationConfig: diff --git a/tests/runtime/test_resync.py b/tests/runtime/test_resync.py index 549eb6c18..51d9f26be 100644 --- a/tests/runtime/test_resync.py +++ b/tests/runtime/test_resync.py @@ -245,13 +245,13 @@ async def test_processes_single_missed_message(self, mock_link, mock_handler): await ctx.stop() async def test_skips_duplicate_message(self, mock_link, mock_handler): - """A message whose ID is already in _processed_ids should be skipped.""" + """A message already recorded as completed should be skipped.""" msg = make_platform_message(msg_id="dup-1", room_id="room-1") mock_link.get_next_message.side_effect = [msg, None] ctx = ExecutionContext("room-1", mock_link, mock_handler) # Mark the message as already processed - ctx._processed_ids["dup-1"] = True + ctx.claims.remember_completed(ctx.room_id, "dup-1") await ctx._resync_pending_messages() diff --git a/tests/runtime/test_runtime.py b/tests/runtime/test_runtime.py index 40a9f9705..f036312e2 100644 --- a/tests/runtime/test_runtime.py +++ b/tests/runtime/test_runtime.py @@ -185,6 +185,40 @@ async def test_execution_idempotent(self, mock_link, mock_handler): await runtime.stop() + async def test_all_contexts_share_one_claim_registry(self, mock_link, mock_handler): + """Message ownership must be shared across every context the runtime + creates — including a context recreated after room removal — while + retaining room-local ownership state.""" + runtime = AgentRuntime(mock_link, "agent-123", mock_handler) + + room_a = await runtime._create_execution("room-a") + room_b = await runtime._create_execution("room-b") + await runtime._on_room_left("room-a") + room_a_again = await runtime._create_execution("room-a") + + assert room_a.claims is room_b.claims + assert room_a_again.claims is room_a.claims + + await runtime.stop() + + async def test_pending_ack_is_retried_only_by_its_room( + self, mock_link, mock_handler + ): + """A shared registry must not acknowledge a different room's message.""" + mock_link.mark_processed = AsyncMock(return_value=True) + runtime = AgentRuntime(mock_link, "agent-123", mock_handler) + room_a = await runtime._create_execution("room-a") + room_b = await runtime._create_execution("room-b") + room_a.claims.remember_ack_pending("room-a", "msg-1") + + assert await room_b._retry_pending_processed_acks() + mock_link.mark_processed.assert_not_awaited() + + assert await room_a._retry_pending_processed_acks() + mock_link.mark_processed.assert_awaited_once_with("room-a", "msg-1") + + await runtime.stop() + async def test_active_sessions_returns_copy(self, mock_link, mock_handler): """active_sessions should return a copy.""" runtime = AgentRuntime(mock_link, "agent-123", mock_handler) diff --git a/tests/test_session_sync.py b/tests/test_session_sync.py index 7c9a435b5..1fb48173b 100644 --- a/tests/test_session_sync.py +++ b/tests/test_session_sync.py @@ -135,7 +135,7 @@ async def test_processed_event_added_to_cache(self, ctx): await ctx._process_event(event) - assert "msg-001" in ctx._processed_ids + assert "msg-001" in ctx.claims.completed_ids @pytest.mark.asyncio async def test_duplicate_event_skipped(self, ctx): @@ -153,26 +153,26 @@ async def test_duplicate_event_skipped(self, ctx): @pytest.mark.asyncio async def test_lru_cache_evicts_oldest(self, ctx): """LRU cache should evict oldest entries when full.""" - limit = ctx._max_processed_ids + limit = ctx.claims.max_completed for i in range(limit): event = make_message_event(msg_id=f"msg-{i:03d}") await ctx._process_event(event) - assert len(ctx._processed_ids) == limit - assert "msg-000" in ctx._processed_ids + assert len(ctx.claims.completed_ids) == limit + assert "msg-000" in ctx.claims.completed_ids event_overflow = make_message_event(msg_id=f"msg-{limit:03d}") await ctx._process_event(event_overflow) - assert len(ctx._processed_ids) == limit - assert "msg-000" not in ctx._processed_ids - assert f"msg-{limit:03d}" in ctx._processed_ids + assert len(ctx.claims.completed_ids) == limit + assert "msg-000" not in ctx.claims.completed_ids + assert f"msg-{limit:03d}" in ctx.claims.completed_ids @pytest.mark.asyncio async def test_duplicate_refreshes_lru_position(self, ctx): """Accessing duplicate should refresh its LRU position.""" - limit = ctx._max_processed_ids + limit = ctx.claims.max_completed for i in range(limit): event = make_message_event(msg_id=f"msg-{i:03d}") @@ -184,8 +184,8 @@ async def test_duplicate_refreshes_lru_position(self, ctx): event_overflow = make_message_event(msg_id=f"msg-{limit:03d}") await ctx._process_event(event_overflow) - assert "msg-000" in ctx._processed_ids - assert "msg-001" not in ctx._processed_ids + assert "msg-000" in ctx.claims.completed_ids + assert "msg-001" not in ctx.claims.completed_ids class TestSynchronizeWithNext: @@ -251,7 +251,7 @@ async def test_sync_point_reached_clears_marker_and_keeps_cache( # Queue should be empty (duplicate removed) assert ctx.queue.empty() # Dedupe cache keeps processed IDs to avoid immediate WS reprocessing - assert "sync-001" in ctx._processed_ids + assert "sync-001" in ctx.claims.completed_ids @pytest.mark.asyncio async def test_sync_point_with_non_message_head_processes_once( From 933dc7e78f6f112f973bd9abae3d5eb59118311f Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 11:36:44 +0300 Subject: [PATCH 5/8] fix(runtime): preserve claim diagnostics type --- src/band/runtime/claims.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/band/runtime/claims.py b/src/band/runtime/claims.py index ab17700a3..e730acc40 100644 --- a/src/band/runtime/claims.py +++ b/src/band/runtime/claims.py @@ -61,7 +61,7 @@ def release(self, room_id: str, message_id: str) -> None: @property def inflight_ids(self) -> set[str]: """Currently claimed message IDs (copy).""" - return set(self._inflight) + return {message_id for _, message_id in self._inflight} def is_completed(self, room_id: str, message_id: str) -> bool: """Whether the message already completed; a hit refreshes LRU recency.""" From 12d1e399a3ea7cb55f36512a9638718fcaa3ec53 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 11:41:02 +0300 Subject: [PATCH 6/8] style(runtime): use absolute runtime imports --- src/band/runtime/execution.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 3cd8fbd20..c067e6e5a 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -47,10 +47,10 @@ SYNTHETIC_SENDER_TYPE, SYNTHETIC_CONTACT_EVENTS_SENDER_ID, ) -from .claims import MessageClaimRegistry -from .retry_tracker import MessageRetryTracker +from band.runtime.claims import MessageClaimRegistry from band.runtime.context_serialization import context_item_to_dict -from .working_state import WorkingStateReporter +from band.runtime.retry_tracker import MessageRetryTracker +from band.runtime.working_state import WorkingStateReporter if TYPE_CHECKING: from band.platform.link import BandLink From 2bf7e43f7d1d383b2c3f650ac4980611f9cac8ca Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 11:54:54 +0300 Subject: [PATCH 7/8] refactor(runtime): strengthen message claim lifecycle --- src/band/runtime/claims.py | 52 ++++++++++------ src/band/runtime/execution.py | 106 ++++++++++++++++++-------------- src/band/runtime/runtime.py | 4 +- tests/runtime/test_claims.py | 33 ++++++++++ tests/runtime/test_execution.py | 67 ++++++++++++++------ tests/test_session_sync.py | 18 +++--- 6 files changed, 185 insertions(+), 95 deletions(-) create mode 100644 tests/runtime/test_claims.py diff --git a/src/band/runtime/claims.py b/src/band/runtime/claims.py index e730acc40..86fb27b81 100644 --- a/src/band/runtime/claims.py +++ b/src/band/runtime/claims.py @@ -3,7 +3,8 @@ from __future__ import annotations from collections import OrderedDict -from typing import TypeAlias +from contextlib import contextmanager +from typing import Iterator, TypeAlias DEFAULT_COMPLETED_CACHE_SIZE = 500 ClaimKey: TypeAlias = tuple[str, str] @@ -44,9 +45,9 @@ def __init__(self, max_completed: int = DEFAULT_COMPLETED_CACHE_SIZE) -> None: self._inflight: set[ClaimKey] = set() self._ack_pending: OrderedDict[ClaimKey, bool] = OrderedDict() self._ack_retries: dict[ClaimKey, int] = {} - self._completed: OrderedDict[ClaimKey, bool] = OrderedDict() + self._completed_by_room: dict[str, OrderedDict[str, bool]] = {} - def try_claim(self, room_id: str, message_id: str) -> bool: + def _try_claim(self, room_id: str, message_id: str) -> bool: """Claim a message for one execution; False if another owner holds it.""" key = (room_id, message_id) if key in self._inflight: @@ -54,37 +55,50 @@ def try_claim(self, room_id: str, message_id: str) -> bool: self._inflight.add(key) return True - def release(self, room_id: str, message_id: str) -> None: + def _release(self, room_id: str, message_id: str) -> None: """Release an in-flight claim (failure, cancellation, or completion).""" self._inflight.discard((room_id, message_id)) - @property - def inflight_ids(self) -> set[str]: - """Currently claimed message IDs (copy).""" - return {message_id for _, message_id in self._inflight} + @contextmanager + def claim(self, room_id: str, message_id: str) -> Iterator[bool]: + """Yield whether a claim was acquired and always release acquired claims. + + Exceptions propagate to the delivery path's existing error handling; + this context only guarantees cleanup. + """ + acquired = self._try_claim(room_id, message_id) + try: + yield acquired + finally: + if acquired: + self._release(room_id, message_id) + + def inflight_ids(self, room_id: str) -> set[str]: + """Currently claimed message IDs for a room (copy).""" + return {message_id for room, message_id in self._inflight if room == room_id} def is_completed(self, room_id: str, message_id: str) -> bool: """Whether the message already completed; a hit refreshes LRU recency.""" - key = (room_id, message_id) - if key not in self._completed: + completed = self._completed_by_room.get(room_id) + if completed is None or message_id not in completed: return False - self._completed.move_to_end(key) + completed.move_to_end(message_id) return True - @property - def completed_ids(self) -> list[str]: - """Completed message IDs, oldest first (copy, no recency side effect).""" - return [message_id for _, message_id in self._completed] + def completed_ids(self, room_id: str) -> list[str]: + """Completed message IDs for a room, oldest first.""" + return list(self._completed_by_room.get(room_id, ())) def remember_completed(self, room_id: str, message_id: str) -> None: """Record durable completion and clear any pending-ack state.""" + completed = self._completed_by_room.setdefault(room_id, OrderedDict()) + completed[message_id] = True + completed.move_to_end(message_id) key = (room_id, message_id) - self._completed[key] = True - self._completed.move_to_end(key) self._ack_pending.pop(key, None) self._ack_retries.pop(key, None) - if len(self._completed) > self.max_completed: - self._completed.popitem(last=False) + if len(completed) > self.max_completed: + completed.popitem(last=False) def is_ack_pending(self, room_id: str, message_id: str) -> bool: """Whether the message completed locally but lacks a durable ack.""" diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index c067e6e5a..7e97c5d70 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -17,13 +17,12 @@ import asyncio import logging from datetime import datetime, timezone -from enum import Enum +from enum import Enum, StrEnum from typing import ( TYPE_CHECKING, Any, Awaitable, Callable, - Literal, Protocol, runtime_checkable, ) @@ -73,6 +72,14 @@ class _BacklogProcessResult(Enum): RETRY_LATER = "retry_later" +class ExecutionState(StrEnum): + """Lifecycle state for one room execution.""" + + STARTING = "starting" + IDLE = "idle" + PROCESSING = "processing" + + def _error_label(e: Exception) -> str: """Return a non-empty label for an exception, falling back to the class name.""" return str(e).strip() or type(e).__name__ @@ -206,8 +213,8 @@ def __init__( schema methods can auto-enable contact tools when this context belongs to the hub room. claim_registry: Optional shared message-claim registry. AgentRuntime - passes one registry to all contexts it creates so a message ID - executes at most once per runtime. Defaults to a private + passes one registry to its default contexts so a room/message + pair executes at most once per runtime. Defaults to a private instance for standalone contexts. """ self.room_id = room_id @@ -233,7 +240,7 @@ def __init__( # Per-room state self.queue: asyncio.Queue[PlatformEvent] = asyncio.Queue() - self.state: Literal["starting", "idle", "processing"] = "starting" + self.state = ExecutionState.STARTING self._is_running = False self._process_loop_task: asyncio.Task[None] | None = None self._context_cache: ConversationContext | None = None @@ -276,9 +283,9 @@ def thread_id(self) -> str: @property def is_processing(self) -> bool: """Check if context is currently processing an event.""" - return self.state == "processing" + return self.state is ExecutionState.PROCESSING - def _set_state(self, new_state: Literal["starting", "idle", "processing"]) -> None: + def _set_state(self, new_state: ExecutionState) -> None: """ Set the execution state and update the idle event accordingly. @@ -286,7 +293,7 @@ def _set_state(self, new_state: Literal["starting", "idle", "processing"]) -> No for graceful shutdown coordination. """ self.state = new_state - if new_state == "processing": + if new_state is ExecutionState.PROCESSING: self._idle_event.clear() else: self._idle_event.set() @@ -863,10 +870,10 @@ async def _process_loop(self) -> None: # If a pending message cannot be claimed yet, stay in startup sync # instead of processing newer WebSocket events out of order. while not await self._synchronize_with_next(): - self._set_state("idle") + self._set_state(ExecutionState.IDLE) await asyncio.sleep(self.config.idle_resync_seconds) - self._set_state("idle") + self._set_state(ExecutionState.IDLE) logger.info( "ExecutionContext %s: Synchronized, switching to WebSocket", self.room_id, @@ -931,7 +938,7 @@ async def _wait_until_resync_complete(self) -> None: and await self._resync_pending_messages() ): return - self._set_state("idle") + self._set_state(ExecutionState.IDLE) await asyncio.sleep(self.config.idle_resync_seconds) async def _synchronize_with_next(self) -> bool: @@ -1176,11 +1183,18 @@ async def _process_backlog_message( return _BacklogProcessResult.ADVANCED return _BacklogProcessResult.RETRY_LATER - if not self.claims.try_claim(self.room_id, msg_id): - logger.debug("Skipping already in-flight backlog message: %s", msg_id) - return _BacklogProcessResult.RETRY_LATER + with self.claims.claim(self.room_id, msg_id) as acquired: + if not acquired: + logger.debug("Deferring in-flight backlog message: %s", msg_id) + return _BacklogProcessResult.RETRY_LATER + return await self._process_claimed_backlog_message(msg) - self._set_state("processing") + async def _process_claimed_backlog_message( + self, msg: PlatformMessage + ) -> _BacklogProcessResult: + """Process a backlog message while its in-flight claim is held.""" + msg_id = msg.id + self._set_state(ExecutionState.PROCESSING) logger.info("Processing backlog message %s in room %s", msg_id, self.room_id) try: @@ -1300,8 +1314,7 @@ async def _process_backlog_message( return _BacklogProcessResult.ADVANCED finally: - self.claims.release(self.room_id, msg_id) - self._set_state("idle") + self._set_state(ExecutionState.IDLE) def _drain_duplicate_from_queue(self, msg_id: str) -> None: """ @@ -1343,7 +1356,7 @@ async def _execute_message_cycle(self, event: PlatformEvent) -> None: ``start()`` emits working:true (+ keep-alive); ``stop()`` in the finally emits the authoritative working:false on success, exception, and cancel — - mirroring the ``_set_state('idle')`` placement. The reporter is a no-op + mirroring the final idle-state transition. The reporter is a no-op when disabled or for the hub room, so call sites need no extra gating. """ await self._working_reporter.start() @@ -1364,24 +1377,22 @@ async def _process_event(self, event: PlatformEvent) -> bool: 5. Mark as processed (success) or failed (exception) """ if isinstance(event, ReconnectedEvent): - self._set_state("processing") + self._set_state(ExecutionState.PROCESSING) logger.debug("Processing %s in room %s", event.type, self.room_id) try: if self._reconnect_sync_requested: self._reconnect_sync_requested = False while not await self._synchronize_with_next(): - self._set_state("idle") + self._set_state(ExecutionState.IDLE) await asyncio.sleep(self.config.idle_resync_seconds) logger.debug("Event %s processed successfully", event.type) finally: - self._set_state("idle") + self._set_state(ExecutionState.IDLE) return True payload = event.payload if isinstance(event, MessageEvent) else None msg_id = payload.id if payload else None - claimed_msg_id: str | None = None - # For messages: check if we should skip if isinstance(event, MessageEvent) and msg_id and payload: # Skip messages from self (agent's own messages) to avoid infinite loops @@ -1421,27 +1432,34 @@ async def _process_event(self, event: PlatformEvent) -> bool: return True return False - if not self.claims.try_claim(self.room_id, msg_id): - # Another execution owns this message. Defer rather than - # treat it as handled: the resync safety net re-checks it, - # so an owner failure never silently loses the message. - logger.debug( - "Message %s owned by another execution; deferring", msg_id - ) - return False - claimed_msg_id = msg_id + with self.claims.claim(self.room_id, msg_id) as acquired: + if not acquired: + # The resync safety net re-checks deferred work, so an + # owner failure never silently loses the message. + logger.debug( + "Message %s owned by another execution; deferring", + msg_id, + ) + return False + return await self._process_event_body(event, msg_id, payload) - if self._message_processed_for_agent(msg_id, payload.metadata): - logger.info( - "Skipping processed replay message %s in room %s", - msg_id, - self.room_id, - ) - self.claims.remember_completed(self.room_id, msg_id) - self.claims.release(self.room_id, msg_id) - return True + return await self._process_event_body(event, msg_id, payload) + + async def _process_event_body( + self, event: PlatformEvent, msg_id: str | None, payload: Any + ) -> bool: + """Process an event after any required in-flight claim is acquired.""" + if isinstance(event, MessageEvent) and msg_id and payload: + if self._message_processed_for_agent(msg_id, payload.metadata): + logger.info( + "Skipping processed replay message %s in room %s", + msg_id, + self.room_id, + ) + self.claims.remember_completed(self.room_id, msg_id) + return True - self._set_state("processing") + self._set_state(ExecutionState.PROCESSING) logger.debug("Processing %s in room %s", event.type, self.room_id) try: @@ -1531,6 +1549,4 @@ async def _process_event(self, event: PlatformEvent) -> bool: return True finally: - if claimed_msg_id: - self.claims.release(self.room_id, claimed_msg_id) - self._set_state("idle") + self._set_state(ExecutionState.IDLE) diff --git a/src/band/runtime/runtime.py b/src/band/runtime/runtime.py index f7babad54..f6925835c 100644 --- a/src/band/runtime/runtime.py +++ b/src/band/runtime/runtime.py @@ -132,8 +132,8 @@ def __init__( # Per-room executions self.executions: dict[str, Execution] = {} - # Shared by all contexts this runtime creates, so a message ID - # executes at most once per runtime even across context recreation. + # Shared by default contexts so a room/message pair executes at most + # once per runtime, including across context recreation. self._claim_registry = MessageClaimRegistry() # Set up presence callbacks diff --git a/tests/runtime/test_claims.py b/tests/runtime/test_claims.py new file mode 100644 index 000000000..84f6f263f --- /dev/null +++ b/tests/runtime/test_claims.py @@ -0,0 +1,33 @@ +"""Tests for the shared inbound message claim registry.""" + +from __future__ import annotations + +from band.runtime.claims import MessageClaimRegistry + + +def test_claim_reports_contention_without_releasing_owner() -> None: + registry = MessageClaimRegistry() + + with registry.claim("room-1", "message-1") as owner: + assert owner + with registry.claim("room-1", "message-1") as contender: + assert not contender + assert registry.inflight_ids("room-1") == {"message-1"} + + +def test_same_message_id_is_independent_between_rooms() -> None: + registry = MessageClaimRegistry() + + with registry.claim("room-1", "message-1") as room_one: + with registry.claim("room-2", "message-1") as room_two: + assert room_one and room_two + + +def test_completed_cache_pressure_is_isolated_per_room() -> None: + registry = MessageClaimRegistry(max_completed=1) + registry.remember_completed("room-1", "old") + registry.remember_completed("room-2", "kept") + registry.remember_completed("room-1", "new") + + assert registry.completed_ids("room-1") == ["new"] + assert registry.completed_ids("room-2") == ["kept"] diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index ff4b3a478..d578aaffe 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -12,6 +12,7 @@ from band.runtime.execution import ( Execution, ExecutionContext, + ExecutionState, _BacklogProcessResult, _error_label, ) @@ -88,7 +89,7 @@ def test_init_starts_idle(self, mock_link, mock_handler): """Should start in starting state, not running.""" ctx = ExecutionContext("room-123", mock_link, mock_handler) - assert ctx.state == "starting" + assert ctx.state is ExecutionState.STARTING assert ctx.is_running is False assert ctx.is_processing is False @@ -554,7 +555,7 @@ async def test_sync_point_clears_marker_and_keeps_dedupe_cache( # Marker should be cleared assert ctx._first_ws_msg_id is None # Dedupe cache should keep processed sync id to avoid WS reprocessing - assert "msg-sync-001" in ctx.claims.completed_ids + assert "msg-sync-001" in ctx.claims.completed_ids(ctx.room_id) await ctx.stop() @@ -662,7 +663,7 @@ async def test_ws_replay_with_processed_metadata_is_not_reopened( mock_handler.assert_not_called() mock_link_with_next.mark_processing.assert_not_called() - assert "msg-processed-replay" in ctx.claims.completed_ids + assert "msg-processed-replay" in ctx.claims.completed_ids(ctx.room_id) await ctx.stop() @@ -701,7 +702,7 @@ async def test_ws_replay_uses_hydrated_context_delivery_status( mock_handler.assert_not_called() mock_link_with_next.mark_processing.assert_not_called() - assert "msg-stale-replay" in ctx.claims.completed_ids + assert "msg-stale-replay" in ctx.claims.completed_ids(ctx.room_id) await ctx.stop() @@ -812,7 +813,7 @@ async def delayed_mark_processing(room_id: str, message_id: str) -> bool: await backlog_task assert mock_handler.await_count == 1 - assert ctx.claims.inflight_ids == set() + assert ctx.claims.inflight_ids(ctx.room_id) == set() async def test_first_message_to_fresh_room_executes_once(self, mock_link_with_next): """A message posted before the room's context was live executes once. @@ -941,7 +942,32 @@ async def test_mark_processing_failure_does_not_execute_message( ) mock_handler.assert_not_called() mock_link_with_next.mark_processed.assert_not_called() - assert ctx.claims.inflight_ids == set() + assert ctx.claims.inflight_ids(ctx.room_id) == set() + + async def test_handler_failure_marks_failed_and_releases_claim( + self, mock_link_with_next + ): + """A real handler failure must not strand ownership of the message.""" + + async def failing_handler(ctx, event): + raise RuntimeError("handler failed") + + mock_link_with_next.mark_processing = AsyncMock(return_value=True) + mock_link_with_next.mark_failed = AsyncMock(return_value=True) + ctx = ExecutionContext( + "room-123", + mock_link_with_next, + failing_handler, + config=SessionConfig(enable_context_hydration=False), + ) + event = make_message_event(room_id="room-123", msg_id="msg-handler-fails") + + assert await ctx._process_event(event) is True + + mock_link_with_next.mark_failed.assert_awaited_once_with( + "room-123", "msg-handler-fails", "handler failed" + ) + assert ctx.claims.inflight_ids(ctx.room_id) == set() async def test_backlog_processed_ack_failure_is_not_remembered( self, mock_link_with_next, mock_handler @@ -976,8 +1002,9 @@ async def test_backlog_processed_ack_failure_is_not_remembered( mock_link_with_next.mark_processed.assert_awaited_once_with( "room-123", "msg-ack-fails" ) - assert "msg-ack-fails" not in ctx.claims.completed_ids + assert "msg-ack-fails" not in ctx.claims.completed_ids(ctx.room_id) assert ctx.claims.is_ack_pending(ctx.room_id, "msg-ack-fails") + assert ctx.claims.inflight_ids(ctx.room_id) == set() async def test_websocket_processed_ack_failure_is_not_remembered( self, mock_link_with_next, mock_handler @@ -999,7 +1026,7 @@ async def test_websocket_processed_ack_failure_is_not_remembered( mock_link_with_next.mark_processed.assert_awaited_once_with( "room-123", "msg-ws-ack-fails" ) - assert "msg-ws-ack-fails" not in ctx.claims.completed_ids + assert "msg-ws-ack-fails" not in ctx.claims.completed_ids(ctx.room_id) assert ctx.claims.is_ack_pending(ctx.room_id, "msg-ws-ack-fails") async def test_backlog_processed_ack_failure_retries_ack_without_handler_replay( @@ -1035,7 +1062,7 @@ async def test_backlog_processed_ack_failure_retries_ack_without_handler_replay( mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-backlog-ack-retry" in ctx.claims.completed_ids + assert "msg-backlog-ack-retry" in ctx.claims.completed_ids(ctx.room_id) assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-backlog-ack-retry") async def test_processed_ack_retry_budget_exhaustion_keeps_local_completion( @@ -1074,7 +1101,7 @@ async def test_processed_ack_retry_budget_exhaustion_keeps_local_completion( mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-ack-budget" in ctx.claims.completed_ids + assert "msg-ack-budget" in ctx.claims.completed_ids(ctx.room_id) assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-ack-budget") async def test_websocket_processed_ack_failure_retries_ack_without_handler_replay( @@ -1096,7 +1123,7 @@ async def test_websocket_processed_ack_failure_retries_ack_without_handler_repla mock_handler.assert_awaited_once() assert mock_link_with_next.mark_processed.await_count == 2 - assert "msg-ws-ack-retry" in ctx.claims.completed_ids + assert "msg-ws-ack-retry" in ctx.claims.completed_ids(ctx.room_id) assert not ctx.claims.is_ack_pending(ctx.room_id, "msg-ws-ack-retry") async def test_ack_pending_message_survives_lru_pressure_without_replay( @@ -1169,8 +1196,8 @@ async def test_websocket_processed_ack_failure_retries_ack_before_newer_queue( "msg-new-ws", ] assert mock_link_with_next.mark_processed.await_count == 3 - assert "msg-old-ws" in ctx.claims.completed_ids - assert "msg-new-ws" in ctx.claims.completed_ids + assert "msg-old-ws" in ctx.claims.completed_ids(ctx.room_id) + assert "msg-new-ws" in ctx.claims.completed_ids(ctx.room_id) assert ctx.claims.pending_ack_ids(ctx.room_id) == [] await ctx.stop() @@ -1209,7 +1236,7 @@ async def test_sync_point_claim_failure_does_not_clear_marker( assert ctx._first_ws_msg_id == "msg-sync-claim-fails" mock_handler.assert_not_called() - assert "msg-sync-claim-fails" not in ctx.claims.completed_ids + assert "msg-sync-claim-fails" not in ctx.claims.completed_ids(ctx.room_id) await ctx.stop() @@ -1248,7 +1275,7 @@ async def test_startup_backlog_claim_failure_does_not_spin( "room-123", "msg-startup-claim-fails" ) mock_handler.assert_not_called() - assert "msg-startup-claim-fails" not in ctx.claims.completed_ids + assert "msg-startup-claim-fails" not in ctx.claims.completed_ids(ctx.room_id) async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler @@ -1328,7 +1355,7 @@ async def test_resync_claim_failure_does_not_spin( "room-123", "msg-resync-claim-fails" ) mock_handler.assert_not_called() - assert "msg-resync-claim-fails" not in ctx.claims.completed_ids + assert "msg-resync-claim-fails" not in ctx.claims.completed_ids(ctx.room_id) async def test_resync_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler @@ -1371,7 +1398,7 @@ async def test_resync_claim_failure_does_not_process_newer_ws_event( "room-123", "msg-resync-older-claim-fails" ) mock_handler.assert_not_called() - assert "msg-resync-newer-ws" not in ctx.claims.completed_ids + assert "msg-resync-newer-ws" not in ctx.claims.completed_ids(ctx.room_id) await ctx.stop() @@ -1548,7 +1575,7 @@ async def slow_handler(ctx, event): assert elapsed < 1.0, f"stop() took {elapsed}s - should cancel processing" # Cancellation must release the local in-flight claim - assert ctx.claims.inflight_ids == set() + assert ctx.claims.inflight_ids(ctx.room_id) == set() class TestContextHydrationConfig: @@ -1994,7 +2021,7 @@ async def test_wait_for_idle_returns_true_when_already_idle( ): """_wait_for_idle should return True immediately when idle.""" ctx = ExecutionContext("room-123", mock_link, mock_handler) - ctx.state = "idle" + ctx.state = ExecutionState.IDLE result = await ctx._wait_for_idle(timeout=1.0) @@ -2005,7 +2032,7 @@ async def test_wait_for_idle_returns_false_on_timeout( ): """_wait_for_idle should return False when timeout exceeded.""" ctx = ExecutionContext("room-123", mock_link, mock_handler) - ctx._set_state("processing") # Use _set_state to properly clear idle event + ctx._set_state(ExecutionState.PROCESSING) start = asyncio.get_running_loop().time() result = await ctx._wait_for_idle(timeout=0.1) diff --git a/tests/test_session_sync.py b/tests/test_session_sync.py index 1fb48173b..4332d4b6d 100644 --- a/tests/test_session_sync.py +++ b/tests/test_session_sync.py @@ -135,7 +135,7 @@ async def test_processed_event_added_to_cache(self, ctx): await ctx._process_event(event) - assert "msg-001" in ctx.claims.completed_ids + assert "msg-001" in ctx.claims.completed_ids(ctx.room_id) @pytest.mark.asyncio async def test_duplicate_event_skipped(self, ctx): @@ -159,15 +159,15 @@ async def test_lru_cache_evicts_oldest(self, ctx): event = make_message_event(msg_id=f"msg-{i:03d}") await ctx._process_event(event) - assert len(ctx.claims.completed_ids) == limit - assert "msg-000" in ctx.claims.completed_ids + assert len(ctx.claims.completed_ids(ctx.room_id)) == limit + assert "msg-000" in ctx.claims.completed_ids(ctx.room_id) event_overflow = make_message_event(msg_id=f"msg-{limit:03d}") await ctx._process_event(event_overflow) - assert len(ctx.claims.completed_ids) == limit - assert "msg-000" not in ctx.claims.completed_ids - assert f"msg-{limit:03d}" in ctx.claims.completed_ids + assert len(ctx.claims.completed_ids(ctx.room_id)) == limit + assert "msg-000" not in ctx.claims.completed_ids(ctx.room_id) + assert f"msg-{limit:03d}" in ctx.claims.completed_ids(ctx.room_id) @pytest.mark.asyncio async def test_duplicate_refreshes_lru_position(self, ctx): @@ -184,8 +184,8 @@ async def test_duplicate_refreshes_lru_position(self, ctx): event_overflow = make_message_event(msg_id=f"msg-{limit:03d}") await ctx._process_event(event_overflow) - assert "msg-000" in ctx.claims.completed_ids - assert "msg-001" not in ctx.claims.completed_ids + assert "msg-000" in ctx.claims.completed_ids(ctx.room_id) + assert "msg-001" not in ctx.claims.completed_ids(ctx.room_id) class TestSynchronizeWithNext: @@ -251,7 +251,7 @@ async def test_sync_point_reached_clears_marker_and_keeps_cache( # Queue should be empty (duplicate removed) assert ctx.queue.empty() # Dedupe cache keeps processed IDs to avoid immediate WS reprocessing - assert "sync-001" in ctx.claims.completed_ids + assert "sync-001" in ctx.claims.completed_ids(ctx.room_id) @pytest.mark.asyncio async def test_sync_point_with_non_message_head_processes_once( From 3ee1b4041e42c8122e12ddbb77be501dbdc55e8f Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Wed, 22 Jul 2026 12:14:48 +0300 Subject: [PATCH 8/8] fix(runtime): reclaim removed room claim cache --- src/band/runtime/claims.py | 10 ++++++++-- src/band/runtime/runtime.py | 5 +++++ tests/runtime/test_runtime.py | 20 ++++++++++++++++++++ 3 files changed, 33 insertions(+), 2 deletions(-) diff --git a/src/band/runtime/claims.py b/src/band/runtime/claims.py index 86fb27b81..49e8e6257 100644 --- a/src/band/runtime/claims.py +++ b/src/band/runtime/claims.py @@ -3,8 +3,9 @@ from __future__ import annotations from collections import OrderedDict +from collections.abc import Iterator from contextlib import contextmanager -from typing import Iterator, TypeAlias +from typing import TypeAlias DEFAULT_COMPLETED_CACHE_SIZE = 500 ClaimKey: TypeAlias = tuple[str, str] @@ -20,7 +21,8 @@ class MessageClaimRegistry: - **ack pending** — the handler completed but the durable processed ack failed; redelivery retries only the ack, never the handler. Never evicted (losing it would replay side effects); drains through the ack - retry budget instead. + retry budget instead. If a room never returns, this state intentionally + remains for the registry's lifetime rather than risk replaying effects. - **completed** — durably processed; kept in a bounded LRU for dedup. ``AgentRuntime`` owns one registry and passes it to every context it @@ -89,6 +91,10 @@ def completed_ids(self, room_id: str) -> list[str]: """Completed message IDs for a room, oldest first.""" return list(self._completed_by_room.get(room_id, ())) + def discard_completed(self, room_id: str) -> None: + """Discard a removed room's durably acknowledged local cache.""" + self._completed_by_room.pop(room_id, None) + def remember_completed(self, room_id: str, message_id: str) -> None: """Record durable completion and clear any pending-ack state.""" completed = self._completed_by_room.setdefault(room_id, OrderedDict()) diff --git a/src/band/runtime/runtime.py b/src/band/runtime/runtime.py index f6925835c..ab5262862 100644 --- a/src/band/runtime/runtime.py +++ b/src/band/runtime/runtime.py @@ -306,6 +306,11 @@ async def _destroy_execution( execution = self.executions.pop(room_id) graceful = await execution.stop(timeout=timeout) + # Durable completion state is safe to release with the room. Pending + # acknowledgements remain in the shared registry so a later rejoin + # retries only the ack instead of replaying handler side effects. + self._claim_registry.discard_completed(room_id) + # Call cleanup callback (for adapter to clean up checkpointer, etc.) if self._on_session_cleanup: try: diff --git a/tests/runtime/test_runtime.py b/tests/runtime/test_runtime.py index f036312e2..7c0eaecfe 100644 --- a/tests/runtime/test_runtime.py +++ b/tests/runtime/test_runtime.py @@ -219,6 +219,26 @@ async def test_pending_ack_is_retried_only_by_its_room( await runtime.stop() + async def test_room_recreation_reclaims_completed_and_retries_pending_ack( + self, mock_link, mock_handler + ): + """Room teardown may drop durable cache, but not unacknowledged success.""" + runtime = AgentRuntime(mock_link, "agent-123", mock_handler) + original = await runtime._create_execution("room-a") + original.claims.remember_completed("room-a", "msg-completed") + original.claims.remember_ack_pending("room-a", "msg-ack-pending") + + await runtime._on_room_left("room-a") + recreated = await runtime._create_execution("room-a") + + assert not recreated.claims.is_completed("room-a", "msg-completed") + + mock_link.mark_processed = AsyncMock(return_value=True) + assert await recreated._retry_pending_processed_acks() + mock_link.mark_processed.assert_awaited_once_with("room-a", "msg-ack-pending") + + await runtime.stop() + async def test_active_sessions_returns_copy(self, mock_link, mock_handler): """active_sessions should return a copy.""" runtime = AgentRuntime(mock_link, "agent-123", mock_handler)