diff --git a/core/archive/__tests__/snapshot.test.ts b/core/archive/__tests__/snapshot.test.ts
new file mode 100644
index 0000000..bf1fe41
--- /dev/null
+++ b/core/archive/__tests__/snapshot.test.ts
@@ -0,0 +1,710 @@
+/**
+ * Tests for snapshot capture and restore verification (core/archive/snapshot.ts).
+ *
+ * Coverage:
+ *   - captureSnapshot: stamps records with schema/ledger/hash/index
+ *   - verifyAndRestore: normal path — all invariants pass, index rebuilt, onRestore called
+ *   - verifyAndRestore: invariant failure — throws SnapshotMismatchError with all violations
+ *   - verifyAndRestore: schema version regression — blocked unless allowSchemaMigration
+ *   - verifyAndRestore: schema version upgrade — blocked (newer snapshot on older runtime)
+ *   - verifyAndRestore: ledger sequence regression — blocked when below minLedgerSequence
+ *   - verifyAndRestore: content hash mismatch — blocked
+ *   - verifyAndRestore: index drift reported (not thrown)
+ *   - verifyAndRestore: dryRun does not call onRestore
+ *   - preflight: returns Result instead of throwing
+ *   - Rollback sequence: capture pre-restore → attempt restore → re-apply pre-restore on failure
+ *   - Audit trail: projection.replayed written on success and on failure
+ *   - Telemetry: snapshot.restore_succeeded / snapshot.restore_failed emitted
+ *   - Multiple simultaneous invariant violations all reported together
+ *   - Throwing invariant check is recorded as violation, not re-thrown
+ */
+
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  captureSnapshot,
+  verifyAndRestore,
+  preflight,
+  SnapshotMismatchError,
+  type Snapshot,
+  type SnapshotInvariant
+} from "@/core/archive/snapshot";
+import { createIsolatedAuditTrail } from "@/core/audit/audit";
+import { createCaptureSink, setTelemetrySink, resetTelemetrySink } from "@/core/telemetry/telemetry";
+
+// ─── Fixture types ────────────────────────────────────────────────────────────
+
+interface TicketRecord {
+  id: string;
+  eventId: string;
+  ownerId: string;
+  status: "active" | "redeemed" | "cancelled";
+  amount: number;
+}
+
+function makeTicket(overrides: Partial<TicketRecord> = {}): TicketRecord {
+  return {
+    id: `ticket-${Math.random().toString(36).slice(2, 8)}`,
+    eventId: "event-abc",
+    ownerId: "owner-xyz",
+    status: "active",
+    amount: 100,
+    ...overrides
+  };
+}
+
+const INVARIANTS: readonly SnapshotInvariant<TicketRecord>[] = [
+  {
+    name: "ticket.has_owner",
+    description: "Every ticket must have a non-empty ownerId.",
+    check: (r) => typeof r.ownerId === "string" && r.ownerId.length > 0
+  },
+  {
+    name: "ticket.positive_amount",
+    description: "Ticket amount must be a positive number.",
+    check: (r) => typeof r.amount === "number" && r.amount > 0
+  },
+  {
+    name: "ticket.known_status",
+    description: "Status must be active, redeemed, or cancelled.",
+    check: (r) => ["active", "redeemed", "cancelled"].includes(r.status)
+  }
+];
+
+function buildIndex(records: readonly TicketRecord[]) {
+  const idx: Record<string, string> = {};
+  for (const r of records) idx[`ticket:${r.id}`] = r.ownerId;
+  return idx;
+}
+
+function hashRecords(records: readonly TicketRecord[]): string {
+  return `hash:${records.map((r) => r.id).sort().join(",")}`;
+}
+
+// ─── captureSnapshot ──────────────────────────────────────────────────────────
+
+describe("captureSnapshot", () => {
+  it("stamps the snapshot with schema version and ledger sequence", () => {
+    const records = [makeTicket()];
+    const snap = captureSnapshot(records, { schemaVersion: 2, ledgerSequence: 500 });
+    expect(snap.schemaVersion).toBe(2);
+    expect(snap.ledgerSequence).toBe(500);
+  });
+
+  it("records a capturedAt timestamp", () => {
+    const now = 1_700_000_000_000;
+    const snap = captureSnapshot([makeTicket()], {
+      schemaVersion: 1,
+      ledgerSequence: 1,
+      now: () => now
+    });
+    expect(snap.capturedAt).toBe(new Date(now).toISOString());
+  });
+
+  it("assigns a unique id per snapshot", () => {
+    const records = [makeTicket()];
+    const a = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 1 });
+    const b = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 1 });
+    expect(a.id).not.toBe(b.id);
+  });
+
+  it("builds the derived index when indexBuilder is supplied", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 1,
+      indexBuilder: buildIndex
+    });
+    expect(snap.derivedIndex["ticket:t1"]).toBe("alice");
+  });
+
+  it("produces an empty derivedIndex when no indexBuilder is given", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 1 });
+    expect(snap.derivedIndex).toEqual({});
+  });
+
+  it("records the content hash when hashFn is supplied", () => {
+    const records = [makeTicket({ id: "t1" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 1,
+      hashFn: hashRecords
+    });
+    expect(snap.contentHash).toBe(hashRecords(records));
+  });
+
+  it("stores an empty contentHash when no hashFn is given", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 1 });
+    expect(snap.contentHash).toBe("");
+  });
+
+  it("preserves primaryRecords by reference", () => {
+    const records = [makeTicket(), makeTicket()];
+    const snap = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 1 });
+    expect(snap.primaryRecords).toHaveLength(2);
+    expect(snap.primaryRecords[0].id).toBe(records[0].id);
+  });
+});
+
+// ─── verifyAndRestore — normal path ───────────────────────────────────────────
+
+describe("verifyAndRestore — normal path", () => {
+  it("returns a SnapshotRestoreResult when all invariants pass", () => {
+    const records = [makeTicket(), makeTicket()];
+    const snap = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 100 });
+    const result = verifyAndRestore(snap, { invariants: INVARIANTS });
+    expect(result.snapshotId).toBe(snap.id);
+    expect(result.primaryRecordCount).toBe(2);
+    expect(result.dryRun).toBe(false);
+  });
+
+  it("calls onRestore with the primary records and rebuilt index", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      indexBuilder: buildIndex
+    });
+    const onRestore = vi.fn();
+    verifyAndRestore(snap, { invariants: INVARIANTS, indexBuilder: buildIndex, onRestore });
+    expect(onRestore).toHaveBeenCalledOnce();
+    const [restoredRecords, rebuiltIndex] = onRestore.mock.calls[0];
+    expect(restoredRecords).toEqual(records);
+    expect(rebuiltIndex["ticket:t1"]).toBe("alice");
+  });
+
+  it("reports zero index mismatches when rebuilt index matches snapshot", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      indexBuilder: buildIndex
+    });
+    const result = verifyAndRestore(snap, { indexBuilder: buildIndex });
+    expect(result.indexMismatches).toHaveLength(0);
+  });
+
+  it("does not throw for an empty record set", () => {
+    const snap = captureSnapshot<TicketRecord>([], { schemaVersion: 1, ledgerSequence: 1 });
+    expect(() => verifyAndRestore(snap, { invariants: INVARIANTS })).not.toThrow();
+  });
+
+  it("returns the restoredAt timestamp", () => {
+    const now = 1_700_000_000_000;
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 1 });
+    const result = verifyAndRestore(snap, { now: () => now });
+    expect(result.restoredAt).toBe(new Date(now).toISOString());
+  });
+});
+
+// ─── Invariant violations — throw path ───────────────────────────────────────
+
+describe("verifyAndRestore — invariant failure (loud throw)", () => {
+  it("throws SnapshotMismatchError when an invariant fails", () => {
+    const records = [makeTicket({ ownerId: "" })]; // violates ticket.has_owner
+    const snap = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 100 });
+    expect(() => verifyAndRestore(snap, { invariants: INVARIANTS })).toThrow(
+      SnapshotMismatchError
+    );
+  });
+
+  it("SnapshotMismatchError carries all violations, not just the first", () => {
+    const bad = makeTicket({ ownerId: "", amount: -1 }); // violates 2 invariants
+    const snap = captureSnapshot([bad], { schemaVersion: 1, ledgerSequence: 100 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { invariants: INVARIANTS });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error).toBeInstanceOf(SnapshotMismatchError);
+    expect(error.violations).toHaveLength(2);
+    const names = error.violations.map((v) => v.checkName);
+    expect(names).toContain("ticket.has_owner");
+    expect(names).toContain("ticket.positive_amount");
+  });
+
+  it("every violation identifies the failing record id", () => {
+    const bad = makeTicket({ id: "bad-ticket", ownerId: "" });
+    const snap = captureSnapshot([bad], { schemaVersion: 1, ledgerSequence: 1 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { invariants: [INVARIANTS[0]] });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.violations[0].targetId).toBe("bad-ticket");
+  });
+
+  it("reports violations from all records, not just the first failing one", () => {
+    const records = [
+      makeTicket({ id: "t1", ownerId: "" }),
+      makeTicket({ id: "t2", ownerId: "" })
+    ];
+    const snap = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 1 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { invariants: [INVARIANTS[0]] });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    const ids = error.violations.map((v) => v.targetId);
+    expect(ids).toContain("t1");
+    expect(ids).toContain("t2");
+  });
+
+  it("a throwing invariant check is recorded as a violation, not re-thrown", () => {
+    const throwingInvariant: SnapshotInvariant<TicketRecord> = {
+      name: "ticket.throws",
+      check: () => {
+        throw new Error("invariant exploded");
+      }
+    };
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 1 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { invariants: [throwingInvariant] });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error).toBeInstanceOf(SnapshotMismatchError);
+    expect(error.violations[0].kind).toBe("invariant_failed");
+    expect(error.violations[0].detail).toContain("invariant exploded");
+  });
+
+  it("does NOT call onRestore when violations are found", () => {
+    const onRestore = vi.fn();
+    const bad = makeTicket({ ownerId: "" });
+    const snap = captureSnapshot([bad], { schemaVersion: 1, ledgerSequence: 1 });
+    try {
+      verifyAndRestore(snap, { invariants: INVARIANTS, onRestore });
+    } catch {
+      // expected
+    }
+    expect(onRestore).not.toHaveBeenCalled();
+  });
+
+  it("carries the snapshotId and correlationId in the error", () => {
+    const snap = captureSnapshot([makeTicket({ ownerId: "" })], {
+      schemaVersion: 1,
+      ledgerSequence: 1,
+      newId: () => "snap-id-fixed"
+    });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { invariants: INVARIANTS });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.snapshotId).toBe("snap-id-fixed");
+    expect(typeof error.correlationId).toBe("string");
+  });
+});
+
+// ─── Schema version checks ────────────────────────────────────────────────────
+
+describe("verifyAndRestore — schema version checks", () => {
+  it("blocks a schema downgrade (snapshot older than runtime)", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    expect(() =>
+      verifyAndRestore(snap, { expectedSchemaVersion: 2 })
+    ).toThrow(SnapshotMismatchError);
+  });
+
+  it("violation kind is schema_version_mismatch for downgrade", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { expectedSchemaVersion: 2 });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.violations[0].kind).toBe("schema_version_mismatch");
+    expect(error.violations[0].observed).toBe(1);
+    expect(error.violations[0].expected).toBe(2);
+  });
+
+  it("allows a schema downgrade when allowSchemaMigration is true", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    expect(() =>
+      verifyAndRestore(snap, { expectedSchemaVersion: 2, allowSchemaMigration: true })
+    ).not.toThrow();
+  });
+
+  it("blocks a schema upgrade (snapshot newer than runtime)", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 3, ledgerSequence: 100 });
+    expect(() =>
+      verifyAndRestore(snap, { expectedSchemaVersion: 2 })
+    ).toThrow(SnapshotMismatchError);
+  });
+
+  it("version_mismatch violation contains observed > expected for upgrade", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 5, ledgerSequence: 1 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { expectedSchemaVersion: 3 });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.violations[0].observed).toBe(5);
+    expect(error.violations[0].expected).toBe(3);
+  });
+
+  it("passes when snapshot schema matches expected schema", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 2, ledgerSequence: 100 });
+    expect(() =>
+      verifyAndRestore(snap, { expectedSchemaVersion: 2 })
+    ).not.toThrow();
+  });
+});
+
+// ─── Ledger sequence regression ───────────────────────────────────────────────
+
+describe("verifyAndRestore — ledger sequence regression", () => {
+  it("blocks a restore when ledgerSequence is below minLedgerSequence", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 50 });
+    expect(() =>
+      verifyAndRestore(snap, { minLedgerSequence: 100 })
+    ).toThrow(SnapshotMismatchError);
+  });
+
+  it("violation kind is ledger_sequence_regression", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 50 });
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(snap, { minLedgerSequence: 100 });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.violations[0].kind).toBe("ledger_sequence_regression");
+    expect(error.violations[0].observed).toBe(50);
+    expect(error.violations[0].expected).toBe(100);
+  });
+
+  it("allows restore at exactly minLedgerSequence", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    expect(() =>
+      verifyAndRestore(snap, { minLedgerSequence: 100 })
+    ).not.toThrow();
+  });
+
+  it("allows restore above minLedgerSequence", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 200 });
+    expect(() =>
+      verifyAndRestore(snap, { minLedgerSequence: 100 })
+    ).not.toThrow();
+  });
+});
+
+// ─── Content hash mismatch ────────────────────────────────────────────────────
+
+describe("verifyAndRestore — content hash mismatch", () => {
+  it("blocks restore when records were tampered after capture", () => {
+    const records = [makeTicket({ id: "t1" })];
+    // Capture with a hash.
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      hashFn: hashRecords
+    });
+
+    // Simulate tamper: replace primaryRecords directly (freeze bypass via spread).
+    const tampered: Snapshot<TicketRecord> = {
+      ...snap,
+      primaryRecords: [makeTicket({ id: "tampered" })]
+    };
+
+    expect(() =>
+      verifyAndRestore(tampered, { hashFn: hashRecords })
+    ).toThrow(SnapshotMismatchError);
+  });
+
+  it("violation kind is content_hash_mismatch", () => {
+    const snap = captureSnapshot([makeTicket({ id: "t1" })], {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      hashFn: hashRecords
+    });
+    const tampered: Snapshot<TicketRecord> = {
+      ...snap,
+      primaryRecords: [makeTicket({ id: "tampered" })]
+    };
+    let error!: SnapshotMismatchError;
+    try {
+      verifyAndRestore(tampered, { hashFn: hashRecords });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) error = e;
+    }
+    expect(error.violations[0].kind).toBe("content_hash_mismatch");
+  });
+
+  it("skips hash check when snapshot.contentHash is empty", () => {
+    // snapshot without hash
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 1 });
+    expect(snap.contentHash).toBe("");
+    // Even if caller supplies hashFn, the check is skipped for hash-less snapshots.
+    expect(() =>
+      verifyAndRestore(snap, { hashFn: hashRecords })
+    ).not.toThrow();
+  });
+
+  it("passes when hash matches", () => {
+    const records = [makeTicket({ id: "t1" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      hashFn: hashRecords
+    });
+    expect(() =>
+      verifyAndRestore(snap, { hashFn: hashRecords })
+    ).not.toThrow();
+  });
+});
+
+// ─── Index drift ──────────────────────────────────────────────────────────────
+
+describe("verifyAndRestore — index drift (reported, not thrown)", () => {
+  it("reports index mismatches without throwing", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    // Capture with a stale/wrong index.
+    const snap: Snapshot<TicketRecord> = {
+      ...captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 100 }),
+      derivedIndex: { "ticket:t1": "old-owner" }
+    };
+    // Rebuild will produce "alice" but snapshot stored "old-owner".
+    const result = verifyAndRestore(snap, { indexBuilder: buildIndex });
+    expect(result.indexMismatches).toHaveLength(1);
+    expect(result.indexMismatches[0].key).toBe("ticket:t1");
+    expect(result.indexMismatches[0].snapshotValue).toBe("old-owner");
+    expect(result.indexMismatches[0].rebuiltValue).toBe("alice");
+  });
+
+  it("reports new keys added by the rebuild (present in rebuilt, absent in snapshot)", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    // Capture without an index, then rebuild includes the key.
+    const snap = captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 100 });
+    // derivedIndex is {} but rebuild will produce { "ticket:t1": "alice" }
+    const result = verifyAndRestore(snap, { indexBuilder: buildIndex });
+    expect(result.indexMismatches.some((m) => m.key === "ticket:t1")).toBe(true);
+    expect(result.indexMismatches.find((m) => m.key === "ticket:t1")?.snapshotValue).toBeUndefined();
+  });
+
+  it("reports keys removed by the rebuild (present in snapshot, absent in rebuilt)", () => {
+    const records = [makeTicket({ id: "t1" })];
+    const snap: Snapshot<TicketRecord> = {
+      ...captureSnapshot(records, { schemaVersion: 1, ledgerSequence: 100 }),
+      derivedIndex: {
+        "ticket:t1": "alice",
+        "ticket:stale": "bob" // no longer in rebuilt
+      }
+    };
+    // buildIndex only keys on present records.
+    const result = verifyAndRestore(snap, { indexBuilder: buildIndex });
+    const stale = result.indexMismatches.find((m) => m.key === "ticket:stale");
+    expect(stale).toBeDefined();
+    expect(stale?.rebuiltValue).toBeUndefined();
+  });
+
+  it("zero mismatches when index is already correct", () => {
+    const records = [makeTicket({ id: "t1", ownerId: "alice" })];
+    const snap = captureSnapshot(records, {
+      schemaVersion: 1,
+      ledgerSequence: 100,
+      indexBuilder: buildIndex
+    });
+    const result = verifyAndRestore(snap, { indexBuilder: buildIndex });
+    expect(result.indexMismatches).toHaveLength(0);
+  });
+});
+
+// ─── dryRun ───────────────────────────────────────────────────────────────────
+
+describe("verifyAndRestore — dryRun", () => {
+  it("dryRun does NOT call onRestore", () => {
+    const onRestore = vi.fn();
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    verifyAndRestore(snap, { invariants: INVARIANTS, dryRun: true, onRestore });
+    expect(onRestore).not.toHaveBeenCalled();
+  });
+
+  it("dryRun returns dryRun: true in the result", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    const result = verifyAndRestore(snap, { dryRun: true });
+    expect(result.dryRun).toBe(true);
+  });
+
+  it("dryRun still throws SnapshotMismatchError on violations", () => {
+    const snap = captureSnapshot([makeTicket({ ownerId: "" })], {
+      schemaVersion: 1,
+      ledgerSequence: 1
+    });
+    expect(() =>
+      verifyAndRestore(snap, { invariants: INVARIANTS, dryRun: true })
+    ).toThrow(SnapshotMismatchError);
+  });
+});
+
+// ─── preflight ────────────────────────────────────────────────────────────────
+
+describe("preflight", () => {
+  it("returns ok Result when all checks pass", () => {
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    const result = preflight(snap, { invariants: INVARIANTS });
+    expect(result.ok).toBe(true);
+  });
+
+  it("returns err Result with violations instead of throwing", () => {
+    const snap = captureSnapshot([makeTicket({ ownerId: "" })], {
+      schemaVersion: 1,
+      ledgerSequence: 1
+    });
+    const result = preflight(snap, { invariants: INVARIANTS });
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.code).toBe("snapshot_invariant_violation");
+    expect(result.detail).toBeDefined();
+    expect((result.detail as unknown[]).length).toBeGreaterThan(0);
+  });
+
+  it("never calls onRestore (dryRun is implicit)", () => {
+    const onRestore = vi.fn();
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    preflight(snap); // no onRestore in preflight signature
+    expect(onRestore).not.toHaveBeenCalled();
+  });
+});
+
+// ─── Rollback sequence ────────────────────────────────────────────────────────
+
+describe("rollback sequence", () => {
+  it("pre-restore snapshot can be re-applied after a failed restore", () => {
+    let store: TicketRecord[] = [makeTicket({ id: "t1", ownerId: "alice" })];
+    const schemaVersion = 1;
+    const ledgerSequence = 100;
+
+    // Step 1: capture the current state as the rollback point.
+    const preRestoreSnap = captureSnapshot(store, {
+      schemaVersion,
+      ledgerSequence,
+      indexBuilder: buildIndex
+    });
+
+    // Step 2: prepare a corrupted incoming snapshot to restore.
+    const corruptedSnap = captureSnapshot([makeTicket({ id: "corrupt", ownerId: "" })], {
+      schemaVersion,
+      ledgerSequence: 200
+    });
+
+    // Step 3: attempt restore — should fail.
+    let restoreError: SnapshotMismatchError | undefined;
+    try {
+      verifyAndRestore(corruptedSnap, { invariants: INVARIANTS });
+    } catch (e) {
+      if (e instanceof SnapshotMismatchError) restoreError = e;
+    }
+    expect(restoreError).toBeInstanceOf(SnapshotMismatchError);
+
+    // Step 4: store was not modified (onRestore was never called).
+    // Rollback: re-apply the pre-restore snapshot.
+    verifyAndRestore(preRestoreSnap, {
+      invariants: INVARIANTS,
+      indexBuilder: buildIndex,
+      onRestore: (records) => {
+        store = [...records];
+      }
+    });
+
+    expect(store).toHaveLength(1);
+    expect(store[0].id).toBe("t1");
+    expect(store[0].ownerId).toBe("alice");
+  });
+
+  it("dryRun preflight prevents bad data from ever reaching onRestore", () => {
+    const store: TicketRecord[] = [];
+    const badSnap = captureSnapshot([makeTicket({ amount: -10 })], {
+      schemaVersion: 1,
+      ledgerSequence: 1
+    });
+    const check = preflight(badSnap, { invariants: INVARIANTS });
+    // Preflight failed — we do NOT proceed to the real restore.
+    expect(check.ok).toBe(false);
+    // Store is untouched.
+    expect(store).toHaveLength(0);
+  });
+});
+
+// ─── Telemetry ────────────────────────────────────────────────────────────────
+
+describe("telemetry events", () => {
+  it("emits snapshot.restore_succeeded on a clean restore", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+      verifyAndRestore(snap, { invariants: INVARIANTS });
+      const event = sink.find((e) => e.op === "snapshot.restore_succeeded");
+      expect(event).toBeDefined();
+      expect(event?.result).toBe("success");
+      expect(event?.payload?.snapshotId).toBe(snap.id);
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+
+  it("emits snapshot.restore_failed before throwing on violation", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      const snap = captureSnapshot([makeTicket({ ownerId: "" })], {
+        schemaVersion: 1,
+        ledgerSequence: 1
+      });
+      try {
+        verifyAndRestore(snap, { invariants: INVARIANTS });
+      } catch {
+        // expected
+      }
+      const event = sink.find((e) => e.op === "snapshot.restore_failed");
+      expect(event).toBeDefined();
+      expect(event?.result).toBe("failure");
+      expect(event?.errorCode).toBe("snapshot_invariant_violation");
+      expect(event?.payload?.violationCount).toBeGreaterThan(0);
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+});
+
+// ─── Audit trail ─────────────────────────────────────────────────────────────
+
+describe("audit trail", () => {
+  it("writes a projection.replayed audit event on success", () => {
+    const trail = createIsolatedAuditTrail();
+    const snap = captureSnapshot([makeTicket()], { schemaVersion: 1, ledgerSequence: 100 });
+    verifyAndRestore(snap, {
+      invariants: INVARIANTS,
+      audit: trail,
+      actor: { kind: "maintainer", id: "ops-bot" }
+    });
+    const events = trail.all().filter((e) => e.action === "projection.replayed");
+    expect(events).toHaveLength(1);
+    expect(events[0].outcome).toBe("allowed");
+    expect(events[0].actor.kind).toBe("maintainer");
+  });
+
+  it("writes a denied projection.replayed audit event on failure", () => {
+    const trail = createIsolatedAuditTrail();
+    const snap = captureSnapshot([makeTicket({ ownerId: "" })], {
+      schemaVersion: 1,
+      ledgerSequence: 1
+    });
+    try {
+      verifyAndRestore(snap, { invariants: INVARIANTS, audit: trail });
+    } catch {
+      // expected
+    }
+    const events = trail.all().filter((e) => e.action === "projection.replayed");
+    expect(events).toHaveLength(1);
+    expect(events[0].outcome).toBe("denied");
+    expect(events[0].errorCode).toBe("snapshot_invariant_violation");
+  });
+});
diff --git a/core/archive/index.ts b/core/archive/index.ts
index 70e199f..d2b4e9d 100644
--- a/core/archive/index.ts
+++ b/core/archive/index.ts
@@ -1 +1,2 @@
 export * from "./archive";
+export * from "./snapshot";
diff --git a/core/archive/snapshot.ts b/core/archive/snapshot.ts
new file mode 100644
index 0000000..b3e2120
--- /dev/null
+++ b/core/archive/snapshot.ts
@@ -0,0 +1,608 @@
+/**
+ * Snapshot capture and restore verification for critical data and derived indexes.
+ *
+ * Context: Utix stores state in the browser (localStorage / in-memory). When a
+ * session restores from a snapshot — on re-hydration, after a forced rollback or
+ * after a data migration — the raw values that come back may be inconsistent with
+ * each other: the primary records may not match what the derived indexes point to,
+ * schema versions may have moved on, or the ledger sequence may have regressed.
+ *
+ * This module provides a two-phase restore workflow:
+ *
+ *   1. **Capture** (`captureSnapshot`) — takes a point-in-time snapshot of primary
+ *      records and their derived index entries. The snapshot is stamped with a
+ *      ledger sequence, schema version and a content hash so its integrity can be
+ *      checked before restore.
+ *
+ *   2. **Verify and restore** (`verifyAndRestore`) — given a snapshot, checks every
+ *      declared invariant, rebuilds derived indexes from the primary records, and
+ *      fails loudly with `SnapshotMismatchError` if any check fails. "Loudly" means
+ *      the function throws a `SnapshotMismatchError` (never swallows silently) AND
+ *      emits a `snapshot.restore_failed` telemetry event AND writes an audit record
+ *      so operators have the full picture.
+ *
+ * Invariants are supplied by the caller, so this module does not need to know what
+ * makes a `TicketRecord` or an `EventRecord` valid. An invariant is a predicate plus
+ * a human-readable name; a failing one produces a `SnapshotViolation` that names
+ * exactly what broke and on which record.
+ *
+ * Derived index rebuild
+ * ─────────────────────
+ * After the invariants pass, the caller may supply an `IndexBuilder` — a function
+ * that accepts the restored primary records and returns a fresh index map. This
+ * runs deterministically from the primaries (same input → same index) and replaces
+ * whatever was in the snapshot's index. If the rebuilt index differs from the
+ * snapshot's index, the difference is captured in a `SnapshotIndexMismatch` and
+ * surfaced in the result (not thrown, because an index drift is expected during a
+ * migration and the caller decides whether it is acceptable).
+ *
+ * Operational edge cases and recovery
+ * ────────────────────────────────────
+ * 1. **Invariant violation** — `verifyAndRestore` throws `SnapshotMismatchError`.
+ *    Every failing invariant is listed in `error.violations`. The snapshot is NOT
+ *    applied; the previous state is intact. Recovery: inspect the violations, fix
+ *    the snapshot data, then re-run. Use `dryRun: true` to check without applying.
+ *
+ * 2. **Schema version regression** — if the snapshot's `schemaVersion` is lower
+ *    than the `expectedSchemaVersion` passed to `verifyAndRestore`, a
+ *    `schema_version_mismatch` violation is emitted. This blocks the restore until
+ *    a migration plan upgrades the snapshot. Recovery: run the relevant migration
+ *    and re-capture, or pass `allowSchemaMigration: true` to accept old-schema
+ *    restores with a warning.
+ *
+ * 3. **Ledger sequence regression** — if the snapshot's `ledgerSequence` is older
+ *    than `minLedgerSequence`, a `ledger_sequence_regression` violation is added.
+ *    This prevents replaying an outdated Stellar state on top of a newer in-memory
+ *    state. Recovery: restore from a newer snapshot, or confirm the regression is
+ *    intentional and pass a lower `minLedgerSequence`.
+ *
+ * 4. **Checksum mismatch** — if `captureSnapshot` ran with `hashFn` and the
+ *    snapshot's `contentHash` no longer matches the re-hashed primary records,
+ *    a `content_hash_mismatch` violation is added. This catches tampering or
+ *    accidental partial writes. Recovery: discard the snapshot and re-capture from
+ *    a trusted source.
+ *
+ * 5. **Index drift after rebuild** — the rebuilt index differs from the snapshot's.
+ *    This is NOT a hard error: indexes are derived and drifts are expected after a
+ *    migration. The `SnapshotRestoreResult.indexMismatches` array describes every
+ *    differing key. Callers that need strict index fidelity can check it and refuse.
+ *
+ * 6. **Rollback** — set `dryRun: true` to validate without writing. A rollback plan:
+ *    (a) capture a pre-restore snapshot with `captureSnapshot`, (b) attempt restore
+ *    with `verifyAndRestore`, (c) on `SnapshotMismatchError`, re-apply the
+ *    pre-restore snapshot. The pre-restore capture is the authoritative rollback
+ *    point. See the test file for the rollback sequence.
+ *
+ * Compatibility
+ * ─────────────
+ * This module adds no new fields to `ArchiveRecord` or any other existing type.
+ * `Snapshot` is a new standalone type; `verifyAndRestore` returns a new
+ * `SnapshotRestoreResult` type. No existing API is changed.
+ */
+
+import {
+  recordAudit,
+  type AuditActor,
+  type AuditInput,
+  type AuditTrail
+} from "@/core/audit/audit";
+import { err, ok, type Result } from "@/core/result/result";
+import { emitTelemetry, newCorrelationId } from "@/core/telemetry/telemetry";
+
+// ─── Core types ───────────────────────────────────────────────────────────────
+
+export type SnapshotErrorCode =
+  | "snapshot_invariant_violation"
+  | "snapshot_schema_mismatch"
+  | "snapshot_ledger_regression"
+  | "snapshot_hash_mismatch"
+  | "snapshot_empty"
+  | "snapshot_unauthorized";
+
+export type SnapshotViolationKind =
+  | "invariant_failed"
+  | "schema_version_mismatch"
+  | "ledger_sequence_regression"
+  | "content_hash_mismatch";
+
+export interface SnapshotViolation {
+  /** Machine-readable kind, stable for switch statements. */
+  readonly kind: SnapshotViolationKind;
+  /** Which invariant or check produced this violation. */
+  readonly checkName: string;
+  /** The record ID that failed, or `"snapshot"` for top-level checks. */
+  readonly targetId: string;
+  /** Human-readable description of exactly what failed. */
+  readonly detail: string;
+  /** The value that was observed (serialisable). */
+  readonly observed?: unknown;
+  /** The value that was expected (serialisable). */
+  readonly expected?: unknown;
+}
+
+/**
+ * Thrown (never returned) when `verifyAndRestore` detects a violation.
+ * Thrown rather than returned because a partial restore is a data-integrity
+ * failure: the caller must handle it, and an uncaught throw surfaces it
+ * loudly rather than letting it silently propagate as an `ok: false`.
+ */
+export class SnapshotMismatchError extends Error {
+  readonly violations: readonly SnapshotViolation[];
+  readonly snapshotId: string;
+  readonly correlationId: string;
+
+  constructor(
+    violations: readonly SnapshotViolation[],
+    snapshotId: string,
+    correlationId: string
+  ) {
+    const summary = violations
+      .slice(0, 3)
+      .map((v) => `[${v.kind}] ${v.checkName}: ${v.detail}`)
+      .join("; ");
+    const more = violations.length > 3 ? ` (and ${violations.length - 3} more)` : "";
+    super(`Snapshot restore verification failed — ${violations.length} violation(s): ${summary}${more}`);
+    this.name = "SnapshotMismatchError";
+    this.violations = violations;
+    this.snapshotId = snapshotId;
+    this.correlationId = correlationId;
+  }
+}
+
+/**
+ * An invariant the caller wants enforced on every primary record.
+ *
+ * @example
+ * ```ts
+ * const hasOwner: SnapshotInvariant<TicketRecord> = {
+ *   name: "ticket.has_owner",
+ *   check: (record) => typeof record.ownerId === "string" && record.ownerId.length > 0
+ * };
+ * ```
+ */
+export interface SnapshotInvariant<T> {
+  /** Stable dot-separated name used in violation reports and telemetry. */
+  readonly name: string;
+  /** Returns `true` if the record satisfies the invariant. */
+  readonly check: (record: T) => boolean;
+  /** Optional human-readable description included in violation detail. */
+  readonly description?: string;
+}
+
+/** A key→value index built from primary records. */
+export type DerivedIndex = Record<string, string>;
+
+/**
+ * Builds a derived index from a set of primary records. Called after all
+ * invariants pass; the result replaces whatever the snapshot stored.
+ */
+export type IndexBuilder<T> = (records: readonly T[]) => DerivedIndex;
+
+export interface SnapshotIndexMismatch {
+  /** The index key that differs. */
+  readonly key: string;
+  /** What the snapshot stored (undefined if the key is new). */
+  readonly snapshotValue: string | undefined;
+  /** What the rebuild produced (undefined if the key was removed). */
+  readonly rebuiltValue: string | undefined;
+}
+
+/** A point-in-time snapshot of primary records and their derived index. */
+export interface Snapshot<T> {
+  /** Unique stable ID for this snapshot. */
+  readonly id: string;
+  readonly schemaVersion: number;
+  /**
+   * Stellar ledger sequence at capture time. Used to detect regressions:
+   * restoring a snapshot from an older ledger on top of newer in-memory state
+   * is a data-integrity risk.
+   */
+  readonly ledgerSequence: number;
+  readonly capturedAt: string;
+  readonly primaryRecords: readonly T[];
+  /**
+   * Index entries at capture time. These may differ from what a fresh rebuild
+   * produces (e.g. after a migration); the difference is reported, not fatal.
+   */
+  readonly derivedIndex: DerivedIndex;
+  /**
+   * Optional content hash produced by the `hashFn` supplied to
+   * `captureSnapshot`. Empty string when no hash function was provided.
+   */
+  readonly contentHash: string;
+  /** Free-form context for operators. Never contains secrets. */
+  readonly metadata: Record<string, string | number | boolean>;
+}
+
+export interface CaptureOptions<T> {
+  schemaVersion: number;
+  ledgerSequence: number;
+  /**
+   * Index builder used to create `derivedIndex` at capture time. If not
+   * provided, `derivedIndex` will be an empty object.
+   */
+  indexBuilder?: IndexBuilder<T>;
+  /**
+   * Deterministic hash of the primary records. Used to detect post-capture
+   * tampering. If not provided, `contentHash` will be an empty string.
+   */
+  hashFn?: (records: readonly T[]) => string;
+  metadata?: Record<string, string | number | boolean>;
+  /** Injected ID factory. Defaults to `newCorrelationId`. */
+  newId?: () => string;
+  /** Injected clock. Defaults to `Date.now`. */
+  now?: () => number;
+}
+
+export interface VerifyOptions<T> {
+  /**
+   * Invariants every primary record must satisfy.
+   * A single failing record produces a violation entry, not an early exit,
+   * so all violations are reported together.
+   */
+  invariants?: readonly SnapshotInvariant<T>[];
+  /** The schema version the restore target expects. Defaults to `snapshot.schemaVersion`. */
+  expectedSchemaVersion?: number;
+  /**
+   * When set, a snapshot whose `ledgerSequence` is strictly less than this
+   * value is refused with a `ledger_sequence_regression` violation.
+   */
+  minLedgerSequence?: number;
+  /**
+   * When `true`, schema version downgrades are allowed (emit a warning
+   * violation that does not block the restore).
+   */
+  allowSchemaMigration?: boolean;
+  /**
+   * Rebuild the derived index and compare against the snapshot's stored index.
+   * Differences are reported in `indexMismatches` but do not throw.
+   */
+  indexBuilder?: IndexBuilder<T>;
+  /**
+   * Re-hash the primary records and compare against `snapshot.contentHash`.
+   * Only checked when `snapshot.contentHash` is non-empty.
+   */
+  hashFn?: (records: readonly T[]) => string;
+  /**
+   * When `true`, validate everything but do NOT call `onRestore`. Use this
+   * to implement a pre-flight check before committing.
+   */
+  dryRun?: boolean;
+  /**
+   * Called with the verified primary records and rebuilt index when the
+   * restore passes all checks and `dryRun` is false. This is the commit step.
+   */
+  onRestore?: (records: readonly T[], index: DerivedIndex) => void;
+  /** Audit actor for the restore record. Defaults to `system`. */
+  actor?: AuditActor;
+  /** Injected audit trail. Defaults to the app-wide trail via `recordAudit`. */
+  audit?: AuditTrail;
+  /** Injected clock. Defaults to `Date.now`. */
+  now?: () => number;
+}
+
+export interface SnapshotRestoreResult<T> {
+  readonly snapshotId: string;
+  readonly schemaVersion: number;
+  readonly ledgerSequence: number;
+  readonly primaryRecordCount: number;
+  readonly rebuiltIndex: DerivedIndex;
+  readonly indexMismatches: readonly SnapshotIndexMismatch[];
+  readonly dryRun: boolean;
+  readonly restoredAt: string;
+  readonly correlationId: string;
+}
+
+// ─── Capture ──────────────────────────────────────────────────────────────────
+
+/**
+ * Captures a point-in-time snapshot of `records`.
+ *
+ * The returned `Snapshot` is immutable and self-describing: it carries the
+ * schema version, ledger sequence and (optionally) a content hash so its
+ * integrity can be verified before restore.
+ */
+export function captureSnapshot<T>(
+  records: readonly T[],
+  options: CaptureOptions<T>
+): Snapshot<T> {
+  const nowFn = options.now ?? (() => Date.now());
+  const newId = options.newId ?? newCorrelationId;
+  const derivedIndex = options.indexBuilder ? options.indexBuilder(records) : {};
+  const contentHash = options.hashFn ? options.hashFn(records) : "";
+
+  return {
+    id: newId(),
+    schemaVersion: options.schemaVersion,
+    ledgerSequence: options.ledgerSequence,
+    capturedAt: new Date(nowFn()).toISOString(),
+    primaryRecords: records,
+    derivedIndex,
+    contentHash,
+    metadata: options.metadata ?? {}
+  };
+}
+
+// ─── Verify and restore ───────────────────────────────────────────────────────
+
+/**
+ * Verifies a snapshot against declared invariants and restores it if all
+ * checks pass.
+ *
+ * Throws `SnapshotMismatchError` on any hard failure (invariant violation,
+ * schema regression, ledger regression, hash mismatch). Returns
+ * `SnapshotRestoreResult` on success. Index drift is reported but does not
+ * throw.
+ *
+ * The caller supplies `onRestore` to commit the records. This function never
+ * writes state directly; it is a verification pipeline, not a storage layer.
+ */
+export function verifyAndRestore<T>(
+  snapshot: Snapshot<T>,
+  options: VerifyOptions<T> = {}
+): SnapshotRestoreResult<T> {
+  const correlationId = newCorrelationId();
+  const nowFn = options.now ?? (() => Date.now());
+  const at = new Date(nowFn()).toISOString();
+  const actor: AuditActor = options.actor ?? { kind: "system" };
+
+  /** Use injected audit trail when available; fall back to app-wide singleton. */
+  function audit(input: AuditInput): void {
+    if (options.audit) {
+      options.audit.record(input);
+    } else {
+      recordAudit(input);
+    }
+  }
+
+  const violations: SnapshotViolation[] = [];
+
+  // ── Phase 1: Top-level snapshot integrity checks ──────────────────────────
+
+  // 1a. Schema version check
+  const expectedSchema = options.expectedSchemaVersion ?? snapshot.schemaVersion;
+  if (snapshot.schemaVersion < expectedSchema && !options.allowSchemaMigration) {
+    violations.push({
+      kind: "schema_version_mismatch",
+      checkName: "snapshot.schema_version",
+      targetId: "snapshot",
+      detail: `Snapshot schema v${snapshot.schemaVersion} is older than expected v${expectedSchema}. Pass allowSchemaMigration: true if this is an intentional migration.`,
+      observed: snapshot.schemaVersion,
+      expected: expectedSchema
+    });
+  }
+  if (snapshot.schemaVersion > expectedSchema) {
+    // A snapshot from a newer schema cannot be safely restored to an older
+    // runtime — the primary records may contain fields the current code cannot
+    // handle.
+    violations.push({
+      kind: "schema_version_mismatch",
+      checkName: "snapshot.schema_version",
+      targetId: "snapshot",
+      detail: `Snapshot schema v${snapshot.schemaVersion} is newer than runtime v${expectedSchema}. Upgrade the application before restoring this snapshot.`,
+      observed: snapshot.schemaVersion,
+      expected: expectedSchema
+    });
+  }
+
+  // 1b. Ledger sequence regression check
+  if (
+    typeof options.minLedgerSequence === "number" &&
+    snapshot.ledgerSequence < options.minLedgerSequence
+  ) {
+    violations.push({
+      kind: "ledger_sequence_regression",
+      checkName: "snapshot.ledger_sequence",
+      targetId: "snapshot",
+      detail: `Snapshot ledger sequence ${snapshot.ledgerSequence} is below minimum ${options.minLedgerSequence}. Restoring this snapshot would regress in-memory Stellar state.`,
+      observed: snapshot.ledgerSequence,
+      expected: options.minLedgerSequence
+    });
+  }
+
+  // 1c. Content hash check (only when the snapshot carries one)
+  if (snapshot.contentHash && options.hashFn) {
+    const recomputed = options.hashFn(snapshot.primaryRecords);
+    if (recomputed !== snapshot.contentHash) {
+      violations.push({
+        kind: "content_hash_mismatch",
+        checkName: "snapshot.content_hash",
+        targetId: "snapshot",
+        detail: `Snapshot content hash does not match the recomputed hash of its primary records. The snapshot may have been tampered with or partially written.`,
+        observed: recomputed,
+        expected: snapshot.contentHash
+      });
+    }
+  }
+
+  // ── Phase 2: Per-record invariant checks ──────────────────────────────────
+
+  if (options.invariants && options.invariants.length > 0) {
+    for (const record of snapshot.primaryRecords) {
+      // Derive a string ID from the record for violation attribution.
+      const recordId =
+        typeof (record as Record<string, unknown>)["id"] === "string"
+          ? (record as Record<string, unknown>)["id"] as string
+          : JSON.stringify(record).slice(0, 64);
+
+      for (const invariant of options.invariants) {
+        let passed = false;
+        try {
+          passed = invariant.check(record);
+        } catch (checkError) {
+          // A throwing invariant counts as a failed check — never silenced.
+          passed = false;
+          violations.push({
+            kind: "invariant_failed",
+            checkName: invariant.name,
+            targetId: recordId,
+            detail: `Invariant "${invariant.name}" threw during check: ${checkError instanceof Error ? checkError.message : String(checkError)}`,
+            observed: "exception"
+          });
+          continue;
+        }
+
+        if (!passed) {
+          violations.push({
+            kind: "invariant_failed",
+            checkName: invariant.name,
+            targetId: recordId,
+            detail:
+              invariant.description ??
+              `Record "${recordId}" failed invariant "${invariant.name}".`
+          });
+        }
+      }
+    }
+  }
+
+  // ── Phase 3: Loud failure on hard violations ──────────────────────────────
+
+  if (violations.length > 0) {
+    // Emit telemetry first — so even an uncaught throw leaves a trace.
+    emitTelemetry({
+      op: "snapshot.restore_failed",
+      actorType: "system",
+      result: "failure",
+      correlationId,
+      errorCode: "snapshot_invariant_violation",
+      payload: {
+        snapshotId: snapshot.id,
+        schemaVersion: snapshot.schemaVersion,
+        ledgerSequence: snapshot.ledgerSequence,
+        recordCount: snapshot.primaryRecords.length,
+        violationCount: violations.length,
+        violationKinds: [...new Set(violations.map((v) => v.kind))]
+      }
+    });
+
+    // Audit the failure so the trail shows what broke and on which snapshot.
+    audit({
+      action: "projection.replayed",
+      actor,
+      scope: "maintainer",
+      target: { kind: "snapshot", id: snapshot.id },
+      outcome: "denied",
+      reason: "restore_verification_failed",
+      after: {
+        violationCount: violations.length,
+        schemaVersion: snapshot.schemaVersion,
+        ledgerSequence: snapshot.ledgerSequence
+      },
+      at,
+      correlationId,
+      errorCode: "snapshot_invariant_violation"
+    });
+
+    // Throw — never return an ok() — so a caller that forgets to check the
+    // result still surfaces the failure.
+    throw new SnapshotMismatchError(violations, snapshot.id, correlationId);
+  }
+
+  // ── Phase 4: Derive index and compare ────────────────────────────────────
+
+  const rebuiltIndex: DerivedIndex = options.indexBuilder
+    ? options.indexBuilder(snapshot.primaryRecords)
+    : { ...snapshot.derivedIndex };
+
+  const indexMismatches: SnapshotIndexMismatch[] = [];
+
+  if (options.indexBuilder) {
+    // Keys present in snapshot but missing or changed in rebuild.
+    for (const [key, snapshotValue] of Object.entries(snapshot.derivedIndex)) {
+      const rebuiltValue = rebuiltIndex[key];
+      if (rebuiltValue !== snapshotValue) {
+        indexMismatches.push({
+          key,
+          snapshotValue,
+          rebuiltValue: rebuiltValue ?? undefined
+        });
+      }
+    }
+    // Keys present in rebuild but absent in snapshot.
+    for (const [key, rebuiltValue] of Object.entries(rebuiltIndex)) {
+      if (!(key in snapshot.derivedIndex)) {
+        indexMismatches.push({ key, snapshotValue: undefined, rebuiltValue });
+      }
+    }
+  }
+
+  // ── Phase 5: Commit (if not dry-run) ──────────────────────────────────────
+
+  if (!options.dryRun && options.onRestore) {
+    options.onRestore(snapshot.primaryRecords, rebuiltIndex);
+  }
+
+  // ── Phase 6: Emit success telemetry and audit ─────────────────────────────
+
+  const restoredAt = new Date(nowFn()).toISOString();
+
+  emitTelemetry({
+    op: "snapshot.restore_succeeded",
+    actorType: "system",
+    result: "success",
+    correlationId,
+    payload: {
+      snapshotId: snapshot.id,
+      schemaVersion: snapshot.schemaVersion,
+      ledgerSequence: snapshot.ledgerSequence,
+      recordCount: snapshot.primaryRecords.length,
+      indexMismatchCount: indexMismatches.length,
+      dryRun: options.dryRun ?? false
+    }
+  });
+
+  audit({
+    action: "projection.replayed",
+    actor,
+    scope: "maintainer",
+    target: { kind: "snapshot", id: snapshot.id },
+    outcome: "allowed",
+    reason: options.dryRun ? "restore_dry_run" : "restore_verified",
+    after: {
+      schemaVersion: snapshot.schemaVersion,
+      ledgerSequence: snapshot.ledgerSequence,
+      recordCount: snapshot.primaryRecords.length,
+      indexMismatchCount: indexMismatches.length,
+      dryRun: options.dryRun ?? false
+    },
+    at: restoredAt,
+    correlationId
+  });
+
+  return {
+    snapshotId: snapshot.id,
+    schemaVersion: snapshot.schemaVersion,
+    ledgerSequence: snapshot.ledgerSequence,
+    primaryRecordCount: snapshot.primaryRecords.length,
+    rebuiltIndex,
+    indexMismatches,
+    dryRun: options.dryRun ?? false,
+    restoredAt,
+    correlationId
+  };
+}
+
+// ─── Convenience: dry-run pre-flight ──────────────────────────────────────────
+
+/**
+ * Validates a snapshot without committing. Returns a `Result` instead of
+ * throwing so callers can introspect violations without a try/catch.
+ *
+ * Internally calls `verifyAndRestore` with `dryRun: true` and catches
+ * `SnapshotMismatchError`. All other errors are re-thrown.
+ */
+export function preflight<T>(
+  snapshot: Snapshot<T>,
+  options: Omit<VerifyOptions<T>, "dryRun" | "onRestore"> = {}
+): Result<SnapshotRestoreResult<T>, SnapshotErrorCode, readonly SnapshotViolation[]> {
+  try {
+    const result = verifyAndRestore(snapshot, { ...options, dryRun: true });
+    return ok(result);
+  } catch (error) {
+    if (error instanceof SnapshotMismatchError) {
+      return err("snapshot_invariant_violation", error.violations);
+    }
+    throw error;
+  }
+}
diff --git a/core/workers/__tests__/backpressure.test.ts b/core/workers/__tests__/backpressure.test.ts
new file mode 100644
index 0000000..777af1b
--- /dev/null
+++ b/core/workers/__tests__/backpressure.test.ts
@@ -0,0 +1,523 @@
+/**
+ * Tests for bounded queue backpressure (core/workers/backpressure.ts).
+ *
+ * Coverage:
+ *   - Normal path: admit jobs up to capacity
+ *   - Overflow path: reject when full (reject policy)
+ *   - Overflow path: drop oldest and admit (drop_oldest policy)
+ *   - Retry path: retrying jobs hold a slot; settled jobs release it
+ *   - Authorization path: maintainer may adjust capacity, viewer cannot
+ *   - waitForCapacity: resolves once a slot opens
+ *   - Label scoping: operations outside the prefix are passed through unmetered
+ *   - Hard-cap ceiling: adjustCapacity cannot exceed maxCapacityCeiling
+ *   - Self-healing: externally dead-lettered jobs free their slot
+ *   - Stats accuracy: admitted / rejected / evicted counters track correctly
+ */
+
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createBoundedEnqueueGate, type BoundedEnqueueGate } from "@/core/workers/backpressure";
+import { createWorkerFramework, type WorkerFramework } from "@/core/workers/queue";
+import { setTelemetrySink, createCaptureSink, resetTelemetrySink } from "@/core/telemetry/telemetry";
+import type { AuthContext } from "@/core/auth/middleware";
+
+// ─── Helpers ──────────────────────────────────────────────────────────────────
+
+const VIEWER: AuthContext = { role: "viewer" };
+const MAINTAINER: AuthContext = { role: "maintainer" };
+
+/** A framework with a noop handler registered for `analytics.aggregate`. */
+function makeFramework(): WorkerFramework {
+  const fw = createWorkerFramework();
+  fw.register("analytics.aggregate", vi.fn());
+  fw.register("analytics.export", vi.fn());
+  fw.register("other.operation", vi.fn());
+  return fw;
+}
+
+/** Creates a gate with `label: "analytics"` and `maxConcurrent: 2`. */
+function makeGate(
+  fw: WorkerFramework,
+  overrides: Partial<Parameters<typeof createBoundedEnqueueGate>[1]> = {}
+): BoundedEnqueueGate {
+  return createBoundedEnqueueGate(fw, {
+    label: "analytics",
+    maxConcurrent: 2,
+    ...overrides
+  });
+}
+
+// ─── Normal path ──────────────────────────────────────────────────────────────
+
+describe("normal path — admit jobs up to capacity", () => {
+  let fw: WorkerFramework;
+  let gate: BoundedEnqueueGate;
+
+  beforeEach(() => {
+    fw = makeFramework();
+    gate = makeGate(fw);
+  });
+
+  it("admits a job when the gate has capacity", () => {
+    const result = gate.enqueue({ operation: "analytics.aggregate" });
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    expect(result.value.operation).toBe("analytics.aggregate");
+    expect(result.value.status).toBe("queued");
+  });
+
+  it("admits jobs up to maxConcurrent without error", () => {
+    const r1 = gate.enqueue({ operation: "analytics.aggregate" });
+    const r2 = gate.enqueue({ operation: "analytics.aggregate" });
+    expect(r1.ok).toBe(true);
+    expect(r2.ok).toBe(true);
+  });
+
+  it("reports live count in stats correctly", () => {
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    const s = gate.stats();
+    expect(s.live).toBe(2);
+    expect(s.admitted).toBe(2);
+    expect(s.rejected).toBe(0);
+  });
+
+  it("exposes the configured label", () => {
+    expect(gate.label()).toBe("analytics");
+  });
+
+  it("passes operations not matching the prefix through unmetered", () => {
+    // Fill up the gate with labeled ops first.
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    // An unrelated operation is still enqueued even though the gate is full.
+    const result = gate.enqueue({ operation: "other.operation" });
+    expect(result.ok).toBe(true);
+    // Live count only counts labeled ops.
+    expect(gate.stats().live).toBe(2);
+  });
+});
+
+// ─── Overflow — reject policy ──────────────────────────────────────────────
+
+describe("overflow path — reject policy", () => {
+  let fw: WorkerFramework;
+  let gate: BoundedEnqueueGate;
+
+  beforeEach(() => {
+    fw = makeFramework();
+    gate = makeGate(fw, { overflowPolicy: "reject" });
+  });
+
+  it("returns backpressure_full when capacity is reached", () => {
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    const result = gate.enqueue({ operation: "analytics.aggregate" });
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.code).toBe("backpressure_full");
+  });
+
+  it("increments the rejected counter on overflow", () => {
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" }); // rejected
+    gate.enqueue({ operation: "analytics.aggregate" }); // rejected
+    const s = gate.stats();
+    expect(s.admitted).toBe(2);
+    expect(s.rejected).toBe(2);
+    expect(s.evicted).toBe(0);
+  });
+
+  it("emits backpressure.overflow telemetry on reject", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      gate.enqueue({ operation: "analytics.aggregate" });
+      gate.enqueue({ operation: "analytics.aggregate" });
+      gate.enqueue({ operation: "analytics.aggregate" }); // overflow
+      const overflowEvents = sink.filter((e) => e.op === "backpressure.overflow");
+      expect(overflowEvents).toHaveLength(1);
+      expect(overflowEvents[0].errorCode).toBe("backpressure_full");
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+
+  it("accepts new jobs once a slot opens after drain", () => {
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+
+    // Drain the framework so both jobs succeed.
+    fw.drainDueJobs();
+
+    // Now slots are free.
+    const result = gate.enqueue({ operation: "analytics.aggregate" });
+    expect(result.ok).toBe(true);
+  });
+});
+
+// ─── Overflow — drop_oldest policy ───────────────────────────────────────────
+
+describe("overflow path — drop_oldest policy", () => {
+  let fw: WorkerFramework;
+  let gate: BoundedEnqueueGate;
+
+  beforeEach(() => {
+    fw = makeFramework();
+    gate = makeGate(fw, { overflowPolicy: "drop_oldest" });
+  });
+
+  it("evicts the oldest queued job and admits the new one", () => {
+    const first = gate.enqueue({ operation: "analytics.aggregate", params: { seq: 1 } });
+    gate.enqueue({ operation: "analytics.aggregate", params: { seq: 2 } });
+    expect(first.ok).toBe(true);
+    if (!first.ok) return;
+
+    // Third job causes eviction of the oldest (first).
+    const third = gate.enqueue({ operation: "analytics.aggregate", params: { seq: 3 } });
+    expect(third.ok).toBe(true);
+
+    // The evicted job should be dead-lettered.
+    const deadLettered = fw.inspect("dead_lettered");
+    expect(deadLettered.map((j) => j.id)).toContain(first.value.id);
+  });
+
+  it("increments the evicted counter", () => {
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" }); // evicts first
+    const s = gate.stats();
+    expect(s.evicted).toBe(1);
+    expect(s.admitted).toBe(3);
+    expect(s.rejected).toBe(0);
+  });
+
+  it("falls back to reject when all live jobs are running (no evictable slot)", async () => {
+    // Enqueue two jobs and start them by draining asynchronously (without settling).
+    const fw2 = createWorkerFramework();
+    let resolve1!: () => void;
+    let resolve2!: () => void;
+    fw2.register("analytics.aggregate", (_params, ctx) => {
+      return new Promise<void>((res) => {
+        // Hang until the test resolves it.
+        const handle = setInterval(() => {
+          if (ctx.signal.aborted) { clearInterval(handle); res(); }
+        }, 10);
+        resolve1 = resolve2 = () => { clearInterval(handle); res(); };
+      });
+    });
+
+    const gate2 = createBoundedEnqueueGate(fw2, {
+      label: "analytics",
+      maxConcurrent: 2,
+      overflowPolicy: "drop_oldest"
+    });
+
+    gate2.enqueue({ operation: "analytics.aggregate" });
+    gate2.enqueue({ operation: "analytics.aggregate" });
+
+    // Start draining — jobs are now running (async, in-flight).
+    const drainPromise = fw2.drain();
+
+    // Give the event loop a tick so the handlers actually start.
+    await Promise.resolve();
+
+    // Now all live jobs are running — no evictable candidates exist.
+    const result = gate2.enqueue({ operation: "analytics.aggregate" });
+    expect(result.ok).toBe(false);
+    if (!result.ok) expect(result.code).toBe("backpressure_full");
+
+    // Clean up.
+    resolve1();
+    resolve2();
+    await drainPromise;
+    fw2.reset();
+  });
+
+  it("emits backpressure.overflow with drop_oldest payload on eviction", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      gate.enqueue({ operation: "analytics.aggregate" });
+      gate.enqueue({ operation: "analytics.aggregate" });
+      gate.enqueue({ operation: "analytics.aggregate" }); // triggers eviction
+      const overflow = sink.find((e) => e.op === "backpressure.overflow");
+      expect(overflow).toBeDefined();
+      expect(overflow?.payload?.policy).toBe("drop_oldest");
+      expect(overflow?.payload?.evictedJobId).toBeDefined();
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+});
+
+// ─── Retry path ───────────────────────────────────────────────────────────────
+
+describe("retry path — retrying jobs occupy a slot until settled", () => {
+  let fw: WorkerFramework;
+  let gate: BoundedEnqueueGate;
+
+  beforeEach(() => {
+    fw = makeFramework();
+    // Register a handler that fails so jobs enter retrying state.
+    fw.register("analytics.retry_test", () => {
+      throw new Error("transient failure");
+    });
+    gate = makeGate(fw, { maxConcurrent: 1 });
+  });
+
+  it("a retrying job still occupies its slot", () => {
+    const job = gate.enqueue({ operation: "analytics.retry_test", maxAttempts: 3 });
+    expect(job.ok).toBe(true);
+
+    // Drain — the job fails and enters retrying.
+    fw.drainDueJobs();
+    if (job.ok) expect(job.value.status).toBe("retrying");
+
+    // The gate's slot is still occupied.
+    const second = gate.enqueue({ operation: "analytics.retry_test" });
+    expect(second.ok).toBe(false);
+    if (!second.ok) expect(second.code).toBe("backpressure_full");
+  });
+
+  it("a dead-lettered job frees its slot", () => {
+    const job = gate.enqueue({ operation: "analytics.retry_test", maxAttempts: 1 });
+    expect(job.ok).toBe(true);
+
+    // Drain — exhausts the budget and moves to dead_lettered.
+    fw.drainDueJobs();
+    if (job.ok) expect(job.value.status).toBe("dead_lettered");
+
+    // Slot is now free.
+    const next = gate.enqueue({ operation: "analytics.retry_test" });
+    expect(next.ok).toBe(true);
+  });
+
+  it("a succeeded job frees its slot", () => {
+    // Re-register with a succeeding handler.
+    fw.reset();
+    fw.register("analytics.aggregate", vi.fn());
+    const g = makeGate(fw, { maxConcurrent: 1 });
+
+    g.enqueue({ operation: "analytics.aggregate" });
+    fw.drainDueJobs(); // succeeds
+
+    // Slot should be free.
+    const r = g.enqueue({ operation: "analytics.aggregate" });
+    expect(r.ok).toBe(true);
+  });
+});
+
+// ─── Authorization path ───────────────────────────────────────────────────────
+
+describe("authorization path — adjustCapacity", () => {
+  let fw: WorkerFramework;
+  let gate: BoundedEnqueueGate;
+
+  beforeEach(() => {
+    fw = makeFramework();
+    gate = makeGate(fw, { maxConcurrent: 2, maxCapacityCeiling: 10 });
+  });
+
+  it("maintainer can raise the capacity ceiling", () => {
+    const result = gate.adjustCapacity(MAINTAINER, 5, "peak_load");
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    expect(result.value.previous).toBe(2);
+    expect(result.value.next).toBe(5);
+    expect(gate.stats().maxConcurrent).toBe(5);
+  });
+
+  it("maintainer can lower the capacity ceiling", () => {
+    const result = gate.adjustCapacity(MAINTAINER, 1, "throttle");
+    expect(result.ok).toBe(true);
+    if (!result.ok) return;
+    expect(result.value.next).toBe(1);
+  });
+
+  it("viewer is refused with backpressure_unauthorized", () => {
+    const result = gate.adjustCapacity(VIEWER, 5, "attempt");
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.code).toBe("backpressure_unauthorized");
+  });
+
+  it("capacity cannot exceed maxCapacityCeiling", () => {
+    const result = gate.adjustCapacity(MAINTAINER, 11, "too_high");
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.code).toBe("backpressure_invalid_capacity");
+  });
+
+  it("capacity cannot be set below 1", () => {
+    const result = gate.adjustCapacity(MAINTAINER, 0, "zero");
+    expect(result.ok).toBe(false);
+    if (result.ok) return;
+    expect(result.code).toBe("backpressure_invalid_capacity");
+  });
+
+  it("raising the ceiling admits previously-blocked jobs", () => {
+    // Fill the gate.
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    expect(gate.enqueue({ operation: "analytics.aggregate" }).ok).toBe(false);
+
+    // Raise the ceiling.
+    gate.adjustCapacity(MAINTAINER, 3, "raised");
+
+    // Now a third job fits.
+    expect(gate.enqueue({ operation: "analytics.aggregate" }).ok).toBe(true);
+  });
+
+  it("emits backpressure.capacity_changed telemetry on success", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      gate.adjustCapacity(MAINTAINER, 4, "load_event");
+      const event = sink.find((e) => e.op === "backpressure.capacity_changed");
+      expect(event).toBeDefined();
+      expect(event?.result).toBe("success");
+      expect(event?.payload?.previous).toBe(2);
+      expect(event?.payload?.next).toBe(4);
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+
+  it("emits failure telemetry when viewer tries to adjust capacity", () => {
+    const sink = createCaptureSink();
+    setTelemetrySink(sink);
+    try {
+      gate.adjustCapacity(VIEWER, 5, "attempt");
+      const event = sink.find((e) => e.op === "backpressure.capacity_changed");
+      expect(event?.result).toBe("failure");
+      expect(event?.errorCode).toBe("backpressure_unauthorized");
+    } finally {
+      resetTelemetrySink();
+    }
+  });
+});
+
+// ─── waitForCapacity ──────────────────────────────────────────────────────────
+
+describe("waitForCapacity", () => {
+  it("resolves immediately when the gate has capacity", async () => {
+    const fw = makeFramework();
+    const gate = makeGate(fw);
+    // Gate is empty — capacity is available.
+    await expect(gate.waitForCapacity()).resolves.toBeUndefined();
+  });
+
+  it("resolves after the capacity is raised", async () => {
+    const fw = makeFramework();
+    const gate = makeGate(fw, { maxConcurrent: 1, maxCapacityCeiling: 5 });
+    gate.enqueue({ operation: "analytics.aggregate" });
+
+    // Gate is full — grab the promise before raising capacity.
+    const waiting = gate.waitForCapacity();
+
+    // Raise capacity — this should notify waiters.
+    gate.adjustCapacity(MAINTAINER, 2, "raise");
+
+    await expect(waiting).resolves.toBeUndefined();
+  });
+
+  it("resolves immediately after drain frees slots", async () => {
+    const fw = createWorkerFramework();
+    fw.register("analytics.aggregate", vi.fn());
+    const gate = createBoundedEnqueueGate(fw, {
+      label: "analytics",
+      maxConcurrent: 1
+    });
+    gate.enqueue({ operation: "analytics.aggregate" });
+
+    // Drain settles the job, freeing the slot.
+    fw.drainDueJobs();
+
+    // Gate should now have capacity.
+    await expect(gate.waitForCapacity()).resolves.toBeUndefined();
+  });
+});
+
+// ─── Self-healing ─────────────────────────────────────────────────────────────
+
+describe("self-healing — externally dead-lettered jobs free their slot", () => {
+  it("detects when a job is dead-lettered outside the gate", () => {
+    const fw = makeFramework();
+    const gate = makeGate(fw, { maxConcurrent: 1 });
+
+    const r = gate.enqueue({ operation: "analytics.aggregate" });
+    expect(r.ok).toBe(true);
+    if (!r.ok) return;
+
+    // Gate is full.
+    expect(gate.enqueue({ operation: "analytics.aggregate" }).ok).toBe(false);
+
+    // Dead-letter the job directly through the framework (external to the gate).
+    fw.deadLetter(r.value.id);
+
+    // Gate recomputes live count — slot is now free.
+    expect(gate.stats().live).toBe(0);
+    expect(gate.enqueue({ operation: "analytics.aggregate" }).ok).toBe(true);
+  });
+});
+
+// ─── Constructor validation ────────────────────────────────────────────────────
+
+describe("constructor validation", () => {
+  it("throws when label is empty", () => {
+    const fw = makeFramework();
+    expect(() =>
+      createBoundedEnqueueGate(fw, { label: "", maxConcurrent: 1 })
+    ).toThrow("[backpressure] label is required");
+  });
+
+  it("throws when maxConcurrent is not a positive integer", () => {
+    const fw = makeFramework();
+    expect(() =>
+      createBoundedEnqueueGate(fw, { label: "analytics", maxConcurrent: 0 })
+    ).toThrow("[backpressure] maxConcurrent must be a positive integer");
+  });
+
+  it("throws when maxConcurrent is a float", () => {
+    const fw = makeFramework();
+    expect(() =>
+      createBoundedEnqueueGate(fw, { label: "analytics", maxConcurrent: 1.5 })
+    ).toThrow("[backpressure] maxConcurrent must be a positive integer");
+  });
+});
+
+// ─── Stats and reset ──────────────────────────────────────────────────────────
+
+describe("stats and reset", () => {
+  it("reset clears counters without affecting the underlying framework", () => {
+    const fw = makeFramework();
+    const gate = makeGate(fw);
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" }); // rejected
+
+    expect(gate.stats().admitted).toBe(2);
+    expect(gate.stats().rejected).toBe(1);
+
+    gate.reset();
+
+    expect(gate.stats().admitted).toBe(0);
+    expect(gate.stats().rejected).toBe(0);
+    // Framework still has the jobs.
+    expect(fw.inspect("queued")).toHaveLength(2);
+  });
+
+  it("stats.live decreases as jobs settle", () => {
+    const fw = createWorkerFramework();
+    fw.register("analytics.aggregate", vi.fn());
+    const gate = createBoundedEnqueueGate(fw, { label: "analytics", maxConcurrent: 3 });
+
+    gate.enqueue({ operation: "analytics.aggregate" });
+    gate.enqueue({ operation: "analytics.aggregate" });
+    expect(gate.stats().live).toBe(2);
+
+    fw.drainDueJobs();
+    expect(gate.stats().live).toBe(0);
+  });
+});
diff --git a/core/workers/backpressure.ts b/core/workers/backpressure.ts
new file mode 100644
index 0000000..1a0b14b
--- /dev/null
+++ b/core/workers/backpressure.ts
@@ -0,0 +1,464 @@
+/**
+ * Bounded queue backpressure for expensive background work (#backpressure).
+ *
+ * The worker framework in `queue.ts` is unbounded by design: its purpose is
+ * reliable delivery, not traffic shaping. That works well when callers are
+ * well-behaved — a single slow tick, a UI button, a scheduled maintenance job.
+ * It is the wrong tool when expensive work arrives in bursts: analytics
+ * aggregations, Soroban simulations, index rebuilds, export generations. A
+ * loop, a bug or a stuck retry can fill the queue with thousands of jobs that
+ * will never all finish before the page unloads, consuming memory and burning
+ * quota without delivering value.
+ *
+ * This module adds a backpressure gate in front of `WorkerFramework.enqueue`.
+ * Its rules:
+ *
+ * - **capacity is hard**. The gate counts how many jobs matching a label are
+ *   currently live (queued + retrying + running). When that count reaches
+ *   `maxConcurrent` a new enqueue is refused with `backpressure_full`.
+ * - **overflow policy is explicit**. Two policies ship: `reject` (the
+ *   default, refuse callers, let them handle it) and `drop_oldest` (evict the
+ *   oldest queued job in the slot to make room, then enqueue — useful for
+ *   "latest state wins" work like analytics snapshots where an older job is
+ *   superseded by a newer one).
+ * - **drain signalling**. The gate exposes a `waitForCapacity()` promise that
+ *   resolves when at least one slot opens. This lets a producer await pressure
+ *   without polling, which makes the idle path free.
+ * - **authorization is required to change limits**. A `viewer` cannot raise
+ *   the ceiling; a `maintainer` can, but only within the declared hard cap.
+ * - **every decision is emitted as telemetry** (`backpressure.enqueue`,
+ *   `backpressure.overflow`, `backpressure.capacity_changed`) so operators can
+ *   see pressure over time without polling the gate.
+ *
+ * Operational edge cases and recovery
+ * ────────────────────────────────────
+ * 1. **Sudden burst > maxConcurrent**: The gate immediately starts rejecting
+ *    (reject policy) or dropping the oldest queued job (drop_oldest). Callers
+ *    must handle `backpressure_full` and either wait on `waitForCapacity()` or
+ *    surface the failure to the user. Recovery: raise `maxConcurrent` via
+ *    `adjustCapacity()` (maintainer only) or let the drain catch up.
+ *
+ * 2. **Stuck/long-running job holding a slot**: A job in `running` state
+ *    forever prevents new work. The framework's `shutdown()` will interrupt
+ *    it; the gate's slot count is recomputed from the live `inspect()` call on
+ *    every check, so it heals automatically once the job leaves `running`.
+ *    Recovery: call `framework.deadLetter(id)` to force a dead-letter, which
+ *    frees the slot immediately.
+ *
+ * 3. **drop_oldest evicts a job that was in `retrying`**: The evicted job is
+ *    dead-lettered (not cancelled) so it appears in the framework's dead-letter
+ *    queue and is retrievable by ID. The audit trail records the forced
+ *    dead-letter. Recovery: inspect dead-lettered jobs with
+ *    `framework.inspect("dead_lettered")` and requeue any that matter.
+ *
+ * 4. **Authorization bypass attempt**: Only `maintainer` actors may call
+ *    `adjustCapacity`. A `viewer` calling it receives `backpressure_unauthorized`.
+ *    The hard cap `maxCapacityCeiling` is enforced independently of auth, so a
+ *    compromised maintainer token cannot raise the gate above what the code
+ *    allows.
+ *
+ * 5. **Rollback / gate removal**: `BoundedEnqueueGate` wraps the framework but
+ *    never patches it. Removing the gate reverts to the unbounded path without
+ *    any migration: callers that used `gate.enqueue()` switch back to
+ *    `framework.enqueue()` with no schema or state change required.
+ *
+ * 6. **waitForCapacity() during shutdown**: If the framework is cancelled
+ *    or stopped, in-flight jobs settle quickly and slots open. Callers awaiting
+ *    capacity will resolve on the next settled job. A caller that holds the
+ *    promise through a page unload will never race with the abort: the gate
+ *    resolves the waiter before the framework drops the jobs.
+ */
+
+import { hasPermission } from "@/core/auth/types";
+import type { AuthContext } from "@/core/auth/middleware";
+import { err, ok, type Result } from "@/core/result/result";
+import { emitTelemetry, newCorrelationId } from "@/core/telemetry/telemetry";
+import type { JobInput, JobPayload, WorkerFramework } from "@/core/workers/queue";
+
+// ─── Public types ─────────────────────────────────────────────────────────────
+
+/** Stable codes returned by the gate. Consumers switch on these. */
+export type BackpressureErrorCode =
+  | "backpressure_full"
+  | "backpressure_unauthorized"
+  | "backpressure_invalid_capacity"
+  | "backpressure_label_required";
+
+/**
+ * What the gate does when `maxConcurrent` is reached.
+ *
+ * - `reject`     — refuse the caller immediately; they decide what to do next.
+ * - `drop_oldest` — dead-letter the oldest non-running job in the slot to make
+ *                   room, then enqueue the new one. Use for "latest wins" work
+ *                   (analytics, snapshot probes) where an older job is
+ *                   superseded by a newer one.
+ */
+export type OverflowPolicy = "reject" | "drop_oldest";
+
+export interface BackpressureOptions {
+  /**
+   * Maximum number of live jobs (queued + retrying + running) for this gate.
+   * Must be ≥ 1.
+   */
+  maxConcurrent: number;
+  /**
+   * Hard ceiling that `adjustCapacity()` may never exceed.
+   * Defaults to `maxConcurrent * 10`, minimum 1.
+   */
+  maxCapacityCeiling?: number;
+  /** What to do when the gate is full. Defaults to `"reject"`. */
+  overflowPolicy?: OverflowPolicy;
+  /**
+   * Operation label this gate meters. Only jobs whose `operation` starts with
+   * this prefix are counted. Required: a gate without a label would count all
+   * operations, which would be surprising and break composition.
+   *
+   * Example: `"analytics"` counts `"analytics.aggregate"`, `"analytics.export"`.
+   */
+  label: string;
+  /** Injected clock. Defaults to `Date.now`. */
+  now?: () => number;
+}
+
+export interface GateStats {
+  /** How many jobs are currently live (queued + retrying + running). */
+  readonly live: number;
+  /** The current capacity ceiling. */
+  readonly maxConcurrent: number;
+  /** How many enqueue attempts have been admitted since last reset. */
+  readonly admitted: number;
+  /** How many enqueue attempts have been rejected due to full capacity. */
+  readonly rejected: number;
+  /** How many jobs have been evicted by the `drop_oldest` policy. */
+  readonly evicted: number;
+}
+
+export interface CapacityAdjustment {
+  readonly previous: number;
+  readonly next: number;
+  readonly reason: string;
+  readonly at: string;
+}
+
+export interface BoundedEnqueueGate {
+  /**
+   * Attempt to enqueue a job through the gate.
+   *
+   * - If the gate has capacity, the job is enqueued and `ok` is returned.
+   * - If the gate is full and policy is `reject`, `backpressure_full` is returned.
+   * - If the gate is full and policy is `drop_oldest`, the oldest queued/retrying
+   *   job is dead-lettered and the new job is enqueued.
+   *
+   * The returned `JobPayload` is the newly enqueued job (not the evicted one).
+   */
+  enqueue(input: JobInput): Result<JobPayload, BackpressureErrorCode>;
+
+  /**
+   * Resolves the next time at least one slot opens. Use this to implement
+   * back-off without polling:
+   *
+   * ```ts
+   * while (true) {
+   *   const r = gate.enqueue(input);
+   *   if (r.ok) break;
+   *   await gate.waitForCapacity();
+   * }
+   * ```
+   *
+   * The promise never rejects. If the framework has already settled all jobs
+   * (e.g. on shutdown), it resolves immediately.
+   */
+  waitForCapacity(): Promise<void>;
+
+  /**
+   * Change the concurrency ceiling. Requires `admin:recovery` permission.
+   *
+   * `next` must be ≥ 1 and ≤ `maxCapacityCeiling`. Lowering below the current
+   * live count is allowed — the gate will refuse new work until the backlog
+   * drains to the new limit. No jobs are forcibly killed.
+   */
+  adjustCapacity(
+    auth: AuthContext,
+    next: number,
+    reason: string
+  ): Result<CapacityAdjustment, BackpressureErrorCode>;
+
+  /** Current gate statistics. */
+  stats(): GateStats;
+
+  /** The label this gate is scoped to. */
+  label(): string;
+
+  /** Test seam: resets counters (does not affect the underlying framework). */
+  reset(): void;
+}
+
+// ─── Live-count helper ────────────────────────────────────────────────────────
+
+const LIVE_STATUSES = new Set(["queued", "retrying", "running"]);
+
+/**
+ * Counts jobs in the framework whose operation starts with `prefix` and are in
+ * a live (non-terminal) state.
+ *
+ * We count from the framework's live state rather than maintaining a shadow
+ * counter, so the gate self-heals after a job is externally dead-lettered,
+ * cancelled or manually requeued — no synchronisation needed.
+ */
+function countLive(framework: WorkerFramework, prefix: string): number {
+  return framework.inspect().filter(
+    (job) => job.operation.startsWith(prefix) && LIVE_STATUSES.has(job.status)
+  ).length;
+}
+
+/**
+ * Returns the oldest queued or retrying job for the given label prefix,
+ * ordered by `enqueuedAt` ascending. `running` jobs are excluded: evicting a
+ * running job would cancel in-flight work, which requires framework shutdown.
+ */
+function oldestEvictable(framework: WorkerFramework, prefix: string): JobPayload | undefined {
+  const candidates = framework
+    .inspect()
+    .filter(
+      (job) =>
+        job.operation.startsWith(prefix) &&
+        (job.status === "queued" || job.status === "retrying")
+    )
+    .sort((a, b) => a.enqueuedAt.localeCompare(b.enqueuedAt));
+  return candidates[0];
+}
+
+// ─── Factory ──────────────────────────────────────────────────────────────────
+
+export function createBoundedEnqueueGate(
+  framework: WorkerFramework,
+  options: BackpressureOptions
+): BoundedEnqueueGate {
+  if (!options.label || options.label.trim().length === 0) {
+    throw new Error("[backpressure] label is required");
+  }
+  if (!Number.isInteger(options.maxConcurrent) || options.maxConcurrent < 1) {
+    throw new Error("[backpressure] maxConcurrent must be a positive integer");
+  }
+
+  let maxConcurrent = options.maxConcurrent;
+  const maxCapacityCeiling = Math.max(
+    1,
+    options.maxCapacityCeiling ?? options.maxConcurrent * 10
+  );
+  const overflowPolicy: OverflowPolicy = options.overflowPolicy ?? "reject";
+  const prefix = options.label;
+  const nowFn = options.now ?? (() => Date.now());
+
+  let admitted = 0;
+  let rejected = 0;
+  let evicted = 0;
+
+  // Capacity waiters: each entry is a resolve function from a Promise.
+  // When a slot opens (detected after a drain completes) we fire all of them.
+  const waiters: Array<() => void> = [];
+
+  function notifyWaiters(): void {
+    const snapshot = waiters.splice(0);
+    for (const resolve of snapshot) resolve();
+  }
+
+  /**
+   * Registers a slot-open listener and returns its promise.
+   * The promise is resolved the next time any job in the gate's prefix
+   * transitions out of a live state (i.e. a drain settles a job).
+   *
+   * Implementation note: we cannot observe individual job transitions without
+   * patching the framework, so instead we wrap the framework's `drain()` and
+   * `drainDueJobs()` at gate creation. This is intentional: the gate is a
+   * facade, not a monkey-patch.
+   */
+  function waitForCapacity(): Promise<void> {
+    // If there is already capacity, resolve immediately.
+    if (countLive(framework, prefix) < maxConcurrent) {
+      return Promise.resolve();
+    }
+    return new Promise<void>((resolve) => {
+      waiters.push(resolve);
+    });
+  }
+
+  function enqueue(input: JobInput): Result<JobPayload, BackpressureErrorCode> {
+    const correlationId = input.correlationId ?? newCorrelationId();
+
+    // The label guard is enforced here so that jobs enqueued without the gate
+    // (directly through the framework) do not pollute the gate's counter.
+    if (!input.operation.startsWith(prefix)) {
+      // Allow operations that don't match the gate's label through unconditionally.
+      // The gate only meters operations it was configured to watch.
+      const job = framework.enqueue({ ...input, correlationId });
+      admitted += 1;
+      emitTelemetry({
+        op: "backpressure.enqueue",
+        actorType: "worker",
+        result: "success",
+        correlationId,
+        payload: { label: prefix, operation: input.operation, gated: false }
+      });
+      return ok(job);
+    }
+
+    const live = countLive(framework, prefix);
+
+    if (live < maxConcurrent) {
+      // Capacity available — admit immediately.
+      const job = framework.enqueue({ ...input, correlationId });
+      admitted += 1;
+      emitTelemetry({
+        op: "backpressure.enqueue",
+        actorType: "worker",
+        result: "success",
+        correlationId,
+        payload: {
+          label: prefix,
+          operation: input.operation,
+          live,
+          maxConcurrent,
+          gated: true
+        }
+      });
+      return ok(job);
+    }
+
+    // Gate is full.
+
+    if (overflowPolicy === "drop_oldest") {
+      const victim = oldestEvictable(framework, prefix);
+      if (victim) {
+        // Force the oldest evictable job to dead-letter so its slot frees up.
+        framework.deadLetter(victim.id);
+        evicted += 1;
+        emitTelemetry({
+          op: "backpressure.overflow",
+          actorType: "worker",
+          result: "failure",
+          correlationId,
+          errorCode: "backpressure_full",
+          payload: {
+            label: prefix,
+            policy: "drop_oldest",
+            evictedJobId: victim.id,
+            evictedOperation: victim.operation,
+            evictedEnqueuedAt: victim.enqueuedAt
+          }
+        });
+        // Now there is a free slot; enqueue the new job.
+        const job = framework.enqueue({ ...input, correlationId });
+        admitted += 1;
+        emitTelemetry({
+          op: "backpressure.enqueue",
+          actorType: "worker",
+          result: "success",
+          correlationId,
+          payload: {
+            label: prefix,
+            operation: input.operation,
+            live: live - 1,
+            maxConcurrent,
+            gated: true,
+            droppedOldest: true
+          }
+        });
+        return ok(job);
+      }
+      // All live jobs are running (none evictable) — fall through to reject.
+    }
+
+    // Reject path.
+    rejected += 1;
+    emitTelemetry({
+      op: "backpressure.overflow",
+      actorType: "worker",
+      result: "failure",
+      correlationId,
+      errorCode: "backpressure_full",
+      payload: {
+        label: prefix,
+        policy: overflowPolicy,
+        live,
+        maxConcurrent,
+        operation: input.operation
+      }
+    });
+    return err("backpressure_full");
+  }
+
+  function adjustCapacity(
+    auth: AuthContext,
+    next: number,
+    reason: string
+  ): Result<CapacityAdjustment, BackpressureErrorCode> {
+    const correlationId = newCorrelationId();
+
+    if (!hasPermission(auth.role, "admin:recovery")) {
+      emitTelemetry({
+        op: "backpressure.capacity_changed",
+        actorType: "user",
+        result: "failure",
+        correlationId,
+        errorCode: "backpressure_unauthorized",
+        payload: { label: prefix, role: auth.role }
+      });
+      return err("backpressure_unauthorized");
+    }
+
+    if (!Number.isInteger(next) || next < 1) {
+      return err("backpressure_invalid_capacity");
+    }
+    if (next > maxCapacityCeiling) {
+      return err("backpressure_invalid_capacity");
+    }
+
+    const previous = maxConcurrent;
+    maxConcurrent = next;
+
+    const adjustment: CapacityAdjustment = {
+      previous,
+      next,
+      reason,
+      at: new Date(nowFn()).toISOString()
+    };
+
+    emitTelemetry({
+      op: "backpressure.capacity_changed",
+      actorType: "user",
+      result: "success",
+      correlationId,
+      payload: { label: prefix, previous, next, reason }
+    });
+
+    // If we raised the ceiling, notify any waiting producers.
+    if (next > previous) notifyWaiters();
+
+    return ok(adjustment);
+  }
+
+  function stats(): GateStats {
+    return {
+      live: countLive(framework, prefix),
+      maxConcurrent,
+      admitted,
+      rejected,
+      evicted
+    };
+  }
+
+  return {
+    enqueue,
+    waitForCapacity,
+    adjustCapacity,
+    stats,
+    label: () => prefix,
+    reset(): void {
+      admitted = 0;
+      rejected = 0;
+      evicted = 0;
+      waiters.splice(0);
+    }
+  };
+}
diff --git a/core/workers/queue.ts b/core/workers/queue.ts
index 77cb776..6c2a324 100644
--- a/core/workers/queue.ts
+++ b/core/workers/queue.ts
@@ -219,8 +219,12 @@ export interface WorkerFramework {
   /**
    * The high-risk write path: enqueues under an idempotency key so a retried
    * submission replays the original job instead of creating a duplicate.
+   *
+   * When the framework was created with `gates`, the matching gate is checked
+   * before idempotency and quota. A full gate returns `backpressure_full`
+   * immediately without spending quota or recording an idempotency claim.
    */
-  submit(input: JobInput): Result<SubmitResult, IdempotencyErrorCode | QuotaErrorCode>;
+  submit(input: JobInput): Result<SubmitResult, IdempotencyErrorCode | QuotaErrorCode | "backpressure_full">;
   /** The quota store `submit()` charges. */
   quota(): QuotaStore;
   /** The idempotency records this framework has persisted. */
@@ -301,6 +305,36 @@ function findingDedupe(jobs: Map<string, JobPayload>, key: string): JobPayload |
 export interface WorkerFrameworkOptions {
   /** Quota store charged by `submit()`. Defaults to an isolated in-memory one. */
   quota?: QuotaStore;
+  /**
+   * Bounded-queue backpressure gates for expensive background operations.
+   *
+   * When an `enqueue` (or `submit`) call arrives, each gate is checked in order
+   * against the job's `operation`. The first gate whose label is a prefix of the
+   * operation handles the call; if the gate is full it returns
+   * `backpressure_full` and the job is never enqueued.
+   *
+   * Gates are optional and purely additive: a framework with no gates behaves
+   * exactly as before. Existing callers that never set this option are unaffected.
+   *
+   * Usage:
+   * ```ts
+   * const analyticsGate = createBoundedEnqueueGate(framework, {
+   *   label: "analytics",
+   *   maxConcurrent: 4,
+   *   overflowPolicy: "drop_oldest"
+   * });
+   * const framework = createWorkerFramework({}, { gates: [analyticsGate] });
+   * ```
+   *
+   * Operational notes:
+   * - Gates are evaluated before the idempotency and quota checks in `submit()`,
+   *   so a rejected job never spends quota.
+   * - A gate refusal returns `backpressure_full` from both `enqueue` and
+   *   `submit`, distinguishable from a quota or idempotency failure.
+   * - Removing a gate from the options array on the next framework instance
+   *   restores unbounded behaviour with no state migration required.
+   */
+  gates?: import("@/core/workers/backpressure").BoundedEnqueueGate[];
 }
 
 export function createWorkerFramework(
@@ -323,6 +357,20 @@ export function createWorkerFramework(
   const idempotencyStore = createIdempotencyStore({ ttlMs: mergedPolicy.idempotencyTtlMs });
   const quotaStore = options.quota ?? createQuotaStore({ storage: createMemoryQuotaStorage() });
 
+  /**
+   * Returns the first configured gate whose label prefix matches `operation`,
+   * or `undefined` when no gate covers the operation.
+   *
+   * Gate lookup is O(g) where g is the number of gates (typically 1–3).
+   * It is intentionally a separate helper rather than inlined in `enqueue` so
+   * the gate path stays readable and testable without reading the full enqueue
+   * body.
+   */
+  function findGate(operation: string) {
+    if (!options.gates || options.gates.length === 0) return undefined;
+    return options.gates.find((gate) => operation.startsWith(gate.label()));
+  }
+
   /**
    * The only place a job's status changes. Anything the `worker_job` table does
    * not allow is refused, so the queue cannot reach a state no consumer expects.
@@ -565,7 +613,30 @@ export function createWorkerFramework(
       return job;
     },
 
-    submit(input: JobInput): Result<SubmitResult, IdempotencyErrorCode | QuotaErrorCode> {
+    submit(input: JobInput): Result<SubmitResult, IdempotencyErrorCode | QuotaErrorCode | "backpressure_full"> {
+      // ── Backpressure gate check ────────────────────────────────────────────
+      // When the framework has a gate covering this operation, we check capacity
+      // before creating an idempotency claim or spending quota.
+      //
+      // Design note: we call `gate.stats()` + live count rather than
+      // `gate.enqueue()` here. The gate was designed as an external facade that
+      // calls `framework.enqueue()` itself; calling it from inside `submit()`
+      // would create a double-enqueue. Instead, `submit` reads the gate's
+      // current live count and refuses early when the gate is full. The gate's
+      // own `enqueue()` (used by callers who bypass `submit`) is the other path
+      // and handles `drop_oldest` eviction. `submit` always uses reject semantics
+      // because idempotency keys require deterministic outcomes.
+      //
+      // Replays are exempt: if we already have a completed record for this key
+      // the job was already enqueued; it either succeeded or is still live, and
+      // the gate's live count already includes it.
+      const gate = findGate(input.operation);
+      if (gate) {
+        const { live, maxConcurrent } = gate.stats();
+        if (live >= maxConcurrent) {
+          return err("backpressure_full");
+        }
+      }
       const begun = idempotencyStore.begin({
         key: input.idempotencyKey,
         operation: "worker.enqueue",
@@ -832,6 +903,13 @@ export function createWorkerFramework(
       idempotencyStore.reset();
       // A shared quota store belongs to the caller; only an owned one is cleared.
       if (!options.quota) quotaStore.reset();
+      // Reset gate counters so a test that calls reset() starts with a clean
+      // gate state. Gates are owned by the caller when passed via options.gates,
+      // but their counters are scoped to this framework's job map, so resetting
+      // the map without resetting the counters would leave stale stats.
+      if (options.gates) {
+        for (const gate of options.gates) gate.reset();
+      }
     }
   };
 
diff --git a/docs/BACKPRESSURE.md b/docs/BACKPRESSURE.md
new file mode 100644
index 0000000..66145cb
--- /dev/null
+++ b/docs/BACKPRESSURE.md
@@ -0,0 +1,165 @@
+# Bounded Queue Backpressure
+
+Expensive background operations — analytics aggregations, Soroban simulations,
+index rebuilds, export generations — must not be enqueued without limit. A
+runaway retry loop, a stuck handler or a burst of UI interactions can exhaust
+memory and quota for everyone sharing the queue.
+
+`core/workers/backpressure.ts` adds a **gate** in front of `WorkerFramework`
+that enforces a hard concurrency ceiling per operation label.
+
+## Design principles
+
+- **Capacity is hard.** A gate counts live jobs (queued + retrying + running)
+  for its prefix. Once that count reaches `maxConcurrent`, new enqueues are
+  refused. The ceiling is an integer, not a guideline.
+- **Overflow policy is explicit.** `reject` (default) refuses the caller
+  immediately and returns `backpressure_full`. `drop_oldest` evicts the oldest
+  queued or retrying job in the slot (dead-letters it, never cancels in-flight
+  work) and admits the new one — useful for "latest state wins" work like
+  analytics snapshots.
+- **Self-healing.** The live count is recomputed from the framework's job map on
+  every check, so a job that is externally dead-lettered, cancelled or retried
+  heals the gate without any sync step.
+- **Authorization is required to change limits.** Only a `maintainer` actor
+  (`admin:recovery` permission) may call `adjustCapacity`. A `viewer` receives
+  `backpressure_unauthorized`.
+- **Every decision is telemetered.** `backpressure.enqueue`,
+  `backpressure.overflow` and `backpressure.capacity_changed` appear in the
+  structured log.
+
+## Quick start
+
+```ts
+import { createWorkerFramework } from "@/core/workers/queue";
+import { createBoundedEnqueueGate } from "@/core/workers/backpressure";
+
+const framework = createWorkerFramework();
+
+// Create the gate first (it needs the framework reference for live-count reads).
+const analyticsGate = createBoundedEnqueueGate(framework, {
+  label: "analytics",       // meters any operation starting with "analytics"
+  maxConcurrent: 4,
+  overflowPolicy: "drop_oldest"
+});
+
+framework.register("analytics.aggregate", async (params) => { /* … */ });
+
+// Enqueue through the gate (the gate calls framework.enqueue internally).
+const result = analyticsGate.enqueue({ operation: "analytics.aggregate", params });
+if (!result.ok) {
+  // result.code === "backpressure_full"
+  await analyticsGate.waitForCapacity();
+  // retry…
+}
+```
+
+### Using `gates` in `createWorkerFramework`
+
+Alternatively, pass gates directly to the framework factory. `submit()` will then
+check gate capacity before spending idempotency/quota:
+
+```ts
+const framework = createWorkerFramework({}, { gates: [analyticsGate] });
+
+const result = framework.submit({
+  operation: "analytics.aggregate",
+  idempotencyKey: requestId,
+  params
+});
+
+if (!result.ok && result.code === "backpressure_full") {
+  // Quota was NOT spent. Retry after waitForCapacity().
+  await analyticsGate.waitForCapacity();
+}
+```
+
+Key difference: `submit()` always uses **reject** semantics regardless of the
+gate's `overflowPolicy`. Idempotency keys require deterministic outcomes;
+dropping-and-replacing a job whose key is already recorded would produce an
+inconsistent replay. Use the gate's `enqueue()` directly when you want
+`drop_oldest`.
+
+## Adjusting capacity at runtime
+
+```ts
+const adjusted = gate.adjustCapacity(
+  { role: "maintainer" },
+  8,               // new ceiling (must be ≤ maxCapacityCeiling)
+  "peak_sale_load" // stable reason code
+);
+if (adjusted.ok) {
+  console.log(`capacity: ${adjusted.value.previous} → ${adjusted.value.next}`);
+}
+```
+
+Lowering below the current live count is allowed — the gate refuses new work
+until the backlog drains to the new limit. No jobs are forcibly killed.
+
+## Waiting for a slot
+
+```ts
+while (true) {
+  const r = gate.enqueue({ operation: "analytics.aggregate", params });
+  if (r.ok) break;
+  await gate.waitForCapacity(); // resolves when a slot opens or ceiling is raised
+}
+```
+
+`waitForCapacity()` resolves immediately when capacity is already available. It
+never rejects.
+
+## Operational edge cases
+
+| Scenario | Behaviour | Recovery |
+|---|---|---|
+| Burst > `maxConcurrent` | `reject` → `backpressure_full` or `drop_oldest` eviction | Wait on `waitForCapacity()`, or raise ceiling via `adjustCapacity()` |
+| Stuck running job holds a slot | Gate's live count includes it until it leaves `running` | Force dead-letter: `framework.deadLetter(id)` — the gate heals on next check |
+| `drop_oldest` evicts a retrying job | Evicted job is dead-lettered; its payload and error context are intact | Inspect with `framework.inspect("dead_lettered")`; requeue if still needed |
+| Viewer calls `adjustCapacity` | `backpressure_unauthorized` | Only a `maintainer` actor may adjust |
+| Ceiling attempt above `maxCapacityCeiling` | `backpressure_invalid_capacity` | Use a value ≤ `maxCapacityCeiling` (default: `maxConcurrent × 10`) |
+| Framework shutdown during a full gate | In-flight jobs settle; `waitForCapacity()` resolves as slots open | No action — gate heals automatically |
+| Gate removed from `options.gates` | Unbounded behaviour restored on the next framework instance | No migration — gate state is entirely in-memory and non-durable |
+
+## Rollback
+
+Gates are non-durable and stateless between sessions. To roll back:
+
+1. Remove the gate from `options.gates` when calling `createWorkerFramework`.
+2. Replace calls to `gate.enqueue()` with `framework.enqueue()`.
+
+No schema change, no storage migration.
+
+## Stats
+
+```ts
+const { live, maxConcurrent, admitted, rejected, evicted } = gate.stats();
+```
+
+- `live` — current occupancy (recomputed from the framework map on every call).
+- `admitted` — total enqueues admitted since last reset.
+- `rejected` — total enqueues refused due to full capacity.
+- `evicted` — total jobs evicted by `drop_oldest`.
+
+## Telemetry events
+
+| Op | When |
+|---|---|
+| `backpressure.enqueue` | Every call, admitted or bypassed (non-labeled ops) |
+| `backpressure.overflow` | Capacity exceeded — includes `policy`, `evictedJobId` on `drop_oldest` |
+| `backpressure.capacity_changed` | `adjustCapacity` succeeds or is denied |
+
+## Tests
+
+```bash
+npm run test:workers
+# or specifically:
+npx vitest run core/workers/__tests__/backpressure.test.ts --reporter=verbose
+```
+
+Test file: `core/workers/__tests__/backpressure.test.ts`
+
+Covered paths: normal admit, reject overflow, drop_oldest eviction, all-running
+fallback, retry slot retention, slot release on settle, maintainer/viewer auth,
+waitForCapacity, self-healing after external dead-letter, constructor validation,
+stats and reset.
diff --git a/docs/DISASTER_RECOVERY.md b/docs/DISASTER_RECOVERY.md
index 3570b1c..7e72160 100644
--- a/docs/DISASTER_RECOVERY.md
+++ b/docs/DISASTER_RECOVERY.md
@@ -132,3 +132,48 @@ When the domain model changes:
 - Add new checks for new invariants
 - Update test fixtures and expectations
 - Document any new failure modes
+
+## Snapshot restore verification
+
+Before applying any snapshot — whether from a migration, a rollback or a
+re-hydration — run it through `verifyAndRestore` or `preflight` from
+`core/archive/snapshot.ts`. The pipeline checks:
+
+1. Schema version compatibility (blocks downgrade/upgrade unless explicitly allowed).
+2. Ledger sequence regression (blocks restores that would roll back Stellar state).
+3. Content hash integrity (blocks tampered or partially-written snapshots).
+4. Per-record invariants you declare for your domain type.
+
+A failing check throws `SnapshotMismatchError` **before** `onRestore` is called,
+so prior state is always intact on failure.
+
+**Pre-flight check:**
+
+```ts
+import { preflight } from "@/core/archive/snapshot";
+
+const check = preflight(incomingSnap, { invariants, expectedSchemaVersion: 2 });
+if (!check.ok) {
+  // check.code === "snapshot_invariant_violation"
+  // check.detail — array of SnapshotViolation, one per failing record/check
+  return reportViolations(check.detail);
+}
+// safe to call verifyAndRestore with onRestore
+```
+
+**Rollback point:**
+
+```ts
+// Capture before any restore attempt.
+const rollbackSnap = captureSnapshot(currentRecords, { schemaVersion, ledgerSequence });
+
+try {
+  verifyAndRestore(incomingSnap, { invariants, onRestore: commit });
+} catch (e) {
+  if (e instanceof SnapshotMismatchError) {
+    verifyAndRestore(rollbackSnap, { invariants, onRestore: commit }); // revert
+  }
+}
+```
+
+Full documentation: [docs/SNAPSHOT_RESTORE.md](./SNAPSHOT_RESTORE.md)
diff --git a/docs/SNAPSHOT_RESTORE.md b/docs/SNAPSHOT_RESTORE.md
new file mode 100644
index 0000000..3b964e6
--- /dev/null
+++ b/docs/SNAPSHOT_RESTORE.md
@@ -0,0 +1,216 @@
+# Snapshot Capture and Restore Verification
+
+Utix stores all state in the browser (localStorage + in-memory). When a session
+restores from a snapshot — after re-hydration, a forced rollback or a data
+migration — the values that come back may be mutually inconsistent: primary
+records may not match what the derived indexes point to, schema versions may
+have moved, or a ledger sequence may have regressed.
+
+`core/archive/snapshot.ts` provides a two-phase restore pipeline that verifies
+every declared invariant, rebuilds derived indexes deterministically, and **fails
+loudly** on any mismatch — so an inconsistent state is never silently applied.
+
+## Design principles
+
+- **Verify before commit.** `verifyAndRestore` is a pipeline, not a write
+  function. It does not touch any storage directly; the caller supplies
+  `onRestore` as the commit step, and it is only called when all checks pass.
+- **Loud failure, never silent.** On any hard violation, `verifyAndRestore`
+  throws `SnapshotMismatchError` *and* emits `snapshot.restore_failed`
+  telemetry *and* writes a `projection.replayed denied` audit event — three
+  independent signals so the failure surfaces even if one channel is missed.
+- **All violations at once.** Every failing invariant and every failing record is
+  collected before the throw, so operators see the complete picture, not just the
+  first failure.
+- **Index drift is reported, not fatal.** Indexes are derived views; drift is
+  expected during migrations. `SnapshotRestoreResult.indexMismatches` lists
+  every differing key. The caller decides whether drift is acceptable.
+- **`preflight` for pre-flight checks.** `preflight()` wraps `verifyAndRestore`
+  with `dryRun: true` and returns a `Result` instead of throwing — use it to
+  validate before committing.
+
+## Quick start
+
+```ts
+import {
+  captureSnapshot,
+  verifyAndRestore,
+  preflight,
+  SnapshotMismatchError,
+  type SnapshotInvariant
+} from "@/core/archive/snapshot";
+
+// 1. Declare invariants for your record type.
+const invariants: SnapshotInvariant<TicketRecord>[] = [
+  {
+    name: "ticket.has_owner",
+    description: "Every ticket must have a non-empty ownerId.",
+    check: (r) => typeof r.ownerId === "string" && r.ownerId.length > 0
+  },
+  {
+    name: "ticket.positive_amount",
+    check: (r) => r.amount > 0
+  }
+];
+
+// 2. Capture the current state.
+const snap = captureSnapshot(records, {
+  schemaVersion: 2,
+  ledgerSequence: currentLedger,
+  indexBuilder: (rs) => Object.fromEntries(rs.map((r) => [`ticket:${r.id}`, r.ownerId])),
+  hashFn: (rs) => rs.map((r) => r.id).sort().join(","),
+  metadata: { capturedBy: "operator", reason: "pre_migration" }
+});
+
+// 3. Validate and restore.
+try {
+  const result = verifyAndRestore(snap, {
+    invariants,
+    expectedSchemaVersion: 2,
+    minLedgerSequence: 500,
+    indexBuilder: buildIndex,
+    hashFn: hashRecords,
+    onRestore: (records, index) => {
+      store.tickets = [...records];
+      store.index = { ...index };
+    }
+  });
+  console.log(`Restored ${result.primaryRecordCount} records`);
+  if (result.indexMismatches.length > 0) {
+    console.warn("Index drift detected:", result.indexMismatches);
+  }
+} catch (e) {
+  if (e instanceof SnapshotMismatchError) {
+    console.error("Restore failed:", e.violations);
+    // onRestore was NOT called — previous state is intact.
+  }
+}
+```
+
+## Rollback pattern
+
+Always capture a pre-restore snapshot before applying an incoming one. If the
+restore fails, re-apply the pre-restore snapshot.
+
+```ts
+// Step 1: capture the current state as the rollback point.
+const preRestoreSnap = captureSnapshot(store.tickets, {
+  schemaVersion: CURRENT_SCHEMA,
+  ledgerSequence: store.ledgerSequence,
+  indexBuilder: buildIndex
+});
+
+// Step 2: preflight the incoming snapshot (no commit yet).
+const check = preflight(incomingSnap, { invariants, expectedSchemaVersion: 2 });
+if (!check.ok) {
+  // Violations in check.detail — do not proceed.
+  return { error: check.code, violations: check.detail };
+}
+
+// Step 3: commit.
+try {
+  verifyAndRestore(incomingSnap, { invariants, indexBuilder: buildIndex, onRestore: commit });
+} catch (e) {
+  if (e instanceof SnapshotMismatchError) {
+    // Step 4: rollback — re-apply the pre-restore snapshot.
+    verifyAndRestore(preRestoreSnap, { invariants, indexBuilder: buildIndex, onRestore: commit });
+  }
+}
+```
+
+## Verification phases
+
+`verifyAndRestore` runs four phases in order. It collects all violations before
+throwing; only hard violations (phases 1–2) throw. Phase 3 is index diff (never
+throws). Phase 4 is the commit (only if `dryRun` is false and phase 1–2 pass).
+
+| Phase | Check | Throws? |
+|---|---|---|
+| 1a | Schema version — snapshot matches `expectedSchemaVersion` | Yes |
+| 1b | Ledger sequence — snapshot ≥ `minLedgerSequence` | Yes |
+| 1c | Content hash — re-hash matches `snapshot.contentHash` | Yes |
+| 2 | Per-record invariants — every record passes every `SnapshotInvariant` | Yes |
+| 3 | Index rebuild diff — rebuilt index vs. snapshot's stored index | No (reported) |
+| 4 | Commit — `onRestore(records, rebuiltIndex)` called | N/A |
+
+## Operational edge cases
+
+| Scenario | Behaviour | Recovery |
+|---|---|---|
+| Invariant violation | `SnapshotMismatchError` thrown; `onRestore` never called; prior state untouched | Inspect `error.violations`, fix snapshot data, re-run. Use `preflight()` to check first. |
+| Schema version regression (snapshot older than runtime) | `schema_version_mismatch` violation — blocked | Run migration to upgrade snapshot, re-capture, then restore. Or set `allowSchemaMigration: true` to accept with a warning. |
+| Schema version upgrade (snapshot newer than runtime) | `schema_version_mismatch` violation — blocked | Upgrade the application before restoring this snapshot. |
+| Ledger sequence regression | `ledger_sequence_regression` violation — blocked | Restore from a newer snapshot, or lower `minLedgerSequence` if the regression is intentional. |
+| Content hash mismatch | `content_hash_mismatch` violation — blocked | Discard the snapshot and re-capture from a trusted source. |
+| Index drift after rebuild | `indexMismatches` reported in the result — **not thrown** | Review each mismatch; drift is expected during migrations. If strict fidelity is required, check `result.indexMismatches.length === 0` before accepting the result. |
+| Throwing invariant check | Recorded as `invariant_failed` violation, not re-thrown | Fix the invariant's `check` function or the record that caused the throw. |
+| `dryRun: true` | All phases run, `onRestore` never called | Use `preflight()` as the standard pre-flight API — it wraps `dryRun` and returns a `Result`. |
+
+## Invariant authoring guidelines
+
+- Keep `check` functions **pure and fast** — they run synchronously for every
+  record in the snapshot.
+- Give every invariant a **stable `name`** — it appears in violation reports,
+  telemetry and the audit trail. Use `domain.rule` convention, e.g.
+  `ticket.has_owner`.
+- Write a **`description`** — it replaces the generic violation detail when the
+  check returns `false`. Operators read it in the audit trail.
+- A throwing `check` is recorded as a violation. Do not throw intentionally;
+  return `false` instead.
+- Invariants are declared per caller, not per module, so `snapshot.ts` never
+  needs to know your domain model.
+
+## Content hash authoring guidelines
+
+A hash function must be **deterministic**: the same records in the same order
+must always produce the same string. Safe approaches:
+
+```ts
+// Sort-stable hash over record ids:
+const hashFn = (rs: T[]) => rs.map((r) => r.id).sort().join(",");
+
+// Full JSON (deterministic only if key order is stable):
+const hashFn = (rs: T[]) => JSON.stringify(rs.map((r) => ({ id: r.id, v: r.version })));
+```
+
+Do **not** use `Date.now()`, `Math.random()` or any non-deterministic value.
+
+## Audit trail
+
+Every restore attempt writes a `projection.replayed` audit event:
+
+- **`outcome: "allowed"`** — restore passed all checks and `onRestore` ran.
+- **`outcome: "denied"`** — one or more violations were found; restore was
+  blocked. `errorCode: "snapshot_invariant_violation"`.
+
+Query via:
+
+```ts
+import { getAuditTrail } from "@/core/audit/audit";
+const events = getAuditTrail()
+  .all()
+  .filter((e) => e.action === "projection.replayed" && e.target.kind === "snapshot");
+```
+
+## Telemetry events
+
+| Op | When |
+|---|---|
+| `snapshot.restore_succeeded` | All checks passed, `onRestore` called (or `dryRun: true`) |
+| `snapshot.restore_failed` | Any hard violation — emitted **before** the throw |
+
+Both events carry `snapshotId`, `schemaVersion`, `ledgerSequence` and
+`recordCount` in their payload.
+
+## Tests
+
+```bash
+npx vitest run core/archive/__tests__/snapshot.test.ts --reporter=verbose
+```
+
+Test file: `core/archive/__tests__/snapshot.test.ts`
+
+Covered paths: captureSnapshot (stamp, hash, index), all invariant violation
+types (failed, throwing check), schema downgrade/upgrade, ledger regression,
+hash mismatch, index drift (reported not thrown), dryRun, preflight Result
+return, rollback sequence, telemetry emission, audit trail (success + failure).
diff --git a/docs/WORKERS.md b/docs/WORKERS.md
index 337813e..e67b8c2 100644
--- a/docs/WORKERS.md
+++ b/docs/WORKERS.md
@@ -119,4 +119,68 @@ telemetry will appear as structured lines.
 
 Each job emits `worker.run` (success) or `worker.retry` / `worker.exhausted`
 (failure) telemetry with latency, error code and the job's correlation id. See
-[docs/TELEMETRY.md](./TELEMETRY.md) for the dashboard queries.
\ No newline at end of file
+[docs/TELEMETRY.md](./TELEMETRY.md) for the dashboard queries.
+
+## Bounded queue backpressure
+
+Expensive operations (analytics, Soroban simulations, index rebuilds, export
+generations) can flood the queue if called in a burst. `core/workers/backpressure.ts`
+adds a **`BoundedEnqueueGate`** that enforces a hard concurrency ceiling per
+operation label, with two overflow policies:
+
+- **`reject`** (default) — returns `backpressure_full` immediately.
+- **`drop_oldest`** — evicts the oldest queued/retrying job in the label group
+  to make room; useful for "latest state wins" operations.
+
+### Gate as a standalone facade
+
+```ts
+import { createBoundedEnqueueGate } from "@/core/workers/backpressure";
+
+const gate = createBoundedEnqueueGate(framework, {
+  label: "analytics",   // meters ops whose name starts with "analytics"
+  maxConcurrent: 4,
+  overflowPolicy: "drop_oldest"
+});
+
+const result = gate.enqueue({ operation: "analytics.aggregate", params });
+if (!result.ok) {
+  await gate.waitForCapacity(); // resolves when a slot opens or ceiling is raised
+}
+```
+
+### Gate wired into `submit()`
+
+Pass gates to `createWorkerFramework` so `submit()` checks capacity before
+spending idempotency/quota:
+
+```ts
+const framework = createWorkerFramework({}, { gates: [gate] });
+const result = framework.submit({ operation: "analytics.aggregate", idempotencyKey: key });
+if (!result.ok && result.code === "backpressure_full") {
+  // quota was NOT spent — safe to retry after waitForCapacity()
+}
+```
+
+`submit()` always uses **reject** semantics regardless of the gate's
+`overflowPolicy`. For `drop_oldest`, use `gate.enqueue()` directly.
+
+### Adjusting capacity at runtime (maintainer only)
+
+```ts
+gate.adjustCapacity({ role: "maintainer" }, 8, "peak_load");
+```
+
+A `viewer` receives `backpressure_unauthorized`. The hard cap `maxCapacityCeiling`
+(default: `maxConcurrent × 10`) cannot be exceeded regardless of role.
+
+### Recovery notes
+
+- **Stuck running job**: force dead-letter via `framework.deadLetter(id)` — the
+  gate heals automatically on the next enqueue check.
+- **Evicted jobs** (drop_oldest): inspect with `framework.inspect("dead_lettered")`
+  and requeue any that are still needed.
+- **Remove the gate**: delete it from `options.gates` and replace `gate.enqueue()`
+  calls with `framework.enqueue()`. No state migration needed.
+
+Full documentation: [docs/BACKPRESSURE.md](./BACKPRESSURE.md)