"""
Runloop sandbox (https://runloop.ai) implementation.

This module provides a Runloop-backed sandbox client/session implementation backed by
`runloop_api_client.sdk.AsyncRunloopSDK`.

The `runloop_api_client` dependency is optional, so package-level exports should guard imports of
this module. Within this module, Runloop SDK imports are lazy so users without the extra can still
import the package.
"""

from __future__ import annotations

import asyncio
import base64
import io
import json
import logging
import posixpath
import shlex
import uuid
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING, Any, Literal, cast
from urllib.parse import urlsplit

from pydantic import BaseModel, Field
from runloop_api_client.types import (
    AfterIdle as _RunloopSdkAfterIdle,
    LaunchParameters as _RunloopSdkLaunchParameters,
)
from runloop_api_client.types.shared.launch_parameters import (
    UserParameters as _RunloopSdkUserParameters,
)

from ....sandbox.entries import Mount
from ....sandbox.errors import (
    ExecTimeoutError,
    ExecTransportError,
    ExposedPortUnavailableError,
    WorkspaceArchiveReadError,
    WorkspaceArchiveWriteError,
    WorkspaceReadNotFoundError,
    WorkspaceWriteTypeError,
)
from ....sandbox.manifest import Manifest
from ....sandbox.session import SandboxSession, SandboxSessionState
from ....sandbox.session.base_sandbox_session import BaseSandboxSession
from ....sandbox.session.dependencies import Dependencies
from ....sandbox.session.manager import Instrumentation
from ....sandbox.session.runtime_helpers import RESOLVE_WORKSPACE_PATH_HELPER, RuntimeHelperScript
from ....sandbox.session.sandbox_client import BaseSandboxClient, BaseSandboxClientOptions
from ....sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot
from ....sandbox.types import ExecResult, ExposedPortEndpoint, User
from ....sandbox.util.tar_utils import UnsafeTarMemberError, validate_tar_bytes
from ....sandbox.workspace_paths import coerce_posix_path, posix_path_as_path, sandbox_path_str

if TYPE_CHECKING:
    from runloop_api_client.sdk.async_execution_result import (
        AsyncExecutionResult as RunloopAsyncExecutionResult,
    )
    from runloop_api_client.sdk.async_snapshot import AsyncSnapshot as RunloopAsyncSnapshot
    from runloop_api_client.types.devbox_view import DevboxView as RunloopDevboxView

DEFAULT_RUNLOOP_WORKSPACE_ROOT = "/home/user"
DEFAULT_RUNLOOP_ROOT_WORKSPACE_ROOT = "/root"
_RUNLOOP_DEFAULT_HOME = PurePosixPath("/home/user")
_RUNLOOP_ROOT_HOME = PurePosixPath("/root")
_RUNLOOP_SANDBOX_SNAPSHOT_MAGIC = b"RUNLOOP_SANDBOX_SNAPSHOT_V1\n"

logger = logging.getLogger(__name__)

RunloopAfterIdle = _RunloopSdkAfterIdle
RunloopLaunchParameters = _RunloopSdkLaunchParameters
RunloopUserParameters = _RunloopSdkUserParameters


@dataclass(frozen=True)
class _RunloopSdkImports:
    async_sdk: type[Any]
    api_connection_error: type[BaseException]
    api_response_validation_error: type[BaseException]
    api_status_error: type[BaseException]
    api_timeout_error: type[BaseException]
    not_found_error: type[BaseException]
    polling_config: type[Any] | None
    polling_timeout: type[BaseException] | None
    runloop_error: type[BaseException]


_RUNLOOP_SDK_IMPORTS: _RunloopSdkImports | None = None


def _import_runloop_sdk() -> _RunloopSdkImports:
    global _RUNLOOP_SDK_IMPORTS
    if _RUNLOOP_SDK_IMPORTS is not None:
        return _RUNLOOP_SDK_IMPORTS

    try:
        from runloop_api_client import (
            APIConnectionError,
            APIResponseValidationError,
            APIStatusError,
            APITimeoutError,
            NotFoundError,
            RunloopError,
        )
        from runloop_api_client.sdk import AsyncRunloopSDK
    except ImportError as e:
        raise ImportError(
            "RunloopSandboxClient requires the optional `runloop_api_client` dependency.\n"
            "Install the Runloop extra before using this sandbox backend."
        ) from e

    polling_config: type[Any] | None = None
    polling_timeout: type[BaseException] | None = None
    try:
        from runloop_api_client.lib.polling import (
            PollingConfig as RunloopPollingConfig,
            PollingTimeout as RunloopPollingTimeout,
        )
    except ImportError:
        pass
    else:
        polling_config = RunloopPollingConfig
        polling_timeout = RunloopPollingTimeout

    _RUNLOOP_SDK_IMPORTS = _RunloopSdkImports(
        async_sdk=AsyncRunloopSDK,
        api_connection_error=APIConnectionError,
        api_response_validation_error=APIResponseValidationError,
        api_status_error=APIStatusError,
        api_timeout_error=APITimeoutError,
        not_found_error=NotFoundError,
        polling_config=polling_config,
        polling_timeout=polling_timeout,
        runloop_error=RunloopError,
    )
    return _RUNLOOP_SDK_IMPORTS


def _encode_runloop_snapshot_ref(*, snapshot_id: str) -> bytes:
    body = json.dumps({"snapshot_id": snapshot_id}, separators=(",", ":"), sort_keys=True).encode(
        "utf-8"
    )
    return _RUNLOOP_SANDBOX_SNAPSHOT_MAGIC + body


def _decode_runloop_snapshot_ref(raw: bytes) -> str | None:
    if not raw.startswith(_RUNLOOP_SANDBOX_SNAPSHOT_MAGIC):
        return None
    body = raw[len(_RUNLOOP_SANDBOX_SNAPSHOT_MAGIC) :]
    try:
        obj = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    snapshot_id = obj.get("snapshot_id") if isinstance(obj, dict) else None
    return snapshot_id if isinstance(snapshot_id, str) and snapshot_id else None


def _runloop_json_safe_body(body: object) -> tuple[str, object] | None:
    if isinstance(body, str | int | float | bool) or body is None:
        return ("provider_body", body)
    if isinstance(body, dict | list):
        try:
            json.dumps(body)
        except TypeError:
            return ("provider_body_repr", repr(body))
        return ("provider_body", body)
    return ("provider_body_repr", repr(body))


def _runloop_error_context(
    exc: BaseException,
    *,
    backend_detail: str | None = None,
) -> dict[str, object]:
    context: dict[str, object] = {
        "backend": "runloop",
        "cause_type": type(exc).__name__,
    }
    if backend_detail is not None:
        context["detail"] = backend_detail

    message = getattr(exc, "message", None)
    if isinstance(message, str) and message:
        context["provider_message"] = message
    else:
        provider_message = str(exc)
        if provider_message:
            context["provider_message"] = provider_message

    status_code = getattr(exc, "status_code", None)
    response = getattr(exc, "response", None)
    if not isinstance(status_code, int):
        response_status = getattr(response, "status_code", None)
        if isinstance(response_status, int):
            status_code = response_status
    if isinstance(status_code, int):
        context["http_status"] = status_code

    request = getattr(exc, "request", None)
    request_url = getattr(request, "url", None)
    if request_url is not None:
        context["request_url"] = str(request_url)
    request_method = getattr(request, "method", None)
    if isinstance(request_method, str) and request_method:
        context["request_method"] = request_method

    if hasattr(exc, "body"):
        safe_body = _runloop_json_safe_body(getattr(exc, "body", None))
        if safe_body is not None:
            context[safe_body[0]] = safe_body[1]

    return context


def _is_runloop_timeout(exc: BaseException) -> bool:
    polling_timeout = _import_runloop_sdk().polling_timeout
    if polling_timeout is not None and isinstance(exc, polling_timeout):
        return True
    if isinstance(exc, _import_runloop_sdk().api_timeout_error):
        return True
    if isinstance(exc, _import_runloop_sdk().api_status_error):
        status_code = getattr(exc, "status_code", None)
        response = getattr(exc, "response", None)
        if not isinstance(status_code, int):
            response_status = getattr(response, "status_code", None)
            if isinstance(response_status, int):
                status_code = response_status
        return status_code == 408
    return False


def _runloop_status_code(exc: BaseException) -> int | None:
    status_code = getattr(exc, "status_code", None)
    response = getattr(exc, "response", None)
    if not isinstance(status_code, int):
        response_status = getattr(response, "status_code", None)
        if isinstance(response_status, int):
            status_code = response_status
    return status_code if isinstance(status_code, int) else None


def _runloop_error_message(exc: BaseException) -> str | None:
    body = getattr(exc, "body", None)
    if isinstance(body, dict):
        message = body.get("message") or body.get("error")
        if isinstance(message, str) and message:
            return message

    message = getattr(exc, "message", None)
    if isinstance(message, str) and message:
        return message

    if exc.args:
        first = exc.args[0]
        if isinstance(first, str) and first:
            return first

    return None


def _runloop_provider_error_types() -> tuple[type[BaseException], ...]:
    sdk_imports = _import_runloop_sdk()
    return (
        sdk_imports.api_connection_error,
        sdk_imports.api_response_validation_error,
        sdk_imports.api_status_error,
        sdk_imports.runloop_error,
    )


def _is_runloop_not_found(exc: BaseException) -> bool:
    return isinstance(exc, _import_runloop_sdk().not_found_error)


def _is_runloop_conflict(exc: BaseException) -> bool:
    if not isinstance(exc, _import_runloop_sdk().api_status_error):
        return False

    status_code = _runloop_status_code(exc)
    if status_code == 409:
        return True

    message = _runloop_error_message(exc)
    if status_code == 400 and isinstance(message, str):
        return "already exists" in message.lower()

    return False


def _runloop_polling_config(*, timeout_s: float | None) -> object | None:
    if timeout_s is None:
        return None
    polling_config = _import_runloop_sdk().polling_config
    if polling_config is None:
        return None
    return cast(object, polling_config(timeout_seconds=max(float(timeout_s), 0.001)))


def _is_runloop_provider_error(exc: BaseException) -> bool:
    return isinstance(
        exc,
        _runloop_provider_error_types(),
    )


class RunloopTimeouts(BaseModel):
    """Timeout configuration for Runloop sandbox operations."""

    model_config = {"frozen": True}

    exec_timeout_unbounded_s: float = Field(default=24 * 60 * 60, ge=1)
    create_s: float = Field(default=300.0, ge=1)
    keepalive_s: float = Field(default=10.0, ge=1)
    cleanup_s: float = Field(default=30.0, ge=1)
    fast_op_s: float = Field(default=30.0, ge=1)
    file_upload_s: float = Field(default=1800.0, ge=1)
    file_download_s: float = Field(default=1800.0, ge=1)
    snapshot_s: float = Field(default=300.0, ge=1)
    suspend_s: float = Field(default=120.0, ge=1)
    resume_s: float = Field(default=300.0, ge=1)


class RunloopTunnelConfig(BaseModel):
    """Runloop public tunnel configuration."""

    model_config = {"frozen": True}

    auth_mode: Literal["open", "authenticated"] | None = None
    http_keep_alive: bool | None = None
    wake_on_http: bool | None = None


class RunloopGatewaySpec(BaseModel):
    """Runloop agent gateway binding."""

    model_config = {"frozen": True}

    gateway: str = Field(min_length=1)
    secret: str = Field(min_length=1)


class RunloopMcpSpec(BaseModel):
    """Runloop MCP gateway binding."""

    model_config = {"frozen": True}

    mcp_config: str = Field(min_length=1)
    secret: str = Field(min_length=1)


def _normalize_runloop_user_parameters(
    user_parameters: RunloopUserParameters | dict[str, object] | None,
) -> RunloopUserParameters | None:
    if isinstance(user_parameters, RunloopUserParameters):
        return user_parameters
    if user_parameters is None:
        return None
    if isinstance(user_parameters, BaseModel):
        return RunloopUserParameters.model_validate(user_parameters.model_dump(mode="json"))
    return RunloopUserParameters.model_validate(user_parameters)


def _normalize_runloop_launch_parameters(
    launch_parameters: RunloopLaunchParameters | dict[str, object] | None,
) -> RunloopLaunchParameters | None:
    if isinstance(launch_parameters, RunloopLaunchParameters):
        return launch_parameters
    if launch_parameters is None:
        return None
    if isinstance(launch_parameters, BaseModel):
        return RunloopLaunchParameters.model_validate(launch_parameters.model_dump(mode="json"))
    return RunloopLaunchParameters.model_validate(launch_parameters)


def _normalize_runloop_tunnel_config(
    tunnel: RunloopTunnelConfig | dict[str, object] | None,
) -> RunloopTunnelConfig | None:
    if isinstance(tunnel, RunloopTunnelConfig):
        return tunnel
    if tunnel is None:
        return None
    if isinstance(tunnel, BaseModel):
        return RunloopTunnelConfig.model_validate(tunnel.model_dump(mode="json"))
    return RunloopTunnelConfig.model_validate(tunnel)


class RunloopSandboxClientOptions(BaseSandboxClientOptions):
    """Client options for the Runloop sandbox."""

    type: Literal["runloop"] = "runloop"
    blueprint_id: str | None = None
    blueprint_name: str | None = None
    env_vars: dict[str, str] | None = None
    pause_on_exit: bool = False
    name: str | None = None
    timeouts: RunloopTimeouts | dict[str, object] | None = None
    exposed_ports: tuple[int, ...] = ()
    user_parameters: RunloopUserParameters | dict[str, object] | None = None
    launch_parameters: RunloopLaunchParameters | dict[str, object] | None = None
    tunnel: RunloopTunnelConfig | dict[str, object] | None = None
    gateways: dict[str, RunloopGatewaySpec] | None = None
    mcp: dict[str, RunloopMcpSpec] | None = None
    metadata: dict[str, str] | None = None
    managed_secrets: dict[str, str] | None = None

    def __init__(
        self,
        blueprint_id: str | None = None,
        blueprint_name: str | None = None,
        env_vars: dict[str, str] | None = None,
        pause_on_exit: bool = False,
        name: str | None = None,
        timeouts: RunloopTimeouts | dict[str, object] | None = None,
        exposed_ports: tuple[int, ...] = (),
        user_parameters: RunloopUserParameters | dict[str, object] | None = None,
        launch_parameters: RunloopLaunchParameters | dict[str, object] | None = None,
        tunnel: RunloopTunnelConfig | dict[str, object] | None = None,
        gateways: dict[str, RunloopGatewaySpec] | None = None,
        mcp: dict[str, RunloopMcpSpec] | None = None,
        metadata: dict[str, str] | None = None,
        managed_secrets: dict[str, str] | None = None,
        *,
        type: Literal["runloop"] = "runloop",
    ) -> None:
        super().__init__(
            type=type,
            blueprint_id=blueprint_id,
            blueprint_name=blueprint_name,
            env_vars=env_vars,
            pause_on_exit=pause_on_exit,
            name=name,
            timeouts=timeouts,
            exposed_ports=exposed_ports,
            user_parameters=user_parameters,
            launch_parameters=launch_parameters,
            tunnel=tunnel,
            gateways=gateways,
            mcp=mcp,
            metadata=metadata,
            managed_secrets=managed_secrets,
        )


class RunloopSandboxSessionState(SandboxSessionState):
    """Serializable state for a Runloop-backed session."""

    type: Literal["runloop"] = "runloop"
    devbox_id: str
    blueprint_id: str | None = None
    blueprint_name: str | None = None
    base_env_vars: dict[str, str] = Field(default_factory=dict)
    pause_on_exit: bool = False
    name: str | None = None
    timeouts: RunloopTimeouts = Field(default_factory=RunloopTimeouts)
    user_parameters: RunloopUserParameters | None = None
    launch_parameters: RunloopLaunchParameters | None = None
    tunnel: RunloopTunnelConfig | None = None
    gateways: dict[str, RunloopGatewaySpec] = Field(default_factory=dict)
    mcp: dict[str, RunloopMcpSpec] = Field(default_factory=dict)
    metadata: dict[str, str] = Field(default_factory=dict)
    secret_refs: dict[str, str] = Field(default_factory=dict)


@dataclass(frozen=True)
class RunloopPlatformBlueprintsClient:
    _sdk: Any

    async def list(self, **params: object) -> object:
        return await self._sdk.blueprint.list(**params)

    async def list_public(self, **params: object) -> object:
        return await self._sdk.api.blueprints.list_public(**params)

    def get(self, blueprint_id: str) -> Any:
        return self._sdk.blueprint.from_id(blueprint_id)

    async def logs(self, blueprint_id: str, **params: object) -> object:
        return await self._sdk.api.blueprints.logs(blueprint_id, **params)

    async def create(self, **params: object) -> object:
        return await self._sdk.blueprint.create(**params)

    async def await_build_complete(self, blueprint_id: str, **params: object) -> object:
        return await self._sdk.api.blueprints.await_build_complete(blueprint_id, **params)

    async def delete(self, blueprint_id: str, **params: object) -> object:
        return await self.get(blueprint_id).delete(**params)


@dataclass(frozen=True)
class RunloopPlatformBenchmarksClient:
    _sdk: Any

    async def list(self, **params: object) -> object:
        return await self._sdk.benchmark.list(**params)

    async def list_public(self, **params: object) -> object:
        return await self._sdk.api.benchmarks.list_public(**params)

    def get(self, benchmark_id: str) -> Any:
        return self._sdk.benchmark.from_id(benchmark_id)

    async def create(self, **params: object) -> object:
        return await self._sdk.benchmark.create(**params)

    async def update(self, benchmark_id: str, **params: object) -> object:
        return await self.get(benchmark_id).update(**params)

    async def definitions(self, benchmark_id: str, **params: object) -> object:
        return await self._sdk.api.benchmarks.definitions(benchmark_id, **params)

    async def start_run(self, benchmark_id: str, **params: object) -> object:
        return await self.get(benchmark_id).start_run(**params)

    async def update_scenarios(
        self,
        benchmark_id: str,
        *,
        scenarios_to_add: tuple[str, ...] | Sequence[str] | None = None,
        scenarios_to_remove: tuple[str, ...] | Sequence[str] | None = None,
        **params: object,
    ) -> object:
        return await self._sdk.api.benchmarks.update_scenarios(
            benchmark_id,
            scenarios_to_add=scenarios_to_add,
            scenarios_to_remove=scenarios_to_remove,
            **params,
        )


@dataclass(frozen=True)
class RunloopPlatformSecretsClient:
    _sdk: Any

    async def create(self, *, name: str, value: str, **params: object) -> object:
        return await self._sdk.secret.create(name=name, value=value, **params)

    async def list(self, **params: object) -> object:
        return await self._sdk.secret.list(**params)

    async def get(self, name: str, **params: object) -> object:
        return await self._sdk.api.secrets.retrieve(name, **params)

    async def update(self, *, name: str, value: str, **params: object) -> object:
        return await self._sdk.secret.update(name, value=value, **params)

    async def delete(self, name: str, **params: object) -> object:
        return await self._sdk.secret.delete(name, **params)


@dataclass(frozen=True)
class RunloopPlatformNetworkPoliciesClient:
    _sdk: Any

    async def create(self, **params: object) -> object:
        return await self._sdk.network_policy.create(**params)

    async def list(self, **params: object) -> object:
        return await self._sdk.network_policy.list(**params)

    def get(self, network_policy_id: str) -> Any:
        return self._sdk.network_policy.from_id(network_policy_id)

    async def update(self, network_policy_id: str, **params: object) -> object:
        return await self.get(network_policy_id).update(**params)

    async def delete(self, network_policy_id: str, **params: object) -> object:
        return await self.get(network_policy_id).delete(**params)


@dataclass(frozen=True)
class RunloopPlatformAxonsClient:
    _sdk: Any

    async def create(self, **params: object) -> object:
        return await self._sdk.axon.create(**params)

    async def list(self, **params: object) -> object:
        return await self._sdk.axon.list(**params)

    def get(self, axon_id: str) -> Any:
        return self._sdk.axon.from_id(axon_id)

    async def publish(self, axon_id: str, **params: object) -> object:
        return await self.get(axon_id).publish(**params)

    async def query_sql(self, axon_id: str, **params: object) -> object:
        return await self.get(axon_id).sql.query(**params)

    async def batch_sql(self, axon_id: str, **params: object) -> object:
        return await self.get(axon_id).sql.batch(**params)


@dataclass(frozen=True)
class RunloopPlatformClient:
    """Thin facade over the Runloop SDK's non-devbox platform resources."""

    _sdk: Any

    @property
    def blueprints(self) -> RunloopPlatformBlueprintsClient:
        return RunloopPlatformBlueprintsClient(self._sdk)

    @property
    def benchmarks(self) -> RunloopPlatformBenchmarksClient:
        return RunloopPlatformBenchmarksClient(self._sdk)

    @property
    def secrets(self) -> RunloopPlatformSecretsClient:
        return RunloopPlatformSecretsClient(self._sdk)

    @property
    def network_policies(self) -> RunloopPlatformNetworkPoliciesClient:
        return RunloopPlatformNetworkPoliciesClient(self._sdk)

    @property
    def axons(self) -> RunloopPlatformAxonsClient:
        return RunloopPlatformAxonsClient(self._sdk)


class RunloopSandboxSession(BaseSandboxSession):
    """Runloop-backed sandbox session implementation."""

    state: RunloopSandboxSessionState
    _sdk: Any
    _devbox: Any
    _skip_start: bool

    def __init__(self, *, state: RunloopSandboxSessionState, sdk: Any, devbox: Any) -> None:
        self.state = state
        self._sdk = sdk
        self._devbox = devbox
        self._skip_start = False

    @classmethod
    def from_state(
        cls,
        state: RunloopSandboxSessionState,
        *,
        sdk: Any,
        devbox: Any,
    ) -> RunloopSandboxSession:
        return cls(state=state, sdk=sdk, devbox=devbox)

    @property
    def devbox_id(self) -> str:
        return self.state.devbox_id

    @property
    def runloop_home(self) -> PurePosixPath:
        return _effective_runloop_home(self.state.user_parameters)

    async def _resolved_envs(self) -> dict[str, str]:
        manifest_envs = await self.state.manifest.environment.resolve()
        return {**self.state.base_env_vars, **manifest_envs}

    def _coerce_exec_timeout(self, timeout_s: float | None) -> float:
        if timeout_s is None:
            return float(self.state.timeouts.exec_timeout_unbounded_s)
        if timeout_s <= 0:
            return 0.001
        return float(timeout_s)

    async def start(self) -> None:
        """Resume a reconnected Runloop devbox without replaying full setup when possible.

        `resume()` marks `_skip_start` when it successfully reconnects to a suspended devbox.
        In that path, Runloop reuses the live machine and only reapplies snapshot or ephemeral
        manifest state if the cached workspace fingerprint no longer matches.
        """
        if self._skip_start:
            if await self.state.snapshot.restorable(dependencies=self.dependencies):
                is_running = await self.running()
                fingerprints_match = await self._can_skip_snapshot_restore_on_resume(
                    is_running=is_running
                )
                if fingerprints_match:
                    await self._reapply_ephemeral_manifest_on_resume()
                else:
                    await self._restore_snapshot_into_workspace_on_resume()
                    if self.should_provision_manifest_accounts_on_resume():
                        await self.provision_manifest_accounts()
                    await self._reapply_ephemeral_manifest_on_resume()
            else:
                await self._reapply_ephemeral_manifest_on_resume()
            return
        await super().start()

    async def shutdown(self) -> None:
        """Suspend or delete the underlying Runloop devbox as the final session cleanup step.

        `pause_on_exit=True` maps to Runloop suspension so the same devbox can be resumed later.
        Otherwise the session shuts the devbox down and treats it as disposable.
        """
        try:
            if self.state.pause_on_exit:
                await self._devbox.suspend(timeout=self.state.timeouts.suspend_s)
                await self._devbox.await_suspended()
            else:
                await self._devbox.shutdown(timeout=self.state.timeouts.cleanup_s)
        except Exception:
            pass

    def supports_pty(self) -> bool:
        return False

    async def _validate_path_access(self, path: Path | str, *, for_write: bool = False) -> Path:
        return await self._validate_remote_path_access(path, for_write=for_write)

    def _runtime_helpers(self) -> tuple[RuntimeHelperScript, ...]:
        return (RESOLVE_WORKSPACE_PATH_HELPER,)

    async def _wrap_command_in_workspace_context(self, command: str) -> str:
        root_q = shlex.quote(self.state.manifest.root)
        envs = await self._resolved_envs()
        if not envs:
            return f"cd {root_q} && {command}"

        env_assignments = " ".join(
            shlex.quote(f"{key}={value}") for key, value in sorted(envs.items())
        )
        return f"cd {root_q} && env -- {env_assignments} {command}"

    async def _exec_internal(
        self,
        *command: str | Path,
        timeout: float | None = None,
    ) -> ExecResult:
        cmd_str = await self._wrap_command_in_workspace_context(shlex.join(str(c) for c in command))
        return await self._run_exec_command(
            cmd_str,
            command=command,
            timeout=timeout,
        )

    async def _run_exec_command(
        self,
        cmd_str: str,
        *,
        command: tuple[str | Path, ...],
        timeout: float | None,
    ) -> ExecResult:
        caller_timeout = self._coerce_exec_timeout(timeout)
        request_timeout = min(caller_timeout, self.state.timeouts.fast_op_s)
        polling_config = _runloop_polling_config(timeout_s=caller_timeout)

        try:
            result: RunloopAsyncExecutionResult = await asyncio.wait_for(
                self._devbox.cmd.exec(
                    cmd_str,
                    timeout=request_timeout,
                    polling_config=polling_config,
                ),
                timeout=caller_timeout,
            )
            stdout = (await result.stdout()).encode("utf-8", errors="replace")
            stderr = (await result.stderr()).encode("utf-8", errors="replace")
            exit_code = int(result.exit_code or 0)
            return ExecResult(stdout=stdout, stderr=stderr, exit_code=exit_code)
        except asyncio.TimeoutError as e:
            raise ExecTimeoutError(
                command=command,
                timeout_s=timeout,
                context=_runloop_error_context(e, backend_detail="exec_timeout"),
                cause=e,
            ) from e
        except Exception as e:
            if _is_runloop_timeout(e):
                raise ExecTimeoutError(
                    command=command,
                    timeout_s=timeout,
                    context=_runloop_error_context(e, backend_detail="exec_timeout"),
                    cause=e,
                ) from e
            if _is_runloop_provider_error(e):
                raise ExecTransportError(
                    command=command,
                    context=_runloop_error_context(e, backend_detail="exec_failed"),
                    cause=e,
                ) from e
            raise ExecTransportError(command=command, cause=e) from e

    async def _ensure_tunnel_url(self, port: int) -> str:
        try:
            url = await self._devbox.get_tunnel_url(port, timeout=self.state.timeouts.fast_op_s)
        except Exception as e:
            if _is_runloop_provider_error(e):
                raise ExposedPortUnavailableError(
                    port=port,
                    exposed_ports=self.state.exposed_ports,
                    reason="backend_unavailable",
                    context=_runloop_error_context(e, backend_detail="get_tunnel_url_failed"),
                    cause=e,
                ) from e
            raise
        if isinstance(url, str) and url:
            return url

        try:
            await self._devbox.net.enable_tunnel(
                auth_mode="open",
                http_keep_alive=True,
                wake_on_http=False,
                timeout=self.state.timeouts.fast_op_s,
            )
        except Exception as e:
            if _is_runloop_provider_error(e):
                raise ExposedPortUnavailableError(
                    port=port,
                    exposed_ports=self.state.exposed_ports,
                    reason="backend_unavailable",
                    context=_runloop_error_context(e, backend_detail="enable_tunnel_failed"),
                    cause=e,
                ) from e
            raise
        try:
            url = await self._devbox.get_tunnel_url(port, timeout=self.state.timeouts.fast_op_s)
        except Exception as e:
            if _is_runloop_provider_error(e):
                context = _runloop_error_context(e, backend_detail="get_tunnel_url_failed")
                context["phase"] = "post_enable"
                raise ExposedPortUnavailableError(
                    port=port,
                    exposed_ports=self.state.exposed_ports,
                    reason="backend_unavailable",
                    context=context,
                    cause=e,
                ) from e
            raise
        if not isinstance(url, str) or not url:
            raise ExposedPortUnavailableError(
                port=port,
                exposed_ports=self.state.exposed_ports,
                reason="backend_unavailable",
                context={"backend": "runloop", "detail": "missing_tunnel_url"},
            )
        return url

    async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
        """Resolve an exposed Runloop port through the provider-managed tunnel endpoint.

        Runloop may not have a tunnel enabled for a devbox yet, so exposed-port resolution can
        trigger tunnel creation before returning the public host, port, and TLS settings.
        """

        return await super().resolve_exposed_port(port)

    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
        try:
            url = await self._ensure_tunnel_url(port)
            split = urlsplit(url)
            host = split.hostname
            if host is None:
                raise ValueError("missing hostname")
            port_value = split.port or (443 if split.scheme == "https" else 80)
            return ExposedPortEndpoint(host=host, port=port_value, tls=split.scheme == "https")
        except ExposedPortUnavailableError:
            raise
        except Exception as e:
            raise ExposedPortUnavailableError(
                port=port,
                exposed_ports=self.state.exposed_ports,
                reason="backend_unavailable",
                context={"backend": "runloop", "detail": "invalid_tunnel_url"},
                cause=e,
            ) from e

    async def read(self, path: Path | str, *, user: str | User | None = None) -> io.IOBase:
        """Read a file via Runloop's binary file API."""
        error_path = posix_path_as_path(coerce_posix_path(path))
        if user is not None:
            await self._check_read_with_exec(path, user=user)

        normalized_path = await self._validate_path_access(path)
        try:
            payload = await self._devbox.file.download(
                path=sandbox_path_str(normalized_path),
                timeout=self.state.timeouts.file_download_s,
            )
            return io.BytesIO(bytes(payload))
        except Exception as e:
            if _is_runloop_not_found(e):
                raise WorkspaceReadNotFoundError(
                    path=error_path,
                    context=_runloop_error_context(e, backend_detail="file_download_failed"),
                    cause=e,
                ) from e
            if _is_runloop_provider_error(e):
                raise WorkspaceArchiveReadError(
                    path=error_path,
                    context=_runloop_error_context(e, backend_detail="file_download_failed"),
                    cause=e,
                ) from e
            raise WorkspaceArchiveReadError(path=error_path, cause=e) from e

    async def write(
        self,
        path: Path | str,
        data: io.IOBase,
        *,
        user: str | User | None = None,
    ) -> None:
        """Write a file through Runloop's upload API using manifest-root workspace paths."""
        error_path = posix_path_as_path(coerce_posix_path(path))
        if user is not None:
            await self._check_write_with_exec(path, user=user)

        payload = data.read()
        if isinstance(payload, str):
            payload = payload.encode("utf-8")
        if not isinstance(payload, bytes | bytearray):
            raise WorkspaceWriteTypeError(path=error_path, actual_type=type(payload).__name__)

        workspace_path = await self._validate_path_access(path, for_write=True)
        await self.mkdir(workspace_path.parent, parents=True)
        try:
            await self._devbox.file.upload(
                path=sandbox_path_str(workspace_path),
                file=bytes(payload),
                timeout=self.state.timeouts.file_upload_s,
            )
        except Exception as e:
            if _is_runloop_provider_error(e):
                raise WorkspaceArchiveWriteError(
                    path=workspace_path,
                    context=_runloop_error_context(e, backend_detail="file_upload_failed"),
                    cause=e,
                ) from e
            raise WorkspaceArchiveWriteError(path=workspace_path, cause=e) from e

    async def running(self) -> bool:
        """Report whether the current Runloop devbox is still in the `running` backend state.

        Resume logic relies on this backend status check before deciding whether a suspended devbox
        can be reused directly or whether snapshot restore must rebuild the workspace elsewhere.
        """
        try:
            info: RunloopDevboxView = await self._devbox.get_info(
                timeout=self.state.timeouts.keepalive_s
            )
            return cast(str, info.status) == "running"
        except Exception:
            return False

    async def mkdir(
        self,
        path: Path | str,
        *,
        parents: bool = False,
        user: str | User | None = None,
    ) -> None:
        """Create directories via raw exec so workspace-root creation does not depend on `cd`."""

        if user is not None:
            path = await self._check_mkdir_with_exec(path, parents=parents, user=user)
        else:
            path = await self._validate_path_access(path, for_write=True)
        cmd = ["mkdir"]
        if parents:
            cmd.append("-p")
        cmd.extend(["--", sandbox_path_str(path)])
        result = await self._run_exec_command(
            shlex.join(cmd),
            command=tuple(cmd),
            timeout=self.state.timeouts.fast_op_s,
        )
        if not result.ok():
            raise WorkspaceArchiveWriteError(
                path=path,
                context={
                    "reason": "mkdir_failed",
                    "exit_code": result.exit_code,
                    "stderr": result.stderr.decode("utf-8", "replace"),
                },
            )

    async def _backup_plain_skip_paths(self, plain_skip: set[Path]) -> bytes | None:
        if not plain_skip:
            return None

        root = sandbox_path_str(self.state.manifest.root)
        root_q = shlex.quote(root)
        checks = "\n".join(
            (
                f"if [ -e {shlex.quote(rel.as_posix())} ]; then "
                f'set -- "$@" {shlex.quote(rel.as_posix())}; fi'
            )
            for rel in sorted(plain_skip, key=lambda p: p.as_posix())
        )
        command = (
            f"cd {root_q}\n"
            "set --\n"
            f"{checks}\n"
            'if [ "$#" -eq 0 ]; then exit 0; fi\n'
            'tar -cf - "$@" | base64 -w0\n'
        )
        result = await self.exec(command, shell=True, timeout=self.state.timeouts.snapshot_s)
        if not result.ok():
            raise WorkspaceArchiveReadError(
                path=self._workspace_root_path(),
                context={
                    "reason": "ephemeral_backup_failed",
                    "exit_code": result.exit_code,
                    "stderr": result.stderr.decode("utf-8", "replace"),
                },
            )
        encoded = result.stdout.decode("utf-8", "replace").strip()
        if not encoded:
            return None
        try:
            return io.BytesIO(base64.b64decode(encoded.encode("utf-8"), validate=True)).read()
        except Exception as e:
            raise WorkspaceArchiveReadError(
                path=self._workspace_root_path(),
                context={"reason": "ephemeral_backup_invalid_base64"},
                cause=e,
            ) from e

    async def _remove_plain_skip_paths(self, plain_skip: set[Path]) -> None:
        if not plain_skip:
            return
        root = self._workspace_root_path()
        command = ["rm", "-rf", "--"] + [(root / rel).as_posix() for rel in sorted(plain_skip)]
        result = await self.exec(*command, shell=False, timeout=self.state.timeouts.cleanup_s)
        if not result.ok():
            raise WorkspaceArchiveReadError(
                path=root,
                context={
                    "reason": "ephemeral_remove_failed",
                    "exit_code": result.exit_code,
                    "stderr": result.stderr.decode("utf-8", "replace"),
                },
            )

    async def _restore_plain_skip_paths(self, backup: bytes | None) -> None:
        if not backup:
            return
        root = self._workspace_root_path()
        temp_path = root / f".sandbox-runloop-restore-{self.state.session_id.hex}.tar"
        await self.write(temp_path, io.BytesIO(backup))
        try:
            result = await self.exec(
                "mkdir",
                "-p",
                root.as_posix(),
                shell=False,
                timeout=self.state.timeouts.cleanup_s,
            )
            if not result.ok():
                raise WorkspaceArchiveReadError(
                    path=root,
                    context={
                        "reason": "ephemeral_restore_mkdir_failed",
                        "exit_code": result.exit_code,
                    },
                )
            result = await self.exec(
                "tar",
                "-xf",
                sandbox_path_str(temp_path),
                "-C",
                root.as_posix(),
                shell=False,
                timeout=self.state.timeouts.snapshot_s,
            )
            if not result.ok():
                raise WorkspaceArchiveReadError(
                    path=root,
                    context={
                        "reason": "ephemeral_restore_failed",
                        "exit_code": result.exit_code,
                        "stderr": result.stderr.decode("utf-8", "replace"),
                    },
                )
        finally:
            try:
                await self.exec("rm", "-f", "--", sandbox_path_str(temp_path), shell=False)
            except Exception:
                pass

    async def persist_workspace(self) -> io.IOBase:
        """Persist the workspace with a native Runloop disk snapshot.

        Before snapshotting, the session temporarily removes ephemeral skip paths and tears down
        ephemeral mounts so the saved disk image contains only durable workspace state, then it
        restores those local-only artifacts afterward.
        """
        root = self._workspace_root_path()
        skip = self._persist_workspace_skip_relpaths()
        mount_targets = self.state.manifest.ephemeral_mount_targets()
        mount_skip_rel_paths: set[Path] = set()
        for _mount_entry, mount_path in mount_targets:
            try:
                mount_skip_rel_paths.add(mount_path.relative_to(root))
            except ValueError:
                continue
        plain_skip = skip - mount_skip_rel_paths

        backup: bytes | None = None
        unmounted_mounts: list[tuple[Mount, Path]] = []
        snapshot_error: WorkspaceArchiveReadError | None = None
        snapshot_id: str | None = None

        try:
            backup = await self._backup_plain_skip_paths(plain_skip)
            await self._remove_plain_skip_paths(plain_skip)

            for mount_entry, mount_path in mount_targets:
                await mount_entry.mount_strategy.teardown_for_snapshot(
                    mount_entry,
                    self,
                    mount_path,
                )
                unmounted_mounts.append((mount_entry, mount_path))

            snapshot: RunloopAsyncSnapshot = await self._devbox.snapshot_disk(
                name=f"sandbox-{self.state.session_id.hex[:12]}",
                metadata={"openai_agents_session_id": self.state.session_id.hex},
                timeout=self.state.timeouts.snapshot_s,
            )
            snapshot_id = snapshot.id
            if not snapshot_id:
                raise WorkspaceArchiveReadError(
                    path=root,
                    context={
                        "reason": "snapshot_unexpected_return",
                        "type": type(snapshot).__name__,
                    },
                )
        except WorkspaceArchiveReadError as e:
            snapshot_error = e
        except Exception as e:
            snapshot_error = WorkspaceArchiveReadError(
                path=root,
                context={"reason": "snapshot_failed"},
                cause=e,
            )
        finally:
            remount_error: WorkspaceArchiveReadError | None = None
            for mount_entry, mount_path in reversed(unmounted_mounts):
                try:
                    await mount_entry.mount_strategy.restore_after_snapshot(
                        mount_entry, self, mount_path
                    )
                except Exception as e:
                    current_error = WorkspaceArchiveReadError(path=root, cause=e)
                    if remount_error is None:
                        remount_error = current_error
                    else:
                        additional = remount_error.context.setdefault(
                            "additional_remount_errors", []
                        )
                        assert isinstance(additional, list)
                        additional.append(
                            {
                                "message": current_error.message,
                                "cause_type": type(e).__name__,
                                "cause": str(e),
                            }
                        )
            try:
                await self._restore_plain_skip_paths(backup)
            except Exception as e:
                restore_error = WorkspaceArchiveReadError(path=root, cause=e)
                if remount_error is None:
                    remount_error = restore_error
                else:
                    additional = remount_error.context.setdefault("additional_restore_errors", [])
                    assert isinstance(additional, list)
                    additional.append(
                        {
                            "message": restore_error.message,
                            "cause_type": type(e).__name__,
                            "cause": str(e),
                        }
                    )

            if remount_error is not None:
                if snapshot_error is not None:
                    remount_error.context["snapshot_error_before_restore_corruption"] = {
                        "message": snapshot_error.message
                    }
                raise remount_error

        if snapshot_error is not None:
            raise snapshot_error

        assert snapshot_id is not None
        return io.BytesIO(_encode_runloop_snapshot_ref(snapshot_id=snapshot_id))

    async def hydrate_workspace(self, data: io.IOBase) -> None:
        """Replace the current devbox from a Runloop snapshot reference or tar archive.

        Runloop restore creates a new devbox from the saved disk snapshot and treats that snapshot
        filesystem as authoritative, including any tools or files that originally came from the
        source blueprint, so restore does not reselect a blueprint. Non-native payloads fall back
        to tar hydration so cross-provider snapshots and file snapshots keep working.
        """
        root = self._workspace_root_path()
        raw = data.read()
        if isinstance(raw, str):
            raw = raw.encode("utf-8")
        if not isinstance(raw, bytes | bytearray):
            raise WorkspaceWriteTypeError(path=root, actual_type=type(raw).__name__)

        snapshot_id = _decode_runloop_snapshot_ref(bytes(raw))
        if snapshot_id is None:
            await self._hydrate_workspace_via_tar(bytes(raw))
            return

        try:
            try:
                await self._devbox.shutdown(timeout=self.state.timeouts.cleanup_s)
            except Exception:
                pass
            envs = await self._resolved_envs()
            create_kwargs = _runloop_create_kwargs(
                blueprint_id=None,
                blueprint_name=None,
                env_vars=envs,
                name=self.state.name,
                user_parameters=self.state.user_parameters,
                launch_parameters=self.state.launch_parameters,
                tunnel=self.state.tunnel,
                gateways=self.state.gateways,
                mcp=self.state.mcp,
                metadata=self.state.metadata,
                secrets=self.state.secret_refs,
            )
            devbox = await self._sdk.devbox.create_from_snapshot(
                snapshot_id,
                timeout=self.state.timeouts.resume_s,
                **create_kwargs,
            )
            self._devbox = devbox
            self.state.devbox_id = devbox.id
        except Exception as e:
            context: dict[str, object] = {
                "reason": "snapshot_restore_failed",
                "snapshot_id": snapshot_id,
            }
            if _is_runloop_provider_error(e):
                context.update(_runloop_error_context(e, backend_detail="snapshot_restore_failed"))
            raise WorkspaceArchiveWriteError(
                path=root,
                context=context,
                cause=e,
            ) from e

    async def _restore_snapshot_into_workspace_on_resume(self) -> None:
        """Restore snapshots on resume, preserving Runloop's native disk-snapshot fast path."""

        root = self._workspace_root_path()
        workspace_archive = await self.state.snapshot.restore(dependencies=self.dependencies)
        try:
            raw = workspace_archive.read()
            if isinstance(raw, str):
                raw = raw.encode("utf-8")
            if not isinstance(raw, bytes | bytearray):
                raise WorkspaceWriteTypeError(path=root, actual_type=type(raw).__name__)

            payload = bytes(raw)
            if _decode_runloop_snapshot_ref(payload) is None:
                # Most providers restore tar snapshots by clearing the workspace first, then
                # extracting into an empty root. Runloop differs only for its native snapshot
                # refs, which already replace the entire devbox disk and therefore should not
                # pre-clear the workspace root on resume.
                await self._clear_workspace_root_on_resume()
            await self.hydrate_workspace(io.BytesIO(payload))
        finally:
            try:
                workspace_archive.close()
            except Exception:
                pass

    async def _hydrate_workspace_via_tar(self, payload: bytes) -> None:
        root = self._workspace_root_path()
        archive_path = root / f".sandbox-runloop-hydrate-{self.state.session_id.hex}.tar"

        try:
            validate_tar_bytes(
                payload,
                allow_external_symlink_targets=False,
            )
        except UnsafeTarMemberError as e:
            raise WorkspaceArchiveWriteError(
                path=root,
                context={
                    "reason": "unsafe_or_invalid_tar",
                    "member": e.member,
                    "detail": str(e),
                },
                cause=e,
            ) from e

        try:
            await self.mkdir(root, parents=True)
            await self.write(archive_path, io.BytesIO(payload))
            result = await self.exec(
                "tar",
                "-C",
                root.as_posix(),
                "-xf",
                archive_path.as_posix(),
                shell=False,
                timeout=self.state.timeouts.snapshot_s,
            )
            if not result.ok():
                raise WorkspaceArchiveWriteError(
                    path=root,
                    context={
                        "reason": "tar_extract_failed",
                        "exit_code": result.exit_code,
                        "stderr": result.stderr.decode("utf-8", errors="replace"),
                    },
                )
        except WorkspaceArchiveWriteError:
            raise
        except Exception as e:
            raise WorkspaceArchiveWriteError(path=root, cause=e) from e
        finally:
            try:
                await self.exec(
                    "rm",
                    "-f",
                    "--",
                    archive_path.as_posix(),
                    shell=False,
                    timeout=self.state.timeouts.cleanup_s,
                )
            except Exception:
                pass


def _runloop_create_kwargs(
    *,
    blueprint_id: str | None,
    blueprint_name: str | None,
    env_vars: dict[str, str] | None,
    name: str | None,
    user_parameters: RunloopUserParameters | None,
    launch_parameters: RunloopLaunchParameters | None,
    tunnel: RunloopTunnelConfig | None,
    gateways: dict[str, RunloopGatewaySpec],
    mcp: dict[str, RunloopMcpSpec],
    metadata: dict[str, str],
    secrets: dict[str, str],
) -> dict[str, object]:
    kwargs: dict[str, object] = {}
    if blueprint_id is not None:
        kwargs["blueprint_id"] = blueprint_id
    if blueprint_name is not None:
        kwargs["blueprint_name"] = blueprint_name
    if env_vars:
        kwargs["environment_variables"] = env_vars
    if name:
        kwargs["name"] = name
    launch_parameters_payload = _runloop_launch_parameters_payload(
        launch_parameters=launch_parameters,
        user_parameters=user_parameters,
    )
    if launch_parameters_payload is not None:
        kwargs["launch_parameters"] = launch_parameters_payload
    if tunnel is not None:
        kwargs["tunnel"] = tunnel.model_dump(mode="json", exclude_none=True)
    if gateways:
        kwargs["gateways"] = {
            key: value.model_dump(mode="json", exclude_none=True) for key, value in gateways.items()
        }
    if mcp:
        kwargs["mcp"] = {
            key: value.model_dump(mode="json", exclude_none=True) for key, value in mcp.items()
        }
    if metadata:
        kwargs["metadata"] = metadata
    if secrets:
        kwargs["secrets"] = secrets
    return kwargs


def _runloop_launch_parameters_payload(
    *,
    launch_parameters: RunloopLaunchParameters | None,
    user_parameters: RunloopUserParameters | None,
) -> dict[str, object] | None:
    payload = (
        launch_parameters.to_dict(mode="json", exclude_none=True, exclude_defaults=True)
        if launch_parameters is not None
        else {}
    )
    if user_parameters is not None:
        payload["user_parameters"] = user_parameters.to_dict(mode="json", exclude_none=True)
    return payload or None


async def _upsert_runloop_managed_secrets(
    sdk: Any,
    *,
    managed_secrets: dict[str, str] | None,
    timeout_s: float,
) -> dict[str, str]:
    if not managed_secrets:
        return {}

    secret_refs: dict[str, str] = {}
    for env_var, secret_value in sorted(managed_secrets.items()):
        try:
            await sdk.secret.create(name=env_var, value=secret_value, timeout=timeout_s)
        except Exception as e:
            if _is_runloop_conflict(e):
                await sdk.secret.update(env_var, value=secret_value, timeout=timeout_s)
            else:
                raise
        secret_refs[env_var] = env_var
    return secret_refs


def _effective_runloop_home(user_parameters: RunloopUserParameters | None) -> PurePosixPath:
    if user_parameters is None:
        return _RUNLOOP_DEFAULT_HOME
    if user_parameters.username == "root" and user_parameters.uid == 0:
        return _RUNLOOP_ROOT_HOME
    return PurePosixPath("/home") / user_parameters.username


def _default_runloop_manifest_root(user_parameters: RunloopUserParameters | None) -> str:
    return str(_effective_runloop_home(user_parameters))


def _validate_runloop_manifest_root(
    manifest: Manifest, *, user_parameters: RunloopUserParameters | None
) -> None:
    root = PurePosixPath(posixpath.normpath(manifest.root))
    runloop_home = _effective_runloop_home(user_parameters)
    try:
        root.relative_to(runloop_home)
    except ValueError as e:
        raise ValueError(
            "RunloopSandboxClient requires manifest.root to be the effective Runloop home "
            f"({runloop_home}) or a subdirectory of it."
        ) from e


class RunloopSandboxClient(BaseSandboxClient[RunloopSandboxClientOptions | None]):
    """Runloop sandbox client managing devbox lifecycle via AsyncRunloopSDK."""

    backend_id = "runloop"
    supports_default_options = True
    _instrumentation: Instrumentation
    _platform: RunloopPlatformClient

    def __init__(
        self,
        *,
        bearer_token: str | None = None,
        base_url: str | None = None,
        instrumentation: Instrumentation | None = None,
        dependencies: Dependencies | None = None,
    ) -> None:
        self._sdk = _import_runloop_sdk().async_sdk(bearer_token=bearer_token, base_url=base_url)
        self._platform = RunloopPlatformClient(self._sdk)
        self._instrumentation = instrumentation or Instrumentation()
        self._dependencies = dependencies

    @property
    def platform(self) -> RunloopPlatformClient:
        return self._platform

    async def create(
        self,
        *,
        snapshot: SnapshotSpec | SnapshotBase | None = None,
        manifest: Manifest | None = None,
        options: RunloopSandboxClientOptions | None,
    ) -> SandboxSession:
        """Create a Runloop devbox and bind it to a manifest rooted under the active home.

        Runloop defaults to the `user` account at `/home/user`, but explicit user parameters can
        switch the active home, including root launch at `/root`. Client creation validates the
        manifest root against that effective home, merges environment variables, and applies any
        configured blueprint selection or user profile when provisioning the devbox. The returned
        session follows the shared sandbox lifecycle and must be started before direct operations.
        """
        resolved_options = options or RunloopSandboxClientOptions()
        if (
            resolved_options.blueprint_id is not None
            and resolved_options.blueprint_name is not None
        ):
            raise ValueError(
                "RunloopSandboxClientOptions cannot set both blueprint_id and blueprint_name"
            )

        user_parameters = _normalize_runloop_user_parameters(resolved_options.user_parameters)
        manifest = manifest or Manifest(root=_default_runloop_manifest_root(user_parameters))
        _validate_runloop_manifest_root(manifest, user_parameters=user_parameters)

        timeouts_in = resolved_options.timeouts
        if isinstance(timeouts_in, RunloopTimeouts):
            timeouts = timeouts_in
        elif timeouts_in is None:
            timeouts = RunloopTimeouts()
        else:
            timeouts = RunloopTimeouts.model_validate(timeouts_in)

        secret_refs = await _upsert_runloop_managed_secrets(
            self._sdk,
            managed_secrets=resolved_options.managed_secrets,
            timeout_s=timeouts.fast_op_s,
        )
        launch_parameters = _normalize_runloop_launch_parameters(resolved_options.launch_parameters)
        tunnel = _normalize_runloop_tunnel_config(resolved_options.tunnel)
        base_envs = dict(resolved_options.env_vars or {})
        manifest_envs = await manifest.environment.resolve()
        envs = {**base_envs, **manifest_envs} or None

        create_kwargs = _runloop_create_kwargs(
            blueprint_id=resolved_options.blueprint_id,
            blueprint_name=resolved_options.blueprint_name,
            env_vars=envs,
            name=resolved_options.name,
            user_parameters=user_parameters,
            launch_parameters=launch_parameters,
            tunnel=tunnel,
            gateways=dict(resolved_options.gateways or {}),
            mcp=dict(resolved_options.mcp or {}),
            metadata=dict(resolved_options.metadata or {}),
            secrets=secret_refs,
        )
        devbox = await self._sdk.devbox.create(timeout=timeouts.create_s, **create_kwargs)

        session_id = uuid.uuid4()
        snapshot_instance = resolve_snapshot(snapshot, str(session_id))
        state = RunloopSandboxSessionState(
            session_id=session_id,
            manifest=manifest,
            snapshot=snapshot_instance,
            devbox_id=devbox.id,
            blueprint_id=resolved_options.blueprint_id,
            blueprint_name=resolved_options.blueprint_name,
            base_env_vars=base_envs,
            pause_on_exit=resolved_options.pause_on_exit,
            name=resolved_options.name,
            timeouts=timeouts,
            exposed_ports=resolved_options.exposed_ports,
            user_parameters=user_parameters,
            launch_parameters=launch_parameters,
            tunnel=tunnel,
            gateways=dict(resolved_options.gateways or {}),
            mcp=dict(resolved_options.mcp or {}),
            metadata=dict(resolved_options.metadata or {}),
            secret_refs=secret_refs,
        )
        inner = RunloopSandboxSession.from_state(state, sdk=self._sdk, devbox=devbox)
        return self._wrap_session(inner, instrumentation=self._instrumentation)

    async def close(self) -> None:
        """Close the shared AsyncRunloopSDK client used for devbox operations."""
        await self._sdk.aclose()

    async def __aenter__(self) -> RunloopSandboxClient:
        return self

    async def __aexit__(self, *_: object) -> None:
        await self.close()

    async def delete(self, session: SandboxSession) -> SandboxSession:
        """Best-effort release the Runloop devbox when callers delete the session."""
        inner = session._inner
        if not isinstance(inner, RunloopSandboxSession):
            raise TypeError("RunloopSandboxClient.delete expects a RunloopSandboxSession")
        try:
            await inner.shutdown()
        except Exception:
            pass
        return session

    async def resume(
        self,
        state: SandboxSessionState,
    ) -> SandboxSession:
        """Resume a persisted Runloop session by reconnecting or reprovisioning a devbox.

        The client first tries to reconnect to the stored devbox id, including after an unclean
        process/client shutdown where the devbox is still running and `shutdown()` was never
        called. If reconnect fails, it creates a fresh devbox with the stored blueprint and
        environment settings.
        """
        if not isinstance(state, RunloopSandboxSessionState):
            raise TypeError("RunloopSandboxClient.resume expects a RunloopSandboxSessionState")

        devbox = None
        reconnected = False
        try:
            devbox = self._sdk.devbox.from_id(state.devbox_id)
            info: RunloopDevboxView = await devbox.get_info(timeout=state.timeouts.keepalive_s)
            status = info.status
            resume_polling_config = _runloop_polling_config(timeout_s=state.timeouts.resume_s)
            if status == "suspended":
                await devbox.resume(timeout=state.timeouts.resume_s)
                await devbox.await_running(polling_config=resume_polling_config)
            elif status == "resuming":
                await devbox.await_running(polling_config=resume_polling_config)
            elif status != "running":
                raise RuntimeError(f"unexpected_status:{status}")
            reconnected = True
        except Exception:
            devbox = None

        if devbox is None:
            manifest_envs = await state.manifest.environment.resolve()
            envs = {**state.base_env_vars, **manifest_envs} or None
            create_kwargs = _runloop_create_kwargs(
                blueprint_id=state.blueprint_id,
                blueprint_name=state.blueprint_name,
                env_vars=envs,
                name=state.name,
                user_parameters=state.user_parameters,
                launch_parameters=state.launch_parameters,
                tunnel=state.tunnel,
                gateways=state.gateways,
                mcp=state.mcp,
                metadata=state.metadata,
                secrets=state.secret_refs,
            )
            devbox = await self._sdk.devbox.create(timeout=state.timeouts.create_s, **create_kwargs)
            state.devbox_id = devbox.id

        inner = RunloopSandboxSession.from_state(state, sdk=self._sdk, devbox=devbox)
        inner._skip_start = state.pause_on_exit and reconnected
        inner._set_start_state_preserved(reconnected, system=reconnected)
        return self._wrap_session(inner, instrumentation=self._instrumentation)

    def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
        return RunloopSandboxSessionState.model_validate(payload)
