sand-boxer/src/sandboxer/models.py

"""Pydantic schemas for profiles, extensions, and sandbox lifecycle."""

from __future__ import annotations

from datetime import datetime
from enum import StrEnum
from typing import Any, Literal

from pydantic import BaseModel, Field


class ActorType(StrEnum):
    ADM = "adm"
    AGT = "agt"
    ATM = "atm"


class SandboxState(StrEnum):
    REQUESTED = "requested"
    PROVISIONING = "provisioning"
    READY = "ready"
    ACTIVE = "active"
    EXPIRED = "expired"
    FAILED = "failed"
    DESTROYING = "destroying"
    DESTROYED = "destroyed"


class RouteStrategy(StrEnum):
    PREFER_SELF_HOSTED = "prefer-self-hosted"
    LOWEST_COST = "lowest-cost"
    LOWEST_LATENCY = "lowest-latency"
    EXPLICIT = "explicit"


class Consumer(BaseModel):
    actor: ActorType
    project: str
    session_id: str | None = None
    run_id: str | None = None


class IsolationSpec(BaseModel):
    level: Literal["container", "microvm", "policy"] = "container"


class NetworkSpec(BaseModel):
    default: Literal["deny", "allow"] = "deny"
    egress: list[str] = Field(default_factory=list)


class WorkspaceSpec(BaseModel):
    mode: Literal["mirror", "remote-canonical"] = "remote-canonical"
    access: Literal["none", "ro", "rw"] = "rw"


class TtlSpec(BaseModel):
    default: str = "4h"
    max: str = "24h"
    idle_reap: str | None = None


class ResourceSpec(BaseModel):
    cpu: str | None = None
    memory_mb: int | None = None


class SetupSpec(BaseModel):
    instructions: str = ""
    secret_refs: list[str] = Field(default_factory=list)


class PlacementSpec(BaseModel):
    prefer: list[str] = Field(default_factory=list)
    fallback: list[str] = Field(default_factory=list)


class ReachabilitySpec(BaseModel):
    tunnel: str = "ops-bridge"
    identity: str = "ops-warden"


class ProfileMetadata(BaseModel):
    cost_class: Literal["self-hosted", "saas-metered"] = "self-hosted"
    latency_class: str = "standard"
    observability: Literal["none", "canary"] = "none"


class RouteSpec(BaseModel):
    strategy: RouteStrategy = RouteStrategy.EXPLICIT
    extensions: list[str] = Field(default_factory=list)
    max_cost_per_hour_usd: float | None = None


class MeterQuote(BaseModel):
    extension_id: str
    estimated_usd: float
    unit: Literal["per_hour", "per_session"] = "per_hour"
    duration_s: int = 3600


class MeterRecord(BaseModel):
    pricing_model: Literal["self-hosted", "metered"] = "self-hosted"
    estimate_usd: float | None = None
    actual_usd: float | None = None
    duration_s: float | None = None
    currency: str = "USD"


class Profile(BaseModel):
    id: str
    version: str
    extension: str
    route: RouteSpec | None = None
    isolation: IsolationSpec = Field(default_factory=IsolationSpec)
    network: NetworkSpec = Field(default_factory=NetworkSpec)
    workspace: WorkspaceSpec = Field(default_factory=WorkspaceSpec)
    scope_default: Literal["session", "agent", "shared"] = "session"
    ttl: TtlSpec = Field(default_factory=TtlSpec)
    resources: ResourceSpec = Field(default_factory=ResourceSpec)
    setup: SetupSpec = Field(default_factory=SetupSpec)
    placement: PlacementSpec = Field(default_factory=PlacementSpec)
    reachability: ReachabilitySpec = Field(default_factory=ReachabilitySpec)
    metadata: ProfileMetadata = Field(default_factory=ProfileMetadata)


class ExtensionCapabilities(BaseModel):
    isolation_levels: list[str] = Field(default_factory=lambda: ["container"])
    regions: list[str] = Field(default_factory=list)
    persistence: bool = False
    pricing_model: Literal["self-hosted", "metered"] = "self-hosted"


class Extension(BaseModel):
    id: str
    title: str
    description: str = ""
    handler: str
    capabilities: ExtensionCapabilities = Field(default_factory=ExtensionCapabilities)
    config: dict[str, Any] = Field(default_factory=dict)


class SandboxCreateRequest(BaseModel):
    profile: str
    inputs: dict[str, str] = Field(default_factory=dict)
    consumer: Consumer
    ttl: str | None = None


class Reachability(BaseModel):
    ssh: str | None = None
    remote_dir: str | None = None
    compose_project: str | None = None
    host: str | None = None
    endpoint: str | None = None


class SandboxStatus(BaseModel):
    sandbox_id: str
    profile_id: str
    extension_id: str
    state: SandboxState
    consumer: Consumer
    host: str | None = None
    reachability: Reachability | None = None
    inputs: dict[str, str] = Field(default_factory=dict)
    error: str | None = None
    meter: MeterRecord | None = None
    telemetry: dict | None = None  # IntrospectionReport JSON when canary
    created_at: datetime
    updated_at: datetime
    ready_at: datetime | None = None
    destroyed_at: datetime | None = None