fix(acp): abort the turn immediately when close() lands mid-prompt

openhands-sdk/openhands/sdk/agent/acp_agent.py

@@ -149,6 +149,16 @@
 # Exception types that indicate transient connection issues worth retrying
 _RETRIABLE_CONNECTION_ERRORS = (OSError, ConnectionError, BrokenPipeError, EOFError)
 
+
+class ACPAgentClosedError(RuntimeError):
+    """Raised when the ACP runtime is torn down while a prompt is in flight.
+
+    ``close()`` runs on whatever thread calls it, so it can land while another
+    thread is inside ``step()``/``astep()``. Subclasses ``RuntimeError`` so
+    callers that already handle the pre-session ``RuntimeError`` keep working.
+    """
+
+
 # ``npm run`` exports package/lifecycle configuration into descendants. ACP
 # providers launched through ``npx`` must not inherit that context, otherwise npm
 # can try to resolve against the parent package instead of the requested one.
@@ -1026,6 +1036,8 @@ def _classify_acp_turn_error(exc: BaseException) -> str:
     policy refusals get their own code, credential failures map to ``ACPAuthRequired``
     (so the client can offer re-auth), everything else is a generic ``ACPPromptError``.
     """
+    if isinstance(exc, ACPAgentClosedError):
+        return "ACPAgentClosed"
     if isinstance(exc, ACPFileCredentialSyncError):
         return "ACPPromptError"
     if isinstance(exc, ACPFileCredentialNeedsReauthError):
@@ -1989,6 +2001,21 @@ def has_live_acp_session(self) -> bool:
             and self._executor is not None
         )
 
+    def _require_executor(self) -> Any:
+        """Return the live executor, or raise if teardown already started.
+
+        Reads ``_executor`` once so a concurrent :meth:`close` cannot null it
+        out between the check and the call. ``_closed`` is set before teardown
+        begins, so it catches the window where the executor is still assigned
+        but its portal is going away.
+        """
+        executor = self._executor
+        if executor is None or self._closed:
+            raise ACPAgentClosedError(
+                "ACP agent was closed while a prompt was in flight"
+            )
+        return executor
+
     def get_all_llms(self) -> Generator[LLM]:
         yield self.llm
 
@@ -3566,11 +3593,15 @@ async def _prompt() -> PromptResponse | None:
 
             for attempt in range(max_retries + 1):
                 try:
-                    response = self._executor.run_async(_prompt)
+                    response = self._require_executor().run_async(_prompt)
                     break
                 except TimeoutError:
                     raise
                 except _RETRIABLE_CONNECTION_ERRORS as e:
+                    if self._closed:
+                        raise ACPAgentClosedError(
+                            "ACP agent was closed while a prompt was in flight"
+                        ) from e
                     if attempt < max_retries:
                         delay = _ACP_PROMPT_RETRY_DELAYS[
                             min(attempt, len(_ACP_PROMPT_RETRY_DELAYS) - 1)
@@ -3595,6 +3626,7 @@ async def _prompt() -> PromptResponse | None:
                     if (
                         e.code in _RETRIABLE_SERVER_ERROR_CODES
                         and attempt < max_retries
+                        and not self._closed
                     ):
                         delay = _ACP_PROMPT_RETRY_DELAYS[
                             min(attempt, len(_ACP_PROMPT_RETRY_DELAYS) - 1)
@@ -3696,12 +3728,14 @@ async def astep(
                 self.acp_prompt_timeout,
                 len(prompt_blocks),
             )
-            portal = self._executor.portal
-
             response: PromptResponse | None = None
             max_retries = _ACP_PROMPT_MAX_RETRIES
             for attempt in range(max_retries + 1):
                 try:
+                    # Re-read per attempt, like step() does, so a teardown
+                    # between retries surfaces as a closed agent rather than
+                    # anyio's "portal is not running".
+                    portal = self._require_executor().portal
                     # Schedule the ACP prompt on the portal loop (where the
                     # connection lives); await the future back on the caller
                     # loop.  Shield the portal task from wait_for timeout so
@@ -3722,6 +3756,10 @@ async def astep(
                 except TimeoutError:
                     raise
                 except _RETRIABLE_CONNECTION_ERRORS as e:
+                    if self._closed:
+                        raise ACPAgentClosedError(
+                            "ACP agent was closed while a prompt was in flight"
+                        ) from e
                     if attempt < max_retries:
                         delay = _ACP_PROMPT_RETRY_DELAYS[
                             min(attempt, len(_ACP_PROMPT_RETRY_DELAYS) - 1)
@@ -3743,6 +3781,7 @@ async def astep(
                     if (
                         e.code in _RETRIABLE_SERVER_ERROR_CODES
                         and attempt < max_retries
+                        and not self._closed
                     ):
                         delay = _ACP_PROMPT_RETRY_DELAYS[
                             min(attempt, len(_ACP_PROMPT_RETRY_DELAYS) - 1)

tests/sdk/agent/test_acp_agent.py

@@ -26,6 +26,7 @@
 import openhands.sdk.utils.files as files_module
 from openhands.sdk.agent.acp_agent import (
     ACPAgent,
+    ACPAgentClosedError,
     _acp_error_detail,
     _acp_error_indicates_auth,
     _apply_acp_model,
@@ -5847,6 +5848,155 @@ def _fake_run_async(_coro, **_kwargs):
         assert call_count == 4
         assert conversation.state.execution_status == ConversationExecutionStatus.ERROR
 
+    def test_close_during_prompt_aborts_without_retry_delay(self, tmp_path):
+        """``close()`` landing mid-prompt aborts now instead of sleeping.
+
+        Killing the subprocess surfaces as a retriable connection error, so
+        without the ``_closed`` check the turn waits out a full retry delay
+        before failing — defeating the point of closing to abort the turn.
+        """
+        agent = _make_agent()
+        conversation = self._make_conversation_with_message(tmp_path)
+        events: list = []
+
+        mock_client = _OpenHandsACPBridge()
+        agent._client = mock_client
+        agent._conn = MagicMock()
+        agent._session_id = "test-session"
+
+        call_count = 0
+
+        def _fake_run_async(_coro, **_kwargs):
+            nonlocal call_count
+            call_count += 1
+            # What close() does on the other thread: _closed is set before
+            # teardown, then the runtime references are dropped.
+            agent._closed = True
+            agent._executor = None
+            raise ConnectionError("Connection closed")
+
+        mock_executor = MagicMock()
+        mock_executor.run_async = _fake_run_async
+        agent._executor = mock_executor
+
+        with patch("openhands.sdk.agent.acp_agent.time.sleep") as mock_sleep:
+            with pytest.raises(ACPAgentClosedError):
+                agent.step(conversation, on_event=events.append)
+
+        assert call_count == 1
+        mock_sleep.assert_not_called()
+        assert conversation.state.execution_status == ConversationExecutionStatus.ERROR
+        assert [e.code for e in events if isinstance(e, ConversationErrorEvent)] == [
+            "ACPAgentClosed"
+        ]
+
+    def test_step_after_teardown_reports_closed_agent(self, tmp_path):
+        """A torn-down executor is reported as a closed agent, not a None deref."""
+        agent = _make_agent()
+        conversation = self._make_conversation_with_message(tmp_path)
+
+        agent._client = _OpenHandsACPBridge()
+        agent._conn = MagicMock()
+        agent._session_id = "test-session"
+        agent._executor = None
+
+        with pytest.raises(ACPAgentClosedError):
+            agent.step(conversation, on_event=lambda _: None)
+
+        assert conversation.state.execution_status == ConversationExecutionStatus.ERROR
+
+    def test_astep_after_teardown_reports_closed_agent(self, tmp_path):
+        """Async path guards the executor read the same way as :meth:`step`."""
+        agent = _make_agent()
+        conversation = self._make_conversation_with_message(tmp_path)
+
+        agent._client = _OpenHandsACPBridge()
+        agent._conn = MagicMock()
+        agent._session_id = "test-session"
+        agent._executor = None
+
+        with pytest.raises(ACPAgentClosedError):
+            asyncio.run(agent.astep(conversation, on_event=lambda _: None))
+
+        assert conversation.state.execution_status == ConversationExecutionStatus.ERROR
+
+    def test_close_during_astep_aborts_without_retry_delay(self, tmp_path):
+        """Async retry loop also stops as soon as teardown starts."""
+        agent = _make_agent()
+        conversation = self._make_conversation_with_message(tmp_path)
+        events: list = []
+
+        mock_client = _OpenHandsACPBridge()
+        agent._client = mock_client
+        agent._conn = MagicMock()
+        agent._session_id = "test-session"
+
+        call_count = 0
+
+        class _ClosingPortal:
+            def start_task_soon(self, fn, *args):  # noqa: ANN001, ANN202
+                nonlocal call_count
+                call_count += 1
+                agent._closed = True
+                agent._executor = None
+                failed: Future = Future()
+                failed.set_exception(ConnectionError("Connection closed"))
+                return failed
+
+        mock_executor = MagicMock()
+        mock_executor.portal = _ClosingPortal()
+        agent._executor = mock_executor
+
+        with pytest.raises(ACPAgentClosedError):
+            asyncio.run(agent.astep(conversation, on_event=events.append))
+
+        assert call_count == 1
+        assert conversation.state.execution_status == ConversationExecutionStatus.ERROR
+
+    def test_astep_reports_closed_agent_when_teardown_lands_between_retries(
+        self, tmp_path
+    ):
+        """The async path re-reads the executor per attempt, like ``step``.
+
+        Teardown that lands after a retriable failure leaves a dead portal
+        behind; reading it once up front would surface anyio's "portal is not
+        running" as a generic prompt error instead of a closed agent.
+        """
+        agent = _make_agent()
+        conversation = self._make_conversation_with_message(tmp_path)
+        events: list = []
+
+        mock_client = _OpenHandsACPBridge()
+        agent._client = mock_client
+        agent._conn = MagicMock()
+        agent._session_id = "test-session"
+
+        call_count = 0
+
+        class _PortalTornDownAfterFirstAttempt:
+            def start_task_soon(self, fn, *args):  # noqa: ANN001, ANN202
+                nonlocal call_count
+                call_count += 1
+                agent._executor = None
+                failed: Future = Future()
+                failed.set_exception(ConnectionError("Connection reset by peer"))
+                return failed
+
+        mock_executor = MagicMock()
+        mock_executor.portal = _PortalTornDownAfterFirstAttempt()
+        agent._executor = mock_executor
+
+        with patch(
+            "openhands.sdk.agent.acp_agent._ACP_PROMPT_RETRY_DELAYS", (0.0, 0.0, 0.0)
+        ):
+            with pytest.raises(ACPAgentClosedError):
+                asyncio.run(agent.astep(conversation, on_event=events.append))
+
+        assert call_count == 1
+        assert [e.code for e in events if isinstance(e, ConversationErrorEvent)] == [
+            "ACPAgentClosed"
+        ]
+
 
 # ---------------------------------------------------------------------------
 # Gemini-specific tests