Add world.snapshots storage interface (local, postgres, vercel)

.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).

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(),
   };
 }

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);
+    }
+  });
+});

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 });
+    },
+  };
+}

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
+);

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
     }
   ]
 }

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',
   {

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),
   };
 }
 

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));
+    },
+  };
+}