feat: remote agents honor interrupt/stop/play control signals by amit-gazal-thenvoi · Pull Request #442 · band-ai/band-sdk-python

amit-gazal-thenvoi · 2026-07-16T07:02:40Z

Summary

SDK joins agent_control:{agent_id} and handles interrupt/stop/play preemptively, out-of-band from the message queue, so a signal can abort a reasoning cycle already in flight (ExecutionContext/AgentRuntime).
interrupt aborts the in-flight cycle and consumes the message (nothing sent, no replay). stop does the same plus durably suppresses new cycles in that room until play, which triggers a rehydration-style /next catch-up.
Bridge (band-bridge) honors the same signals for one-shot/bridged agents: interrupt/stop cancel whatever forward call is currently in flight for a room; play proactively nudges the room via /next instead of waiting for the next natural event.
Fix: the ReconnectedEvent sync path now short-circuits locally on _stopped, consistent with the other stopped-guards, instead of making a /next call already known to return empty.

Test plan

uv run pytest tests/ --ignore=tests/integration/ --ignore=tests/e2e/ — full unit suite green
uv run pytest tests/runtime/ tests/platform/ tests/websocket/ tests/bridge/ tests/test_platform_runtime.py — control-signal suites green (851 passed)
uv run ruff check . / uv run ruff format --check . / uv run pyrefly check — clean
Live integration tests (tests/integration/test_control_signals.py) — require a live platform + BAND_API_KEY/BAND_API_KEY_USER; not run in CI

🤖 Generated with Claude Code

Wires agent_control into the bridge's AgentRunner: interrupt/stop cancel whatever forward task is currently in flight for a room, and play proactively nudges the room via /next instead of waiting for the next natural event. The bridge holds no Band lifecycle logic, so interrupt and stop are handled identically here — the consume-vs-replay distinction and durable per-room suppression already happen correctly downstream via OneShotInvoker's /next claim check and the platform's stop gate.

The ReconnectedEvent path in ExecutionContext._process_event always ran the full /next sync, unlike the idle-timeout and resync-sentinel paths which already short-circuit on _stopped. The platform gate made this safe (a stopped room's /next is guaranteed empty), but it wasted a call on every reconnect while stopped. Skip locally for consistency with the other _stopped short-circuits.

linear-code · 2026-07-16T07:02:43Z

INT-829

… sync A stop control signal landing mid-cycle during _synchronize_with_next or _resync_pending_messages left the message in 'processing' for replay, but the enclosing loop kept polling /next, which returns 'processing' messages same as before. The very next iteration re-claimed and fully re-ran the cycle stop just aborted. Both loops now check self._stopped after processing a backlog message and stop polling until play resumes. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…y budget A cycle cancelled by our own stop/interrupt control signal never actually ran the handler, but record_attempt() was still charging it as a real attempt. At the default max_message_retries=1, a message stopped once and replayed via /next on resume would immediately land in permanently_failed before ever reaching the adapter. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…te-agents-honor-interrupt-and-stopresume # Conflicts: # src/band/client/streaming/client.py # src/band/platform/link.py # src/band/runtime/execution.py # src/band/runtime/platform_runtime.py # uv.lock

AlexanderZ-Band · 2026-07-16T15:02:29Z

+        room_id = getattr(event, "room_id", None)
+        task = asyncio.current_task()
+        if room_id and task is not None:
+            self._active_room_tasks[room_id] = task


_active_room_tasks is set before the semaphore and per-room lock. A second same-room task can overwrite the active forward while it waits, so interrupt cancels the waiter and the real forward continues. Register the task only while it owns the room lock, or track all tasks per room.

AlexanderZ-Band · 2026-07-16T15:02:52Z

+
+        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


Cancelling this await does not cancel asyncio.to_thread() or the AgentCore invocation. The remote cycle can keep consuming tokens and running tools, so this does not meet the in-flight halt requirement. Add remote cancellation or exclude this transport from that guarantee.

AlexanderZ-Band · 2026-07-16T15:02:53Z

+            for room_id in room_ids:
+                self._cancel_active_forward(room_id)
+        elif payload.mode == "play":
+            await asyncio.gather(*(self._nudge_room(rid) for rid in room_ids))


play awaits the full /next forward on the WebSocket callback task. While that forward runs, this receive path cannot process a following stop or interrupt. Schedule the nudge as a tracked background task.

AlexanderZ-Band · 2026-07-16T15:02:54Z

+                else:
+                    logger.warning(
+                        "ExecutionContext %s: durable mark_processed failed for "
+                        "interrupted message %s; idle resync may re-deliver it",


If this ack fails, the interrupted message remains locally replayable and /next can run it again. Record _remember_processed_ack_pending(msg_id) so the interrupt stays consumed while the durable ack retries.

AlexanderZ-Band · 2026-07-16T15:02:55Z

+            # _run_cycle and we send nothing further. Only message-driven
+            # cycles report the working signal (see _invoke_handler);
+            # participant add/remove events are housekeeping and skip it.
+            if not await self._run_cycle(event, msg_id):


A control signal during hydration or mark_processing() sees no _active_cycle_task: interrupt is lost, and stop can still reach this call without another _stopped check. Cover the full claimed-message lifecycle or re-check control state immediately before starting the handler.

AlexanderZ-Band · 2026-07-16T15:02:56Z

+        A cycle cancelled by interrupt/stop never actually ran the handler, so
+        it must not count against the message's retry budget.
+        """
+        self._attempts.pop(msg_id, None)


pop() removes every prior failed attempt, not just the control-cancelled attempt. Decrement by one so interrupting a retry does not reset the message's entire retry budget.

AlexanderZ-Band · 2026-07-16T17:14:50Z


+    # --- Control signals (interrupt/stop/play) ---
+
+    async def _handle_control(self, payload: AgentControlPayload) -> None:


This method does three things: dedup, room resolution, mode dispatch. Split out _is_duplicate_control(cid) and _resolve_control_rooms(payload) to make it easier to read.

AlexanderZ-Band · 2026-07-16T17:14:50Z

+            cid,
+        )
+
+        if payload.mode in ("interrupt", "stop"):


Use match/case here (repo convention in CLAUDE.md). Add a case _: that logs unknown modes — right now they are silently ignored.

AlexanderZ-Band · 2026-07-16T17:14:50Z

@@ -382,20 +403,33 @@ def _backlog_event(self, msg: PlatformMessage) -> MessageEvent:
        )

    async def _safe_handle_event(self, event: object) -> None:


Too many nested levels. Move the inner try/except up to the outer try — same behavior, flatter code.

AlexanderZ-Band · 2026-07-16T17:14:58Z

AgentRunner does forwarding, rehydration, and control signals in one ~800-line file. The control-signal part is self-contained — consider moving it to its own module as a follow-up.

AlexanderZ-Band · 2026-07-16T17:36:17Z

+import httpx
+import pytest
+import pytest_asyncio
+from thenvoi_rest.human_api_chats.types.create_my_chat_room_request_chat import (


This import fails — the package was renamed to band_rest (your conftest already uses it). The file doesn't collect: ModuleNotFoundError: No module named 'thenvoi_rest'.

AlexanderZ-Band · 2026-07-16T17:36:17Z

+        return await client.get(f"{get_base_url()}{path}", headers=_user_headers())
+
+
+async def _send_user_mention(chat_id: str, agent_id: str, text: str) -> str:


No need for raw httpx here — the typed client has human_api_messages.send_my_chat_message with mentions, and user_api_client is already available. Raw httpx is only justified for the control endpoints.

AlexanderZ-Band · 2026-07-16T17:36:17Z

+    return resp.json()["data"]["id"]
+
+
+async def _wait_until(


Generic polling helper, nothing control-specific — put it in tests/conftest_integration.py so the next integration test doesn't grow its own.

AlexanderZ-Band · 2026-07-16T17:36:17Z

+    return pred()
+
+
+class _ControllableHandler:


Same blocking-handler pattern (started/cancelled events + hang until cancelled) is hand-rolled many times in tests/runtime/test_execution_interrupt.py. Extract one shared fake and use it in both places.

AlexanderZ-Band · 2026-07-16T17:36:17Z

+
+
+@pytest_asyncio.fixture
+async def control_room(user_api_client, session_api_client, shared_agent1_info):


This repeats the reuse-or-create + is_room_alive pattern from shared_room in conftest_integration.py. Make it a shared user-owned room fixture there — future control/permission tests will need it too.

AlexanderZ-Band

See inline comments — main blockers: the in-flight cancellation gaps in the bridge (task registration ordering, play blocking the receive path), the interrupt/stop lifecycle windows in execution.py, and tests/integration/test_control_signals.py failing to import (thenvoi_rest).

amit-gazal-thenvoi added 3 commits July 15, 2026 15:02

wip

2024f63

amit-gazal-thenvoi and others added 2 commits July 16, 2026 16:11

amit-gazal-thenvoi requested a review from a team July 16, 2026 14:46

Merge remote-tracking branch 'origin/dev' into int-829-agentflow-remo…

d1573cf

…te-agents-honor-interrupt-and-stopresume # Conflicts: # src/band/client/streaming/client.py # src/band/platform/link.py # src/band/runtime/execution.py # src/band/runtime/platform_runtime.py # uv.lock

AlexanderZ-Band reviewed Jul 16, 2026

View reviewed changes

AlexanderZ-Band requested changes Jul 16, 2026

View reviewed changes


		# --- Control signals (interrupt/stop/play) ---

		async def _handle_control(self, payload: AgentControlPayload) -> None:

		@@ -382,20 +403,33 @@ def _backlog_event(self, msg: PlatformMessage) -> MessageEvent:
		)

		async def _safe_handle_event(self, event: object) -> None:

		return await client.get(f"{get_base_url()}{path}", headers=_user_headers())


		async def _send_user_mention(chat_id: str, agent_id: str, text: str) -> str:



		@pytest_asyncio.fixture
		async def control_room(user_api_client, session_api_client, shared_agent1_info):

Uh oh!

Conversation

amit-gazal-thenvoi commented Jul 16, 2026

Summary

Test plan

Uh oh!

linear-code Bot commented Jul 16, 2026

Uh oh!

AlexanderZ-Band Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AlexanderZ-Band commented Jul 16, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AlexanderZ-Band left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

AlexanderZ-Band Jul 16, 2026 •

edited

Loading