fix: stop remote adapter turns through the shared cancellation owner

docs/adapters/opencode.md

@@ -36,7 +36,12 @@ since the ask carries its own `approval_wait_timeout_s` expiry.
 A manual ask also releases the turn's delivery early (it is marked processed
 while the human decides), so the adapter judges that turn itself when OpenCode
 finishes it: a turn that ends without a reply then posts one missing-reply
-failure. A turn cancelled by cleanup or an interrupt posts nothing.
+failure. A turn ended by cleanup, a STOP or an interrupt, even one already
+released for approval, posts nothing: its watcher is cancelled first, then its
+OpenCode session is aborted once through the adapter's `backend_turns`. A turn
+that already ended on its own, by finishing or by timing out, still posts that
+outcome when a STOP lands while it is being posted; the STOP only finishes the
+abort.
 
 New ordinary work remains outstanding while the session is occupied. Approval,
 status and question replies remain reachable through the shared control hook;

pyproject.toml

@@ -63,7 +63,10 @@ opencode = [
     "mcp>=1.28.1,<2",
 ]
 letta = [
-    "letta-client>=1.0.0",
+    # Floor is `conversations.cancel` (shared-mode STOP), first shipped in
+    # 1.7.1; the `conversations` resource itself arrived in 1.6.5 -- checked
+    # against the PyPI wheels 2026-10-19.
+    "letta-client>=1.7.1",
     # Floor pinned to crewai's own transitive pin (mcp~=1.28.1) -- see `desktop` above.
     "mcp>=1.28.1,<2",
 ]
@@ -245,7 +248,7 @@ dev = [
     # Include Agno for testing
     "agno>=2.6.0",
     # Include letta-client for testing
-    "letta-client>=1.0.0",
+    "letta-client>=1.7.1",
     # Include bridge deps for testing
     "aiohttp>=3.9,<4",
     "python-dotenv>=1.2.2",

src/band/adapters/copilot_sdk.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 import re
@@ -33,6 +34,7 @@
 from band.core.protocols import GENERIC_PROVIDER_FAILURE_MESSAGE, send_event_safe
 from band.core.simple_adapter import SimpleAdapter
 from band.core.tool_filter import filter_tool_schemas
+from band.core.turn_lifecycle import BackendTurns
 from band.core.types import Capability, Emit, MessageType, ToolEventKey, TurnUsage
 from band.integrations.copilot_sdk import CopilotSessionManager
 from band.integrations.copilot_sdk.prompts import TURN_COMPLETION_GUIDANCE
@@ -72,6 +74,7 @@
         AssistantTurnRetryData,
         AssistantUsageData,
         ModelCallFailureData,
+        SessionIdleData,
         SessionWarningData,
         ToolExecutionCompleteData,
     )
@@ -170,6 +173,16 @@ class TurnState:
     incidents: list[str] = field(default_factory=list)
 
 
+@dataclass
+class SessionTurn:
+    """The session a stop aborts, and whether its prompt is still running."""
+
+    session: CopilotSession
+    # Set when the prompt is sent, cleared once it returns; a prompt that is
+    # cancelled or fails leaves it set.
+    prompting: bool = False
+
+
 def _turn_incident(data: object) -> str | None:
     """A one-line account of a session event that can explain a silent turn."""
     match data:
@@ -319,6 +332,9 @@ def __init__(
         self._room_tools: dict[str, AgentToolsProtocol] = {}
         self._session_ids: dict[str, RoomSessionIds] = {}
         self._turn_state: dict[str, TurnState] = {}
+        self._backend_turns: BackendTurns[SessionTurn] = BackendTurns(
+            self._abort_turn, provider=_PROVIDER
+        )
         self._system_prompt: str = ""
 
     # --- Lifecycle -------------------------------------------------------
@@ -395,6 +411,7 @@ async def on_cleanup(self, room_id: str) -> None:
         rooms); ``cleanup_all`` stops it at shutdown (unless it was borrowed
         via ``client=`` — then its owner stops it).
         """
+        await self.backend_turns.settled(room_id)
         if self._session_manager:
             await self._session_manager.cleanup_session(room_id)
         self._room_tools.pop(room_id, None)
@@ -403,6 +420,7 @@ async def on_cleanup(self, room_id: str) -> None:
 
     async def cleanup_all(self) -> None:
         """Stop all sessions and, if adapter-owned, the Copilot client."""
+        await self.backend_turns.drain()
         if self._session_manager:
             await self._session_manager.stop()
         self._room_tools.clear()
@@ -411,6 +429,10 @@ async def cleanup_all(self) -> None:
 
     # --- Message handling --------------------------------------------------
 
+    @property
+    def backend_turns(self) -> BackendTurns[SessionTurn]:
+        return self._backend_turns
+
     async def on_message(
         self,
         msg: PlatformMessage,
@@ -433,52 +455,61 @@ async def on_message(
 
         # Same-session calls must not interleave; other rooms run concurrently.
         async with self._session_manager.turn_lock(room_id):
-            # Bound only once this message holds the room, so a queued
-            # message never redirects the running turn's tool calls.
-            self._room_tools[room_id] = tools
-            try:
-                session, inject_text = await self._obtain_session(
-                    room_id, history, tools, is_session_bootstrap=is_session_bootstrap
-                )
-            except Exception:
-                logger.exception("Room %s: Copilot session setup failed", room_id)
-                await tools.send_failure(
-                    AgentFailure(_PROVIDER, GENERIC_PROVIDER_FAILURE_MESSAGE)
+            # Abort is what stops the runtime's work for a cancelled turn; the
+            # session stays valid for the next message.
+            async with self.backend_turns.running(room_id):
+                # Bound once the previous turn's abort has settled and this
+                # message holds the room, so neither a queued message nor a
+                # late call from the stopped turn reaches this turn's tools.
+                self._room_tools[room_id] = tools
+                try:
+                    session, inject_text = await self._obtain_session(
+                        room_id,
+                        history,
+                        tools,
+                        is_session_bootstrap=is_session_bootstrap,
+                    )
+                except Exception:
+                    logger.exception("Room %s: Copilot session setup failed", room_id)
+                    await tools.send_failure(
+                        AgentFailure(_PROVIDER, GENERIC_PROVIDER_FAILURE_MESSAGE)
+                    )
+                    raise
+                live = SessionTurn(session)
+                self.backend_turns.bind(room_id, live)
+                prompt = self._compose_prompt(
+                    msg,
+                    participants_msg,
+                    contacts_msg,
+                    inject_text=inject_text,
                 )
-                raise
-            prompt = self._compose_prompt(
-                msg,
-                participants_msg,
-                contacts_msg,
-                inject_text=inject_text,
-            )
 
-            # The session's ask_user handler reads this slot; safe because the
-            # turn lock serializes turns per room.
-            turn = TurnState(
-                sender_mention={
-                    "id": msg.sender_id,
-                    "name": msg.sender_name or msg.sender_type,
-                }
-            )
-            self._turn_state[room_id] = turn
-            try:
-                final_text = await self._run_turn(session, prompt, turn)
-            except Exception:
-                logger.exception("Room %s: Copilot turn failed", room_id)
-                # Abort any work the runtime is still doing for this turn and
-                # drop the session; the next message resumes it fresh by id.
-                await self._session_manager.evict_session(room_id)
-                await tools.send_failure(
-                    AgentFailure(_PROVIDER, GENERIC_PROVIDER_FAILURE_MESSAGE)
+                # The session's ask_user handler reads this slot; safe because the
+                # turn lock serializes turns per room.
+                turn = TurnState(
+                    sender_mention={
+                        "id": msg.sender_id,
+                        "name": msg.sender_name or msg.sender_type,
+                    }
                 )
-                raise
-            finally:
-                self._turn_state.pop(room_id, None)
-                # No-op unless Emit.USAGE is on / usage is non-empty; best-effort,
-                # never raises. In the finally so a turn that errors or times out
-                # AFTER a model call still reports the tokens it spent.
-                await self.emit_usage(tools, turn.usage)
+                self._turn_state[room_id] = turn
+                try:
+                    final_text = await self._run_turn(live, prompt, turn)
+                except Exception:
+                    logger.exception("Room %s: Copilot turn failed", room_id)
+                    # Abort any work the runtime is still doing for this turn and
+                    # drop the session; the next message resumes it fresh by id.
+                    await self._session_manager.evict_session(room_id)
+                    await tools.send_failure(
+                        AgentFailure(_PROVIDER, GENERIC_PROVIDER_FAILURE_MESSAGE)
+                    )
+                    raise
+                finally:
+                    self._turn_state.pop(room_id, None)
+                    # No-op unless Emit.USAGE is on / usage is non-empty; best-effort,
+                    # never raises. In the finally so a turn that errors or times out
+                    # AFTER a model call still reports the tokens it spent.
+                    await self.emit_usage(tools, turn.usage)
 
             await self._emit_thoughts(turn, tools)
 
@@ -889,8 +920,31 @@ def _compose_prompt(
         parts.append(msg.format_for_llm())
         return "\n\n".join(parts)
 
+    @staticmethod
+    async def _abort_turn(live: SessionTurn) -> None:
+        """Abort the prompt and, while one is running, wait for the idle that ends it.
+
+        The CLI emits that idle just after acknowledging the abort, and
+        ``send_and_wait`` ends on any idle: a next prompt sent before it
+        arrives returns on it with no reply. Subscribing first also catches
+        an idle that arrives ahead of the acknowledgement.
+        """
+        idled = asyncio.Event()
+
+        def on_event(event: SessionEvent) -> None:
+            if isinstance(event.data, SessionIdleData):
+                idled.set()
+
+        unsubscribe = live.session.on(on_event)
+        try:
+            await live.session.abort()
+            if live.prompting:
+                await idled.wait()
+        finally:
+            unsubscribe()
+
     async def _run_turn(
-        self, session: CopilotSession, prompt: str, turn: TurnState
+        self, live: SessionTurn, prompt: str, turn: TurnState
     ) -> str | None:
         """Send one prompt, wait for idle, and return the final assistant text.
 
@@ -908,13 +962,15 @@ def collect(event: SessionEvent) -> None:
             elif incident := _turn_incident(data):
                 turn.incidents.append(incident)
 
-        unsubscribe = session.on(collect)
+        unsubscribe = live.session.on(collect)
+        live.prompting = True
         try:
-            final_event = await session.send_and_wait(
+            final_event = await live.session.send_and_wait(
                 prompt, timeout=self.config.turn_timeout_s
             )
         finally:
             unsubscribe()
+        live.prompting = False
 
         content = getattr(final_event.data, "content", None) if final_event else None
         # An empty final message means no reply — same as no final event.