import logging
import os
import random
import re
import time
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Dict, Optional
from urllib.parse import quote

import requests
from django.conf import settings

from api.models import AgentComputeSession, MCPServerConfig
from api.sandbox_utils import monotonic_elapsed_ms as _elapsed_ms, normalize_timeout as _normalize_timeout
from api.services.sandbox_compute import (
    SandboxComputeBackend,
    SandboxComputeUnavailable,
    SandboxSessionUpdate,
    _requires_agent_pod_discovery,
)
from api.services.system_settings import (
    get_sandbox_compute_pod_image,
    get_sandbox_compute_require_proxy,
    get_sandbox_egress_proxy_pod_image,
)

logger = logging.getLogger(__name__)

_SERVICE_ACCOUNT_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount")
_SANDBOX_SERVICE_PORT = 8080
_PROXY_POST_MAX_ATTEMPTS = 3
_PROXY_POST_RETRY_BASE_DELAY_SECONDS = 0.25
_PROXY_POST_RETRY_JITTER_SECONDS = 0.1
_TRANSIENT_PROXY_STATUS_CODES = {502, 503, 504}
_DEFAULT_EGRESS_PROXY_HTTP_PORT = 3128
_DEFAULT_EGRESS_PROXY_SOCKS_PORT = 1080
_WORKSPACE_VOLUME_MODE_PVC = "pvc"
_WORKSPACE_VOLUME_MODE_EMPTYDIR = "emptydir"
_QUANTITY_RE = re.compile(r"^([+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)([A-Za-z]{0,2})$")
_DECIMAL_SI_MULTIPLIERS = {
    "": Decimal("1"),
    "n": Decimal("1e-9"),
    "u": Decimal("1e-6"),
    "m": Decimal("1e-3"),
    "k": Decimal("1e3"),
    "M": Decimal("1e6"),
    "G": Decimal("1e9"),
    "T": Decimal("1e12"),
    "P": Decimal("1e15"),
    "E": Decimal("1e18"),
}
_BINARY_SI_MULTIPLIERS = {
    "Ki": Decimal(1024),
    "Mi": Decimal(1024) ** 2,
    "Gi": Decimal(1024) ** 3,
    "Ti": Decimal(1024) ** 4,
    "Pi": Decimal(1024) ** 5,
    "Ei": Decimal(1024) ** 6,
}


def _required_resource_quantity(setting_name: str, value: Any) -> str:
    quantity = str(value or "").strip()
    if not quantity:
        raise SandboxComputeUnavailable(f"{setting_name} is required for kubernetes backend.")
    return quantity


def _optional_resource_quantity(value: Any) -> Optional[str]:
    quantity = str(value or "").strip()
    return quantity or None


def _build_container_resources(
    *,
    cpu_request: Any,
    memory_request: Any,
    cpu_limit: Any,
    memory_limit: Any,
    cpu_request_setting: str,
    memory_request_setting: str,
    cpu_limit_setting: str,
    memory_limit_setting: str,
    ephemeral_storage_request: Any = "",
    ephemeral_storage_limit: Any = "",
) -> Dict[str, Dict[str, str]]:
    requests = {
        "cpu": _required_resource_quantity(cpu_request_setting, cpu_request),
        "memory": _required_resource_quantity(memory_request_setting, memory_request),
    }
    limits = {
        "cpu": _required_resource_quantity(cpu_limit_setting, cpu_limit),
        "memory": _required_resource_quantity(memory_limit_setting, memory_limit),
    }
    ephemeral_request = _optional_resource_quantity(ephemeral_storage_request)
    if ephemeral_request:
        requests["ephemeral-storage"] = ephemeral_request
    ephemeral_limit = _optional_resource_quantity(ephemeral_storage_limit)
    if ephemeral_limit:
        limits["ephemeral-storage"] = ephemeral_limit
    return {
        "requests": requests,
        "limits": limits,
    }


def _normalize_workspace_volume_mode(value: Any) -> str:
    mode = str(value or "").strip().lower()
    if mode in {_WORKSPACE_VOLUME_MODE_EMPTYDIR, "empty-dir", "ephemeral"}:
        return _WORKSPACE_VOLUME_MODE_EMPTYDIR
    if mode in {_WORKSPACE_VOLUME_MODE_PVC, "persistentvolumeclaim", "persistent-volume-claim"}:
        return _WORKSPACE_VOLUME_MODE_PVC
    if not mode:
        return _WORKSPACE_VOLUME_MODE_EMPTYDIR
    raise SandboxComputeUnavailable(
        "SANDBOX_COMPUTE_WORKSPACE_VOLUME_MODE must be 'emptydir' or 'pvc'."
    )


def _is_transient_proxy_request_error(exc: requests.RequestException) -> bool:
    if isinstance(exc, requests.HTTPError):
        status_code = getattr(getattr(exc, "response", None), "status_code", None)
        return status_code in _TRANSIENT_PROXY_STATUS_CODES
    return isinstance(
        exc,
        (
            requests.ConnectionError,
            requests.Timeout,
            requests.exceptions.ChunkedEncodingError,
        ),
    )


class KubernetesApiError(RuntimeError):
    def __init__(self, status_code: int, message: str) -> None:
        super().__init__(message)
        self.status_code = status_code


class KubernetesApiClient:
    def __init__(self, *, base_url: str, token: str, ca_path: Optional[str], timeout: int) -> None:
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.ca_path = ca_path
        self.timeout = timeout

    def request_json(
        self,
        method: str,
        path: str,
        *,
        json_body: Optional[Dict[str, Any]] = None,
        timeout: Optional[int] = None,
        allow_404: bool = False,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Optional[Dict[str, Any]]:
        url = f"{self.base_url}{path}"
        headers = {"Authorization": f"Bearer {self.token}"}
        if extra_headers:
            headers.update(extra_headers)
        try:
            response = requests.request(
                method,
                url,
                json=json_body,
                headers=headers,
                timeout=timeout or self.timeout,
                verify=self.ca_path or True,
            )
        except requests.RequestException as exc:
            raise KubernetesApiError(0, f"Kubernetes API request failed: {exc}") from exc

        if response.status_code == 404 and allow_404:
            return None
        if response.status_code >= 400:
            raise KubernetesApiError(response.status_code, response.text)
        if not response.text:
            return {}
        try:
            return response.json()
        except ValueError as exc:
            raise KubernetesApiError(response.status_code, "Invalid JSON from Kubernetes API") from exc


class KubernetesSandboxBackend(SandboxComputeBackend):
    def custom_tool_workspace_root(self, agent_id: Any) -> str:
        return "/workspace"

    def __init__(self) -> None:
        base_url = _k8s_api_url()
        token = _read_service_account_token()
        if not token:
            raise SandboxComputeUnavailable("Kubernetes service account token not available.")
        ca_path = _service_account_path("ca.crt")
        timeout = int(getattr(settings, "SANDBOX_COMPUTE_K8S_TIMEOUT_SECONDS", 30))
        self._client = KubernetesApiClient(base_url=base_url, token=token, ca_path=ca_path, timeout=timeout)
        self._namespace = _k8s_namespace()
        self._compute_api_token = getattr(settings, "SANDBOX_COMPUTE_API_TOKEN", "") or ""
        self._pod_image = get_sandbox_compute_pod_image()
        self._pod_service_account = getattr(settings, "SANDBOX_COMPUTE_POD_SERVICE_ACCOUNT", "") or ""
        self._pod_runtime_class = getattr(settings, "SANDBOX_COMPUTE_POD_RUNTIME_CLASS", "gvisor")
        self._pod_configmap = getattr(settings, "SANDBOX_COMPUTE_POD_CONFIGMAP_NAME", "gobii-sandbox-common-env")
        self._pod_secret = getattr(settings, "SANDBOX_COMPUTE_POD_SECRET_NAME", "gobii-sandbox-env")
        self._pod_resources = _build_container_resources(
            cpu_request=settings.SANDBOX_COMPUTE_POD_CPU_REQUEST,
            memory_request=settings.SANDBOX_COMPUTE_POD_MEMORY_REQUEST,
            cpu_limit=settings.SANDBOX_COMPUTE_POD_CPU_LIMIT,
            memory_limit=settings.SANDBOX_COMPUTE_POD_MEMORY_LIMIT,
            cpu_request_setting="SANDBOX_COMPUTE_POD_CPU_REQUEST",
            memory_request_setting="SANDBOX_COMPUTE_POD_MEMORY_REQUEST",
            cpu_limit_setting="SANDBOX_COMPUTE_POD_CPU_LIMIT",
            memory_limit_setting="SANDBOX_COMPUTE_POD_MEMORY_LIMIT",
            ephemeral_storage_request=settings.SANDBOX_COMPUTE_POD_EPHEMERAL_STORAGE_REQUEST,
            ephemeral_storage_limit=settings.SANDBOX_COMPUTE_POD_EPHEMERAL_STORAGE_LIMIT,
        )
        self._egress_proxy_image = get_sandbox_egress_proxy_pod_image()
        self._egress_proxy_port = int(
            getattr(settings, "SANDBOX_EGRESS_PROXY_POD_PORT", _DEFAULT_EGRESS_PROXY_HTTP_PORT)
        )
        self._egress_proxy_service_port = int(
            getattr(settings, "SANDBOX_EGRESS_PROXY_SERVICE_PORT", self._egress_proxy_port)
        )
        self._egress_proxy_socks_port = int(
            getattr(settings, "SANDBOX_EGRESS_PROXY_SOCKS_POD_PORT", _DEFAULT_EGRESS_PROXY_SOCKS_PORT)
        )
        self._egress_proxy_socks_service_port = int(
            getattr(settings, "SANDBOX_EGRESS_PROXY_SOCKS_SERVICE_PORT", self._egress_proxy_socks_port)
        )
        self._egress_proxy_runtime_class = getattr(settings, "SANDBOX_EGRESS_PROXY_POD_RUNTIME_CLASS", "") or ""
        self._egress_proxy_service_account = (
            getattr(settings, "SANDBOX_EGRESS_PROXY_POD_SERVICE_ACCOUNT", "") or ""
        )
        self._egress_proxy_resources = _build_container_resources(
            cpu_request=settings.SANDBOX_EGRESS_PROXY_POD_CPU_REQUEST,
            memory_request=settings.SANDBOX_EGRESS_PROXY_POD_MEMORY_REQUEST,
            cpu_limit=settings.SANDBOX_EGRESS_PROXY_POD_CPU_LIMIT,
            memory_limit=settings.SANDBOX_EGRESS_PROXY_POD_MEMORY_LIMIT,
            cpu_request_setting="SANDBOX_EGRESS_PROXY_POD_CPU_REQUEST",
            memory_request_setting="SANDBOX_EGRESS_PROXY_POD_MEMORY_REQUEST",
            cpu_limit_setting="SANDBOX_EGRESS_PROXY_POD_CPU_LIMIT",
            memory_limit_setting="SANDBOX_EGRESS_PROXY_POD_MEMORY_LIMIT",
        )
        self._no_proxy = getattr(settings, "SANDBOX_COMPUTE_NO_PROXY", "") or ""
        self._pod_ready_timeout = int(getattr(settings, "SANDBOX_COMPUTE_POD_READY_TIMEOUT_SECONDS", 300))
        self._service_routable_timeout = int(settings.SANDBOX_COMPUTE_SERVICE_ROUTABLE_TIMEOUT_SECONDS)
        self._pvc_size = getattr(settings, "SANDBOX_COMPUTE_PVC_SIZE", "1Gi")
        self._pvc_storage_class = getattr(settings, "SANDBOX_COMPUTE_PVC_STORAGE_CLASS", "")
        self._workspace_volume_mode = _normalize_workspace_volume_mode(
            getattr(settings, "SANDBOX_COMPUTE_WORKSPACE_VOLUME_MODE", "emptydir")
        )
        self._workspace_emptydir_size = getattr(settings, "SANDBOX_COMPUTE_WORKSPACE_EMPTYDIR_SIZE", "1Gi")
        self._snapshot_class = getattr(settings, "SANDBOX_COMPUTE_SNAPSHOT_CLASS", "")
        self._snapshot_on_idle_stop = bool(getattr(settings, "SANDBOX_COMPUTE_SNAPSHOT_ON_IDLE_STOP", False))
        self._proxy_timeout = int(getattr(settings, "SANDBOX_COMPUTE_HTTP_TIMEOUT_SECONDS", 180))
        self._mcp_timeout = int(getattr(settings, "SANDBOX_COMPUTE_MCP_REQUEST_TIMEOUT_SECONDS", self._proxy_timeout))
        self._tool_timeout = int(getattr(settings, "SANDBOX_COMPUTE_TOOL_REQUEST_TIMEOUT_SECONDS", self._proxy_timeout))
        self._discovery_timeout = int(
            getattr(settings, "SANDBOX_COMPUTE_DISCOVERY_TIMEOUT_SECONDS", self._proxy_timeout)
        )

        if not self._pod_image:
            raise SandboxComputeUnavailable("SANDBOX_COMPUTE_POD_IMAGE is required for kubernetes backend.")
        if not self._compute_api_token:
            raise SandboxComputeUnavailable("SANDBOX_COMPUTE_API_TOKEN is required for kubernetes backend.")

    def _uses_pvc_workspace(self) -> bool:
        return self._workspace_volume_mode == _WORKSPACE_VOLUME_MODE_PVC

    def deploy_or_resume(self, agent, session: AgentComputeSession) -> SandboxSessionUpdate:
        pod_name = _pod_name(agent.id)
        sandbox_service_name = _sandbox_service_name(agent.id)
        pvc_name = _pvc_name(agent.id)
        egress_service_name = None
        no_proxy = None
        workspace_reset = False
        if session.proxy_server:
            if not self._egress_proxy_image:
                raise SandboxComputeUnavailable(
                    "SANDBOX_EGRESS_PROXY_POD_IMAGE is required to use proxy-backed sandbox pods."
                )
            egress_service_name = self._ensure_egress_proxy(agent, session.proxy_server)
            no_proxy = _merge_no_proxy_values(
                self._no_proxy,
                "localhost",
                "127.0.0.1",
                ".svc",
                ".cluster.local",
            )

        snapshot_name = None
        if self._uses_pvc_workspace():
            snapshot_name = session.workspace_snapshot.k8s_snapshot_name if session.workspace_snapshot else None
            if snapshot_name and not _resource_exists(self._client, _snapshot_path(self._namespace, snapshot_name)):
                logger.warning("Snapshot %s not found; provisioning fresh PVC for agent=%s", snapshot_name, agent.id)
                snapshot_name = None
        try:
            if self._uses_pvc_workspace() and not _resource_exists(self._client, _pvc_path(self._namespace, pvc_name)):
                self._create_pvc(pvc_name, agent_id=str(agent.id), snapshot_name=snapshot_name)
                workspace_reset = snapshot_name is None
            if not _resource_exists(self._client, _service_path(self._namespace, sandbox_service_name)):
                self._create_service(sandbox_service_name, agent_id=str(agent.id))

            pod = self._get_pod(pod_name)
            if not pod:
                self._create_pod(
                    pod_name,
                    pvc_name,
                    agent_id=str(agent.id),
                    egress_service_name=egress_service_name,
                    no_proxy=no_proxy,
                )
                workspace_reset = workspace_reset or not self._uses_pvc_workspace()
            else:
                phase = (pod.get("status") or {}).get("phase")
                if not _sandbox_pod_matches(
                    pod,
                    image=self._pod_image,
                    resources=self._pod_resources,
                    workspace_volume_mode=self._workspace_volume_mode,
                    pvc_name=pvc_name,
                    emptydir_size_limit=self._workspace_emptydir_size,
                    egress_service_name=egress_service_name,
                    http_proxy_port=self._egress_proxy_service_port,
                    socks_proxy_port=self._egress_proxy_socks_service_port,
                    no_proxy=no_proxy,
                ):
                    self._delete_pod(pod_name)
                    self._create_pod(
                        pod_name,
                        pvc_name,
                        agent_id=str(agent.id),
                        egress_service_name=egress_service_name,
                        no_proxy=no_proxy,
                    )
                    workspace_reset = workspace_reset or not self._uses_pvc_workspace()
                elif phase not in {"Running", "Pending"}:
                    self._delete_pod(pod_name)
                    self._create_pod(
                        pod_name,
                        pvc_name,
                        agent_id=str(agent.id),
                        egress_service_name=egress_service_name,
                        no_proxy=no_proxy,
                    )
                    workspace_reset = workspace_reset or not self._uses_pvc_workspace()
        except KubernetesApiError as exc:
            raise SandboxComputeUnavailable(f"Kubernetes scheduler failed: {exc}") from exc

        if not self._wait_for_pod_ready(pod_name):
            self._delete_pod(pod_name)
            self._delete_service(sandbox_service_name)
            self._delete_egress_proxy(agent)
            return SandboxSessionUpdate(
                state=AgentComputeSession.State.ERROR,
                pod_name=pod_name,
                namespace=self._namespace,
                workspace_reset=workspace_reset,
            )
        if not self._wait_for_service_routable(
            sandbox_service_name,
            timeout_seconds=self._service_routable_timeout,
        ):
            self._delete_pod(pod_name)
            self._delete_service(sandbox_service_name)
            self._delete_egress_proxy(agent)
            return SandboxSessionUpdate(
                state=AgentComputeSession.State.ERROR,
                pod_name=pod_name,
                namespace=self._namespace,
                workspace_reset=workspace_reset,
            )

        return SandboxSessionUpdate(
            state=AgentComputeSession.State.RUNNING,
            pod_name=pod_name,
            namespace=self._namespace,
            workspace_reset=workspace_reset,
        )

    def run_command(
        self,
        agent,
        session: AgentComputeSession,
        command: str,
        *,
        cwd: Optional[str] = None,
        env: Optional[Dict[str, str]] = None,
        trusted_env_keys: Optional[list[str]] = None,
        timeout: Optional[int] = None,
        interactive: bool = False,
    ) -> Dict[str, Any]:
        if not session.pod_name:
            return {"status": "error", "message": "Sandbox pod not available."}
        timeout_value = _normalize_timeout(
            timeout,
            default=int(getattr(settings, "SANDBOX_COMPUTE_RUN_COMMAND_TIMEOUT_SECONDS", 120)),
        )
        request_timeout = max(self._proxy_timeout, timeout_value + 10)
        payload = {
            "agent_id": str(agent.id),
            "command": command,
            "cwd": cwd,
            "env": env,
            "timeout": timeout_value,
            "interactive": interactive,
        }
        if trusted_env_keys:
            payload["trusted_env_keys"] = [str(key) for key in trusted_env_keys if str(key)]
        return self._proxy_post(_sandbox_service_name(agent.id), "/sandbox/compute/run_command", payload, timeout=request_timeout)

    def mcp_request(
        self,
        agent,
        session: AgentComputeSession,
        server_config_id: str,
        tool_name: str,
        params: Dict[str, Any],
        *,
        full_tool_name: Optional[str] = None,
        server_payload: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if not session.pod_name:
            return {"status": "error", "message": "Sandbox pod not available."}
        payload = {
            "agent_id": str(agent.id),
            "server_id": server_config_id,
            "tool_name": tool_name,
            "params": params,
        }
        if server_payload:
            payload["server"] = server_payload
        timeout_value = getattr(
            self,
            "_mcp_timeout",
            getattr(self, "_proxy_timeout", int(getattr(settings, "SANDBOX_COMPUTE_HTTP_TIMEOUT_SECONDS", 180))),
        )
        return self._proxy_post(
            _sandbox_service_name(agent.id),
            "/sandbox/compute/mcp_request",
            payload,
            timeout=timeout_value,
        )

    def tool_request(
        self,
        agent,
        session: AgentComputeSession,
        tool_name: str,
        params: Dict[str, Any],
    ) -> Dict[str, Any]:
        if not session.pod_name:
            return {"status": "error", "message": "Sandbox pod not available."}
        params_payload = params or {}
        request_timeout = getattr(
            self,
            "_tool_timeout",
            getattr(self, "_proxy_timeout", int(getattr(settings, "SANDBOX_COMPUTE_HTTP_TIMEOUT_SECONDS", 180))),
        )
        if tool_name == "python_exec":
            normalized = _normalize_timeout(
                params_payload.get("timeout_seconds"),
                default=int(getattr(settings, "SANDBOX_COMPUTE_PYTHON_DEFAULT_TIMEOUT_SECONDS", 30)),
                maximum=int(getattr(settings, "SANDBOX_COMPUTE_PYTHON_MAX_TIMEOUT_SECONDS", 120)),
            )
            params_payload = dict(params_payload)
            params_payload["timeout_seconds"] = normalized
            request_timeout = max(request_timeout, normalized + 10)
        payload = {
            "agent_id": str(agent.id),
            "tool_name": tool_name,
            "params": params_payload,
        }
        return self._proxy_post(
            _sandbox_service_name(agent.id),
            "/sandbox/compute/tool_request",
            payload,
            timeout=request_timeout,
        )

    def sync_filespace(
        self,
        agent,
        session: AgentComputeSession,
        *,
        direction: str,
        payload: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if not session.pod_name:
            return {"status": "error", "message": "Sandbox pod not available."}
        body = payload or {}
        body.update({"agent_id": str(agent.id), "direction": direction})
        return self._proxy_post(_sandbox_service_name(agent.id), "/sandbox/compute/sync_filespace", body)

    def snapshot_workspace(self, agent, session: AgentComputeSession, *, reason: str) -> Dict[str, Any]:
        if not self._uses_pvc_workspace():
            return {
                "status": "skipped",
                "message": "Workspace uses emptyDir and is disposable after object-store sync.",
                "workspace_disposable": True,
            }
        if not self._snapshot_on_idle_stop:
            return {
                "status": "skipped",
                "message": "Workspace snapshotting disabled; PVC is disposable after object-store sync.",
                "workspace_disposable": True,
            }
        pvc_name = _pvc_name(agent.id)
        if not _resource_exists(self._client, _pvc_path(self._namespace, pvc_name)):
            return {"status": "error", "message": "Workspace PVC not found."}
        snapshot_name = _snapshot_name(agent.id)
        body = {
            "apiVersion": "snapshot.storage.k8s.io/v1",
            "kind": "VolumeSnapshot",
            "metadata": {
                "name": snapshot_name,
                "labels": {
                    "app": "sandbox-compute",
                    "agent_id": str(agent.id),
                },
            },
            "spec": {
                "source": {"persistentVolumeClaimName": pvc_name},
            },
        }
        if self._snapshot_class:
            body["spec"]["volumeSnapshotClassName"] = self._snapshot_class

        try:
            self._client.request_json("POST", _snapshot_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            return {"status": "error", "message": f"Snapshot create failed: {exc}"}

        ready = self._wait_for_snapshot_ready(snapshot_name)
        if not ready:
            return {"status": "error", "message": "Snapshot did not become ready in time."}

        return {
            "status": "ok",
            "snapshot_name": snapshot_name,
        }

    def terminate(
        self,
        agent,
        session: AgentComputeSession,
        *,
        reason: str,
        delete_workspace: bool = False,
    ) -> SandboxSessionUpdate:
        pod_name = session.pod_name or _pod_name(agent.id)
        self._delete_pod(pod_name)
        self._delete_service(_sandbox_service_name(agent.id))
        self._delete_egress_proxy_for_agent_id(agent.id)
        if delete_workspace and self._uses_pvc_workspace():
            pvc_name = _pvc_name(agent.id)
            self._delete_pvc(pvc_name)
        return SandboxSessionUpdate(
            state=AgentComputeSession.State.STOPPED,
            pod_name=pod_name,
            namespace=self._namespace,
            workspace_reset=(delete_workspace or not self._uses_pvc_workspace()),
        )

    def delete_agent_resources(
        self,
        agent_id: Any,
        *,
        delete_workspace: bool,
        delete_snapshots: bool,
        dry_run: bool = False,
    ) -> Dict[str, Any]:
        agent_id_text = str(agent_id)
        actions: list[Dict[str, Any]] = []

        def _record(kind: str, name: str) -> None:
            actions.append({"kind": kind, "name": name, "deleted": not dry_run})

        pod_name = _pod_name(agent_id_text)
        service_name = _sandbox_service_name(agent_id_text)
        egress_pod_name = _egress_proxy_pod_name(agent_id_text)
        egress_service_name = _egress_proxy_service_name(agent_id_text)

        _record("Pod", pod_name)
        _record("Service", service_name)
        _record("Pod", egress_pod_name)
        _record("Service", egress_service_name)
        if not dry_run:
            self._delete_pod(pod_name)
            self._delete_service(service_name)
            self._delete_pod(egress_pod_name)
            self._delete_service(egress_service_name)

        if delete_workspace and self._uses_pvc_workspace():
            pvc_name = _pvc_name(agent_id_text)
            _record("PersistentVolumeClaim", pvc_name)
            if not dry_run:
                self._delete_pvc(pvc_name)

        if delete_snapshots and self._uses_pvc_workspace():
            for snapshot_name in self._snapshot_names_for_agent(agent_id_text):
                _record("VolumeSnapshot", snapshot_name)
                if not dry_run:
                    self._delete_snapshot(snapshot_name)

        return {"status": "ok", "agent_id": agent_id_text, "actions": actions}

    def discover_mcp_tools(
        self,
        server_config_id: str,
        *,
        reason: str,
        agent=None,
        session: Optional[AgentComputeSession] = None,
        server_payload: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        if not server_payload:
            return {"status": "error", "message": "Missing MCP server payload for discovery."}
        if not _requires_agent_pod_discovery(server_payload):
            from api.agent.tools.mcp_manager import get_mcp_manager

            manager = get_mcp_manager()
            ok = manager.discover_tools_for_server(server_config_id, agent=agent)
            return {
                "status": "ok" if ok else "error",
                "reason": reason,
                "message": "Agent pod discovery skipped for non-sandboxed MCP server.",
            }

        if agent is None or session is None:
            return {"status": "error", "message": "Sandboxed stdio discovery requires an agent session."}

        service_name = _sandbox_service_name(agent.id)
        payload = {
            "agent_id": str(agent.id),
            "server_id": server_config_id,
            "reason": reason,
            "server": server_payload,
        }

        return self._proxy_post(
            service_name,
            "/sandbox/compute/discover_mcp_tools",
            payload,
            timeout=getattr(
                self,
                "_discovery_timeout",
                getattr(
                    self,
                    "_proxy_timeout",
                    int(getattr(settings, "SANDBOX_COMPUTE_HTTP_TIMEOUT_SECONDS", 180)),
                ),
            ),
        )

    def _proxy_post(
        self,
        service_name: str,
        path: str,
        payload: Dict[str, Any],
        *,
        timeout: Optional[int] = None,
    ) -> Dict[str, Any]:
        request_timeout = timeout or self._proxy_timeout
        url = _sandbox_service_url(self._namespace, service_name, path)
        with requests.Session() as session:
            session.trust_env = False
            for attempt in range(1, _PROXY_POST_MAX_ATTEMPTS + 1):
                try:
                    response = session.post(
                        url,
                        json=payload,
                        timeout=request_timeout,
                        headers={"X-Sandbox-Compute-Token": self._compute_api_token},
                    )
                    response.raise_for_status()
                    break
                except requests.RequestException as exc:
                    retryable = _is_transient_proxy_request_error(exc)
                    if not retryable or attempt >= _PROXY_POST_MAX_ATTEMPTS:
                        logger.warning(
                            "Sandbox proxy POST failed service=%s path=%s attempt=%s/%s retryable=%s error=%s",
                            service_name,
                            path,
                            attempt,
                            _PROXY_POST_MAX_ATTEMPTS,
                            retryable,
                            exc,
                        )
                        return {"status": "error", "message": f"Sandbox proxy request failed: {exc}"}
                    delay = (
                        _PROXY_POST_RETRY_BASE_DELAY_SECONDS * attempt
                        + random.uniform(0, _PROXY_POST_RETRY_JITTER_SECONDS)
                    )
                    logger.info(
                        "Retrying sandbox proxy POST service=%s path=%s attempt=%s/%s delay_seconds=%.3f error=%s",
                        service_name,
                        path,
                        attempt,
                        _PROXY_POST_MAX_ATTEMPTS,
                        delay,
                        exc,
                    )
                    time.sleep(delay)
        if not response.text:
            return {"status": "error", "message": "Sandbox proxy returned empty response."}
        try:
            return response.json()
        except ValueError:
            return {"status": "error", "message": "Sandbox proxy returned invalid JSON."}

    def _get_pod(self, pod_name: str) -> Optional[Dict[str, Any]]:
        try:
            return self._client.request_json("GET", _pod_path(self._namespace, pod_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to fetch pod %s: %s", pod_name, exc)
            return None

    def _get_service(self, service_name: str) -> Optional[Dict[str, Any]]:
        try:
            return self._client.request_json(
                "GET",
                _service_path(self._namespace, service_name),
                allow_404=True,
            )
        except KubernetesApiError as exc:
            logger.warning("Failed to fetch service %s: %s", service_name, exc)
            return None

    def _ensure_egress_proxy(self, agent, proxy_server) -> str:
        host = str(getattr(proxy_server, "host", "") or "").strip()
        port = getattr(proxy_server, "port", None)
        if not host or not port:
            raise SandboxComputeUnavailable("Proxy server is missing host/port metadata.")
        _upstream_proxy_scheme(proxy_server)

        pod_name = _egress_proxy_pod_name(agent.id)
        service_name = _egress_proxy_service_name(agent.id)

        try:
            service = self._get_service(service_name)
            if not service:
                self._create_egress_proxy_service(service_name, agent_id=str(agent.id))

            pod = self._get_pod(pod_name)
            if not pod:
                self._create_egress_proxy_pod(pod_name, agent_id=str(agent.id), proxy_server=proxy_server)
            else:
                phase = (pod.get("status") or {}).get("phase")
                if not _egress_proxy_pod_matches(
                    pod,
                    proxy_server=proxy_server,
                    resources=self._egress_proxy_resources,
                    http_listen_port=self._egress_proxy_port,
                    socks_listen_port=self._egress_proxy_socks_port,
                ):
                    self._delete_pod(pod_name)
                    self._create_egress_proxy_pod(pod_name, agent_id=str(agent.id), proxy_server=proxy_server)
                elif phase not in {"Running", "Pending"}:
                    self._delete_pod(pod_name)
                    self._create_egress_proxy_pod(pod_name, agent_id=str(agent.id), proxy_server=proxy_server)
        except KubernetesApiError as exc:
            raise SandboxComputeUnavailable(f"Egress proxy provisioning failed: {exc}") from exc

        if not self._wait_for_pod_ready(pod_name):
            raise SandboxComputeUnavailable("Egress proxy pod did not become ready in time.")

        return service_name

    def _create_pod(
        self,
        pod_name: str,
        pvc_name: str,
        *,
        agent_id: str,
        egress_service_name: Optional[str],
        no_proxy: Optional[str],
    ) -> None:
        body = _build_pod_manifest(
            pod_name=pod_name,
            pvc_name=pvc_name,
            namespace=self._namespace,
            image=self._pod_image,
            runtime_class=self._pod_runtime_class,
            service_account=self._pod_service_account,
            configmap_name=self._pod_configmap,
            secret_name=self._pod_secret,
            agent_id=agent_id,
            resources=self._pod_resources,
            workspace_volume_mode=self._workspace_volume_mode,
            workspace_emptydir_size_limit=self._workspace_emptydir_size,
            egress_service_name=egress_service_name,
            http_proxy_port=self._egress_proxy_service_port,
            socks_proxy_port=self._egress_proxy_socks_service_port,
            no_proxy=no_proxy,
        )
        try:
            self._client.request_json("POST", _pod_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            if exc.status_code != 409:
                raise

    def _create_egress_proxy_pod(self, pod_name: str, *, agent_id: str, proxy_server) -> None:
        body = _build_egress_proxy_pod_manifest(
            pod_name=pod_name,
            namespace=self._namespace,
            image=self._egress_proxy_image,
            runtime_class=self._egress_proxy_runtime_class or None,
            service_account=self._egress_proxy_service_account or None,
            agent_id=agent_id,
            proxy_server=proxy_server,
            resources=self._egress_proxy_resources,
            http_listen_port=self._egress_proxy_port,
            socks_listen_port=self._egress_proxy_socks_port,
        )
        try:
            self._client.request_json("POST", _pod_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            if exc.status_code != 409:
                raise

    def _create_egress_proxy_service(self, service_name: str, *, agent_id: str) -> None:
        body = _build_egress_proxy_service_manifest(
            service_name=service_name,
            namespace=self._namespace,
            agent_id=agent_id,
            http_port=self._egress_proxy_service_port,
            http_target_port=self._egress_proxy_port,
            socks_port=self._egress_proxy_socks_service_port,
            socks_target_port=self._egress_proxy_socks_port,
        )
        try:
            self._client.request_json("POST", _service_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            if exc.status_code != 409:
                raise

    def _create_service(self, service_name: str, *, agent_id: str) -> None:
        body = _build_sandbox_service_manifest(
            service_name=service_name,
            namespace=self._namespace,
            agent_id=agent_id,
            port=_SANDBOX_SERVICE_PORT,
            target_port=_SANDBOX_SERVICE_PORT,
        )
        try:
            self._client.request_json("POST", _service_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            if exc.status_code != 409:
                raise

    def _delete_pod(self, pod_name: str) -> None:
        try:
            self._client.request_json("DELETE", _pod_path(self._namespace, pod_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to delete pod %s: %s", pod_name, exc)

    def _create_pvc(self, pvc_name: str, *, agent_id: str, snapshot_name: Optional[str]) -> None:
        body = _build_pvc_manifest(
            pvc_name=pvc_name,
            namespace=self._namespace,
            agent_id=agent_id,
            size=self._pvc_size,
            storage_class=self._pvc_storage_class,
            snapshot_name=snapshot_name,
        )
        try:
            self._client.request_json("POST", _pvc_collection_path(self._namespace), json_body=body)
        except KubernetesApiError as exc:
            if exc.status_code != 409:
                raise

    def _delete_pvc(self, pvc_name: str) -> None:
        try:
            self._client.request_json("DELETE", _pvc_path(self._namespace, pvc_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to delete PVC %s: %s", pvc_name, exc)

    def _delete_service(self, service_name: str) -> None:
        try:
            self._client.request_json("DELETE", _service_path(self._namespace, service_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to delete service %s: %s", service_name, exc)

    def _delete_egress_proxy(self, agent) -> None:
        self._delete_egress_proxy_for_agent_id(agent.id)

    def _delete_egress_proxy_for_agent_id(self, agent_id: Any) -> None:
        pod_name = _egress_proxy_pod_name(agent_id)
        service_name = _egress_proxy_service_name(agent_id)
        self._delete_pod(pod_name)
        try:
            self._client.request_json("DELETE", _service_path(self._namespace, service_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to delete egress proxy service %s: %s", service_name, exc)

    def _snapshot_names_for_agent(self, agent_id: str) -> list[str]:
        selector = f"app=sandbox-compute,agent_id={agent_id}"
        try:
            response = self._client.request_json(
                "GET",
                _with_label_selector(_snapshot_collection_path(self._namespace), selector),
            )
        except KubernetesApiError as exc:
            logger.warning("Failed to list snapshots for agent=%s: %s", agent_id, exc)
            return []
        return [
            str((item.get("metadata") or {}).get("name"))
            for item in (response or {}).get("items", [])
            if isinstance(item, dict) and (item.get("metadata") or {}).get("name")
        ]

    def _delete_snapshot(self, snapshot_name: str) -> None:
        try:
            self._client.request_json("DELETE", _snapshot_path(self._namespace, snapshot_name), allow_404=True)
        except KubernetesApiError as exc:
            logger.warning("Failed to delete snapshot %s: %s", snapshot_name, exc)

    def _wait_for_pod_ready(self, pod_name: str) -> bool:
        started_at = time.monotonic()
        deadline = started_at + max(float(self._pod_ready_timeout), 0.0)
        attempts = 0
        last_phase = None
        last_summary = ""
        while time.monotonic() < deadline:
            attempts += 1
            pod = self._get_pod(pod_name)
            if not pod:
                time.sleep(2)
                continue
            summary = _pod_readiness_summary(pod)
            status = pod.get("status") or {}
            phase = status.get("phase")
            if phase != last_phase or summary != last_summary:
                logger.info(
                    "Sandbox pod readiness progress pod=%s phase=%s attempts=%s elapsed_ms=%s summary=%s",
                    pod_name,
                    phase,
                    attempts,
                    _elapsed_ms(started_at),
                    summary,
                )
                last_phase = phase
                last_summary = summary
            if phase == "Running":
                for condition in status.get("conditions", []):
                    if condition.get("type") == "Ready" and condition.get("status") == "True":
                        logger.info(
                            "Sandbox pod ready pod=%s attempts=%s elapsed_ms=%s",
                            pod_name,
                            attempts,
                            _elapsed_ms(started_at),
                        )
                        return True
            time.sleep(min(2.0, max(deadline - time.monotonic(), 0.0)))
        logger.warning(
            "Sandbox pod readiness timeout pod=%s attempts=%s elapsed_ms=%s timeout_seconds=%s summary=%s",
            pod_name,
            attempts,
            _elapsed_ms(started_at),
            self._pod_ready_timeout,
            last_summary or f"phase={last_phase or 'unknown'}",
        )
        return False

    def _wait_for_service_routable(self, service_name: str, *, timeout_seconds: float) -> bool:
        started_at = time.monotonic()
        deadline = started_at + max(float(timeout_seconds), 0.0)
        url = _sandbox_service_url(self._namespace, service_name, "/healthz")
        attempts = 0
        last_error = None

        with requests.Session() as session:
            session.trust_env = False
            while time.monotonic() < deadline:
                attempts += 1
                remaining = max(deadline - time.monotonic(), 0.1)
                try:
                    response = session.get(url, timeout=min(5.0, remaining))
                    response.raise_for_status()
                    logger.info(
                        "Sandbox service routable service=%s attempts=%s elapsed_ms=%s",
                        service_name,
                        attempts,
                        _elapsed_ms(started_at),
                    )
                    return True
                except requests.RequestException as exc:
                    last_error = exc
                    if attempts == 1 or attempts % 5 == 0:
                        logger.info(
                            "Sandbox service not routable yet service=%s attempts=%s elapsed_ms=%s error=%s",
                            service_name,
                            attempts,
                            _elapsed_ms(started_at),
                            exc,
                        )
                    time.sleep(min(1.0, remaining))

        logger.warning(
            "Sandbox service routability timeout service=%s attempts=%s elapsed_ms=%s timeout_seconds=%s last_error=%s",
            service_name,
            attempts,
            _elapsed_ms(started_at),
            timeout_seconds,
            last_error,
        )
        return False

    def _wait_for_snapshot_ready(self, snapshot_name: str) -> bool:
        deadline = time.time() + int(getattr(settings, "SANDBOX_COMPUTE_SNAPSHOT_TIMEOUT_SECONDS", 60))
        while time.time() < deadline:
            try:
                snapshot = self._client.request_json(
                    "GET",
                    _snapshot_path(self._namespace, snapshot_name),
                    allow_404=True,
                )
            except KubernetesApiError as exc:
                logger.warning("Snapshot status check failed: %s", exc)
                time.sleep(2)
                continue
            if snapshot and (snapshot.get("status") or {}).get("readyToUse") is True:
                return True
            time.sleep(2)
        return False


def _pod_readiness_summary(pod: Dict[str, Any]) -> str:
    status = pod.get("status") or {}
    parts = [f"phase={status.get('phase') or 'unknown'}"]

    conditions = []
    for condition in status.get("conditions") or []:
        condition_type = condition.get("type")
        condition_status = condition.get("status")
        if not condition_type or condition_status == "True":
            continue
        reason = condition.get("reason") or ""
        message = condition.get("message") or ""
        detail = f"{condition_type}={condition_status}"
        if reason:
            detail += f"/{reason}"
        if message:
            detail += f":{message[:160]}"
        conditions.append(detail)
    if conditions:
        parts.append("conditions=" + ";".join(conditions))

    container_states = []
    for container in status.get("containerStatuses") or []:
        name = container.get("name") or "container"
        state = container.get("state") or {}
        waiting = state.get("waiting")
        terminated = state.get("terminated")
        if waiting:
            reason = waiting.get("reason") or "waiting"
            message = waiting.get("message") or ""
            detail = f"{name}=waiting/{reason}"
            if message:
                detail += f":{message[:160]}"
            container_states.append(detail)
        elif terminated:
            reason = terminated.get("reason") or "terminated"
            exit_code = terminated.get("exitCode")
            detail = f"{name}=terminated/{reason}"
            if exit_code is not None:
                detail += f":exit={exit_code}"
            container_states.append(detail)
        elif not container.get("ready"):
            container_states.append(f"{name}=not_ready")
    if container_states:
        parts.append("containers=" + ";".join(container_states))

    return " ".join(parts)


def _read_service_account_token() -> str:
    path = _service_account_path("token")
    if not path:
        return ""
    try:
        return Path(path).read_text().strip()
    except OSError:
        return ""


def _service_account_path(filename: str) -> Optional[str]:
    candidate = _SERVICE_ACCOUNT_DIR / filename
    if candidate.exists():
        return str(candidate)
    return None


def _k8s_api_url() -> str:
    explicit = getattr(settings, "SANDBOX_COMPUTE_K8S_API_URL", "") or os.environ.get("SANDBOX_COMPUTE_K8S_API_URL")
    if explicit:
        return explicit.rstrip("/")
    host = os.environ.get("KUBERNETES_SERVICE_HOST")
    port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
    if not host:
        raise SandboxComputeUnavailable("Kubernetes service host not configured.")
    return f"https://{host}:{port}"


def _k8s_namespace() -> str:
    explicit = getattr(settings, "SANDBOX_COMPUTE_K8S_NAMESPACE", "") or os.environ.get("SANDBOX_COMPUTE_K8S_NAMESPACE")
    if explicit:
        return explicit
    path = _service_account_path("namespace")
    if path:
        try:
            return Path(path).read_text().strip()
        except OSError:
            pass
    return "default"


def _resource_exists(client: KubernetesApiClient, path: str) -> bool:
    try:
        return client.request_json("GET", path, allow_404=True) is not None
    except KubernetesApiError:
        return False


def _slugify(value: str) -> str:
    cleaned = re.sub(r"[^a-z0-9-]", "-", value.lower())
    cleaned = re.sub(r"-+", "-", cleaned).strip("-")
    return cleaned


def _pod_name(agent_id: Any) -> str:
    return _slugify(f"sandbox-agent-{agent_id}")


def _sandbox_service_name(agent_id: Any) -> str:
    return _pod_name(agent_id)


def _pvc_name(agent_id: Any) -> str:
    return _slugify(f"sandbox-workspace-{agent_id}")


def _egress_proxy_pod_name(agent_id: Any) -> str:
    return _slugify(f"sandbox-egress-{agent_id}")


def _egress_proxy_service_name(agent_id: Any) -> str:
    return _slugify(f"sandbox-egress-{agent_id}")


def _snapshot_name(agent_id: Any) -> str:
    stamp = time.strftime("%Y%m%d%H%M%S", time.gmtime())
    return _slugify(f"sandbox-snap-{str(agent_id)[:8]}-{stamp}")


def _pod_collection_path(namespace: str) -> str:
    return f"/api/v1/namespaces/{namespace}/pods"


def _pod_path(namespace: str, pod_name: str) -> str:
    return f"/api/v1/namespaces/{namespace}/pods/{pod_name}"


def _pod_proxy_path(namespace: str, pod_name: str, path: str) -> str:
    suffix = path if path.startswith("/") else f"/{path}"
    return f"/api/v1/namespaces/{namespace}/pods/{pod_name}/proxy{suffix}"


def _pvc_collection_path(namespace: str) -> str:
    return f"/api/v1/namespaces/{namespace}/persistentvolumeclaims"


def _pvc_path(namespace: str, pvc_name: str) -> str:
    return f"/api/v1/namespaces/{namespace}/persistentvolumeclaims/{pvc_name}"


def _service_collection_path(namespace: str) -> str:
    return f"/api/v1/namespaces/{namespace}/services"


def _service_path(namespace: str, service_name: str) -> str:
    return f"/api/v1/namespaces/{namespace}/services/{service_name}"


def _sandbox_service_url(namespace: str, service_name: str, path: str) -> str:
    suffix = path if path.startswith("/") else f"/{path}"
    return f"http://{service_name}.{namespace}.svc.cluster.local:{_SANDBOX_SERVICE_PORT}{suffix}"


def _snapshot_collection_path(namespace: str) -> str:
    return f"/apis/snapshot.storage.k8s.io/v1/namespaces/{namespace}/volumesnapshots"


def _snapshot_path(namespace: str, snapshot_name: str) -> str:
    return f"/apis/snapshot.storage.k8s.io/v1/namespaces/{namespace}/volumesnapshots/{snapshot_name}"


def _with_label_selector(path: str, selector: str) -> str:
    return f"{path}?labelSelector={quote(selector, safe='')}"


def _build_pvc_manifest(
    *,
    pvc_name: str,
    namespace: str,
    agent_id: str,
    size: str,
    storage_class: str,
    snapshot_name: Optional[str],
) -> Dict[str, Any]:
    spec: Dict[str, Any] = {
        "accessModes": ["ReadWriteOnce"],
        "resources": {"requests": {"storage": size}},
    }
    if storage_class:
        spec["storageClassName"] = storage_class
    if snapshot_name:
        spec["dataSource"] = {
            "name": snapshot_name,
            "kind": "VolumeSnapshot",
            "apiGroup": "snapshot.storage.k8s.io",
        }
    return {
        "apiVersion": "v1",
        "kind": "PersistentVolumeClaim",
        "metadata": {
            "name": pvc_name,
            "namespace": namespace,
            "labels": {
                "app": "sandbox-compute",
                "component": "sandbox-workspace",
                "agent_id": agent_id,
            },
        },
        "spec": spec,
    }


def _build_pod_manifest(
    *,
    pod_name: str,
    pvc_name: str,
    namespace: str,
    image: str,
    runtime_class: str,
    service_account: str,
    configmap_name: str,
    secret_name: str,
    agent_id: str,
    resources: Dict[str, Dict[str, str]],
    workspace_volume_mode: str = _WORKSPACE_VOLUME_MODE_EMPTYDIR,
    workspace_emptydir_size_limit: str = "",
    egress_service_name: Optional[str] = None,
    http_proxy_port: int = _DEFAULT_EGRESS_PROXY_HTTP_PORT,
    socks_proxy_port: int = _DEFAULT_EGRESS_PROXY_SOCKS_PORT,
    no_proxy: Optional[str] = None,
) -> Dict[str, Any]:
    workspace_volume_mode = _normalize_workspace_volume_mode(workspace_volume_mode)
    env = [
        {"name": "SANDBOX_RUNTIME_CACHE_ROOT", "value": "/runtime-cache"},
        {"name": "SANDBOX_AGENT_WORKSPACE_LAYOUT", "value": "isolated"},
    ]
    env.extend(
        _build_proxy_env(
            egress_service_name=egress_service_name,
            http_proxy_port=http_proxy_port,
            socks_proxy_port=socks_proxy_port,
            no_proxy=no_proxy,
        )
    )

    container: Dict[str, Any] = {
        "name": "sandbox-supervisor",
        "image": image,
        "imagePullPolicy": "IfNotPresent",
        "ports": [{"containerPort": 8080}],
        "envFrom": [
            {"secretRef": {"name": secret_name}},
            {"configMapRef": {"name": configmap_name}},
        ],
        "securityContext": {
            "allowPrivilegeEscalation": False,
            "runAsNonRoot": True,
            "runAsUser": 1000,
            "runAsGroup": 1000,
            "capabilities": {"drop": ["ALL"]},
        },
        "resources": {
            "requests": dict(resources["requests"]),
            "limits": dict(resources["limits"]),
        },
        "volumeMounts": [
            {"name": "workspace", "mountPath": "/workspace"},
            {"name": "runtime-cache", "mountPath": "/runtime-cache"},
        ],
        "readinessProbe": {
            "httpGet": {"path": "/healthz", "port": 8080},
            "initialDelaySeconds": 10,
            "periodSeconds": 10,
            "failureThreshold": 3,
        },
    }
    container["env"] = env

    manifest = {
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {
            "name": pod_name,
            "namespace": namespace,
            "labels": {
                "app": "sandbox-compute",
                "component": "sandbox-agent",
                "agent_id": agent_id,
                "workspace_volume_mode": workspace_volume_mode,
            },
        },
        "spec": {
            "automountServiceAccountToken": False,
            "runtimeClassName": runtime_class,
            "terminationGracePeriodSeconds": 300,
            "securityContext": {
                "fsGroup": 1000,
                "fsGroupChangePolicy": "OnRootMismatch",
                "seccompProfile": {"type": "RuntimeDefault"},
            },
            "containers": [container],
            "volumes": [
                _build_workspace_volume(
                    mode=workspace_volume_mode,
                    pvc_name=pvc_name,
                    emptydir_size_limit=workspace_emptydir_size_limit,
                ),
                {"name": "runtime-cache", "emptyDir": {}},
            ],
        },
    }
    if service_account:
        manifest["spec"]["serviceAccountName"] = service_account
    return manifest


def _build_workspace_volume(*, mode: str, pvc_name: str, emptydir_size_limit: str) -> Dict[str, Any]:
    if mode == _WORKSPACE_VOLUME_MODE_PVC:
        return {
            "name": "workspace",
            "persistentVolumeClaim": {"claimName": pvc_name},
        }
    empty_dir: Dict[str, Any] = {}
    size_limit = str(emptydir_size_limit or "").strip()
    if size_limit:
        empty_dir["sizeLimit"] = size_limit
    return {
        "name": "workspace",
        "emptyDir": empty_dir,
    }


def _build_proxy_env(
    *,
    egress_service_name: Optional[str],
    http_proxy_port: int,
    socks_proxy_port: int,
    no_proxy: Optional[str],
) -> list[Dict[str, str]]:
    return [
        {"name": key, "value": value}
        for key, value in _sandbox_sidecar_proxy_env_values(
            egress_service_name=egress_service_name,
            http_proxy_port=http_proxy_port,
            socks_proxy_port=socks_proxy_port,
            no_proxy=no_proxy,
        ).items()
    ]


def _sandbox_sidecar_proxy_env_values(
    *,
    egress_service_name: Optional[str],
    http_proxy_port: int,
    socks_proxy_port: int,
    no_proxy: Optional[str],
) -> Dict[str, str]:
    http_proxy_url = None
    socks_proxy_url = None
    if egress_service_name:
        http_proxy_url = f"http://{egress_service_name}:{http_proxy_port}"
        socks_proxy_url = f"socks5://{egress_service_name}:{socks_proxy_port}"
    env: Dict[str, str] = {}
    if http_proxy_url:
        for key in ("HTTP_PROXY", "HTTPS_PROXY", "FTP_PROXY", "http_proxy", "https_proxy", "ftp_proxy"):
            env[key] = http_proxy_url
    if socks_proxy_url:
        for key in ("ALL_PROXY", "all_proxy"):
            env[key] = socks_proxy_url
    no_proxy_value = str(no_proxy or "").strip()
    if no_proxy_value:
        env["NO_PROXY"] = no_proxy_value
        env["no_proxy"] = no_proxy_value
    return env


def _build_egress_proxy_pod_manifest(
    *,
    pod_name: str,
    namespace: str,
    image: str,
    runtime_class: Optional[str],
    service_account: Optional[str],
    agent_id: str,
    proxy_server: Any,
    resources: Dict[str, Dict[str, str]],
    http_listen_port: int,
    socks_listen_port: int,
) -> Dict[str, Any]:
    host = str(getattr(proxy_server, "host", "") or "").strip()
    port = str(getattr(proxy_server, "port", "") or "").strip()
    username = str(getattr(proxy_server, "username", "") or "").strip()
    password = str(getattr(proxy_server, "password", "") or "").strip()
    proxy_id = str(getattr(proxy_server, "id", "") or "").strip()
    upstream_protocol = _upstream_proxy_protocol(proxy_server)
    upstream_proxy_scheme = _upstream_proxy_scheme(proxy_server)

    env = [
        {"name": "UPSTREAM_PROTOCOL", "value": upstream_protocol},
        {"name": "UPSTREAM_PROXY_SCHEME", "value": upstream_proxy_scheme},
        {"name": "UPSTREAM_HOST", "value": host},
        {"name": "UPSTREAM_PORT", "value": port},
        {"name": "HTTP_LISTEN_PORT", "value": str(http_listen_port)},
        {"name": "SOCKS_LISTEN_PORT", "value": str(socks_listen_port)},
    ]
    if username:
        env.append({"name": "UPSTREAM_USERNAME", "value": username})
    if password:
        env.append({"name": "UPSTREAM_PASSWORD", "value": password})

    labels = {
        "app": "sandbox-egress-proxy",
        "component": "sandbox-egress",
        "agent_id": agent_id,
    }
    if proxy_id:
        labels["proxy_id"] = proxy_id

    spec: Dict[str, Any] = {
        "automountServiceAccountToken": False,
        "terminationGracePeriodSeconds": 30,
        "securityContext": {
            "seccompProfile": {"type": "RuntimeDefault"},
        },
        "containers": [
            {
                "name": "egress-proxy",
                "image": image,
                "imagePullPolicy": "IfNotPresent",
                "ports": [
                    {"name": "http", "containerPort": http_listen_port},
                    {"name": "socks5", "containerPort": socks_listen_port},
                ],
                "env": env,
                "securityContext": {
                    "allowPrivilegeEscalation": False,
                    "runAsNonRoot": True,
                    "runAsUser": 1000,
                    "runAsGroup": 1000,
                    "capabilities": {"drop": ["ALL"]},
                },
                "resources": {
                    "requests": dict(resources["requests"]),
                    "limits": dict(resources["limits"]),
                },
                "readinessProbe": {
                    "tcpSocket": {"port": http_listen_port},
                    "initialDelaySeconds": 3,
                    "periodSeconds": 5,
                    "failureThreshold": 3,
                },
                "livenessProbe": {
                    "tcpSocket": {"port": http_listen_port},
                    "initialDelaySeconds": 5,
                    "periodSeconds": 10,
                    "failureThreshold": 3,
                },
            }
        ],
    }
    if runtime_class:
        spec["runtimeClassName"] = runtime_class
    if service_account:
        spec["serviceAccountName"] = service_account

    return {
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {
            "name": pod_name,
            "namespace": namespace,
            "labels": labels,
        },
        "spec": spec,
    }


def _upstream_proxy_scheme(proxy_server: Any) -> str:
    proxy_type = str(getattr(proxy_server, "proxy_type", "") or "").strip().upper()
    if proxy_type in {"HTTP", "HTTPS", "SOCKS5"}:
        return proxy_type.lower()
    raise SandboxComputeUnavailable(f"Unsupported proxy type for sandbox egress: {proxy_type or 'unknown'}")


def _upstream_proxy_protocol(proxy_server: Any) -> str:
    proxy_type = str(getattr(proxy_server, "proxy_type", "") or "").strip().upper()
    if proxy_type == "SOCKS5":
        return "socks5"
    if proxy_type in {"HTTP", "HTTPS"}:
        return "http"
    raise SandboxComputeUnavailable(f"Unsupported proxy type for sandbox egress: {proxy_type or 'unknown'}")


def _egress_proxy_pod_matches(
    pod: Dict[str, Any],
    *,
    proxy_server: Any,
    resources: Dict[str, Dict[str, str]],
    http_listen_port: int,
    socks_listen_port: int,
) -> bool:
    metadata = (pod.get("metadata") or {}) if isinstance(pod, dict) else {}
    labels = (metadata.get("labels") or {}) if isinstance(metadata, dict) else {}
    desired_proxy_id = str(getattr(proxy_server, "id", "") or "")
    if desired_proxy_id and labels.get("proxy_id") != desired_proxy_id:
        return False

    spec = (pod.get("spec") or {}) if isinstance(pod, dict) else {}
    containers = spec.get("containers") or []
    if not containers:
        return False
    container = containers[0] or {}
    env_entries = container.get("env") or []
    env = {
        str(entry.get("name")): str(entry.get("value", ""))
        for entry in env_entries
        if isinstance(entry, dict) and entry.get("name")
    }

    expected = {
        "UPSTREAM_PROTOCOL": _upstream_proxy_protocol(proxy_server),
        "UPSTREAM_PROXY_SCHEME": _upstream_proxy_scheme(proxy_server),
        "UPSTREAM_HOST": str(getattr(proxy_server, "host", "") or "").strip(),
        "UPSTREAM_PORT": str(getattr(proxy_server, "port", "") or "").strip(),
        "HTTP_LISTEN_PORT": str(http_listen_port),
        "SOCKS_LISTEN_PORT": str(socks_listen_port),
        "UPSTREAM_USERNAME": str(getattr(proxy_server, "username", "") or "").strip(),
        "UPSTREAM_PASSWORD": str(getattr(proxy_server, "password", "") or "").strip(),
    }
    if not all(env.get(key, "") == value for key, value in expected.items()):
        return False
    return _container_resources_match(container, resources)


def _sandbox_pod_matches(
    pod: Dict[str, Any],
    *,
    image: str,
    resources: Dict[str, Dict[str, str]],
    workspace_volume_mode: str,
    pvc_name: str,
    emptydir_size_limit: str,
    egress_service_name: Optional[str],
    http_proxy_port: int,
    socks_proxy_port: int,
    no_proxy: Optional[str],
) -> bool:
    spec = (pod.get("spec") or {}) if isinstance(pod, dict) else {}
    containers = spec.get("containers") or []
    if not containers:
        return False
    container = next(
        (
            entry
            for entry in containers
            if isinstance(entry, dict) and str(entry.get("name", "")).strip() == "sandbox-supervisor"
        ),
        containers[0],
    )
    if not isinstance(container, dict):
        return False
    if str(container.get("image", "") or "").strip() != image:
        return False

    env_entries = container.get("env") or []
    env = {
        str(entry.get("name")): str(entry.get("value", ""))
        for entry in env_entries
        if isinstance(entry, dict) and entry.get("name")
    }
    if env.get("SANDBOX_RUNTIME_CACHE_ROOT", "") != "/runtime-cache":
        return False
    if env.get("SANDBOX_AGENT_WORKSPACE_LAYOUT", "") != "isolated":
        return False
    if not _workspace_volume_matches(
        spec,
        mode=workspace_volume_mode,
        pvc_name=pvc_name,
        emptydir_size_limit=emptydir_size_limit,
    ):
        return False

    expected_proxy_env = {
        str(entry.get("name")): str(entry.get("value", ""))
        for entry in _build_proxy_env(
            egress_service_name=egress_service_name,
            http_proxy_port=http_proxy_port,
            socks_proxy_port=socks_proxy_port,
            no_proxy=no_proxy,
        )
        if isinstance(entry, dict) and entry.get("name")
    }
    proxy_keys = {
        "HTTP_PROXY",
        "HTTPS_PROXY",
        "ALL_PROXY",
        "NO_PROXY",
        "http_proxy",
        "https_proxy",
        "all_proxy",
        "no_proxy",
    }
    if not all(env.get(key, "") == expected_proxy_env.get(key, "") for key in proxy_keys):
        return False
    return _container_resources_match(container, resources)


def _workspace_volume_matches(
    pod_spec: Dict[str, Any],
    *,
    mode: str,
    pvc_name: str,
    emptydir_size_limit: str,
) -> bool:
    mode = _normalize_workspace_volume_mode(mode)
    volumes = pod_spec.get("volumes") or []
    volume = next(
        (
            entry
            for entry in volumes
            if isinstance(entry, dict) and str(entry.get("name", "")).strip() == "workspace"
        ),
        None,
    )
    if not isinstance(volume, dict):
        return False
    if mode == _WORKSPACE_VOLUME_MODE_PVC:
        claim = volume.get("persistentVolumeClaim") or {}
        return isinstance(claim, dict) and str(claim.get("claimName", "")).strip() == pvc_name

    empty_dir = volume.get("emptyDir")
    if not isinstance(empty_dir, dict):
        return False
    size_limit = str(emptydir_size_limit or "").strip()
    if not size_limit:
        return True
    return _resource_quantities_match(empty_dir.get("sizeLimit", ""), size_limit)


def _container_resources_match(container: Dict[str, Any], expected_resources: Dict[str, Dict[str, str]]) -> bool:
    actual_resources = container.get("resources") or {}
    if not isinstance(actual_resources, dict):
        return False
    actual_requests = actual_resources.get("requests") or {}
    actual_limits = actual_resources.get("limits") or {}
    if not isinstance(actual_requests, dict) or not isinstance(actual_limits, dict):
        return False
    expected_requests = expected_resources.get("requests") or {}
    expected_limits = expected_resources.get("limits") or {}
    return (
        all(_resource_quantities_match(actual_requests.get(key, ""), value) for key, value in expected_requests.items())
        and all(_resource_quantities_match(actual_limits.get(key, ""), value) for key, value in expected_limits.items())
    )


def _resource_quantities_match(actual: Any, expected: Any) -> bool:
    actual_value = str(actual or "").strip()
    expected_value = str(expected or "").strip()
    if actual_value == expected_value:
        return True
    try:
        return _parse_kubernetes_quantity(actual_value) == _parse_kubernetes_quantity(expected_value)
    except ValueError:
        return False


def _parse_kubernetes_quantity(value: str) -> Decimal:
    match = _QUANTITY_RE.fullmatch(value)
    if not match:
        raise ValueError(f"Unsupported Kubernetes quantity: {value}")
    number_text, suffix = match.groups()
    try:
        number = Decimal(number_text)
    except InvalidOperation as exc:
        raise ValueError(f"Unsupported Kubernetes quantity: {value}") from exc
    if suffix in _DECIMAL_SI_MULTIPLIERS:
        return number * _DECIMAL_SI_MULTIPLIERS[suffix]
    if suffix in _BINARY_SI_MULTIPLIERS:
        return number * _BINARY_SI_MULTIPLIERS[suffix]
    raise ValueError(f"Unsupported Kubernetes quantity: {value}")


def _build_egress_proxy_service_manifest(
    *,
    service_name: str,
    namespace: str,
    agent_id: str,
    http_port: int,
    http_target_port: int,
    socks_port: int,
    socks_target_port: int,
) -> Dict[str, Any]:
    return {
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {
            "name": service_name,
            "namespace": namespace,
            "labels": {
                "app": "sandbox-egress-proxy",
                "component": "sandbox-egress",
                "agent_id": agent_id,
            },
        },
        "spec": {
            "type": "ClusterIP",
            "selector": {
                "app": "sandbox-egress-proxy",
                "agent_id": agent_id,
            },
            "ports": [
                {
                    "name": "http",
                    "port": http_port,
                    "targetPort": http_target_port,
                    "protocol": "TCP",
                },
                {
                    "name": "socks5",
                    "port": socks_port,
                    "targetPort": socks_target_port,
                    "protocol": "TCP",
                }
            ],
        },
    }


def _build_sandbox_service_manifest(
    *,
    service_name: str,
    namespace: str,
    agent_id: str,
    port: int,
    target_port: int,
) -> Dict[str, Any]:
    return {
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {
            "name": service_name,
            "namespace": namespace,
            "labels": {
                "app": "sandbox-compute",
                "component": "sandbox-agent",
                "agent_id": agent_id,
            },
        },
        "spec": {
            "type": "ClusterIP",
            "selector": {
                "app": "sandbox-compute",
                "agent_id": agent_id,
            },
            "ports": [
                {
                    "name": "http",
                    "port": port,
                    "targetPort": target_port,
                    "protocol": "TCP",
                }
            ],
        },
    }


def _merge_no_proxy_values(*values: str) -> str:
    parts: list[str] = []
    for value in values:
        if not value:
            continue
        for item in value.split(","):
            cleaned = item.strip()
            if not cleaned or cleaned in parts:
                continue
            parts.append(cleaned)
    return ",".join(parts)
