Here's a thinking process that leads to the suggested structure: 1. **Understand the Goal:** The user wants to build a "Unified Sandbox Provider SDK." This SDK needs to allow easy sending of *basic primitive commands* to an *automatically routed* sandbox environment. The scope is small initially (a couple of providers). 2. **Identify Key Challenges/Requirements:** * **Unification:** The SDK must hide the differences between various backend providers (Provider A, Provider B, etc.). * **Simplicity (Primitives):** Commands must be simple (e.g., `create`, `read`, `run_script`, `delete`), not complex, provider-specific APIs. * **Routing:** The SDK needs a mechanism (either explicit or automatic) to decide which backend provider handles the request. * **Extensibility:** It needs to be easy to add a third, fourth, or fifth provider later. * **Observability/Flow:** The SDK needs to manage the asynchronous communication (sending the command and getting a result). 3. **Determine the Core Architectural Pattern:** * *Abstraction Layer:* This is mandatory. An interface defines the common language. * *Strategy/Adapter Pattern:* Each specific provider (e.g., AWS Lambda, Custom VM) implements that common interface. * *Facade/Router:* A top-level class manages the selection and delegation. 4. **Design the Structure (Layer by Layer):** * **Layer 0: The Contract (The Primitives)** * What does the SDK promise to do? Define a common `Command` or `Request` structure.The user wants a design + implementation for a unified sandbox provider SDK. Let me parse the request: 1. **Suggest a structure** — architecture for a unified SDK that abstracts over multiple sandbox providers (like E2B, Daytona, Modal, Firecracker, local Docker, AWS Lambda-ish, etc.) 2. **Allow sending basic primitive commands** — file ops (read/write/list/delete), exec commands, maybe process management 3. **Automatically routed sandbox** — some routing layer that picks a provider based on config/heuristics (capability matching, region, cost, availability) 4. **Build it (small scope, just a couple providers)** — so implement, say, two providers: one remote (E2B is the canonical one) and one local (Docker, or a subprocess-based "local" provider that doesn't require Docker at all — that's actually smarter for testability). The user says "suggest a structure... and then build it (small scope)". So they want both architecture design and code. This is a coding task, so a fairly comprehensive response is appropriate. Format should include code blocks, headers, and a clear architecture explanation. Given the scope ("small scope, do just a couple providers"), I should keep it tight but complete and runnable. Let me think about the right language. Python is the most common for sandbox SDKs (E2B's primary SDK is Python, and sandbox use-cases are usually AI-agent code execution). I'll go with Python. Could also mention TypeScript as an alternative but focus on Python. **Core design decisions:** 1. **Provider-agnostic capability interface.** Define an abstract `Sandbox` interface with primitives: - `exec(command, cwd, env, timeout) -> ExecResult` (stdout, stderr, exit_code) - `read_file(path) -> bytes` - `write_file(path, data)` - `list_files(path) -> list[FileInfo]` - `delete_file(path)` - `mkdir(path)` - maybe `remove()` / context manager for lifecycle: `create()`, `stop()`, `alive()` 2. **Provider abstraction.** A `Provider` knows how to *create* sandboxes. `Sandbox` is the handle to a live sandbox. Providers register themselves via a registry. 3. **Routing.** A `Router` that, given a request spec (capabilities needed, region, labels, cost preference), picks a provider. Simple scoring: check `provider.capabilities` against required capabilities; score by preference; fall back in order of priority. Keep it small: capability matching + priority + optional explicit `provider=` override. 4. **Capability model.** Each provider declares capabilities: `exec`, `fs`, `network`, `snapshot`, `long_running`, etc. Also constraints like region. This drives automatic routing. 5. **Two providers:** - `LocalProcessProvider` — runs commands via `subprocess` in a temp dir on the host. Zero dependencies, great for tests/dev. File ops are just direct filesystem ops (with a root jail to keep it sandbox-ish scoped to a directory). - `E2BProvider` — uses the `e2b` SDK. Create sandbox, exec via `sandbox.commands.run`, file ops via `sandbox.files`. Docker could be a third but "couple providers" — I'll do local + E2B, and maybe stub-mention Docker as an easy extension. Actually, a Docker provider is very useful and not much code (docker SDK: `docker run`, `container.exec_run`, `container.get_archive` / `put_archive`). But the user said small scope, a couple providers. Two providers = local + E2B. I could include Docker as an optional third briefly... Better to keep to two and mention Docker as the next extension with a sketch. Hmm, actually let me reconsider: E2B requires an API key and pip install. Local works everywhere. That's a good pair: one real remote cloud sandbox, one local fallback that makes the whole thing testable without any keys. Actually, I think I should include Docker too since many people can't use E2B (API key needed) but have Docker. That would make three providers — pushing scope. The user said "a couple providers". Two is literal. But Docker adds real value... Let me include Docker as a third *optional* provider but keep it compact. Hmm. Risk: response gets long. Let me think about total length. The response will include: - Architecture diagram (ASCII) + module layout - Core interfaces (~100 lines) - Router (~50 lines) - Local provider (~80 lines) - E2B provider (~70 lines) - Docker provider (~90 lines) — optional - Config example (YAML) - Usage example - Tests sketch - Extension notes That's a lot. Maybe trim. I think local + Docker is actually the more universally runnable pair (both work without any cloud API keys), and E2B is the canonical cloud sandbox. Hmm, but Docker requires Docker installed. Local requires nothing. Decision: implement **LocalProcessProvider** and **E2BProvider** as the two, and include a compact **DockerProvider** since it's genuinely useful and shows how a third provider slots in — or maybe mention it in "next steps" with a brief interface mapping rather than full code. I'll gauge length. The user asked to "build it" so code is the deliverable. I'll do local + E2B fully, and Docker... I'll include it, actually. It's the most commonly deployed option and demonstrates that the abstraction generalizes beyond "the two obvious ones". But keep it tight. Hmm, let me reconsider scope discipline. "small scope, do just a couple providers" — the user explicitly constrained. Respecting that: two providers. I'll do Local (subprocess) + Docker (since both are self-hostable and testable, no API keys), OR Local + E2B. Which pair teaches more? E2B shows a real remote provider with async/REST semantics and auth. Docker shows containerization but its API is similar in spirit to local subprocess (exec_run ≈ subprocess.run). For teaching the abstraction, a remote cloud provider is more valuable — it proves the abstraction isn't just wrapping subprocess. But Docker is more universally usable... Compromise: full code for Local + Docker (both runnable by anyone with Docker, no keys), and a short section showing how E2B would slot in with a code sketch (condensed but functional-ish). That gives "a couple providers" fully built plus a pointer. Actually — alternatively full Local + E2B, sketch Docker. E2B sketch requires knowledge of their SDK API which changes (e2b v1 changed a lot: `Sandbox.create()`, `sandbox.commands.run()`, `sandbox.files.write()`, etc.). I know the v1 API decently: ```python from e2b import Sandbox sbx = Sandbox.create("ubuntu:22.04", timeout=300) r = sbx.commands.run("ls -la", cwd="/root") r.stdout, r.stderr, r.exit_code sbx.files.write("/path", b"data") # accepts str path + bytes/str sbx.files.read("/path") # returns bytes sbx.files.list("/path") # list of FileInfo with name, path, is_dir sbx.remove() ``` Roughly right. E2B `files.read` returns bytes by default (there's a `format` param: "bytes" | "text"). `files.list` returns FileInfo objects with `.name`, `.path`, `.is_dir`. `commands.run` returns `CommandResult` with `.stdout`, `.stderr`, `.exit_code`. Good. Docker SDK (docker-py): ```python import docker client = docker.from_env() container = client.containers.run("ubuntu:22.04", command="sleep infinity", detach=True, auto_remove=...) container.exec_run(["bash","-lc",cmd], workdir=..., environment=..., demux=True) -> (exit_code, (stdout, stderr)) # files: tar_bytes = container.get_archive(path) # (stream (tar), stat) -> need tarfile extraction container.put_archive(path, tar_bytes) ``` File ops via tar are a bit involved but doable compactly. `get_archive` returns (bits: generator of bytes, stat). `put_archive(container_path, data)` where data is a tar in bytes. That's maybe 60-90 lines for a decent Docker provider with a helper for tar packing. It's worth it because file ops via tar demonstrate the pain the abstraction hides — good pedagogy for "why an SDK". Okay, final scope decision: - Fully implement: core (interfaces, registry, router, config), LocalProvider, DockerProvider. - E2B: I'll include as a third full-ish provider? No — keep discipline. I'll give E2B a compact but complete implementation too, since it's ~60 lines and it's *the* sandbox provider most people mean. Hmm, three providers then. "A couple" — two or three. I think three is fine if each is compact, but let me watch total length. Total response will be long regardless (~400-500 lines of code). That's acceptable for a "build it" request. Actually, I worry about bloat. Let me pick **two fully: Local + Docker**, and a **short E2B sketch** in an "adding a cloud provider" section with the adapter mapping (maybe 25-40 lines, marked as "install `pip install e2b`" with the key method bodies). Hmm, but an E2B sketch that's incomplete is less useful than a complete one... Alternative: Local + E2B fully, Docker in "next steps". Meh. Let me think about who asks this: someone building an agent-execution platform, wanting a unified SDK like e2b/daytona/modal abstraction. The most valuable pair to *build* for them: one local (dev/test, zero deps) + one production-grade remote. E2B is the most popular sandbox-for-AI product and Python-first. Docker is the most deployable self-hosted option. You know what, I'll just do all three but keep each tight. Local ~60 lines, Docker ~80 lines, E2B ~55 lines. The user can trim. Three providers still honors "small scope, just a couple" in spirit — hmm, no, it kind of doesn't. The user said "do just a couple providers". I should respect that. Two providers: **Local + Docker**. Both zero-API-key, both immediately runnable, both demonstrate different adapter challenges (path mapping / root jail vs tar streaming). Then a brief "adding E2B" section with a condensed adapter (~30 lines) to show cloud integration, clearly labeled as the next step. That's respectful of scope and maximally runnable. Hmm, but if they wanted E2B specifically... The prompt says "automatically routed sandbox" and "providers" generically. Docker+Local covers routing demo well (route heavy jobs to docker, quick evals to local). Hmm wait, one more consideration: maybe the two most instructive are **local** and **E2B** because local is sync-subprocess and E2B is a managed remote service — maximal contrast in adapter complexity. Docker sits in between. Ugh. Decision paralysis. Let me just commit: **Local (subprocess) + Docker**, with E2B shown as an extension sketch. Rationale I can state: both runnable without cloud credentials; E2B adapter shown briefly so they can slot in the remote provider they actually use. This is defensible and complete. Hmm, actually let me reconsider once more. If I include Docker file ops via tar, that's the fiddliest part. Compact version: ```python def _tar_bytes(self, path: str, content: bytes) -> bytes: buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: ti = tarfile.TarInfo(name=os.path.basename(path)) ti.size = len(content) ti.mtime = int(time.time()) tf.addfile(ti, io.BytesIO(content)) return buf.getvalue() def write_file(self, path, data): parent = posixpath.dirname(path) self.container.put_archive(parent, self._tar_bytes(path, data)) def read_file(self, path): stream, _ = self.container.get_archive(path) raw = b"".join(stream) with tarfile.open(fileobj=io.BytesIO(raw)) as tf: member = tf.getmembers()[0] return tf.extractfile(member).read() ``` Note: `get_archive` returns the tar including the directory prefix (e.g. if you ask for /src/main.py, member is "main.py" at top level of tar... actually member name is the basename). Using getmembers()[0] and extractfile works. Also note put_archive can't create intermediate dirs; fine, note it. `exec_run` with `demux=True` returns `(exit_code, (stdout, stderr))` where missing stream is None. Container lifecycle: `client.containers.run(image, "sleep infinity", detach=True, auto_remove=True)` — auto_remove on stop. `container.stop()` with timeout. Also handle `container.status`. Local provider: - create: make temp dir `tempfile.mkdtemp(prefix="sandbox-")` as root. - exec: `subprocess.run(cmd, shell=True, cwd=root, env={**os.environ, **env}, capture_output=True, timeout=...)`. Note: shell=True to allow "primitive commands" as strings. Or accept both str and list. For a "primitive command" SDK, string commands with shell=True is ergonomic; list for safety. I'll accept str and run with shell=True in local; for docker, exec_run(["bash", "-lc", cmd]). - Files: guard against path traversal with a resolve check: `root = os.path.realpath(root)`; `_resolve(path)` -> if not under root, raise SandboxError. Use os.path.realpath and `os.path.commonpath` check. - list_files: os.scandir -> FileInfo(name, path, size, is_dir, modified_at). - read/write/delete/mkdir straightforward. Core files layout (single package `sandboxsdk` or nicer name like `pysandbox`? Let me name it `boxsdk`? "box" is taken by box.com. `sandboxkit`? `sbx`? I'll use `sandboxkit` — descriptive, available-ish. Or simply `sandbox`. Package name doesn't matter much; I'll call it `sandboxkit`.) Layout: ``` sandboxkit/ __init__.py # public API re-exports core.py # exceptions, dataclasses (ExecResult, FileInfo, SandboxSpec), protocols provider.py # Provider protocol + registry router.py # Router + routing rules config.py # load config (yaml or dict), build router providers/ __init__.py local.py docker.py api.py # top-level facade: SandboxKit class -> sandbox(spec) -> Sandbox handle tests/ test_local.py test_docker.py (skipif no docker) examples/ quickstart.py sandboxkit.yaml ``` Maybe merge registry into provider.py and facade into api.py. Keep file count modest. Could even do a single-module version for "small scope"? A single file `sandboxkit.py` with everything is actually very appealing for copy-paste and the user's "small scope". But structure was explicitly requested ("suggest a structure"), so a package layout is the point. Compromise: present package layout, and the code per file in separate code blocks with filenames as headers. That reads well. Hmm, or provide it as a handful of files but fewer: `core.py`, `providers/local.py`, `providers/docker.py`, `router.py`, `api.py` (+ tiny `config.py` merged into router or api?). Let me merge config parsing into `api.py`/`SandboxKit.from_config()`... Config as YAML file adds yaml dependency; use plain dict + optional yaml load. Keep it: `SandboxKit.from_dict(cfg)` and `from_yaml(path)` guarded by optional import. That's nice. **Key design details to articulate:** 1. **Two-layer split**: `Provider` (factory; stateless; declares capabilities) vs `Sandbox` (live handle; per-instance ops). This is the single most important structural decision. Routers select providers, not sandboxes. 2. **Primitive operation set** (the "instruction set" all providers must implement): - lifecycle: create/stop/health (alive) - exec: run a command, get exit/stdout/stderr, timeout, env, cwd - fs: read/write/list/delete/mkdir — bytes-based to be provider-neutral - Everything else (run code snippets, start servers, snapshots) is built on top of exec/fs, either as convenience methods in the SDK or provider capabilities. 3. **Capability-driven routing**: providers declare `{capabilities: {...}, cost/latency tier, regions, limits}`; the request declares needs (`exec`, `fs`, `network`, `gpu`, min_timeout, image, region, labels). Router = filter (must-have capabilities) → score (preferences: region match, cost tier, priority order) → pick best, with failover if create() fails. Keep the router small: deterministic, no magic. 4. **Uniform result types**: `ExecResult(exit_code, stdout, stderr, duration_ms)`, `FileInfo(...)`. Bytes everywhere for fs payloads. Timeouts normalized (seconds; `None` = provider default). Normalizing error surface: `SandboxError` hierarchy (`ProviderUnavailable`, `SandboxTimeout`, `NotFound`, `OperationFailed`). 5. **Context-manager lifecycle + explicit stop**: sandboxes are resources; `with kit.sandbox(...) as sbx:` auto-stops. 6. **Routing config**: ```yaml providers: - type: docker image: ubuntu:22.04 priority: 10 - type: local priority: 5 routing: default: auto ``` Request-level overrides: `kit.sandbox(provider="docker", needs=["network"], region=...)`. 7. **Sync vs async**: keep sync (small scope) and note async/`trio`/`anyio` as the production evolution. E2B has both sync and async SDKs; asyncio version is where you'd want async. I'll note it. 8. **Path semantics**: define that all fs paths in the API are absolute POSIX paths inside the sandbox (`/` = sandbox root). Providers map as needed (local: root dir; docker: container fs, maybe a workdir mount). This avoids the classic mess of relative paths per provider. 9. **Extensibility hooks**: registry with `@register("docker")` decorator, so third-party providers self-register on import. Config references by name. Let me now write the actual code carefully, making sure it's correct and runnable. --- **core.py** ```python from __future__ import annotations import time from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, Mapping, Optional, Sequence # ---------- errors ---------- class SandboxError(Exception): """Base error for everything raised by sandboxkit.""" class ProviderError(SandboxError): """The provider itself failed (auth, quota, infra).""" class SandboxNotFound(SandboxError): """The sandbox is gone / was never there.""" class SandboxTimeout(SandboxError): """An operation exceeded its deadline.""" class PathOutsideSandbox(SandboxError): """A filesystem path escaped the sandbox root.""" # ---------- result types ---------- @dataclass(frozen=True) class ExecResult: exit_code: int stdout: str stderr: str duration_ms: int @dataclass(frozen=True) class FileInfo: path: str # absolute, POSIX, inside sandbox name: str is_dir: bool size: int # 0 for dirs where unknown modified_at: float # epoch seconds, 0.0 if unknown @dataclass(frozen=True) class SandboxSpec: """What you want. Providers try to satisfy it; routing matches it.""" image: Optional[str] = None env: Mapping[str, str] = field(default_factory=dict) labels: Mapping[str, str] = field(default_factory=dict) needs: frozenset[str] = frozenset() # e.g. {"network", "gpu"} region: Optional[str] = None ttl_seconds: Optional[int] = None provider: Optional[str] = None # pin to one provider, skip routing timeout_seconds: Optional[int] = None ``` Hmm, `needs: frozenset[str]` with default_factory — frozenset default: `field(default_factory=frozenset)` works. `Mapping` for env with default dict — frozen dataclass with mutable default is a problem, hence default_factory. Fine. **Interfaces** — Provider protocol and Sandbox ABC. Where do they live? I'll put Sandbox ABC in core.py and Provider in provider.py along with the registry. ```python class Sandbox(ABC): """A live sandbox handle. All paths are absolute POSIX inside the sandbox.""" @property @abstractmethod def id(self) -> str: ... @property @abstractmethod def provider(self) -> str: ... # --- exec --- @abstractmethod def exec(self, command: str, *, cwd: str = "/", env: Optional[Mapping[str, str]] = None, timeout: Optional[float] = None) -> ExecResult: ... # --- filesystem primitives --- @abstractmethod def read_file(self, path: str) -> bytes: ... @abstractmethod def write_file(self, path: str, data: bytes | str) -> None: ... @abstractmethod def list_files(self, path: str = "/") -> list[FileInfo]: ... @abstractmethod def delete_file(self, path: str) -> None: ... @abstractmethod def mkdir(self, path: str) -> None: ... # --- lifecycle --- @abstractmethod def stop(self, *, force: bool = False) -> None: ... def is_alive(self) -> bool: try: return self.exec("true", timeout=10).exit_code == 0 except SandboxError: return False def __enter__(self): return self def __exit__(self, *exc): self.stop() # convenience built on the primitives def run(self, command: str, **kw) -> str: """exec + assert zero exit, return stdout.""" res = self.exec(command, **kw) if res.exit_code != 0: raise OperationFailed(...) return res.stdout ``` Hmm `OperationFailed` — add `class ExecFailed(SandboxError)` with result attached. Let me define: ```python class ExecFailed(SandboxError): def __init__(self, result: ExecResult): super().__init__(f"command failed with exit code {result.exit_code}: {result.stderr.strip()[:200]}") self.result = result ``` Convenience `run()` raising ExecFailed is handy for the "send basic primitive commands" UX. Good. Also `write_text`/`read_text` conveniences maybe. Keep small: write_file accepts bytes|str and encodes; add `read_text` helper? Minor. I'll skip read_text to keep surface small — actually one-liner conveniences are cheap and nice. I'll include a couple: `write_text`, `read_text`? Eh. Keep `write_file` accepting str|bytes and skip text readers. Users can decode. Fine. **Provider ABC:** ```python @dataclass(frozen=True) class ProviderInfo: name: str capabilities: frozenset[str] # {"exec", "fs", "network", "docker", "gpu", ...} regions: tuple[str, ...] = () # empty = any cost_tier: int = 1 # 0 cheapest ... higher pricier max_timeout_seconds: Optional[int] = None class Provider(ABC): info: ProviderInfo @abstractmethod def create(self, spec: SandboxSpec) -> Sandbox: ... def health(self) -> bool: return True ``` **Registry:** ```python _REGISTRY: dict[str, type[Provider]] = {} def register(cls: type[Provider]) -> type[Provider]: name = cls.__name__.removesuffix("Provider").lower() # DockerProvider -> "docker" _REGISTRY[name] = cls return cls def get_provider_class(name): ... def provider_names(): ... ``` Actually deriving name from class name is cute but fragile; explicit `name = "docker"` class attribute? I'll do `register("docker")` as a decorator with argument, clearer: ```python def register(name: str): def deco(cls): _REGISTRY[name] = cls return cls return deco ``` Simple. And `providers/__init__.py` imports local and docker so registration happens on `import sandboxkit` (via api.py importing providers package). **router.py** ```python @dataclass class ProviderEntry: name: str priority: int = 0 enabled: bool = True instance: Provider = instantiated lazily class Router: def __init__(self, entries: list[ProviderEntry]): ... def providers(self) -> list[Provider]: # priority-sorted, healthy def route(self, spec: SandboxSpec) -> Provider: # 1. explicit pin # 2. capability filter: provider.info.capabilities ⊇ spec.needs # 3. region filter # 4. image support? (keep simple: ignore; providers may raise if they can't) # 5. score: priority + region match bonus + cheaper cost tier preferred # 6. if none -> raise NoRouteError ``` Add `NoRouteError(SandboxError)`. Scoring: `score = entry.priority * 10 + (2 if region matches else 0) - provider.info.cost_tier`. Hmm priorities: higher priority = preferred. Keep deterministic and documented. Failover: in `SandboxKit.sandbox()`, iterate candidates from router (ranked list) and try create() until one succeeds; on failure (ProviderError), log and continue. So `Router.route` returns ranked list, not single. Yes — `route_all(spec) -> list[Provider]` ranked; kit tries in order. Config entries: how do we instantiate providers with their options (image default, docker base_url, local root)? ProviderEntry carries `options: dict` and instantiates `cls(**options)`. ```python @dataclass class ProviderConfig: type: str # registry name options: dict # provider-specific kwargs priority: int = 0 enabled: bool = True ``` Router builds/holds instances lazily. **api.py — facade** ```python class SandboxKit: def __init__(self, providers: list[ProviderConfig], *, failover: bool = True): self.router = Router(providers) @classmethod def from_dict(cls, cfg) -> SandboxKit @classmethod def from_yaml(cls, path) -> SandboxKit # optional yaml @classmethod def auto(cls) -> SandboxKit: # sane default: docker if available else local entries = [] if provider healthcheck docker ok -> add; local always add return cls(entries) def sandbox(self, spec: SandboxSpec | None = None, **overrides) -> Sandbox: spec = replace(spec, **overrides) if spec else build from kwargs errors = [] for provider in self.router.route(spec): try: return provider.create(spec) except ProviderError as e: errors.append((provider.info.name, e)); continue raise NoRouteError(...) ``` `**overrides` maps kwargs to SandboxSpec fields — nice ergonomics: `kit.sandbox(image="ubuntu:24.04", needs={"network"})`. **providers/local.py** ```python @register("local") class LocalProvider(Provider): info = ProviderInfo(name="local", capabilities=frozenset({"exec", "fs"}), cost_tier=0) def __init__(self, base_dir: Optional[str] = None): self.base_dir = base_dir or tempfile.gettempdir() def create(self, spec): return LocalSandbox(root=tempfile.mkdtemp(prefix="sandboxkit-", dir=self.base_dir), env=dict(spec.env)) ``` Hmm, spec.env: apply env to every exec? Or once at create? For local, apply per-exec merged over os.environ. Fine. ```python class LocalSandbox(Sandbox): def __init__(self, root, env): self._root = os.path.realpath(root) self._id = os.path.basename(root) self._env = dict(env) self._alive = True id, provider properties def _resolve(self, path): if not path.startswith("/"): path = "/" + path # treat as sandbox-relative? or reject? full = os.path.realpath(os.path.join(self._root, path.lstrip("/"))) # ensure within root if full != self._root and not full.startswith(self._root + os.sep): raise PathOutsideSandbox(path) return full ``` Wait — realpath of join(root, relpath). If path contains "../" it can escape; realpath resolves then we check containment. Also symlink escape caught by realpath. Good. For non-existent paths realpath still works (doesn't raise), so containment check still valid. Path convention: API paths are absolute POSIX within sandbox; "/" maps to root. I'll normalize: strip leading "/", join, realpath, check containment. If path is "/", full == root; containment check `full != root and not startswith(root+sep)` — wait, if full == root it's allowed (that's "/"). If full == root then it passes (first condition false → raise not triggered). Check logic: raise if `full != root and not full.startswith(root + os.sep)`. Correct. exec: ```python def exec(self, command, *, cwd="/", env=None, timeout=None): if not self._alive: raise SandboxNotFound(...) merged = {**os.environ, **self._env, **(env or {})} cwd_path = self._resolve(cwd) start = time.monotonic() try: proc = subprocess.run(command, shell=True, cwd=cwd_path, env=merged, capture_output=True, text=True, timeout=timeout) except subprocess.TimeoutExpired as e: raise SandboxTimeout(f"...") from e return ExecResult(exit_code=proc.returncode, stdout=proc.stdout or "", stderr=proc.stderr or "", duration_ms=int((time.monotonic()-start)*1000)) ``` text=True may fail on non-UTF8 bytes; use errors="replace"? subprocess text uses locale encoding; safer: capture bytes then decode with errors="replace". Let me do that for robustness: ```python proc = subprocess.run(command, shell=True, cwd=..., env=..., capture_output=True, timeout=timeout) stdout = proc.stdout.decode("utf-8", "replace") ``` TimeoutExpired has .stdout/.stderr partials; for simplicity raise SandboxTimeout with those decoded if available. Keep simple: raise with message; note partial output available on the exception. Fine, small scope. fs ops: ```python def read_file(self, path): full = self._resolve(path) if os.path.isdir(full): raise SandboxError(f"{path} is a directory") with open(full, "rb") as f: return f.read() def write_file(self, path, data): full = self._resolve(path) os.makedirs(os.path.dirname(full), exist_ok=True) # auto-create parents — nice primitive behavior if isinstance(data, str): data = data.encode() with open(full, "wb") as f: f.write(data) def list_files(self, path="/"): full = self._resolve(path) if not os.path.exists(full): raise SandboxNotFound(path) out = [] for entry in os.scandir(full): st = entry.stat() rel = "/" + os.path.relpath(entry.path, self._root).replace(os.sep, "/") out.append(FileInfo(path=rel, name=entry.name, is_dir=entry.is_dir(), size=st.st_size, modified_at=st.st_mtime)) return sorted(out, key=lambda f: f.path) def delete_file(self, path): full = self._resolve(path) if os.path.isdir(full) and not os.path.islink(full): # remove empty dirs only? primitive "delete_file" -> unlink; add delete for dirs? os.rmdir(full)? or shutil.rmtree? else: os.unlink(full) ``` Hmm — primitive set: keep `delete_file` for files, raise on non-empty dir with hint to exec `rm -rf`? Or just rmtree? For a primitive file API, I'll unlink files and `rmdir` empty dirs, and raise a clear error for non-empty dirs suggesting `exec("rm -rf ...")`. That keeps primitives honest and code small. Hmm, or provide `delete` that handles both (files and dirs) — ergonomics beat purity for an SDK. Many SDKs have `files.remove(path, recursive?)`. I'll do `delete_file(path, *, recursive=False)`. Small addition, good UX. Wait — but then the Sandbox ABC signature: `delete_file(self, path: str, *, recursive: bool = False)`. Fine. mkdir: `os.makedirs(full, exist_ok=True)`. Should mkdir create parents? "mkdir -p" behavior is friendlier; note it. stop: `shutil.rmtree(self._root, ignore_errors=True); self._alive = False`. is_alive override for local: just `self._alive` (no need to spawn). Actually the base `is_alive` runs "true" — fine too but cheaper to override. I'll override. **providers/docker.py** ```python import io, os, posixpath, tarfile, time import docker # pip install docker from ..core import ... from ..provider import register, Provider @register("docker") class DockerProvider(Provider): info = ProviderInfo(name="docker", capabilities=frozenset({"exec", "fs", "network"}), cost_tier=1) def __init__(self, image="ubuntu:24.04", base_url=None, network=None, pull=False): kwargs = {} if base_url: kwargs["base_url"] = base_url self.client = docker.DockerClient.from_env(**kwargs) # hmm: from_env accepts base_url kw self.image = image self.network = network if pull: self.client.images.pull(image) def health(self): try: self.client.ping(); return True except Exception: return False def create(self, spec): image = spec.image or self.image try: container = self.client.containers.run( image, command="sleep infinity", detach=True, auto_remove=True, network=self.network if spec.needs_network... ) except docker.errors.ImageNotFound as e: raise ProviderError(...) from e return DockerSandbox(container, env=spec.env) ``` Wait: `client.containers.run(image, "sleep infinity", detach=True, auto_remove=True)` — with detach, auto_remove applies when container stops. Good. Environment: pass at create: `environment=dict(spec.env)` — docker supports env per container. Then exec env merges additionally. Simpler: pass spec.env at container creation, and also allow per-exec env merged into exec environment. exec_run supports `environment={...}` per exec too. I'll pass spec.env at create and per-exec env via exec_run's environment param. Network needs: if "network" in spec.needs, ensure not run with network disabled; default docker has network. Could support `network="none"` when needs lacks network — nice touch but optional. Skip for small scope; mention as note. DockerSandbox: ```python class DockerSandbox(Sandbox): def __init__(self, container, env): self._c = container self._id = container.id self._env = dict(env) @property def id(self): return self._id[:12] def exec(self, command, *, cwd="/", env=None, timeout=None): merged = {**self._env, **(env or {})} try: code, (out, err) = self._c.exec_run( ["bash", "-lc", command], workdir=None if cwd == "/" else cwd, environment=merged, demux=True, ) except Exception as e: ... ``` Hmm — timeout: docker-py exec_run doesn't support timeout directly; low-level API has timeout on socket? `exec_run` has no timeout param. Options: run via low-level `exec_create` + `exec_start(socket=False, timeout=...)`? The API client `exec_start` accepts `timeout` (http read timeout). Actually `container.client.api.exec_create(...)` then `client.api.exec_start(exec_id, timeout=...)`? The APIClient.exec_start signature: `exec_start(exec_id, detach=False, tty=False, stream=False, socket=False, demux=False)` — no timeout, but underlying HTTP request uses client timeout (default 60s). We can pass custom `client = docker.from_env(timeout=...)`. Simplest honest approach: note that Docker provider doesn't support per-exec timeouts precisely; emulate via `timeout` command wrapper: `["bash", "-lc", f"timeout {t}s {shlex.quote...}"]` — messy. Alternative: accept limitation and raise SandboxTimeout if elapsed > timeout after the fact? Wrong semantics (command keeps running). Cleanest small-scope approach: use `coreutils` `timeout` inside the container when timeout is given: `bash -lc "timeout --signal=KILL 30s -- bash -lc ''"` — quoting hell. Use shlex.quote on the inner command: ```python inner = command if timeout is not None: inner = f"timeout -k 1 {timeout} bash -lc {shlex.quote(command)}" code, streams = self._c.exec_run(["bash", "-lc", inner], ...) ``` Hmm wait, if we wrap, then the outer bash runs `timeout ... bash -lc `. That preserves semantics decently (signals, exit code 124/137 on timeout). Then detect exit code 124 or 137 → raise SandboxTimeout. Slight false-positive risk if user command itself exits 124, acceptable for small scope with a note. Alternatively check `code in (124, 137)` and stderr containing "Killed"? Keep simple: treat 124 as timeout (124 is GNU timeout's dedicated exit code; 137 is SIGKILL). I'll check both and note the caveat. Also keep image must contain GNU coreutils `timeout` — ubuntu:24.04 does. Note it. Actually — even simpler and more honest: document that per-exec timeout in Docker provider is best-effort via GNU `timeout`. That's fine. Workdir: exec_run workdir must exist in container; default "/" exists. Pass `workdir=cwd if cwd != "/" else None`? Hmm passing "/" should be fine actually. `workdir="/"`. But some images... fine, pass as-is. Note: cwd must exist, else exec fails with exit code (exec_run returns exit_code 126 or similar? It returns error). Keep: pass workdir=cwd directly. Hmm, if workdir doesn't exist docker returns exit_code 126? It may raise. I'll catch docker.errors.APIError → ProviderError. Note behavior. File ops via tar: ```python def write_file(self, path, data): if isinstance(data, str): data = data.encode() path = posixpath.normpath("/" + path) parent = posixpath.dirname(path) name = posixpath.basename(path) buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: info = tarfile.TarInfo(name=name) info.size = len(data) info.mode = 0o644 info.mtime = int(time.time()) tf.addfile(info, io.BytesIO(data)) self._c.put_archive(parent or "/", buf.getvalue()) ``` Note: put_archive cannot create missing intermediate dirs — but we can pre-create them via exec mkdir? That adds a round trip; or embed parent dirs in the tar! A tar can contain directories entries: put_archive extracts all entries. So write_file can add TarInfo dir entries for each intermediate path segment. Nice trick: ```python def write_file(self, path, data): path = posixpath.normpath("/" + path) parts = path.strip("/").split("/") # entries: dirs for all but last, then file ``` But put_archive's `path` argument is where the tar is extracted. If tar contains "a/b/c.txt" entries with dirs, and we put_archive("/", tar), it creates /a, /a/b, /a/b/c.txt. So: put_archive("/", tar with full relative path entries). Edge: path "/" itself invalid. Let me implement that way — simpler mental model and auto-creates parents: ```python def write_file(self, path, data): if isinstance(data, str): data = data.encode("utf-8") rel = posixpath.normpath(path).lstrip("/") if not rel: raise SandboxError("cannot write '/'") # path traversal check: after normpath, no ".." segments if any(seg == ".." for seg in rel.split("/")): raise PathOutsideSandbox(path) buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: # parent dirs acc = [] for seg in rel.split("/")[:-1]: acc.append(seg) di = tarfile.TarInfo("/".join(acc)) di.type = tarfile.DIRTYPE di.mode = 0o755 tf.addfile(di) fi = tarfile.TarInfo(rel) fi.size = len(data); fi.mode = 0o644 tf.addfile(fi, io.BytesIO(data)) self._c.put_archive("/", buf.getvalue()) ``` Hmm: TarInfo dir entries need name without trailing slash (tarfile handles). Also empty dirs in tar need `di.size = 0` default; addfile without fileobj for dirs works (tarfile writes zeros). Yes: `tf.addfile(di)` without fileobj writes zero-length? For dir type, tarfile writes headers; getsize 0. OK. Actually for DIRTYPE, tarfile.addfile(ti) without fileobj writes header with size 0 — correct. read_file via get_archive: ```python def read_file(self, path): path = posixpath.normpath("/" + path) stream, stat = self._c.get_archive(path) raw = b"".join(stream) with tarfile.open(fileobj=io.BytesIO(raw), mode="r") as tf: members = tf.getmembers() # member names are relative (basename-based) file_members = [m for m in members if m.isfile()] if not file_members: raise SandboxError(f"{path} is not a regular file") return tf.extractfile(file_members[0]).read() ``` get_archive: returns (data: generator of tar bytes, stat: dict). The tar contains the path's *basename* at root (actually the last component). If we request a dir, members include dir and children; we only take first regular file — for read_file that's fine (dirs rejected). If requested file is a symlink? extractfile follows within archive... edge case, skip. Path validation: check for ".." segments after normpath — normpath("/a/../b") = "/b" so normpath already collapses; a result containing ".." can't happen after normpath? posixpath.normpath("/a/../../b") → "/b". Normpath resolves ".." lexically at root. So after normpath, no ".." remains — good, validation is automatic. Keep a defensive check anyway? It's harmless; I'll keep the normpath and skip redundant check. Actually keep one-liner check for clarity. Eh — normpath handles it; skip. Hmm, but paths like "/../etc/passwd" → normpath("/../etc/passwd") = "/etc/passwd" — that's inside container root anyway, fine, containers don't have our jail. Docker sandbox = whole container FS is the sandbox, so no jail needed. Right — local provider needs the jail because it shares host FS; docker doesn't (container boundary is the jail). Worth noting in comments. list_files: ```python def list_files(self, path="/"): path = posixpath.normpath("/" + path) stream, stat = self._c.get_archive(path) raw = b"".join(stream) entries = [] base = posixpath.dirname(path) # tar members are relative to this with tarfile.open(...) as tf: for m in tf.getmembers(): # member name relative; skip the entry for the dir itself when listing a dir full = posixpath.join(base, m.name) if m.name != posixpath.basename(path) or m.isdir()... ``` Hmm, this gets confusing. When you get_archive("/data"), the tar contains a member named "data" (the dir) plus children "data/x.txt". When you get_archive("/data/x.txt"), tar contains "x.txt". So mapping member names back to sandbox paths: if we archived path P, members are relative to dirname(P). So sandbox path = posixpath.join(posixpath.dirname(P), m.name). For P="/data": dirname="/" → join("/", "data") = "/data" (dir itself), "data/x.txt" → "/data/x.txt". For list_files we want only children, excluding P itself: skip member whose sandbox path == P and is dir. Let me write: ```python def list_files(self, path="/"): p = posixpath.normpath("/" + path) if p != "/": stream, _ = self._c.get_archive(p) ... ``` Edge: get_archive("/") — archiving root of container? That would tar the entire container filesystem — bad! Handle path == "/" specially: docker exec `ls` instead? Or disallow "/" listing? Hmm. Using exec for listing would be simpler and uniform: `ls -1A`? But then we don't get sizes/types easily — `ls -lA --time-style=+` parsing... ugh. Alternative: for list_files("/") in docker, get_archive with path "/." trick? get_archive("/.")... the API path would be "/" + "/." — docker archives "/"? That's the whole FS; big but works? It'd be slow/huge. Better: for "/", list top-level via `exec("ls -1A /")`? Mixed approach is inelegant. Cleaner: implement list_files via exec `find` or `ls` parsing? Parsing `ls` is fragile. What about `stat -c`? Also fragile across images. Pragmatic: use get_archive for everything except "/", where we use a small exec-based fallback: `exec("ls -1A --quoting-style=escape /")` and return FileInfo with just names (size=0, is_dir unknown → maybe detect by trailing slash with `ls -1Ap /`: appends "/" for dirs!). `ls -1Ap /` gives names with trailing slash for dirs — parseable: `name.rstrip('/')`, `is_dir = name_orig.endswith('/')`. But then is_dir/size semantics differ between "/" and other paths. Acceptable for small scope with a comment; note that a production version would use a single mechanism. Alternatively: always use the ls -1Ap approach for ALL dirs (uniform!), and use get_archive only for read_file. That's uniform: list_files = exec("ls -1Ap {quoted_path}"). Hmm but then list_files depends on ls being present (ubuntu: yes; alpine: busybox ls supports -1A? busybox ls has -A and -1; -p? busybox ls supports -p? I believe busybox ls supports -p (append / to dirs) — not sure. Also --quoting-style not in busybox. Names with spaces: `ls -1A` outputs one per line, spaces fine; newlines in filenames break it (acceptable caveat). Trade-off: get_archive-based listing gives exact types/sizes without shell parsing, but is archivally heavy and breaks for "/". ls-based is light and uniform but fragile to exotic names and needs ls. Given "primitive commands" SDK where exec is available anyway, and simplicity: I'll use get_archive for non-root paths (rich metadata), and for root "/" use `ls -1Ap` fallback. Hmm, that's two code paths — more code. Or: just use `ls -1Ap` everywhere for list_files. One code path, simple, caveat documented. For small scope I prefer one path. Let me do ls-based: ```python def list_files(self, path="/"): p = posixpath.normpath("/" + path) res = self.exec(f"ls -1Ap --time-style=long-iso -l {shlex.quote(p)}") ``` Ugh, if I want size/mtime I need `ls -l` parsing: `ls -1lAp` columns: perms, links, owner, group, size, date..., name. Parsing `ls -l` reliably = pain. Decision: keep list_files simple — name + is_dir only for docker (size=0, modified_at=0), via `ls -1Ap`. For local, full stat info. Document that FileInfo fields may be zero/unknown per provider. That's an honest trade-off of unified APIs: the *intersection* of provider abilities defines the guaranteed contract; extras are best-effort. Actually that's an important design point to state explicitly: **the abstract contract is the intersection; capabilities declare extras.** Hmm, but is_dir matters — ls -1Ap gives trailing slash → is_dir. Good. Wait, actually — maybe use get_archive for dirs (it gives me all metadata cleanly) and only special-case "/" via ls. You know, I keep going back and forth; let me decide by simplicity-of-code: get_archive path: ~10 lines, gives size/is_dir/mtime for free, works for any dir except "/" (which I handle with an ls fallback ~8 lines). Total ~18 lines + complexity. ls path: ~8 lines total, uniform, but weak metadata for docker provider always. I'll go with get_archive + root fallback. It showcases proper docker API usage and gives better metadata; the root fallback is short. Also `get_archive` on a *file* would also work but read_file already uses get_archive... fine. Hmm wait, one more wrinkle: get_archive for a dir returns the whole subtree recursively — if user does list_files("/data") on a huge dir, we tar everything. That's bad for large dirs. ls doesn't have that problem. Ugh. For a *basic* SDK it's acceptable with a caveat note. OK final: get_archive with root fallback; note the caveat "lists recursively-tar the dir — fine for small dirs; production would stream or use ls". Hmm, that caveat is a bit embarrassing... Actually you know what — use `ls` everywhere. It's simpler, one code path, avoids the tar-everything trap, and I'll set size/modified_at to 0.0 with a comment. Basic primitive = list names + dir/file flag. Good. Wait, but is_dir via trailing slash and names with newline edge... fine, documented. DECISION: ls -1Ap everywhere in docker provider. Moving on. Hmm, hmm. Let me sanity check busybox/`ubuntu` — ubuntu:24.04 has GNU ls supporting -1 -A -p. Fine. I'll also guard: use `ls -1Ap --` and shlex.quote the path. Actually one more idea: `ls -1Ap --full-time`? No, keep plain. Alright — docker list_files: ```python def list_files(self, path="/"): p = posixpath.normpath("/" + path) res = self.exec(f"ls -1Ap {shlex.quote(p)}") if res.exit_code != 0: raise SandboxNotFound(path) if "No such file" in res.stderr else ExecFailed-ish out = [] for line in res.stdout.splitlines(): if not line: continue is_dir = line.endswith("/") name = line.rstrip("/") out.append(FileInfo(path=posixpath.join(p, name), name=name, is_dir=is_dir, size=0, modified_at=0.0)) return out ``` Hmm "No such file or directory" check — message is "ls: cannot access '...': No such file or directory". Raise SandboxNotFound. Else ExecFailed. Need to import ExecFailed. OK. Note: `ls -1Ap /` includes "." and ".."? -A excludes "." and ".." (that's -a vs -A). -A = all except . and .. Good. delete_file for docker: exec `rm -rf`? Using exec for delete is pragmatic: `rm -rf -- ` with quoting. But "primitive" file API implemented via exec — acceptable and honest (that's what many SDKs do internally). Or get_archive/unlink impossible in docker API; there's no delete API. So exec it is: ```python def delete_file(self, path, *, recursive=False): p = posixpath.normpath("/" + path) args = "rm -rf" if recursive else "rm -r"? ``` Hmm, local delete: file→unlink, dir→rmdir unless recursive→rmtree. For docker: `rm -f file`, `rm -r -f dir`? Non-recursive dir delete fails if non-empty (rmdir semantics): use `rmdir` for dir non-recursive? Simpler: if recursive: `rm -rf -- p` else `rm -f -- p`... `rm -f` on a dir errors ("cannot remove: Is a directory") which matches "non-empty/not-dir" semantics roughly. Local raises a clear error for non-empty non-recursive. To keep parity, docker: recursive → `rm -rf -- p`; non-recursive → `rm -d -- p`? `rm -d` removes empty dirs and files? `rm -d` = unlink dirs if empty (GNU). `rm -f -- p` for files; for empty dir it errors without -d. Hmm. Use: non-recursive → `rm -f -- p` and if stderr contains "Is a directory" raise SandboxError("use recursive=True")? Or just `rm -d -- p` which handles both file and empty dir. I'll do: recursive → `rm -rf -- p`, else → `rm -d -- p`, then check exit code; if failed and it's a dir-nonempty case raise helpful error. Simplify: ```python flag = "-rf" if recursive else "-d" res = self.exec(f"rm {flag}f? ...") ``` `rm -d` also needs -f? `rm -d p` removes empty dir. `rm -d -f` suppresses errors. I want errors surfaced. Let me: recursive: `rm -rf -- p`; else: `rm -d -- p`; if exit != 0 raise ExecFailed (its stderr will say why). For parity, local non-recursive dir non-empty raises error too. Good enough. mkdir docker: exec `mkdir -p -- p`. Local: os.makedirs(exist_ok=True). Parity ✓. stop docker: `self._c.stop(timeout=10)` then with auto_remove=True it's removed. Docker SDK stop has timeout param (seconds to wait before SIGKILL). If container already gone → docker.errors.NotFound → ignore. is_alive docker: check `self._c.reload(); self._c.status == "running"`? Cheaper than exec. Override: ```python def is_alive(self): try: self._c.reload() return self._c.status == "running" except docker.errors.NotFound: return False ``` Also guard exec on stopped container: exec_run raises APIError if not running — catch and raise SandboxError. Fine. **E2B sketch** — I decided local+docker as the two, E2B as extension. Let me draft a compact E2B adapter for the "adding a provider" section (~35-45 lines): ```python # providers/e2b.py — pip install e2b (set E2B_API_KEY) @register("e2b") class E2BProvider(Provider): info = ProviderInfo(name="e2b", capabilities=frozenset({"exec", "fs", "network", "long-running"}), cost_tier=2) def __init__(self, template="ubuntu-24.04", timeout=300): self.template, self.timeout = template, timeout def create(self, spec): from e2b import Sandbox as E2BSandbox # lazy import sbx = E2BSandbox(template=self.template, timeout=(spec.ttl_seconds or self.timeout)) return E2BAdapter(sbx) class E2BAdapter(Sandbox): def __init__(self, sbx): self._s = sbx @property def id(self): return self._s.sandbox_id @property def provider(self): return "e2b" def exec(self, command, *, cwd="/", env=None, timeout=None): r = self._s.commands.run(command, cwd=cwd or None, envs=dict(env or {}), timeout=timeout) return ExecResult(r.exit_code, r.stdout, r.stderr, 0) def read_file(self, path): return self._s.files.read(path) def write_file(self, path, data): if isinstance(data, str): data = data.encode() self._s.files.write(path, data) def list_files(self, path="/"): return [FileInfo(path=f.path, name=f.name, is_dir=f.is_dir, size=0, modified_at=0.0) for f in self._s.files.list(path)] def delete_file(self, path, *, recursive=False): self._s.files.remove(path) def mkdir(self, path): self.exec(f"mkdir -p {shlex.quote(path)}") def stop(self, force=False): self._s.remove() ``` Check e2b API from memory (v1.x): `Sandbox(template=..., timeout=...)`? In v1: `from e2b import Sandbox; sbx = Sandbox(template="ubuntu-24.04", timeout=300)` — I believe `Sandbox()` constructor creates synchronously (or `Sandbox.create()`). In e2b v1.0, `Sandbox(...)` still works as sync constructor? They refactored: `Sandbox.create()` returns new sync sandbox; `AsyncSandbox.create()`. And older `Sandbox("template-id")` positional was deprecated... The class is constructed with `Sandbox(template=..., timeout=...)`? Hmm. In e2b 1.x docs: `sbx = Sandbox.create()` and `sbx = Sandbox.create(template="ubuntu-24.04")`. Yes, `Sandbox.create` is the documented path (keepalive, metadata kwargs etc.). I'll use `Sandbox.create(...)` and mark it "verify against current e2b docs, API moves fast". `commands.run(cmd, cwd=..., envs=..., timeout=...)` — envs param name: I think `envs: dict[str,str]`. CommandResult has `.stdout`, `.stderr`, `.exit_code`. `files.write(path, data)` accepts bytes|str|Buffer. `files.read(path)` returns bytes (format="bytes" default). `files.list(path)` returns list of `FileInfo` with name/path/is_dir. `files.remove(path)`. `sbx.remove()`. Good — reasonably confident. Since I'm presenting it as an extension sketch, I'll add a note to double-check names against the installed version. **Config example:** ```yaml providers: - type: docker priority: 20 options: image: ubuntu:24.04 - type: local priority: 0 # routing is automatic: try docker first (higher priority), fall back to local. ``` `from_dict`: ```python @classmethod def from_dict(cls, cfg: dict) -> "SandboxKit": entries = [] for p in cfg.get("providers", []): entries.append(ProviderConfig( type=p["type"], options=p.get("options", {}), priority=p.get("priority", 0), enabled=p.get("enabled", True), )) return cls(entries) ``` And yaml: ```python @classmethod def from_yaml(cls, path): import yaml return cls.from_dict(yaml.safe_load(open(path))) ``` Note optional dependency. **Usage example:** ```python from sandboxkit import SandboxKit kit = SandboxKit.auto() # docker if available, else local with kit.sandbox(image="ubuntu:24.04", needs={"network"}) as sbx: res = sbx.exec("python3 --version") sbx.write_file("/work/hello.py", "print('hi from the sandbox')") out = sbx.run("python3 /work/hello.py") print(sbx.read_file("/work/hello.py").decode()) for f in sbx.list_files("/work"): print(f.name, f.is_dir) ``` Also showing explicit routing pin: `kit.sandbox(provider="local")`. `auto()`: ```python @classmethod def auto(cls) -> "SandboxKit": entries = [] d = ProviderConfig(type="docker", options={}, priority=20) l = ProviderConfig(type="local", options={}, priority=0) return cls([d, l]) # router skips unhealthy providers ``` Router should call provider.health() during routing (cache it briefly? keep simple: check each time — cheap for docker ping, free for local). **Router implementation:** ```python class Router: def __init__(self, configs: Sequence[ProviderConfig]): self._configs = list(configs) self._instances: dict[str, Provider] = {} def _instance(self, cfg) -> Provider: if cfg.type not in self._instances: cls = get_provider_class(cfg.type) self._instances[cfg.type] = cls(**cfg.options) return self._instances[cfg.type] def route(self, spec: SandboxSpec) -> list[Provider]: if spec.provider: pin = ... find matching config; must include; return [instance] candidates = [] for cfg in self._configs: if not cfg.enabled: continue p = self._instance(cfg) if spec.needs and not spec.needs <= p.info.capabilities: continue if spec.region and p.info.regions and spec.region not in p.info.regions: continue if not p.health(): continue candidates.append((self._score(p, cfg, spec), p)) candidates.sort(key=lambda t: -t[0]) return [p for _, p in candidates] def _score(self, p, cfg, spec): s = cfg.priority * 10 if spec.region and spec.region in p.info.regions: s += 5 s -= p.info.cost_tier # prefer cheaper on ties return s ``` Plus failover note: kit tries ranked list. Explicit pin: if spec.provider given, only that one (even if unhealthy? try anyway, raise if fails). Also `route` raising NoRouteError if empty (kit collects per-provider errors for message). `needs` as frozenset; comparing `spec.needs <= p.info.capabilities` — both frozensets. Note: sandbox spec.needs default frozenset() — `frozenset() <= anything` is True. Good. Hmm — SandboxSpec is frozen dataclass; kit.sandbox(**kwargs) builds spec: ```python def sandbox(self, spec: SandboxSpec | None = None, **kw) -> Sandbox: if spec is None: spec = SandboxSpec(**{k: v for k, v in kw.items()}) elif kw: spec = dataclasses.replace(spec, **kw) ...route & create with failover... ``` Validate kw keys are valid SandboxSpec fields → TypeError from dataclass, fine. Failover loop: ```python attempts = [] for provider in self.router.route(spec): try: return provider.create(spec) except SandboxError as e: attempts.append(f"{provider.info.name}: {e}") raise NoRouteError("no provider could create a sandbox: " + "; ".join(attempts)) ``` Catch SandboxError broadly for failover? Provider.create may raise various; catching SandboxError catches our hierarchy. docker errors wrapped as ProviderError inside create. Local create rarely fails. Good. **Tests sketch** — parametrized over providers with docker skipif: ```python import pytest from sandboxkit import SandboxKit kit = SandboxKit.from_dict({"providers": [{"type": "local"}]}) no_docker = pytest.mark.skipif(not _docker_ok(), reason="docker not available") def test_roundtrip(kit): with kit.sandbox() as s: s.write_file("/tmp/a.txt", "hello") assert s.read_file("/tmp/a.txt") == b"hello" assert s.run("cat /tmp/a.txt") == "hello\n" assert [f.name for f in s.list_files("/tmp")] == ["a.txt"] def test_exec_env_cwd... def test_timeout... def test_path_escape_local... ``` Provide a small conftest with fixtures parametrizing kit over providers. Keep the test section compact — show a representative subset. **What to explain in prose (structure section):** - Layered architecture ASCII diagram: ``` ┌─────────────────────────────┐ │ your app / agents │ └──────────────┬──────────────┘ │ kit.sandbox(...) → Sandbox handle ┌──────────────▼──────────────┐ │ SandboxKit (facade) │ failover, context mgmt, conveniences ├──────────────────────────────┤ │ Router │ capability match → score → ranked list ├──────────────────────────────┤ │ Provider registry │ name → Provider class │ ┌─────────┐ ┌─────────┐ │ │ │ docker │ │ local │ │ (add e2b/modal/daytona here) │ └─────────┘ └─────────┘ │ └──────────────────────────────┘ ``` - Key abstractions table: Provider / Sandbox / SandboxSpec / ExecResult / FileInfo / Router / registry. - Contract notes: paths absolute POSIX; bytes; intersection semantics; errors taxonomy; sync-first with note on async. - Directory tree. - Code. - How routing decisions are made (explicit pin > capability filter > region > priority score > failover). - Extension guide: add provider = subclass Provider + Sandbox, register, add config line. E2B sketch. - Next steps: async interface (anyio), streaming exec (generator of chunks), process management primitives (`start_process` returning PID handle), snapshots, usage metering, retries/idempotency, connection pooling. Also mention why primitives chosen: exec + fs cover ~everything (e.g., "run python script" = write_file + exec); keep the abstract surface small so adapters stay thin. Length/format: This is a build request → long response with code is appropriate. Use headers, a diagram, code blocks with file paths. Keep prose tight; let code carry. Avoid over-explaining obvious things. I should make sure each code file is complete and consistent (imports!). Let me now write the final code mentally with care for import correctness. Package: `sandboxkit/` **`sandboxkit/core.py`**: ```python """Core abstractions: errors, result types, and the Sandbox interface. Contract notes: - All filesystem paths are absolute POSIX paths *inside* the sandbox ("/" = sandbox root). - File payloads are bytes; providers/decide encoding. - The abstract Sandbox interface is the *intersection* of what all providers can do; provider-specific extras live on the concrete provider or are declared via capabilities. """ from __future__ import annotations import dataclasses from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Mapping, Optional # -------------------------------------------------------------------------- # Errors # -------------------------------------------------------------------------- class SandboxError(Exception): """Base class for every error raised by sandboxkit.""" class ProviderError(SandboxError): """Provider-level failure: auth, quota, image pull, infrastructure.""" class NoRouteError(SandboxError): """No provider satisfies the request (or all attempts failed).""" class SandboxNotFound(SandboxError): """Sandbox or path does not exist.""" class SandboxTimeout(SandboxError): """Operation exceeded its deadline.""" class PathOutsideSandbox(SandboxError): """Path escapes the sandbox root (local provider jail).""" class ExecFailed(SandboxError): """Command returned non-zero exit code (raised by Sandbox.run).""" def __init__(self, result: "ExecResult"): tail = (result.stderr or result.stdout).strip() super().__init__( f"exit code {result.exit_code}" + (f": {tail[:200]}" if tail else "") ) self.result = result # -------------------------------------------------------------------------- # Result types # -------------------------------------------------------------------------- @dataclass(frozen=True) class ExecResult: exit_code: int stdout: str stderr: str duration_ms: int @dataclass(frozen=True) class FileInfo: path: str # absolute POSIX path inside the sandbox name: str is_dir: bool size: int # 0 when unknown (provider-dependent) modified_at: float # epoch seconds, 0.0 when unknown @dataclass(frozen=True) class SandboxSpec: """What you want out of a sandbox. Used for routing and creation.""" image: Optional[str] = None env: Mapping[str, str] = field(default_factory=dict) labels: Mapping[str, str] = field(default_factory=dict) needs: frozenset = frozenset() # capability names, e.g. {"network"} region: Optional[str] = None ttl_seconds: Optional[int] = None timeout_seconds: Optional[int] = None # default per-exec timeout provider: Optional[str] = None # pin to a specific provider, skip routing ``` Hmm `needs: frozenset = frozenset()` — plain default with immutable type is fine in dataclass (frozenset is immutable so no mutable-default error). **`sandboxkit/provider.py`** — Provider ABC + registry: ```python """Provider interface + registry. A Provider creates sandboxes; a Sandbox is a live handle. Routing selects providers, never concrete sandboxes.""" from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Optional from .core import Sandbox, SandboxSpec @dataclass(frozen=True) class ProviderInfo: name: str capabilities: frozenset # what this provider can do, e.g. {"exec", "fs", "network"} regions: tuple = () # empty = location-agnostic cost_tier: int = 1 # lower = cheaper; tiebreaker in routing class Provider(ABC): """A factory for sandboxes. Must be cheap to instantiate and side-effect free.""" info: ProviderInfo @abstractmethod def create(self, spec: SandboxSpec) -> Sandbox: """Provision a sandbox. Raise SandboxError subclasses on failure so routing can fail over.""" def health(self) -> bool: """Cheap liveness check used by the router. Override if you can do better.""" return True # ------------------------- registry ------------------------- _REGISTRY: dict = {} def register(name: str): """Class decorator: registers a Provider under a routing name.""" def deco(cls): if name in _REGISTRY: raise ValueError(f"provider {name!r} already registered") _REGISTRY[name] = cls return cls return deco def get_provider(name: str) -> type: try: return _REGISTRY[name] except KeyError: raise ValueError(f"unknown provider {name!r}; known: {sorted(_REGISTRY)}") from None def provider_names(): return sorted(_REGISTRY) ``` **`sandboxkit/router.py`**: ```python """Capability-driven routing: given a SandboxSpec, produce a ranked list of providers. Routing is intentionally boring: filter -> score -> sort.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Optional from .core import SandboxSpec, SandboxError from .provider import Provider, get_provider @dataclass class ProviderConfig: type: str # registry name, e.g. "docker" options: dict = field(default_factory=dict) priority: int = 0 # higher = preferred enabled: bool = True class Router: def __init__(self, configs): self._configs = list(configs) self._instances: dict = {} # -- lifecycle ---------------------------------------------------------- def _instance(self, cfg: ProviderConfig) -> Provider: if cfg.type not in self._instances: self._instances[cfg.type] = get_provider(cfg.type)(**cfg.options) return self._instances[cfg.type] def close(self): for inst in self._instances.values(): closer = getattr(inst, "close", None) if closer: closer() # -- routing ------------------------------------------------------------ def route(self, spec: SandboxSpec) -> list: """Return providers that can satisfy `spec`, best first.""" if spec.provider: # explicit pin skips all logic p = self._instance(self._config_of(spec.provider)) return [p] ranked = [] for cfg in self._configs: if not cfg.enabled: continue p = self._instance(cfg) if spec.needs and not spec.needs <= p.info.capabilities: continue # missing a required capability if spec.region and p.info.regions and spec.region not in p.info.regions: continue # region constraint if not p.health(): continue # dead provider, try next ranked.append((self._score(p, cfg, spec), p)) ranked.sort(key=lambda pair: -pair[0]) return [p for _, p in ranked] def _config_of(self, name: str) -> ProviderConfig: for cfg in self._configs: if cfg.type == name: return cfg raise SandboxError(f"provider {name!r} is not configured") @staticmethod def _score(p: Provider, cfg: ProviderConfig, spec: SandboxSpec) -> int: score = cfg.priority * 10 if spec.region and spec.region in p.info.regions: score += 5 # region match bonus score -= p.info.cost_tier # prefer cheaper on ties return score ``` **`sandboxkit/api.py`** — facade: ```python """Public facade. SandboxKit is the only class most applications touch.""" from __future__ import annotations import dataclasses from typing import Optional from .core import Sandbox, SandboxError, SandboxSpec, NoRouteError, ProviderError from .provider import provider_names # noqa: F401 (helps error messages) from .router import ProviderConfig, Router # import built-in providers so their @register decorators run from .providers import local as _local # noqa: F401 from .providers import docker as _docker # noqa: F401 class SandboxKit: def __init__(self, providers, *, failover: bool = True): self.router = Router(providers) self.failover = failover # ---------------- factories ---------------- @classmethod def from_dict(cls, cfg: dict) -> "SandboxKit": providers = [ ProviderConfig( type=p["type"], options=p.get("options", {}), priority=p.get("priority", 0), enabled=p.get("enabled", True), ) for p in cfg.get("providers", []) ] return cls(providers) @classmethod def from_yaml(cls, path) -> "SandboxKit": import yaml # optional dependency: pip install pyyaml with open(path) as f: return cls.from_dict(yaml.safe_load(f) or {}) @classmethod def auto(cls) -> "SandboxKit": """Sensible default: docker if installed, local always as fallback.""" return cls([ ProviderConfig(type="docker", priority=10), ProviderConfig(type="local", priority=0), ]) # ---------------- the one method you call ---------------- def sandbox(self, spec: Optional[SandboxSpec] = None, **overrides) -> Sandbox: """Open a sandbox, routed automatically. Overrides are SandboxSpec fields: kit.sandbox(image="ubuntu:24.04", needs={"network"}, provider="local") """ if spec is None: spec = SandboxSpec(**overrides) elif overrides: spec = dataclasses.replace(spec, **overrides) attempts = [] for provider in self.router.route(spec): try: return provider.create(spec) except SandboxError as e: if not self.failover: raise attempts.append(f"{provider.info.name}: {e}") raise NoRouteError( "could not create a sandbox (providers tried: " + (", ".join(attempts) or "none matched the request") + ")" ) ``` Wait: NoRouteError import — from .core import NoRouteError. Adjust imports. Also when failover False and first create fails, we raise that original error — good. Note: ProviderError imported but unused — drop it. `provider_names` unused — drop or keep for error msg? Drop. **`sandboxkit/providers/__init__.py`**: just make it a package (maybe empty or import submodules? If api imports them explicitly, empty is fine). **`sandboxkit/providers/local.py`**: ```python """Local provider: sandboxes are temp directories on the host, commands run via subprocess. Zero dependencies, no isolation beyond a directory jail — great for tests and quick evals, not for untrusted code.""" from __future__ import annotations import os import shutil import subprocess import tempfile import time from typing import Mapping, Optional from ..core import ( ExecResult, FileInfo, PathOutsideSandbox, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("local") class LocalProvider(Provider): info = ProviderInfo( name="local", capabilities=frozenset({"exec", "fs"}), cost_tier=0, ) def __init__(self, base_dir: Optional[str] = None): self.base_dir = base_dir def create(self, spec: SandboxSpec) -> "LocalSandbox": root = tempfile.mkdtemp(prefix="sandboxkit-", dir=self.base_dir) return LocalSandbox(root=root, env=dict(spec.env)) class LocalSandbox(Sandbox): def __init__(self, root: str, env: Mapping[str, str]): self._root = os.path.realpath(root) self._id = os.path.basename(self._root) self._env = dict(env) self._alive = True @property def id(self) -> str: return self._id @property def provider(self) -> str: return "local" # ---------------- internal ---------------- def _resolve(self, path: str) -> str: """Map an in-sandbox POSIX path to a host path, refusing escapes.""" if not path.startswith("/"): raise SandboxError(f"path must be absolute inside the sandbox: {path!r}") full = os.path.realpath(os.path.join(self._root, path.lstrip("/"))) if full != self._root and not full.startswith(self._root + os.sep): raise PathOutsideSandbox(path) return full # ---------------- exec ---------------- def exec(self, command, *, cwd="/", env=None, timeout=None) -> ExecResult: if not self._alive: raise SandboxNotFound(f"sandbox {self._id} is stopped") start = time.monotonic() merged = {**os.environ, **self._env, **(env or {})} try: proc = subprocess.run( command, shell=True, cwd=self._resolve(cwd), env=merged, capture_output=True, timeout=timeout, ) except subprocess.TimeoutExpired as e: raise SandboxTimeout(f"command timed out after {timeout}s: {command!r}") from e return ExecResult( exit_code=proc.returncode, stdout=(proc.stdout or b"").decode("utf-8", "replace"), stderr=(proc.stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - start) * 1000), ) # ---------------- filesystem ---------------- def read_file(self, path: str) -> bytes: full = self._resolve(path) if os.path.isdir(full): raise SandboxError(f"{path} is a directory") try: with open(full, "rb") as f: return f.read() except FileNotFoundError: raise SandboxNotFound(path) from None def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") full = self._resolve(path) os.makedirs(os.path.dirname(full), exist_ok=True) # parents, like `mkdir -p` with open(full, "wb") as f: f.write(data) def list_files(self, path: str = "/") -> list[FileInfo]: full = self._resolve(path) if not os.path.exists(full): raise SandboxNotFound(path) out = [] with os.scandir(full) as it: for entry in it: st = entry.stat(follow_symlinks=False) in_sandbox = "/" + os.path.relpath(entry.path, self._root).replace(os.sep, "/") out.append(FileInfo( path=in_sandbox if in_sandbox != "//" else "/", name=entry.name, is_dir=entry.is_dir(follow_symlinks=False), size=st.st_size, modified_at=st.st_mtime, )) return sorted(out, key=lambda f: (not f.is_dir, f.name)) ``` Hmm, the relpath edge: entry.path for root child = root + "/" + name → relpath = name → "/name". For listing "/" it's fine. The "//" edge can't happen since relpath of a child never equals ".". Actually if path is "/" then entries are direct children; relpath gives "name". Fine — remove the weird guard. Simplify: ```python rel = os.path.relpath(entry.path, self._root).replace(os.sep, "/") path_in_sandbox = "/" + rel ``` If listing a subdirectory, rel = "sub/name" → "/sub/name". Correct. But wait — `entry.stat()` on broken symlink raises; wrap? Edge case, skip (small scope). Also `list_files` on a *file* path? os.scandir(file) raises NotADirectoryError → convert to SandboxError. Add try/except: ```python try: with os.scandir(full) as it: ... except NotADirectoryError: raise SandboxError(f"{path} is not a directory") from None ``` delete + mkdir: ```python def delete_file(self, path: str, *, recursive: bool = False) -> None: full = self._resolve(path) if not os.path.exists(full) and not os.path.islink(full): raise SandboxNotFound(path) if os.path.isdir(full) and not os.path.islink(full): if recursive: shutil.rmtree(full) else: try: os.rmdir(full) except OSError as e: raise SandboxError(f"{path} is not empty (use recursive=True)") from e else: os.unlink(full) def mkdir(self, path: str) -> None: os.makedirs(self._resolve(path), exist_ok=True) # behaves like `mkdir -p` # ---------------- lifecycle ---------------- def stop(self, *, force: bool = False) -> None: if self._alive: shutil.rmtree(self._root, ignore_errors=True) self._alive = False def is_alive(self) -> bool: return self._alive ``` Wait — Sandbox ABC: should delete_file have `recursive` param? If abstract signature has it, docker must accept it too. Yes, include in ABC. Also delete on a symlink: os.path.exists follows symlinks; broken symlink → exists False, islink True → handled by the `or os.path.islink` condition then unlink. OK. One issue: `Sandbox.run` convenience uses ExecFailed — in core. And base `is_alive` uses exec — local overrides. Docker overrides. **`sandboxkit/core.py` Sandbox ABC** — full: ```python class Sandbox(ABC): """A live sandbox. All paths are absolute POSIX paths inside the sandbox.""" @property @abstractmethod def id(self) -> str: ... @property @abstractmethod def provider(self) -> str: ... # ---- process / command primitives ---- @abstractmethod def exec( self, command: str, *, cwd: str = "/", env: Optional[Mapping[str, str]] = None, timeout: Optional[float] = None, ) -> ExecResult: """Run a shell command. Never raises on non-zero exit; inspect ExecResult.""" # ---- filesystem primitives ---- @abstractmethod def read_file(self, path: str) -> bytes: ... @abstractmethod def write_file(self, path: str, data) -> None: ... @abstractmethod def list_files(self, path: str = "/") -> list: ... @abstractmethod def delete_file(self, path: str, *, recursive: bool = False) -> None: ... @abstractmethod def mkdir(self, path: str) -> None: ... # ---- lifecycle ---- @abstractmethod def stop(self, *, force: bool = False) -> None: ... def is_alive(self) -> bool: try: return self.exec("true", timeout=10).exit_code == 0 except SandboxError: return False # ---- conveniences built only on the primitives above ---- def run(self, command: str, **kwargs) -> str: """exec + raise ExecFailed on non-zero exit + return stdout.""" result = self.exec(command, **kwargs) if result.exit_code != 0: raise ExecFailed(result) return result.stdout def write_text(self, path: str, text: str) -> None: self.write_file(path, text) def read_text(self, path: str) -> str: return self.read_file(path).decode("utf-8", "replace") def __enter__(self) -> "Sandbox": return self def __exit__(self, *exc_info) -> None: try: self.stop() except SandboxError: pass ``` Type hints: keep light (`list` → `list[FileInfo]` with `from __future__ import annotations` it's fine as string annotation). I'll include `from __future__ import annotations` in core and use plain annotations. **`sandboxkit/providers/docker.py`** — full: ```python """Docker provider: each sandbox is a container running `sleep infinity`. Requires the `docker` package and a reachable daemon. Container FS = sandbox FS, so no path-jail is needed (the container boundary is the jail).""" from __future__ import annotations import io import posixpath import shlex import tarfile import time from typing import Mapping, Optional import docker from docker.errors import APIError, ImageNotFound, NotFound from ..core import ( ExecFailed, ExecResult, FileInfo, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("docker") class DockerProvider(Provider): info = ProviderInfo( name="docker", capabilities=frozenset({"exec", "fs", "network"}), cost_tier=1, ) def __init__(self, image: str = "ubuntu:24.04", base_url: Optional[str] = None, network: Optional[str] = None, pull: bool = False): kwargs = {"base_url": base_url} if base_url else {} self._client = docker.from_env(**kwargs) self._default_image = image self._network = network if pull: self._client.images.pull(image) def health(self) -> bool: try: self._client.ping() return True except Exception: return False def create(self, spec: SandboxSpec) -> "DockerSandbox": image = spec.image or self._default_image try: container = self._client.containers.run( image, command="sleep infinity", detach=True, auto_remove=True, network=self._network, environment=dict(spec.env), labels=dict(spec.labels) or None, ) except ImageNotFound as e: raise SandboxError(f"image not found: {image}") from e except APIError as e: raise SandboxError(f"docker create failed: {e}") from e return DockerSandbox(container, default_env=dict(spec.env)) ``` Hmm — SandboxError for provider infra errors: I defined ProviderError for provider-level failures. Use ProviderError here (create failures = provider failure → failover to next provider is desired). Yes: ImageNotFound/APIError → ProviderError. And SandboxError base catches it in failover loop anyway. Use ProviderError — more precise. `labels=dict(spec.labels) or None` — docker accepts dict; empty dict fine, no need for None. Just pass `dict(spec.labels)`. network=None means default bridge; fine. DockerSandbox: ```python class DockerSandbox(Sandbox): def __init__(self, container, default_env: Mapping[str, str]): self._c = container self._id = container.id self._default_env = dict(default_env) @property def id(self) -> str: return self._id[:12] @property def provider(self) -> str: return "docker" # ---------------- exec ---------------- def exec(self, command, *, cwd="/", env=None, timeout=None) -> ExecResult: if timeout is not None: # Best-effort per-command timeout via GNU `timeout` (present on ubuntu/debian images). inner = f"timeout -k 1 {timeout} bash -c {shlex.quote(command)}" else: inner = command start = time.monotonic() try: code, streams = self._c.exec_run( ["bash", "-c", inner], workdir=cwd if cwd != "/" else None, environment={**self._default_env, **(env or {})}, demux=True, ) except NotFound: raise SandboxNotFound(f"sandbox {self._id} is gone") from None except APIError as e: raise SandboxError(f"docker exec failed: {e}") from e stdout, stderr = streams result = ExecResult( exit_code=int(code), stdout=(stdout or b"").decode("utf-8", "replace"), stderr=(stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - start) * 1000), ) if timeout is not None and result.exit_code in (124, 137): raise SandboxTimeout(f"command timed out after {timeout}s: {command!r}", ) return result ``` Hmm wait — SandboxTimeout raised means we lose result; that's fine (timeouts are exceptional). But false positive: command legitimately exits 124. Note in comment. Alternatively attach result: `raise SandboxTimeout(...)` — keep simple, note caveat in docstring/comment. Also: `workdir=cwd if cwd != "/" else None` — passing None uses container's default WORKDIR (often "/"). Good. exec envs: docker exec_run `environment` param expects dict — yes (it converts to env list). Note about `bash` availability: ubuntu has bash; alpine needs `sh`. Could make shell configurable on provider. Add note or param `shell=["bash", "-c"]` on DockerProvider, threaded to sandbox. Small addition, useful. I'll pass shell to DockerSandbox constructor with default `["bash", "-c"]`. Then exec uses `self._shell + [inner]`. And for timeout wrapper: `timeout -k 1 {t} {shlex.join(shell)} {shlex.quote(command)}`? Getting complicated. Keep bash hard assumption + note "swap to sh for alpine". Simpler. Actually, let me add the `shell` param — it's 3 lines and makes alpine work: Provider init: `self._shell = tuple(shell)` default `("bash", "-c")`. create passes to sandbox. exec: `cmd = [*self._shell, inner]`. Timeout wrapper: `f"timeout -k 1 {timeout} {shlex.join(self._shell)} {shlex.quote(command)}"` — shlex.join(["bash","-c"]) = "bash -c", then quote(command) appended: `timeout -k 1 5 bash -c 'echo hi'`. Wait that's wrong: inner would be run as `bash -c timeout...`? No: cmd = [*shell, inner] = ["bash", "-c", "timeout -k 1 5 bash -c 'echo hi'"] → outer bash -c runs the timeout string, which runs `timeout ... bash -c 'echo hi'`. Correct. OK, include shell param. Hmm, but timeout+alpine: busybox timeout exists with different flags (-k unsupported? busybox timeout supports `-s SIG` and `-k`? busybox timeout: `timeout [-s SIG] SECS PROG [ARGS]` — no -k maybe). Eh — default bash+GNU timeout is fine; alpine users can adjust. Note it. Actually for scope, keep shell param but leave timeout as GNU assumption with comment. Fine. filesystem: ```python # ---------------- filesystem ---------------- def _norm(self, path: str) -> str: p = posixpath.normpath("/" + path) # collapses "..", resolves to absolute return p def read_file(self, path: str) -> bytes: p = self._norm(path) try: stream, _stat = self._c.get_archive(p) except NotFound: raise SandboxNotFound(path) from None raw = b"".join(stream) with tarfile.open(fileobj=io.BytesIO(raw), mode="r") as tf: files = [m for m in tf.getmembers() if m.isfile()] if not files: raise SandboxError(f"{path} is not a regular file") return tf.extractfile(files[0]).read() def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") rel = self._norm(path).lstrip("/") if not rel: raise SandboxError("cannot write to '/'") buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: acc = [] for seg in rel.split("/")[:-1]: # create parent dirs inside the tar acc.append(seg) d = tarfile.TarInfo("/".join(acc)) d.type = tarfile.DIRTYPE d.mode = 0o755 tf.addfile(d) f = tarfile.TarInfo(rel) f.size = len(data) f.mode = 0o644 tf.addfile(f, io.BytesIO(data)) try: self._c.put_archive("/", buf.getvalue()) except APIError as e: raise SandboxError(f"docker put_archive failed: {e}") from e ``` Wait — normpath("/"+path): if user passes "work/hello.py" (relative-ish), we make "/work/hello.py" — actually the contract says paths are absolute; normpath with leading "/" is forgiving. For local I raise on non-absolute. Inconsistent — make both forgiving? Local contract says absolute; hmm. For friendliness, let both accept and normalize: local `_resolve` — if not startswith "/", prepend? I wrote raise. Let me make both forgiving: treat any path as sandbox-absolute (prepend "/"). Simpler UX. But local raise exists to catch bugs... honestly forgiving normalization is friendlier and consistent: "all paths are interpreted as absolute in-sandbox paths; leading '/' optional." I'll do: local `_resolve`: strip leading "/", join, realpath, containment check. That handles "work/x" and "/work/x" identically. And ".." segments: realpath resolves and containment check rejects escapes → PathOutsideSandbox. Good. For docker, normpath collapses ".." → stays in container → no escape possible (container root). Consistent-ish. OK, update local `_resolve`: ```python def _resolve(self, path: str) -> str: full = os.path.realpath(os.path.join(self._root, path.lstrip("/"))) if full != self._root and not full.startswith(self._root + os.sep): raise PathOutsideSandbox(path) return full ``` Note: lstrip("/") on "/tmp/../../etc" → "tmp/../../etc" → realpath resolves → /root-of-sandbox/../... wait: join(root, "tmp/../../etc") → root/tmp/../../etc → realpath → parent-of-root/etc → escape detected → raise. list_files (ls-based, uniform): ```python def list_files(self, path: str = "/") -> list: p = self._norm(path) result = self.exec(f"ls -1Ap -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) out = [] for line in result.stdout.splitlines(): if not line: continue is_dir = line.endswith("/") name = line.rstrip("/") out.append(FileInfo( path=posixpath.join(p if p != "/" else "", name) or "/" + name, name=name, is_dir=is_dir, size=0, # ls-based listing: metadata not guaranteed modified_at=0.0, )) return out ``` Hmm `posixpath.join(p, name)` where p="/" gives "/name" — posixpath.join("/", "x") = "/x". Good, so just `posixpath.join(p, name)` works even for "/". Simplify: `path=posixpath.join(p, name)`. delete/mkdir: ```python def delete_file(self, path: str, *, recursive: bool = False) -> None: p = self._norm(path) flag = "-rf" if recursive else "-d" result = self.exec(f"rm {flag} -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) def mkdir(self, path: str) -> None: p = self._norm(path) result = self.exec(f"mkdir -p -- {shlex.quote(p)}") if result.exit_code != 0: raise ExecFailed(result) ``` `rm -d` on non-empty dir: GNU rm -d removes empty dirs only, errors otherwise → ExecFailed with message. Good parity with local. stop/is_alive: ```python # ---------------- lifecycle ---------------- def stop(self, *, force: bool = False) -> None: try: self._c.stop(timeout=0 if force else 10) except NotFound: pass # already gone; auto_remove cleans up except APIError as e: raise SandboxError(f"docker stop failed: {e}") from e def is_alive(self) -> bool: try: self._c.reload() return self._c.status == "running" except APIError: return False ``` `stop(timeout=0)` — docker-py stop timeout=0 means immediate SIGKILL? In docker-py, timeout is seconds to wait before killing; timeout=0 → immediate kill? The API waits `timeout` seconds then SIGKILLs; passing 0... docker-py: `stop(timeout=10)`; internally if timeout is passed uses it; 0 might mean "kill immediately". I believe docker API stop with t=0 sends SIGKILL immediately. OK. **`sandboxkit/__init__.py`**: ```python """sandboxkit — one tiny interface over many sandbox providers.""" from .api import SandboxKit from .core import ( ExecFailed, ExecResult, FileInfo, NoRouteError, PathOutsideSandbox, ProviderError, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from .provider import Provider, ProviderInfo, provider_names, register from .router import ProviderConfig, Router __all__ = [...] __version__ = "0.1.0" ``` Careful: api.py imports providers.local and providers.docker; docker.py imports `docker` package at module top — if user hasn't installed docker-py, importing sandboxkit fails! That's bad for the local-only use case. Fix: lazy import inside DockerProvider.__init__? Then DockerProvider class definition needs no docker import... but it references docker.errors inside methods — could import module-level in methods, or catch. Cleanest small approach: try/except at module top: ```python try: import docker from docker.errors import APIError, ImageNotFound, NotFound except ImportError: docker = None APIError = ImageNotFound = NotFound = () # dummy ``` Then in __init__: `if docker is None: raise ProviderError("pip install docker")`. And health() returns False if docker is None. And exception handlers reference APIError which is `()` (empty tuple → except () catches nothing — actually `except ()` never matches, which is what we want when docker missing since methods would fail earlier). Hmm, `except APIError:` with APIError=() → "except ()" is valid? `except` requires exception classes or tuple; empty tuple is allowed (matches nothing). Yes valid. And ImageNotFound similar. This makes sandboxkit importable without docker installed. Alternatively raise inside health and __init__ only. Let me implement: ```python try: import docker from docker.errors import APIError, ImageNotFound, NotFound except ImportError: # docker optional; provider surfaces as "not installed" docker = None APIError = () ImageNotFound = () NotFound = () ``` In DockerProvider.__init__: `if docker is None: raise ProviderError("docker package not installed: pip install docker")`. Wait, but __init__ is called by router._instance during route() — should raise before... routing calls health() first. health(): `if docker is None: return False`. Then __init__ raise is belt-and-braces. But _instance() is called before health() in my Router.route flow — `p = self._instance(cfg)` then health check. So __init__ raising ProviderError would propagate out of route() → kit.sandbox → NoRouteError catch? In kit.sandbox, router.route(spec) raising ProviderError isn't caught (catch wraps create only). Fix: make route() itself resilient — wrap `self._instance(cfg)` and health in try/except SandboxError → skip candidate. Let me restructure route: ```python for cfg in self._configs: if not cfg.enabled: continue try: p = self._instance(cfg) ok = p.health() except SandboxError: ok = False if not ok: continue ... ``` Then a broken provider config is skipped gracefully (with failover). And if all fail, NoRouteError message lists attempts? The attempts list only covers create failures. For route-skip reasons we lose detail... acceptable for small scope; maybe router can collect skip reasons and kit can include them. Keep simple: router exposes last-skip reasons? Meh — keep simple, note it. Actually let me have route() return ranked list, and kit's NoRouteError message mention providers configured. Fine for scope. Also ProviderError raised from DockerProvider.__init__ when docker-py missing — caught by route's try/except SandboxError → skipped. And in kit.sandbox failover catches SandboxError from create. Good. But wait: health() for local returns True always (base). For DockerProvider with docker=None → False → skipped. If docker package installed but daemon down → ping fails → False → skipped → fallback to local. **Docker health caching**: ping per sandbox creation is fine. Now — `DockerProvider.create` also passes `spec.image`; if user pins image not pulled, docker auto-pulls on containers.run (yes, docker-py pulls automatically if missing? `containers.run` will pull if image not present locally — yes, it does pull). Good. **Environment at create vs exec**: I pass spec.env as container env AND as default exec env. Redundant but harmless; container env makes it visible to any process. Keep. **Local sandbox `exec` and `os.environ` merge**: includes full host env — that leaks host env into "sandbox". For an eval sandbox that's undesirable; but for dev convenience it's needed (PATH etc.). Compromise: pass os.environ by default (PATH needed to find binaries) — document the caveat. OK. **Now the example usage file** (`examples/quickstart.py`): ```python from sandboxkit import SandboxKit, SandboxSpec kit = SandboxKit.auto() # 1) automatic routing: docker if the daemon is up, else local temp dir with kit.sandbox() as sbx: print("routed to:", sbx.provider, sbx.id) # primitive: exec print(sbx.exec("echo hello from the sandbox").stdout) # primitive: filesystem sbx.write_file("/work/greet.py", 'print("hi from a file")\n') print(sbx.run("python3 /work/greet.py")) print(sbx.read_file("/work/greet.py").decode()) for f in sbx.list_files("/work"): print(" ", f.name, "dir" if f.is_dir else "file") # 2) pin a provider / state requirements — routing becomes explicit with kit.sandbox(provider="local") as sbx: ... # 3) ask for capabilities; router filters providers that can't deliver spec = SandboxSpec(needs={"network"}, image="ubuntu:24.04") with kit.sandbox(spec) as sbx: print(sbx.run("curl -s https://example.com -o /dev/null -w '%{http_code}'")) ``` **Config file** example: ```yaml # sandboxkit.yaml providers: - type: docker priority: 10 # preferred options: image: ubuntu:24.04 - type: local priority: 0 # fallback options: base_dir: /tmp/sandboxes ``` **Tests** (`tests/test_providers.py`): ```python import pytest from sandboxkit import SandboxKit, NoRouteError, SandboxNotFound, SandboxTimeout def make_kit(include_docker: bool): cfg = {"providers": [{"type": "local"}]} if include_docker: cfg["providers"].insert(0, {"type": "docker", "priority": 10}) return SandboxKit.from_dict(cfg) def docker_available(): try: from sandboxkit.providers.docker import DockerProvider return DockerProvider().health() except Exception: return False ``` Hmm DockerProvider() instantiation needs no docker necessarily... it raises ProviderError if missing. Wrapped in except. OK. Or simpler: ```python @pytest.fixture(params=["local", "docker"]) def kit(request): if request.param == "docker": pytest.importorskip("docker") # also require a running daemon ... ``` Keep tests compact; show a handful: ```python def test_exec_and_fs_roundtrip(): with make_kit(False).sandbox(provider="local") as s: s.write_file("/x/a.txt", "hi") assert s.read_file("/x/a.txt") == b"hi" assert s.run("cat /x/a.txt") == "hi\n" assert [f.name for f in s.list_files("/x")] == ["a.txt"] def test_exit_codes_and_env(): with make_kit(False).sandbox() as s: r = s.exec("exit 3") assert r.exit_code == 3 r = s.exec("echo $FOO", env={"FOO": "bar"}) assert r.stdout == "bar\n" def test_timeout(): with make_kit(False).sandbox() as s: with pytest.raises(SandboxTimeout): s.exec("sleep 5", timeout=1) def test_local_path_jail(): with make_kit(False).sandbox() as s: with pytest.raises(PathOutsideSandbox): s.read_file("/../etc/passwd") def test_routing_failover(): # docker config with an image that doesn't exist → should fall back to local? ``` Hmm — failover test: docker create with bad image raises SandboxError → kit falls back to local. But route() ranks docker first (priority 10) and local second. create(image="nope:not-real") → docker raises SandboxError("image not found") → caught → try local → success. Nice test but requires docker daemon. Make it conditional. Or test NoRouteError: pin provider="docker" with no docker config → Router._config_of raises SandboxError("not configured") — propagates from route(). Hmm — should pinning to unconfigured provider raise NoRouteError? _config_of raises SandboxError which kit doesn't catch inside route... kit catches around create only. So it propagates as SandboxError — fine-ish; better to raise NoRouteError. Let me wrap: in kit.sandbox, wrap `self.router.route(spec)` in try/except SandboxError → attempts = [f"routing: {e}"] → raise NoRouteError. Cleaner: router.route raises NoRouteError directly for pin-not-configured and empty-candidate list. Then kit's failover loop handles empty list → NoRouteError with attempts. Let me make router raise NoRouteError for unconfigured pin, and return [] if no candidates (kit raises NoRouteError with attempts). Also if pin configured but create fails — kit catches SandboxError, attempts has one entry, then raises NoRouteError("...docker: ..."). Acceptable. I'll show a few tests only — keep short. **Directory tree** to present: ``` sandboxkit/ ├── __init__.py # public API ├── core.py # errors, result types, Sandbox interface ├── provider.py # Provider interface + registry ├── router.py # routing rules ├── api.py # SandboxKit facade (+ config loading) └── providers/ ├── __init__.py ├── local.py # subprocess + tempdir (zero deps) └── docker.py # docker daemon (pip install docker) tests/ └── test_basics.py examples/ └── quickstart.py sandboxkit.yaml ``` **Prose structure of the answer:** 1. Design in one paragraph + the four key decisions (Provider/Sandbox split; small primitive instruction set; capability-based routing with failover; bytes + absolute paths contract). 2. Architecture diagram + directory tree. 3. The primitive contract (table of ops). Keep brief. 4. Code: each file. 5. Usage. 6. Config + routing rules explained (bullet list of the routing algorithm). 7. Adding a third provider (E2B sketch, short). 8. Deliberate scope cuts / next steps (async, streaming exec, long-running processes, snapshots, retries, real sandboxing for local). I must watch total length — the code is ~450 lines. Prose should be tight. Use markdown headers and code blocks. The user asked "suggest a structure... and then build it" — so both, and code-heavy is right. Let me reconsider a couple of implementation details for correctness: 1. `subprocess.run(..., shell=True)` with `timeout` — raises TimeoutExpired; the child may leave orphans; on POSIX subprocess kills the process group? `run` kills the direct child (shell); grandchildren may linger. Acceptable caveat, don't mention or mention briefly? Skip — small scope. 2. Local `exec` with `cwd` not existing → subprocess raises FileNotFoundError (not caught) → leaks as raw exception. Wrap: try/except FileNotFoundError → SandboxError? Add to exec try block: catch FileNotFoundError/PermissionError → SandboxError. Let me add: ```python try: proc = subprocess.run(...) except subprocess.TimeoutExpired as e: raise SandboxTimeout(...) except FileNotFoundError as e: raise SandboxError(f"cwd does not exist: {cwd}") from e ``` Hmm FileNotFoundError could also be from shell binary? shell=True uses /bin/sh, present. So it's cwd. Fine, or generic OSError → SandboxError. I'll catch OSError broadly → SandboxError. But TimeoutExpired is not OSError subclass? TimeoutExpired inherits SubprocessError, Exception. OK order matters: TimeoutExpired first, then OSError. Good. 3. Docker `exec_run` demux returns `(exit_code, (stdout_bytes_or_None, stderr_bytes_or_None))`. With demux=True, output is tuple of bytes; None if empty. Handled. 4. Docker `get_archive` on missing path → raises NotFound (docker.errors.NotFound). Handled → SandboxNotFound. 5. `put_archive("/")` with tar containing "x/a.txt": extracts to /x/a.txt, creating /x. Correct. 6. TarInfo dir entries: `tf.addfile(d)` with no fileobj — tarfile writes size 0 and type dir. Correct. But: if a parent dir entry appears after... order: we add dirs in ascending depth before file — required for correct extraction. Our loop does parents first. Good. 7. `exec_run` env param: `environment={"FOO": "bar"}` — docker-py accepts dict[str, str]. Good. 8. Windows local: os.sep checks — I use `self._root + os.sep`; realpath on Windows returns backslashes; join with lstrip'd path uses os.path.join — fine. relpath replace os.sep → "/" for FileInfo path. POSIX-ish. OK. 9. `Sandbox.run` kwargs pass-through to exec — cwd/env/timeout. Good. 10. `kit.sandbox()` returns Sandbox — context manager via ABC's __enter__/__exit__. Good. 11. Local sandbox id = basename of mkdtemp root: "sandboxkit-xxxx". Fine. 12. Router `_instance` caching per type — if two configs of same type with different options (e.g., two docker pools, different images/regions) → cache collision! _instances keyed by type string; two entries same type different options → second gets first's instance (wrong options). Fix: key by (type, id(options))? Or instantiate per config: `self._instances[id(cfg)] = ...`. Better: key by `id(cfg)` (config object identity). Let me store instances on config? Config is a plain dataclass — attach via `cfg._instance`? Hacky but fine for scope: store `self._instances[id(cfg)]`. I'll do dict keyed by id(cfg). Document. Hmm, actually multiple same-type configs (two docker pools, different regions) is a realistic routing scenario — supporting it correctly matters. id(cfg) keyed dict it is. 13. `ProviderInfo.capabilities` as frozenset — declared as class attribute on providers; needs frozenset literal. Fine. 14. Region support: local/docker have no regions (empty tuple) → region spec: `if spec.region and p.info.regions and spec.region not in p.info.regions: continue` — providers with no regions pass any region spec (treated as location-agnostic). Document. With E2B sketch I could set regions=() too; cloud providers would declare regions. 15. SandboxSpec.needs — used only for capability filter; docker declares "network" capability, local doesn't → spec with needs={"network"} skips local. But wait: local subprocesses DO have network. Declaring it as capability=false for local is a *policy* choice (keep local for untrusted-ish/no-network work). Since local is your own machine, network exists but you probably don't want to route network jobs there. Hmm — capability means "can do", not "should do". For routing honesty, local CAN exec network commands. But semantically for routing "needs network" means "sandbox must provide network" — local does. If I mark local without network, then needs={"network"} → only docker. That's actually a nice demo of routing filtering! And justifiable: local is a dev/test target; routing network-required work to docker is a policy. But it's a bit of a lie in the capability model. Alternative: give both network capability, and demo filtering via something else... like `gpu`? Neither has gpu. Hmm. I think it's fine and even instructive: capabilities encode what the provider is *intended* for; I'll note "local declares no `network` capability because we keep it as an offline dev target — declare it if you want". Or be honest: local has network but we treat capability as policy-declared. I'll add a comment in local.py: "note: subprocesses can reach the network; we deliberately don't declare it so network-needing work routes to docker. Change if you disagree." That demonstrates capabilities as policy + shows routing working with two providers. Good teaching moment. 16. `SandboxSpec.timeout_seconds` — "default per-exec timeout" — should sandbox.exec default to it? I created the field but never use it. Either wire it (sandbox instances could receive it and use as default timeout in exec when timeout=None) or drop it. Wiring: in create, providers pass `spec.timeout_seconds` to sandbox as default_timeout; exec uses `timeout if timeout is not None else self._default_timeout`. Local: subprocess timeout=None means no timeout — with default None fine. Docker: timeout=None means no wrapper. Wiring is ~4 lines per provider. Worth it? It's a nice touch but adds surface. Small scope... I'll keep the field but implement wiring in both providers minimally (3 lines each). Actually — to keep code tight, I'll drop `timeout_seconds` from spec and mention default-timeout as a next-step. Hmm. It's cheap: LocalSandbox takes `default_timeout`; exec: `timeout = timeout if timeout is not None else self._default_timeout`. Docker same. 6 lines total. I'll keep it — it makes the spec feel real. OK keep. Wait, docker default timeout: when timeout is None and default_timeout set → wrapper applies. Implement: ```python t = timeout if timeout is not None else self._default_timeout if t is not None: inner = f"timeout -k 1 {t} bash -c {shlex.quote(command)}" ``` Careful with shell tuple: inner uses shell tuple joined. Let me finalize docker exec: ```python def exec(self, command, *, cwd="/", env=None, timeout=None): t = timeout if timeout is not None else self._default_timeout if t is not None: shell_str = shlex.join(self._shell) inner = f"timeout -k 1 {t} {shell_str} {shlex.quote(command)}" else: inner = command ... cmd = [*self._shell, inner] code, streams = self._c.exec_run(cmd, workdir=..., environment=..., demux=True) ``` shlex.join(["bash","-c"]) → "bash -c". So inner = `timeout -k 1 30 bash -c 'echo hi'` run via ["bash","-c", inner]. Works. GNU timeout exit code on timeout = 124 (unless -k kill → 137 if killed by SIGKILL after grace). With -k 1: SIGTERM first → exit 124; if ignores → SIGKILL → exit 137. So check `in (124, 137)`. Note false-positive caveat in comment. 17. `SandboxTimeout` needs to be importable — yes from core. 18. Providers package `__init__.py` — empty file content: maybe re-export for convenience: `from .local import LocalProvider` etc. But docker import would pull docker pkg... with try/except it's safe. api.py imports them anyway. Empty is fine; I'll add docstring only. 19. `core.py` Sandbox ABC — abstract properties id/provider. Subclasses implement. Local overrides is_alive. Fine. 20. ExecFailed in docker list/delete — imported. In docker exec — not raised (exec returns result; run() convenience raises). list_files raises ExecFailed if ls fails for non-NotFound reason. Import ExecFailed in docker.py. Yes included. 21. One more: `Router.close()` — docker client close? DockerClient has .close(). Minor; include close protocol optional. Eh — skip close entirely for scope? Router.close exists in my draft; keep it minimal or drop. Drop to reduce surface. Actually keeping is fine (2 lines). I'll drop it — not needed for scope. Hmm, providers might hold clients... small scope: drop. 22. `SandboxKit.from_dict` — providers list; also allow `default_provider`? Skip. 23. `auto()`: docker priority 10, local 0 — if docker daemon is down, health() False → local. If docker package missing → local. 24. Type annotations with `from __future__ import annotations` everywhere → can use `list[FileInfo]`, `frozenset[str]` on 3.8? Fine since postponed. I'll include the future import in each file and target 3.9+. 25. `LocalProvider.create` — spec.env passed. `LocalSandbox(root, env=dict(spec.env), default_timeout=spec.timeout_seconds)`. 26. base `is_alive` uses exec "true" — docker overrides with reload. Good. 27. `Sandbox.write_file(path, data)` type: `data: bytes | str` — with future annotations OK (3.10 union syntax in string annotation evaluates lazily — but if user introspects... fine for 3.10+; to be safe use `Union[bytes, str]`? With `from __future__ import annotations`, `bytes | str` in annotations is fine even on 3.8 (never evaluated). OK.) 28. `Mapping` import in provider files for env types. 29. kit.sandbox **overrides validation: `SandboxSpec(**overrides)` — unknown key → TypeError (clear enough). 30. Docker provider options in config: `options: {image: ubuntu:24.04}` → DockerProvider(image=...). Local options: base_dir. Good. Now the E2B extension sketch — keep ~35 lines, marked as sketch ("verify method names against e2b's current docs; their API churns"): ```python # sandboxkit/providers/e2b.py # pip install e2b (export E2B_API_KEY=...) @register("e2b") class E2BProvider(Provider): info = ProviderInfo( name="e2b", capabilities=frozenset({"exec", "fs", "network", "internet"}), cost_tier=2, ) def __init__(self, template: str = "ubuntu-24.04", default_timeout: int = 300): self.template = template self.default_timeout = default_timeout def create(self, spec: SandboxSpec) -> Sandbox: from e2b import Sandbox as _E2B # lazy: heavy import sbx = _E2B.create(template=spec.image or self.template, timeout=spec.ttl_seconds or self.default_timeout, env_vars=dict(spec.env)) return _E2BAdapter(sbx) class _E2BAdapter(Sandbox): def __init__(self, sbx): self._s = sbx @property def id(self): return self._s.sandbox_id @property def provider(self): return "e2b" def exec(self, command, *, cwd="/", env=None, timeout=None): r = self._s.commands.run(command, cwd=cwd if cwd != "/" else None, envs=dict(env or {}), timeout=timeout) return ExecResult(r.exit_code, r.stdout, r.stderr, 0) def read_file(self, path): return self._s.files.read(path) def write_file(self, path, data): self._s.files.write(path, data.encode() if isinstance(data, str) else data) def list_files(self, path="/"): return [FileInfo(path=f.path, name=f.name, is_dir=f.is_dir, size=0, modified_at=0.0) for f in self._s.files.list(path)] def delete_file(self, path, *, recursive=False): self._s.files.remove(path) def mkdir(self, path): self.run(f"mkdir -p {shlex.quote(path)}") def stop(self, *, force=False): self._s.remove() ``` Then config line: `- type: e2b, priority: 20`. Good demo of "add a provider = 1 file + 1 config line". Also capabilities naming: I used "network" for docker; e2b "network" too. Keep consistent single name "network". In my earlier draft I wrote {"exec","fs","network","internet"} — just use "network". OK. **Routing algorithm summary** (prose): 1. If spec.provider pinned → use it, no fallback beyond create-failover (well, pin means only that provider; create failure → NoRouteError). Hmm — with failover loop, pinned provider create failure → attempts=[pin] → NoRouteError. Should pin failover? No — pin means explicit. OK. 2. Else filter: enabled, capabilities ⊇ needs, region ok, health ok. 3. Score: priority*10 + region match(5) − cost_tier. 4. Kit tries ranked order, first create success wins; failures logged into NoRouteError. **Contract decisions prose (short bullets):** - Paths: absolute POSIX in sandbox; leading "/" optional; local enforces a directory jail, docker's jail is the container itself. - Payloads: bytes for file APIs; str accepted for convenience. - exec never raises on non-zero exit (returns ExecResult); `run()` convenience wraps. - Errors: one hierarchy; routing can fail over on ProviderError/create failures. - The interface is the intersection; provider-specific extras stay provider-side (or new capabilities). - Sync-only for now; async is the first production upgrade. **Scope cuts / next steps:** - Streaming exec (stdout as iterator), long-lived processes (start/kill/send signal), code-upload-and-run helpers, snapshots/pooling (warm pools), per-provider retries with backoff, usage/metering, async (anyio), real isolation for local (bwrap/nsjail), Windows note. Also a note on `local` being *not* secure isolation — important safety callout. Good. **Testing note:** parametrized conftest so the same tests run against both providers: Maybe include a short `conftest.py`: ```python import pytest from sandboxkit import SandboxKit def _docker_ok(): try: import docker docker.from_env().ping() return True except Exception: return False @pytest.fixture(params=["local"] + (["docker"] if _docker_ok() else [])) def kit(request): providers = [{"type": "local"}] if request.param == "docker": providers.insert(0, {"type": "docker", "priority": 10}) return SandboxKit.from_dict({"providers": providers}) ``` And test: ```python def test_same_contract_everywhere(kit): with kit.sandbox() as s: s.write_file("/work/hello.txt", "hello") assert s.read_file("/work/hello.txt") == b"hello" assert s.run("cat /work/hello.txt").strip() == "hello" assert {f.name for f in s.list_files("/work")} == {"hello.txt"} s.delete_file("/work/hello.txt") with pytest.raises(SandboxNotFound): s.read_file("/work/hello.txt") ``` Wait — docker sandbox: /work doesn't exist → write_file creates it via tar parent dirs. ✓. Local: created via makedirs. ✓. pytest.raises import needed in that snippet — fine, add import line. Length check: This is going to be a long answer, but it's a "build it" request; that's appropriate. I'll keep prose lean, code complete. Let me also double check docker.py exception usage when docker is None: methods reference `NotFound` = () and `APIError` = () in except clauses — only reachable if a client exists, which requires docker not None. But `self._c.exec_run` etc. would need self._c — only exists if created. OK. Also `docker.from_env(**kwargs)` with base_url kw: `docker.from_env(**kwargs)` — DockerClient.from_env accepts **kwargs passed to APIClient including base_url. Yes. One more design point worth a sentence: **why a Provider/Sandbox split** (factories vs handles) — routing picks factories cheaply, health checks without creating sandboxes; sandbox ops never consult the router. And **why registry + decorator** — third parties can ship providers in their own packages: `import my_provider` then add `- type: myprovider` to config. Also should `SandboxKit` expose `stop_all` / tracking open sandboxes? For scope, no — context manager discipline. Mention in next steps (refcounting/GC-based cleanup). Now, let me also sanity-check the local `exec` timeout: `subprocess.run(..., timeout=t)` raises TimeoutExpired and kills child. Local default_timeout wired: `t = timeout if timeout is not None else self._default_timeout`. Local exec code: ```python t = timeout if timeout is not None else self._default_timeout try: proc = subprocess.run( command, shell=True, cwd=self._resolve(cwd), env=merged, capture_output=True, timeout=t, ) except subprocess.TimeoutExpired as e: raise SandboxTimeout(f"command timed out after {t}s") from e except OSError as e: raise SandboxError(f"could not run command in {cwd}: {e}") from e ``` Good. Also: local `_resolve(cwd)` where cwd default "/" → full=root ✓. FileInfo sorted output: `sorted(out, key=lambda f: (not f.is_dir, f.name))` — dirs first then files, alphabetical. Nice deterministic. Docker list_files: also sort? ls output is already alphabetical (locale-dependent). Add same sort for determinism: `sorted(out, key=...)`. One line. OK. Now write the final answer. Structure: # Structure (design) - short intro sentence - diagram - key decisions (bulleted, ~6) - primitive contract table (small) - layout tree # Build - files with code - usage example - config yaml - routing rules recap (short) - adding a provider (e2b sketch) - tests - what I'd add next Headers with ##, keep prose minimal between code. The response will be long but the majority is code, which the user asked for. Let me write out every file completely and correctly now, being careful about imports. **core.py** final: ```python """Core contracts: errors, result types, and the Sandbox interface every provider must implement. Conventions ----------- - Filesystem paths are absolute POSIX paths *inside* the sandbox ("/" = root). A leading "/" is optional; "work/x.txt" and "/work/x.txt" are the same file. - File payloads are bytes; `str` is accepted and encoded as UTF-8. - `exec` never raises for non-zero exit codes — inspect ExecResult. (`Sandbox.run` is the assert-success convenience.) - This interface is the *intersection* of all providers. Provider-specific extras stay on the concrete class; cross-provider features are added here only if every provider can implement them. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Mapping, Optional # ---------------------------------------------------------------- errors --- class SandboxError(Exception): """Base class for everything raised by sandboxkit.""" class ProviderError(SandboxError): """Provider-level failure: auth, quota, image pull, daemon down.""" class NoRouteError(SandboxError): """No configured provider satisfies the request, or every attempt failed.""" class SandboxNotFound(SandboxError): """The sandbox (or a path inside it) does not exist.""" class SandboxTimeout(SandboxError): """An operation exceeded its deadline.""" class PathOutsideSandbox(SandboxError): """A path escaped the sandbox root (enforced by the local provider).""" class ExecFailed(SandboxError): """Raised by Sandbox.run when a command exits non-zero.""" def __init__(self, result: "ExecResult"): tail = (result.stderr or result.stdout).strip() super().__init__(f"exit code {result.exit_code}" + (f": {tail[:300]}" if tail else "")) self.result = result # ------------------------------------------------------------ result types --- @dataclass(frozen=True) class ExecResult: exit_code: int stdout: str stderr: str duration_ms: int @dataclass(frozen=True) class FileInfo: path: str # absolute in-sandbox path name: str is_dir: bool size: int # 0 when a provider can't know it cheaply modified_at: float # epoch seconds; 0.0 when unknown @dataclass(frozen=True) class SandboxSpec: """What you want from a sandbox. Drives both routing and creation.""" image: Optional[str] = None env: Mapping[str, str] = field(default_factory=dict) labels: Mapping[str, str] = field(default_factory=dict) needs: frozenset = frozenset() # required capabilities, e.g. {"network"} region: Optional[str] = None ttl_seconds: Optional[int] = None # how long the sandbox itself may live timeout_seconds: Optional[int] = None # default per-command timeout provider: Optional[str] = None # pin routing to one provider # ------------------------------------------------------------- the contract --- class Sandbox(ABC): """A live sandbox handle. All paths are absolute POSIX inside the sandbox.""" @property @abstractmethod def id(self) -> str: ... @property @abstractmethod def provider(self) -> str: ... # ---- command primitive ---- @abstractmethod def exec( self, command: str, *, cwd: str = "/", env: Optional[Mapping[str, str]] = None, timeout: Optional[float] = None, ) -> ExecResult: ... # ---- filesystem primitives ---- @abstractmethod def read_file(self, path: str) -> bytes: ... @abstractmethod def write_file(self, path: str, data) -> None: ... @abstractmethod def list_files(self, path: str = "/") -> list: ... @abstractmethod def delete_file(self, path: str, *, recursive: bool = False) -> None: ... @abstractmethod def mkdir(self, path: str) -> None: ... # ---- lifecycle ---- @abstractmethod def stop(self, *, force: bool = False) -> None: ... def is_alive(self) -> bool: try: return self.exec("true", timeout=10).exit_code == 0 except SandboxError: return False # ---- conveniences (built only on the primitives above) ---- def run(self, command: str, **kwargs) -> str: """exec + raise ExecFailed on non-zero exit + return stdout.""" result = self.exec(command, **kwargs) if result.exit_code != 0: raise ExecFailed(result) return result.stdout def write_text(self, path: str, text: str) -> None: self.write_file(path, text) def read_text(self, path: str) -> str: return self.read_file(path).decode("utf-8", "replace") def __enter__(self) -> "Sandbox": return self def __exit__(self, *exc_info) -> None: try: self.stop() except SandboxError: pass ``` **provider.py** final — as drafted, plus ProviderInfo with `capabilities: frozenset`, `regions: tuple = ()`, cost_tier. **router.py** final — as drafted with id(cfg) instance cache: ```python class Router: def __init__(self, configs): self._configs = list(configs) self._instances: dict = {} # id(config) -> Provider (same-type configs may differ in options) def _instance(self, cfg) -> Provider: key = id(cfg) if key not in self._instances: self._instances[key] = get_provider(cfg.type)(**cfg.options) return self._instances[key] ``` And route: ```python def route(self, spec) -> list: if spec.provider: # explicit pin bypasses routing return [self._instance(self._config_of(spec.provider))] ranked = [] for cfg in self._configs: if not cfg.enabled: continue try: p = self._instance(cfg) healthy = p.health() except SandboxError: healthy = False if not healthy: continue if spec.needs and not spec.needs <= p.info.capabilities: continue if spec.region and p.info.regions and spec.region not in p.info.regions: continue ranked.append((self._score(p, cfg, spec), p)) ranked.sort(key=lambda pair: -pair[0]) return [p for _, p in ranked] ``` `_config_of` raising NoRouteError (not generic SandboxError): ```python def _config_of(self, name: str) -> ProviderConfig: for cfg in self._configs: if cfg.type == name: return cfg raise NoRouteError(f"provider {name!r} is not configured") ``` Import NoRouteError from core. **api.py** final — as drafted. imports: dataclasses, Optional, core (Sandbox, SandboxSpec, SandboxError, NoRouteError), router (ProviderConfig, Router), providers imports. ```python class SandboxKit: """The only class most applications touch.""" def __init__(self, providers, *, failover: bool = True): self.router = Router(providers) self.failover = failover # ---- construction ---- @classmethod def from_dict(cls, cfg: dict) -> "SandboxKit": entries = [ ProviderConfig( type=p["type"], options=p.get("options") or {}, priority=p.get("priority", 0), enabled=p.get("enabled", True), ) for p in cfg.get("providers", []) ] return cls(entries) @classmethod def from_yaml(cls, path) -> "SandboxKit": import yaml # optional: pip install pyyaml with open(path) as f: return cls.from_dict(yaml.safe_load(f) or {}) @classmethod def auto(cls) -> "SandboxKit": """Docker first (if the daemon is reachable), local temp-dir as fallback.""" return cls([ ProviderConfig(type="docker", priority=10), ProviderConfig(type="local", priority=0), ]) # ---- main entry point ---- def sandbox(self, spec: Optional[SandboxSpec] = None, **overrides) -> Sandbox: """ Open a sandbox, routed automatically. Keyword overrides are SandboxSpec fields, e.g.: kit.sandbox(image="ubuntu:24.04", needs={"network"}) kit.sandbox(provider="local") """ if spec is None: spec = SandboxSpec(**overrides) elif overrides: spec = dataclasses.replace(spec, **overrides) attempts = [] try: candidates = self.router.route(spec) except SandboxError as e: candidates, attempts = [], [f"routing: {e}"] for provider in candidates: try: return provider.create(spec) except SandboxError as e: if not self.failover: raise attempts.append(f"{provider.info.name}: {e}") raise NoRouteError("could not create a sandbox — " + ("; ".join(attempts) if attempts else "no provider matched")) ``` **providers/local.py** final — assembled above. Let me write it fully and double check. ```python """Local provider — every sandbox is a temp directory on the host and commands run through the shell. Zero dependencies, instant, perfect for tests and local development. NOT an isolation boundary: it only jails filesystem access to its root directory, so don't route untrusted code here.""" from __future__ import annotations import os import shutil import subprocess import tempfile import time from typing import Mapping, Optional from ..core import ( ExecResult, FileInfo, PathOutsideSandbox, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("local") class LocalProvider(Provider): info = ProviderInfo( name="local", # Deliberately no "network" capability: keep local as the offline/dev # target so `needs={"network"}` work routes to docker. Remove this # comment's premise by adding "network" here if you disagree. capabilities=frozenset({"exec", "fs"}), cost_tier=0, ) def __init__(self, base_dir: Optional[str] = None): self.base_dir = base_dir def create(self, spec: SandboxSpec) -> "LocalSandbox": root = tempfile.mkdtemp(prefix="sandboxkit-", dir=self.base_dir) return LocalSandbox(root, env=dict(spec.env), default_timeout=spec.timeout_seconds) class LocalSandbox(Sandbox): def __init__(self, root: str, env: Mapping[str, str], default_timeout: Optional[float] = None): self._root = os.path.realpath(root) self._id = os.path.basename(self._root) self._env = dict(env) self._default_timeout = default_timeout self._alive = True @property def id(self) -> str: return self._id @property def provider(self) -> str: return "local" # ---- internal ---- def _resolve(self, path: str) -> str: """Map an in-sandbox path to a host path, refusing directory escapes.""" full = os.path.realpath(os.path.join(self._root, path.lstrip("/"))) if full != self._root and not full.startswith(self._root + os.sep): raise PathOutsideSandbox(path) return full # ---- exec ---- def exec(self, command: str, *, cwd: str = "/", env=None, timeout=None) -> ExecResult: if not self._alive: raise SandboxNotFound(f"sandbox {self._id} is stopped") t = timeout if timeout is not None else self._default_timeout merged = {**os.environ, **self._env, **(env or {})} started = time.monotonic() try: proc = subprocess.run( command, shell=True, cwd=self._resolve(cwd), env=merged, capture_output=True, timeout=t, ) except subprocess.TimeoutExpired as e: raise SandboxTimeout(f"command timed out after {t}s: {command!r}") from e except OSError as e: raise SandboxError(f"could not run command in {cwd!r}: {e}") from e return ExecResult( exit_code=proc.returncode, stdout=(proc.stdout or b"").decode("utf-8", "replace"), stderr=(proc.stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - started) * 1000), ) # ---- filesystem ---- def read_file(self, path: str) -> bytes: full = self._resolve(path) if os.path.isdir(full): raise SandboxError(f"{path} is a directory") try: with open(full, "rb") as f: return f.read() except FileNotFoundError: raise SandboxNotFound(path) from None def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") full = self._resolve(path) os.makedirs(os.path.dirname(full), exist_ok=True) # parent dirs, mkdir -p style with open(full, "wb") as f: f.write(data) def list_files(self, path: str = "/") -> list: full = self._resolve(path) try: entries = list(os.scandir(full)) except (FileNotFoundError, NotADirectoryError): raise SandboxNotFound(path) from None except PermissionError as e: raise SandboxError(f"cannot list {path}: {e}") from e out = [] for entry in entries: try: st = entry.stat(follow_symlinks=False) size, mtime = st.st_size, st.st_mtime except OSError: size, mtime = 0, 0.0 rel = os.path.relpath(entry.path, self._root).replace(os.sep, "/") out.append(FileInfo( path="/" + rel, name=entry.name, is_dir=entry.is_dir(follow_symlinks=False), size=size, modified_at=mtime, )) return sorted(out, key=lambda f: (not f.is_dir, f.name)) def delete_file(self, path: str, *, recursive: bool = False) -> None: full = self._resolve(path) if not os.path.exists(full) and not os.path.islink(full): raise SandboxNotFound(path) if os.path.isdir(full) and not os.path.islink(full): if recursive: shutil.rmtree(full) else: try: os.rmdir(full) except OSError as e: raise SandboxError(f"{path} is not empty; use recursive=True") from e else: os.unlink(full) def mkdir(self, path: str) -> None: os.makedirs(self._resolve(path), exist_ok=True) # ---- lifecycle ---- def stop(self, *, force: bool = False) -> None: if self._alive: shutil.rmtree(self._root, ignore_errors=True) self._alive = False def is_alive(self) -> bool: return self._alive ``` Note `NotADirectoryError` → SandboxNotFound? Listing a *file* path... scandir(file) raises NotADirectoryError. Raising SandboxNotFound is slightly off; SandboxError better. Change: except FileNotFoundError → SandboxNotFound; except NotADirectoryError → SandboxError(f"{path} is not a directory"). Fine, small. **providers/docker.py** final: ```python """Docker provider — every sandbox is a container running `sleep infinity`. File operations ride on the tar archive API (put_archive / get_archive), which is what the docker daemon natively speaks. Requires `pip install docker` and a reachable daemon. The container boundary is the isolation; no path jail needed.""" from __future__ import annotations import io import posixpath import shlex import tarfile import time from typing import Mapping, Optional try: import docker from docker.errors import APIError, ImageNotFound, NotFound except ImportError: # docker is optional; provider just reports unhealthy docker = None APIError = () ImageNotFound = () NotFound = () from ..core import ( ExecFailed, ExecResult, FileInfo, ProviderError, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("docker") class DockerProvider(Provider): info = ProviderInfo( name="docker", capabilities=frozenset({"exec", "fs", "network"}), cost_tier=1, ) def __init__( self, image: str = "ubuntu:24.04", base_url: Optional[str] = None, network: Optional[str] = None, shell = ("bash", "-c"), ): if docker is None: raise ProviderError("docker provider needs the `docker` package (pip install docker)") kwargs = {"base_url": base_url} if base_url else {} self._client = docker.from_env(**kwargs) self._default_image = image self._network = network self._shell = tuple(shell) def health(self) -> bool: if docker is None: return False try: self._client.ping() return True except Exception: return False def create(self, spec: SandboxSpec) -> "DockerSandbox": image = spec.image or self._default_image try: container = self._client.containers.run( image, command="sleep infinity", # keep the container alive as a sandbox detach=True, auto_remove=True, # gone for good once stopped network=self._network, environment=dict(spec.env), labels=dict(spec.labels), ) except ImageNotFound as e: raise ProviderError(f"image not found: {image}") from e except APIError as e: raise ProviderError(f"docker create failed: {e}") from e return DockerSandbox( container, default_env=dict(spec.env), default_timeout=spec.timeout_seconds, shell=self._shell, ) class DockerSandbox(Sandbox): def __init__(self, container, *, default_env: Mapping[str, str], default_timeout: Optional[float], shell): self._c = container self._id = container.id self._default_env = dict(default_env) self._default_timeout = default_timeout self._shell = tuple(shell) @property def id(self) -> str: return self._id[:12] @property def provider(self) -> str: return "docker" # ---- internal ---- def _norm(self, path: str) -> str: """Normalize to an absolute in-container path; '..' collapses inside the container root, so the container itself is the jail.""" return posixpath.normpath("/" + path) # ---- exec ---- def exec(self, command: str, *, cwd: str = "/", env=None, timeout=None) -> ExecResult: t = timeout if timeout is not None else self._default_timeout if t is not None: # Best-effort deadline via GNU `timeout` (present on ubuntu/debian # images). Exit code 124/137 maps to SandboxTimeout; a command that # genuinely exits 124 would be misreported — acceptable for now. inner = f"timeout -k 1 {t} {shlex.join(self._shell)} {shlex.quote(command)}" else: inner = command started = time.monotonic() try: code, streams = self._c.exec_run( [*self._shell, inner], workdir=None if cwd == "/" else self._norm(cwd), environment={**self._default_env, **(env or {})}, demux=True, ) except NotFound: raise SandboxNotFound(f"sandbox {self._id} is gone") from None except APIError as e: raise SandboxError(f"docker exec failed: {e}") from e stdout, stderr = streams result = ExecResult( exit_code=int(code), stdout=(stdout or b"").decode("utf-8", "replace"), stderr=(stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - started) * 1000), ) if t is not None and result.exit_code in (124, 137): raise SandboxTimeout(f"command timed out after {t}s: {command!r}") return result # ---- filesystem ---- def read_file(self, path: str) -> bytes: p = self._norm(path) try: stream, _stat = self._c.get_archive(p) except NotFound: raise SandboxNotFound(path) from None except APIError as e: raise SandboxError(f"docker get_archive failed: {e}") from e raw = b"".join(stream) with tarfile.open(fileobj=io.BytesIO(raw), mode="r") as tf: members = [m for m in tf.getmembers() if m.isfile()] if not members: raise SandboxError(f"{path} is not a regular file") return tf.extractfile(members[0]).read() def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") rel = self._norm(path).lstrip("/") if not rel: raise SandboxError("cannot write to '/'") buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: acc: list = [] for seg in rel.split("/")[:-1]: # embed parent dirs so put_archive creates them acc.append(seg) d = tarfile.TarInfo("/".join(acc)) d.type = tarfile.DIRTYPE d.mode = 0o755 tf.addfile(d) f = tarfile.TarInfo(rel) f.size = len(data) f.mode = 0o644 tf.addfile(f, io.BytesIO(data)) try: self._c.put_archive("/", buf.getvalue()) except (NotFound, APIError) as e: raise SandboxError(f"docker put_archive failed: {e}") from e def list_files(self, path: str = "/") -> list: p = self._norm(path) result = self.exec(f"ls -1Ap -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) out = [] for line in result.stdout.splitlines(): if not line: continue is_dir = line.endswith("/") name = line.rstrip("/") out.append(FileInfo( path=posixpath.join(p, name), name=name, is_dir=is_dir, size=0, # ls-based listing: metadata not guaranteed here modified_at=0.0, )) return sorted(out, key=lambda f: (not f.is_dir, f.name)) def delete_file(self, path: str, *, recursive: bool = False) -> None: p = self._norm(path) result = self.exec(f"rm {'-rf' if recursive else '-d'} -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) def mkdir(self, path: str) -> None: result = self.exec(f"mkdir -p -- {shlex.quote(self._norm(path))}") if result.exit_code != 0: raise ExecFailed(result) # ---- lifecycle ---- def stop(self, *, force: bool = False) -> None: try: self._c.stop(timeout=0 if force else 10) except NotFound: pass # already gone; auto_remove cleaned it up except APIError as e: raise SandboxError(f"docker stop failed: {e}") from e def is_alive(self) -> bool: try: self._c.reload() return self._c.status == "running" except Exception: return False ``` Check: `rm -d -- file` — removes file? GNU rm -d is like unlink for files and empty dirs: "rm -d, --dir remove empty directories" — for a regular file, `rm -d file` removes it (acts as unlink). Yes: "rm -d" works on files too. Good. `exec_run` with `workdir` param name — docker-py `Container.exec_run(cmd, stdout=True, stderr=True, stdin=False, tty=False, privileged=False, user='', detach=False, stream=False, socket=False, environment=None, workdir=None, demux=False)`. Yes workdir exists. Good. `shlex.join` — Python 3.8+. Fine. `posixpath.join(p, name)` where p="/" → "/name". Good. Wait — in list_files, if p is "/" then `ls -1Ap -- /` outputs entries like "bin/", "etc/", "home/", ... names fine. posixpath.join("/", "bin") = "/bin". ✓. `exec` env param on stopped container → APIError("container is not running")? docker raises APIError for exec on stopped container — we raise SandboxError. Acceptable. Could check is_alive first; skip. Also note docker exec_run demux: `code, (stdout, stderr)` — with detach=False, demux=True. Correct. One subtlety: `containers.run(..., labels=dict(spec.labels))` — docker-py accepts labels dict. Empty dict fine. ✓. `environment=dict(spec.env)` at create + per-exec merged env — both applied. ✓. **`__init__.py`**: ```python """sandboxkit: one tiny interface over many sandbox providers.""" from .api import SandboxKit from .core import ( ExecFailed, ExecResult, FileInfo, NoRouteError, PathOutsideSandbox, ProviderError, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from .provider import Provider, ProviderInfo, provider_names, register from .router import ProviderConfig, Router __version__ = "0.1.0" __all__ = [ "SandboxKit", "Sandbox", "SandboxSpec", "SandboxConfig"?? no... ] ``` __all__ list: SandboxKit, Sandbox, SandboxSpec, SandboxError, ProviderError, NoRouteError, SandboxNotFound, SandboxTimeout, PathOutsideSandbox, ExecFailed, ExecResult, FileInfo, Provider, ProviderInfo, ProviderConfig, Router, register, provider_names. Fine — I'll write it compactly. **provider.py** final text: ```python """Provider interface + registry. A Provider is a *factory* for sandboxes: cheap to instantiate, stateless apart from its client/connection, and able to answer a health check without creating anything. Routing picks providers; a Sandbox is the live handle you get back. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass from .core import Sandbox, SandboxSpec @dataclass(frozen=True) class ProviderInfo: name: str capabilities: frozenset # what this provider is intended for: {"exec", "fs", "network", ...} regions: tuple = () # empty = location-agnostic (matches any region request) cost_tier: int = 1 # lower = cheaper; tiebreaker when priorities tie class Provider(ABC): info: ProviderInfo @abstractmethod def create(self, spec: SandboxSpec) -> Sandbox: """Provision a live sandbox. Raise a SandboxError subclass on failure so the caller can fail over to the next routed provider.""" def health(self) -> bool: """Cheap liveness check used by the router. Default: always healthy.""" return True # ------------------------------------------------------------- registry --- _REGISTRY: dict = {} def register(name: str): """Class decorator: make a Provider routable under `name` (config `type: name`).""" def deco(cls): if name in _REGISTRY: raise ValueError(f"provider {name!r} is already registered") _REGISTRY[name] = cls return cls return deco def get_provider(name: str) -> type: try: return _REGISTRY[name] except KeyError: raise ValueError(f"unknown provider {name!r}; registered: {sorted(_REGISTRY)}") from None def provider_names() -> list: return sorted(_REGISTRY) ``` **router.py** full: ```python """Routing: given a SandboxSpec, produce a ranked list of candidate providers. Deliberately boring: filter (capability/region/health) -> score (priority, region bonus, cost) -> sort. The caller then tries them in order for failover. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Optional from .core import NoRouteError, SandboxError, SandboxSpec from .provider import Provider, get_provider @dataclass class ProviderConfig: type: str # registry name, e.g. "docker" options: dict = field(default_factory=dict) # provider-specific kwargs priority: int = 0 # higher = preferred (weighted 10x in scoring) enabled: bool = True class Router: def __init__(self, configs): self._configs = list(configs) self._instances = {} # id(config) -> Provider; keyed by config so # two entries of the same type can coexist # ---- instance management ---- def _instance(self, cfg: ProviderConfig) -> Provider: key = id(cfg) if key not in self._instances: self._instances[key] = get_provider(cfg.type)(**cfg.options) return self._instances[key] # ---- the decision ---- def route(self, spec: SandboxSpec) -> list: """Providers that can satisfy `spec`, best first. May be empty.""" if spec.provider: # explicit pin skips all logic return [self._instance(self._config_of(spec.provider))] ranked = [] for cfg in self._configs: if not cfg.enabled: continue try: provider = self._instance(cfg) healthy = provider.health() except SandboxError: # broken config/package → skip healthy = False if not healthy: continue if spec.needs and not spec.needs <= provider.info.capabilities: continue # can't provide what's required if spec.region and provider.info.regions and spec.region not in provider.info.regions: continue # region-agnostic providers pass ranked.append((self._score(provider, cfg, spec), provider)) ranked.sort(key=lambda pair: -pair[0]) return [p for _, p in ranked] def _config_of(self, name: str) -> ProviderConfig: for cfg in self._configs: if cfg.type == name: return cfg raise NoRouteError(f"provider {name!r} is not configured") @staticmethod def _score(provider: Provider, cfg: ProviderConfig, spec: SandboxSpec) -> int: score = cfg.priority * 10 if spec.region and spec.region in provider.info.regions: score += 5 # exact region match bonus return score - provider.info.cost_tier # cheaper wins ties ``` OK. Now the usage example, config, tests, e2b sketch — drafted above. Assemble. Also quickstart: python3 availability — ubuntu:24.04 has python3? ubuntu:24.04 includes python3.12? Ubuntu base images (ubuntu:24.04) include python3? I believe ubuntu:24.04 docker image includes python3 (24.04 does include python3 in the minimal image? Ubuntu 24.04 base image does have python3.12? Hmm — ubuntu:22.04 does NOT include python3 (minimal images have no python since 22.04? Actually ubuntu:22.04 image lacks python3; ubuntu:24.04 also lacks it I think). Let me recall: Ubuntu minimal docker images (ubuntu:22.04+) do NOT ship python3 — you have to apt-get install. Yes, that's right: `ubuntu:22.04` and `ubuntu:24.04` don't include python3. So quickstart using python3 in docker would fail! Safer quickstart demo: use shell built-ins (echo, cat, ls) — those work everywhere (ubuntu has coreutils). E.g.: ```python sbx.write_file("/work/notes.txt", "hello sandbox\n") print(sbx.run("cat /work/notes.txt")) # hello sandbox print(sbx.exec("uname -a").stdout) for f in sbx.list_files("/work"): ... ``` And a curl/network example would fail (no curl in ubuntu base). Network demo: `getent hosts example.com`? getent exists. Or omit network demo, just show needs={"network"} pinning routing (docker gets chosen; local skipped). Without actually running network commands. Good — demo routing filter, not the network itself. Or use `timeout 5 bash -c 'exec 3<>/dev/tcp/example.com/80 ...'` — too clever. Skip actual network command; or mention "e.g. run curl/pip inside". Keep the demo shell-only. Note in comments: "ubuntu base images lack python3; run `apt-get update && apt-get install -y python3` or bake your own image." For the quickstart, also demonstrate env + timeout + error semantics: ```python res = sbx.exec("echo $GREETING", env={"GREETING": "hi"}) r = sbx.exec("ls /nope") # exit code 2, doesn't raise ``` Good. Tests file (compact): ```python import pytest from sandboxkit import ExecFailed, SandboxKit, SandboxNotFound, SandboxTimeout def _docker_ok() -> bool: try: import docker docker.from_env().ping() return True except Exception: return False @pytest.fixture(params=["local"] + (["docker"] if _docker_ok() else [])) def kit(request): providers = [{"type": "local"}] if request.param == "docker": providers.insert(0, {"type": "docker", "priority": 10}) return SandboxKit.from_dict({"providers": providers}) def test_fs_roundtrip_and_exec(kit): with kit.sandbox() as s: s.write_file("/work/hello.txt", "hello") assert s.read_file("/work/hello.txt") == b"hello" assert s.run("cat /work/hello.txt").strip() == "hello" assert {f.name for f in s.list_files("/work")} == {"hello.txt"} s.delete_file("/work/hello.txt") with pytest.raises(SandboxNotFound): s.read_file("/work/hello.txt") def test_env_cwd_exit_code(kit): with kit.sandbox() as s: s.mkdir("/app") r = s.exec("pwd && echo $WHO", cwd="/app", env={"WHO": "agent"}) assert r.exit_code == 0 assert r.stdout.splitlines() == ["/app", "agent"] def test_exec_does_not_raise_on_failure(kit): with kit.sandbox() as s: assert s.exec("definitely-not-a-command").exit_code != 0 with pytest.raises(ExecFailed): s.run("definitely-not-a-command") def test_timeout(kit): with kit.sandbox() as s: with pytest.raises(SandboxTimeout): s.exec("sleep 5", timeout=1) def test_local_jail_rejects_escapes(): with SandboxKit.from_dict({"providers": [{"type": "local"}]}).sandbox() as s: with pytest.raises(PathOutsideSandbox): s.read_file("../../etc/passwd") ``` Wait — `s.exec("pwd && echo $WHO", cwd="/app")` — for docker, workdir="/app" exists (mkdir created). For local, cwd resolved. `echo $WHO` env passed via exec env. Local: env merged into subprocess env ✓. Docker: environment={"WHO": "agent"} ✓. pwd in docker workdir /app → "/app" ✓. Timeout test local: subprocess timeout=1 kills → TimeoutExpired → SandboxTimeout ✓. Docker: `timeout -k 1 1 bash -c 'sleep 5'` → SIGTERM after 1s, sleep dies (sleep doesn't catch), timeout exits 124 ✓ → SandboxTimeout ✓. `test_local_jail_rejects_escapes` — need PathOutsideSandbox import. Add. Also local: read_file("../../etc/passwd") → resolve: join(root, "../../etc/passwd") → realpath → outside → PathOutsideSandbox ✓. Also maybe a routing test: ```python def test_needs_filter_routes_to_docker_only(): kit = SandboxKit.from_dict({"providers": [ {"type": "docker", "priority": 10}, {"type": "local"}, ]}) if not _docker_ok(): pytest.skip("docker unavailable") with kit.sandbox(needs={"network"}) as s: assert s.provider == "docker" ``` Since local lacks network capability, needs={"network"} → docker only. Nice, shows auto-routing. Include. Also NoRouteError test: kit with only local, sandbox(provider="docker") → NoRouteError. Include quickly? Maybe skip to keep tests short. I'll include 5-6 tests max. One catch in `test_env_cwd_exit_code` for docker: `s.mkdir("/app")` → `mkdir -p -- /app` ✓ then workdir /app ✓. Note: docker `ls -1Ap -- /work` — the `--` end-of-options for GNU ls ✓. `uname -a` works on both. In quickstart I'll use `uname -a` or `echo`. Fine. **Quickstart final:** ```python """examples/quickstart.py — the whole SDK in ~30 lines.""" from sandboxkit import SandboxKit kit = SandboxKit.auto() # docker if its daemon is reachable, else local temp dirs # 1) Automatic routing: no provider named, the kit picks + fails over. with kit.sandbox() as sbx: print(f"routed to: {sbx.provider} ({sbx.id})") # primitive: exec — never raises on non-zero exit r = sbx.exec("echo 'hello from' $(uname -sr)") print(r.stdout.strip(), "| exit:", r.exit_code, "| took:", r.duration_ms, "ms") # primitives: filesystem sbx.write_file("/work/notes.txt", "sandbox writes files\n") print(sbx.run("cat /work/notes.txt"), end="") # run() = exec + assert success print(sbx.read_file("/work/notes.txt").decode(), end="") for f in sbx.list_files("/work"): print(" entry:", f.name, "(dir)" if f.is_dir else "(file)") # env + cwd + timeout on the exec primitive print(sbx.exec("pwd && echo $WHO", cwd="/work", env={"WHO": "agent"}).stdout.strip()) # 2) State your needs; routing filters providers that can't satisfy them. with kit.sandbox(needs={"network"}) as sbx: # local declares no network → docker wins print("network job routed to:", sbx.provider) # 3) Or pin a provider explicitly when you don't want the router involved. with kit.sandbox(provider="local") as sbx: sbx.write_file("hello.txt", "pinned to local\n") print(sbx.read_text("hello.txt")) ``` `sbx.exec("echo 'hello from' $(uname -sr)")` fine both providers. `sandbox(needs={"network"})` — if docker daemon unavailable → NoRouteError (local filtered out). Edge in demo; note in comment "requires docker daemon". OK note it. **Config yaml** (shown in prose): ```yaml # sandboxkit.yaml providers: - type: docker priority: 10 # preferred when healthy options: image: ubuntu:24.04 - type: local priority: 0 # fallback options: base_dir: /tmp/sandboxes # kit = SandboxKit.from_yaml("sandboxkit.yaml") ``` **E2B sketch** — as drafted. Present in "adding a provider" section with 1-file+1-config-line message. Keep code but flag: "method names current as of e2b 1.x — verify". Also mention third-party providers self-register: any module that calls `@register("x")` and gets imported becomes routable — config `type: x`. **Next steps** (short bullet list): - `exec` streaming (`stream=True` → iterator of chunks) for long output - long-running processes: `start()` → Process handle (pid, kill, wait, read_stream) - async: mirror the interface with async methods (or `anyio.from_thread`); E2B/Modal are async-native - warm pools + snapshots: keep N sandboxes pre-created; restore from image/snapshot - retries/backoff around provider create; per-provider circuit breakers - billing/metering hooks on the facade (wrap create/stop) - real isolation for local (bubblewrap/nsjail) if you ever route untrusted code - `list_files` metadata parity (docker could use `stat -c` for size/mtime) Also safety note: local provider is not a security boundary. I think that's a complete, well-scoped build. Now format the response. Response outline: 1. **The structure** — 1-para + diagram + core decisions (6 bullets) + primitive set table. 2. **Layout** — tree. 3. **Code** — files. 4. **Using it** — quickstart + yaml. 5. **How routing decides** — 4 numbered points (short). 6. **Adding a provider (E2B sketch)**. 7. **Tests**. 8. **What I deliberately left out / next**. Keep sections tight. The code dominates. Total will be long but justified. Let me make sure headers are used and no em-dash overuse... whatever, write naturally. One more check on imports in api.py: `from .providers import local as _local` and `from .providers import docker as _docker` — these run register decorators. But providers/__init__.py must exist (it does, empty). Also `from .core import ... NoRouteError ...`. Also unused import Optional — used in sandbox signature. dataclasses imported for replace. OK. In router.py imports: Optional unused? `spec.region` typing... I import Optional but may not use — drop it. Field import used in ProviderConfig. OK. In core.py: `field` used for env/labels defaults ✓. Mapping, Optional used ✓. docker.py: `Mapping` used in __init__ signature (default_env: Mapping) ✓, Optional used ✓. ImageNotFound imported ✓. ExecFailed used in list/delete ✓. ProviderError used ✓. SandboxTimeout used ✓. FileInfo used ✓. NotFound used ✓. tarfile/io/posixpath/shlex/time used ✓. SandboxSpec used in create signature ✓. SandboxNotFound used ✓. SandboxError used ✓. ExecResult used ✓. Sandbox base ✓. local.py: all imports used? PathOutsideSandbox ✓, SandboxNotFound ✓, SandboxTimeout ✓, SandboxError ✓, FileInfo ✓, ExecResult ✓, SandboxSpec ✓ (create signature), Sandbox ✓, Mapping/Optional ✓, tempfile/shutil/subprocess/os/time ✓. provider.py: dataclass ✓, ABC/abstractmethod ✓, Sandbox/SandboxSpec used in type hints ✓. router.py: NoRouteError ✓, SandboxError ✓, SandboxSpec ✓, Provider/get_provider ✓, dataclass/field ✓. Now — one thing: in `core.py` `Sandbox.list_files` annotation `-> list:` — with future annotations it's fine; better `-> "list[FileInfo]"`. I'll use `list[FileInfo]` string-evaluated-later; fine with future import. For docker `_norm` docstring: container is jail — normpath collapses ".." → stays inside "/". Note: `/../../etc` → normpath("/../../etc") = "/etc" — inside container ✓ (that's fine; container /etc is sandbox root's /etc). Also in `Sandbox.run` docstring fine. Also: `SandboxKit.auto()` — providers list with docker priority 10; if docker missing → health False → skipped; local remains. Also worth one line: why `frozenset` for needs/capabilities (subset operator makes capability checks a one-liner). Let me also double check `ProviderInfo` frozen dataclass with `frozenset` and `tuple` defaults — immutable ✓. `ProviderConfig` dataclass not frozen (mutable default via field) ✓. Alright — also think: should `SandboxKit.sandbox` accept ttl/env directly as kwargs ✓ via SandboxSpec fields: env=..., labels=..., ttl_seconds=..., timeout_seconds=..., image=..., region=..., provider=..., needs=... ✓. Potential pitfall: `dataclasses.replace(spec, **overrides)` — replace on frozen dataclass works ✓. Edge: `kit.sandbox()` no args → SandboxSpec() with defaults → router.route: needs empty → all healthy providers ✓. I'm confident in the code. Time to also decide: include `pyproject.toml`? Brief mention: `pip install docker pyyaml` as needed. A tiny pyproject could help but adds length; mention in one line instead. Skip file, mention deps. Docker note: `sleep infinity` — busybox `sleep infinity`? Not needed (we use ubuntu default image which has GNU coreutils sleep supporting infinity ✓). Also: docker container `command="sleep infinity"` — if image is distroless etc. it'd fail; note that image must have a shell+sleep. One-line caveat in docker.py docstring: "requires an image with a shell and `sleep` (e.g. ubuntu/debian)". Alright. Also mention Python version: 3.9+ (uses shlex.join 3.8+, walrus none). Fine. Now let me think about whether to include the tests section — yes, compact, since "build it" implies verify. Keep to ~40 lines. Final answer assembly. I'll aim for clear section headers: - ## Structure - ## The primitive contract (tiny table) - ## Layout - ## The code (each file under `####` or bold filename headers — use `**\`path\`**` or `#### sandboxkit/core.py`). I'll use `####` headers with backticked filenames. - ## Using it - ## How routing decides (recap) - ## Adding a third provider (E2B sketch) - ## Tests - ## Deliberately out of scope / next steps Keep prose in each minimal. The diagram: ``` your app / agents │ ▼ ┌──────────────────────────────────────────┐ │ SandboxKit (facade) │ │ sandbox(spec) → Sandbox │ │ - builds spec, failover loop, │ │ context-manager lifecycles │ ├──────────────────────────────────────────┤ │ Router │ │ filter: capabilities/region/health │ │ score: priority, region, cost │ │ output: ranked providers │ ├──────────────────────────────────────────┤ │ Provider registry (name → class) │ │ ┌────────────┐ ┌────────────┐ │ │ │ docker │ │ local │ ... │ │ │ containers │ │ subprocess │ │ │ └────────────┘ └────────────┘ │ ├──────────────────────────────────────────┤ │ Sandbox interface (the one contract) │ │ exec / read / write / list / delete / │ │ mkdir / stop / is_alive │ └──────────────────────────────────────────┘ ``` Good. Write it all out now. I'll be careful to keep the code blocks complete and consistent. Length estimate: prose ~40 lines + diagram + ~500 lines code + notes. That's a hefty but appropriate response for "build it". Go. # Unified Sandbox Provider SDK — Structure + Implementation ## The structure Two abstractions carry the whole design: - **`Provider`** — a *factory* for sandboxes. Cheap to instantiate, stateless, declares its capabilities, answers health checks without creating anything. Routing selects providers, never sandboxes. - **`Sandbox`** — a *live handle*. This is the one contract every provider implements. ``` your app / agents │ ▼ ┌─────────────────────────────────────────────┐ │ SandboxKit (facade) │ │ kit.sandbox(spec) → Sandbox │ │ failover loop, context-manager lifecycle │ ├─────────────────────────────────────────────┤ │ Router │ │ filter: capabilities ⊇ needs, region, │ │ health │ │ score: priority, region match, cost │ │ output: ranked provider list │ ├─────────────────────────────────────────────┤ │ Provider registry (name → class) │ │ ┌───────────┐ ┌───────────┐ │ │ │ docker │ │ local │ … more │ │ │ containers│ │ subprocess│ │ │ └───────────┘ └───────────┘ │ ├─────────────────────────────────────────────┤ │ Sandbox interface — the one contract │ │ exec · read · write · list · delete · │ │ mkdir · stop · is_alive │ └─────────────────────────────────────────────┘ ``` **Key decisions** 1. **Tiny primitive instruction set.** `exec` + 5 filesystem ops + lifecycle covers ~everything ("run this script" = `write_file` + `exec`). Small surface = thin adapters = providers you can add in an afternoon. 2. **Capabilities drive routing.** Providers declare what they're *for* (`exec`, `fs`, `network`, …). A request declares `needs`. Routing is filter → score → rank → failover. Boring and debuggable. 3. **Normal contracts everywhere:** paths are absolute POSIX *inside* the sandbox (`"/"` = root, leading `/` optional); file payloads are bytes; `exec` never raises on non-zero exit — you inspect `ExecResult` (only `run()` asserts success); one error hierarchy so failover is uniform. 4. **The interface is the intersection** of all providers. Provider-specific extras stay on the concrete class; metadata like `FileInfo.size` may be 0 where a provider can't know it cheaply. 5. **Registry + decorator.** Third parties can ship providers in their own packages; one `@register("name")` + one config line makes them routable. 6. **Sync-first** (small scope); async is the first production upgrade, noted at the end. ## Layout ``` sandboxkit/ ├── __init__.py # public API ├── core.py # errors, result types, Sandbox interface ├── provider.py # Provider interface + registry ├── router.py # routing rules ├── api.py # SandboxKit facade + config loading └── providers/ ├── __init__.py ├── local.py # subprocess + temp dir (zero deps) └── docker.py # docker daemon (pip install docker) tests/test_basics.py examples/quickstart.py ``` Dependencies: none required for core + local; `pip install docker` for the docker provider; `pyyaml` only if you use YAML config. --- ## The code #### `sandboxkit/core.py` ```python """Core contracts: errors, result types, and the Sandbox interface. Conventions ----------- - Paths are absolute POSIX paths *inside* the sandbox ("/" = root). A leading "/" is optional; "work/x.txt" and "/work/x.txt" are the same file. - File payloads are bytes; str is accepted and encoded as UTF-8. - exec() never raises on non-zero exit — inspect ExecResult. (Sandbox.run() is the assert-success convenience.) - This interface is the *intersection* of all providers. Metadata a provider can't know cheaply is zero, not missing. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Mapping, Optional # ------------------------------------------------------------------ errors class SandboxError(Exception): """Base class for everything raised by sandboxkit.""" class ProviderError(SandboxError): """Provider-level failure: auth, quota, image pull, daemon down.""" class NoRouteError(SandboxError): """No configured provider satisfies the request, or every attempt failed.""" class SandboxNotFound(SandboxError): """The sandbox (or a path inside it) does not exist.""" class SandboxTimeout(SandboxError): """An operation exceeded its deadline.""" class PathOutsideSandbox(SandboxError): """A path escaped the sandbox root (enforced by the local provider).""" class ExecFailed(SandboxError): """Raised by Sandbox.run when a command exits non-zero.""" def __init__(self, result: "ExecResult"): tail = (result.stderr or result.stdout).strip() super().__init__(f"exit code {result.exit_code}" + (f": {tail[:300]}" if tail else "")) self.result = result # ------------------------------------------------------------ result types @dataclass(frozen=True) class ExecResult: exit_code: int stdout: str stderr: str duration_ms: int @dataclass(frozen=True) class FileInfo: path: str # absolute in-sandbox path name: str is_dir: bool size: int # 0 when a provider can't know it cheaply modified_at: float # epoch seconds; 0.0 when unknown @dataclass(frozen=True) class SandboxSpec: """What you want from a sandbox. Drives both routing and creation.""" image: Optional[str] = None env: Mapping[str, str] = field(default_factory=dict) labels: Mapping[str, str] = field(default_factory=dict) needs: frozenset = frozenset() # required capabilities, e.g. {"network"} region: Optional[str] = None ttl_seconds: Optional[int] = None # how long the sandbox itself may live timeout_seconds: Optional[int] = None # default per-command timeout provider: Optional[str] = None # pin routing to one provider # ---------------------------------------------------------- the contract class Sandbox(ABC): """A live sandbox handle. All paths are absolute POSIX inside the sandbox.""" @property @abstractmethod def id(self) -> str: ... @property @abstractmethod def provider(self) -> str: ... # ---- command primitive ---- @abstractmethod def exec( self, command: str, *, cwd: str = "/", env: Optional[Mapping[str, str]] = None, timeout: Optional[float] = None, ) -> ExecResult: ... # ---- filesystem primitives ---- @abstractmethod def read_file(self, path: str) -> bytes: ... @abstractmethod def write_file(self, path: str, data) -> None: ... @abstractmethod def list_files(self, path: str = "/") -> list[FileInfo]: ... @abstractmethod def delete_file(self, path: str, *, recursive: bool = False) -> None: ... @abstractmethod def mkdir(self, path: str) -> None: ... # ---- lifecycle ---- @abstractmethod def stop(self, *, force: bool = False) -> None: ... def is_alive(self) -> bool: try: return self.exec("true", timeout=10).exit_code == 0 except SandboxError: return False # ---- conveniences (built only on the primitives above) ---- def run(self, command: str, **kwargs) -> str: """exec + raise ExecFailed on non-zero exit + return stdout.""" result = self.exec(command, **kwargs) if result.exit_code != 0: raise ExecFailed(result) return result.stdout def write_text(self, path: str, text: str) -> None: self.write_file(path, text) def read_text(self, path: str) -> str: return self.read_file(path).decode("utf-8", "replace") def __enter__(self) -> "Sandbox": return self def __exit__(self, *exc_info) -> None: try: self.stop() except SandboxError: pass ``` #### `sandboxkit/provider.py` ```python """Provider interface + registry. A Provider is a *factory* for sandboxes: cheap to instantiate, side-effect free apart from its client connection, able to answer a health check without creating anything. Routing picks providers; Sandbox is the live handle.""" from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass from .core import Sandbox, SandboxSpec @dataclass(frozen=True) class ProviderInfo: name: str capabilities: frozenset # what this provider is intended for: {"exec", "fs", "network", ...} regions: tuple = () # empty = location-agnostic (matches any region request) cost_tier: int = 1 # lower = cheaper; tiebreaker when priorities tie class Provider(ABC): info: ProviderInfo @abstractmethod def create(self, spec: SandboxSpec) -> Sandbox: """Provision a live sandbox. Raise a SandboxError subclass on failure so the caller can fail over to the next routed provider.""" def health(self) -> bool: """Cheap liveness check used by the router. Default: always healthy.""" return True # --------------------------------------------------------------- registry _REGISTRY: dict = {} def register(name: str): """Class decorator: make a Provider routable under `name` (config: `type: name`).""" def deco(cls): if name in _REGISTRY: raise ValueError(f"provider {name!r} is already registered") _REGISTRY[name] = cls return cls return deco def get_provider(name: str) -> type: try: return _REGISTRY[name] except KeyError: raise ValueError(f"unknown provider {name!r}; registered: {sorted(_REGISTRY)}") from None def provider_names() -> list: return sorted(_REGISTRY) ``` #### `sandboxkit/router.py` ```python """Routing: given a SandboxSpec, produce a ranked list of candidate providers. Deliberately boring: filter (capability / region / health) → score (priority, region bonus, cost) → sort. The facade then tries them in order for failover.""" from __future__ import annotations from dataclasses import dataclass, field from .core import NoRouteError, SandboxError, SandboxSpec from .provider import Provider, get_provider @dataclass class ProviderConfig: type: str # registry name, e.g. "docker" options: dict = field(default_factory=dict) # provider-specific kwargs priority: int = 0 # higher = preferred (10x weight) enabled: bool = True class Router: def __init__(self, configs): self._configs = list(configs) self._instances = {} # id(config) -> Provider; keyed by config so two # entries of the same type can coexist def _instance(self, cfg: ProviderConfig) -> Provider: key = id(cfg) if key not in self._instances: self._instances[key] = get_provider(cfg.type)(**cfg.options) return self._instances[key] def route(self, spec: SandboxSpec) -> list: """Providers that can satisfy `spec`, best first. May be empty.""" if spec.provider: # explicit pin skips all logic return [self._instance(self._config_of(spec.provider))] ranked = [] for cfg in self._configs: if not cfg.enabled: continue try: provider = self._instance(cfg) healthy = provider.health() except SandboxError: # broken config/package → skip healthy = False if not healthy: continue if spec.needs and not spec.needs <= provider.info.capabilities: continue # can't provide what's required if spec.region and provider.info.regions and spec.region not in provider.info.regions: continue # region-agnostic providers pass ranked.append((self._score(provider, cfg, spec), provider)) ranked.sort(key=lambda pair: -pair[0]) return [p for _, p in ranked] def _config_of(self, name: str) -> ProviderConfig: for cfg in self._configs: if cfg.type == name: return cfg raise NoRouteError(f"provider {name!r} is not configured") @staticmethod def _score(provider: Provider, cfg: ProviderConfig, spec: SandboxSpec) -> int: score = cfg.priority * 10 if spec.region and spec.region in provider.info.regions: score += 5 # exact region match bonus return score - provider.info.cost_tier # cheaper wins ties ``` #### `sandboxkit/api.py` ```python """Public facade. SandboxKit is the only class most applications touch.""" from __future__ import annotations import dataclasses from typing import Optional from .core import NoRouteError, Sandbox, SandboxError, SandboxSpec from .router import ProviderConfig, Router # import built-in providers so their @register decorators run from .providers import local as _local # noqa: F401 from .providers import docker as _docker # noqa: F401 class SandboxKit: def __init__(self, providers, *, failover: bool = True): self.router = Router(providers) self.failover = failover # ---- construction ---- @classmethod def from_dict(cls, cfg: dict) -> "SandboxKit": entries = [ ProviderConfig( type=p["type"], options=p.get("options") or {}, priority=p.get("priority", 0), enabled=p.get("enabled", True), ) for p in cfg.get("providers", []) ] return cls(entries) @classmethod def from_yaml(cls, path) -> "SandboxKit": import yaml # optional: pip install pyyaml with open(path) as f: return cls.from_dict(yaml.safe_load(f) or {}) @classmethod def auto(cls) -> "SandboxKit": """Docker first (if its daemon is reachable), local temp dirs as fallback.""" return cls([ ProviderConfig(type="docker", priority=10), ProviderConfig(type="local", priority=0), ]) # ---- the one method you call ---- def sandbox(self, spec: Optional[SandboxSpec] = None, **overrides) -> Sandbox: """ Open a sandbox, routed automatically. Keyword overrides are SandboxSpec fields, e.g.: kit.sandbox(image="ubuntu:24.04", needs={"network"}) kit.sandbox(provider="local") """ if spec is None: spec = SandboxSpec(**overrides) elif overrides: spec = dataclasses.replace(spec, **overrides) attempts = [] try: candidates = self.router.route(spec) except SandboxError as e: candidates, attempts = [], [f"routing: {e}"] for provider in candidates: try: return provider.create(spec) except SandboxError as e: if not self.failover: raise attempts.append(f"{provider.info.name}: {e}") raise NoRouteError("could not create a sandbox — " + ("; ".join(attempts) if attempts else "no provider matched")) ``` #### `sandboxkit/providers/__init__.py` ```python # Providers self-register via @register when imported (api.py imports them). ``` #### `sandboxkit/providers/local.py` ```python """Local provider — every sandbox is a temp directory on the host, commands run through the shell. Zero dependencies, instant, perfect for tests and dev. NOT an isolation boundary: it only jails filesystem access to its root dir. Don't route untrusted code here.""" from __future__ import annotations import os import shutil import subprocess import tempfile import time from typing import Mapping, Optional from ..core import ( ExecResult, FileInfo, PathOutsideSandbox, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("local") class LocalProvider(Provider): info = ProviderInfo( name="local", # Deliberately NO "network" capability: keeps local as the offline/dev # target so `needs={"network"}` work routes to docker. Declare it if # you're happy running network jobs on the host. capabilities=frozenset({"exec", "fs"}), cost_tier=0, ) def __init__(self, base_dir: Optional[str] = None): self.base_dir = base_dir def create(self, spec: SandboxSpec) -> "LocalSandbox": root = tempfile.mkdtemp(prefix="sandboxkit-", dir=self.base_dir) return LocalSandbox(root, env=dict(spec.env), default_timeout=spec.timeout_seconds) class LocalSandbox(Sandbox): def __init__(self, root: str, env: Mapping[str, str], default_timeout: Optional[float] = None): self._root = os.path.realpath(root) self._id = os.path.basename(self._root) self._env = dict(env) self._default_timeout = default_timeout self._alive = True @property def id(self) -> str: return self._id @property def provider(self) -> str: return "local" # ---- internal ---- def _resolve(self, path: str) -> str: """Map an in-sandbox path to a host path, refusing directory escapes.""" full = os.path.realpath(os.path.join(self._root, path.lstrip("/"))) if full != self._root and not full.startswith(self._root + os.sep): raise PathOutsideSandbox(path) return full # ---- exec ---- def exec(self, command: str, *, cwd: str = "/", env=None, timeout=None) -> ExecResult: if not self._alive: raise SandboxNotFound(f"sandbox {self._id} is stopped") t = timeout if timeout is not None else self._default_timeout merged = {**os.environ, **self._env, **(env or {})} started = time.monotonic() try: proc = subprocess.run( command, shell=True, cwd=self._resolve(cwd), env=merged, capture_output=True, timeout=t, ) except subprocess.TimeoutExpired as e: raise SandboxTimeout(f"command timed out after {t}s: {command!r}") from e except OSError as e: raise SandboxError(f"could not run command in {cwd!r}: {e}") from e return ExecResult( exit_code=proc.returncode, stdout=(proc.stdout or b"").decode("utf-8", "replace"), stderr=(proc.stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - started) * 1000), ) # ---- filesystem ---- def read_file(self, path: str) -> bytes: full = self._resolve(path) if os.path.isdir(full): raise SandboxError(f"{path} is a directory") try: with open(full, "rb") as f: return f.read() except FileNotFoundError: raise SandboxNotFound(path) from None def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") full = self._resolve(path) os.makedirs(os.path.dirname(full), exist_ok=True) # parent dirs, mkdir -p style with open(full, "wb") as f: f.write(data) def list_files(self, path: str = "/") -> list: full = self._resolve(path) try: entries = list(os.scandir(full)) except FileNotFoundError: raise SandboxNotFound(path) from None except NotADirectoryError: raise SandboxError(f"{path} is not a directory") from None out = [] for entry in entries: try: st = entry.stat(follow_symlinks=False) size, mtime = st.st_size, st.st_mtime except OSError: size, mtime = 0, 0.0 rel = os.path.relpath(entry.path, self._root).replace(os.sep, "/") out.append(FileInfo( path="/" + rel, name=entry.name, is_dir=entry.is_dir(follow_symlinks=False), size=size, modified_at=mtime, )) return sorted(out, key=lambda f: (not f.is_dir, f.name)) def delete_file(self, path: str, *, recursive: bool = False) -> None: full = self._resolve(path) if not os.path.exists(full) and not os.path.islink(full): raise SandboxNotFound(path) if os.path.isdir(full) and not os.path.islink(full): if recursive: shutil.rmtree(full) else: try: os.rmdir(full) except OSError as e: raise SandboxError(f"{path} is not empty; use recursive=True") from e else: os.unlink(full) def mkdir(self, path: str) -> None: os.makedirs(self._resolve(path), exist_ok=True) # ---- lifecycle ---- def stop(self, *, force: bool = False) -> None: if self._alive: shutil.rmtree(self._root, ignore_errors=True) self._alive = False def is_alive(self) -> bool: return self._alive ``` #### `sandboxkit/providers/docker.py` ```python """Docker provider — every sandbox is a container running `sleep infinity`. File operations ride on the tar archive API (put_archive / get_archive), which is what the docker daemon natively speaks. Requires `pip install docker` and a reachable daemon. The container boundary is the isolation; no path jail needed. Note: the image needs a shell, `sleep`, and coreutils (ubuntu/debian images do). """ from __future__ import annotations import io import posixpath import shlex import tarfile import time from typing import Mapping, Optional try: import docker from docker.errors import APIError, ImageNotFound, NotFound except ImportError: # docker is optional; provider reports unhealthy docker = None APIError = () ImageNotFound = () NotFound = () from ..core import ( ExecFailed, ExecResult, FileInfo, ProviderError, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from ..provider import Provider, ProviderInfo, register @register("docker") class DockerProvider(Provider): info = ProviderInfo( name="docker", capabilities=frozenset({"exec", "fs", "network"}), cost_tier=1, ) def __init__(self, image: str = "ubuntu:24.04", base_url: Optional[str] = None, network: Optional[str] = None, shell=("bash", "-c")): if docker is None: raise ProviderError("docker provider needs the `docker` package (pip install docker)") kwargs = {"base_url": base_url} if base_url else {} self._client = docker.from_env(**kwargs) self._default_image = image self._network = network self._shell = tuple(shell) def health(self) -> bool: if docker is None: return False try: self._client.ping() return True except Exception: return False def create(self, spec: SandboxSpec) -> "DockerSandbox": image = spec.image or self._default_image try: container = self._client.containers.run( image, command="sleep infinity", # keep the container alive as a sandbox detach=True, auto_remove=True, # gone for good once stopped network=self._network, environment=dict(spec.env), labels=dict(spec.labels), ) except ImageNotFound as e: raise ProviderError(f"image not found: {image}") from e except APIError as e: raise ProviderError(f"docker create failed: {e}") from e return DockerSandbox( container, default_env=dict(spec.env), default_timeout=spec.timeout_seconds, shell=self._shell, ) class DockerSandbox(Sandbox): def __init__(self, container, *, default_env: Mapping[str, str], default_timeout: Optional[float], shell): self._c = container self._id = container.id self._default_env = dict(default_env) self._default_timeout = default_timeout self._shell = tuple(shell) @property def id(self) -> str: return self._id[:12] @property def provider(self) -> str: return "docker" # ---- internal ---- def _norm(self, path: str) -> str: """Normalize to an absolute in-container path. normpath collapses '..' inside the container root — the container itself is the jail.""" return posixpath.normpath("/" + path) # ---- exec ---- def exec(self, command: str, *, cwd: str = "/", env=None, timeout=None) -> ExecResult: t = timeout if timeout is not None else self._default_timeout if t is not None: # Best-effort deadline via GNU `timeout`. Exit 124/137 maps to # SandboxTimeout; a command that genuinely exits 124 would be # misreported — acceptable for this scope. inner = f"timeout -k 1 {t} {shlex.join(self._shell)} {shlex.quote(command)}" else: inner = command started = time.monotonic() try: code, streams = self._c.exec_run( [*self._shell, inner], workdir=None if cwd == "/" else self._norm(cwd), environment={**self._default_env, **(env or {})}, demux=True, ) except NotFound: raise SandboxNotFound(f"sandbox {self._id} is gone") from None except APIError as e: raise SandboxError(f"docker exec failed: {e}") from e stdout, stderr = streams result = ExecResult( exit_code=int(code), stdout=(stdout or b"").decode("utf-8", "replace"), stderr=(stderr or b"").decode("utf-8", "replace"), duration_ms=int((time.monotonic() - started) * 1000), ) if t is not None and result.exit_code in (124, 137): raise SandboxTimeout(f"command timed out after {t}s: {command!r}") return result # ---- filesystem ---- def read_file(self, path: str) -> bytes: p = self._norm(path) try: stream, _stat = self._c.get_archive(p) except NotFound: raise SandboxNotFound(path) from None except APIError as e: raise SandboxError(f"docker get_archive failed: {e}") from e raw = b"".join(stream) with tarfile.open(fileobj=io.BytesIO(raw), mode="r") as tf: members = [m for m in tf.getmembers() if m.isfile()] if not members: raise SandboxError(f"{path} is not a regular file") return tf.extractfile(members[0]).read() def write_file(self, path: str, data) -> None: if isinstance(data, str): data = data.encode("utf-8") rel = self._norm(path).lstrip("/") if not rel: raise SandboxError("cannot write to '/'") buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tf: acc: list = [] for seg in rel.split("/")[:-1]: # embed parent dirs so put_archive creates them acc.append(seg) d = tarfile.TarInfo("/".join(acc)) d.type = tarfile.DIRTYPE d.mode = 0o755 tf.addfile(d) f = tarfile.TarInfo(rel) f.size = len(data) f.mode = 0o644 tf.addfile(f, io.BytesIO(data)) try: self._c.put_archive("/", buf.getvalue()) except (NotFound, APIError) as e: raise SandboxError(f"docker put_archive failed: {e}") from e def list_files(self, path: str = "/") -> list: p = self._norm(path) result = self.exec(f"ls -1Ap -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) out = [] for line in result.stdout.splitlines(): if not line: continue is_dir = line.endswith("/") name = line.rstrip("/") out.append(FileInfo( path=posixpath.join(p, name), name=name, is_dir=is_dir, size=0, # ls-based listing: metadata not guaranteed here modified_at=0.0, )) return sorted(out, key=lambda f: (not f.is_dir, f.name)) def delete_file(self, path: str, *, recursive: bool = False) -> None: p = self._norm(path) result = self.exec(f"rm {'-rf' if recursive else '-d'} -- {shlex.quote(p)}") if result.exit_code != 0: if "No such file or directory" in result.stderr: raise SandboxNotFound(path) raise ExecFailed(result) def mkdir(self, path: str) -> None: result = self.exec(f"mkdir -p -- {shlex.quote(self._norm(path))}") if result.exit_code != 0: raise ExecFailed(result) # ---- lifecycle ---- def stop(self, *, force: bool = False) -> None: try: self._c.stop(timeout=0 if force else 10) except NotFound: pass # already gone; auto_remove cleaned it up except APIError as e: raise SandboxError(f"docker stop failed: {e}") from e def is_alive(self) -> bool: try: self._c.reload() return self._c.status == "running" except Exception: return False ``` #### `sandboxkit/__init__.py` ```python """sandboxkit: one tiny interface over many sandbox providers.""" from .api import SandboxKit from .core import ( ExecFailed, ExecResult, FileInfo, NoRouteError, PathOutsideSandbox, ProviderError, Sandbox, SandboxError, SandboxNotFound, SandboxSpec, SandboxTimeout, ) from .provider import Provider, ProviderInfo, provider_names, register from .router import ProviderConfig, Router __version__ = "0.1.0" __all__ = [ "SandboxKit", "Sandbox", "SandboxSpec", "ExecResult", "FileInfo", "SandboxError", "ProviderError", "NoRouteError", "SandboxNotFound", "SandboxTimeout", "PathOutsideSandbox", "ExecFailed", "Provider", "ProviderInfo", "ProviderConfig", "Router", "register", "provider_names", ] ``` --- ## Using it ```python """examples/quickstart.py — the whole SDK in ~30 lines.""" from sandboxkit import SandboxKit kit = SandboxKit.auto() # docker if its daemon is reachable, else local temp dirs # 1) Automatic routing: no provider named; the kit picks and fails over. with kit.sandbox() as sbx: print(f"routed to: {sbx.provider} ({sbx.id})") # primitive: exec — never raises on non-zero exit r = sbx.exec("echo 'hello from' $(uname -sr)") print(r.stdout.strip(), "| exit:", r.exit_code, "| took:", r.duration_ms, "ms") # primitives: filesystem sbx.write_file("/work/notes.txt", "sandbox writes files\n") print(sbx.run("cat /work/notes.txt"), end="") # run() = exec + assert success print(sbx.read_file("/work/notes.txt").decode(), end="") for f in sbx.list_files("/work"): print(" entry:", f.name, "(dir)" if f.is_dir else "(file)") # env + cwd + timeout on the exec primitive print(sbx.exec("pwd && echo $WHO", cwd="/work", env={"WHO": "agent"}).stdout.strip()) # 2) State your needs; routing filters providers that can't satisfy them. # (local declares no "network" capability, so this routes to docker — # requires the docker daemon to be up.) with kit.sandbox(needs={"network"}) as sbx: print("network job routed to:", sbx.provider) # 3) Or pin a provider explicitly when you don't want the router involved. with kit.sandbox(provider="local") as sbx: sbx.write_file("hello.txt", "pinned to local\n") print(sbx.read_text("hello.txt")) ``` Config-driven, with automatic failover baked in: ```yaml # sandboxkit.yaml providers: - type: docker priority: 10 # preferred when healthy options: image: ubuntu:24.04 - type: local priority: 0 # fallback if docker is down options: base_dir: /tmp/sandboxes # kit = SandboxKit.from_yaml("sandboxkit.yaml") ``` ## How routing decides 1. **Pinned?** `spec.provider` set → use exactly that provider (create failure → `NoRouteError`, no fallback). 2. **Filter** — enabled, `provider.capabilities ⊇ spec.needs`, region compatible, `health()` passes. Broken providers (missing package, dead daemon) are skipped silently. 3. **Score** — `priority * 10 + 5 (exact region match) − cost_tier`; sort descending. 4. **Failover** — the facade tries ranked providers in order; first successful `create()` wins, failures accumulate into the `NoRouteError` message. ## Adding a third provider (e.g. E2B) One file + one config line. Sketch (verify method names against your installed `e2b` version — their API moves fast): ```python # sandboxkit/providers/e2b.py (pip install e2b, export E2B_API_KEY=...) @register("e2b") class E2BProvider(Provider): info = ProviderInfo(name="e2b", capabilities=frozenset({"exec", "fs", "network"}), cost_tier=2) def __init__(self, template: str = "ubuntu-24.04", default_ttl: int = 300): self.template, self.default_ttl = template, default_ttl def create(self, spec: SandboxSpec) -> Sandbox: from e2b import Sandbox as _E2B # lazy: heavy import sbx = _E2B.create(template=spec.image or self.template, timeout=spec.ttl_seconds or self.default_ttl, env_vars=dict(spec.env)) return _E2BAdapter(sbx) class _E2BAdapter(Sandbox): def __init__(self, sbx): self._s = sbx @property def id(self): return self._s.sandbox_id @property def provider(self): return "e2b" def exec(self, command, *, cwd="/", env=None, timeout=None): r = self._s.commands.run(command, cwd=cwd if cwd != "/" else None, envs=dict(env or {}), timeout=timeout) return ExecResult(r.exit_code, r.stdout, r.stderr, 0) def read_file(self, path): return self._s.files.read(path) def write_file(self, path, data): self._s.files.write(path, data.encode() if isinstance(data, str) else data) def list_files(self, path="/"): return [FileInfo(path=f.path, name=f.name, is_dir=f.is_dir, size=0, modified_at=0.0) for f in self._s.files.list(path)] def delete_file(self, path, *, recursive=False): self._s.files.remove(path) def mkdir(self, path): self.run(f"mkdir -p {shlex.quote(path)}") def stop(self, *, force=False): self._s.remove() ``` ```yaml - type: e2b priority: 20 # cloud sandboxes become the preferred target ``` ## Tests (same contract, parametrized over providers) ```python # tests/test_basics.py import pytest from sandboxkit import ExecFailed, PathOutsideSandbox, SandboxKit, SandboxNotFound, SandboxTimeout def _docker_ok() -> bool: try: import docker docker.from_env().ping() return True except Exception: return False @pytest.fixture(params=["local"] + (["docker"] if _docker_ok() else [])) def kit(request): providers = [{"type": "local"}] if request.param == "docker": providers.insert(0, {"type": "docker", "priority": 10}) return SandboxKit.from_dict({"providers": providers}) def test_fs_roundtrip_and_exec(kit): with kit.sandbox() as s: s.write_file("/work/hello.txt", "hello") assert s.read_file("/work/hello.txt") == b"hello" assert s.run("cat /work/hello.txt").strip() == "hello" assert {f.name for f in s.list_files("/work")} == {"hello.txt"} s.delete_file("/work/hello.txt") with pytest.raises(SandboxNotFound): s.read_file("/work/hello.txt") def test_env_cwd_exit_code(kit): with kit.sandbox() as s: s.mkdir("/app") r = s.exec("pwd && echo $WHO", cwd="/app", env={"WHO": "agent"}) assert r.exit_code == 0 assert r.stdout.splitlines() == ["/app", "agent"] def test_exec_does_not_raise_on_failure(kit): with kit.sandbox() as s: assert s.exec("definitely-not-a-command").exit_code != 0 with pytest.raises(ExecFailed): s.run("definitely-not-a-command") def test_timeout(kit): with kit.sandbox() as s: with pytest.raises(SandboxTimeout): s.exec("sleep 5", timeout=1) def test_local_jail_rejects_escapes(): kit = SandboxKit.from_dict({"providers": [{"type": "local"}]}) with kit.sandbox() as s: with pytest.raises(PathOutsideSandbox): s.read_file("../../etc/passwd") ``` ## Deliberately out of scope (natural next steps) - **Streaming exec** — `stream=True` yielding output chunks for long-running commands. - **Long-lived processes** — `start()` returning a handle (pid, kill, signal, poll) instead of run-to-completion `exec`. - **Async** — mirror the interface with `async` methods; cloud providers (E2B/Modal) are async-native, so this is the first real upgrade. - **Warm pools / snapshots** — pre-created sandbox pools to kill cold-start latency; snapshot/restore as a declared capability. - **Retries + circuit breakers** per provider around `create`. - **Metering** — wrap `create`/`stop` on the facade to bill per sandbox-second. - **Real isolation for `local`** (bubblewrap/nsjail) if you ever route untrusted code there — today it's a convenience target, not a security boundary. The core trick to preserve as you grow: keep `Sandbox` as the intersection contract and push everything provider-specific into capabilities and concrete classes — that's what keeps adding provider #4, #5, #10 cheap.

suggest a structure for unified sandbox provider sdk that ca...