fix: apply agent.control signals off the agent_control topic processor

band-bridge/bridge_core/bridge.py

@@ -120,8 +120,8 @@ def __init__(
         # 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
+        # Background tasks spawned by control signals (play nudges) so the
+        # next control signal isn't queued behind 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()
@@ -234,8 +234,6 @@ 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:
@@ -244,6 +242,9 @@ async def close(self) -> None:
                 self._config.agent_id,
                 exc_info=True,
             )
+        # After disconnect: the link applies control signals until it returns,
+        # so a PLAY can still spawn a nudge during it.
+        await self._cancel_control_tasks()
         try:
             await self._forwarder.close()
         except Exception:
@@ -253,6 +254,13 @@ async def close(self) -> None:
                 exc_info=True,
             )
 
+    async def _cancel_control_tasks(self) -> None:
+        pending = tuple(self._control_tasks)
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.wait(pending)
+
     async def _connect_and_consume(self) -> None:
         """Connect, subscribe, and consume events until shutdown."""
         # BandLink.connect() joins the control channel, whose pushes may arrive
@@ -589,10 +597,10 @@ def _remember_processed_message(self, message_id: str) -> None:
     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``
+        Runs off BandLink's control lane 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:

band-bridge/bridge_core/control.py

@@ -40,8 +40,8 @@ def __init__(self, runner: AgentRunner) -> None:
         # 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).
+        # BandLink delivers control signals one at a time (as for
+        # AgentRuntime's ControlDeduplicator on the long-running SDK path).
         self._seen_control_ids: OrderedDict[str, bool] = OrderedDict()
 
     async def handle(self, payload: AgentControlPayload) -> None:
@@ -76,7 +76,8 @@ async def handle(self, payload: AgentControlPayload) -> None:
                     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.
+                # queued behind this /next + forward: BandLink applies control
+                # signals one at a time.
                 self._runner._spawn_play_nudge(room_ids)
             case _:
                 logger.warning(

src/band/client/streaming/client.py

@@ -261,6 +261,9 @@ class AgentControlPayload(WirePayload):
     correlation_id: str | None = None
 
 
+ControlHook = Callable[[AgentControlPayload], Awaitable[None]]
+
+
 class SupersedePayload(WirePayload):
     """Payload for terminal agent_control supersede events."""
 
@@ -764,7 +767,7 @@ async def join_agent_control_channel(
         self,
         agent_id: str,
         on_supersede: Callable[[SupersedePayload], Awaitable[None]],
-        on_control: Callable[[AgentControlPayload], Awaitable[None]] | None = None,
+        on_control: ControlHook | None = None,
     ):
         """Subscribe to agent-control events for this agent.
 

src/band/platform/link.py

@@ -9,7 +9,7 @@
 
 import asyncio
 import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from band_sdk_core import AgentTopicKind, SessionState
@@ -20,6 +20,7 @@
     WebSocketDisconnectReason,
     WebSocketUpgradeError,
 )
+from band.client.streaming.client import ControlHook
 from band.config.settings import DEFAULT_REST_URL, DEFAULT_WS_URL
 from band.core.exceptions import AgentAlreadyRunningError
 from band.core.types import ConflictPolicy, PlatformConnection
@@ -134,10 +135,13 @@ def __init__(
         self._last_disconnect_reason: WebSocketDisconnectReason | None = None
 
         # Preemptive control-signal hook (interrupt/stop/play). Set by the
-        # runtime. Invoked DIRECTLY from the WebSocket receive task — never via
-        # the serialized _event_queue — so a control signal can act on a cycle
-        # already in flight instead of queuing behind it.
-        self.on_control: Callable[[AgentControlPayload], Awaitable[None]] | None = None
+        # runtime. Never run via the serialized _event_queue, so a control
+        # signal can act on a cycle already in flight instead of queuing
+        # behind it. Signals apply one at a time, so a hook that never
+        # returns holds every later signal until disconnect().
+        self.on_control: ControlHook | None = None
+        # Control signals still being applied, in arrival order.
+        self._control_lane: list[asyncio.Task[None]] = []
 
     @property
     def is_connected(self) -> bool:
@@ -232,6 +236,7 @@ async def disconnect(self) -> None:
         except Exception as e:  # noqa: BLE001 -- best-effort event emission must not crash the turn/link
             logger.warning("Error unsubscribing from agent_control: %s", e)
 
+        await self._cancel_control_lane()
         await self._ws.__aexit__(None, None, None)
         self._ws = None
         self._is_connected = False
@@ -416,18 +421,57 @@ async def _on_supersede(self, payload: SupersedePayload) -> None:
     async def _on_control(self, payload: AgentControlPayload) -> None:
         """Handle an ``agent.control`` push (interrupt/stop/play).
 
-        Invoked directly from the WebSocket receive task. Forwards to the
-        registered ``on_control`` hook WITHOUT touching the serialized event
-        queue, so the signal can preempt a cycle already in flight. If no hook
-        is registered, the push is a safe no-op.
+        Runs the ``on_control`` hook on a task ordered after every earlier
+        signal. The hook can take seconds (a STOP drains every room's turn),
+        and awaiting it here would hold a ``supersede`` queued behind it on
+        the same topic, where a reconnect discards queued messages unseen.
+        If no hook is registered, the push is a safe no-op.
         """
-        if self.on_control is None:
+        hook = self.on_control
+        if hook is None:
             logger.debug(
                 "agent.control received (mode=%s) but no on_control hook registered",
                 payload.mode,
             )
             return
-        await self.on_control(payload)
+        previous = self._control_lane[-1] if self._control_lane else None
+        task = asyncio.create_task(
+            self._apply_control(previous, hook, payload),
+            name=f"agent.control {payload.mode} correlation_id={payload.correlation_id}",
+        )
+        self._control_lane.append(task)
+        task.add_done_callback(self._control_lane.remove)
+
+    @staticmethod
+    async def _apply_control(
+        previous: asyncio.Task[None] | None,
+        hook: ControlHook,
+        payload: AgentControlPayload,
+    ) -> None:
+        if previous is not None:
+            await asyncio.wait({previous})
+        try:
+            await hook(payload)
+        except Exception:
+            logger.exception(
+                "agent.control %s failed (correlation_id=%s room_id=%s)",
+                payload.mode,
+                payload.correlation_id,
+                payload.room_id,
+            )
+
+    async def _cancel_control_lane(self) -> None:
+        if not self._control_lane:
+            return
+        pending = list(self._control_lane)
+        logger.info(
+            "Cancelling %d unfinished control signal(s) on disconnect: %s",
+            len(pending),
+            ", ".join(task.get_name() for task in pending),
+        )
+        for task in pending:
+            task.cancel()
+        await asyncio.wait(pending)
 
     async def _on_disconnected(self, error: Exception | None) -> None:
         """Handle PHX client disconnection."""

src/band/runtime/control.py

@@ -23,7 +23,7 @@
 class ControlDeduplicator:
     """Drops repeated signals by correlation id; the server does not.
 
-    Bounded LRU, only touched from the WebSocket receive task.
+    Bounded LRU; ``BandLink`` delivers control signals one at a time.
     """
 
     def __init__(self) -> None:

src/band/runtime/execution.py

@@ -252,9 +252,10 @@ async def request_resync(self) -> None:
     def interrupt(self, *, kind: ControlMode | str = ControlMode.INTERRUPT) -> bool:
         """Abort the in-flight reasoning cycle for this room.
 
-        Called preemptively from the WebSocket receive task on an ``interrupt``
-        or ``stop`` control signal. ``AgentRuntime`` ``hasattr``-guards this, so
-        custom ``Execution`` implementations that omit it degrade to a no-op.
+        Called preemptively, outside the room's message queue, on an
+        ``interrupt`` or ``stop`` control signal. ``AgentRuntime``
+        ``hasattr``-guards this, so custom ``Execution`` implementations that
+        omit it degrade to a no-op.
         ``kind`` keeps ``str`` in the union because ``Protocol`` signatures
         aren't runtime-enforced and a custom implementation may still pass a
         plain string.
@@ -691,8 +692,8 @@ async def request_resync(self) -> None:
     def interrupt(self, *, kind: ControlMode | str = ControlMode.INTERRUPT) -> bool:
         """Abort the in-flight reasoning cycle, if any.
 
-        Called from the WebSocket receive task. It only terminates the running
-        turn scope; all message-status / dedupe bookkeeping happens once in
+        Called on a control signal, outside the cycle. It only terminates the
+        running turn scope; all message-status / dedupe bookkeeping happens once in
         ``_run_cycle``, never here, so the two coroutines do not race on shared
         state.
 

src/band/runtime/runtime.py

@@ -307,8 +307,8 @@ async def _on_reconnected(self) -> None:
     async def handle_control(self, payload: AgentControlPayload) -> None:
         """Apply an ``agent.control`` signal (interrupt/stop/play) to executions.
 
-        Invoked directly from the WebSocket receive task (via
-        ``BandLink.on_control``) so it can preempt a cycle already in flight.
+        Invoked via ``BandLink.on_control``, outside the serialized event
+        queue, so it can preempt a cycle already in flight.
 
         Routing:
         - ``scope == "agent"`` with no ``room_id`` → fan out to all executions.
@@ -318,8 +318,8 @@ async def handle_control(self, payload: AgentControlPayload) -> None:
         Dedupes on ``correlation_id`` (the server does not). A signal that
         reached no room is not recorded, so a later copy still applies once
         the room is routed. Note: exceptions raised here are
-        swallowed-and-logged by the channel handler, so this method must not
-        rely on propagating errors to signal anything.
+        only logged by ``BandLink``, so this method must not rely on
+        propagating errors to signal anything.
         """
         targets = control_targets(payload, self.executions)
         if not targets or not self._control_ids.first_sighting(payload):

tests/bridge/test_close_after_control.py

@@ -0,0 +1,43 @@
+"""``AgentRunner.close()`` cancels a play nudge spawned while its link is still
+disconnecting: the link applies control signals until ``disconnect()`` returns."""
+
+from __future__ import annotations
+
+import asyncio
+
+from band.client.streaming import AgentControlPayload
+from tests.bridge.test_bridge import _build_runner
+
+
+class TestCloseAfterControl:
+    async def test_a_play_applied_during_disconnect_never_outlives_close(
+        self,
+    ) -> None:
+        runner, _, link = _build_runner()
+        story: list[str] = []
+        nudge_started = asyncio.Event()
+
+        async def next_message(room_id: str) -> None:
+            story.append(f"nudge {room_id}")
+            nudge_started.set()
+            try:
+                await asyncio.Future()
+            except asyncio.CancelledError:
+                story.append(f"nudge {room_id} cancelled")
+                raise
+
+        async def disconnect() -> None:
+            """A PLAY queued in the link's control lane applies before it ends."""
+            await runner._control.handle(
+                AgentControlPayload(
+                    mode="play", scope="room", agent_id="agent-1", room_id="r1"
+                )
+            )
+            await nudge_started.wait()
+
+        link.get_next_message.side_effect = next_message
+        link.disconnect.side_effect = disconnect
+
+        await asyncio.wait_for(runner.close(), timeout=5.0)
+
+        assert story == ["nudge r1", "nudge r1 cancelled"]

tests/conftest.py

@@ -354,6 +354,18 @@ async def __call__(self, ctx, event) -> None:
 # =============================================================================
 
 
+# The platform's eviction of an older same-key session (``agent_control``
+# ``supersede``), as the wire carries it.
+TERMINAL_SUPERSEDE: dict[str, Any] = {
+    "reason": "session.already_connected",
+    "message": "Superseded by a newer session",
+    "retryable": False,
+    "retry_after": 15,
+    "target_socket_id": "agent_socket:agent-123",
+    "correlation_id": "evict-1",
+}
+
+
 def spy_on_reconciliation_drain(link: BandLink) -> asyncio.Event:
     """Arm a spy on ``_drain_reconciliation`` -- the last step
     ``_on_reconnected`` runs -- and return an event set the instant the real

tests/platform/conftest.py

@@ -12,6 +12,7 @@
 from band_sdk_core import DeadReason, SessionState
 
 from band.client.streaming import WebSocketClient
+from band.platform.link import BandLink
 
 
 class AllTopicsJoined:
@@ -34,6 +35,14 @@ def __contains__(self, topic: object) -> bool:
         return topic not in self._missing
 
 
+@pytest.fixture
+def link() -> BandLink:
+    """An unconnected BandLink, for driving its handlers directly."""
+    return BandLink(
+        agent_id="agent-123", api_key="k", ws_url="wss://x/ws", rest_url="https://x"
+    )
+
+
 @pytest.fixture
 def mock_ws_client():
     """Autospecced WebSocketClient for testing BandLink.