fix: retire a room's ACP process after it refuses to open a session

CHANGELOG.md

@@ -6,6 +6,13 @@ The format is based on [Conventional Commits](https://www.conventionalcommits.or
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 This changelog is automatically generated by [Release Please](https://github.com/googleapis/release-please).
 
+## [Unreleased]
+
+### Bug fixes
+
+* **acp:** retire a refused session setup only while it still owns the room, so
+  later turns use a fresh process without tearing down a newer session.
+
 ## [5.1.0](https://github.com/band-ai/band-sdk-python/compare/band-sdk-v5.0.0...band-sdk-v5.1.0) (2026-10-13)
 
 

src/band/adapters/cursor_acp.py

@@ -325,13 +325,7 @@ async def _get_or_create_session(
             self._cursor_profile.bind_session(session_id)
         return session_id, created
 
-    async def on_cleanup(
-        self, room_id: str, *, expected_runtime: ACPRuntime | None = None
-    ) -> None:
-        if expected_runtime is not None:
-            async with self._session_lock:
-                if self._runtimes.get(room_id) is not expected_runtime:
-                    return
+    def _on_room_retiring(self, room_id: str) -> None:
         # Wakes any decision _run_turn is parked on; the task itself keeps
         # running detached and winds down on its own (via _on_background_task_done)
         # once the runtime this stops out from under it closes the connection.
@@ -341,7 +335,6 @@ async def on_cleanup(
             turn.session_id = None
             self._cursor_profile.bind_session(None)
         self._cancel_room_decisions(room_id)
-        await super().on_cleanup(room_id, expected_runtime=expected_runtime)
 
     def _forget_session(self, session_id: str) -> None:
         self._cursor_profile.forget_session(session_id)

src/band/integrations/acp/client_adapter.py

@@ -1109,6 +1109,18 @@ async def _get_or_create_session(
 
         try:
             return await asyncio.shield(initializer.task)
+        except ACPConfigError:
+            raise
+        except Exception:
+            # The agent refused to open a session (e.g. an internal error from
+            # session/new). A process in that state can keep refusing every
+            # later session, so retire this room's runtime and the next turn
+            # spawns a fresh process. Only while this setup is still the
+            # room's own: one another turn started meanwhile owns the runtime.
+            await self._retire_room(
+                room_id, expected_runtime=runtime, failed_setup=initializer
+            )
+            raise
         finally:
             await self._release_session_initializer(room_id, initializer)
 
@@ -1498,12 +1510,36 @@ def _build_prompt_text(
     async def on_cleanup(
         self, room_id: str, *, expected_runtime: ACPRuntime | None = None
     ) -> None:
+        await self._retire_room(room_id, expected_runtime=expected_runtime)
+
+    def _on_room_retiring(self, room_id: str) -> None:
+        """Called under the session lock just before a room's resources go;
+        subclasses release per-room state that depends on them."""
+
+    async def _retire_room(
+        self,
+        room_id: str,
+        *,
+        expected_runtime: ACPRuntime | None = None,
+        failed_setup: SessionInitializer | None = None,
+    ) -> None:
+        """Release a room's session, setup and runtime. With ``expected_runtime``,
+        only while it is still the room's; with ``failed_setup``, only while no
+        other setup or published session has replaced it. Both checks and the
+        release happen under one lock, so no turn can slip in between."""
         async with self._session_lock:
             if (
                 expected_runtime is not None
                 and self._runtimes.get(room_id) is not expected_runtime
             ):
                 return
+            if failed_setup is not None:
+                current = self._session_initializers.get(room_id)
+                if (
+                    current is not None and current is not failed_setup
+                ) or room_id in self._room_to_session:
+                    return
+            self._on_room_retiring(room_id)
             session = self._room_to_session.pop(room_id, None)
             initializer = self._session_initializers.pop(room_id, None)
             self._room_tools.pop(room_id, None)

tests/integrations/acp/acp_toolkit/agent.py

@@ -88,6 +88,10 @@ def __init__(
         self._supports_session_load = supports_session_load
         self._persisted_sessions: set[str] = set()
         self._session_load_error: RequestError | None = None
+        # The connection whose process refuses every session/new, modelling an
+        # agent process left broken; a fresh connection is unaffected.
+        self._wedges_next_connection = False
+        self._wedged_conn: AgentSideConnection | None = None
         # A room-owned ACPRuntime opens its own connection, so a genuinely
         # concurrent multi-room turn (see test_independent_rooms_configure_
         # without_waiting_for_each_other) can have more than one live
@@ -221,6 +225,12 @@ def breaks_session_load(self, error: RequestError | None = None) -> FakeACPAgent
         self._session_load_error = error or RequestError.internal_error()
         return self
 
+    def wedges_its_process(self) -> FakeACPAgent:
+        """Make the current process refuse every ``session/new`` with an
+        internal error for as long as it lives; a respawned process works."""
+        self._wedges_next_connection = True
+        return self
+
     def will_stream(self, *parts: str) -> FakeACPAgent:
         """Emit several agent_message_chunk deltas in one turn (streaming reply)."""
 
@@ -621,6 +631,11 @@ async def new_session(
         self, cwd: str, mcp_servers: Any = None, **kwargs: Any
     ) -> NewSessionResponse:
         del kwargs
+        if self._wedges_next_connection:
+            self._wedges_next_connection = False
+            self._wedged_conn = self._current_conn
+        if self._wedged_conn is not None and self._current_conn is self._wedged_conn:
+            raise RequestError.internal_error()
         sid = f"fake-session-{len(self.sessions) + 1}"
         self.sessions.append(
             {"session_id": sid, "cwd": cwd, "mcp_servers": list(mcp_servers or [])}

tests/integrations/acp/test_client_adapter.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import logging
+from contextlib import asynccontextmanager
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 from urllib.parse import urlsplit
@@ -12,6 +13,8 @@
 from acp.exceptions import RequestError
 from acp.helpers import update_agent_message_text
 from acp.schema import (
+    AgentCapabilities,
+    InitializeResponse,
     McpServerStdio,
     NewSessionResponse,
     PermissionOption,
@@ -35,7 +38,7 @@
     _resolve_launcher,
 )
 from band.integrations.acp.client_profiles import CursorACPClientProfile
-from band.integrations.acp.client_runtime import ACPCollectingClient
+from band.integrations.acp.client_runtime import ACPCollectingClient, ACPRuntime
 from band.integrations.acp.client_types import (
     ACPClientSessionState,
     BandACPClient,
@@ -2312,3 +2315,106 @@ def test_retired_acp_delivery_mode_is_rejected() -> None:
         ACPClientAdapterConfig.model_validate(
             {"command": "codex", "assistant_text_mode": "thought"}
         )
+
+
+async def _overlap_with_failed_setup(
+    workspace: Path, *, hold_configuration: bool
+) -> tuple[BaseException | None, BaseException | None, AsyncMock, bool]:
+    """Turn one's session/new fails while turn two already waits to set up the
+    same room, and turn two starts its own setup before turn one hears of the
+    failure. Returns both outcomes, the connection, and whether turn two's held
+    session configuration was cancelled."""
+    arrival = asyncio.Event()
+    second_waiting = asyncio.Event()
+    release_configuration = asyncio.Event()
+    configuration_cancelled = False
+    requests = 0
+    conn = AsyncMock()
+    conn.initialize.return_value = InitializeResponse(
+        protocol_version=1, agent_capabilities=AgentCapabilities()
+    )
+
+    async def new_session(**kwargs: object) -> NewSessionResponse:
+        nonlocal requests
+        requests += 1
+        if requests == 1:
+            # Wake the waiting second turn before this setup fails, so it runs
+            # before the first turn's waiter resumes.
+            arrival.set()
+            raise RequestError.internal_error()
+        return NewSessionResponse(session_id="newer-session")
+
+    conn.new_session = new_session
+
+    @asynccontextmanager
+    async def spawn(*args: object, **kwargs: object):
+        yield conn, None
+
+    async def configure(request: object) -> None:
+        nonlocal configuration_cancelled
+        if hold_configuration:
+            try:
+                await release_configuration.wait()
+            except asyncio.CancelledError:
+                configuration_cancelled = True
+                raise
+
+    adapter = ACPClientAdapter(
+        ACPClientAdapterConfig(
+            command=("fake-agent",), inject_band_tools=False, cwd=str(workspace)
+        ),
+        resolve_session_config=configure,
+    )
+    adapter._build_runtime = lambda workspace=None: ACPRuntime(
+        command=["fake-agent"],
+        cwd=workspace,
+        spawn_process=spawn,
+        pass_builtin_transport_options=False,
+    )
+    await adapter.on_started("Probe", "")
+
+    async def send(content: str) -> None:
+        await adapter.on_message(
+            make_platform_message(content, room_id="room-1"),
+            FakeAgentTools(),
+            ACPClientSessionState(),
+            None,
+            None,
+            is_session_bootstrap=False,
+            room_id="room-1",
+        )
+
+    async def second_turn() -> None:
+        second_waiting.set()
+        await arrival.wait()
+        await send("second")
+
+    try:
+        async with asyncio.timeout(3):
+            second = asyncio.create_task(second_turn())
+            await second_waiting.wait()
+            first = asyncio.create_task(send("first"))
+            (first_result,) = await asyncio.gather(first, return_exceptions=True)
+            release_configuration.set()
+            (second_result,) = await asyncio.gather(second, return_exceptions=True)
+    finally:
+        release_configuration.set()
+        await adapter.stop()
+    return first_result, second_result, conn, configuration_cancelled
+
+
+@pytest.mark.parametrize("hold_configuration", [False, True])
+async def test_a_failed_setup_never_tears_down_a_newer_one(
+    tmp_path: Path, hold_configuration: bool
+) -> None:
+    """A room's runtime is retired after its session setup fails, but only while
+    that failed setup is still the room's own: a setup another turn started on
+    the room meanwhile keeps its runtime, and its turn runs."""
+    first, second, conn, cancelled = await _overlap_with_failed_setup(
+        tmp_path, hold_configuration=hold_configuration
+    )
+
+    assert isinstance(first, RequestError)
+    assert second is None
+    assert conn.prompt.await_count == 1
+    assert not cancelled

tests/integrations/acp/test_client_adapter_behavior.py

@@ -950,6 +950,22 @@ async def test_replay_injected_when_session_load_errors() -> None:
     assert reply.thoughts == ["Blue."], "the turn must complete despite the load error"
 
 
+@pytest.mark.asyncio
+async def test_a_room_recovers_after_its_agent_process_refuses_a_session() -> None:
+    """An agent process can refuse every session/new once it is broken. The turn
+    that hit it fails, and the room's next message must reach a fresh process
+    instead of the broken one."""
+    agent = FakeACPAgent().will_say("Recovered.")
+    agent.wedges_its_process()
+
+    async with acp_adapter(agent) as session:
+        with pytest.raises(RequestError):
+            await session.send("first", room="room-1")
+        reply = await session.send("second", room="room-1")
+
+    assert reply.thoughts == ["Recovered."]
+
+
 @pytest.mark.asyncio
 async def test_no_replay_when_remote_session_loads() -> None:
     """A restored session already holds the conversation remotely; replaying the