fix(sdk): wait boundedly for late ACP read_only mode confirmation

AGENTS.md

@@ -85,6 +85,7 @@ All pull requests must comply with [`.agents/skills/custom-codereview-guide.md`]
 - Agent-server persistence detects whether settings or secret payloads contain values by exercising the real Pydantic secret-serialization pipeline with `_SecretProbeCipher` (`persistence/models.py`). Do not replace this with a hardcoded field list or a recursive `SecretStr` walk: fields such as `AgentContext.secrets` store plain strings and only become secret-bearing inside their serializers.
 
 - `LookupSecret` normalizes hostless URLs against `OH_INTERNAL_SERVER_URL` (set by `openhands-agent-server.__main__` from the bound host/port, rewriting wildcard binds to loopback) and otherwise falls back to `http://127.0.0.1:8000`, so relative secret URLs can safely target the current agent-server instance.
+- read_only ACP session-mode confirmation (`_apply_acp_session_mode`) waits at most `_ACP_SESSION_MODE_CONFIRM_TIMEOUT` (5s) on a per-session `asyncio.Event` in `_OpenHandsACPBridge.wait_for_current_mode` when `set_mode` returned with no observed mode and the advertised current mode differs; a different observed mode or a timeout still raises `ACPSessionModeError` (fail-closed, no provider/CLI version conditions). `_verify_read_only_mode_after_config` is deliberately unchanged (sync, generation-based, post-config), so the wait applies only to the initial `session/set_mode` confirmation.
 
 
 

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

@@ -217,6 +217,10 @@
     os.environ.get("ACP_RUNTIME_SHUTDOWN_TIMEOUT", "5.0")
 )
 
+# Upper bound on how long a read_only session/set_mode waits for its
+# CurrentModeUpdate confirmation when the notification trails the response.
+_ACP_SESSION_MODE_CONFIRM_TIMEOUT: float = 5.0
+
 # Minimum interval between on_activity heartbeat signals (seconds).
 # Throttled to avoid excessive calls while still keeping the idle timer
 # well below the ~20 min runtime-api kill threshold.
@@ -873,9 +877,16 @@ async def _apply_acp_session_mode(
     observed = client.get_current_mode_id(session_id)
     if observed == mode_id or (observed is None and current_mode_id == mode_id):
         return
+    waited = ""
+    if observed is None:
+        timeout = _ACP_SESSION_MODE_CONFIRM_TIMEOUT
+        observed = await client.wait_for_current_mode(session_id, timeout)
+        if observed == mode_id:
+            return
+        waited = f" within {timeout:g}s" if observed is None else ""
     raise ACPSessionModeError(
         f"ACP server {agent_name!r} session {session_id} did not confirm "
-        f"read_only session mode {mode_id!r} "
+        f"read_only session mode {mode_id!r}{waited} "
         f"(advertised current={current_mode_id!r}, observed={observed!r})."
     )
 
@@ -2021,6 +2032,7 @@ def __init__(self, permission_policy: ACPPermissionPolicy = "writable") -> None:
         self._config_options_by_session: dict[str, list[SessionConfigOption]] = {}
         self._current_mode_by_session: dict[str, str] = {}
         self._current_mode_generation_by_session: dict[str, int] = {}
+        self._current_mode_updated: dict[str, asyncio.Event] = {}
         # Fork session state for ask_agent() — guarded by _fork_lock to
         # prevent concurrent ask_agent() calls from colliding.
         self._fork_lock = threading.Lock()
@@ -2275,6 +2287,18 @@ def get_current_mode_id(self, session_id: str) -> str | None:
         """Return the last confirmed session mode id for *session_id*."""
         return self._current_mode_by_session.get(session_id)
 
+    async def wait_for_current_mode(
+        self, session_id: str, timeout: float
+    ) -> str | None:
+        """Wait up to *timeout* seconds for a CurrentModeUpdate for *session_id*.
+
+        Returns the confirmed mode id, or ``None`` if none arrived in time.
+        """
+        event = self._current_mode_updated.setdefault(session_id, asyncio.Event())
+        with contextlib.suppress(TimeoutError):
+            await asyncio.wait_for(event.wait(), timeout)
+        return self.get_current_mode_id(session_id)
+
     def get_current_mode_generation(self, session_id: str) -> int:
         """Return how many CurrentModeUpdate notifications this session has seen."""
         return self._current_mode_generation_by_session.get(session_id, 0)
@@ -2367,6 +2391,7 @@ async def session_update(
                 self._current_mode_generation_by_session[session_id] = (
                     self._current_mode_generation_by_session.get(session_id, 0) + 1
                 )
+                self._current_mode_updated.setdefault(session_id, asyncio.Event()).set()
         elif isinstance(update, ToolCallStart):
             entry = {
                 "tool_call_id": update.tool_call_id,

tests/sdk/agent/test_acp_permission_policy.py

@@ -20,6 +20,7 @@
     SetSessionConfigOptionResponse,
 )
 
+from openhands.sdk.agent import acp_agent as acp_agent_module
 from openhands.sdk.agent.acp_agent import (
     ACPAgent,
     ACPSessionModeError,
@@ -57,6 +58,11 @@
 ]
 
 
+@pytest.fixture(autouse=True)
+def _short_mode_confirm_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(acp_agent_module, "_ACP_SESSION_MODE_CONFIRM_TIMEOUT", 0.05)
+
+
 def _claude_session_modes(current: str = "default") -> SessionModeState:
     return SessionModeState(
         available_modes=[
@@ -602,6 +608,105 @@ async def _set_mode(*, mode_id: str, session_id: str) -> None:
     assert client.get_current_mode_id("sess-1") == "default"
 
 
+def _mode_update(mode_id: str) -> CurrentModeUpdate:
+    return CurrentModeUpdate(
+        session_update="current_mode_update", current_mode_id=mode_id
+    )
+
+
+async def _apply_default_mode(conn: MagicMock, client: _OpenHandsACPBridge) -> None:
+    await _apply_acp_session_mode(
+        conn,
+        client,
+        policy="read_only",
+        mode_id="default",
+        agent_name="claude-agent-acp",
+        session_id="sess-1",
+        session_response=NewSessionResponse(
+            session_id="sess-1",
+            modes=_claude_session_modes("bypassPermissions"),
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_read_only_waits_for_mode_update_after_set_mode_response(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(acp_agent_module, "_ACP_SESSION_MODE_CONFIRM_TIMEOUT", 5.0)
+    conn = MagicMock()
+    conn.set_session_mode = AsyncMock()
+    client = _OpenHandsACPBridge(permission_policy="read_only")
+
+    async def _late_update() -> None:
+        await asyncio.sleep(0.05)
+        await client.session_update("sess-1", _mode_update("default"))
+
+    late = asyncio.create_task(_late_update())
+    await _apply_default_mode(conn, client)
+    await late
+    assert client.get_current_mode_id("sess-1") == "default"
+
+
+@pytest.mark.asyncio
+async def test_read_only_mode_confirmation_timeout_reports_wait() -> None:
+    conn = MagicMock()
+    conn.set_session_mode = AsyncMock()
+    client = _OpenHandsACPBridge(permission_policy="read_only")
+    with pytest.raises(ACPSessionModeError, match=r"did not confirm.*within 0\.05s"):
+        await _apply_default_mode(conn, client)
+
+
+@pytest.mark.asyncio
+async def test_read_only_wrong_mode_update_fails_without_waiting(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(acp_agent_module, "_ACP_SESSION_MODE_CONFIRM_TIMEOUT", 60.0)
+    conn = MagicMock()
+    conn.set_session_mode = AsyncMock()
+    client = _OpenHandsACPBridge(permission_policy="read_only")
+
+    async def _wrong_update() -> None:
+        await asyncio.sleep(0.01)
+        await client.session_update("sess-1", _mode_update("bypassPermissions"))
+
+    wrong = asyncio.create_task(_wrong_update())
+    with pytest.raises(ACPSessionModeError, match="observed='bypassPermissions'"):
+        await asyncio.wait_for(_apply_default_mode(conn, client), timeout=5)
+    await wrong
+
+
+@pytest.mark.asyncio
+async def test_read_only_wrong_mode_before_response_fails_without_waiting(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(acp_agent_module, "_ACP_SESSION_MODE_CONFIRM_TIMEOUT", 60.0)
+    conn = MagicMock()
+    client = _OpenHandsACPBridge(permission_policy="read_only")
+
+    async def _set_mode(*, mode_id: str, session_id: str) -> None:
+        await client.session_update(session_id, _mode_update("bypassPermissions"))
+
+    conn.set_session_mode = AsyncMock(side_effect=_set_mode)
+    with pytest.raises(ACPSessionModeError, match="observed='bypassPermissions'"):
+        await asyncio.wait_for(_apply_default_mode(conn, client), timeout=5)
+
+
+@pytest.mark.asyncio
+async def test_read_only_confirmed_before_response_does_not_wait(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(acp_agent_module, "_ACP_SESSION_MODE_CONFIRM_TIMEOUT", 60.0)
+    conn = MagicMock()
+    client = _OpenHandsACPBridge(permission_policy="read_only")
+
+    async def _set_mode(*, mode_id: str, session_id: str) -> None:
+        await client.session_update(session_id, _mode_update(mode_id))
+
+    conn.set_session_mode = AsyncMock(side_effect=_set_mode)
+    await asyncio.wait_for(_apply_default_mode(conn, client), timeout=5)
+
+
 @pytest.mark.asyncio
 async def test_writable_does_not_require_advertised_modes() -> None:
     conn = MagicMock()