deer-flow/docker/provisioner/app.py

"""DeerFlow Sandbox Provisioner Service.

Dynamically creates and manages per-sandbox Pods in Kubernetes.
Each ``sandbox_id`` gets its own Pod + Service.  The backend accesses sandboxes
through NodePort or Kubernetes service DNS, depending on configuration.

The provisioner connects to the host machine's Kubernetes cluster via a
mounted kubeconfig (``~/.kube/config``) or in-cluster config.  Sandbox Pods
run in K8s and are accessed by the backend via the configured Service mode.

Endpoints:
    POST   /api/sandboxes              — Create a sandbox Pod + Service
    DELETE /api/sandboxes/{sandbox_id} — Destroy a sandbox Pod + Service
    GET    /api/sandboxes/{sandbox_id} — Get sandbox status & URL
    GET    /api/sandboxes              — List all sandboxes
    GET    /health                     — Provisioner health check

Architecture (docker-compose-dev):
    ┌────────────┐  HTTP  ┌─────────────┐  K8s API  ┌──────────────┐
    │ remote     │ ─────▸ │ provisioner │ ────────▸ │  host K8s    │
    │ _backend   │        │ :8002       │           │  API server  │
    └────────────┘        └─────────────┘           └──────┬───────┘
                                                           │ creates
                          ┌─────────────┐           ┌──────▼───────┐
                          │   backend   │ ────────▸ │   sandbox    │
                          │             │ direct/DNS│   Pod(s)     │
                          └─────────────┘           └──────────────┘
"""

from __future__ import annotations

import logging
import os
import re
import secrets
import time
from contextlib import asynccontextmanager

import urllib3
from fastapi import FastAPI, HTTPException, Request, Response
from kubernetes import client as k8s_client
from kubernetes import config as k8s_config
from kubernetes.client.rest import ApiException
from pydantic import BaseModel, Field

# Suppress only the InsecureRequestWarning from urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

logger = logging.getLogger(__name__)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)

# ── Configuration (all tuneable via environment variables) ───────────────

K8S_NAMESPACE = os.environ.get("K8S_NAMESPACE", "deer-flow")
SANDBOX_IMAGE = os.environ.get(
    "SANDBOX_IMAGE",
    "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest",
)
SKILLS_HOST_PATH = os.environ.get("SKILLS_HOST_PATH", "/skills")
THREADS_HOST_PATH = os.environ.get("THREADS_HOST_PATH", "/.deer-flow/threads")
DEER_FLOW_HOST_BASE_DIR = os.environ.get("DEER_FLOW_HOST_BASE_DIR", "/.deer-flow")
SKILLS_PVC_NAME = os.environ.get("SKILLS_PVC_NAME", "")
USERDATA_PVC_NAME = os.environ.get("USERDATA_PVC_NAME", "")
SKILLS_PVC_SUBPATH_TEMPLATE = os.environ.get("SKILLS_PVC_SUBPATH_TEMPLATE", "")
SANDBOX_CONTAINER_PORT_RAW = os.environ.get("SANDBOX_CONTAINER_PORT", "8080")
SANDBOX_SERVICE_TYPE = os.environ.get("SANDBOX_SERVICE_TYPE", "NodePort")
try:
    SANDBOX_CONTAINER_PORT = int(SANDBOX_CONTAINER_PORT_RAW)
except ValueError as exc:
    raise RuntimeError(f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT_RAW!r}; expected an integer TCP port") from exc
if not (1 <= SANDBOX_CONTAINER_PORT <= 65535):
    raise RuntimeError(f"Invalid SANDBOX_CONTAINER_PORT={SANDBOX_CONTAINER_PORT}; expected a value in [1, 65535]")
if SANDBOX_SERVICE_TYPE not in {"NodePort", "ClusterIP"}:
    raise RuntimeError(f"Invalid SANDBOX_SERVICE_TYPE={SANDBOX_SERVICE_TYPE!r}; expected 'NodePort' or 'ClusterIP'")
SAFE_THREAD_ID_PATTERN = r"^[A-Za-z0-9_\-]+$"
SAFE_USER_ID_PATTERN = r"^[A-Za-z0-9_\-]+$"
DEFAULT_USER_ID = "default"

# Path to the kubeconfig *inside* the provisioner container.
# Typically the host's ~/.kube/config is mounted here.
KUBECONFIG_PATH = os.environ.get("KUBECONFIG_PATH", "/root/.kube/config")
PROVISIONER_API_KEY = os.environ.get("PROVISIONER_API_KEY", "")

# The hostname / IP that the backend uses to reach NodePort services. On Docker
# Desktop for macOS this is ``host.docker.internal``; on Linux it may be the
# host's LAN IP. Ignored when SANDBOX_SERVICE_TYPE=ClusterIP.
NODE_HOST = os.environ.get("NODE_HOST", "host.docker.internal")


def join_host_path(base: str, *parts: str) -> str:
    """Join host filesystem path segments while preserving native style."""
    if not parts:
        return base

    if re.match(r"^[A-Za-z]:[\\/]", base) or base.startswith("\\\\") or "\\" in base:
        from pathlib import PureWindowsPath

        result = PureWindowsPath(base)
        for part in parts:
            result /= part
        return str(result)

    from pathlib import Path

    result = Path(base)
    for part in parts:
        result /= part
    return str(result)


# ── K8s client setup ────────────────────────────────────────────────────

core_v1: k8s_client.CoreV1Api | None = None


def _init_k8s_client() -> k8s_client.CoreV1Api:
    """Load kubeconfig from the mounted host config and return a CoreV1Api.

    Tries the mounted kubeconfig first, then falls back to in-cluster
    config (useful if the provisioner itself runs inside K8s).
    """
    if os.path.exists(KUBECONFIG_PATH):
        if os.path.isdir(KUBECONFIG_PATH):
            raise RuntimeError(f"KUBECONFIG_PATH points to a directory, expected a file: {KUBECONFIG_PATH}")
        try:
            k8s_config.load_kube_config(config_file=KUBECONFIG_PATH)
            logger.info(f"Loaded kubeconfig from {KUBECONFIG_PATH}")
        except Exception as exc:
            raise RuntimeError(f"Failed to load kubeconfig from {KUBECONFIG_PATH}: {exc}") from exc
    else:
        logger.warning(f"Kubeconfig not found at {KUBECONFIG_PATH}; trying in-cluster config")
        try:
            k8s_config.load_incluster_config()
        except Exception as exc:
            raise RuntimeError(f"Failed to initialize Kubernetes client. No kubeconfig at {KUBECONFIG_PATH}, and in-cluster config is unavailable: {exc}") from exc

    # When connecting from inside Docker to the host's K8s API, the
    # kubeconfig may reference ``localhost`` or ``127.0.0.1``.  We
    # optionally rewrite the server address so it reaches the host.
    k8s_api_server = os.environ.get("K8S_API_SERVER")
    if k8s_api_server:
        configuration = k8s_client.Configuration.get_default_copy()
        configuration.host = k8s_api_server
        # Self-signed certs are common for local clusters
        configuration.verify_ssl = False
        api_client = k8s_client.ApiClient(configuration)
        return k8s_client.CoreV1Api(api_client)

    return k8s_client.CoreV1Api()


def _wait_for_kubeconfig(timeout: int = 30) -> None:
    """Wait for kubeconfig file if configured, then continue with fallback support."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if os.path.exists(KUBECONFIG_PATH):
            if os.path.isfile(KUBECONFIG_PATH):
                logger.info(f"Found kubeconfig file at {KUBECONFIG_PATH}")
                return
            if os.path.isdir(KUBECONFIG_PATH):
                raise RuntimeError(f"Kubeconfig path is a directory. Please mount a kubeconfig file at {KUBECONFIG_PATH}.")
            raise RuntimeError(f"Kubeconfig path exists but is not a regular file: {KUBECONFIG_PATH}")
        logger.info(f"Waiting for kubeconfig at {KUBECONFIG_PATH} …")
        time.sleep(2)
    logger.warning(f"Kubeconfig not found at {KUBECONFIG_PATH} after {timeout}s; will attempt in-cluster Kubernetes config")


def _ensure_namespace() -> None:
    """Create the K8s namespace if it does not yet exist."""
    try:
        core_v1.read_namespace(K8S_NAMESPACE)
        logger.info(f"Namespace '{K8S_NAMESPACE}' already exists")
    except ApiException as exc:
        if exc.status == 404:
            ns = k8s_client.V1Namespace(
                metadata=k8s_client.V1ObjectMeta(
                    name=K8S_NAMESPACE,
                    labels={
                        "app.kubernetes.io/name": "deer-flow",
                        "app.kubernetes.io/component": "sandbox",
                    },
                )
            )
            core_v1.create_namespace(ns)
            logger.info(f"Created namespace '{K8S_NAMESPACE}'")
        else:
            raise


# ── FastAPI lifespan ─────────────────────────────────────────────────────


@asynccontextmanager
async def lifespan(_app: FastAPI):
    global core_v1
    _wait_for_kubeconfig()
    core_v1 = _init_k8s_client()
    _ensure_namespace()
    logger.info("Provisioner is ready (using host Kubernetes)")
    yield


app = FastAPI(title="DeerFlow Sandbox Provisioner", lifespan=lifespan)


@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    if request.url.path.startswith("/api/"):
        key = request.headers.get("X-API-Key", "")
        if not PROVISIONER_API_KEY or not secrets.compare_digest(key, PROVISIONER_API_KEY):
            logger.warning("provisioner auth rejected: %s %s", request.method, request.url.path)
            return Response(status_code=401, content="Unauthorized")
    return await call_next(request)


# ── Request / Response models ───────────────────────────────────────────


class CreateSandboxRequest(BaseModel):
    sandbox_id: str
    thread_id: str = Field(pattern=SAFE_THREAD_ID_PATTERN)
    user_id: str = Field(default=DEFAULT_USER_ID, pattern=SAFE_USER_ID_PATTERN)
    include_legacy_skills: bool = False


class SandboxResponse(BaseModel):
    sandbox_id: str
    sandbox_url: str
    status: str


# ── K8s resource helpers ─────────────────────────────────────────────────


def _pod_name(sandbox_id: str) -> str:
    return f"sandbox-{sandbox_id}"


def _svc_name(sandbox_id: str) -> str:
    return f"sandbox-{sandbox_id}-svc"


def _sandbox_url(sandbox_id: str, node_port: int | None = None) -> str:
    """Build the sandbox access URL for the configured Service mode."""
    if SANDBOX_SERVICE_TYPE == "ClusterIP":
        return f"http://{_svc_name(sandbox_id)}.{K8S_NAMESPACE}.svc.cluster.local:{SANDBOX_CONTAINER_PORT}"
    if node_port is None:
        raise RuntimeError("node_port is required when SANDBOX_SERVICE_TYPE=NodePort")
    return f"http://{NODE_HOST}:{node_port}"


def _build_volumes(
    thread_id: str,
    user_id: str = DEFAULT_USER_ID,
    *,
    include_legacy_skills: bool = False,
) -> list[k8s_client.V1Volume]:
    """Build volume list: PVC when configured, otherwise hostPath.

    Skills are split into public, per-user custom, and legacy (global-custom)
    volumes so that ``/mnt/skills/{public,custom,legacy}/`` paths resolve
    correctly inside the sandbox — matching the hostPath layout produced by
    ``LocalSandboxProvider`` and ``AioSandboxProvider``.
    """
    volumes: list[k8s_client.V1Volume] = []

    # ── Skills volumes ────────────────────────────────────────────────

    if SKILLS_PVC_NAME:
        # PVC mode: three-way subPath not yet supported; fall back to
        # single-volume mount for backward compatibility.
        logger.warning("SKILLS_PVC_NAME is set — three-way skills layout is not supported in PVC mode yet; falling back to single /mnt/skills mount")
        volumes.append(
            k8s_client.V1Volume(
                name="skills",
                persistent_volume_claim=k8s_client.V1PersistentVolumeClaimVolumeSource(
                    claim_name=SKILLS_PVC_NAME,
                    read_only=True,
                ),
            )
        )
    else:
        # hostPath mode: three-way layout
        public_path = join_host_path(SKILLS_HOST_PATH, "public")
        volumes.append(
            k8s_client.V1Volume(
                name="skills-public",
                host_path=k8s_client.V1HostPathVolumeSource(
                    path=public_path,
                    type="Directory",
                ),
            )
        )

        user_custom_path = join_host_path(
            DEER_FLOW_HOST_BASE_DIR,
            "users",
            user_id,
            "skills",
            "custom",
        )
        volumes.append(
            k8s_client.V1Volume(
                name="skills-custom",
                host_path=k8s_client.V1HostPathVolumeSource(
                    path=user_custom_path,
                    type="DirectoryOrCreate",
                ),
            )
        )

        if include_legacy_skills:
            legacy_path = join_host_path(SKILLS_HOST_PATH, "custom")
            volumes.append(
                k8s_client.V1Volume(
                    name="skills-legacy",
                    host_path=k8s_client.V1HostPathVolumeSource(
                        path=legacy_path,
                        type="Directory",
                    ),
                )
            )

    # ── User-data volume ──────────────────────────────────────────────

    if USERDATA_PVC_NAME:
        userdata_vol = k8s_client.V1Volume(
            name="user-data",
            persistent_volume_claim=k8s_client.V1PersistentVolumeClaimVolumeSource(
                claim_name=USERDATA_PVC_NAME,
            ),
        )
    else:
        userdata_vol = k8s_client.V1Volume(
            name="user-data",
            host_path=k8s_client.V1HostPathVolumeSource(
                path=join_host_path(THREADS_HOST_PATH, thread_id, "user-data"),
                type="DirectoryOrCreate",
            ),
        )

    volumes.append(userdata_vol)
    return volumes


def _build_volume_mounts(
    thread_id: str,
    user_id: str = DEFAULT_USER_ID,
    *,
    include_legacy_skills: bool = False,
) -> list[k8s_client.V1VolumeMount]:
    """Build volume mount list, mirroring three-way skills layout.

    Skills are mounted to ``/mnt/skills/{public,custom,legacy}/`` so that
    category-aware ``Skill.get_container_path()`` paths resolve correctly.
    PVC mode falls back to a single ``/mnt/skills`` mount and can optionally
    scope that mount with ``SKILLS_PVC_SUBPATH_TEMPLATE``.
    """
    mounts: list[k8s_client.V1VolumeMount] = []

    if SKILLS_PVC_NAME:
        skills_mount = k8s_client.V1VolumeMount(
            name="skills",
            mount_path="/mnt/skills",
            read_only=True,
        )
        if SKILLS_PVC_SUBPATH_TEMPLATE:
            skills_mount.sub_path = SKILLS_PVC_SUBPATH_TEMPLATE.format(
                user_id=user_id,
                thread_id=thread_id,
            )
        mounts.append(skills_mount)
    else:
        mounts.extend(
            [
                k8s_client.V1VolumeMount(
                    name="skills-public",
                    mount_path="/mnt/skills/public",
                    read_only=True,
                ),
                k8s_client.V1VolumeMount(
                    name="skills-custom",
                    mount_path="/mnt/skills/custom",
                    read_only=True,
                ),
            ]
        )
        if include_legacy_skills:
            mounts.append(
                k8s_client.V1VolumeMount(
                    name="skills-legacy",
                    mount_path="/mnt/skills/legacy",
                    read_only=True,
                )
            )

    userdata_mount = k8s_client.V1VolumeMount(
        name="user-data",
        mount_path="/mnt/user-data",
        read_only=False,
    )
    if USERDATA_PVC_NAME:
        userdata_mount.sub_path = f"deer-flow/users/{user_id}/threads/{thread_id}/user-data"
    mounts.append(userdata_mount)

    return mounts


def _build_pod(
    sandbox_id: str,
    thread_id: str,
    user_id: str = DEFAULT_USER_ID,
    *,
    include_legacy_skills: bool = False,
) -> k8s_client.V1Pod:
    """Construct a Pod manifest for a single sandbox."""
    return k8s_client.V1Pod(
        metadata=k8s_client.V1ObjectMeta(
            name=_pod_name(sandbox_id),
            namespace=K8S_NAMESPACE,
            labels={
                "app": "deer-flow-sandbox",
                "sandbox-id": sandbox_id,
                "app.kubernetes.io/name": "deer-flow",
                "app.kubernetes.io/component": "sandbox",
            },
        ),
        spec=k8s_client.V1PodSpec(
            containers=[
                k8s_client.V1Container(
                    name="sandbox",
                    image=SANDBOX_IMAGE,
                    image_pull_policy="IfNotPresent",
                    ports=[
                        k8s_client.V1ContainerPort(
                            name="http",
                            container_port=SANDBOX_CONTAINER_PORT,
                            protocol="TCP",
                        )
                    ],
                    readiness_probe=k8s_client.V1Probe(
                        http_get=k8s_client.V1HTTPGetAction(
                            path="/v1/sandbox",
                            port=SANDBOX_CONTAINER_PORT,
                        ),
                        initial_delay_seconds=5,
                        period_seconds=5,
                        timeout_seconds=3,
                        failure_threshold=3,
                    ),
                    liveness_probe=k8s_client.V1Probe(
                        http_get=k8s_client.V1HTTPGetAction(
                            path="/v1/sandbox",
                            port=SANDBOX_CONTAINER_PORT,
                        ),
                        initial_delay_seconds=10,
                        period_seconds=10,
                        timeout_seconds=3,
                        failure_threshold=3,
                    ),
                    resources=k8s_client.V1ResourceRequirements(
                        requests={
                            "cpu": "100m",
                            "memory": "256Mi",
                            "ephemeral-storage": "500Mi",
                        },
                        limits={
                            "cpu": "1000m",
                            "memory": "1Gi",
                            "ephemeral-storage": "500Mi",
                        },
                    ),
                    volume_mounts=_build_volume_mounts(
                        thread_id,
                        user_id=user_id,
                        include_legacy_skills=include_legacy_skills,
                    ),
                    security_context=k8s_client.V1SecurityContext(
                        privileged=False,
                        allow_privilege_escalation=True,
                    ),
                )
            ],
            volumes=_build_volumes(
                thread_id,
                user_id=user_id,
                include_legacy_skills=include_legacy_skills,
            ),
            restart_policy="Always",
        ),
    )


def _build_service(sandbox_id: str) -> k8s_client.V1Service:
    """Construct a Service manifest for the configured access mode."""
    return k8s_client.V1Service(
        metadata=k8s_client.V1ObjectMeta(
            name=_svc_name(sandbox_id),
            namespace=K8S_NAMESPACE,
            labels={
                "app": "deer-flow-sandbox",
                "sandbox-id": sandbox_id,
                "app.kubernetes.io/name": "deer-flow",
                "app.kubernetes.io/component": "sandbox",
            },
        ),
        spec=k8s_client.V1ServiceSpec(
            type=SANDBOX_SERVICE_TYPE,
            ports=[
                k8s_client.V1ServicePort(
                    name="http",
                    port=SANDBOX_CONTAINER_PORT,
                    target_port=SANDBOX_CONTAINER_PORT,
                    protocol="TCP",
                )
            ],
            selector={
                "sandbox-id": sandbox_id,
            },
        ),
    )


def _url_from_service(svc, sandbox_id: str) -> str | None:
    """Build the backend-facing sandbox URL from an already-fetched Service."""
    if SANDBOX_SERVICE_TYPE == "ClusterIP":
        return _sandbox_url(sandbox_id)

    for port in svc.spec.ports or []:
        if port.name == "http" and port.node_port:
            return _sandbox_url(sandbox_id, node_port=port.node_port)
    return None


def _sandbox_access_url(sandbox_id: str, *, tolerate_read_errors: bool = False) -> str | None:
    """Read the sandbox Service and return its backend-facing URL when ready."""
    try:
        svc = core_v1.read_namespaced_service(_svc_name(sandbox_id), K8S_NAMESPACE)
    except ApiException as exc:
        if exc.status == 404:
            return None
        if tolerate_read_errors and exc.status not in {401, 403}:
            logger.warning(
                "Transient error reading Service %s: status=%s reason=%s",
                _svc_name(sandbox_id),
                exc.status,
                exc.reason,
            )
            return None
        raise

    return _url_from_service(svc, sandbox_id)


def _get_pod_phase(sandbox_id: str) -> str:
    """Return the Pod phase (Pending / Running / Succeeded / Failed / Unknown)."""
    try:
        pod = core_v1.read_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE)
        return pod.status.phase or "Unknown"
    except ApiException:
        return "NotFound"


# ── API endpoints ────────────────────────────────────────────────────────


@app.get("/health")
async def health():
    """Provisioner health check."""
    return {"status": "ok"}


@app.post("/api/sandboxes", response_model=SandboxResponse)
def create_sandbox(req: CreateSandboxRequest):
    """Create a sandbox Pod + Service for *sandbox_id*.

    If the sandbox already exists, returns the existing information
    (idempotent).
    """
    sandbox_id = req.sandbox_id
    thread_id = req.thread_id
    user_id = req.user_id
    include_legacy_skills = req.include_legacy_skills

    logger.info(
        "Received request to create sandbox '%s' for thread '%s' user '%s' include_legacy_skills=%s",
        sandbox_id,
        thread_id,
        user_id,
        include_legacy_skills,
    )

    # ── Fast path: sandbox already exists ────────────────────────────
    existing_url = _sandbox_access_url(sandbox_id, tolerate_read_errors=True)
    if existing_url:
        return SandboxResponse(
            sandbox_id=sandbox_id,
            sandbox_url=existing_url,
            status=_get_pod_phase(sandbox_id),
        )

    # ── Create Pod ───────────────────────────────────────────────────
    try:
        core_v1.create_namespaced_pod(
            K8S_NAMESPACE,
            _build_pod(
                sandbox_id,
                thread_id,
                user_id=user_id,
                include_legacy_skills=include_legacy_skills,
            ),
        )
        logger.info(f"Created Pod {_pod_name(sandbox_id)}")
    except ApiException as exc:
        if exc.status != 409:  # 409 = AlreadyExists
            raise HTTPException(status_code=500, detail=f"Pod creation failed: {exc.reason}")

    # ── Create Service ───────────────────────────────────────────────
    try:
        core_v1.create_namespaced_service(K8S_NAMESPACE, _build_service(sandbox_id))
        logger.info(f"Created Service {_svc_name(sandbox_id)}")
    except ApiException as exc:
        if exc.status != 409:
            # Roll back the Pod on failure
            try:
                core_v1.delete_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE)
            except ApiException:
                pass
            raise HTTPException(status_code=500, detail=f"Service creation failed: {exc.reason}")

    # ── Wait until the Service has a usable access URL ───────────────
    sandbox_url: str | None = None
    for _ in range(20):
        sandbox_url = _sandbox_access_url(sandbox_id, tolerate_read_errors=True)
        if sandbox_url:
            break
        time.sleep(0.5)

    if not sandbox_url:
        raise HTTPException(status_code=500, detail="Service access URL was not available in time")

    return SandboxResponse(
        sandbox_id=sandbox_id,
        sandbox_url=sandbox_url,
        status=_get_pod_phase(sandbox_id),
    )


@app.delete("/api/sandboxes/{sandbox_id}")
def destroy_sandbox(sandbox_id: str):
    """Destroy a sandbox Pod + Service."""
    errors: list[str] = []

    # Delete Service
    try:
        core_v1.delete_namespaced_service(_svc_name(sandbox_id), K8S_NAMESPACE)
        logger.info(f"Deleted Service {_svc_name(sandbox_id)}")
    except ApiException as exc:
        if exc.status != 404:
            errors.append(f"service: {exc.reason}")

    # Delete Pod
    try:
        core_v1.delete_namespaced_pod(_pod_name(sandbox_id), K8S_NAMESPACE)
        logger.info(f"Deleted Pod {_pod_name(sandbox_id)}")
    except ApiException as exc:
        if exc.status != 404:
            errors.append(f"pod: {exc.reason}")

    if errors:
        raise HTTPException(status_code=500, detail=f"Partial cleanup: {', '.join(errors)}")

    return {"ok": True, "sandbox_id": sandbox_id}


@app.get("/api/sandboxes/{sandbox_id}", response_model=SandboxResponse)
def get_sandbox(sandbox_id: str):
    """Return current status and URL for a sandbox."""
    sandbox_url = _sandbox_access_url(sandbox_id)
    if not sandbox_url:
        raise HTTPException(status_code=404, detail=f"Sandbox '{sandbox_id}' not found")

    return SandboxResponse(
        sandbox_id=sandbox_id,
        sandbox_url=sandbox_url,
        status=_get_pod_phase(sandbox_id),
    )


@app.get("/api/sandboxes")
def list_sandboxes():
    """List every sandbox currently managed in the namespace."""
    try:
        services = core_v1.list_namespaced_service(
            K8S_NAMESPACE,
            label_selector="app=deer-flow-sandbox",
        )
    except ApiException as exc:
        raise HTTPException(status_code=500, detail=f"Failed to list services: {exc.reason}")

    sandboxes: list[SandboxResponse] = []
    for svc in services.items:
        sid = (svc.metadata.labels or {}).get("sandbox-id")
        if not sid:
            continue
        sandbox_url = _url_from_service(svc, sid)
        if not sandbox_url:
            continue
        sandboxes.append(
            SandboxResponse(
                sandbox_id=sid,
                sandbox_url=sandbox_url,
                status=_get_pod_phase(sid),
            )
        )

    return {"sandboxes": sandboxes, "count": len(sandboxes)}