feat(app-server): route ACP agents to the ACP conversation endpoint (#14004)

Co-authored-by: Debug Agent <debug@example.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

enterprise/tests/unit/storage/test_saas_sql_app_conversation_info_service.py

@@ -280,6 +280,7 @@ class TestSaasSQLAppConversationInfoService:
         stored_metadata.per_turn_token = 0
         stored_metadata.public = None
         stored_metadata.tags = {}
+        stored_metadata.agent_kind = None
 
         saas_metadata = MagicMock(spec=StoredConversationMetadataSaas)
         saas_metadata.user_id = UUID('a1111111-1111-1111-1111-111111111111')

openhands/app_server/app_conversation/app_conversation_models.py

@@ -97,6 +97,7 @@ class AppConversationInfo(BaseModel):
     trigger: ConversationTrigger | None = None
     pr_number: list[int] = Field(default_factory=list)
     llm_model: str | None = None
+    agent_kind: str = 'llm'
 
     metrics: MetricsSnapshot | None = None
 

openhands/app_server/app_conversation/live_status_app_conversation_service.py

@@ -16,8 +16,10 @@ from fastapi import Request
 from pydantic import Field, SecretStr, TypeAdapter
 
 from openhands.agent_server.models import (
+    ACPConversationInfo,
     ConversationInfo,
     SendMessageRequest,
+    StartACPConversationRequest,
     StartConversationRequest,
     TextContent,
 )
@@ -93,6 +95,7 @@ from openhands.app_server.utils.llm_metadata import (
 from openhands.integrations.provider import PROVIDER_TOKEN_TYPE, ProviderType
 from openhands.integrations.service_types import SuggestedTask
 from openhands.sdk import Agent, AgentContext, LocalWorkspace
+from openhands.sdk.agent.acp_agent import ACPAgent
 from openhands.sdk.hooks import HookConfig
 from openhands.sdk.llm import LLM
 from openhands.sdk.plugin import PluginSource
@@ -109,10 +112,38 @@ from openhands.tools.preset.planning import (
 )
 from openhands.utils._redact_compat import sanitize_config
 from openhands.utils.git import ensure_valid_git_branch_name
+from openhands.utils.sdk_settings_compat import ACPAgentSettings
 
 _conversation_info_type_adapter = TypeAdapter(list[ConversationInfo | None])
+_acp_conversation_info_type_adapter = TypeAdapter(list[ACPConversationInfo | None])
 _logger = logging.getLogger(__name__)
 
+
+def _agent_kind_to_router_path(agent_kind: str) -> str:
+    """Map agent_kind discriminator to the agent-server router path prefix."""
+    if agent_kind == 'acp':
+        return 'acp/conversations'
+    return 'conversations'
+
+
+def _split_ids_by_kind(
+    conversation_ids: list[str],
+    conversation_kind_by_id: dict[str, str],
+) -> tuple[list[str], list[str]]:
+    """Split conversation IDs into (llm_ids, acp_ids) based on their agent_kind."""
+    llm_ids = [
+        cid
+        for cid in conversation_ids
+        if conversation_kind_by_id.get(cid, 'llm') != 'acp'
+    ]
+    acp_ids = [
+        cid
+        for cid in conversation_ids
+        if conversation_kind_by_id.get(cid, 'llm') == 'acp'
+    ]
+    return llm_ids, acp_ids
+
+
 # Planning agent instruction to prevent "Ready to proceed?" behavior
 PLANNING_AGENT_INSTRUCTION = """<IMPORTANT_PLANNING_BOUNDARIES>
 You are a Planning Agent that can ONLY create plans - you CANNOT execute code or make changes.
@@ -334,15 +365,24 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
                 if sandbox.session_api_key
                 else {}
             )
+            is_acp = isinstance(start_conversation_request, StartACPConversationRequest)
+            router_path = 'acp/conversations' if is_acp else 'conversations'
             response = await self.httpx_client.post(
-                f'{agent_server_url}/api/conversations',
+                f'{agent_server_url}/api/{router_path}',
                 json=body_json,
                 headers=headers,
                 timeout=self.sandbox_startup_timeout,
             )
 
             response.raise_for_status()
-            info = ConversationInfo.model_validate(response.json())
+            if is_acp:
+                info = ACPConversationInfo.model_validate(response.json())
+                agent_kind = 'acp'
+                display_model = start_conversation_request.agent.acp_model
+            else:
+                info = ConversationInfo.model_validate(response.json())
+                agent_kind = 'llm'
+                display_model = start_conversation_request.agent.llm.model
 
             # Store info...
             user_id = await self.user_context.get_user_id()
@@ -351,7 +391,8 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
                 title=f'Conversation {info.id.hex[:5]}',
                 sandbox_id=sandbox.id,
                 created_by_user_id=user_id,
-                llm_model=start_conversation_request.agent.llm.model,
+                llm_model=display_model,
+                agent_kind=agent_kind,
                 # Git parameters
                 selected_repository=request.selected_repository,
                 selected_branch=request.selected_branch,
@@ -411,6 +452,13 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
             app_conversation_infos
         )
 
+        # Build a mapping from conversation id -> agent_kind for routing
+        conversation_kind_by_id: dict[str, str] = {
+            str(info.id): info.agent_kind
+            for info in app_conversation_infos
+            if info is not None
+        }
+
         # Get referenced sandboxes in a single batch operation...
         sandboxes = await self.sandbox_service.batch_get_sandboxes(
             list(sandbox_id_to_conversation_ids)
@@ -420,7 +468,9 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         # Gather the running conversations
         tasks = [
             self._get_live_conversation_info(
-                sandbox, sandbox_id_to_conversation_ids.get(sandbox.id)
+                sandbox,
+                sandbox_id_to_conversation_ids.get(sandbox.id),
+                conversation_kind_by_id,
             )
             for sandbox in sandboxes
             if sandbox and sandbox.status == SandboxStatus.RUNNING
@@ -456,42 +506,73 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         self,
         sandbox: SandboxInfo,
         conversation_ids: list[str],
-    ) -> list[ConversationInfo]:
-        """Get agent status for multiple conversations from the Agent Server."""
-        try:
-            # Build the URL with query parameters
-            agent_server_url = self._get_agent_server_url(sandbox)
-            url = f'{agent_server_url.rstrip("/")}/api/conversations'
-            params = {'ids': conversation_ids}
+        conversation_kind_by_id: dict[str, str] | None = None,
+    ) -> list[ConversationInfo | ACPConversationInfo]:
+        """Get agent status for multiple conversations from the Agent Server.
 
-            # Set up headers
-            headers = {}
-            if sandbox.session_api_key:
-                headers['X-Session-API-Key'] = sandbox.session_api_key
-
-            response = await self.httpx_client.get(url, params=params, headers=headers)
-            response.raise_for_status()
-
-            data = response.json()
-            conversation_info = _conversation_info_type_adapter.validate_python(data)
-            conversation_info = [c for c in conversation_info if c]
-            return conversation_info
-        except httpx.HTTPStatusError:
-            # The runtime API stops idle sandboxes all the time and they return a 404 or a 503.
-            # This is normal and should not be considered an error.
-            _logger.warning(
-                f'Error getting conversation status from sandbox {sandbox.id}',
-                exc_info=True,
-            )
-            return []
-        except Exception:
-            # Not getting a status is not a fatal error - we just mark the conversation as stopped
-            _logger.exception(
-                f'Error getting conversation status from sandbox {sandbox.id}',
-                stack_info=True,
-            )
+        Splits conversation_ids by agent_kind and fetches from the appropriate
+        endpoint (/api/conversations for LLM, /api/acp/conversations for ACP).
+        """
+        if not conversation_ids:
             return []
 
+        conversation_kind_by_id = conversation_kind_by_id or {}
+
+        llm_ids, acp_ids = _split_ids_by_kind(conversation_ids, conversation_kind_by_id)
+
+        agent_server_url = self._get_agent_server_url(sandbox)
+        headers: dict[str, str] = {}
+        if sandbox.session_api_key:
+            headers['X-Session-API-Key'] = sandbox.session_api_key
+
+        results: list[ConversationInfo | ACPConversationInfo] = []
+
+        # Fetch LLM conversations
+        if llm_ids:
+            try:
+                url = f'{agent_server_url.rstrip("/")}/api/conversations'
+                response = await self.httpx_client.get(
+                    url, params={'ids': llm_ids}, headers=headers
+                )
+                response.raise_for_status()
+                data = response.json()
+                infos = _conversation_info_type_adapter.validate_python(data)
+                results.extend(c for c in infos if c)
+            except httpx.HTTPStatusError:
+                _logger.warning(
+                    f'Error getting LLM conversation status from sandbox {sandbox.id}',
+                    exc_info=True,
+                )
+            except Exception:
+                _logger.exception(
+                    f'Error getting LLM conversation status from sandbox {sandbox.id}',
+                    stack_info=True,
+                )
+
+        # Fetch ACP conversations
+        if acp_ids:
+            try:
+                url = f'{agent_server_url.rstrip("/")}/api/acp/conversations'
+                response = await self.httpx_client.get(
+                    url, params={'ids': acp_ids}, headers=headers
+                )
+                response.raise_for_status()
+                data = response.json()
+                infos = _acp_conversation_info_type_adapter.validate_python(data)
+                results.extend(c for c in infos if c)
+            except httpx.HTTPStatusError:
+                _logger.warning(
+                    f'Error getting ACP conversation status from sandbox {sandbox.id}',
+                    exc_info=True,
+                )
+            except Exception:
+                _logger.exception(
+                    f'Error getting ACP conversation status from sandbox {sandbox.id}',
+                    stack_info=True,
+                )
+
+        return results
+
     def _build_conversation(
         self,
         app_conversation_info: AppConversationInfo | None,
@@ -516,7 +597,10 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
                 None,
             )
             if conversation_url:
-                conversation_url += f'/api/conversations/{app_conversation_info.id.hex}'
+                router_path = _agent_kind_to_router_path(
+                    app_conversation_info.agent_kind
+                )
+                conversation_url += f'/api/{router_path}/{app_conversation_info.id.hex}'
             session_api_key = sandbox.session_api_key
 
         return AppConversation(
@@ -1219,7 +1303,7 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         selected_repository: str | None = None,
         plugins: list[PluginSpec] | None = None,
         api_secrets: dict[str, SecretStr] | None = None,
-    ) -> StartConversationRequest:
+    ) -> StartConversationRequest | StartACPConversationRequest:
         """Build a complete StartConversationRequest for a user.
 
         Resolves LLM, MCP, tools, secrets and agent context, then
@@ -1228,6 +1312,8 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         skills, hooks) are applied to the agent after creation.
         Finally delegates to ``ConversationSettings.create_request()``.
 
+        For ACP agent settings, routes to ``_build_acp_start_conversation_request``.
+
         Args:
             sandbox: Sandbox information
             conversation_id: Unique conversation identifier
@@ -1247,6 +1333,18 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         """
         user = await self.user_context.get_user_info()
 
+        # Route ACP agent settings to the ACP-specific builder
+        if isinstance(user.agent_settings, ACPAgentSettings):
+            return await self._build_acp_start_conversation_request(
+                sandbox=sandbox,
+                conversation_id=conversation_id,
+                initial_message=initial_message,
+                working_dir=working_dir,
+                selected_repository=selected_repository,
+                plugins=plugins,
+                api_secrets=api_secrets,
+            )
+
         project_dir = get_project_dir(working_dir, selected_repository)
         workspace = LocalWorkspace(working_dir=project_dir)
 
@@ -1388,6 +1486,128 @@ class LiveStatusAppConversationService(AppConversationServiceBase):
         # prompts, LLM metadata, skills) applied after create_agent().
         return conv_settings.create_request(StartConversationRequest, agent=agent)
 
+    @staticmethod
+    def _acp_provider_env(user: UserInfo) -> dict[str, str]:
+        """Translate user credentials into ACP provider environment variables.
+
+        The ACP subprocess reads provider credentials from environment variables.
+        Maps the user's LLM API key to the env var expected by the active ACP
+        server via ``ACPAgentSettings.api_key_env_var``. Custom servers return
+        ``None`` — users manage credentials entirely via ``acp_env``.
+
+        Args:
+            user: User information containing credentials
+
+        Returns:
+            Dictionary of environment variable name -> value pairs for the ACP subprocess.
+        """
+        env: dict[str, str] = {}
+
+        if not isinstance(user.agent_settings, ACPAgentSettings):
+            return env
+
+        acp_settings = user.agent_settings
+
+        # Pass through any explicit per-key overrides first; the API-key
+        # injection below will not overwrite keys already set here.
+        if acp_settings.acp_env:
+            env.update(acp_settings.acp_env)
+
+        # Map the user's LLM API key to the env var expected by the ACP server.
+        api_key_env = acp_settings.api_key_env_var
+        if api_key_env and api_key_env not in env:
+            llm_api_key = acp_settings.llm.api_key
+            if llm_api_key:
+                key_value = (
+                    llm_api_key.get_secret_value()
+                    if isinstance(llm_api_key, SecretStr)
+                    else str(llm_api_key)
+                )
+                if key_value and key_value.strip():
+                    env[api_key_env] = key_value
+
+        return env
+
+    async def _build_acp_start_conversation_request(
+        self,
+        sandbox: SandboxInfo,
+        conversation_id: UUID,
+        initial_message: SendMessageRequest | None,
+        working_dir: str,
+        selected_repository: str | None = None,
+        plugins: list[PluginSpec] | None = None,
+        api_secrets: dict[str, SecretStr] | None = None,
+    ) -> StartACPConversationRequest:
+        """Build a StartACPConversationRequest for ACP agent conversations.
+
+        Unlike the LLM path, ACP agents run as separate subprocesses; we pass
+        credentials via environment variables rather than injecting an LLM object.
+
+        Args:
+            sandbox: Sandbox information
+            conversation_id: Unique conversation identifier
+            initial_message: Optional initial message to send
+            working_dir: Working directory path
+            selected_repository: Optional repository name
+            plugins: Optional list of plugins to load
+            api_secrets: Optional secrets passed directly via the API.
+        """
+        user = await self.user_context.get_user_info()
+
+        project_dir = get_project_dir(working_dir, selected_repository)
+        workspace = LocalWorkspace(working_dir=project_dir)
+
+        # --- secrets --------------------------------------------------------
+        secrets = await self._setup_secrets_for_git_providers(user)
+
+        if api_secrets:
+            from openhands.app_server.constants import (
+                validate_secret_name,
+                validate_secrets_dict,
+            )
+
+            validate_secrets_dict(cast('Mapping[str, object]', api_secrets))
+            for name, value in api_secrets.items():
+                validate_secret_name(name)
+                if name in secrets:
+                    _logger.warning(
+                        'API-provided secret %r overrides existing secret', name
+                    )
+                secrets[name] = StaticSecret(value=value)
+
+        # --- build the ACP agent ------------------------------------------
+        acp_settings = user.agent_settings  # already verified to be ACPAgentSettings
+        assert isinstance(acp_settings, ACPAgentSettings)
+
+        # Merge provider env vars (API keys etc.) into acp_env
+        provider_env = self._acp_provider_env(user)
+        merged_env = {**(acp_settings.acp_env or {}), **provider_env}
+
+        acp_agent = ACPAgent(
+            acp_command=acp_settings.acp_command,
+            acp_args=acp_settings.acp_args,
+            acp_env=merged_env,
+            acp_model=acp_settings.acp_model,
+            acp_session_mode=acp_settings.acp_session_mode,
+            acp_prompt_timeout=acp_settings.acp_prompt_timeout,
+        )
+
+        sdk_plugins: list[PluginSource] | None = None
+        if plugins:
+            sdk_plugins = [
+                PluginSource(source=p.source, ref=p.ref, repo_path=p.repo_path)
+                for p in plugins
+            ]
+
+        return StartACPConversationRequest(
+            workspace=workspace,
+            conversation_id=conversation_id,
+            initial_message=initial_message,
+            secrets=secrets,
+            plugins=sdk_plugins,
+            agent=acp_agent,
+        )
+
     async def _process_pending_messages(
         self,
         task_id: UUID,

openhands/app_server/app_conversation/sql_app_conversation_info_service.py

@@ -100,6 +100,7 @@ class StoredConversationMetadata(Base):
 
     # LLM model used for the conversation
     llm_model: Mapped[str | None] = mapped_column(String, nullable=True)
+    agent_kind: Mapped[str | None] = mapped_column(String, nullable=True)
 
     conversation_version: Mapped[str] = mapped_column(
         String, nullable=False, default='V0', index=True
@@ -370,6 +371,7 @@ class SQLAppConversationInfoService(AppConversationInfoService):
             context_window=usage.context_window,
             per_turn_token=usage.per_turn_token,
             llm_model=info.llm_model,
+            agent_kind=info.agent_kind,
             conversation_version='V1',
             sandbox_id=info.sandbox_id,
             parent_conversation_id=(
@@ -558,6 +560,7 @@ class SQLAppConversationInfoService(AppConversationInfoService):
             trigger=ConversationTrigger(stored.trigger) if stored.trigger else None,
             pr_number=stored.pr_number or [],
             llm_model=stored.llm_model,
+            agent_kind=stored.agent_kind or 'llm',
             metrics=metrics,
             parent_conversation_id=(
                 UUID(stored.parent_conversation_id)

openhands/app_server/app_lifespan/alembic/versions/009.py

@@ -0,0 +1,26 @@
+"""Add agent_kind column to conversation_metadata table
+
+Revision ID: 009
+Revises: 008
+Create Date: 2026-04-27 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '009'
+down_revision: Union[str, None] = '008'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'conversation_metadata', sa.Column('agent_kind', sa.String, nullable=True)
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('conversation_metadata', 'agent_kind')

openhands/app_server/sandbox/process_sandbox_service.py

@@ -134,22 +134,22 @@ class ProcessSandboxService(SandboxService):
         )
 
         try:
-            # Start the process
-            process = subprocess.Popen(
-                cmd,
-                env=env,
-                cwd=working_dir,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-            )
+            # Start the process, directing output to a log file to avoid pipe-buffer deadlocks
+            log_path = os.path.join(working_dir, '.openhands-agent-server.log')
+            with open(log_path, 'a', buffering=1) as log_handle:
+                process = subprocess.Popen(
+                    cmd, env=env, cwd=working_dir, stdout=log_handle, stderr=log_handle
+                )
 
             # Wait a moment for the process to start
             await asyncio.sleep(1)
 
             # Check if process is still running
             if process.poll() is not None:
-                stdout, stderr = process.communicate()
-                raise SandboxError(f'Agent process failed to start: {stderr.decode()}')
+                raise SandboxError(
+                    f'Agent process failed to start (exit code {process.returncode}). '
+                    f'See {log_path} for details.'
+                )
 
             return process
 

openhands/app_server/settings/settings_models.py

@@ -29,8 +29,15 @@ from openhands.core.config.mcp_config import MCPConfig
 from openhands.core.config.utils import load_openhands_config
 from openhands.integrations.provider import ProviderToken
 from openhands.integrations.service_types import ProviderType
-from openhands.sdk.settings import AgentSettings, ConversationSettings
+from openhands.sdk.settings import ConversationSettings
 from openhands.utils.jsonpatch_compat import deep_merge
+from openhands.utils.sdk_settings_compat import (
+    ACPAgentSettings,
+    AgentSettingsConfig,
+    LLMAgentSettings,
+    default_agent_settings,
+    validate_agent_settings,
+)
 
 
 def _coerce_value(value: Any) -> Any:
@@ -130,7 +137,7 @@ class Settings(BaseModel):
     git_user_name: str | None = None
     git_user_email: str | None = None
     v1_enabled: bool = True
-    agent_settings: AgentSettings = Field(default_factory=AgentSettings)
+    agent_settings: AgentSettingsConfig = Field(default_factory=default_agent_settings)
     conversation_settings: ConversationSettings = Field(
         default_factory=ConversationSettings
     )
@@ -198,9 +205,7 @@ class Settings(BaseModel):
 
             # Use object.__setattr__ to avoid validate_assignment
             # side-effects on other fields.
-            object.__setattr__(
-                self, 'agent_settings', AgentSettings.model_validate(merged)
-            )
+            object.__setattr__(self, 'agent_settings', validate_agent_settings(merged))
 
         conv_update = payload.get('conversation_settings_diff')
         if isinstance(conv_update, dict):
@@ -245,7 +250,9 @@ class Settings(BaseModel):
 
     @field_serializer('agent_settings')
     def agent_settings_serializer(
-        self, agent_settings: AgentSettings, info: SerializationInfo
+        self,
+        agent_settings: LLMAgentSettings | ACPAgentSettings,
+        info: SerializationInfo,
     ) -> dict[str, Any]:
         context = info.context or {}
         if context.get('expose_secrets', False):
@@ -268,7 +275,7 @@ class Settings(BaseModel):
         agent_settings = data.get('agent_settings')
         if isinstance(agent_settings, dict):
             data['agent_settings'] = _coerce_dict_secrets(agent_settings)
-        elif isinstance(agent_settings, AgentSettings):
+        elif isinstance(agent_settings, (LLMAgentSettings, ACPAgentSettings)):
             data['agent_settings'] = agent_settings.model_dump(
                 mode='json', context={'expose_secrets': True}
             )
@@ -337,7 +344,8 @@ class Settings(BaseModel):
             remote_runtime_resource_factor=app_config.sandbox.remote_runtime_resource_factor,
             search_api_key=app_config.search_api_key,
             max_budget_per_task=app_config.max_budget_per_task,
-            agent_settings=AgentSettings(**agent_settings_dict),
+            # Always LLM for config-file-sourced settings
+            agent_settings=LLMAgentSettings(**agent_settings_dict),
             conversation_settings=ConversationSettings.model_validate(
                 {
                     'confirmation_mode': bool(app_config.security.confirmation_mode),
@@ -349,6 +357,8 @@ class Settings(BaseModel):
 
     def merge_with_config_settings(self) -> 'Settings':
         """Merge config.toml MCP settings with stored SDK agent_settings."""
+        if not isinstance(self.agent_settings, LLMAgentSettings):
+            return self
         config_settings = Settings.from_config()
         if not config_settings:
             return self
@@ -363,7 +373,7 @@ class Settings(BaseModel):
         self.agent_settings.mcp_config = merged_mcp
         return self
 
-    def to_agent_settings(self) -> AgentSettings:
+    def to_agent_settings(self) -> LLMAgentSettings | ACPAgentSettings:
         return self.agent_settings
 
     def get_agent_settings_display(self) -> dict[str, Any]:

openhands/app_server/settings/settings_router.py

@@ -22,7 +22,7 @@ from openhands.integrations.provider import (
     PROVIDER_TOKEN_TYPE,
     ProviderType,
 )
-from openhands.sdk.settings import AgentSettings, ConversationSettings
+from openhands.sdk.settings import ConversationSettings
 from openhands.server.shared import config
 from openhands.server.user_auth import (
     get_provider_tokens,
@@ -35,6 +35,10 @@ from openhands.utils.llm import (
     is_openhands_model,
     resolve_llm_base_url,
 )
+from openhands.utils.sdk_settings_compat import (
+    LLMAgentSettings,
+    export_agent_settings_schema,
+)
 
 LITE_LLM_API_URL = os.environ.get(
     'LITE_LLM_API_URL', 'https://llm-proxy.app.all-hands.dev'
@@ -55,6 +59,8 @@ def _post_merge_llm_fixups(settings: Settings) -> None:
     rules to :func:`openhands.utils.llm.resolve_llm_base_url` so the
     personal-save and enterprise org-defaults paths stay in lockstep.
     """
+    if not isinstance(settings.agent_settings, LLMAgentSettings):
+        return
     llm = settings.agent_settings.llm
     llm.base_url = resolve_llm_base_url(
         model=llm.model,
@@ -256,7 +262,7 @@ async def store_settings(
 @router.get('/agent-schema')
 async def load_settings_schema() -> dict[str, Any]:
     """Load the schema for settings"""
-    return AgentSettings.export_schema().model_dump(mode='json')
+    return export_agent_settings_schema().model_dump(mode='json')
 
 
 @router.get('/conversation-schema')

openhands/utils/sdk_settings_compat.py

@@ -0,0 +1,53 @@
+"""SDK compatibility shim for the discriminated-union ``AgentSettings`` rework."""
+
+from typing import Any
+
+try:
+    from openhands.sdk.settings import (  # type: ignore[attr-defined]
+        ACPAgentSettings,
+        AgentSettingsConfig,
+        LLMAgentSettings,
+        default_agent_settings,
+        export_agent_settings_schema,
+        validate_agent_settings,
+    )
+
+    _HAS_DISCRIMINATED_UNION = True
+except ImportError:
+    _HAS_DISCRIMINATED_UNION = False
+    from openhands.sdk.settings import AgentSettings
+
+    LLMAgentSettings = AgentSettings  # type: ignore[misc, assignment]
+
+    class _ACPAgentSettingsStub:
+        """Sentinel — older SDK builds cannot produce ACPAgentSettings instances."""
+
+    ACPAgentSettings = _ACPAgentSettingsStub  # type: ignore[misc, assignment]
+    AgentSettingsConfig = AgentSettings  # type: ignore[misc, assignment]
+
+    def default_agent_settings() -> AgentSettings:  # type: ignore[misc]
+        return AgentSettings()
+
+    def validate_agent_settings(data: dict[str, Any]) -> AgentSettings:  # type: ignore[misc]
+        if isinstance(data, dict) and data.get('kind') == 'acp':
+            raise RuntimeError(
+                "Stored settings contain kind='acp' but the installed "
+                'openhands-sdk does not support ACP agents. Upgrade to an '
+                'SDK release that includes the discriminated-union rework '
+                '(OpenHands/software-agent-sdk#2861).'
+            )
+        return AgentSettings.model_validate(data)
+
+    def export_agent_settings_schema():  # type: ignore[misc]
+        return AgentSettings.export_schema()
+
+
+__all__ = [
+    'ACPAgentSettings',
+    'AgentSettingsConfig',
+    'LLMAgentSettings',
+    '_HAS_DISCRIMINATED_UNION',
+    'default_agent_settings',
+    'export_agent_settings_schema',
+    'validate_agent_settings',
+]

tests/unit/utils/test_sdk_settings_compat.py

@@ -0,0 +1,72 @@
+"""Tests for SDK settings compatibility shim."""
+
+import pytest
+
+from openhands.utils.sdk_settings_compat import (
+    _HAS_DISCRIMINATED_UNION,
+    ACPAgentSettings,
+    AgentSettingsConfig,
+    LLMAgentSettings,
+    default_agent_settings,
+    export_agent_settings_schema,
+    validate_agent_settings,
+)
+
+
+def test_has_discriminated_union_is_bool():
+    assert isinstance(_HAS_DISCRIMINATED_UNION, bool)
+
+
+def test_default_agent_settings_returns_instance():
+    settings = default_agent_settings()
+    assert settings is not None
+
+
+def test_validate_agent_settings_with_llm_dict():
+    """validate_agent_settings should accept a plain dict and return an agent settings object."""
+    result = validate_agent_settings({})
+    assert result is not None
+
+
+def test_validate_agent_settings_with_acp_kind_raises_or_returns():
+    """validate_agent_settings with kind='acp' either returns ACP settings or raises RuntimeError."""
+    data = {'kind': 'acp', 'agent_kind': 'acp'}
+    if _HAS_DISCRIMINATED_UNION:
+        # With SDK support, kind='acp' produces an ACPAgentSettings
+        result = validate_agent_settings(data)
+        assert isinstance(result, ACPAgentSettings)
+    else:
+        # Without SDK support, kind='acp' raises RuntimeError
+        with pytest.raises(RuntimeError, match='kind=.acp.'):
+            validate_agent_settings(data)
+
+
+def test_export_agent_settings_schema_returns_schema():
+    schema = export_agent_settings_schema()
+    assert schema is not None
+    # Schema should have a model_dump method (Pydantic model)
+    dumped = schema.model_dump(mode='json')
+    assert isinstance(dumped, dict)
+
+
+def test_llm_agent_settings_is_usable():
+    """LLMAgentSettings should be constructable."""
+    settings = LLMAgentSettings()
+    assert settings is not None
+
+
+@pytest.mark.skipif(
+    not _HAS_DISCRIMINATED_UNION,
+    reason='Discriminated union not available in installed SDK',
+)
+def test_acp_agent_settings_is_usable():
+    """ACPAgentSettings should be constructable when SDK supports it."""
+    settings = ACPAgentSettings()
+    assert settings is not None
+    assert hasattr(settings, 'agent_kind')
+    assert settings.agent_kind == 'acp'
+
+
+def test_agent_settings_config_is_type():
+    """AgentSettingsConfig should be a usable type."""
+    assert AgentSettingsConfig is not None