"""Extension author SDK — base contract for sandbox backends.""" from __future__ import annotations import uuid from abc import ABC, abstractmethod from typing import Any from sandboxer.models import MeterQuote, Profile class SandboxExtension(ABC): """Base class for self-hosted and SaaS sandbox extensions.""" def __init__(self, config: dict[str, Any] | None = None) -> None: self.config: dict[str, Any] = config or {} @staticmethod def new_sandbox_id(inputs: dict[str, str]) -> str: return inputs.get("sandbox_id") or str(uuid.uuid4())[:8] @abstractmethod def provision( self, profile: Profile, inputs: dict[str, str], host: str ) -> dict[str, str]: """Create or attach sandbox resources. Returns a handle dict for later ops.""" @abstractmethod def wait_ready(self, handle: dict[str, str]) -> dict[str, str]: """Confirm reachability. Returns reachability descriptor fields.""" @abstractmethod def teardown(self, handle: dict[str, str]) -> dict[str, str]: """Release sandbox resources. Returns cleanup report fields.""" def estimate_cost( self, profile: Profile, inputs: dict[str, str], *, duration_s: int = 3600, ) -> MeterQuote | None: """Optional pre-create cost quote (metered SaaS extensions).""" return None def meter_actual(self, handle: dict[str, str], *, duration_s: float) -> float | None: """Optional post-destroy actual cost in USD.""" return None def supports_snapshots(self) -> bool: """Whether this extension implements checkpoint snapshot/restore.""" return False def snapshot(self, handle: dict[str, str]) -> dict[str, str]: """Capture workspace checkpoint. Returns snapshot metadata including snapshot_id.""" raise NotImplementedError(f"{type(self).__name__} does not support snapshots") def restore_from_snapshot( self, profile: Profile, snapshot_meta: dict[str, str], inputs: dict[str, str], host: str, ) -> dict[str, str]: """Provision a new sandbox from a prior checkpoint.""" raise NotImplementedError(f"{type(self).__name__} does not support restore")