LangBot/plugins/Runner/acp-agent-runner/components/runner/default.py

"""ACP default Runner implementation."""

from __future__ import annotations

import contextlib
import json
import logging
import os
import shlex
import time
import typing

from langbot_plugin.api.agent_tools.external_tools import AgentRunExternalTools
from langbot_plugin.api.agent_tools.mcp_access import AgentRunMCPAccess
from langbot_plugin.api.agent_tools.mcp_config import AgentMCPServerConfig
from langbot_plugin.api.definition.components.runner.runner import Runner
from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
from langbot_plugin.api.entities.builtin.runner import (
    AgentInput,
    RunnerContext,
    RunnerResult,
)
from pkg.acp_client import AcpError, AcpStdioClient
from pkg.daemon_relay import (
    AgentRuntimeDaemonError,
    agent_runtime_daemon_config_from_plugin_config,
    get_agent_runtime_daemon_hub,
)
from pkg.prompt import acp_prompt_blocks, has_acp_prompt_input, prompt_capabilities
from pkg.runtime_support import local_workspace
from pkg.session import CLAUDE_ENV_VARS_TO_UNSET, build_session_params
from pkg.steering import run_with_steering

logger = logging.getLogger(__name__)

ACP_SESSION_STATE_KEY = "external.acp_session_id"
# Common ACP agent launch presets from the public ACP registry and vendor docs.
# Pin adapters when runner behavior depends on a specific ACP lifecycle fix;
# set acp-command to use a different tested version or install path.
DEFAULT_PROVIDER_COMMANDS = {
    "auggie": "npx -y @augmentcode/auggie --acp",
    "autohand": "npx -y @autohandai/autohand-acp",
    "claude-code": "npx -y @agentclientprotocol/[email protected]",
    "codebuddy-code": "npx -y @tencent-ai/codebuddy-code --acp",
    "codex": "npx -y @zed-industries/codex-acp",
    "deepagents": "npx -y deepagents-acp",
    "dimcode": "npx -y dimcode acp",
    "dirac": "npx -y dirac-cli --acp",
    "factory-droid": "npx -y droid exec --output-format acp-daemon",
    "gemini": "npx -y @google/gemini-cli --acp",
    "glm-agent": "npx -y glm-acp-agent",
    "kilo": "npx -y @kilocode/cli acp",
    "opencode": "opencode acp",
    "pi-acp": "npx -y pi-acp",
    "qwen-code": "npx -y @qwen-code/qwen-code --acp --experimental-skills",
}
SUPPORTED_PROVIDERS = set(DEFAULT_PROVIDER_COMMANDS) | {"custom"}
SUPPORTED_LOCATIONS = {"local", "remote-ssh", "daemon"}
SUPPORTED_REMOTE_SHELLS = {"bash", "powershell", "none"}
SUPPORTED_LANGBOT_ASSET_MODES = {"auto", "ephemeral", "gateway"}
ACP_COMMAND_CONFIG_KEYS = ("acp-command", "remote-command", "local-command")
WORKSPACE_CONFIG_KEYS = ("workspace",)
REMOTE_WORKSPACE_CONFIG_KEYS = ("remote-workspace", "session-cwd")
LOCAL_WORKSPACE_CONFIG_KEYS = ("local-workspace", "cwd", "session-cwd")
SSH_TARGET_CONFIG_KEYS = ("ssh-target", "ssh_target")
DAEMON_ID_CONFIG_KEYS = ("daemon-id", "daemon_id")
SSH_IDENTITY_FILE_CONFIG_KEYS = ("ssh-identity-file", "ssh-key-file")
ASSET_GATEWAY_PUBLIC_URL_CONFIG_KEYS = ("langbot-assets-gateway-public-url", "mcp-public-url")


def _to_bool(value: typing.Any, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, int | float):
        return bool(value)
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "y", "on"}
    return bool(value)


def _to_float(value: typing.Any, default: float) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _to_int(value: typing.Any, default: int) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _parse_args(value: typing.Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(item) for item in value if str(item)]
    text = str(value).strip()
    if not text:
        return []
    return shlex.split(text)


def _parse_json_object(value: typing.Any, *, label: str) -> dict[str, typing.Any]:
    if value in (None, ""):
        return {}
    if isinstance(value, dict):
        return dict(value)
    try:
        parsed = json.loads(str(value))
    except json.JSONDecodeError as exc:
        raise AcpError(f"{label} must be a JSON object: {exc}", code="acp.config_invalid") from exc
    if not isinstance(parsed, dict):
        raise AcpError(f"{label} must be a JSON object", code="acp.config_invalid")
    return parsed


def _parse_json_list(value: typing.Any, *, label: str) -> list[typing.Any]:
    if value in (None, ""):
        return []
    if isinstance(value, list):
        return list(value)
    try:
        parsed = json.loads(str(value))
    except json.JSONDecodeError as exc:
        raise AcpError(f"{label} must be a JSON array: {exc}", code="acp.config_invalid") from exc
    if not isinstance(parsed, list):
        raise AcpError(f"{label} must be a JSON array", code="acp.config_invalid")
    return parsed


def _first_config_value(config: dict[str, typing.Any], keys: tuple[str, ...]) -> str:
    for key in keys:
        value = config.get(key)
        if value not in (None, ""):
            return str(value).strip()
    return ""


def _shell_command(command: str, command_args: typing.Sequence[str]) -> str:
    parts = (command, *command_args)
    return " ".join(shlex.quote(part) for part in parts if part)


def _posix_quote(value: str) -> str:
    return shlex.quote(str(value))


def _powershell_quote(value: str) -> str:
    return "'" + str(value).replace("'", "''") + "'"


def _remote_shell_command(
    *,
    remote_shell: str,
    workspace: str,
    acp_command: str,
    env: dict[str, str] | None = None,
    unset_env: typing.Sequence[str] = (),
) -> str:
    remote_env = dict(env or {})
    if remote_shell == "none":
        unset_prefix = " ".join(f"-u {_posix_quote(name)}" for name in unset_env)
        env_prefix = " ".join(f"{_posix_quote(name)}={_posix_quote(value)}" for name, value in remote_env.items())
        prefixes = " ".join(part for part in (unset_prefix, env_prefix) if part)
        return f"env {prefixes} {acp_command}" if prefixes else acp_command

    if remote_shell == "powershell":
        script_parts = ["$ErrorActionPreference='Stop'"]
        script_parts.extend(f"Remove-Item Env:{name} -ErrorAction SilentlyContinue" for name in unset_env)
        script_parts.extend(
            f"Set-Item -Path {_powershell_quote(f'Env:{name}')} -Value {_powershell_quote(value)}"
            for name, value in remote_env.items()
            if name not in unset_env
        )
        if workspace:
            quoted_workspace = _powershell_quote(workspace)
            script_parts.append(f"New-Item -ItemType Directory -Force -Path {quoted_workspace} | Out-Null")
            script_parts.append(f"Set-Location -LiteralPath {quoted_workspace}")
        script_parts.append(acp_command)
        return f"powershell -NoProfile -ExecutionPolicy Bypass -Command {_posix_quote('; '.join(script_parts))}"

    script_parts = []
    script_parts.extend(
        f"export {_posix_quote(name)}={_posix_quote(value)}"
        for name, value in remote_env.items()
        if name not in unset_env
    )
    if workspace:
        quoted_workspace = _posix_quote(workspace)
        script_parts.append(f"mkdir -p {quoted_workspace}")
        script_parts.append(f"cd {quoted_workspace}")
    unset_prefix = " ".join(f"-u {_posix_quote(name)}" for name in unset_env)
    command = f"env {unset_prefix} {acp_command}" if unset_prefix else acp_command
    script_parts.append(f"exec {command}")
    return f"bash -lc {_posix_quote(' && '.join(script_parts))}"


def _mcp_bridge_tool_names(ctx: RunnerContext) -> list[str]:
    tool_names = ["langbot_get_current_event", "langbot_list_assets"]
    if ctx.context.available_apis.history_page:
        tool_names.append("langbot_history_page")
    if ctx.resources.knowledge_bases:
        tool_names.append("langbot_retrieve_knowledge")
    if ctx.resources.tools:
        tool_names.append("langbot_get_tool_detail")
        tool_names.append("langbot_call_tool")
    return tool_names


def _resource_summary(ctx: RunnerContext) -> dict[str, typing.Any]:
    ordinary_tools = [item for item in ctx.resources.tools if item.tool_type != "platform"]
    platform_tools = [item for item in ctx.resources.tools if item.tool_type == "platform"]
    return {
        "knowledge_bases": [
            {
                "kb_id": item.kb_id,
                "name": item.kb_name,
                "type": item.kb_type,
            }
            for item in ctx.resources.knowledge_bases
        ],
        "tools": [
            {
                "tool_name": item.tool_name,
                "type": item.tool_type,
                "description": item.description,
            }
            for item in ordinary_tools
        ],
        "platform_tools": [
            {
                "tool_name": item.tool_name,
                "description": item.description,
            }
            for item in platform_tools
        ],
        "mcp_bridge_tools": [{"tool_name": name} for name in _mcp_bridge_tool_names(ctx)],
    }


def _extract_session_id(result: typing.Any) -> str:
    if not isinstance(result, dict):
        return ""
    session_id = result.get("sessionId") or result.get("session_id") or result.get("id")
    return str(session_id or "").strip()


def _runtime_has_method(capabilities: dict[str, typing.Any], method: str) -> bool:
    if method == "session/load":
        return bool(capabilities.get("loadSession"))
    if method.startswith("session/"):
        session_capabilities = capabilities.get("sessionCapabilities")
        if isinstance(session_capabilities, dict):
            capability_key = method.split("/", 1)[1].replace("-", "_")
            for key in (capability_key, method.split("/", 1)[1]):
                if key in session_capabilities:
                    value = session_capabilities[key]
                    return value is not None and value is not False
    methods = capabilities.get("methods")
    if isinstance(methods, list) and method in {str(item) for item in methods}:
        return True
    key = method.replace("/", "_").replace("-", "_")
    value = capabilities.get(method) or capabilities.get(key)
    return bool(value)


def _content_text(value: typing.Any) -> str:
    if isinstance(value, str):
        return value
    if not isinstance(value, dict):
        return ""
    if value.get("type") == "text" and isinstance(value.get("text"), str):
        return str(value["text"])
    text = value.get("text")
    if isinstance(text, str):
        return text
    return ""


def _agent_text_from_update(update: dict[str, typing.Any]) -> str:
    payload = update.get("update") if isinstance(update.get("update"), dict) else update
    if not isinstance(payload, dict):
        return ""

    update_kind = str(payload.get("sessionUpdate") or payload.get("kind") or payload.get("type") or "")
    content = payload.get("content")

    if update_kind == "agent_message_chunk":
        return _content_text(content)

    if "agent_message" in update_kind or payload.get("role") == "assistant":
        if isinstance(content, list):
            return "".join(_content_text(item) for item in content)
        return _content_text(content)

    return ""


def _agent_message_id(update: dict[str, typing.Any]) -> str:
    payload = update.get("update") if isinstance(update.get("update"), dict) else update
    if not isinstance(payload, dict):
        return ""
    return str(payload.get("messageId") or payload.get("message_id") or "")


def _tool_update_payload(update: dict[str, typing.Any]) -> dict[str, typing.Any] | None:
    payload = update.get("update") if isinstance(update.get("update"), dict) else update
    if not isinstance(payload, dict):
        return None
    update_kind = str(payload.get("sessionUpdate") or payload.get("kind") or payload.get("type") or "")
    if "tool_call" not in update_kind:
        return None
    return payload


def _tool_update_name(payload: dict[str, typing.Any]) -> str:
    metadata = payload.get("_meta")
    if isinstance(metadata, dict):
        for value in metadata.values():
            if isinstance(value, dict) and value.get("toolName"):
                return str(value["toolName"])
    return str(payload.get("name") or payload.get("title") or "acp_tool")


def _mcp_env_to_acp(env: typing.Any) -> list[dict[str, str]]:
    if isinstance(env, dict):
        return [{"name": str(key), "value": str(value)} for key, value in env.items()]
    if isinstance(env, list):
        converted = []
        for item in env:
            if isinstance(item, dict) and "name" in item and "value" in item:
                converted.append({"name": str(item["name"]), "value": str(item["value"])})
        return converted
    return []


def _mcp_headers_to_acp(headers: typing.Any) -> list[dict[str, str]]:
    if isinstance(headers, dict):
        return [{"name": str(key), "value": str(value)} for key, value in headers.items()]
    if isinstance(headers, list):
        converted = []
        for item in headers:
            if isinstance(item, dict) and "name" in item and "value" in item:
                converted.append({"name": str(item["name"]), "value": str(item["value"])})
        return converted
    return []


class DefaultRunner(Runner):
    """Runner for Agent Client Protocol compatible agents."""

    def _validate_config(self, ctx: RunnerContext) -> dict[str, typing.Any]:
        config = ctx.config or {}
        provider = str(config.get("provider", "claude-code") or "claude-code").strip()
        if provider not in SUPPORTED_PROVIDERS:
            raise AcpError(
                f"provider must be one of: {', '.join(sorted(SUPPORTED_PROVIDERS))}",
                code="acp.config_invalid",
            )

        location = str(config.get("location", "local") or "local").strip()
        if location not in SUPPORTED_LOCATIONS:
            raise AcpError(
                f"location must be one of: {', '.join(sorted(SUPPORTED_LOCATIONS))}",
                code="acp.config_invalid",
            )

        command = str(config.get("command", "") or "").strip()
        command_args = _parse_args(config.get("args"))
        acp_command = _first_config_value(config, ACP_COMMAND_CONFIG_KEYS)
        if not acp_command and command:
            acp_command = _shell_command(command, command_args)
        if not acp_command:
            acp_command = DEFAULT_PROVIDER_COMMANDS.get(provider, "")
        if not acp_command:
            raise AcpError("acp-command is required when provider=custom", code="acp.config_invalid")

        workspace = _first_config_value(config, WORKSPACE_CONFIG_KEYS)
        if not workspace:
            if location == "remote-ssh":
                workspace = _first_config_value(config, REMOTE_WORKSPACE_CONFIG_KEYS)
            else:
                workspace = _first_config_value(config, LOCAL_WORKSPACE_CONFIG_KEYS) or os.getcwd()
        if location == "remote-ssh" and not workspace:
            raise AcpError("workspace is required when location=remote-ssh", code="acp.config_invalid")

        if location == "local":
            workspace = local_workspace(
                _first_config_value(config, WORKSPACE_CONFIG_KEYS)
                or _first_config_value(config, LOCAL_WORKSPACE_CONFIG_KEYS)
            )

        ssh_target = _first_config_value(config, SSH_TARGET_CONFIG_KEYS)
        if location == "remote-ssh" and not ssh_target:
            raise AcpError("ssh-target is required when location=remote-ssh", code="acp.config_invalid")

        daemon_id = _first_config_value(config, DAEMON_ID_CONFIG_KEYS)
        if location == "daemon" and not daemon_id:
            raise AcpError("daemon-id is required when location=daemon", code="acp.config_invalid")
        # The daemon hub is a process-wide socket keyed by daemon ID. Until it
        # can bind clients and jobs to a full installation capability, accepting
        # daemon runs in a shared Worker would grant another tenant access.
        if location == "daemon" and os.environ.get("LANGBOT_PLUGIN_RUNTIME_PROFILE") == "shared":
            raise AcpError("daemon mode is unavailable in shared runtime", code="acp.config_invalid")

        remote_shell = str(config.get("remote-shell", "bash") or "bash").strip()
        if remote_shell not in SUPPORTED_REMOTE_SHELLS:
            raise AcpError(
                f"remote-shell must be one of: {', '.join(sorted(SUPPORTED_REMOTE_SHELLS))}",
                code="acp.config_invalid",
            )

        permission_decision = str(config.get("permission-decision", "allow_once") or "allow_once").strip()
        if permission_decision not in {"allow_once", "reject_once", "first"}:
            raise AcpError(
                "permission-decision must be allow_once, reject_once, or first",
                code="acp.config_invalid",
            )

        langbot_assets_mode = str(config.get("langbot-assets-mode", "auto") or "auto").strip()
        if langbot_assets_mode not in SUPPORTED_LANGBOT_ASSET_MODES:
            raise AcpError(
                "langbot-assets-mode must be auto, ephemeral, or gateway",
                code="acp.config_invalid",
            )

        return {
            "provider": provider,
            "location": location,
            "acp_command": acp_command,
            "workspace": workspace,
            "cwd": workspace if location == "local" else None,
            "session_cwd": workspace,
            "env": {str(k): str(v) for k, v in _parse_json_object(config.get("env-json"), label="env-json").items()},
            "ssh_target": ssh_target,
            "ssh_port": _to_int(config.get("ssh-port"), 22),
            "ssh_identity_file": _first_config_value(config, SSH_IDENTITY_FILE_CONFIG_KEYS),
            "ssh_connect_timeout": _to_int(config.get("ssh-connect-timeout"), 10),
            "ssh_extra_options": _parse_args(config.get("ssh-extra-options")),
            "daemon_id": daemon_id,
            "daemon_connect_timeout": _to_float(config.get("daemon-connect-timeout"), 30.0),
            "remote_shell": remote_shell,
            "timeout": _to_float(config.get("timeout"), 300.0),
            "startup_timeout": _to_float(config.get("startup-timeout"), 30.0),
            "initialize_timeout": _to_float(config.get("initialize-timeout"), 30.0),
            "reuse_session": _to_bool(config.get("reuse-session"), True),
            "create_session_if_missing": _to_bool(config.get("create-session-if-missing"), True),
            "streaming": _to_bool(config.get("streaming"), True),
            "permission_decision": permission_decision,
            "append_run_scope_prompt": _to_bool(config.get("append-run-scope-prompt"), True),
            "mcp_bridge_enabled": _to_bool(
                config.get("langbot-assets-enabled", config.get("mcp-bridge-enabled")),
                True,
            ),
            "mcp_bridge_host": str(config.get("mcp-bridge-host", "127.0.0.1") or "127.0.0.1").strip(),
            "mcp_bridge_port": _to_int(config.get("mcp-bridge-port"), 0),
            "mcp_bridge_request_timeout": _to_float(config.get("mcp-bridge-request-timeout"), 60.0),
            "mcp_bridge_transport": str(config.get("mcp-bridge-transport", "auto") or "auto").strip(),
            "mcp_public_url": str(config.get("mcp-public-url", "") or "").strip(),
            "mcp_servers": _parse_json_list(config.get("mcp-servers-json"), label="mcp-servers-json"),
            "langbot_assets_mode": langbot_assets_mode,
            "asset_gateway_host": str(
                config.get("langbot-assets-gateway-host", config.get("mcp-bridge-host", "127.0.0.1")) or "127.0.0.1"
            ).strip(),
            "asset_gateway_port": _to_int(
                config.get("langbot-assets-gateway-port", config.get("mcp-bridge-port")),
                0,
            ),
            "asset_gateway_request_timeout": _to_float(
                config.get("langbot-assets-gateway-request-timeout", config.get("mcp-bridge-request-timeout")),
                60.0,
            ),
            "asset_gateway_token_ttl": _to_float(config.get("langbot-assets-token-ttl"), 3600.0),
            "asset_gateway_public_url": _first_config_value(config, ASSET_GATEWAY_PUBLIC_URL_CONFIG_KEYS),
            "daemon_hub": agent_runtime_daemon_config_from_plugin_config(
                self.get_plugin_config(),
                env_prefix="LANGBOT_ACP_DAEMON",
            ),
        }

    def _input_text(self, ctx: RunnerContext) -> str:
        return ctx.input.to_text()

    def _with_run_scope_prompt(self, ctx: RunnerContext, input_text: str) -> str:
        resources = json.dumps(_resource_summary(ctx), ensure_ascii=True, separators=(",", ":"))
        return (
            "System instructions from LangBot:\n"
            f"- Current LangBot run_id: {ctx.run_id}\n"
            "- The injected LangBot MCP server is already scoped to this run. Follow its tool schemas exactly.\n"
            "- Call LangBot MCP tools directly in this ACP session when possible.\n"
            "- Do not launch background agents, workflows, or tasks. If this model can only call MCP tools through an Agent, "
            "launch at most one Agent with run_in_background=false and wait for it to finish.\n"
            "- Wait for each LangBot MCP tool result before replying to the user.\n"
            "- Do not add run_id or other fields to MCP tool calls unless the tool schema asks for them.\n"
            "- If a LangBot MCP call is rejected, stop and report the error.\n"
            f"- Authorized LangBot resources and MCP bridge tools for this run: {resources}\n\n"
            "User input:\n"
            f"{input_text}"
        )

    def _stored_session_id(self, ctx: RunnerContext) -> str:
        return str(ctx.state.conversation.get(ACP_SESSION_STATE_KEY) or "").strip()

    def _mcp_server_to_acp(self, server: AgentMCPServerConfig) -> dict[str, typing.Any]:
        if server.transport == "http":
            return {
                "name": server.name,
                "type": "http",
                "url": server.url,
                "headers": _mcp_headers_to_acp(server.headers),
            }
        if server.transport != "stdio":
            raise AcpError(f"unsupported MCP transport: {server.transport}", code="acp.mcp_bridge_invalid")
        return {
            "name": server.name,
            "type": "stdio",
            "command": server.command,
            "args": list(server.args),
            "env": _mcp_env_to_acp(server.env),
        }

    def _mcp_servers(
        self,
        ctx: RunnerContext,
        config: dict[str, typing.Any],
    ) -> tuple[AgentRunMCPAccess | None, list[dict[str, typing.Any]]]:
        servers = [server for server in config["mcp_servers"] if isinstance(server, dict)]
        if not config["mcp_bridge_enabled"]:
            return None, servers

        if config["mcp_bridge_transport"] not in {"auto", "stdio", "http"}:
            raise AcpError("mcp-bridge-transport must be auto, stdio, or http", code="acp.config_invalid")

        access = AgentRunMCPAccess(
            self.get_run_api(ctx),
            ctx,
            enabled=True,
            location=config["location"],
            mode=config["langbot_assets_mode"],
            transport=config["mcp_bridge_transport"],
            bridge_host=config["mcp_bridge_host"],
            bridge_port=config["mcp_bridge_port"],
            bridge_public_url=config["mcp_public_url"],
            bridge_request_timeout=config["mcp_bridge_request_timeout"],
            gateway_host=config["asset_gateway_host"],
            gateway_port=config["asset_gateway_port"],
            gateway_public_url=config["asset_gateway_public_url"],
            gateway_request_timeout=config["asset_gateway_request_timeout"],
            gateway_token_ttl=config["asset_gateway_token_ttl"],
        )
        access.start()
        if access.server_config is not None:
            servers.append(self._mcp_server_to_acp(access.server_config))
        return access, servers

    def _launch_config(self, config: dict[str, typing.Any], access: AgentRunMCPAccess | None) -> dict[str, typing.Any]:
        unset_env = CLAUDE_ENV_VARS_TO_UNSET if config["provider"] == "claude-code" else ()
        if config["location"] == "local":
            argv = _parse_args(config["acp_command"])
            if not argv:
                raise AcpError("acp-command is required", code="acp.config_invalid")
            return {
                "command": argv[0],
                "args": argv[1:],
                "cwd": config["cwd"],
                "unset_env": unset_env,
            }

        ssh_args = [
            "-T",
            "-o",
            "BatchMode=yes",
            "-o",
            "ExitOnForwardFailure=yes",
            "-o",
            f"ConnectTimeout={config['ssh_connect_timeout']}",
            "-p",
            str(config["ssh_port"]),
        ]
        if config["ssh_identity_file"]:
            ssh_args.extend(["-i", config["ssh_identity_file"]])
        ssh_args.extend(config["ssh_extra_options"])

        if access is not None and access.reverse_tunnel is not None:
            ssh_args.extend(access.reverse_tunnel.ssh_args())

        ssh_args.append(config["ssh_target"])
        ssh_args.append(
            _remote_shell_command(
                remote_shell=config["remote_shell"],
                workspace=config["workspace"],
                acp_command=config["acp_command"],
                env=config["env"],
                unset_env=unset_env,
            )
        )
        return {
            "command": "ssh",
            "args": ssh_args,
            "cwd": None,
            "unset_env": unset_env,
        }

    async def _create_or_resume_session(
        self,
        client: AcpStdioClient,
        initialize_result: dict[str, typing.Any],
        ctx: RunnerContext,
        config: dict[str, typing.Any],
        mcp_servers: list[dict[str, typing.Any]],
        stored_session_id: str,
    ) -> tuple[str, bool]:
        capabilities = initialize_result.get("agentCapabilities")
        if not isinstance(capabilities, dict):
            capabilities = {}

        if stored_session_id and config["reuse_session"]:
            if _runtime_has_method(capabilities, "session/resume"):
                result = await client.request(
                    "session/resume",
                    build_session_params(
                        provider=config["provider"],
                        session_id=stored_session_id,
                        cwd=config["session_cwd"],
                        mcp_servers=mcp_servers,
                    ),
                    timeout=config["timeout"],
                )
                return _extract_session_id(result) or stored_session_id, False
            if _runtime_has_method(capabilities, "session/load"):
                result = await client.request(
                    "session/load",
                    build_session_params(
                        provider=config["provider"],
                        session_id=stored_session_id,
                        cwd=config["session_cwd"],
                        mcp_servers=mcp_servers,
                    ),
                    timeout=config["timeout"],
                )
                await client.drain_updates()
                return _extract_session_id(result) or stored_session_id, False

        if not config["create_session_if_missing"]:
            raise AcpError(
                "no stored ACP session and create-session-if-missing is disabled", code="acp.session_missing"
            )

        result = await client.request(
            "session/new",
            build_session_params(
                provider=config["provider"],
                cwd=config["session_cwd"],
                mcp_servers=mcp_servers,
            ),
            timeout=config["timeout"],
        )
        session_id = _extract_session_id(result)
        if not session_id:
            raise AcpError(f"ACP session/new did not return a session id: {result!r}", code="acp.response_invalid")
        return session_id, True

    async def _stream_prompt_results(
        self,
        client: AcpStdioClient,
        ctx: RunnerContext,
        session_id: str,
        prompt_blocks: list[dict[str, typing.Any]],
        *,
        timeout: float,
        streaming: bool,
    ) -> typing.AsyncGenerator[RunnerResult, None]:
        prompt_request = client.send_request(
            "session/prompt",
            {
                "sessionId": session_id,
                "prompt": prompt_blocks,
            },
        )

        sequence = 0
        final_text_parts: list[str] = []
        active_message_id = ""
        active_tool_calls: dict[str, str] = {}
        deadline = time.monotonic() + timeout

        while True:
            if prompt_request.future.done():
                update = client.next_update_nowait()
                if update is None:
                    break
            else:
                if time.monotonic() >= deadline:
                    raise TimeoutError
                update = await client.next_update(timeout=0.1)
                if update is None:
                    continue

            text = _agent_text_from_update(update)
            if text:
                message_id = _agent_message_id(update)
                if message_id:
                    if active_message_id and message_id != active_message_id:
                        final_text_parts.clear()
                    active_message_id = message_id
                final_text_parts.append(text)
                if streaming:
                    sequence += 1
                    yield RunnerResult.message_delta(
                        ctx.run_id,
                        MessageChunk(
                            role="assistant",
                            content=text,
                            all_content="".join(final_text_parts),
                            msg_sequence=sequence,
                        ),
                    )

            tool_payload = _tool_update_payload(update)
            if tool_payload:
                tool_call_id = str(tool_payload.get("toolCallId") or tool_payload.get("id") or "")
                observed_tool_name = _tool_update_name(tool_payload)
                status = str(tool_payload.get("status") or "")
                if tool_call_id and tool_call_id not in active_tool_calls:
                    active_tool_calls[tool_call_id] = observed_tool_name
                    tool_name = observed_tool_name
                    yield RunnerResult.tool_call_started(ctx.run_id, tool_call_id, tool_name, {})
                if tool_call_id and status in {"completed", "failed", "cancelled"}:
                    tool_name = active_tool_calls.get(tool_call_id, observed_tool_name)
                    yield RunnerResult.tool_call_completed(
                        ctx.run_id,
                        tool_call_id,
                        tool_name,
                        result=tool_payload if status == "completed" else None,
                        error=None if status == "completed" else json.dumps(tool_payload, ensure_ascii=False),
                    )

        await prompt_request.wait(timeout=timeout)
        final_text = "".join(final_text_parts).strip()
        if not final_text:
            yield RunnerResult.run_failed(
                ctx.run_id,
                error="ACP agent returned no assistant text",
                code="acp.empty_response",
            )
            return

        yield RunnerResult.message_completed(ctx.run_id, Message(role="assistant", content=final_text))
        yield RunnerResult.run_completed(ctx.run_id, finish_reason="stop")

    def _daemon_payload(
        self,
        ctx: RunnerContext,
        config: dict[str, typing.Any],
        prompt_text: str,
        stored_session_id: str,
    ) -> dict[str, typing.Any]:
        return {
            "run_id": ctx.run_id,
            "prompt_text": prompt_text,
            "input": ctx.input.model_dump(mode="json") if hasattr(ctx.input, "model_dump") else {},
            "input_text": self._input_text(ctx),
            "config": {
                "provider": config["provider"],
                "acp_command": config["acp_command"],
                "workspace": config["workspace"],
                "cwd": config["workspace"] or None,
                "session_cwd": config["session_cwd"],
                "env": config["env"],
                "timeout": config["timeout"],
                "startup_timeout": config["startup_timeout"],
                "initialize_timeout": config["initialize_timeout"],
                "reuse_session": config["reuse_session"],
                "create_session_if_missing": config["create_session_if_missing"],
                "streaming": config["streaming"],
                "permission_decision": config["permission_decision"],
                "stored_session_id": stored_session_id,
                "mcp_servers": config["mcp_servers"],
                "langbot_assets_enabled": config["mcp_bridge_enabled"],
                "mcp_request_timeout": config["mcp_bridge_request_timeout"],
            },
        }

    async def _run_daemon(
        self,
        ctx: RunnerContext,
        config: dict[str, typing.Any],
        prompt_text: str,
        stored_session_id: str,
    ) -> typing.AsyncGenerator[RunnerResult, None]:
        hub = get_agent_runtime_daemon_hub("acp", error_code_prefix="acp")
        await hub.start(
            host=config["daemon_hub"]["host"],
            port=config["daemon_hub"]["port"],
            token=config["daemon_hub"]["token"],
        )

        await hub.wait_for_daemon(config["daemon_id"], config["daemon_connect_timeout"])
        tools = AgentRunExternalTools(self.get_run_api(ctx), ctx) if config["mcp_bridge_enabled"] else None
        payload = self._daemon_payload(ctx, config, prompt_text, stored_session_id)
        async with contextlib.aclosing(
            hub.run_job(
                daemon_id=config["daemon_id"],
                payload=payload,
                tools=tools,
                timeout=config["timeout"],
            )
        ) as event_stream:
            async for event in event_stream:
                event.setdefault("run_id", ctx.run_id)
                yield RunnerResult.model_validate(event)

    async def run(self, ctx: RunnerContext) -> typing.AsyncGenerator[RunnerResult, None]:
        try:
            config = self._validate_config(ctx)
        except AcpError as exc:
            yield RunnerResult.run_failed(ctx.run_id, error=exc.message, code=exc.code, retryable=exc.retryable)
            return

        input_text = self._input_text(ctx)
        if not has_acp_prompt_input(input_text, ctx.input):
            yield RunnerResult.run_failed(ctx.run_id, error="input text is required", code="acp.empty_input")
            return

        prompt_text = self._with_run_scope_prompt(ctx, input_text) if config["append_run_scope_prompt"] else input_text

        # The first turn carries the run-scope system prompt and the original
        # multimodal input; steering follow-ups are plain text resumed against the
        # same agent session (the Host queues them while this run is active).
        first_turn = {"value": True}

        def run_turn(turn_prompt: str, resume_session_id: str) -> typing.AsyncGenerator[RunnerResult, None]:
            if first_turn["value"]:
                first_turn["value"] = False
                return self._run_acp_turn(ctx, config, turn_prompt, resume_session_id, ctx.input)
            return self._run_acp_turn(ctx, config, turn_prompt, resume_session_id, AgentInput(text=turn_prompt))

        initial_resume = self._stored_session_id(ctx) if config["reuse_session"] else ""
        async for result in run_with_steering(
            ctx,
            lambda: self.get_run_api(ctx),
            run_turn,
            initial_prompt=prompt_text,
            initial_resume_session_id=initial_resume,
            session_state_key=ACP_SESSION_STATE_KEY,
        ):
            yield result

    async def _run_acp_turn(
        self,
        ctx: RunnerContext,
        config: dict[str, typing.Any],
        prompt_text: str,
        resume_session_id: str,
        agent_input: AgentInput,
    ) -> typing.AsyncGenerator[RunnerResult, None]:
        """Run a single ACP prompt turn, resuming ``resume_session_id`` when set."""
        stored_session_id = resume_session_id or (self._stored_session_id(ctx) if config["reuse_session"] else "")

        if config["location"] == "daemon":
            try:
                async for result in self._run_daemon(ctx, config, prompt_text, stored_session_id):
                    yield result
            except AgentRuntimeDaemonError as exc:
                yield RunnerResult.run_failed(ctx.run_id, error=exc.message, code=exc.code, retryable=exc.retryable)
            except Exception as exc:
                logger.exception("ACP daemon transport failed: %s", exc)
                yield RunnerResult.run_failed(ctx.run_id, error=str(exc), code="acp.daemon_unexpected")
            return

        access = None
        client: AcpStdioClient | None = None
        try:
            access, mcp_servers = self._mcp_servers(ctx, config)
            launch_config = self._launch_config(config, access)
            client = AcpStdioClient(
                command=launch_config["command"],
                args=launch_config["args"],
                cwd=launch_config["cwd"],
                env=config["env"],
                unset_env=launch_config["unset_env"],
                permission_decision=config["permission_decision"],
                startup_timeout=config["startup_timeout"],
            )
            async with client:
                initialize_result = await client.initialize(timeout=config["initialize_timeout"])
                session_id, created = await self._create_or_resume_session(
                    client,
                    initialize_result,
                    ctx,
                    config,
                    mcp_servers,
                    stored_session_id,
                )

                if created or stored_session_id != session_id:
                    yield RunnerResult.state_updated(
                        ctx.run_id,
                        ACP_SESSION_STATE_KEY,
                        session_id,
                        scope="conversation",
                    )

                prompt_blocks = acp_prompt_blocks(prompt_text, agent_input, prompt_capabilities(initialize_result))
                async for result in self._stream_prompt_results(
                    client,
                    ctx,
                    session_id,
                    prompt_blocks,
                    timeout=config["timeout"],
                    streaming=config["streaming"],
                ):
                    yield result

        except AcpError as exc:
            if client and client.stderr_tail.strip():
                logger.warning("ACP runner failed with stderr tail: %s", client.stderr_tail.strip())
            yield RunnerResult.run_failed(ctx.run_id, error=exc.message, code=exc.code, retryable=exc.retryable)
        except TimeoutError:
            yield RunnerResult.run_failed(ctx.run_id, error="ACP request timed out", code="acp.timeout", retryable=True)
        except Exception as exc:
            logger.exception("ACP runner unexpected error: %s", exc)
            yield RunnerResult.run_failed(ctx.run_id, error=f"ACP runner error: {exc}", code="acp.unexpected_error")
        finally:
            if access is not None:
                access.stop()