feat: add reusable sandbox snapshots

CHANGELOG.md

@@ -19,6 +19,8 @@ Versions follow [Semantic Versioning](https://semver.org/).
 
 ### Added
 
+- Reusable disk and full snapshot create, get, list, and delete operations,
+  plus sandbox restoration through `CreateSandboxRequest.snapshot_id`.
 - Sandbox labels on create and response models, AND-matched label filtering in
   `list_sandboxes()`, and label merging through `SandboxInstance.set_labels()`.
 

README.md

@@ -85,6 +85,21 @@ sandbox.set_labels({"environment": "production"})
 List filters use AND semantics. `set_labels()` merges the supplied keys into
 the existing label map.
 
+## Capture and restore snapshots
+
+```python
+snapshot = sandbox.create_snapshot(
+    CreateSnapshotRequest(type=SnapshotType.DISK, name="configured-workspace")
+)
+snapshot = client.snapshots.get(snapshot.id)  # poll until READY
+restored = client.create_sandbox(CreateSandboxRequest(snapshot_id=snapshot.id))
+```
+
+Disk snapshots capture the filesystem. Full snapshots also retain memory and
+must be restored with their captured CPU, memory, and disk sizes. Snapshot
+capture requires a running source sandbox with no attached external disks.
+Each account may retain up to five snapshots across both types.
+
 ## Delegate access to one sandbox
 
 The owner can create one token for a sandbox. Creation and rotation return the
@@ -384,6 +399,7 @@ Runnable examples live under [`examples/`](examples/):
 - [HTTP execution server](examples/execution_server/README.md)
 - [Command streaming](examples/command_streaming/README.md)
 - [Files and snapshots](examples/files_and_snapshots/README.md)
+- [Reusable snapshots](examples/reusable_snapshots/README.md)
 - [Ingress preview](examples/ingress_preview/README.md)
 - [Private overlay network](examples/network/README.md)
 - [Custom template](examples/custom_template/README.md)
@@ -428,7 +444,7 @@ to run before opening a pull request.
 ```text
 src/createos/client.py      client configuration and account-level operations
 src/createos/instance.py    stateful sandbox lifecycle and command operations
-src/createos/services.py    files, processes, desktop, templates, disks, networks
+src/createos/services.py    files, processes, desktop, templates, disks, networks, snapshots
 src/createos/models.py      public requests, responses, options, and enums
 src/createos/_transport.py  HTTP, authentication, retries, and JSend handling
 src/createos/_streams.py    NDJSON, SSE, command, process, and binary streams

examples/reusable_snapshots/README.md

@@ -0,0 +1,11 @@
+# Reusable snapshots
+
+Capture a reusable disk snapshot, wait for it to become ready, and restore a
+new sandbox from it.
+
+```sh
+export CREATEOS_API_KEY="your-api-key"
+python examples/reusable_snapshots/main.py
+```
+
+The example creates live resources and cleans them up when it finishes.

examples/reusable_snapshots/main.py

@@ -0,0 +1,60 @@
+"""Capture a reusable disk snapshot and restore a sandbox from it."""
+
+import os
+import time
+
+from createos import (
+    Client,
+    CreateSandboxRequest,
+    CreateSnapshotRequest,
+    Snapshot,
+    SnapshotStatus,
+    SnapshotType,
+)
+
+
+def wait_until_ready(client: Client, snapshot_id: str) -> Snapshot:
+    """Poll a snapshot for up to five minutes."""
+    deadline = time.monotonic() + 300
+    while time.monotonic() < deadline:
+        snapshot = client.snapshots.get(snapshot_id)
+        if snapshot.status is SnapshotStatus.READY:
+            return snapshot
+        if snapshot.status is SnapshotStatus.FAILED:
+            raise RuntimeError(f"snapshot {snapshot_id} failed")
+        time.sleep(1)
+    raise TimeoutError(f"snapshot {snapshot_id} did not become ready")
+
+
+def main() -> None:
+    """Create, capture, restore, and clean up live resources."""
+    with Client(api_key=os.environ["CREATEOS_API_KEY"]) as client:
+        source = client.create_sandbox(
+            CreateSandboxRequest(shape="s-1vcpu-1gb", rootfs="alpine:3.20")
+        )
+        snapshot = None
+        restored = None
+        try:
+            snapshot = source.create_snapshot(
+                CreateSnapshotRequest(
+                    type=SnapshotType.DISK, name="configured-workspace"
+                )
+            )
+            snapshot = wait_until_ready(client, snapshot.id)
+            restored = client.create_sandbox(
+                CreateSandboxRequest(snapshot_id=snapshot.id)
+            )
+            print(f"restored sandbox {restored.id} from snapshot {snapshot.id}")
+        finally:
+            if restored is not None:
+                restored.destroy()
+            source.destroy()
+            if snapshot is not None and snapshot.status in {
+                SnapshotStatus.READY,
+                SnapshotStatus.FAILED,
+            }:
+                client.snapshots.delete(snapshot.id)
+
+
+if __name__ == "__main__":
+    main()

src/createos/__init__.py

@@ -41,6 +41,7 @@
     ComputerWindowResizeRequest,
     CreateSandboxRequest,
     CreateSandboxResponse,
+    CreateSnapshotRequest,
     DestroyedResponse,
     DetachDiskOptions,
     Disk,
@@ -94,6 +95,10 @@
     SandboxDisk,
     SandboxStatus,
     Shape,
+    Snapshot,
+    SnapshotDeletedResponse,
+    SnapshotStatus,
+    SnapshotType,
     StrEnum,
     Template,
     TemplateCreateRequest,

src/createos/client.py

@@ -29,6 +29,7 @@
 from .services import (
     DisksService,
     NetworksService,
+    SnapshotsService,
     TemplatesService,
     _fetch_all,
     _many,
@@ -69,6 +70,7 @@ def __init__(
         self.templates = TemplatesService(self._transport)
         self.networks = NetworksService(self._transport)
         self.disks = DisksService(self._transport)
+        self.snapshots = SnapshotsService(self._transport)
 
     @property
     def base_url(self) -> str:
@@ -134,6 +136,7 @@ def create_sandbox(
         )
         created = _one(CreateSandboxResponse, data)
         sandbox = Sandbox(
+            snapshot_id=created.snapshot_id,
             id=created.id,
             status=created.status,
             ip_address=created.ip_address,

src/createos/instance.py

@@ -13,6 +13,7 @@
 from .models import (
     AttachDiskOptions,
     BandwidthView,
+    CreateSnapshotRequest,
     DestroyedResponse,
     DetachDiskOptions,
     DiskAttachment,
@@ -28,6 +29,7 @@
     SandboxAccessTokenMetadata,
     SandboxDisk,
     SandboxStatus,
+    Snapshot,
     WaitOptions,
 )
 from .services import (
@@ -145,6 +147,15 @@ def resume(self) -> SandboxInstance:
         """Restore the paused sandbox."""
         return self._lifecycle("/resume")
 
+    def create_snapshot(self, request: CreateSnapshotRequest) -> Snapshot:
+        """Start a reusable snapshot capture from this running sandbox."""
+        return _one(
+            Snapshot,
+            self._transport.request(
+                "POST", self._path("/snapshots"), body=request
+            ),
+        )
+
     def fork(
         self, request: ForkSandboxRequest | None = None
     ) -> SandboxInstance:

src/createos/models.py

@@ -33,6 +33,18 @@ class SandboxStatus(StrEnum):
     FAILED = "failed"
 
 
+class SnapshotType(StrEnum):
+    DISK = "disk"
+    FULL = "full"
+
+
+class SnapshotStatus(StrEnum):
+    CREATING = "creating"
+    READY = "ready"
+    FAILED = "failed"
+    DELETING = "deleting"
+
+
 class ExecStreamEventType(StrEnum):
     STDOUT = "stdout"
     STDERR = "stderr"
@@ -235,7 +247,8 @@ class DiskAttachment(Model):
 
 @dataclass(slots=True)
 class CreateSandboxRequest(Model):
-    shape: str = ""
+    snapshot_id: str = _json("snapshot_id", omit_default=True, default="")
+    shape: str = _json("shape", omit_default=True, default="")
     rootfs: str = _json("rootfs", omit_default=True, default="")
     name: str = _json("name", omit_default=True, default="")
     labels: dict[str, str] = _json(
@@ -270,6 +283,12 @@ class CreateSandboxRequest(Model):
     )
 
 
+@dataclass(slots=True)
+class CreateSnapshotRequest(Model):
+    type: SnapshotType = SnapshotType.DISK
+    name: str = _json("name", omit_default=True, default="")
+
+
 @dataclass(slots=True)
 class ForkSandboxRequest(Model):
     start_paused: bool = _json("start_paused", omit_default=True, default=False)
@@ -562,6 +581,7 @@ class CreateSandboxResponse(Model):
     egress_rules: list[str] = _json("egress", default_factory=list)
     bandwidth_quota_bytes: int = 0
     ingress_url_template: str = ""
+    snapshot_id: str | None = None
 
 
 @dataclass(slots=True)
@@ -592,6 +612,29 @@ class Sandbox(Model):
     last_resumed_at: datetime | None = None
     forked_from: str | None = None
     auto_pause_after_seconds: int | None = None
+    snapshot_id: str | None = None
+
+
+@dataclass(slots=True)
+class Snapshot(Model):
+    id: str
+    sandbox_id: str
+    type: SnapshotType
+    status: SnapshotStatus
+    name: str | None = None
+    created_at: datetime | None = None
+    ready_at: datetime | None = None
+    shape: str = ""
+    vcpu: int = 0
+    memory_mib: int = _json("mem_mib", default=0)
+    disk_mib: int = _json("disk_mib", default=0)
+    size_bytes: int = 0
+
+
+@dataclass(slots=True)
+class SnapshotDeletedResponse(Model):
+    id: str
+    status: SnapshotStatus
 
 
 @dataclass(slots=True)

src/createos/services.py

@@ -45,6 +45,8 @@
     PaginationOptions,
     PTYSize,
     RequestOptions,
+    Snapshot,
+    SnapshotDeletedResponse,
     Template,
     TemplateCreateRequest,
     TemplateLogsOptions,
@@ -288,6 +290,41 @@ def rotate_credentials(
         return _one(Disk, data)
 
 
+class SnapshotsService:
+    """Manage account-level reusable sandbox snapshots."""
+
+    def __init__(self, transport: Transport) -> None:
+        self._transport = transport
+
+    def list(self, options: PaginationOptions | None = None) -> list[Snapshot]:
+        """Return reusable snapshots owned by the caller."""
+        return _fetch_all(
+            self._transport,
+            "/v1/snapshots",
+            Snapshot,
+            options or PaginationOptions(),
+            legacy_key="snapshots",
+        )
+
+    def get(self, snapshot_id: str) -> Snapshot:
+        """Return an owned reusable snapshot by ID."""
+        return _one(
+            Snapshot,
+            self._transport.request(
+                "GET", f"/v1/snapshots/{_segment(snapshot_id)}"
+            ),
+        )
+
+    def delete(self, snapshot_id: str) -> SnapshotDeletedResponse:
+        """Remove an owned ready or failed snapshot from API listings."""
+        return _one(
+            SnapshotDeletedResponse,
+            self._transport.request(
+                "DELETE", f"/v1/snapshots/{_segment(snapshot_id)}"
+            ),
+        )
+
+
 class FilesService:
     """Transfer files to and from one sandbox."""