From 76e80931896dc56a7c1ffeb4739e39e15ff44eed Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 14:06:04 +0300
Subject: [PATCH 1/7] feat: adopt band_sdk_core's one-shot delivery lifecycle
 in OneShotInvoker

OneShotInvoker's handle_event/_process_message_event hand-rolled the same
ignore/cleanup/self-echo/invocation routing, next-message comparison, drain
classification, and ack decision that band_sdk_core's evaluate_delivery_event/
evaluate_next_message/evaluate_drain_candidate/evaluate_adapter_result were
extracted from as the reference implementation. Route through those four
functions instead, so core's stricter payload validation and one definition
of each decision replace the local logic. Two intentional behavior changes:
a room-cleanup event with no resolvable room id now raises OneShotEnvelopeError
(400) instead of silently no-op'ing, and a malformed self-echo payload raises
before classification instead of returning skipped_self.

Also introduces OneShotStatus, an SDK-owned StrEnum for the response
"status" vocabulary hosts branch on (kept spelled exactly as it always has
been), replacing six re-typed magic-string literals in the producer.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 src/band/runtime/__init__.py  |   3 +-
 src/band/runtime/oneshot.py   | 314 +++++++++++++++++++++-------------
 tests/runtime/test_oneshot.py |  40 +++--
 3 files changed, 224 insertions(+), 133 deletions(-)

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/oneshot.py b/src/band/runtime/oneshot.py
index 1d66be520..5b51efd2f 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,57 @@ 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.
+        # 1. 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,
-            }
-
-        # 3. Claim.
+        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
+
+        # 2. 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.
+            # 3. 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 +410,22 @@ 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)
+            # 4. Mark the triggering message processed.
+            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.
+            # 6. 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
+        # 5. Drain — 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 +443,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 +482,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 +608,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_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"
 

From d711e4c92067aedd92867dab468fd6bc4a25a11f Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 14:06:13 +0300
Subject: [PATCH 2/7] feat: consolidate ExecutionContext self-echo detection
 via band_sdk_core.is_self_echo

Both self-echo checks in ExecutionContext (the backlog path and the live
WebSocket path) hand-rolled the same sender_type == "Agent" and sender_id ==
agent_id comparison. Route both through band_sdk_core's new public
is_self_echo, the same predicate OneShotInvoker now uses via
evaluate_delivery_event/evaluate_drain_candidate, so the two SDKs share one
definition instead of two independently-maintained copies.

Adds regression coverage for both call sites -- self-echo had no test
anywhere in tests/runtime/ before this.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 src/band/runtime/execution.py   | 18 ++++-----
 tests/runtime/test_execution.py | 65 +++++++++++++++++++++++++++++++++
 2 files changed, 74 insertions(+), 9 deletions(-)

diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py
index 151d26a8c..10b4fe949 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
@@ -1415,10 +1415,10 @@ 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
+        if self._agent_id and is_self_echo(
+            sender_id=msg.sender_id,
+            sender_type=msg.sender_type,
+            agent_id=self._agent_id,
         ):
             logger.debug("Skipping self-message %s", msg_id)
             return BacklogProcessResult.ADVANCED
@@ -1812,10 +1812,10 @@ 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
+            if self._agent_id and is_self_echo(
+                sender_id=payload.sender_id,
+                sender_type=payload.sender_type,
+                agent_id=self._agent_id,
             ):
                 logger.debug("Skipping self-message %s", msg_id)
                 return True
diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py
index f77242bdf..7cc65c679 100644
--- a/tests/runtime/test_execution.py
+++ b/tests/runtime/test_execution.py
@@ -919,6 +919,71 @@ 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."""
+        from band.runtime.types import PlatformMessage
+
+        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
     ):

From 9aa1320ac7b5e549cecad88e6204d36a975ed2b8 Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 14:06:23 +0300
Subject: [PATCH 3/7] docs: note band_sdk_core's role in the one-shot delivery
 lifecycle

docs/websocket-events.md read as if band_sdk_core were only an
event-validation dependency; note the delivery-lifecycle decisions and that
OneShotInvoker's routing is core's, not the SDK's own logic. AGENTS.md gets
a short band-sdk-core pointer naming which decisions live in core and the
standing obligation to extend the CI wheel-smoke when a new core symbol
starts being used. ARCHITECTURE.md now references OneShotStatus instead of
a bare status-string literal.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 AGENTS.md                          | 14 ++++++++++++++
 docs/websocket-events.md           | 12 ++++++++++++
 examples/agentcore/ARCHITECTURE.md |  2 +-
 3 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/AGENTS.md b/AGENTS.md
index ec1992266..bb17c3325 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -28,6 +28,20 @@ never enable it in an example or default config. See
 [docs/capability-negotiation.md](docs/capability-negotiation.md) for the full
 emit/capabilities API and how requests get pruned against `AgentMe.feature_flags`.
 
+## band-sdk-core
+
+`band-sdk-core` (Rust, PyO3-bound, shared with band-sdk-typescript) is the
+decision layer underneath both the WebSocket and REST paths: inbound
+event-payload validation (`validate_event_payload`), delivery/retry/session
+state (`ClaimRegistry`/`RetryTracker`/`ParticipantRoster`/`Session`), the
+shared `is_self_echo` predicate, and the one-shot delivery lifecycle
+(`evaluate_delivery_event`/`evaluate_next_message`/`evaluate_drain_candidate`/
+`evaluate_adapter_result`) that `OneShotInvoker` is a thin wrapper around —
+see [docs/websocket-events.md](docs/websocket-events.md). Whenever code here
+starts calling a 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.
+
 ## REST Client
 
 The SDK uses a Fern-generated REST client with a property-based namespace API
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..385bf8cf7 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": 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;

From 40e6f43b82891f4eeaa2b399460a770f69517158 Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 14:34:16 +0300
Subject: [PATCH 4/7] chore: bump band-sdk-core to 2.2.0, prove is_self_echo in
 CI

2.2.0 adds the public is_self_echo used by ExecutionContext and
OneShotInvoker (band-sdk-core#64). Extends the wheel-smoke CI step to prove
is_self_echo and evaluate_drain_candidate are callable from the isolated
pinned install, not just importable -- the standing obligation whenever a
new core symbol starts being used.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 .github/workflows/ci.yml | 10 +++++++++-
 pyproject.toml           |  2 +-
 uv.lock                  | 20 ++++++++++----------
 3 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 92c638150..5dd407781 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -255,7 +255,13 @@ 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_drain_candidate,
+            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.
@@ -265,6 +271,8 @@ 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"}
         print('Core imports successful')
         PYEOF
 
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/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]]

From 2ddd3f3e849cb8dd950f2f7873fb1030d4bbf872 Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 15:13:35 +0300
Subject: [PATCH 5/7] docs: fold band-sdk-core pointer into the existing
 WebSocket doc, not a new section

The previous commit added a standalone AGENTS.md heading duplicating the
existing WebSocket Channels & Events pointer to docs/websocket-events.md,
working against the AGENTS.md slim-down this branch built on (#601). Fold
the one durable fact (extend the CI wheel-smoke when a new core symbol
starts being used) into that existing section instead.

Also fixes ARCHITECTURE.md's OneShotStatus reference: the wire response
carries the string "no_pending", not Python enum syntax.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 AGENTS.md                          | 21 +++++----------------
 examples/agentcore/ARCHITECTURE.md |  2 +-
 2 files changed, 6 insertions(+), 17 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index bb17c3325..96ea0dd19 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -28,20 +28,6 @@ never enable it in an example or default config. See
 [docs/capability-negotiation.md](docs/capability-negotiation.md) for the full
 emit/capabilities API and how requests get pruned against `AgentMe.feature_flags`.
 
-## band-sdk-core
-
-`band-sdk-core` (Rust, PyO3-bound, shared with band-sdk-typescript) is the
-decision layer underneath both the WebSocket and REST paths: inbound
-event-payload validation (`validate_event_payload`), delivery/retry/session
-state (`ClaimRegistry`/`RetryTracker`/`ParticipantRoster`/`Session`), the
-shared `is_self_echo` predicate, and the one-shot delivery lifecycle
-(`evaluate_delivery_event`/`evaluate_next_message`/`evaluate_drain_candidate`/
-`evaluate_adapter_result`) that `OneShotInvoker` is a thin wrapper around —
-see [docs/websocket-events.md](docs/websocket-events.md). Whenever code here
-starts calling a 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.
-
 ## REST Client
 
 The SDK uses a Fern-generated REST client with a property-based namespace API
@@ -56,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/examples/agentcore/ARCHITECTURE.md b/examples/agentcore/ARCHITECTURE.md
index 385bf8cf7..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": OneShotStatus.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;

From 830baa1699ef9f0ff1645c9ac3af17684fffffac Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Tue, 1 Sep 2026 15:13:42 +0300
Subject: [PATCH 6/7] style: trim narration comments, consolidate redundant
 inline imports

oneshot.py's numbered inline comments ("# 1. Claim.", "# 4. Mark...")
duplicated _process_message_event's own docstring step list without adding
information; drop the ones that only restate the following line, keep the
ones stating an actual invariant (unnumbered).

test_execution.py re-imported PlatformMessage, MessageMetadata, and
datetime/timezone inside ~20 individual test methods despite each already
being available at module scope (or trivially promotable there) -- moves
them to the top-level import, one source of truth per name.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018aLCfpQh5vzVzEneDe2kVY
---
 src/band/runtime/oneshot.py     |  7 +------
 tests/runtime/test_execution.py | 28 ++--------------------------
 2 files changed, 3 insertions(+), 32 deletions(-)

diff --git a/src/band/runtime/oneshot.py b/src/band/runtime/oneshot.py
index 5b51efd2f..eba0c0ea7 100644
--- a/src/band/runtime/oneshot.py
+++ b/src/band/runtime/oneshot.py
@@ -342,7 +342,6 @@ async def _process_message_event(
         """
         msg_id = payload["id"]
 
-        # 1. 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
@@ -371,12 +370,10 @@ async def _process_message_event(
             case {"decision": "ready_to_claim"}:
                 pass
 
-        # 2. Claim.
         logger.info("Claiming msg %s in room %s", msg_id, room_id)
         await self._link.mark_processing(room_id, msg_id)
 
         try:
-            # 3. Build AgentInput and run the adapter.
             participants = await self._fetch_participants(room_id)
             sender_name = _lookup_sender_name(participants, payload.get("sender_id"))
 
@@ -410,10 +407,8 @@ async def _process_message_event(
 
             await self._adapter.on_event(inp)
 
-            # 4. Mark the triggering message processed.
             await self._acknowledge(room_id=room_id, message_id=msg_id, succeeded=True)
         except Exception as exc:
-            # 6. Mark failed so the platform can surface the error.
             logger.exception(
                 "Adapter failed for message %s in room %s", msg_id, room_id
             )
@@ -425,7 +420,7 @@ async def _process_message_event(
             )
             raise
 
-        # 5. 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] = []
diff --git a/tests/runtime/test_execution.py b/tests/runtime/test_execution.py
index 7cc65c679..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",
@@ -954,7 +948,6 @@ async def test_backlog_skips_self_authored_message(
     ):
         """The backlog self-echo guard (band_sdk_core.is_self_echo) must
         never reach the handler or mark anything."""
-        from band.runtime.types import PlatformMessage
 
         msg = PlatformMessage(
             id="msg-backlog-self-echo",
@@ -988,7 +981,6 @@ 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",
@@ -1046,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()
@@ -1101,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()
@@ -1251,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",
@@ -1311,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",
@@ -1347,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",
@@ -1488,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",
@@ -1530,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",
@@ -1568,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",
@@ -1605,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",
@@ -1649,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",
@@ -1685,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",
@@ -1730,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",
@@ -1767,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"))
@@ -1808,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(

From 1a8b4a34cefe4f4aa39503dfedf50be997834e7a Mon Sep 17 00:00:00 2001
From: Alexander Zaikman <alexander.zaikman@band.ai>
Date: Fri, 4 Sep 2026 17:36:57 +0300
Subject: [PATCH 7/7] fix: guard is_self_echo against null sender fields,
 extend wheel-smoke coverage

The band-sdk-core migration swapped ExecutionContext's null-safe `==`
self-echo comparisons for band_sdk_core.is_self_echo, a pyo3 extension with
non-Optional str params. The /next backlog path reads sender_id/sender_type
from a Fern REST model that can carry a backend null despite its str type
hint (the same hazard oneshot.py's _drain_candidate already guards against);
without the guard, a null sender crashes the drain with TypeError instead of
being treated as a no-match. Apply the same `or ""` guard on the live
WebSocket path for consistency with the one established pattern.

Also extend the CI wheel-smoke step to prove evaluate_delivery_event,
evaluate_next_message, and evaluate_adapter_result are callable from the
pinned wheel, per this PR's own AGENTS.md rule that new band_sdk_core
symbols get wheel-smoke coverage -- oneshot.py now calls all three but the
smoke step only checked evaluate_drain_candidate and is_self_echo.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDYPGiaDSYNZnyC5mg18ch
---
 .github/workflows/ci.yml      | 13 +++++++++++++
 src/band/runtime/execution.py | 20 ++++++++++++++------
 2 files changed, 27 insertions(+), 6 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5dd407781..8b50b9eda 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -259,7 +259,10 @@ jobs:
             ClaimRegistry,
             ParticipantRoster,
             RetryTracker,
+            evaluate_adapter_result,
+            evaluate_delivery_event,
             evaluate_drain_candidate,
+            evaluate_next_message,
             is_self_echo,
         )
 
@@ -273,6 +276,16 @@ jobs:
         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/src/band/runtime/execution.py b/src/band/runtime/execution.py
index 10b4fe949..edd2efbd9 100644
--- a/src/band/runtime/execution.py
+++ b/src/band/runtime/execution.py
@@ -1414,10 +1414,14 @@ async def _process_backlog_message(
         """
         msg_id = msg.id
 
-        # Skip messages from self (agent's own messages) to avoid infinite loops
+        # 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,
-            sender_type=msg.sender_type,
+            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)
@@ -1811,10 +1815,14 @@ 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
+            # 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,
-                sender_type=payload.sender_type,
+                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)