feat: add hosted runtime transport and portable checkpoints

cookbook/frameworks/sandbox/README.md

@@ -127,3 +127,32 @@ limits cannot be requested. These values do not promise unlimited or durable dis
 Docker continues to require both limits. Portable checkpoint limits apply separately
 from the live filesystem size. Provider-specific setup and live acceptance are
 required before claiming hosted execution support.
+
+### Prepared hosted runtime
+
+The hosted transport requires a prepared Linux image containing this package and
+its chosen native SDK, installed into root-owned directories. Provider control
+commands run as root. Agent commands run as UID/GID 1000 with Linux
+`no_new_privs`; `/workspace` and `/runtime` must initially be empty. Provider
+daemons must run under other users and must not write the managed roots.
+
+`HostedCheckpointTransfer(provider, checkpoint_store)` implements the same portable
+archive format as Docker. Its initial transfer limit is 64 MiB per checkpoint;
+the service buffers those bytes during upload/download. Configure a smaller
+`CheckpointArchive(CheckpointLimits(...))` when needed. This limit is independent
+of the provider's live disk allocation. Large repositories and dependencies can
+exceed it and must fail completion rather than silently discard files.
+
+Capture freezes all UID 1000 processes and leaves them frozen until teardown.
+Only managed workspace and native files are restored; applications must restart.
+Quiesce database/application writers first if logical consistency is required.
+A captured computer cannot run another command. Lost capture responses can be
+recaptured during finalization without repeating model work. Imports require an
+unused computer; failures stop it. Known native credential files are excluded,
+but that exclusion is not a general secret scanner.
+
+The runtime helper is trusted control code. The initial deployment remains for
+customer-operated trusted teams; it is not a hostile multitenant security claim.
+The provider adapter must never auto-resume a stopped execution during inspection.
+Local acceptance exercises the helper with real Linux processes through a Docker
+test bridge; each hosted provider still requires live acceptance.

cookbook/frameworks/sandbox/TEST_LOG.md

@@ -163,3 +163,13 @@ Python 3.9 follow-up: 76 sandbox unit tests, 40 real Postgres cases and 12 Agent
 **Result:** 10 real Postgres tests and 96 sandbox unit tests passed. Required format and validation scripts passed. No hosted provider calls were made; this is allocation contract evidence, not hosted acceptance.
 
 ---
+
+### Hosted runtime and portable transfer
+
+**Status:** PASS
+
+**Description:** Local Docker bridge with real Postgres verifies UID/GID 1000, `no_new_privs`, inaccessible root control files, runtime events, background-writer freeze, native/workspace/symlink restoration, and rejection/teardown of imports into used computers. Provider-contract tests inject ambiguous creates, duplicate inventory, stale ownership, lost RPC/command replies, cancellation and resource mismatches.
+
+**Result:** 6 real Linux/Postgres cases passed across sync/async database adapters; hosted contract tests passed within the sandbox unit suite. No hosted platform or native-model calls were made.
+
+---

libs/agno/agno/sandbox/hosted.py

@@ -0,0 +1,290 @@
+"""Shared ownership and transport rules for explicit hosted provider adapters."""
+
+import asyncio
+import base64
+import hashlib
+import json
+import math
+from abc import abstractmethod
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Mapping, Optional, Sequence
+from uuid import uuid4
+
+from agno.sandbox.allocation import PostgresAllocationJournal, SandboxAllocationUnknown
+from agno.sandbox.base import ExecResult, SandboxError, SandboxHandle, SandboxProvider, SandboxSpec
+
+
+@dataclass(frozen=True)
+class HostedComputer:
+    provider_ref: str
+    metadata: Mapping[str, str]
+    state: str
+
+
+async def _drain(awaitable: Any) -> Any:
+    task = asyncio.ensure_future(awaitable)
+    try:
+        return await asyncio.shield(task)
+    except asyncio.CancelledError:
+        try:
+            await task
+        finally:
+            raise
+
+
+class HostedSandboxProvider(SandboxProvider):
+    """Ownership, capacity and root transport shared by hosted implementations.
+
+    Providers must supply authoritative identity/inventory and permanently stop
+    compute on deletion. No execution or create operation may be retried inside
+    the SDK. Prepared images must contain this package at a root-owned location.
+    Hosted disk/PID controls are provider defaults; checkpoint limits are separate.
+    """
+
+    backend = "abstract"
+
+    def __init__(
+        self, namespace: str, journal: PostgresAllocationJournal, *, max_allocations: int = 10, lifetime: int = 1800
+    ) -> None:
+        if not namespace or len(namespace) > 90 or type(max_allocations) is not int or max_allocations < 1:
+            raise ValueError("Hosted namespace and allocation capacity must be bounded and nonempty")
+        if type(lifetime) is not int or lifetime < 60:
+            raise ValueError("Hosted lifetime must be at least 60 seconds")
+        self.namespace = self.backend + ":" + namespace
+        self.journal, self.max_allocations, self.lifetime = journal, max_allocations, lifetime
+
+    def _validate_spec(self, spec: SandboxSpec) -> None:
+        if spec.pids_limit is not None or spec.workspace_mb is not None:
+            raise ValueError("Hosted providers require explicit pids_limit=None and workspace_mb=None")
+        if spec.network not in ("none", "internet"):
+            raise ValueError("Hosted network must be none or internet")
+
+    def _name(self, spec: SandboxSpec) -> str:
+        identity = [self.namespace, spec.agent_id, spec.session_id, spec.generation]
+        return "agno-" + hashlib.sha256(json.dumps(identity).encode()).hexdigest()[:40]
+
+    def _digest(self, spec: SandboxSpec) -> str:
+        values = {key: getattr(spec, key) for key in spec.__dataclass_fields__}
+        values["env"] = dict(spec.env)
+        values.update(namespace=self.namespace, lifetime=self.lifetime)
+        return hashlib.sha256(json.dumps(values, sort_keys=True, allow_nan=False).encode()).hexdigest()
+
+    def _metadata(self, spec: SandboxSpec, token: str) -> Dict[str, str]:
+        return dict(
+            namespace=self.namespace,
+            name=self._name(spec),
+            agent=spec.agent_id,
+            session=spec.session_id,
+            generation=str(spec.generation),
+            config=self._digest(spec),
+            token=token,
+        )
+
+    def _handle(self, computer: HostedComputer) -> SandboxHandle:
+        metadata = computer.metadata
+        if metadata.get("namespace") != self.namespace:
+            raise SandboxError("Hosted computer belongs to a different namespace")
+        try:
+            return SandboxHandle(
+                computer.provider_ref,
+                metadata["name"],
+                metadata["agent"],
+                metadata["session"],
+                int(metadata["generation"]),
+                computer.state,
+                metadata["config"],
+            )
+        except (KeyError, TypeError, ValueError) as error:
+            raise SandboxError("Hosted computer has incomplete ownership metadata") from error
+
+    async def _verify(self, computer: HostedComputer) -> SandboxHandle:
+        handle = self._handle(computer)
+        row = await self.journal.ainspect(self.namespace, handle.name)
+        if (
+            row is None
+            or row["state"] != "ready"
+            or row["token"] != computer.metadata.get("token")
+            or row["provider_ref"] != handle.provider_ref
+            or row["digest"] != handle.config_digest
+        ):
+            raise SandboxAllocationUnknown("Provider identity does not match its durable allocation")
+        return handle
+
+    async def afind(self, spec: SandboxSpec) -> Optional[SandboxHandle]:
+        self._validate_spec(spec)
+        row = await self.journal.ainspect(self.namespace, self._name(spec))
+        if row is None or row["state"] == "retired":
+            return None
+        if row["digest"] != self._digest(spec):
+            raise SandboxError("Allocation generation configuration changed")
+        if row["provider_ref"] is None:
+            matches = await _drain(self._inventory(row["token"]))
+            if len(matches) != 1:
+                raise SandboxAllocationUnknown("Pending create requires exactly one matching provider allocation")
+            computer = matches[0]
+            if dict(computer.metadata) != self._metadata(spec, row["token"]):
+                raise SandboxAllocationUnknown("Reconciled allocation metadata differs from the requested generation")
+            await self.journal._bind(self.namespace, row["name"], row["token"], computer.provider_ref)
+        else:
+            computer = await _drain(self._get(row["provider_ref"]))
+            if computer is None:
+                await self.journal._retire(self.namespace, row["name"], row["token"], row["provider_ref"])
+                return None
+        return await self._verify(computer)
+
+    async def acreate(self, spec: SandboxSpec) -> SandboxHandle:
+        self._validate_spec(spec)
+        existing = await self.afind(spec)
+        if existing is not None:
+            if existing.state != "running":
+                raise SandboxError("Hosted generation is stopped; restore into a new generation")
+            computer = await _drain(self._get(existing.provider_ref))
+            if computer is None:
+                raise SandboxError("Hosted computer disappeared during reconciliation")
+            await self._check_resources(spec, computer)
+            await self._control(existing, "init", {})
+            return existing
+        row, owner = await self.journal._reserve(
+            self.namespace, self._name(spec), self._digest(spec), max_allocations=self.max_allocations
+        )
+        if not owner:
+            raise SandboxAllocationUnknown("Another replica owns this allocation intent; reconcile before retrying")
+        computer = await _drain(self._create(spec, self._metadata(spec, row["token"])))
+        # Bind before doing further work so a failed initialization is still
+        # visible and reclaimable by a replacement coordinator.
+        await self.journal._bind(self.namespace, row["name"], row["token"], computer.provider_ref)
+        handle = await self._verify(computer)
+        if dict(computer.metadata) != self._metadata(spec, row["token"]):
+            raise SandboxAllocationUnknown("Created computer has unexpected ownership metadata")
+        try:
+            if computer.state != "running":
+                raise SandboxError("Created hosted computer is not running")
+            await self._check_resources(spec, computer)
+            await self._control(handle, "init", {})
+        except BaseException:
+            await _drain(self.adestroy(handle))
+            raise
+        return handle
+
+    async def aget(self, provider_ref: str) -> Optional[SandboxHandle]:
+        computer = await _drain(self._get(provider_ref))
+        return None if computer is None else await self._verify(computer)
+
+    async def _owned(self, handle: SandboxHandle, *, running: bool = False) -> Optional[SandboxHandle]:
+        current = await self.aget(handle.provider_ref)
+        if current is not None:
+            expected, actual = asdict(handle), asdict(current)
+            expected.pop("state")
+            actual.pop("state")
+            if expected != actual:
+                raise SandboxError("Hosted computer identity changed")
+        if running and (current is None or current.state != "running"):
+            raise SandboxError("Hosted computer is not running")
+        return current
+
+    async def _control(self, handle: SandboxHandle, operation: str, payload: Mapping[str, Any]) -> Dict[str, Any]:
+        await self._owned(handle, running=True)
+        body = json.dumps(dict(payload), allow_nan=False).encode()
+        if len(body) > 1024 * 1024:
+            raise ValueError("Hosted control request exceeds 1 MiB")
+        argv = ["python", "-I", "-m", "agno.sandbox.hosted_runtime", operation]
+        if operation != "init":
+            name = uuid4().hex + ".json"
+            await _drain(self._write(handle.provider_ref, "/var/lib/agno-control/" + name, body))
+            argv.append(name)
+        result = await _drain(self._command(handle.provider_ref, argv, timeout=float(payload.get("timeout", 120)) + 15))
+        if result.exit_code or result.timed_out or result.output_limited:
+            raise SandboxError("Hosted control operation failed: " + result.stderr[-1000:])
+        if len(result.stdout.encode()) > 4 * 1024 * 1024:
+            raise SandboxError("Hosted control reply exceeds 4 MiB")
+        value = json.loads(result.stdout)
+        if not isinstance(value, dict):
+            raise SandboxError("Hosted control reply must be an object")
+        return value
+
+    async def arpc(self, handle: SandboxHandle, operation: str, payload: bytes) -> ExecResult:
+        if operation not in ("start", "poll") or not isinstance(payload, bytes) or len(payload) > 1024 * 1024:
+            raise ValueError("Runtime RPC requires start/poll and a payload no larger than 1 MiB")
+        # Lost control replies do not stop or repeat the native execution.
+        value = await self._control(handle, operation, json.loads(payload))
+        return ExecResult(0, json.dumps(value), "")
+
+    async def aexec(
+        self,
+        handle: SandboxHandle,
+        argv: Sequence[str],
+        *,
+        timeout: float = 60,
+        max_output_bytes: int = 1_048_576,
+        stdin: bytes = b"",
+    ) -> ExecResult:
+        if isinstance(argv, (str, bytes)) or not argv or any(not isinstance(arg, str) or "\x00" in arg for arg in argv):
+            raise ValueError("Commands must be a nonempty sequence of strings")
+        if (
+            not math.isfinite(timeout)
+            or timeout <= 0
+            or type(max_output_bytes) is not int
+            or not 1 <= max_output_bytes <= 1024 * 1024
+        ):
+            raise ValueError("Hosted timeout must be positive and output limit between 1 byte and 1 MiB")
+        if not isinstance(stdin, bytes) or len(stdin) > 512 * 1024:
+            raise ValueError("Hosted command input must be bytes no larger than 512 KiB")
+        await self._owned(handle, running=True)
+        try:
+            value = await self._control(
+                handle,
+                "exec",
+                dict(
+                    argv=list(argv),
+                    timeout=timeout,
+                    max_output_bytes=max_output_bytes,
+                    stdin=base64.b64encode(stdin).decode(),
+                ),
+            )
+            result = ExecResult(**value)
+            if result.timed_out or result.output_limited:
+                await _drain(self.astop(handle))
+            return result
+        except BaseException:
+            await _drain(self.astop(handle))
+            raise
+
+    async def astop(self, handle: SandboxHandle) -> None:
+        # The initial hosted contract uses cold replacement; stop is permanent.
+        await self.adestroy(handle)
+
+    async def adestroy(self, handle: SandboxHandle) -> None:
+        row = await self.journal.ainspect(self.namespace, handle.name)
+        if row is None or row["provider_ref"] != handle.provider_ref or row["digest"] != handle.config_digest:
+            raise SandboxAllocationUnknown("Refusing to delete an unowned allocation")
+        current = await self._owned(handle) if row["state"] != "retired" else None
+        if current is not None:
+            await _drain(self._delete(handle.provider_ref))
+            remaining = await _drain(self._get(handle.provider_ref))
+            if remaining is not None and remaining.state not in ("exited", "dead"):
+                raise SandboxError("Provider did not confirm compute termination")
+        await self.journal._retire(self.namespace, handle.name, row["token"], handle.provider_ref)
+
+    @abstractmethod
+    async def _create(self, spec: SandboxSpec, metadata: Dict[str, str]) -> HostedComputer: ...
+
+    async def _check_resources(self, spec: SandboxSpec, computer: HostedComputer) -> None:
+        """Adapters with fixed template resources verify them after durable binding."""
+
+    @abstractmethod
+    async def _get(self, provider_ref: str) -> Optional[HostedComputer]: ...
+
+    @abstractmethod
+    async def _inventory(self, token: str) -> Sequence[HostedComputer]: ...
+
+    @abstractmethod
+    async def _delete(self, provider_ref: str) -> None: ...
+
+    @abstractmethod
+    async def _command(self, provider_ref: str, argv: Sequence[str], *, timeout: float) -> ExecResult: ...
+
+    @abstractmethod
+    async def _write(self, provider_ref: str, path: str, payload: bytes) -> None: ...
+
+    @abstractmethod
+    async def _read(self, provider_ref: str, path: str, *, max_bytes: int) -> bytes: ...

libs/agno/agno/sandbox/hosted_runtime.py

@@ -0,0 +1,219 @@
+"""Root-owned transport helper for prepared, trusted Linux sandbox images.
+
+The transport runs as root; harnesses and commands always run as UID/GID 1000
+with privilege elevation disabled. Root provider daemons must not write managed
+roots. Capturing freezes that user until the computer is destroyed. This is a
+filesystem checkpoint, not an application-consistent database backup.
+"""
+
+import argparse
+import ctypes
+import hashlib
+import json
+import os
+import re
+import shutil
+import signal
+import stat
+import time
+from pathlib import Path
+from typing import Any, Dict
+from uuid import uuid4
+
+from agno.sandbox.base import SandboxError
+from agno.sandbox.checkpoint import CheckpointArchive, CheckpointLimits
+
+_CONTROL = Path("/var/lib/agno-control")
+_MAX_REQUEST = 1024 * 1024
+
+
+def _control() -> None:
+    if os.geteuid() != 0:
+        raise SandboxError("Hosted transport requires root; commands drop privileges before execution")
+    for parent in (Path("/var"), Path("/var/lib")):
+        info = parent.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
+            raise SandboxError("Transport directory has an unsafe parent")
+    _CONTROL.mkdir(mode=0o700, exist_ok=True)
+    info = _CONTROL.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077:
+        raise SandboxError("Transport directory must be owned exclusively by root")
+
+
+def _path(name: str) -> Path:
+    if not re.fullmatch(r"[a-f0-9]{32}\.(json|tar)", name):
+        raise ValueError("Invalid transport object name")
+    return _CONTROL / name
+
+
+def _read(name: str, limit: int) -> bytes:
+    path = _path(name)
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+    try:
+        info = os.fstat(descriptor)
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_size > limit:
+            raise SandboxError("Invalid transport object")
+        with os.fdopen(descriptor, "rb", closefd=False) as stream:
+            value = stream.read(limit + 1)
+        if len(value) > limit:
+            raise SandboxError("Transport object exceeds its limit")
+        return value
+    finally:
+        os.close(descriptor)
+
+
+def _drop() -> None:
+    if (_CONTROL / "frozen").exists():
+        raise SandboxError("Captured compute cannot execute again; replace it from its checkpoint")
+    os.setgroups([])
+    os.setgid(1000)
+    os.setuid(1000)
+    # Linux PR_SET_NO_NEW_PRIVS: setuid binaries and file capabilities cannot
+    # turn a user command back into a root checkpoint writer.
+    libc = ctypes.CDLL(None, use_errno=True)
+    if libc.prctl(38, 1, 0, 0, 0) != 0:
+        raise OSError(ctypes.get_errno(), "Could not disable privilege elevation")
+    os.chdir("/workspace")
+    os.environ["HOME"] = "/runtime/native"
+    os.environ["CODEX_HOME"] = "/runtime/native/codex"
+
+
+def _initialize() -> Dict[str, Any]:
+    marker = _CONTROL / "initialized"
+    if marker.exists():
+        return {"initialized": True}
+    for path in (Path("/workspace"), Path("/runtime")):
+        if path.exists() or path.is_symlink():
+            if not stat.S_ISDIR(path.lstat().st_mode) or any(path.iterdir()):
+                raise SandboxError("Hosted runtime requires empty managed directories in the prepared image")
+        else:
+            path.mkdir(mode=0o700)
+        path.chmod(0o700)
+        os.chown(path, 1000, 1000)
+    Path("/runtime/native").mkdir(mode=0o700)
+    os.chown("/runtime/native", 1000, 1000)
+    marker.write_text("1")
+    return {"initialized": True}
+
+
+def _freeze() -> None:
+    (_CONTROL / "frozen").touch(mode=0o600)
+    deadline = time.monotonic() + 10
+    stable = 0
+    while time.monotonic() < deadline:
+        running = False
+        for entry in Path("/proc").iterdir():
+            if not entry.name.isdigit():
+                continue
+            try:
+                fields = dict(line.split(":", 1) for line in (entry / "status").read_text().splitlines() if ":" in line)
+                if "1000" not in fields.get("Uid", "").split():
+                    continue
+                state = fields["State"].strip().split()[0]
+                if state not in ("T", "t", "Z", "X"):
+                    running = True
+                    os.kill(int(entry.name), signal.SIGSTOP)
+            except (FileNotFoundError, ProcessLookupError):
+                continue
+        stable = 0 if running else stable + 1
+        if stable == 3:
+            return
+        time.sleep(0.02)
+    raise SandboxError("Could not freeze all managed filesystem writers")
+
+
+def _capture(request: Dict[str, Any]) -> Dict[str, Any]:
+    archive = CheckpointArchive(CheckpointLimits(**request["limits"]))
+    _freeze()
+    path = _path(request["object"])
+    archive.capture(
+        {"workspace": Path("/workspace"), "native": Path("/runtime/native")},
+        path,
+        state=request["state"],
+        exclude=request["exclude"],
+    )
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        while chunk := stream.read(1024 * 1024):
+            digest.update(chunk)
+    return {"sha256": digest.hexdigest(), "size": path.stat().st_size}
+
+
+def _restore(request: Dict[str, Any]) -> Dict[str, Any]:
+    if (_CONTROL / "used").exists() or (_CONTROL / "frozen").exists():
+        raise SandboxError("Restore requires an unused computer")
+    # Even a failed or interrupted import consumes this computer. The service
+    # stops it instead of letting a later turn observe partially restored roots.
+    (_CONTROL / "used").touch(mode=0o600)
+    roots = (Path("/workspace"), Path("/runtime/native"))
+    for path in roots:
+        if not stat.S_ISDIR(path.lstat().st_mode) or any(path.iterdir()):
+            raise SandboxError("Restore requires empty managed roots")
+    source = _path(request["object"])
+    limits = CheckpointLimits(**request["limits"])
+    payload = _read(request["object"], limits.max_bytes)
+    if len(payload) != request["size"] or hashlib.sha256(payload).hexdigest() != request["sha256"]:
+        raise SandboxError("Uploaded checkpoint digest mismatch")
+    del payload
+    destination = _CONTROL / uuid4().hex
+    archive = CheckpointArchive(limits)
+    try:
+        state = archive.restore(source, destination)
+        for root, target in zip(("workspace", "native"), roots):
+            restored = destination / root
+            for directory, dirs, files in os.walk(restored, followlinks=False):
+                os.chown(directory, 1000, 1000, follow_symlinks=False)
+                for name in dirs + files:
+                    os.chown(Path(directory) / name, 1000, 1000, follow_symlinks=False)
+            target.rmdir()
+            os.rename(restored, target)
+        return {"state": state}
+    finally:
+        shutil.rmtree(destination, ignore_errors=True)
+        source.unlink(missing_ok=True)
+
+
+def _dispatch(operation: str, request: Dict[str, Any]) -> Dict[str, Any]:
+    if operation == "init":
+        return _initialize()
+    if not (_CONTROL / "initialized").exists():
+        raise SandboxError("Hosted runtime is not initialized")
+    if operation == "capture":
+        return _capture(request)
+    if operation == "restore":
+        return _restore(request)
+    if operation == "remove":
+        _path(request["object"]).unlink(missing_ok=True)
+        return {}
+    if operation not in ("exec", "start", "poll"):
+        raise ValueError("Unsupported hosted operation")
+    (_CONTROL / "used").touch(mode=0o600)
+    _drop()
+    if operation == "exec":
+        from agno.sandbox._exec import _execute
+
+        return _execute(request)
+    from agno.sandbox.runtime import _poll, _start
+
+    return (
+        _start(Path("/runtime"), request)
+        if operation == "start"
+        else _poll(Path("/runtime"), request["execution_id"], request["cursor"])
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Hosted sandbox control transport")
+    parser.add_argument("operation", choices=("init", "exec", "start", "poll", "capture", "restore", "remove"))
+    parser.add_argument("request", nargs="?")
+    args = parser.parse_args()
+    _control()
+    request = {}
+    if args.request is not None:
+        request = json.loads(_read(args.request, _MAX_REQUEST))
+        _path(args.request).unlink()
+    print(json.dumps(_dispatch(args.operation, request), allow_nan=False))
+
+
+if __name__ == "__main__":
+    main()

libs/agno/agno/sandbox/hosted_transfer.py

@@ -0,0 +1,90 @@
+"""Portable bounded checkpoints transported through hosted filesystem APIs."""
+
+import hashlib
+import tempfile
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any, Mapping, Optional, Sequence
+from uuid import uuid4
+
+from agno.sandbox.base import SandboxError, SandboxHandle, _sync
+from agno.sandbox.checkpoint import CheckpointArchive, CheckpointLimits, CheckpointRef, FileCheckpointStore, _thread
+from agno.sandbox.hosted import HostedSandboxProvider, _drain
+from agno.sandbox.transfer import DEFAULT_NATIVE_EXCLUSIONS
+
+
+class HostedCheckpointTransfer:
+    """Capture managed roots, excluding known native credential files.
+
+    The initial transport buffers at most 64 MiB of archive bytes per operation.
+    Capture freezes user processes permanently; callers destroy this computer
+    after publishing recovery and restore into a fresh computer for the next turn.
+    Quiesce transactional applications before capture when logical consistency is
+    required. Mounted external storage and running processes are not checkpointed.
+    """
+
+    def __init__(
+        self,
+        provider: HostedSandboxProvider,
+        store: FileCheckpointStore,
+        *,
+        archive: Optional[CheckpointArchive] = None,
+    ) -> None:
+        self.provider, self.store = provider, store
+        self.archive = archive or CheckpointArchive(CheckpointLimits(max_bytes=64 * 1024 * 1024))
+        if self.archive.limits.max_bytes > 64 * 1024 * 1024:
+            raise ValueError("Hosted archive transfer currently supports at most 64 MiB")
+
+    async def acapture(
+        self, handle: SandboxHandle, *, state: Mapping[str, Any], exclude: Sequence[str] = DEFAULT_NATIVE_EXCLUSIONS
+    ) -> CheckpointRef:
+        name = uuid4().hex + ".tar"
+        metadata = await self.provider._control(
+            handle,
+            "capture",
+            dict(object=name, state=dict(state), exclude=list(exclude), limits=asdict(self.archive.limits)),
+        )
+        reference = CheckpointRef(**metadata)
+        if reference.size > self.archive.limits.max_bytes:
+            raise SandboxError("Hosted checkpoint exceeds its transfer limit")
+        await self.provider._owned(handle, running=True)
+        payload = await _drain(
+            self.provider._read(handle.provider_ref, "/var/lib/agno-control/" + name, max_bytes=reference.size)
+        )
+        if len(payload) != reference.size or hashlib.sha256(payload).hexdigest() != reference.sha256:
+            raise SandboxError("Hosted checkpoint download differs from the captured object")
+        with tempfile.TemporaryDirectory(prefix=".hosted-capture-", dir=self.store.root) as directory:
+            path = Path(directory) / "checkpoint.tar"
+            await _thread(path.write_bytes, payload)
+            await self.archive.avalidate(path)
+            result = await self.store.aput(path)
+        await self.provider._control(handle, "remove", {"object": name})
+        return result
+
+    def capture(
+        self, handle: SandboxHandle, *, state: Mapping[str, Any], exclude: Sequence[str] = DEFAULT_NATIVE_EXCLUSIONS
+    ) -> CheckpointRef:
+        return _sync(self.acapture(handle, state=state, exclude=exclude))
+
+    async def arestore(self, handle: SandboxHandle, reference: CheckpointRef) -> Mapping[str, Any]:
+        if reference.size > self.archive.limits.max_bytes:
+            raise SandboxError("Hosted checkpoint exceeds its transfer limit")
+        source = await self.store.averify(reference)
+        state = await self.archive.avalidate(source)
+        payload = await _thread(source.read_bytes)
+        name = uuid4().hex + ".tar"
+        await self.provider._owned(handle, running=True)
+        try:
+            await _drain(self.provider._write(handle.provider_ref, "/var/lib/agno-control/" + name, payload))
+            response = await self.provider._control(
+                handle, "restore", dict(object=name, limits=asdict(self.archive.limits), **asdict(reference))
+            )
+            if response.get("state") != state:
+                raise SandboxError("Hosted restored manifest differs from the verified checkpoint")
+            return state
+        except BaseException:
+            await _drain(self.provider.astop(handle))
+            raise
+
+    def restore(self, handle: SandboxHandle, reference: CheckpointRef) -> Mapping[str, Any]:
+        return _sync(self.arestore(handle, reference))

libs/agno/tests/integration/sandbox/test_hosted.py

@@ -0,0 +1,206 @@
+"""Real Linux hosted transport semantics through a local Docker test bridge."""
+
+import json
+import os
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+
+from agno.sandbox.allocation import PostgresAllocationJournal
+from agno.sandbox.base import ExecResult, SandboxError, SandboxSpec
+from agno.sandbox.checkpoint import FileCheckpointStore
+from agno.sandbox.client import SandboxRuntimeClient
+from agno.sandbox.docker import DockerSandboxProvider
+from agno.sandbox.hosted import HostedComputer, HostedSandboxProvider
+from agno.sandbox.hosted_transfer import HostedCheckpointTransfer
+from tests.integration.sandbox.test_postgres import URL
+from tests.integration.sandbox.test_postgres import store as _store_fixture
+
+store = _store_fixture
+IMAGE = os.environ.get("AGNO_TEST_HOSTED_IMAGE")
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.skipif(not URL or not IMAGE, reason="Set hosted image and sandbox Postgres URL"),
+]
+
+
+class LocalHosted(HostedSandboxProvider):
+    backend = "local-test"
+
+    def __init__(self, namespace, journal):
+        super().__init__(namespace, journal)
+        self.docker = DockerSandboxProvider(namespace)
+
+    async def _create(self, spec, metadata):
+        args = [
+            "container",
+            "run",
+            "-d",
+            "--user",
+            "0:0",
+            "--network",
+            "none",
+            "--memory",
+            "512m",
+            "--pids-limit",
+            "256",
+        ]
+        for key, value in metadata.items():
+            args.extend(["--label", key + "=" + value])
+        for key, value in spec.env.items():
+            args.extend(["--env", key + "=" + value])
+        code, out, err = await self.docker._docker(*args, spec.image)
+        assert code == 0, err
+        return await self._get(out.decode().strip())
+
+    async def _get(self, provider_ref):
+        code, out, err = await self.docker._docker("container", "inspect", provider_ref)
+        if code:
+            if b"No such" in err:
+                return None
+            raise SandboxError(err.decode())
+        record = json.loads(out)[0]
+        return HostedComputer(record["Id"], record["Config"]["Labels"], record["State"]["Status"])
+
+    async def _inventory(self, token):
+        code, out, err = await self.docker._docker("container", "ls", "-aq", "--filter", "label=token=" + token)
+        assert code == 0, err
+        return [await self._get(ref) for ref in out.decode().splitlines()]
+
+    async def _delete(self, provider_ref):
+        code, _, err = await self.docker._docker("container", "rm", "-f", provider_ref)
+        assert code == 0, err
+
+    async def _command(self, provider_ref, argv, *, timeout):
+        code, out, err = await self.docker._docker(
+            "container", "exec", "--user", "0:0", provider_ref, *argv, timeout=timeout
+        )
+        return ExecResult(code, out.decode(), err.decode())
+
+    async def _write(self, provider_ref, path, payload):
+        code, _, err = await self.docker._docker(
+            "container",
+            "exec",
+            "-i",
+            "--user",
+            "0:0",
+            provider_ref,
+            "python",
+            "-I",
+            "-c",
+            "import sys; from pathlib import Path; Path(sys.argv[1]).write_bytes(sys.stdin.buffer.read())",
+            path,
+            input_data=payload,
+        )
+        assert code == 0, err
+
+    async def _read(self, provider_ref, path, *, max_bytes):
+        code, out, err = await self.docker._docker(
+            "container",
+            "exec",
+            "--user",
+            "0:0",
+            provider_ref,
+            "python",
+            "-I",
+            "-c",
+            "import sys; from pathlib import Path; p=Path(sys.argv[1]); assert p.stat().st_size <= int(sys.argv[2]); sys.stdout.buffer.write(p.read_bytes())",
+            path,
+            str(max_bytes),
+        )
+        assert code == 0, err
+        return out
+
+
+@pytest_asyncio.fixture
+async def hosted(store, tmp_path):
+    journal = PostgresAllocationJournal(store, table_name=store.bindings.name + "_alloc")
+    await journal.asetup()
+    provider = LocalHosted(uuid4().hex, journal)
+    spec = SandboxSpec("a", "s", 1, IMAGE, pids_limit=None, workspace_mb=None, env={"AGNO_SANDBOX_TEST_RUNTIME": "1"})
+    handles = []
+
+    async def create(generation=1):
+        handle = await provider.acreate(replace(spec, generation=generation))
+        handles.append(handle)
+        return handle
+
+    try:
+        yield provider, create, HostedCheckpointTransfer(provider, FileCheckpointStore(tmp_path / "objects"))
+    finally:
+        for handle in handles:
+            await provider.adestroy(handle)
+        await store._transaction(lambda connection: journal.table.drop(connection))
+
+
+async def test_commands_are_unprivileged_and_control_files_inaccessible(hosted):
+    provider, create, _ = hosted
+    handle = await create()
+    result = await provider.aexec(
+        handle,
+        [
+            "python",
+            "-c",
+            "import os; from pathlib import Path; print(os.getuid(), os.getgid()); print(next(l for l in Path('/proc/self/status').read_text().splitlines() if l.startswith('NoNewPrivs:'))); assert not os.access('/var/lib/agno-control', os.R_OK); assert not os.access('/usr/local/lib/python3.12/site-packages/agno', os.W_OK)",
+        ],
+    )
+    assert result.exit_code == 0 and "1000 1000" in result.stdout and "NoNewPrivs:\t1" in result.stdout
+
+
+async def test_runtime_and_frozen_checkpoint_restore_into_replacement(hosted):
+    provider, create, transfer = hosted
+    first = await create()
+    client = SandboxRuntimeClient(provider)
+    execution = await client.astart(first, execution_id="e", run_id="r", sdk="fixture", input="hello")
+    pages = [page async for page in client.apages(execution, timeout=20)]
+    assert pages[-1].result["type"] == "execution_finished"
+    writer = "import time; from pathlib import Path\np=Path('/workspace/ticks')\nwhile True:\n p.write_text(str(time.monotonic())); time.sleep(.01)"
+    result = await provider.aexec(
+        first,
+        [
+            "python",
+            "-c",
+            "import subprocess; from pathlib import Path; Path('/runtime/native/native.txt').write_text('native'); Path('/workspace/link').symlink_to('fixture-effects.txt'); subprocess.Popen(['python','-c',"
+            + repr(writer)
+            + "], start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)",
+        ],
+    )
+    assert result.exit_code == 0
+    reference = await transfer.acapture(first, state={"turn": 1})
+    result = await provider._command(
+        first.provider_ref,
+        [
+            "python",
+            "-I",
+            "-c",
+            "import time; from pathlib import Path; p=Path('/workspace/ticks'); a=p.read_bytes(); time.sleep(.1); assert a == p.read_bytes()",
+        ],
+        timeout=5,
+    )
+    assert result.exit_code == 0, result.stderr
+    with pytest.raises(SandboxError, match="Captured compute"):
+        await provider.aexec(first, ["true"])
+    second = await create(2)
+    assert await transfer.arestore(second, reference) == {"turn": 1}
+    restored = await provider.aexec(
+        second,
+        [
+            "python",
+            "-c",
+            "from pathlib import Path; assert Path('link').read_text() == 'r\\n'; assert Path('/runtime/native/native.txt').read_text() == 'native'; assert Path('ticks').exists(); print('restored')",
+        ],
+    )
+    assert restored.exit_code == 0 and restored.stdout == "restored\n"
+
+
+async def test_restore_rejects_used_computer_and_stops_it(hosted):
+    provider, create, transfer = hosted
+    source = await create()
+    reference = await transfer.acapture(source, state={})
+    target = await create(2)
+    assert (await provider.aexec(target, ["true"])).exit_code == 0
+    with pytest.raises(SandboxError, match="unused"):
+        await transfer.arestore(target, reference)
+    assert await provider._get(target.provider_ref) is None

libs/agno/tests/unit/sandbox/test_hosted.py

@@ -0,0 +1,177 @@
+import asyncio
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+
+from agno.sandbox.allocation import SandboxAllocationUnknown
+from agno.sandbox.base import ExecResult, SandboxError, SandboxSpec
+from agno.sandbox.hosted import HostedComputer, HostedSandboxProvider
+
+
+class Journal:
+    def __init__(self):
+        self.rows = {}
+
+    async def ainspect(self, namespace, name):
+        row = self.rows.get((namespace, name))
+        return dict(row) if row else None
+
+    async def _reserve(self, namespace, name, digest, *, max_allocations):
+        row = self.rows.get((namespace, name))
+        if row and row["state"] != "retired":
+            return dict(row), False
+        row = dict(name=name, digest=digest, token=uuid4().hex, state="pending", provider_ref=None)
+        self.rows[namespace, name] = row
+        return dict(row), True
+
+    async def _bind(self, namespace, name, token, provider_ref):
+        row = self.rows[namespace, name]
+        assert row["token"] == token
+        row.update(provider_ref=provider_ref, state="ready")
+
+    async def _retire(self, namespace, name, token, provider_ref):
+        row = self.rows[namespace, name]
+        assert row["token"] == token and row["provider_ref"] == provider_ref
+        row["state"] = "retired"
+
+
+class Provider(HostedSandboxProvider):
+    backend = "fixture"
+
+    def __init__(self):
+        super().__init__("deployment", Journal())
+        self.computers = {}
+        self.creates = 0
+        self.lost_create = False
+        self.hide_inventory = False
+        self.failed_command = False
+        self.deletes = []
+
+    async def _create(self, spec, metadata):
+        self.creates += 1
+        computer = HostedComputer(uuid4().hex, metadata, "running")
+        self.computers[computer.provider_ref] = computer
+        if self.lost_create:
+            raise TimeoutError("lost create")
+        return computer
+
+    async def _get(self, provider_ref):
+        return self.computers.get(provider_ref)
+
+    async def _inventory(self, token):
+        return [] if self.hide_inventory else [v for v in self.computers.values() if v.metadata["token"] == token]
+
+    async def _delete(self, provider_ref):
+        self.deletes.append(provider_ref)
+        self.computers.pop(provider_ref)
+
+    async def _command(self, provider_ref, argv, *, timeout):
+        if self.failed_command:
+            raise TimeoutError("lost command")
+        return ExecResult(0, "{}", "")
+
+    async def _write(self, provider_ref, path, payload):
+        pass
+
+    async def _read(self, provider_ref, path, *, max_bytes):
+        return b""
+
+
+def spec():
+    return SandboxSpec("a", "s", 1, "prepared", pids_limit=None, workspace_mb=None)
+
+
+@pytest.mark.asyncio
+async def test_lost_create_is_reconciled_without_reallocation():
+    provider = Provider()
+    provider.lost_create = True
+    with pytest.raises(TimeoutError):
+        await provider.acreate(spec())
+    provider.hide_inventory = True
+    with pytest.raises(SandboxAllocationUnknown):
+        await provider.acreate(spec())
+    assert provider.creates == 1
+    provider.hide_inventory = False
+    found = await provider.afind(spec())
+    assert found == await provider.acreate(spec())
+    assert provider.creates == 1
+
+
+@pytest.mark.asyncio
+async def test_duplicate_inventory_and_foreign_metadata_fail_closed():
+    provider = Provider()
+    provider.lost_create = True
+    with pytest.raises(TimeoutError):
+        await provider.acreate(spec())
+    original = next(iter(provider.computers.values()))
+    provider.computers["second"] = replace(original, provider_ref="second")
+    with pytest.raises(SandboxAllocationUnknown):
+        await provider.afind(spec())
+    provider.computers.pop("second")
+    provider.computers[original.provider_ref] = replace(original, metadata={**original.metadata, "session": "foreign"})
+    with pytest.raises(SandboxAllocationUnknown):
+        await provider.afind(spec())
+    assert not provider.deletes
+
+
+@pytest.mark.asyncio
+async def test_control_failure_keeps_compute_but_command_failure_stops_it():
+    provider = Provider()
+    handle = await provider.acreate(spec())
+    provider.failed_command = True
+    with pytest.raises(TimeoutError):
+        await provider.arpc(handle, "poll", b"{}")
+    assert not provider.deletes
+    with pytest.raises(TimeoutError):
+        await provider.aexec(handle, ["example"])
+    assert provider.deletes == [handle.provider_ref]
+    await provider.adestroy(handle)
+    assert provider.deletes == [handle.provider_ref]
+
+
+@pytest.mark.asyncio
+async def test_cancelled_create_drains_then_reconciles():
+    provider = Provider()
+    entered, finish = asyncio.Event(), asyncio.Event()
+    original = provider._create
+
+    async def delayed(*args):
+        entered.set()
+        await finish.wait()
+        return await original(*args)
+
+    provider._create = delayed
+    task = asyncio.create_task(provider.acreate(spec()))
+    await entered.wait()
+    task.cancel()
+    await asyncio.sleep(0)
+    assert not task.done()
+    finish.set()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert await provider.afind(spec()) is not None
+    assert provider.creates == 1
+
+
+@pytest.mark.asyncio
+async def test_stale_generation_never_deletes_live_compute():
+    provider = Provider()
+    handle = await provider.acreate(spec())
+    with pytest.raises(SandboxError, match="identity"):
+        await provider.adestroy(replace(handle, generation=2))
+    assert not provider.deletes
+
+
+@pytest.mark.asyncio
+async def test_resource_validation_failure_reclaims_bound_allocation():
+    provider = Provider()
+
+    async def reject(*args):
+        raise SandboxError("template resources differ")
+
+    provider._check_resources = reject
+    with pytest.raises(SandboxError, match="resources"):
+        await provider.acreate(spec())
+    assert len(provider.deletes) == 1
+    assert next(iter(provider.journal.rows.values()))["state"] == "retired"