diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c6763b8aa..e18a797c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -256,7 +256,16 @@ jobs: /tmp/test-install/bin/python <<'PYEOF' from band import Agent, BandLink, AgentRuntime from band.config import load_agent_config - from band_sdk_core import ClaimRegistry, ParticipantRoster, RetryTracker + from band_sdk_core import ( + ClaimRegistry, + ParticipantRoster, + RetryTracker, + evaluate_adapter_result, + evaluate_delivery_event, + evaluate_drain_candidate, + evaluate_next_message, + is_self_echo, + ) # Prove the pinned band_sdk_core wheel is callable, not just importable, # from this isolated install -- not only from the dev checkout via uv sync. @@ -266,6 +275,18 @@ jobs: assert tracker.max_retries == 1 roster = ParticipantRoster() assert roster.list() == [] + assert is_self_echo("agent-1", "Agent", "agent-1") is True + assert evaluate_drain_candidate(None, [], "agent-1") == {"decision": "no_candidate"} + assert evaluate_delivery_event("room_added", "room-1", {}, "agent-1") == { + "decision": "ignored", + "event_type": "room_added", + } + assert evaluate_next_message("msg-1", None) == {"decision": "no_pending"} + assert evaluate_adapter_result("room-1", "msg-1", True) == { + "decision": "processed", + "room_id": "room-1", + "message_id": "msg-1", + } print('Core imports successful') PYEOF diff --git a/AGENTS.md b/AGENTS.md index ec1992266..96ea0dd19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,8 +42,11 @@ validation; build a `kwargs` dict and omit the key instead. See The SDK subscribes to Phoenix Channels (agent/chat/user rooms, participants, tasks) and hydrates each event's payload into a typed, rule-free `WirePayload` projection without re-validating. See -[docs/websocket-events.md](docs/websocket-events.md) for the channel table -and payload field reference. +[docs/websocket-events.md](docs/websocket-events.md) for the channel table, +payload field reference, and `band-sdk-core`'s delivery-lifecycle decisions. +Whenever code here starts calling a `band_sdk_core` symbol it didn't use +before, extend `.github/workflows/ci.yml`'s wheel-smoke step to prove that +symbol is callable from the isolated pinned wheel, not just importable. ## Contact Event Handling diff --git a/docs/websocket-events.md b/docs/websocket-events.md index 013497836..cf85d976e 100644 --- a/docs/websocket-events.md +++ b/docs/websocket-events.md @@ -22,6 +22,18 @@ re-validating by `WirePayload.from_wire` (`src/band/client/streaming/wire.py`). Every model inherits `WirePayload`, which sets `ConfigDict(extra="allow")` once for all of them. +`band-sdk-core` is also where the one-shot delivery-lifecycle *decisions* +live — `evaluate_delivery_event`/`evaluate_next_message`/ +`evaluate_drain_candidate`/`evaluate_adapter_result` — not just payload +validation. `OneShotInvoker` (`src/band/runtime/oneshot.py`) is a thin +caller-owns-the-loop wrapper around those four functions: the +ignore/cleanup/self-echo/invocation routing, the drain-candidate +classification, and the ack decision are core's, not the SDK's own logic. +`ExecutionContext` is a different machine with its own dedup model +(`metadata.delivery_status`) and does not call `evaluate_delivery_event`/ +`evaluate_next_message`/`evaluate_drain_candidate` — it shares only the +`is_self_echo` predicate with `OneShotInvoker`. + ```python notest MessageCreatedPayload: id, content, message_type, sender_id, sender_type, diff --git a/examples/agentcore/ARCHITECTURE.md b/examples/agentcore/ARCHITECTURE.md index 2632e70a4..7b1b7f034 100644 --- a/examples/agentcore/ARCHITECTURE.md +++ b/examples/agentcore/ARCHITECTURE.md @@ -158,7 +158,7 @@ Inside `OneShotInvoker.handle_event`, each invocation: When weather invocation B starts (after A finishes), B's `get_next_message` returns `204 No Content` — A drained it. B exits with -`{"status": "no_pending"}` without an LLM call. +`{"status": "no_pending"}` (`OneShotStatus.NO_PENDING`) without an LLM call. This is the same in-band claim/process semantics the SDK's `ExecutionContext` uses in the normal long-running Agent flow; diff --git a/pyproject.toml b/pyproject.toml index 6a151f8b1..735e61f03 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ dependencies = [ "band-client-rest==0.0.27", - "band-sdk-core==2.0.0", + "band-sdk-core==2.2.0", "phoenix-channels-python-client>=0.2.4", "python-dotenv>=1.2.2", "pydantic>=2.0", diff --git a/src/band/runtime/__init__.py b/src/band/runtime/__init__.py index 5e8ba7311..85157e0c6 100644 --- a/src/band/runtime/__init__.py +++ b/src/band/runtime/__init__.py @@ -31,7 +31,7 @@ from .presence import RoomPresence from .execution import Execution, ExecutionContext, ExecutionHandler from .runtime import AgentRuntime -from .oneshot import OneShotInvoker, OneShotEnvelopeError +from .oneshot import OneShotEnvelopeError, OneShotInvoker, OneShotStatus # Tools from .tools import ( @@ -73,6 +73,7 @@ "AgentRuntime", "OneShotInvoker", "OneShotEnvelopeError", + "OneShotStatus", # Tools "AgentTools", "HumanTools", diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 151d26a8c..edd2efbd9 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -29,7 +29,7 @@ runtime_checkable, ) -from band_sdk_core import ClaimRegistry, ParticipantRoster, RetryTracker +from band_sdk_core import ClaimRegistry, ParticipantRoster, RetryTracker, is_self_echo from band.client.rest import DEFAULT_REQUEST_OPTIONS from band.client.streaming import ControlMode, DeliveryStatus @@ -1414,11 +1414,15 @@ async def _process_backlog_message( """ msg_id = msg.id - # Skip messages from self (agent's own messages) to avoid infinite loops - if ( - self._agent_id - and msg.sender_type == "Agent" - and msg.sender_id == self._agent_id + # Skip messages from self (agent's own messages) to avoid infinite loops. + # ``PlatformMessage`` declares these as ``str``, but it's a plain + # dataclass filled from Fern models -- a backend null reaches here as + # ``None`` and would otherwise raise ``TypeError`` from is_self_echo + # (a pyo3 extension with non-Optional ``str`` params). + if self._agent_id and is_self_echo( + sender_id=msg.sender_id or "", + sender_type=msg.sender_type or "", + agent_id=self._agent_id, ): logger.debug("Skipping self-message %s", msg_id) return BacklogProcessResult.ADVANCED @@ -1811,11 +1815,15 @@ async def _process_event(self, event: PlatformEvent) -> bool: # 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 - if ( - self._agent_id - and payload.sender_type == "Agent" - and payload.sender_id == self._agent_id + # Skip messages from self (agent's own messages) to avoid infinite + # loops. ``payload`` is hydrated via ``model_construct``, which + # bypasses pydantic validation -- a null sender field would + # otherwise raise ``TypeError`` from is_self_echo's non-Optional + # ``str`` params. + if self._agent_id and is_self_echo( + sender_id=payload.sender_id or "", + sender_type=payload.sender_type or "", + agent_id=self._agent_id, ): logger.debug("Skipping self-message %s", msg_id) return True diff --git a/src/band/runtime/oneshot.py b/src/band/runtime/oneshot.py index 1d66be520..eba0c0ea7 100644 --- a/src/band/runtime/oneshot.py +++ b/src/band/runtime/oneshot.py @@ -42,9 +42,19 @@ import logging from collections import deque from datetime import datetime, timezone -from typing import Any +from enum import StrEnum +from typing import Any, cast + +from band_sdk_core import ( + DrainCandidate, + evaluate_adapter_result, + evaluate_delivery_event, + evaluate_drain_candidate, + evaluate_next_message, +) from band.client.rest import DEFAULT_REQUEST_OPTIONS +from band.logging_config import current_traceparent from band.runtime.capabilities import prune_unsupported from band.runtime.participants import participant_snapshot from band.core.protocols import FrameworkAdapter @@ -63,9 +73,30 @@ ) from band.runtime.tools import AgentTools +# BandLink.get_next_message returns this dataclass, not band.core.types' +# same-named one that _build_platform_message below constructs. +from band.runtime.types import PlatformMessage as RestPlatformMessage + logger = logging.getLogger(__name__) +class OneShotStatus(StrEnum): + """``handle_event``/``_process_message_event``'s ``status`` vocabulary. + + A public contract hosts branch on (e.g. ``result["status"] == "done"``), + kept spelled exactly as it always has been — deliberately not renamed to + band_sdk_core's own decision literals (``skip_self``, ``cleanup``), + which are a distinct vocabulary consumed inline at each call site. + """ + + IGNORED = "ignored" + CLEANED_UP = "cleaned_up" + SKIPPED_SELF = "skipped_self" + NO_PENDING = "no_pending" + ALREADY_PROCESSED = "already_processed" + DONE = "done" + + # Defensive cap on the drain loop. The platform shouldn't backlog dozens of # messages for a single agent in normal operation; if it does, surface it via # ``drain_truncated`` rather than draining indefinitely. @@ -193,67 +224,101 @@ async def shutdown(self) -> None: async def handle_event(self, body: dict[str, Any]) -> dict[str, Any]: """Process one forwarded platform event from the bridge envelope. - Non-message events return ``{"status": "ignored", ...}`` without side - effects; in v1 only ``message_created`` drives an LLM call. + Routing is band_sdk_core's ``evaluate_delivery_event`` — see + ``docs/websocket-events.md``. Non-message events return + ``{"status": "ignored", ...}`` without side effects; in v1 only + ``message_created`` drives an LLM call. Raises: - OneShotEnvelopeError: envelope is missing ``room_id`` or - ``payload.id`` for a ``message_created`` event. + OneShotEnvelopeError: the envelope fails core's validation — + missing/empty ``room_id`` or ``payload.id`` for a + ``message_created`` or room-cleanup event, or a + ``message_created`` payload missing a required field. RuntimeError: ``startup()`` was not called first. """ if not self._started: raise RuntimeError("OneShotInvoker.startup() not called") - event_type = body.get("event_type") - - # Long-running containers keep one invoker (and one adapter) alive - # across many rooms over the container's lifetime. Adapters cache - # per-room state on ``self`` (e.g. Anthropic's ``_message_history``, - # Claude SDK's live per-room sessions, langgraph checkpoints); the - # only thing that frees those entries is ``adapter.on_cleanup``. - # Without this hookup the cache grows unbounded — and for adapters - # that spawn subprocesses per room, those subprocesses leak too. - # Mirrors ``AgentRuntime._destroy_execution``'s cleanup-callback hook - # in the long-running path. - if event_type in {"room_removed", "room_deleted"}: - room_id = body.get("room_id") or (body.get("payload") or {}).get("id") - if room_id: - try: - await self._adapter.on_cleanup(room_id) - except Exception: - logger.warning( - "Adapter on_cleanup failed for room %s", - room_id, - exc_info=True, - ) - return { - "status": "cleaned_up", - "event_type": event_type, - "room_id": room_id, - } - - # Other forwardable event types intentionally fall through to - # "ignored": + event_type: str = body.get("event_type") or "" + payload = body.get("payload") or {} + try: + decision = evaluate_delivery_event( + event_type, + body.get("room_id"), + payload, + self._agent_id, + current_traceparent(), + ) + except ValueError as exc: + raise OneShotEnvelopeError(str(exc)) from exc + + # Other forwardable event types intentionally route to "ignored": # - room_added: bridge already subscribed the WS; no per-room # context to create on this side. # - participant_added/removed: OneShot fetches participants fresh # on every invocation, so there's no cache to update. # - contact_*: routed via the separate ContactEventConfig flow in # long-running mode; not wired into OneShot. - if event_type != "message_created": - logger.debug("Ignoring non-message event: %s", event_type) - return {"status": "ignored", "event_type": event_type} + match decision: + case {"decision": "ignored", "event_type": ignored_event_type}: + logger.debug("Ignoring non-message event: %s", ignored_event_type) + return { + "status": OneShotStatus.IGNORED, + "event_type": ignored_event_type, + } + case {"decision": "cleanup", "room_id": room_id}: + # Long-running containers keep one invoker (and one adapter) + # alive across many rooms over the container's lifetime. + # Adapters cache per-room state on ``self`` (e.g. Anthropic's + # ``_message_history``, Claude SDK's live per-room sessions, + # langgraph checkpoints); the only thing that frees those + # entries is ``adapter.on_cleanup``. Without this hookup the + # cache grows unbounded — and for adapters that spawn + # subprocesses per room, those subprocesses leak too. Mirrors + # ``AgentRuntime._destroy_execution``'s cleanup-callback hook + # in the long-running path. + return await self._cleanup_room(event_type, cast(str, room_id)) + case {"decision": "skip_self", "message_id": message_id}: + logger.debug("Skipping self-message %s", message_id) + return {"status": OneShotStatus.SKIPPED_SELF, "message_id": message_id} + case {"decision": "invocation", "room_id": room_id}: + return await self._process_message_event( + room_id=cast(str, room_id), payload=payload + ) + case _: + raise AssertionError(f"unreachable delivery decision: {decision!r}") - payload = body.get("payload") or {} - room_id = body.get("room_id") or payload.get("chat_room_id") - if not room_id: - raise OneShotEnvelopeError("missing room_id") - if not payload.get("id"): - raise OneShotEnvelopeError("missing message id in payload") + # --- Internal: the lifecycle dance --- - return await self._process_message_event(room_id=room_id, payload=payload) + async def _cleanup_room(self, event_type: str, room_id: str) -> dict[str, Any]: + try: + await self._adapter.on_cleanup(room_id) + except Exception: + logger.warning( + "Adapter on_cleanup failed for room %s", room_id, exc_info=True + ) + return { + "status": OneShotStatus.CLEANED_UP, + "event_type": event_type, + "room_id": room_id, + } - # --- Internal: the lifecycle dance --- + async def _acknowledge( + self, *, room_id: str, message_id: str, succeeded: bool, error: str = "" + ) -> None: + match evaluate_adapter_result(room_id, message_id, succeeded): + case {"decision": "processed"}: + await self._link.mark_processed(room_id, message_id) + case {"decision": "failed"}: + try: + await self._link.mark_failed(room_id, message_id, error) + except Exception: + logger.warning( + "Could not mark %s failed in room %s", + message_id, + room_id, + exc_info=True, + ) async def _process_message_event( self, *, room_id: str, payload: dict[str, Any] @@ -261,62 +326,54 @@ async def _process_message_event( """Run the SDK agent loop for one forwarded message_created event. Steps (the message case of ``ExecutionContext._process_event``, - adapted for request/response): + adapted for request/response); self-filtering already happened in + ``handle_event`` via ``evaluate_delivery_event``: - 1. Self-filter — skip the agent's own echo without an LLM call. - 2. ``get_next_message`` — if the triggering message isn't the next + 1. ``get_next_message`` — if the triggering message isn't the next open one for this agent, exit early (a sibling invocation already claimed it, or there's an older unprocessed message ahead of it). - 3. ``mark_processing`` — claim it. - 4. Fetch participants + history, build ``AgentInput``, run adapter. - 5. ``mark_processed`` on success. - 6. Drain — only swallow messages the LLM actually saw (``seen_ids``). + 2. ``mark_processing`` — claim it. + 3. Fetch participants + history, build ``AgentInput``, run adapter. + 4. ``mark_processed`` on success. + 5. Drain — only swallow messages the LLM actually saw (``seen_ids``). A message that arrived after the history snapshot is left open so the next invocation handles it with fresh context. - 7. ``mark_failed`` on exception. + 6. ``mark_failed`` on exception. """ msg_id = payload["id"] - # 1. Self-message filter — Band echoes the agent's own outbound - # messages back on its WS subscription, which the bridge forwards here. - if ( - payload.get("sender_type") == "Agent" - and payload.get("sender_id") == self._agent_id - ): - return {"status": "skipped_self", "message_id": msg_id} - - # 2. Verify the triggering message is the next open one for this agent. # The platform's ``/next`` returns the oldest actionable message — # anything not yet in ``processed`` state, including ones stuck in # ``processing`` from a previous crash — so a single call covers both # the normal claim case and stuck-message reclaim. next_msg = await self._link.get_next_message(room_id) - if next_msg is None: - logger.info( - "Skip: room %s has no pending messages (triggering=%s)", - room_id, - msg_id, - ) - return {"status": "no_pending", "message_id": msg_id} - if next_msg.id != msg_id: - logger.info( - "Skip: room %s next-open=%s != triggering=%s", - room_id, - next_msg.id, - msg_id, - ) - return { - "status": "already_processed", - "message_id": msg_id, - "next_open": next_msg.id, - } + match evaluate_next_message(msg_id, next_msg.id if next_msg else None): + case {"decision": "no_pending"}: + logger.info( + "Skip: room %s has no pending messages (triggering=%s)", + room_id, + msg_id, + ) + return {"status": OneShotStatus.NO_PENDING, "message_id": msg_id} + case {"decision": "already_processed", "next_open_id": next_open_id}: + logger.info( + "Skip: room %s next-open=%s != triggering=%s", + room_id, + next_open_id, + msg_id, + ) + return { + "status": OneShotStatus.ALREADY_PROCESSED, + "message_id": msg_id, + "next_open": next_open_id, + } + case {"decision": "ready_to_claim"}: + pass - # 3. Claim. logger.info("Claiming msg %s in room %s", msg_id, room_id) await self._link.mark_processing(room_id, msg_id) try: - # 4. Build AgentInput and run the adapter. participants = await self._fetch_participants(room_id) sender_name = _lookup_sender_name(participants, payload.get("sender_id")) @@ -350,25 +407,20 @@ async def _process_message_event( await self._adapter.on_event(inp) - # 5. Mark the triggering message processed. - await self._link.mark_processed(room_id, msg_id) + await self._acknowledge(room_id=room_id, message_id=msg_id, succeeded=True) except Exception as exc: - # 7. Mark failed so the platform can surface the error. logger.exception( "Adapter failed for message %s in room %s", msg_id, room_id ) - try: - await self._link.mark_failed(room_id, msg_id, str(exc)[:500] or "error") - except Exception: - logger.warning( - "Could not mark %s failed in room %s", - msg_id, - room_id, - exc_info=True, - ) + await self._acknowledge( + room_id=room_id, + message_id=msg_id, + succeeded=False, + error=str(exc)[:500] or "error", + ) raise - # 6. Drain — scoped to what the LLM saw (seen_ids). A message that + # Drain is scoped to what the LLM saw (seen_ids). A message that # arrived after the history snapshot is NOT swallowed; it's left open # so the next invocation processes it with fresh context. drained: list[str] = [] @@ -386,22 +438,29 @@ async def _process_message_event( exc_info=True, ) break - if stale is None: - break - # Defensive: the platform shouldn't return our own messages here, - # but the SDK guards against it (execution.py self-message skip). - if stale.sender_type == "Agent" and stale.sender_id == self._agent_id: - continue - if stale.id not in seen_ids: - logger.info( - "Drain stopped at %s in room %s — arrived after history snapshot", - stale.id, - room_id, - ) - break - await self._link.mark_processing(room_id, stale.id) - await self._link.mark_processed(room_id, stale.id) - drained.append(stale.id) + match evaluate_drain_candidate( + _drain_candidate(stale), seen_ids, self._agent_id + ): + case {"decision": "no_candidate"}: + break + case {"decision": "self_echo"}: + # Defensive: the platform shouldn't return our own + # messages here, but the SDK guards against it + # (execution.py self-message skip). An echo never halts + # the drain, so it consumes a cap iteration and continues. + continue + case {"decision": "out_of_snapshot", "message_id": stale_id}: + logger.info( + "Drain stopped at %s in room %s — arrived after history snapshot", + stale_id, + room_id, + ) + break + case {"decision": "drain", "message_id": stale_id}: + stale_id = cast(str, stale_id) + await self._link.mark_processing(room_id, stale_id) + await self._link.mark_processed(room_id, stale_id) + drained.append(stale_id) else: drain_truncated = True logger.warning( @@ -418,7 +477,7 @@ async def _process_message_event( ) result: dict[str, Any] = { - "status": "done", + "status": OneShotStatus.DONE, "room_id": room_id, "message_id": msg_id, } @@ -544,6 +603,26 @@ async def _fetch_history( # --- Module-level helpers (no state, easy to unit-test) --- +def _drain_candidate(msg: RestPlatformMessage | None) -> DrainCandidate | None: + """``/next``'s dataclass fields are unvalidated; core rejects a non-string. + + ``PlatformMessage`` declares ``sender_id``/``sender_type`` as ``str``, but + it's a plain dataclass filled from Fern models — a backend null reaches + here as ``None`` and would otherwise raise ``TypeError`` mid-drain, after + the triggering message was already marked processed. ``or ""`` keeps + today's behavior: an empty sender never matches self-echo, so it falls + through to the snapshot check exactly as a real "no match" candidate + would. + """ + if msg is None: + return None + return { + "id": msg.id, + "sender_id": msg.sender_id or "", + "sender_type": msg.sender_type or "", + } + + def _lookup_sender_name( participants: list[dict[str, Any]], sender_id: str | None ) -> str | None: diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py index f77242bdf..384aa3198 100644 --- a/tests/runtime/test_execution.py +++ b/tests/runtime/test_execution.py @@ -10,6 +10,7 @@ from band_sdk_core import ClaimRegistry, RetryTracker +from band.client.streaming import MessageMetadata from band.logging_config import TRACE_CONTEXT, trace_context_scope from band.runtime.execution import ( Execution, @@ -18,7 +19,7 @@ BacklogProcessResult, _error_label, ) -from band.runtime.types import ConversationContext, SessionConfig +from band.runtime.types import ConversationContext, PlatformMessage, SessionConfig # Import test helpers from conftest from tests.conftest import ( @@ -689,8 +690,6 @@ async def test_sync_processes_backlog_messages( self, mock_link_with_next, mock_handler ): """Sync should process backlog messages from /next.""" - from datetime import datetime, timezone - from band.runtime.types import PlatformMessage # Setup get_next_message to return one backlog message, then None backlog_msg = PlatformMessage( @@ -731,8 +730,6 @@ async def test_sync_point_clears_marker_and_keeps_dedupe_cache( self, mock_link_with_next, mock_handler ): """When sync point is reached, marker is cleared and dedupe is preserved.""" - from datetime import datetime, timezone - from band.runtime.types import PlatformMessage # Setup: WS message arrives, then /next returns same message sync_msg = PlatformMessage( @@ -776,8 +773,6 @@ async def test_sync_removes_duplicate_from_ws_queue( self, mock_link_with_next, mock_handler ): """Sync should dedupe when non-message events are ahead of sync-point WS copy.""" - from datetime import datetime, timezone - from band.runtime.types import PlatformMessage sync_msg = PlatformMessage( id="msg-sync-001", @@ -851,7 +846,6 @@ async def test_ws_replay_with_processed_metadata_is_not_reopened( self, mock_link_with_next, mock_handler ): """Processed WebSocket replay should not call mark_processing or execute.""" - from band.client.streaming import MessageMetadata ctx = ExecutionContext( "room-123", @@ -919,11 +913,74 @@ async def test_ws_replay_uses_hydrated_context_delivery_status( await ctx.stop() + async def test_ws_skips_self_authored_message( + self, mock_link_with_next, mock_handler + ): + """The live-event self-echo guard (band_sdk_core.is_self_echo) must + never reach the handler or mark anything.""" + ctx = ExecutionContext( + "room-123", + mock_link_with_next, + mock_handler, + agent_id="agent-123", + config=SessionConfig(enable_context_hydration=False), + ) + await ctx.start() + await asyncio.sleep(0.05) + + event = make_message_event( + room_id="room-123", + msg_id="msg-self-echo", + sender_id="agent-123", + sender_type="Agent", + ) + await ctx.on_event(event) + await asyncio.sleep(0.1) + + mock_handler.assert_not_called() + mock_link_with_next.mark_processing.assert_not_called() + assert "msg-self-echo" not in ctx.claims.completed_ids(ctx.room_id) + + await ctx.stop() + + async def test_backlog_skips_self_authored_message( + self, mock_link_with_next, mock_handler + ): + """The backlog self-echo guard (band_sdk_core.is_self_echo) must + never reach the handler or mark anything.""" + + msg = PlatformMessage( + id="msg-backlog-self-echo", + room_id="room-123", + content="echo", + sender_id="agent-123", + sender_type="Agent", + sender_name=None, + message_type="text", + metadata={}, + created_at=datetime.now(timezone.utc), + ) + ctx = ExecutionContext( + "room-123", + mock_link_with_next, + mock_handler, + agent_id="agent-123", + config=SessionConfig(enable_context_hydration=False), + ) + + result = await ctx._process_backlog_message(msg) + + assert result == BacklogProcessResult.ADVANCED + mock_handler.assert_not_awaited() + mock_link_with_next.mark_processing.assert_not_awaited() + mock_link_with_next.mark_processed.assert_not_awaited() + + await ctx.stop() + async def test_pending_next_message_present_in_context_still_executes( self, mock_link_with_next, mock_handler ): """A pending /next message is work even when it appears in room context.""" - from band.runtime.types import PlatformMessage pending_msg = PlatformMessage( id="msg-pending-down", @@ -981,7 +1038,6 @@ async def test_same_id_backlog_and_ws_paths_are_locally_inflight_deduped( self, mock_link_with_next, mock_handler ): """Only one path should execute when /next and WebSocket race on an id.""" - from band.runtime.types import PlatformMessage processing_started = asyncio.Event() release_processing = asyncio.Event() @@ -1036,7 +1092,6 @@ async def test_first_message_to_fresh_room_executes_once(self, mock_link_with_ne 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() @@ -1186,7 +1241,6 @@ async def test_backlog_processed_ack_failure_is_not_remembered( self, mock_link_with_next, mock_handler ): """Local success without durable processed ack must not enter processed dedupe.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-ack-fails", @@ -1246,7 +1300,6 @@ async def test_backlog_processed_ack_failure_retries_ack_without_handler_replay( self, mock_link_with_next, mock_handler ): """Redelivery after local success should retry only the processed ack.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-backlog-ack-retry", @@ -1282,7 +1335,6 @@ async def test_processed_ack_retry_budget_exhaustion_keeps_local_completion( self, mock_link_with_next, mock_handler ): """Permanent processed ack failure should not deadlock or replay local side effects.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-ack-budget", @@ -1423,7 +1475,6 @@ async def test_resync_retries_pending_ack_before_advancing_to_newer_backlog( before processing a newer /next backlog message -- normal resync cannot get past a stuck pending ACK to reach newer backlog. Once the ACK confirms, resync proceeds normally to the newer message.""" - from band.runtime.types import PlatformMessage newer_msg = PlatformMessage( id="msg-newer", @@ -1465,7 +1516,6 @@ async def test_sync_point_claim_failure_does_not_clear_marker( self, mock_link_with_next, mock_handler ): """A failed durable claim is not a completed sync point.""" - from band.runtime.types import PlatformMessage sync_msg = PlatformMessage( id="msg-sync-claim-fails", @@ -1503,7 +1553,6 @@ async def test_startup_backlog_claim_failure_does_not_spin( self, mock_link_with_next, mock_handler ): """Startup sync should stop after one unclaimable non-sync backlog message.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-startup-claim-fails", @@ -1540,7 +1589,6 @@ async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler ): """Startup sync should not switch to WebSocket after an unclaimable backlog message.""" - from band.runtime.types import PlatformMessage backlog_msg = PlatformMessage( id="msg-older-claim-fails", @@ -1584,7 +1632,6 @@ async def test_resync_claim_failure_does_not_spin( self, mock_link_with_next, mock_handler ): """Resync should stop after one unclaimable /next message.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-resync-claim-fails", @@ -1620,7 +1667,6 @@ async def test_resync_claim_failure_does_not_process_newer_ws_event( self, mock_link_with_next, mock_handler ): """Phase 2 resync should block queued WebSocket events behind older /next work.""" - from band.runtime.types import PlatformMessage msg = PlatformMessage( id="msg-resync-older-claim-fails", @@ -1665,8 +1711,6 @@ async def test_sync_skips_permanently_failed( self, mock_link_with_next, mock_handler ): """Sync should skip permanently failed messages.""" - from datetime import datetime, timezone - from band.runtime.types import PlatformMessage failed_msg = PlatformMessage( id="msg-failed-001", @@ -1702,8 +1746,6 @@ async def test_sync_skips_permanently_failed( async def test_retry_tracker_records_failures(self, mock_link_with_next): """Retry tracker should record failed processing attempts.""" - from datetime import datetime, timezone - from band.runtime.types import PlatformMessage # Handler that fails failing_handler = AsyncMock(side_effect=Exception("Processing failed")) @@ -1743,7 +1785,6 @@ async def test_retry_saturation_skips_handler_on_next_delivery( """Once a message's attempts exceed max_retries it becomes permanently failed, and a *subsequent* delivery of that same message must skip the handler entirely rather than invoke it again.""" - from band.runtime.types import PlatformMessage failing_handler = AsyncMock(side_effect=Exception("Processing failed")) msg = PlatformMessage( diff --git a/tests/runtime/test_oneshot.py b/tests/runtime/test_oneshot.py index a3c037bc2..814102f84 100644 --- a/tests/runtime/test_oneshot.py +++ b/tests/runtime/test_oneshot.py @@ -112,6 +112,7 @@ def _msg_body( "sender_type": sender_type, "message_type": "user", "inserted_at": "2026-05-21T10:00:00Z", + "updated_at": "2026-05-21T10:00:00Z", }, } @@ -331,14 +332,27 @@ async def test_room_removed_swallows_cleanup_errors(self) -> None: ) assert result["status"] == "cleaned_up" + async def test_room_removed_with_no_resolvable_room_id_raises_envelope_error( + self, + ) -> None: + """A room-cleanup event with no room identity anywhere (envelope nor + payload) is a malformed envelope, not a silent no-op — unlike the + pre-band_sdk_core behavior, which returned ``{"status": "cleaned_up", + "room_id": None}`` without calling ``on_cleanup``. + """ + link = make_link_mock() + adapter = _make_adapter_mock() + invoker = await _make_invoker(link, adapter) + + with pytest.raises(OneShotEnvelopeError, match="room_id"): + await invoker.handle_event({"event_type": "room_removed", "payload": {}}) + adapter.on_cleanup.assert_not_awaited() + async def test_missing_room_id_raises_envelope_error(self) -> None: link = make_link_mock() invoker = await _make_invoker(link) - body = { - "event_type": "message_created", - "agent_id": "agent-1", - "payload": {"id": "msg-1", "sender_id": "u", "content": "x"}, - } + body = _msg_body() + del body["room_id"] with pytest.raises(OneShotEnvelopeError, match="room_id"): await invoker.handle_event(body) @@ -357,18 +371,10 @@ async def test_missing_message_id_raises_envelope_error(self) -> None: async def test_falls_back_to_payload_chat_room_id(self) -> None: link = make_link_mock(next_messages=[platform_msg("msg-1"), None]) invoker = await _make_invoker(link) - body = { - "event_type": "message_created", - "agent_id": "agent-1", - "payload": { - "id": "msg-1", - "chat_room_id": "fallback-room", - "sender_id": "u", - "sender_type": "User", - "content": "hi", - "inserted_at": "2026-05-21T10:00:00Z", - }, - } + body = _msg_body() + del body["room_id"] + body["payload"]["chat_room_id"] = "fallback-room" + result = await invoker.handle_event(body) assert result["room_id"] == "fallback-room" diff --git a/uv.lock b/uv.lock index 7c1e5a58d..176caf97c 100644 --- a/uv.lock +++ b/uv.lock @@ -774,7 +774,7 @@ requires-dist = [ { name = "anthropic", marker = "extra == 'dev-parlant'", specifier = ">=0.75.0" }, { name = "async-lru", specifier = ">=2.3.0" }, { name = "band-client-rest", specifier = "==0.0.27" }, - { name = "band-sdk-core", specifier = "==2.0.0" }, + { name = "band-sdk-core", specifier = "==2.2.0" }, { name = "band-testing-python", marker = "extra == 'dev'", specifier = "==0.1.4" }, { name = "band-testing-python", marker = "extra == 'dev-crewai'", specifier = "==0.1.4" }, { name = "band-testing-python", marker = "extra == 'dev-parlant'", specifier = "==0.1.4" }, @@ -918,17 +918,17 @@ provides-extras = ["logging", "desktop", "codex", "opencode", "letta", "pydantic [[package]] name = "band-sdk-core" -version = "2.0.0" +version = "2.2.0" source = { registry = "https://pypi.org/simple" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e6/b0/c0450d0df122aa46ffaf11aabacd5dcf27b81e35ff1f5b08639c0cfa1457/band_sdk_core-2.0.0-cp311-abi3-macosx_10_12_x86_64.whl", hash = "sha256:7db8f08f4dc4616ecee2aa5e8a7bd61d846d959a309a6b25da598e649d4ac278", size = 457389, upload-time = "2026-08-29T12:00:47.019Z" }, - { url = "https://files.pythonhosted.org/packages/a6/99/12d8e5a5fb4ef42e684117de2687cbb7bf8df0b706eeb65587caa9867b49/band_sdk_core-2.0.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:556f420711b7026b0b3dbb9abf0a2dcdd6f7126f2c09ec40ca281c081229851c", size = 458700, upload-time = "2026-08-29T12:00:48.396Z" }, - { url = "https://files.pythonhosted.org/packages/16/c3/a894dd0a873a80184bdf443e770917c2ec85e671a8c72f4c8400584692e9/band_sdk_core-2.0.0-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:19e58438cf932bb0324c0358a8c332437eade63bb253e6f7cddbfa02c928051e", size = 509782, upload-time = "2026-08-29T12:00:49.711Z" }, - { url = "https://files.pythonhosted.org/packages/b1/95/18d976f8a12d24819e3788a2c880ec7d07945673c6a34b68fd7653142ee8/band_sdk_core-2.0.0-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:9f581c2173967e44a219d86f124a65650fc915c8c99a1f9ddf69c0e98612253f", size = 510429, upload-time = "2026-08-29T12:00:50.849Z" }, - { url = "https://files.pythonhosted.org/packages/67/87/1eef985931ddddf2f2ff9623622f15bab460d4d943aa7dc96d1da8227d37/band_sdk_core-2.0.0-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:430fb51e32080fa58a1c84f8d2342c443865c85a05f8ad727288063e5bdb425a", size = 687161, upload-time = "2026-08-29T12:00:51.987Z" }, - { url = "https://files.pythonhosted.org/packages/08/8d/61d49368320438ad8ee7ebfc689cfc62163cdb9a5d7592e8d48a3438aae4/band_sdk_core-2.0.0-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:3606b58f5a3d9ff006fcd9db425cc4ac47c4482380146a282e23299ca6f0fcab", size = 724455, upload-time = "2026-08-29T12:00:53.376Z" }, - { url = "https://files.pythonhosted.org/packages/ea/17/c795774be4b1b1c96633c68b5a8f2966bfd0752004835920f6f0a676153a/band_sdk_core-2.0.0-cp311-abi3-win_amd64.whl", hash = "sha256:8ac8e95b466e8110f4e3dd99afaa4520f825c294e2b731d04c246efe1edd7685", size = 331961, upload-time = "2026-08-29T12:00:54.47Z" }, - { url = "https://files.pythonhosted.org/packages/ab/93/aa331d33db1d8436dba57df83d4592c6534223ffc11bf94c1eb5090714cc/band_sdk_core-2.0.0-cp311-abi3-win_arm64.whl", hash = "sha256:d802cb5a411afa5b7e6089962cdc3b4573ad6d43048fc2b28a8a9079a247aa0b", size = 314808, upload-time = "2026-08-29T12:00:55.707Z" }, + { url = "https://files.pythonhosted.org/packages/f2/b6/6432cd80745c68517ffb2efb9f8dc5f5d2ad423eaccb002d7a14dd1d28f9/band_sdk_core-2.2.0-cp311-abi3-macosx_10_12_x86_64.whl", hash = "sha256:2eaa8f62a5b806d10993ea4feb73f1c48d3c959b9539bfbcdbd1ffa1f426b998", size = 475871, upload-time = "2026-09-01T11:30:26.358Z" }, + { url = "https://files.pythonhosted.org/packages/9a/1d/a20cbfce00fd32ec5e1145528138fd2b0d8af12dd09c337a59c6c4b79f69/band_sdk_core-2.2.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:52c4c6af089e81d877e34b839694e8b01547d0f25891cf1bc9c717cf32278f59", size = 477493, upload-time = "2026-09-01T11:30:27.934Z" }, + { url = "https://files.pythonhosted.org/packages/e5/d8/4ec2c9a0b37027072e92dbe2912fce0d593ee88c3f96e1281882429588ff/band_sdk_core-2.2.0-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:9c99fad0f3edb0fc9a9d42079a2be76d573c4dd7a9579ce4b5aea1d5fabaaf9e", size = 526652, upload-time = "2026-09-01T11:30:29.398Z" }, + { url = "https://files.pythonhosted.org/packages/61/1a/548e4f355517ee7cd4422906055765b955ec543efc95ef4993ac8a2a84ed/band_sdk_core-2.2.0-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:431e7185d2a8659371a6fb22aa0a3383e935607d1c8944f488eebf70b0821cf7", size = 528152, upload-time = "2026-09-01T11:30:30.953Z" }, + { url = "https://files.pythonhosted.org/packages/37/9a/6b19bf76fc55d4f5a1315ac036506aea71c499651b0fbe328004c9e40011/band_sdk_core-2.2.0-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:1996fb61fb22d5750df214b3d180e183cdbf3f7e36cea960ca8ca2442a93b888", size = 705290, upload-time = "2026-09-01T11:30:32.514Z" }, + { url = "https://files.pythonhosted.org/packages/c7/8a/5bceb2a5732fb51b1fa62d6dea172104ea79080c768a0aa4a4d42ddf5d5c/band_sdk_core-2.2.0-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:8688f4c55c7cd778e5bb2846b6ada03e7c392d9eb8565a698fdb3df6e29faa8b", size = 743509, upload-time = "2026-09-01T11:30:34.036Z" }, + { url = "https://files.pythonhosted.org/packages/0e/00/c267b0fc208f121ad25b41f41a18c5d77c9280049eae8a359821d37c2a33/band_sdk_core-2.2.0-cp311-abi3-win_amd64.whl", hash = "sha256:704bff82b7494f1997df1aa20def09d7431075f0ad0a1dd4895a4b916409a9f2", size = 348322, upload-time = "2026-09-01T11:30:35.664Z" }, + { url = "https://files.pythonhosted.org/packages/79/d5/46a9dce9b20509904d69b181685fd2d1d498c657466dcd58ba0fedb78248/band_sdk_core-2.2.0-cp311-abi3-win_arm64.whl", hash = "sha256:7f49e75f6f890f38ba7366fb9264c8ca01c34f231db7ab8f08f4c6cc9c161295", size = 331334, upload-time = "2026-09-01T11:30:36.826Z" }, ] [[package]]