diff --git a/CHANGELOG.md b/CHANGELOG.md index ed6027494..d032a65aa 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/src/band/adapters/cursor_acp.py b/src/band/adapters/cursor_acp.py index 41a55195d..a7044b71a 100644 --- a/src/band/adapters/cursor_acp.py +++ b/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) diff --git a/src/band/integrations/acp/client_adapter.py b/src/band/integrations/acp/client_adapter.py index 8e24fdd23..abee3c5dc 100644 --- a/src/band/integrations/acp/client_adapter.py +++ b/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) diff --git a/tests/integrations/acp/acp_toolkit/agent.py b/tests/integrations/acp/acp_toolkit/agent.py index 6284f97dc..5b3a11942 100644 --- a/tests/integrations/acp/acp_toolkit/agent.py +++ b/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 [])} diff --git a/tests/integrations/acp/test_client_adapter.py b/tests/integrations/acp/test_client_adapter.py index 14cee3051..bf3f3da2e 100644 --- a/tests/integrations/acp/test_client_adapter.py +++ b/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 diff --git a/tests/integrations/acp/test_client_adapter_behavior.py b/tests/integrations/acp/test_client_adapter_behavior.py index b08cc1b62..dee344a71 100644 --- a/tests/integrations/acp/test_client_adapter_behavior.py +++ b/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