From b88d5344bf69f55d3d74863fcbdb00b63cd35331 Mon Sep 17 00:00:00 2001 From: onatozmenn Date: Mon, 3 Aug 2026 14:19:48 +0300 Subject: [PATCH 1/2] fix(acp): abort the turn immediately when close() lands mid-prompt close() runs on whatever thread calls it and drops _executor while a concurrent step()/astep() may be between its own None check and the call, so the turn died with a confusing AttributeError: 'NoneType' object has no attribute 'run_async'. Killing the subprocess also surfaces as a retriable connection error first, so the turn slept through a full retry delay (5s by default, 50s across all attempts on the async path) before failing, even though a closed agent can never succeed. close() is the only way to abort a step() that is blocked for the whole turn, so that delay defeats it. Read the executor once behind _require_executor() and stop retrying as soon as _closed is set. The turn now fails immediately with ACPAgentClosedError, surfaced as an ACPAgentClosed error event. Fixes #4329 Signed-off-by: onatozmenn --- .../openhands/sdk/agent/acp_agent.py | 41 ++++++- tests/sdk/agent/test_acp_agent.py | 106 ++++++++++++++++++ 2 files changed, 145 insertions(+), 2 deletions(-) diff --git a/openhands-sdk/openhands/sdk/agent/acp_agent.py b/openhands-sdk/openhands/sdk/agent/acp_agent.py index 8bd2295d72..2747ddefcb 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,7 +3728,7 @@ async def astep( self.acp_prompt_timeout, len(prompt_blocks), ) - portal = self._executor.portal + portal = self._require_executor().portal response: PromptResponse | None = None max_retries = _ACP_PROMPT_MAX_RETRIES @@ -3722,6 +3754,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 +3779,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..25759b1f74 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,111 @@ 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 + # --------------------------------------------------------------------------- # Gemini-specific tests From b296caab635504d027a664db5db319d505fc1ef4 Mon Sep 17 00:00:00 2001 From: onatozmenn Date: Mon, 3 Aug 2026 22:30:09 +0300 Subject: [PATCH 2/2] fix(acp): re-read the executor per attempt in astep astep read the portal once before the retry loop while step re-read the executor on every attempt. Teardown landing between two attempts would then hit a dead portal, and anyio raises a plain RuntimeError there, which is not retriable and so bypasses the _closed check and classifies as ACPPromptError rather than ACPAgentClosed. Unreachable through close() today, because _closed is set before the teardown that produces the first retriable error, so the loop never reaches a second attempt. Making both paths read the executor the same way removes the asymmetry rather than relying on that ordering. Raised by the automated review on #4334. Signed-off-by: onatozmenn --- .../openhands/sdk/agent/acp_agent.py | 6 ++- tests/sdk/agent/test_acp_agent.py | 44 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/openhands-sdk/openhands/sdk/agent/acp_agent.py b/openhands-sdk/openhands/sdk/agent/acp_agent.py index 2747ddefcb..b13d980635 100644 --- a/openhands-sdk/openhands/sdk/agent/acp_agent.py +++ b/openhands-sdk/openhands/sdk/agent/acp_agent.py @@ -3728,12 +3728,14 @@ async def astep( self.acp_prompt_timeout, len(prompt_blocks), ) - portal = self._require_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 diff --git a/tests/sdk/agent/test_acp_agent.py b/tests/sdk/agent/test_acp_agent.py index 25759b1f74..b548349859 100644 --- a/tests/sdk/agent/test_acp_agent.py +++ b/tests/sdk/agent/test_acp_agent.py @@ -5953,6 +5953,50 @@ def start_task_soon(self, fn, *args): # noqa: ANN001, ANN202 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