diff --git a/openhands-sdk/openhands/sdk/agent/acp_agent.py b/openhands-sdk/openhands/sdk/agent/acp_agent.py index 8bd2295d72..b13d980635 100644 --- a/openhands-sdk/openhands/sdk/agent/acp_agent.py +++ b/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) diff --git a/tests/sdk/agent/test_acp_agent.py b/tests/sdk/agent/test_acp_agent.py index 89f7e5b950..b548349859 100644 --- a/tests/sdk/agent/test_acp_agent.py +++ b/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