diff --git a/.changeset/world-snapshots-interface.md b/.changeset/world-snapshots-interface.md
new file mode 100644
index 0000000000..a4e8876784
--- /dev/null
+++ b/.changeset/world-snapshots-interface.md
@@ -0,0 +1,8 @@
+---
+'@workflow/world': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+'@workflow/world-vercel': minor
+---
+
+Add an OPTIONAL `snapshots` storage interface (`save`/`load`/`delete` + `SnapshotMetadata`, plus `encodeSnapshotEnvelope`/`decodeSnapshotEnvelope` helpers that pack metadata and bytes into one atomically-storable blob) for VM-memory snapshotting, with implementations in world-local (single envelope file per run), world-postgres (`workflow_snapshots` table storing the envelope), and world-vercel (workflow-server `/v2/runs/:runId/snapshot` endpoints with W3C trace-context injection; delete is 404-idempotent). Inert until the runtime opts in; community World implementations that don't provide `snapshots` are unaffected (the runtime feature-detects and falls back to full replay).
diff --git a/packages/world-local/src/storage/index.ts b/packages/world-local/src/storage/index.ts
index 48b492d694..c98e1df13f 100644
--- a/packages/world-local/src/storage/index.ts
+++ b/packages/world-local/src/storage/index.ts
@@ -3,6 +3,7 @@ import { instrumentObject } from '../instrumentObject.js';
 import { createEventsStorage } from './events-storage.js';
 import { createHooksStorage } from './hooks-storage.js';
 import { createRunsStorage, type LocalRunsStorage } from './runs-storage.js';
+import { createSnapshotsStorage } from './snapshots-storage.js';
 import { createStepsStorage } from './steps-storage.js';
 
 /**
@@ -30,6 +31,7 @@ export function createStorage(basedir: string, tag?: string): LocalStorage {
   const steps = createStepsStorage(basedir, tag);
   const events = createEventsStorage(basedir, tag);
   const hooks = createHooksStorage(basedir, tag);
+  const snapshots = createSnapshotsStorage(basedir);
 
   // Instrument all storage methods with tracing
   // NOTE: Span names are lowercase per OTEL semantic conventions
@@ -38,6 +40,7 @@ export function createStorage(basedir: string, tag?: string): LocalStorage {
     steps: instrumentObject('world.steps', steps),
     events: instrumentObject('world.events', events),
     hooks: instrumentObject('world.hooks', hooks),
+    snapshots: instrumentObject('world.snapshots', snapshots),
     clearCache: () => events.clearCache(),
   };
 }
diff --git a/packages/world-local/src/storage/snapshots-storage.test.ts b/packages/world-local/src/storage/snapshots-storage.test.ts
new file mode 100644
index 0000000000..4b76af232d
--- /dev/null
+++ b/packages/world-local/src/storage/snapshots-storage.test.ts
@@ -0,0 +1,115 @@
+import { promises as fs } from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { createSnapshotsStorage } from './snapshots-storage.js';
+
+describe('snapshots storage (world-local)', () => {
+  let testDir: string;
+  let snapshots: ReturnType<typeof createSnapshotsStorage>;
+
+  beforeEach(async () => {
+    testDir = await fs.mkdtemp(path.join(os.tmpdir(), 'snapshots-test-'));
+    snapshots = createSnapshotsStorage(testDir);
+  });
+
+  afterEach(async () => {
+    await fs.rm(testDir, { recursive: true, force: true });
+  });
+
+  it('returns null when no snapshot exists', async () => {
+    expect(await snapshots.load('wrun_missing')).toBeNull();
+  });
+
+  it('round-trips snapshot bytes and metadata', async () => {
+    const data = new Uint8Array([1, 2, 3, 250, 251, 252]);
+    const createdAt = new Date('2025-06-01T12:00:00Z');
+    await snapshots.save('wrun_a', data, {
+      eventsCursor: 'evnt_123',
+      createdAt,
+    });
+
+    const loaded = await snapshots.load('wrun_a');
+    expect(loaded).not.toBeNull();
+    expect(new Uint8Array(loaded!.data)).toEqual(data);
+    expect(loaded!.metadata.eventsCursor).toBe('evnt_123');
+    expect(+loaded!.metadata.createdAt).toBe(+createdAt);
+  });
+
+  it('overwrites the previous snapshot on save', async () => {
+    await snapshots.save('wrun_a', new Uint8Array([1]), {
+      eventsCursor: null,
+      createdAt: new Date(),
+    });
+    await snapshots.save('wrun_a', new Uint8Array([9, 9]), {
+      eventsCursor: 'evnt_9',
+      createdAt: new Date(),
+    });
+
+    const loaded = await snapshots.load('wrun_a');
+    expect(new Uint8Array(loaded!.data)).toEqual(new Uint8Array([9, 9]));
+    expect(loaded!.metadata.eventsCursor).toBe('evnt_9');
+  });
+
+  it('supports a null events cursor', async () => {
+    await snapshots.save('wrun_b', new Uint8Array([7]), {
+      eventsCursor: null,
+      createdAt: new Date(),
+    });
+    const loaded = await snapshots.load('wrun_b');
+    expect(loaded!.metadata.eventsCursor).toBeNull();
+  });
+
+  it('delete removes the snapshot (idempotent)', async () => {
+    await snapshots.save('wrun_c', new Uint8Array([1]), {
+      eventsCursor: null,
+      createdAt: new Date(),
+    });
+    await snapshots.delete('wrun_c');
+    expect(await snapshots.load('wrun_c')).toBeNull();
+    // Deleting again is a no-op.
+    await snapshots.delete('wrun_c');
+  });
+
+  it('stores metadata and bytes in ONE file (atomic pairing — no torn save)', async () => {
+    await snapshots.save('wrun_atomic', new Uint8Array([1, 2, 3]), {
+      eventsCursor: 'evnt_x',
+      createdAt: new Date(),
+    });
+    const files = await fs.readdir(path.join(testDir, 'snapshots'));
+    expect(files).toEqual(['wrun_atomic.snapshot']);
+  });
+
+  it('treats a corrupt envelope file as a miss instead of returning torn state', async () => {
+    await snapshots.save('wrun_corrupt', new Uint8Array([1, 2, 3]), {
+      eventsCursor: 'evnt_x',
+      createdAt: new Date(),
+    });
+    await fs.writeFile(
+      path.join(testDir, 'snapshots', 'wrun_corrupt.snapshot'),
+      Buffer.from([0xde, 0xad])
+    );
+    expect(await snapshots.load('wrun_corrupt')).toBeNull();
+  });
+
+  it('rejects path-traversal runIds on every operation', async () => {
+    const hostile = [
+      '../escape',
+      '..',
+      'a/b',
+      'a\\b',
+      'nul\0byte',
+      '.hidden',
+      'dotted.name',
+      '',
+    ];
+    const metadata = { eventsCursor: null, createdAt: new Date() };
+    for (const runId of hostile) {
+      await expect(
+        snapshots.save(runId, new Uint8Array([1]), metadata)
+      ).rejects.toThrow(/unsafe|invalid/i);
+      await expect(snapshots.load(runId)).rejects.toThrow(/unsafe|invalid/i);
+      await expect(snapshots.delete(runId)).rejects.toThrow(/unsafe|invalid/i);
+    }
+  });
+});
diff --git a/packages/world-local/src/storage/snapshots-storage.ts b/packages/world-local/src/storage/snapshots-storage.ts
new file mode 100644
index 0000000000..cd4dd13def
--- /dev/null
+++ b/packages/world-local/src/storage/snapshots-storage.ts
@@ -0,0 +1,90 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import type { SnapshotMetadata } from '@workflow/world';
+import {
+  decodeSnapshotEnvelope,
+  encodeSnapshotEnvelope,
+} from '@workflow/world';
+import {
+  assertSafeEntityId,
+  ensureDir,
+  readBuffer,
+  resolveWithinBase,
+  write,
+} from '../fs.js';
+
+/**
+ * Create the snapshots sub-storage for a local World implementation.
+ *
+ * Snapshots are stored as ONE file per run:
+ *   {basedir}/snapshots/{runId}.snapshot — snapshot envelope (metadata +
+ *   opaque VM snapshot bytes; see `encodeSnapshotEnvelope`)
+ *
+ * A single file matters: `write()` is atomic per file (temp + rename),
+ * but two files (bytes + metadata) written separately can tear — a crash
+ * between the renames, or a concurrent `load` interleaving them, pairs a
+ * heap image from one suspension with an `eventsCursor` from another,
+ * and the restore silently replays from the wrong log position. The
+ * envelope makes the pairing structurally atomic.
+ *
+ * Compression and encryption are handled by `@workflow/core`'s snapshot
+ * entrypoint (`compress → encrypt → save`); this world layer stores the
+ * resulting bytes verbatim.
+ */
+export function createSnapshotsStorage(basedir: string) {
+  const snapshotsDir = path.join(basedir, 'snapshots');
+
+  // `runId` arrives from the request body: validate it before it touches a
+  // filesystem path (primary defense — rejects `../`, `/`, `\`, NUL, `.`),
+  // and contain the join under the snapshots dir (defense in depth), the
+  // same two-layer scheme the other world-local storages use.
+  function envelopePath(runId: string): string {
+    assertSafeEntityId('runId', runId);
+    return resolveWithinBase(snapshotsDir, `${runId}.snapshot`);
+  }
+
+  return {
+    async save(
+      runId: string,
+      data: Uint8Array,
+      metadata: SnapshotMetadata
+    ): Promise<void> {
+      await ensureDir(snapshotsDir);
+      await write(
+        envelopePath(runId),
+        Buffer.from(encodeSnapshotEnvelope(metadata, data)),
+        { overwrite: true }
+      );
+    },
+
+    async load(
+      runId: string
+    ): Promise<{ data: Uint8Array; metadata: SnapshotMetadata } | null> {
+      let envelope: Buffer;
+      try {
+        envelope = await readBuffer(envelopePath(runId));
+      } catch (error: any) {
+        if (error.code === 'ENOENT') {
+          return null;
+        }
+        throw error;
+      }
+      // Anything that fails to decode (truncated, corrupt, unknown
+      // version, schema-invalid metadata) is a clean miss — the caller
+      // falls back to full replay, never to fabricated metadata.
+      return decodeSnapshotEnvelope(
+        new Uint8Array(
+          envelope.buffer,
+          envelope.byteOffset,
+          envelope.byteLength
+        )
+      );
+    },
+
+    async delete(runId: string): Promise<void> {
+      // `force: true` — idempotent by contract (terminal-state cleanup
+      // retries, and runs that never snapshotted delete too).
+      await fs.rm(envelopePath(runId), { force: true });
+    },
+  };
+}
diff --git a/packages/world-postgres/src/drizzle/migrations/0019_add_snapshots_table.sql b/packages/world-postgres/src/drizzle/migrations/0019_add_snapshots_table.sql
new file mode 100644
index 0000000000..83c8b55257
--- /dev/null
+++ b/packages/world-postgres/src/drizzle/migrations/0019_add_snapshots_table.sql
@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_snapshots" (
+	"run_id" varchar PRIMARY KEY NOT NULL,
+	"data" "bytea" NOT NULL,
+	"events_cursor" varchar,
+	"created_at" timestamp DEFAULT now() NOT NULL
+);
diff --git a/packages/world-postgres/src/drizzle/migrations/meta/_journal.json b/packages/world-postgres/src/drizzle/migrations/meta/_journal.json
index 4ce969faa2..a1692d0d1c 100644
--- a/packages/world-postgres/src/drizzle/migrations/meta/_journal.json
+++ b/packages/world-postgres/src/drizzle/migrations/meta/_journal.json
@@ -134,6 +134,13 @@
       "when": 1785619990000,
       "tag": "0018_add_hook_token_retention",
       "breakpoints": true
+    },
+    {
+      "idx": 19,
+      "version": "7",
+      "when": 1785700000000,
+      "tag": "0019_add_snapshots_table",
+      "breakpoints": true
     }
   ]
 }
diff --git a/packages/world-postgres/src/drizzle/schema.ts b/packages/world-postgres/src/drizzle/schema.ts
index 578c78cf2d..89a4bce0a4 100644
--- a/packages/world-postgres/src/drizzle/schema.ts
+++ b/packages/world-postgres/src/drizzle/schema.ts
@@ -267,6 +267,24 @@ const bytea = customType<{ data: Buffer; notNull: false; default: false }>({
   },
 });
 
+/**
+ * VM snapshots for VM-memory snapshotting.
+ *
+ * Each row is a 1-to-1 mapping with a workflow run — a snapshot captures
+ * the QuickJS VM state at a suspension point so execution can resume from
+ * there without replaying the full event log.
+ *
+ * The binary data is stored as opaque bytes in the `data` column (the SDK
+ * applies compression/encryption before handing bytes to the world).
+ * Metadata (`eventsCursor`, `createdAt`) lives alongside for cheap loads.
+ */
+export const snapshots = schema.table('workflow_snapshots', {
+  runId: varchar('run_id').primaryKey(),
+  data: bytea('data').notNull(),
+  eventsCursor: varchar('events_cursor'),
+  createdAt: timestamp('created_at').defaultNow().notNull(),
+});
+
 export const streams = schema.table(
   'workflow_stream_chunks',
   {
diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts
index 463ba42fba..cbb6f8e3d9 100644
--- a/packages/world-postgres/src/index.ts
+++ b/packages/world-postgres/src/index.ts
@@ -4,6 +4,7 @@ import { Pool } from 'pg';
 import type { PostgresWorldConfig } from './config.js';
 import { createClient, type Drizzle } from './drizzle/index.js';
 import { createQueue } from './queue.js';
+import { createSnapshotsStorage } from './snapshots.js';
 import {
   createEventsStorage,
   createHooksStorage,
@@ -18,6 +19,7 @@ function createStorage(drizzle: Drizzle): Storage {
     events: createEventsStorage(drizzle),
     hooks: createHooksStorage(drizzle),
     steps: createStepsStorage(drizzle),
+    snapshots: createSnapshotsStorage(drizzle),
   };
 }
 
diff --git a/packages/world-postgres/src/snapshots.ts b/packages/world-postgres/src/snapshots.ts
new file mode 100644
index 0000000000..0b424863f1
--- /dev/null
+++ b/packages/world-postgres/src/snapshots.ts
@@ -0,0 +1,86 @@
+import type { SnapshotMetadata, Storage } from '@workflow/world';
+import {
+  decodeSnapshotEnvelope,
+  encodeSnapshotEnvelope,
+} from '@workflow/world';
+import { eq } from 'drizzle-orm';
+import { type Drizzle, Schema } from './drizzle/index.js';
+
+/**
+ * Snapshot storage for world-postgres.
+ *
+ * Compression and encryption are handled by `@workflow/core`'s
+ * snapshot entrypoint (`compress(snapshot) → encrypt → save`). This
+ * world layer treats the bytes as opaque — it does NOT add its own
+ * compression.
+ *
+ * The `data` column stores a snapshot ENVELOPE (metadata + bytes in one
+ * blob; see `encodeSnapshotEnvelope`): the full metadata object
+ * round-trips losslessly through the envelope, so new metadata fields
+ * never require a schema migration, and the metadata/bytes pairing is
+ * atomic by construction. The `events_cursor` / `created_at` columns
+ * are denormalized copies kept for observability (SQL inspection) —
+ * loads read the envelope, never the columns.
+ *
+ * Each run has at most one row; `save()` upserts the latest
+ * suspension's bytes.
+ */
+export function createSnapshotsStorage(
+  drizzle: Drizzle
+): NonNullable<Storage['snapshots']> {
+  const { snapshots } = Schema;
+
+  return {
+    async save(
+      runId: string,
+      data: Uint8Array,
+      metadata: SnapshotMetadata
+    ): Promise<void> {
+      const blob = Buffer.from(encodeSnapshotEnvelope(metadata, data));
+      await drizzle
+        .insert(snapshots)
+        .values({
+          runId,
+          data: blob,
+          eventsCursor: metadata.eventsCursor,
+          createdAt: metadata.createdAt,
+        })
+        .onConflictDoUpdate({
+          target: snapshots.runId,
+          set: {
+            data: blob,
+            eventsCursor: metadata.eventsCursor,
+            createdAt: metadata.createdAt,
+          },
+        });
+    },
+
+    async load(
+      runId: string
+    ): Promise<{ data: Uint8Array; metadata: SnapshotMetadata } | null> {
+      const [row] = await drizzle
+        .select()
+        .from(snapshots)
+        .where(eq(snapshots.runId, runId))
+        .limit(1);
+
+      if (!row) return null;
+
+      // Anything that fails to decode (corrupt, unknown version,
+      // schema-invalid metadata) is a clean miss — the caller falls back
+      // to full replay, never to fabricated metadata.
+      return decodeSnapshotEnvelope(
+        new Uint8Array(
+          row.data.buffer,
+          row.data.byteOffset,
+          row.data.byteLength
+        )
+      );
+    },
+
+    async delete(runId: string): Promise<void> {
+      // Plain DELETE — naturally idempotent (0 rows affected is success).
+      await drizzle.delete(snapshots).where(eq(snapshots.runId, runId));
+    },
+  };
+}
diff --git a/packages/world-vercel/src/snapshots.test.ts b/packages/world-vercel/src/snapshots.test.ts
new file mode 100644
index 0000000000..c4be67539a
--- /dev/null
+++ b/packages/world-vercel/src/snapshots.test.ts
@@ -0,0 +1,247 @@
+import { createServer, type Server } from 'node:http';
+import type { AddressInfo } from 'node:net';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+// Mock getHttpConfig to return a localhost URL pointing at the test server.
+// Set per-test via setBaseUrl().
+let baseUrl = 'http://127.0.0.1:0';
+vi.mock('./utils.js', () => ({
+  getHttpConfig: vi.fn(() =>
+    Promise.resolve({
+      baseUrl,
+      headers: new Headers(),
+      usingProxy: false,
+    })
+  ),
+}));
+
+// Bypass the OIDC token fetch in getHttpConfig — handled by the mock above.
+
+import { createSnapshotsStorage } from './snapshots.js';
+
+interface RequestRecord {
+  method: string;
+  path: string;
+  contentLength?: string;
+  bodyBytes: number;
+  bodyError?: string;
+}
+
+/**
+ * HTTP test server with programmable response handlers.
+ *
+ * Each test installs a handler via `server.handle = (req, res, attempt) => …`.
+ * The server tracks per-request body sizes and content-length so tests can
+ * assert that the FULL body was received on every attempt (not 0 bytes,
+ * which is the symptom of the undici fetch+RetryAgent+Buffer-body bug).
+ */
+class TestServer {
+  server!: Server;
+  url = '';
+  records: RequestRecord[] = [];
+  handle:
+    | ((
+        req: import('node:http').IncomingMessage,
+        res: import('node:http').ServerResponse,
+        attempt: number
+      ) => void)
+    | undefined;
+
+  async start(): Promise<void> {
+    this.records = [];
+    this.server = createServer((req, res) => {
+      const cl = req.headers['content-length'] as string | undefined;
+      let bodyBytes = 0;
+      const record: RequestRecord = {
+        method: req.method ?? '?',
+        path: req.url ?? '?',
+        contentLength: cl,
+        bodyBytes: 0,
+      };
+      req.on('data', (chunk) => {
+        bodyBytes += chunk.length;
+      });
+      req.on('end', () => {
+        record.bodyBytes = bodyBytes;
+        this.records.push(record);
+        const attempt = this.records.length;
+        if (this.handle) {
+          this.handle(req, res, attempt);
+        } else {
+          res.writeHead(200);
+          res.end('ok');
+        }
+      });
+      req.on('error', (err) => {
+        record.bodyError = err.message;
+        record.bodyBytes = bodyBytes;
+        this.records.push(record);
+      });
+    });
+    await new Promise<void>((resolve) => this.server.listen(0, resolve));
+    const { port } = this.server.address() as AddressInfo;
+    this.url = `http://127.0.0.1:${port}`;
+  }
+
+  async stop(): Promise<void> {
+    if (this.server) {
+      await new Promise<void>((resolve) => {
+        this.server.close(() => resolve());
+      });
+    }
+  }
+}
+
+describe('snapshots storage', () => {
+  let server: TestServer;
+
+  beforeEach(async () => {
+    server = new TestServer();
+    await server.start();
+    baseUrl = server.url;
+  });
+
+  afterEach(async () => {
+    await server.stop();
+  });
+
+  describe('save', () => {
+    it('sends a single PUT with the full body when the server responds 200', async () => {
+      server.handle = (_req, res) => {
+        res.writeHead(200);
+        res.end('ok');
+      };
+      const storage = createSnapshotsStorage();
+      const data = new Uint8Array(1024).fill(7);
+      await storage.save('wrun_test', data, {
+        eventsCursor: 'eid:test',
+        createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      });
+
+      expect(server.records).toHaveLength(1);
+      const r = server.records[0]!;
+      expect(r.method).toBe('PUT');
+      expect(r.path).toBe('/v2/runs/wrun_test/snapshot');
+      expect(r.bodyBytes).toBeGreaterThan(0);
+      expect(r.bodyBytes).toBe(Number(r.contentLength));
+    });
+
+    it('retries on transient 503 and sends the full body on every attempt (regression: undici fetch+RetryAgent loses Buffer body on retry)', async () => {
+      // First attempt: 503 (transient). Second: 200.
+      // The undici fetch() + RetryAgent combo wraps Buffer bodies in a
+      // one-shot ReadableStream, so the second attempt sends 0 bytes
+      // and triggers UND_ERR_REQ_CONTENT_LENGTH_MISMATCH. Switching
+      // the snapshot save path to undici.request() preserves the body
+      // across retries.
+      server.handle = (_req, res, attempt) => {
+        if (attempt === 1) {
+          res.writeHead(503);
+          res.end('try again');
+        } else {
+          res.writeHead(200);
+          res.end('ok');
+        }
+      };
+
+      const storage = createSnapshotsStorage();
+      const data = new Uint8Array(64 * 1024).fill(42);
+      await storage.save('wrun_retry', data, {
+        eventsCursor: null,
+        createdAt: new Date('2024-01-02T00:00:00.000Z'),
+      });
+
+      expect(server.records).toHaveLength(2);
+      // BOTH attempts must include the full body. If the body were lost
+      // on retry, attempt 2 would have bodyBytes === 0 and the request
+      // would fail with UND_ERR_REQ_CONTENT_LENGTH_MISMATCH before
+      // reaching the server at all.
+      for (const r of server.records) {
+        expect(r.method).toBe('PUT');
+        expect(r.bodyBytes).toBeGreaterThan(0);
+        expect(r.bodyBytes).toBe(Number(r.contentLength));
+      }
+    });
+
+    it('throws WorkflowWorldError when the server returns 4xx', async () => {
+      server.handle = (_req, res) => {
+        res.writeHead(400);
+        res.end('bad request');
+      };
+      const storage = createSnapshotsStorage();
+      const data = new Uint8Array(16);
+      await expect(
+        storage.save('wrun_bad', data, {
+          eventsCursor: null,
+          createdAt: new Date(),
+        })
+      ).rejects.toThrow(/HTTP 400/);
+    });
+  });
+
+  describe('load', () => {
+    it('round-trips the FULL metadata through the envelope body (no header fabrication)', async () => {
+      const { encodeSnapshotEnvelope } = await import('@workflow/world');
+      const stored = encodeSnapshotEnvelope(
+        {
+          eventsCursor: 'evnt_cursor_1',
+          createdAt: new Date('2025-06-01T12:00:00.000Z'),
+        },
+        new Uint8Array([1, 2, 3, 4])
+      );
+      server.handle = (_req, res) => {
+        // Deliberately NO X-Snapshot-* headers: metadata must come from
+        // the envelope, never be invented from headers/wall time.
+        res.writeHead(200, { 'Content-Type': 'application/octet-stream' });
+        res.end(Buffer.from(stored));
+      };
+      const storage = createSnapshotsStorage();
+      const loaded = await storage.load('wrun_env');
+      expect(loaded).not.toBeNull();
+      expect(loaded!.metadata.eventsCursor).toBe('evnt_cursor_1');
+      expect(loaded!.metadata.createdAt).toEqual(
+        new Date('2025-06-01T12:00:00.000Z')
+      );
+      expect(Array.from(loaded!.data)).toEqual([1, 2, 3, 4]);
+    });
+
+    it('treats an undecodable body as a miss instead of fabricating metadata', async () => {
+      server.handle = (_req, res) => {
+        // A body that predates the envelope format (or got truncated).
+        res.writeHead(200, {
+          'Content-Type': 'application/octet-stream',
+          // Even with plausible-looking headers present, the load must
+          // NOT synthesize metadata from them.
+          'X-Snapshot-Events-Cursor': 'evnt_header_only',
+          'X-Snapshot-Created-At': new Date().toISOString(),
+        });
+        res.end(Buffer.from([0xde, 0xad, 0xbe, 0xef]));
+      };
+      const storage = createSnapshotsStorage();
+      await expect(storage.load('wrun_legacy')).resolves.toBeNull();
+    });
+  });
+
+  describe('delete', () => {
+    it('treats 404 as success (idempotent terminal-state cleanup)', async () => {
+      server.handle = (_req, res) => {
+        res.writeHead(404);
+        res.end('not found');
+      };
+      const storage = createSnapshotsStorage();
+      await expect(storage.delete('wrun_never_snapshotted')).resolves.toBe(
+        undefined
+      );
+    });
+
+    it('still throws on non-retryable server errors', async () => {
+      // 403 rather than 500: the shared dispatcher's RetryAgent retries
+      // 5xx (with backoff), which is orthogonal to what this asserts.
+      server.handle = (_req, res) => {
+        res.writeHead(403);
+        res.end('forbidden');
+      };
+      const storage = createSnapshotsStorage();
+      await expect(storage.delete('wrun_err')).rejects.toThrow(/HTTP 403/);
+    });
+  });
+});
diff --git a/packages/world-vercel/src/snapshots.ts b/packages/world-vercel/src/snapshots.ts
new file mode 100644
index 0000000000..b244000b30
--- /dev/null
+++ b/packages/world-vercel/src/snapshots.ts
@@ -0,0 +1,229 @@
+import { WorkflowWorldError } from '@workflow/errors';
+import type { SnapshotMetadata, Storage } from '@workflow/world';
+import {
+  decodeSnapshotEnvelope,
+  encodeSnapshotEnvelope,
+} from '@workflow/world';
+import { request as undiciRequest } from 'undici';
+import { HTTP_DEBUG_ENABLED } from './http-core.js';
+import { getDispatcher } from './http-client.js';
+import { injectTraceContextIntoHeaders } from './telemetry.js';
+import { type APIConfig, getHttpConfig } from './utils.js';
+
+/**
+ * Convert a Web `Headers` object into a plain record for undici's
+ * lower-level `request()` API. Headers in undici-request take
+ * `Record<string, string | string[]>`, not the Headers object.
+ */
+function headersToRecord(headers: Headers): Record<string, string> {
+  const record: Record<string, string> = {};
+  for (const [key, value] of headers) {
+    record[key] = value;
+  }
+  return record;
+}
+
+/**
+ * Per-operation diagnostic (wire bytes + HTTP cost, grep-able by runId
+ * alongside @workflow/core's QUICKJS_VM diagnostics). Runs on every
+ * suspension/resume, so it is gated behind the package's HTTP debug
+ * flag like every other request log in this package.
+ */
+function snapshotDiag(fields: Record<string, unknown>): void {
+  if (!HTTP_DEBUG_ENABLED) return;
+  console.debug('[workflow:world-vercel:http] WORLD_SNAPSHOT_DIAG', fields);
+}
+
+/**
+ * Create snapshot storage backed by the workflow-server API.
+ *
+ * Compression and encryption are handled by `@workflow/core`'s
+ * snapshot entrypoint (`compress(snapshot) → encrypt → save`). This
+ * world layer transports the bytes opaquely — it does not compress
+ * (encryption produces ciphertext that doesn't compress) and it does
+ * not encrypt.
+ *
+ * The request/response body is a snapshot ENVELOPE (metadata + bytes in
+ * one blob; see `encodeSnapshotEnvelope`): the workflow-server stores
+ * it opaquely, so the FULL metadata object round-trips losslessly
+ * without any server-side schema involvement, and the metadata/bytes
+ * pairing is atomic by construction. The `X-Snapshot-*` headers on save
+ * are denormalized copies for server-side observability only — loads
+ * decode the envelope and never trust headers (fabricating metadata
+ * from missing headers is exactly the silent-wrong-answer direction:
+ * an invented null cursor means "replay from the beginning").
+ *
+ * Snapshot endpoints use raw binary transfer:
+ *   - PUT  /v2/runs/:runId/snapshot — envelope body
+ *   - GET  /v2/runs/:runId/snapshot — envelope response
+ *   - DELETE /v2/runs/:runId/snapshot — no body; 404 is success
+ */
+export function createSnapshotsStorage(
+  config?: APIConfig
+): NonNullable<Storage['snapshots']> {
+  return {
+    async save(
+      runId: string,
+      data: Uint8Array,
+      metadata: SnapshotMetadata
+    ): Promise<void> {
+      const t0 = performance.now();
+      const { baseUrl, headers } = await getHttpConfig(config);
+      const url = `${baseUrl}/v2/runs/${encodeURIComponent(runId)}/snapshot`;
+
+      const envelope = encodeSnapshotEnvelope(metadata, data);
+
+      headers.set('Content-Type', 'application/octet-stream');
+      // Observability-only denormalized copies (see module docstring).
+      headers.set('X-Snapshot-Events-Cursor', metadata.eventsCursor ?? '');
+      headers.set('X-Snapshot-Created-At', metadata.createdAt.toISOString());
+      // Explicit W3C trace-context injection: this path routes around
+      // `makeRequest` (raw undici for Buffer-body retry correctness), so
+      // ambient auto-instrumentation does not reliably hook it. Required
+      // for workflow-server spans to join the caller's trace — see
+      // telemetry.ts and trace-propagation.test.ts.
+      await injectTraceContextIntoHeaders(headers);
+
+      // Use undici.request() rather than the global fetch() because
+      // fetch() + RetryAgent is broken for Buffer/Uint8Array bodies:
+      // fetch wraps the body in a one-shot ReadableStream (per the
+      // WHATWG fetch spec), so when the RetryAgent retries (on 5xx or
+      // network errors), the second attempt sends 0 bytes and undici
+      // throws `UND_ERR_REQ_CONTENT_LENGTH_MISMATCH`. The lower-level
+      // `request()` API hands the Buffer to the connection layer
+      // directly, which can be replayed on retry.
+      //
+      // Upstream context: nodejs/undici#3288 (filed May 2024) reported
+      // this exact failure. The "fix" in nodejs/undici#3294 made
+      // RetryAgent skip stateful bodies rather than rewind them, and
+      // the maintainers explicitly recommended switching to
+      // `undici.request()` for any retried request with a body. Don't
+      // simplify this back to `fetch()` without first verifying that
+      // upstream now copies Buffers across retries.
+      //
+      // Snapshot bodies are 5-15 MB so the bug fires constantly under
+      // network turbulence; a single failed save poisons the run
+      // (handler returns 500 -> queue retries handler -> save fails
+      // again -> 5xx loop until the run TTL).
+      const putStart = performance.now();
+      const response = await undiciRequest(url, {
+        method: 'PUT',
+        body: envelope,
+        headers: headersToRecord(headers),
+        dispatcher: getDispatcher(config) as never,
+      });
+      const putDurationMs = Math.round(performance.now() - putStart);
+
+      if (response.statusCode < 200 || response.statusCode >= 300) {
+        const text = await response.body.text().catch(() => '');
+        throw new WorkflowWorldError(
+          `PUT /v2/runs/${runId}/snapshot -> HTTP ${response.statusCode}: ${text}`,
+          { url, status: response.statusCode }
+        );
+      }
+
+      // Consume the response body to release the connection
+      await response.body.text();
+
+      snapshotDiag({
+        op: 'save',
+        runId,
+        // Bytes received from the core — already compressed and
+        // encrypted upstream. The world transports them opaquely
+        // (envelope framing adds the metadata header).
+        wireBytes: envelope.byteLength,
+        putDurationMs,
+        totalDurationMs: Math.round(performance.now() - t0),
+      });
+    },
+
+    async load(
+      runId: string
+    ): Promise<{ data: Uint8Array; metadata: SnapshotMetadata } | null> {
+      const t0 = performance.now();
+      const { baseUrl, headers } = await getHttpConfig(config);
+      const url = `${baseUrl}/v2/runs/${encodeURIComponent(runId)}/snapshot`;
+
+      headers.set('Accept', 'application/octet-stream');
+      await injectTraceContextIntoHeaders(headers);
+
+      const getStart = performance.now();
+      const response = await fetch(url, {
+        method: 'GET',
+        headers,
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici dispatcher
+        dispatcher: getDispatcher(config),
+      } as any);
+      const getDurationMs = Math.round(performance.now() - getStart);
+
+      if (response.status === 404) {
+        // Consume the response body to release the connection
+        await response.text().catch(() => {});
+        snapshotDiag({
+          op: 'load',
+          runId,
+          outcome: 'not_found',
+          getDurationMs,
+          totalDurationMs: Math.round(performance.now() - t0),
+        });
+        return null;
+      }
+
+      if (!response.ok) {
+        const text = await response.text().catch(() => '');
+        throw new WorkflowWorldError(
+          `GET /v2/runs/${runId}/snapshot -> HTTP ${response.status}: ${text}`,
+          { url, status: response.status }
+        );
+      }
+
+      const buffer = await response.arrayBuffer();
+      const envelope = new Uint8Array(buffer);
+
+      // Decode the envelope — the ONLY source of metadata. A body that
+      // does not decode (pre-envelope write, truncation, schema-invalid
+      // metadata) is a clean miss: the caller falls back to full
+      // replay. Never fabricate metadata from headers or wall time.
+      const decoded = decodeSnapshotEnvelope(envelope);
+
+      snapshotDiag({
+        op: 'load',
+        runId,
+        outcome: decoded ? 'ok' : 'undecodable_envelope',
+        wireBytes: envelope.byteLength,
+        getDurationMs,
+        totalDurationMs: Math.round(performance.now() - t0),
+      });
+
+      return decoded;
+    },
+
+    async delete(runId: string): Promise<void> {
+      const { baseUrl, headers } = await getHttpConfig(config);
+      const url = `${baseUrl}/v2/runs/${encodeURIComponent(runId)}/snapshot`;
+      await injectTraceContextIntoHeaders(headers);
+
+      const response = await fetch(url, {
+        method: 'DELETE',
+        headers,
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any -- undici dispatcher
+        dispatcher: getDispatcher(config),
+      } as any);
+
+      // 404 is success: delete is idempotent by interface contract —
+      // terminal-state cleanup retries, runs twice, and runs for runs
+      // that never snapshotted (matching local's `force: true` and
+      // postgres's plain DELETE).
+      if (!response.ok && response.status !== 404) {
+        const text = await response.text().catch(() => '');
+        throw new WorkflowWorldError(
+          `DELETE /v2/runs/${runId}/snapshot -> HTTP ${response.status}: ${text}`,
+          { url, status: response.status }
+        );
+      }
+
+      // Consume the response body to release the connection
+      await response.text();
+    },
+  };
+}
diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts
index 46fced038c..7ef4667fe7 100644
--- a/packages/world-vercel/src/storage.ts
+++ b/packages/world-vercel/src/storage.ts
@@ -17,10 +17,12 @@ import {
   getWorkflowRuns,
   listWorkflowRuns,
 } from './runs.js';
+import { createSnapshotsStorage } from './snapshots.js';
 import { getStep, listWorkflowRunSteps } from './steps.js';
 import type { APIConfig } from './utils.js';
 
 export function createStorage(config?: APIConfig): Storage {
+  const snapshots = createSnapshotsStorage(config);
   const storage: Storage = {
     // Storage interface with namespaced methods
     runs: {
@@ -57,6 +59,7 @@ export function createStorage(config?: APIConfig): Storage {
       getByToken: (token) => getHookByToken(token, config),
       list: (params) => listHooks(params, config),
     },
+    snapshots,
   };
 
   // Instrument all storage methods with tracing
@@ -66,5 +69,6 @@ export function createStorage(config?: APIConfig): Storage {
     steps: instrumentObject('world.steps', storage.steps),
     events: instrumentObject('world.events', storage.events),
     hooks: instrumentObject('world.hooks', storage.hooks),
+    snapshots: instrumentObject('world.snapshots', snapshots),
   };
 }
diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts
index c071d59960..b430d44863 100644
--- a/packages/world-vercel/src/trace-propagation.test.ts
+++ b/packages/world-vercel/src/trace-propagation.test.ts
@@ -314,6 +314,87 @@ describe('streamer write trace propagation', () => {
   });
 });
 
+describe('snapshot requests trace propagation', () => {
+  it('injects traceparent on snapshots.load, propagating the active context to workflow-server', async () => {
+    const { encodeSnapshotEnvelope } = await import('@workflow/world');
+    const body = encodeSnapshotEnvelope(
+      { eventsCursor: 'evnt_1', createdAt: new Date() },
+      new Uint8Array([1, 2, 3])
+    );
+    const fetchMock = vi.fn().mockResolvedValue(
+      new Response(body, {
+        status: 200,
+        headers: { 'Content-Type': 'application/octet-stream' },
+      })
+    );
+    vi.stubGlobal('fetch', fetchMock);
+    vi.stubEnv(WORKFLOW_SERVER_URL_OVERRIDE, 'https://workflow-server.test');
+
+    const { createSnapshotsStorage } = await import('./snapshots.js');
+    const snapshots = createSnapshotsStorage({ token: 'test-token' });
+
+    const tracer = otelTrace.getTracer('test');
+    let traceId = '';
+    await tracer.startActiveSpan('flow-invocation', async (span) => {
+      traceId = span.spanContext().traceId;
+      const loaded = await snapshots.load('wrun_1');
+      expect(loaded?.data).toBeInstanceOf(Uint8Array);
+      span.end();
+    });
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const calledInit = fetchMock.mock.calls[0][1];
+    const sent = new Headers(calledInit?.headers as HeadersInit);
+    // Without the explicit injection this header is absent — the snapshot
+    // path routes around makeRequest (raw undici / global fetch), which
+    // ambient auto-instrumentation does not reliably hook.
+    const traceparent = sent.get('traceparent');
+    expect(traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/);
+    expect(traceparent).toContain(traceId);
+  });
+
+  it('injects traceparent on snapshots.save (undici request path)', async () => {
+    const agent = new MockAgent();
+    agent.disableNetConnect();
+    let capturedHeaders: Record<string, unknown> | undefined;
+    agent
+      .get('https://vercel-workflow.com')
+      .intercept({
+        path: '/api/v2/runs/wrun_1/snapshot',
+        method: 'PUT',
+      })
+      .reply(200, (opts) => {
+        capturedHeaders = opts.headers as Record<string, unknown>;
+        return 'ok';
+      });
+
+    const { createSnapshotsStorage } = await import('./snapshots.js');
+    const snapshots = createSnapshotsStorage({
+      token: 'test-token',
+      dispatcher: agent,
+    });
+
+    const tracer = otelTrace.getTracer('test');
+    let traceId = '';
+    await tracer.startActiveSpan('flow-invocation', async (span) => {
+      traceId = span.spanContext().traceId;
+      await snapshots.save('wrun_1', new Uint8Array([1, 2, 3]), {
+        eventsCursor: 'evnt_1',
+        createdAt: new Date(),
+      });
+      span.end();
+    });
+
+    const headers = new Headers(
+      (capturedHeaders ?? {}) as Record<string, string>
+    );
+    const traceparent = headers.get('traceparent');
+    expect(traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/);
+    expect(traceparent).toContain(traceId);
+    agent.assertNoPendingInterceptors();
+  });
+});
+
 describe('ws events transport upgrade trace propagation', () => {
   // `openWsChannel` is gated, unlike the `resolveWsTransport` lookup it
   // replaced: nothing opens a channel on the HTTP default.
diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts
index 68d0d2de65..b0447f99d8 100644
--- a/packages/world/src/index.ts
+++ b/packages/world/src/index.ts
@@ -111,6 +111,12 @@ export {
   PaginatedResponseSchema,
   StructuredErrorSchema,
 } from './shared.js';
+export type * from './snapshots.js';
+export {
+  decodeSnapshotEnvelope,
+  encodeSnapshotEnvelope,
+  SnapshotMetadataSchema,
+} from './snapshots.js';
 export type { SpecVersion } from './spec-version.js';
 export {
   isLegacySpecVersion,
diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts
index e0184a0d2e..6bd3a27e49 100644
--- a/packages/world/src/interfaces.ts
+++ b/packages/world/src/interfaces.ts
@@ -29,6 +29,7 @@ import type {
   StreamChunksResponse,
   StreamInfoResponse,
 } from './shared.js';
+import type { SnapshotMetadata } from './snapshots.js';
 import type {
   GetStepParams,
   ListWorkflowRunStepsParams,
@@ -315,6 +316,67 @@ export interface Storage {
      */
     list(params: ListHooksParams): Promise<PaginatedResponse<Hook>>;
   };
+
+  /**
+   * VM snapshot storage for VM-memory snapshotting. OPTIONAL: a World
+   * that cannot (or chooses not to) store multi-MB blobs simply omits
+   * it, and the runtime falls back to full event replay — snapshots are
+   * an optimization, never a correctness requirement. Consumers must
+   * feature-detect (`world.snapshots?.…`).
+   *
+   * Snapshots capture the state of the QuickJS WASM VM at a suspension
+   * point, allowing workflow execution to resume from the exact point of
+   * suspension instead of replaying the full event log.
+   *
+   * The metadata (including eventsCursor) is stored alongside the snapshot
+   * data so that on restore, only events created after the snapshot need
+   * to be fetched. Implementations MUST round-trip the metadata object
+   * losslessly and atomically with the bytes it describes — a snapshot
+   * paired with another suspension's metadata replays from the wrong log
+   * position and silently diverges. The `encodeSnapshotEnvelope` /
+   * `decodeSnapshotEnvelope` helpers pack both into one self-describing
+   * blob so a plain blob store satisfies this with a single atomic
+   * write; worlds with transactional metadata storage may store the
+   * fields natively instead.
+   */
+  snapshots?: {
+    /**
+     * Save a VM snapshot for a workflow run.
+     * Each save overwrites the previous snapshot for this run.
+     *
+     * @param runId - The workflow run ID
+     * @param data - The serialized snapshot bytes (from QuickJS.serializeSnapshot())
+     * @param metadata - Snapshot metadata including the events cursor
+     */
+    save(
+      runId: string,
+      data: Uint8Array,
+      metadata: SnapshotMetadata
+    ): Promise<void>;
+
+    /**
+     * Load the most recent VM snapshot for a workflow run.
+     * Returns null if no snapshot exists (first invocation).
+     *
+     * @param runId - The workflow run ID
+     * @returns The snapshot data and metadata, or null if not found
+     */
+    load(
+      runId: string
+    ): Promise<{ data: Uint8Array; metadata: SnapshotMetadata } | null>;
+
+    /**
+     * Delete the snapshot for a workflow run.
+     * Called when the workflow reaches a terminal state (completed,
+     * failed, cancelled). MUST be idempotent: terminal-state cleanup is
+     * exactly the path most likely to retry or run for a run that never
+     * snapshotted, so deleting a nonexistent snapshot resolves
+     * successfully.
+     *
+     * @param runId - The workflow run ID
+     */
+    delete(runId: string): Promise<void>;
+  };
 }
 
 /**
diff --git a/packages/world/src/snapshots.test.ts b/packages/world/src/snapshots.test.ts
new file mode 100644
index 0000000000..4bded6e82d
--- /dev/null
+++ b/packages/world/src/snapshots.test.ts
@@ -0,0 +1,86 @@
+import { describe, expect, it } from 'vitest';
+import {
+  decodeSnapshotEnvelope,
+  encodeSnapshotEnvelope,
+  type SnapshotMetadata,
+} from './snapshots.js';
+
+const metadata: SnapshotMetadata = {
+  eventsCursor: 'evnt_01ABC',
+  createdAt: new Date('2025-06-01T12:00:00.000Z'),
+};
+
+describe('snapshot envelope', () => {
+  it('round-trips metadata and data through one blob', () => {
+    const data = new Uint8Array(4096).map((_, i) => i % 251);
+    const envelope = encodeSnapshotEnvelope(metadata, data);
+    const decoded = decodeSnapshotEnvelope(envelope);
+    expect(decoded).not.toBeNull();
+    expect(decoded!.metadata.eventsCursor).toBe('evnt_01ABC');
+    expect(decoded!.metadata.createdAt).toEqual(metadata.createdAt);
+    expect(Buffer.from(decoded!.data)).toEqual(Buffer.from(data));
+  });
+
+  it('round-trips a null events cursor (no truthiness coercion)', () => {
+    const envelope = encodeSnapshotEnvelope(
+      { ...metadata, eventsCursor: null },
+      new Uint8Array([1])
+    );
+    const decoded = decodeSnapshotEnvelope(envelope);
+    expect(decoded!.metadata.eventsCursor).toBeNull();
+  });
+
+  it('round-trips metadata fields this schema does not know about (forward compat)', () => {
+    // The envelope carries the WHOLE metadata object — fields added by a
+    // NEWER schema must survive a decode by this one, or adding metadata
+    // would silently drop it through older storage layers.
+    const rich = {
+      ...metadata,
+      eventCount: 42,
+      rngDraws: 7,
+      formatVersion: 2,
+    } as SnapshotMetadata;
+    const decoded = decodeSnapshotEnvelope(
+      encodeSnapshotEnvelope(rich, new Uint8Array([9]))
+    );
+    expect(decoded!.metadata).toMatchObject({ eventsCursor: 'evnt_01ABC' });
+    expect((decoded!.metadata as Record<string, unknown>).eventCount).toBe(42);
+    expect((decoded!.metadata as Record<string, unknown>).rngDraws).toBe(7);
+  });
+
+  it('returns null for non-envelope bytes (legacy/foreign blob)', () => {
+    expect(decodeSnapshotEnvelope(new Uint8Array([1, 2, 3]))).toBeNull();
+    expect(decodeSnapshotEnvelope(new Uint8Array(64).fill(0xab))).toBeNull();
+    expect(decodeSnapshotEnvelope(new Uint8Array(0))).toBeNull();
+  });
+
+  it('returns null for an unknown envelope version', () => {
+    const envelope = encodeSnapshotEnvelope(metadata, new Uint8Array([1]));
+    envelope[4] = 99;
+    expect(decodeSnapshotEnvelope(envelope)).toBeNull();
+  });
+
+  it('returns null for a truncated envelope', () => {
+    const envelope = encodeSnapshotEnvelope(metadata, new Uint8Array(100));
+    // Cut into the metadata region.
+    expect(decodeSnapshotEnvelope(envelope.subarray(0, 12))).toBeNull();
+  });
+
+  it('returns null for schema-invalid metadata', () => {
+    const bad = encodeSnapshotEnvelope(
+      { eventsCursor: 123, createdAt: 'not-a-date' } as never,
+      new Uint8Array([1])
+    );
+    expect(decodeSnapshotEnvelope(bad)).toBeNull();
+  });
+
+  it('decodes from a non-zero byteOffset view', () => {
+    const envelope = encodeSnapshotEnvelope(metadata, new Uint8Array([7, 8]));
+    const padded = new Uint8Array(envelope.length + 16);
+    padded.set(envelope, 16);
+    const view = padded.subarray(16);
+    const decoded = decodeSnapshotEnvelope(view);
+    expect(decoded).not.toBeNull();
+    expect(Array.from(decoded!.data)).toEqual([7, 8]);
+  });
+});
diff --git a/packages/world/src/snapshots.ts b/packages/world/src/snapshots.ts
new file mode 100644
index 0000000000..f87bd9011b
--- /dev/null
+++ b/packages/world/src/snapshots.ts
@@ -0,0 +1,96 @@
+import { z } from 'zod';
+
+export const SnapshotMetadataSchema = z.object({
+  /**
+   * Pagination cursor for events.list() — the snapshot was taken at
+   * this point in the event log. On restore, only events AFTER this
+   * cursor need to be fetched.
+   */
+  eventsCursor: z.string().nullable(),
+  /** Timestamp when the snapshot was created */
+  createdAt: z.coerce.date(),
+});
+
+export type SnapshotMetadata = z.infer<typeof SnapshotMetadataSchema>;
+
+// ---------------------------------------------------------------------------
+// Snapshot envelope
+//
+// Packs the metadata and the snapshot bytes into ONE self-describing blob so
+// that a World backed by a plain blob store can persist both with a single
+// atomic write. Two separate writes (bytes here, metadata there) can tear: a
+// crash between them — or a concurrent load interleaving them — pairs bytes
+// from one suspension with an eventsCursor from another, and the restore
+// replays from the wrong log position and silently diverges. The envelope
+// makes that structurally impossible, and it also means new metadata fields
+// round-trip through every envelope-based World without storage changes.
+//
+// Layout (little-endian):
+//   bytes 0..4   magic "WSNP"
+//   byte  4      envelope format version (1)
+//   bytes 5..9   u32 metadata JSON byte length
+//   bytes 9..9+N metadata JSON (UTF-8, SnapshotMetadataSchema-valid)
+//   bytes 9+N..  snapshot data (opaque)
+// ---------------------------------------------------------------------------
+
+const ENVELOPE_MAGIC = [0x57, 0x53, 0x4e, 0x50]; // "WSNP"
+const ENVELOPE_VERSION = 1;
+const ENVELOPE_HEADER_LEN = 9;
+
+/** Encode a snapshot's metadata and bytes into one atomic blob. */
+export function encodeSnapshotEnvelope(
+  metadata: SnapshotMetadata,
+  data: Uint8Array
+): Uint8Array {
+  const metaBytes = new TextEncoder().encode(JSON.stringify(metadata));
+  const out = new Uint8Array(
+    ENVELOPE_HEADER_LEN + metaBytes.length + data.length
+  );
+  out.set(ENVELOPE_MAGIC, 0);
+  out[4] = ENVELOPE_VERSION;
+  new DataView(out.buffer).setUint32(5, metaBytes.length, true);
+  out.set(metaBytes, ENVELOPE_HEADER_LEN);
+  out.set(data, ENVELOPE_HEADER_LEN + metaBytes.length);
+  return out;
+}
+
+/**
+ * Decode a snapshot envelope. Returns null for anything that is not a
+ * well-formed, schema-valid envelope (wrong magic, unknown version,
+ * truncated, invalid JSON, schema violation) — the caller treats that as
+ * a clean miss (full replay) rather than restoring from fabricated or
+ * torn state. Never invents metadata.
+ */
+export function decodeSnapshotEnvelope(
+  bytes: Uint8Array
+): { metadata: SnapshotMetadata; data: Uint8Array } | null {
+  if (bytes.length < ENVELOPE_HEADER_LEN) return null;
+  for (let i = 0; i < ENVELOPE_MAGIC.length; i++) {
+    if (bytes[i] !== ENVELOPE_MAGIC[i]) return null;
+  }
+  if (bytes[4] !== ENVELOPE_VERSION) return null;
+  const metaLen = new DataView(
+    bytes.buffer,
+    bytes.byteOffset,
+    bytes.byteLength
+  ).getUint32(5, true);
+  if (ENVELOPE_HEADER_LEN + metaLen > bytes.length) return null;
+  try {
+    const metaJson = new TextDecoder().decode(
+      bytes.subarray(ENVELOPE_HEADER_LEN, ENVELOPE_HEADER_LEN + metaLen)
+    );
+    // `.passthrough()`: metadata fields introduced by a newer schema
+    // survive a decode by this one (the whole point of enveloping the
+    // metadata is that new fields never require storage changes) —
+    // known fields are still validated.
+    const metadata = SnapshotMetadataSchema.passthrough().parse(
+      JSON.parse(metaJson)
+    ) as SnapshotMetadata;
+    return {
+      metadata,
+      data: bytes.subarray(ENVELOPE_HEADER_LEN + metaLen),
+    };
+  } catch {
+    return null;
+  }
+}