feat: remote agents honor interrupt/stop/play control signals

band-bridge/bridge_core/bridge.py

@@ -43,6 +43,7 @@
 from band.platform.link import BandLink
 
 from .config import AgentConfig, BridgeConfig, ReconnectConfig
+from .control import ControlSignalHandler
 from .forwarder import Forwarder, build_forwarder
 from .health import HealthServer
 
@@ -105,6 +106,23 @@ def __init__(
         self._shutdown_event = shutdown_event
         self._connected_event = asyncio.Event()
         self._processed_message_ids: OrderedDict[str, None] = OrderedDict()
+        # In-flight forward tasks per room, so a control signal (interrupt/stop)
+        # can cancel whatever forwards are currently live for a room. Tracked as
+        # a set because same-room events serialize on the per-room lock: while
+        # one task forwards, others wait on that lock, and all of them must be
+        # cancellable — otherwise interrupt cancels a queued waiter while the
+        # running forward keeps going. Populated by _safe_handle_event at the
+        # top of processing, discarded in its finally.
+        self._active_room_tasks: dict[str, set[asyncio.Task[None]]] = {}
+        # Control-signal handling (interrupt/stop/play) lives in its own type;
+        # it delegates cancel/nudge/room-list back to this runner but owns the
+        # correlation-id dedup and routing (see control.py).
+        self._control = ControlSignalHandler(self)
+        # Background tasks spawned by control signals (play nudges) so the WS
+        # receive task isn't blocked awaiting them. Tracked here to be
+        # cancelled on close() and to keep a strong ref (create_task alone
+        # doesn't).
+        self._control_tasks: set[asyncio.Task[None]] = set()
         # Per-room locks: events for the same room are forwarded sequentially
         # so the container always sees a settled history (its own prior reply
         # is already posted before the next invocation reads room context).
@@ -214,6 +232,8 @@ def _backoff_sleep_seconds(self, delay: float) -> float:
 
     async def close(self) -> None:
         """Disconnect link and close forwarder. Idempotent."""
+        for task in tuple(self._control_tasks):
+            task.cancel()
         try:
             await self._link.disconnect()
         except Exception:
@@ -236,6 +256,11 @@ async def _connect_and_consume(self) -> None:
         await self._link.connect()
         self._connected_event.set()
 
+        # Route control signals (interrupt/stop/play) to the control handler.
+        # Set as soon as connected so the hook is live for the agent_control
+        # channel BandLink.connect() already joined.
+        self._link.on_control = self._control.handle
+
         await self._link.subscribe_agent_rooms(self._config.agent_id)
 
         existing_rooms = await self._fetch_existing_rooms()
@@ -331,31 +356,36 @@ async def _rehydrate_backlog(self, room_ids: list[str]) -> None:
         marking, so ``/next`` would return the same message until the container
         marks it processed. The container's own drain loop pulls the rest.
         """
+        # Rooms are independent; nudge concurrently. Per-room locks and dedup
+        # still serialize each nudge against any live event for that room.
+        await asyncio.gather(*(self._nudge_room(rid) for rid in room_ids))
 
-        async def nudge(room_id: str) -> None:
-            try:
-                msg = await self._link.get_next_message(room_id)
-            except Exception:
-                logger.warning(
-                    "Agent %s: rehydration /next failed for room %s",
-                    self._config.agent_id,
-                    room_id,
-                    exc_info=True,
-                )
-                return
-            if msg is None:
-                return
-            logger.info(
-                "Agent %s: rehydrating room %s with backlog message %s",
+    async def _nudge_room(self, room_id: str) -> None:
+        """Forward one backlog message for a room, if ``/next`` has one.
+
+        Shared by startup rehydration (:meth:`_rehydrate_backlog`) and a live
+        ``play`` control signal (``ControlSignalHandler.handle``) — both cases
+        are the same "catch up this room via /next" operation.
+        """
+        try:
+            msg = await self._link.get_next_message(room_id)
+        except Exception:
+            logger.warning(
+                "Agent %s: rehydration /next failed for room %s",
                 self._config.agent_id,
                 room_id,
-                msg.id,
+                exc_info=True,
             )
-            await self._safe_handle_event(self._backlog_event(msg))
-
-        # Rooms are independent; nudge concurrently. Per-room locks and dedup
-        # still serialize each nudge against any live event for that room.
-        await asyncio.gather(*(nudge(rid) for rid in room_ids))
+            return
+        if msg is None:
+            return
+        logger.info(
+            "Agent %s: rehydrating room %s with backlog message %s",
+            self._config.agent_id,
+            room_id,
+            msg.id,
+        )
+        await self._safe_handle_event(self._backlog_event(msg))
 
     def _backlog_event(self, msg: PlatformMessage) -> MessageEvent:
         """Wrap a ``/next`` PlatformMessage as a synthetic message_created event.
@@ -382,20 +412,36 @@ def _backlog_event(self, msg: PlatformMessage) -> MessageEvent:
         )
 
     async def _safe_handle_event(self, event: object) -> None:
-        # Semaphore caps concurrent in-flight forwards per agent. Holding it
-        # across ``_handle_event`` (which includes the per-room lock + the
-        # forwarder call) is the point — bursts wait here instead of stacking
-        # up inside the forwarder.
-        async with self._forward_semaphore:
-            try:
+        # Track this task among the in-flight forwards for its room so a control
+        # signal (interrupt/stop) can cancel it — whether it's the one holding
+        # the per-room lock and forwarding, or a same-room task still waiting on
+        # that lock. Room-less events (e.g. contact events) aren't tracked —
+        # there's nothing room-scoped to cancel for them.
+        room_id = getattr(event, "room_id", None)
+        task = asyncio.current_task()
+        if room_id and task is not None:
+            self._active_room_tasks.setdefault(room_id, set()).add(task)
+        try:
+            # Semaphore caps concurrent in-flight forwards per agent. Holding
+            # it across ``_handle_event`` (which includes the per-room lock +
+            # the forwarder call) is the point — bursts wait here instead of
+            # stacking up inside the forwarder.
+            async with self._forward_semaphore:
                 await self._handle_event(event)
-            except Exception:
-                logger.warning(
-                    "Agent %s: error handling event %s",
-                    self._config.agent_id,
-                    type(event).__name__,
-                    exc_info=True,
-                )
+        except Exception:
+            logger.warning(
+                "Agent %s: error handling event %s",
+                self._config.agent_id,
+                type(event).__name__,
+                exc_info=True,
+            )
+        finally:
+            if room_id:
+                tasks = self._active_room_tasks.get(room_id)
+                if tasks is not None:
+                    tasks.discard(task)
+                    if not tasks:
+                        self._active_room_tasks.pop(room_id, None)
 
     async def _handle_event(self, event: object) -> None:
         """Manage room subscriptions; forward forwardable events.
@@ -533,6 +579,56 @@ def _remember_processed_message(self, message_id: str) -> None:
         if len(self._processed_message_ids) > _DEDUP_MAX_SIZE:
             self._processed_message_ids.popitem(last=False)
 
+    # --- Control signals (interrupt/stop/play) ---
+    #
+    # Signal routing/dedup lives in ControlSignalHandler (control.py); it calls
+    # back into the operations below, which touch this runner's forwarding
+    # state (the per-room task registry) and the /next nudge path.
+
+    def _spawn_play_nudge(self, room_ids: list[str]) -> None:
+        """Fire ``play`` /next nudges as tracked background tasks.
+
+        Runs off the WS receive task so a following interrupt/stop isn't queued
+        behind the nudge's /next + forward. Each spawned forward still lands in
+        ``_active_room_tasks`` (via ``_safe_handle_event``), so a later
+        interrupt/stop can cancel it. Tasks are tracked in ``_control_tasks``
+        and cancelled on ``close()``.
+        """
+        for room_id in room_ids:
+            task = asyncio.create_task(self._nudge_room(room_id))
+            self._control_tasks.add(task)
+            task.add_done_callback(self._control_tasks.discard)
+
+    def _cancel_active_forward(self, room_id: str) -> None:
+        """Cancel all in-flight forward tasks for a room, if any.
+
+        Cancels both the task currently forwarding and any same-room tasks
+        still queued on the per-room lock, so a signal can't leave a waiter
+        behind to run the moment the cancelled forward releases the lock.
+
+        Best-effort: this stops the bridge from waiting and releases the room
+        lock immediately. It does not guarantee the remote invocation itself
+        stops — ``AgentCoreForwarder``'s underlying boto3 call runs in a
+        thread and isn't killed by cancelling the awaiting task (see
+        ``forwarder.py``); that is a pre-existing, documented limitation of
+        that transport, not something this signal can fix.
+        """
+        tasks = self._active_room_tasks.get(room_id)
+        if not tasks:
+            return
+        cancelled = 0
+        for task in tuple(tasks):
+            if not task.done():
+                task.cancel()
+                cancelled += 1
+        if cancelled:
+            logger.info(
+                "Agent %s: cancelled %d in-flight forward(s) for room %s",
+                self._config.agent_id,
+                cancelled,
+                room_id,
+            )
+
 
 class BandBridge:
     """Bridge orchestrator: N AgentRunners + signal handling + health server."""

band-bridge/bridge_core/control.py

@@ -0,0 +1,135 @@
+"""Agent control-signal handling (interrupt / stop / play) for the bridge.
+
+Split out of ``AgentRunner``: the runner owns forwarding, rehydration, and the
+per-room in-flight task registry; this module owns only the control-signal
+concern — correlation-id dedup, target-room resolution, and mode dispatch —
+delegating the actual cancel / nudge / room-list operations back to the runner.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import OrderedDict
+from typing import TYPE_CHECKING
+
+from band.client.streaming import AgentControlPayload
+
+if TYPE_CHECKING:
+    from .bridge import AgentRunner
+
+logger = logging.getLogger(__name__)
+
+_CONTROL_DEDUP_MAX_SIZE = 256
+
+
+class ControlSignalHandler:
+    """Applies ``agent.control`` signals to a runner's in-flight forwards.
+
+    The bridge holds no Band lifecycle logic, so ``interrupt`` and ``stop`` are
+    handled identically: cancel whatever forward task is currently in flight
+    for the target room(s), if any. There is nothing to do if no forward is in
+    flight — the interrupt-vs-stop message-lifecycle distinction (consume vs.
+    leave-for-replay) is already handled downstream by the container via its
+    own ``/next`` claim check. ``play`` proactively nudges the room(s) via
+    ``/next`` so a queued message is picked up without waiting for the next
+    natural bridge event.
+    """
+
+    def __init__(self, runner: AgentRunner) -> None:
+        self._runner = runner
+        # Control-signal dedup. The server does not deduplicate agent.control
+        # pushes, so a stale duplicate could otherwise reach out and cancel
+        # whatever unrelated new task now occupies a room slot. Bounded LRU;
+        # only touched from the WebSocket receive task (same as AgentRuntime's
+        # _seen_control_ids on the long-running SDK path).
+        self._seen_control_ids: OrderedDict[str, bool] = OrderedDict()
+
+    async def handle(self, payload: AgentControlPayload) -> None:
+        """Apply an ``agent.control`` signal to in-flight forwards.
+
+        Routing mirrors ``AgentRuntime.handle_control``: a ``room_id`` targets
+        that room only; ``scope == "agent"`` with no ``room_id`` fans out to
+        all of this agent's rooms; any other combination is a no-op. Dedupes on
+        ``correlation_id`` (the server does not).
+        """
+        if self._is_duplicate(payload):
+            return
+
+        room_ids = await self._resolve_rooms(payload)
+        if room_ids is None:
+            return
+
+        agent_id = self._runner.agent_id
+        logger.info(
+            "Agent %s: applying control mode=%s scope=%s to %d room(s) "
+            "(correlation_id=%s)",
+            agent_id,
+            payload.mode,
+            payload.scope,
+            len(room_ids),
+            payload.correlation_id,
+        )
+
+        match payload.mode:
+            case "interrupt" | "stop":
+                for room_id in room_ids:
+                    self._runner._cancel_active_forward(room_id)
+            case "play":
+                # Nudge in the background so a following stop/interrupt isn't
+                # blocked behind this /next + forward on the WS receive task.
+                self._runner._spawn_play_nudge(room_ids)
+            case _:
+                logger.warning(
+                    "Agent %s: ignoring control signal with unknown mode=%s",
+                    agent_id,
+                    payload.mode,
+                )
+
+    def _is_duplicate(self, payload: AgentControlPayload) -> bool:
+        """Return True if this signal's ``correlation_id`` was already handled.
+
+        Records unseen ids in a bounded LRU. Signals without a
+        ``correlation_id`` are never treated as duplicates (the server omits
+        it on some pushes) but are logged.
+        """
+        agent_id = self._runner.agent_id
+        cid = payload.correlation_id
+        if cid is None:
+            logger.debug(
+                "Agent %s: control signal mode=%s has no correlation_id; not deduped",
+                agent_id,
+                payload.mode,
+            )
+            return False
+
+        if cid in self._seen_control_ids:
+            logger.debug(
+                "Agent %s: duplicate control signal %s ignored",
+                agent_id,
+                cid,
+            )
+            return True
+
+        self._seen_control_ids[cid] = True
+        self._seen_control_ids.move_to_end(cid)
+        if len(self._seen_control_ids) > _CONTROL_DEDUP_MAX_SIZE:
+            self._seen_control_ids.popitem(last=False)
+        return False
+
+    async def _resolve_rooms(self, payload: AgentControlPayload) -> list[str] | None:
+        """Resolve the signal's target room ids, or None for a no-op.
+
+        A ``room_id`` targets that room; ``scope == "agent"`` with no
+        ``room_id`` fans out to all of the agent's rooms (possibly empty); any
+        other combination returns None so the caller skips dispatch.
+        """
+        if payload.room_id is not None:
+            return [payload.room_id]
+        if payload.scope == "agent":
+            return await self._runner._fetch_existing_rooms()
+        logger.warning(
+            "Agent %s: control signal scope=%s with no room_id; no-op",
+            self._runner.agent_id,
+            payload.scope,
+        )
+        return None

band-bridge/bridge_core/forwarder.py

@@ -157,6 +157,10 @@ def _call() -> None:
                     body.close()
 
         try:
+            # interrupt/stop are best-effort for this transport: cancelling the
+            # await frees the bridge but can't kill the in-flight boto3 thread —
+            # the remote invocation runs to completion (or its own timeout)
+            # regardless.
             await asyncio.wait_for(
                 asyncio.to_thread(_call),
                 timeout=self._target.timeout,

src/band/client/streaming/__init__.py

@@ -23,6 +23,7 @@
     ContactAddedPayload,
     ContactRemovedPayload,
     SupersedePayload,
+    AgentControlPayload,
 )
 from band.client.streaming.errors import WebSocketUpgradeError
 
@@ -44,4 +45,5 @@
     "ContactAddedPayload",
     "ContactRemovedPayload",
     "SupersedePayload",
+    "AgentControlPayload",
 ]