From 07975afea7d4dab8b8f3024c96b5bf2c0a9f9c8f Mon Sep 17 00:00:00 2001
From: gbemisola299 <gbemisola299@gmail.com>
Date: Tue, 20 Oct 2026 10:00:38 +0100
Subject: [PATCH] feat(infra): flag-gated slices for spikes 603/604/605/606
 (#467 #468 #469 #470)

phase-170 (#470): structured log pipeline with deep secret redaction,
per-correlation-id sampling, closed index labels and a bounded
non-blocking https batcher wired into api-observability.

phase-171 (#469): lock-consistent backup snapshots (VACUUM INTO + sidecars),
checksum/integrity verification, restore marker that gates the faucet-health
cron, AES-256-GCM sealed secrets envelope, and a DR drill script.

phase-172 (#468): SQLite to Postgres rehearsal helpers: dialect translation,
typed per-table parity with coercion detection, per-test schemas, and a
bounded dual-writer with drift log and primary-only rollback.

phase-173 (#467): preview isolation: per-preview data root, production
issuer/cron/contract/mainnet fencing in validatePhaseEnv, preview cron gating.

All behind default-off flags; docs, flag rows and findings updated.
---
 .env.local.example                            |  13 +
 PROJECT_ARCHITECTURE.md                       |   4 +
 app/api/cron/faucet-health/route.ts           |  19 +-
 docs/TECHNICAL.md                             |  46 +++
 docs/spikes/603-preview-isolation-findings.md |  39 ++
 .../spikes/604-postgres-rehearsal-findings.md |  35 ++
 docs/spikes/605-backup-dr-findings.md         |  38 ++
 docs/spikes/606-log-pipeline-findings.md      |  34 ++
 lib/__tests__/backup-snapshot.test.ts         | 170 ++++++++
 lib/__tests__/db-parity.test.ts               | 178 +++++++++
 lib/__tests__/log-pipeline.test.ts            | 204 ++++++++++
 lib/__tests__/preview-isolation.test.ts       | 121 ++++++
 lib/api-observability.ts                      |   5 +-
 lib/backup-snapshot.ts                        | 367 ++++++++++++++++++
 lib/db-parity.ts                              | 335 ++++++++++++++++
 lib/env-validation.ts                         |  27 +-
 lib/feature-flags.ts                          |  18 +-
 lib/log-pipeline.ts                           | 294 ++++++++++++++
 lib/preview-isolation.ts                      | 187 +++++++++
 lib/security-counters.ts                      |  20 +-
 lib/server-data-paths.ts                      |  19 +-
 package.json                                  |   4 +-
 scripts/backup-drill.ts                       |  51 +++
 scripts/db-parity.ts                          |  34 ++
 24 files changed, 2254 insertions(+), 8 deletions(-)
 create mode 100644 docs/spikes/603-preview-isolation-findings.md
 create mode 100644 docs/spikes/604-postgres-rehearsal-findings.md
 create mode 100644 docs/spikes/605-backup-dr-findings.md
 create mode 100644 docs/spikes/606-log-pipeline-findings.md
 create mode 100644 lib/__tests__/backup-snapshot.test.ts
 create mode 100644 lib/__tests__/db-parity.test.ts
 create mode 100644 lib/__tests__/log-pipeline.test.ts
 create mode 100644 lib/__tests__/preview-isolation.test.ts
 create mode 100644 lib/backup-snapshot.ts
 create mode 100644 lib/db-parity.ts
 create mode 100644 lib/log-pipeline.ts
 create mode 100644 lib/preview-isolation.ts
 create mode 100644 scripts/backup-drill.ts
 create mode 100644 scripts/db-parity.ts

diff --git a/.env.local.example b/.env.local.example
index 5ce48d8e..27d962ec 100644
--- a/.env.local.example
+++ b/.env.local.example
@@ -208,3 +208,16 @@ NEXT_PUBLIC_CLASSIC_LIQ_ISSUER=
 # phase-124: metadata version migration tool v1→v2 (scripts/reset-phase --migrate-metadata)
 # NEXT_PUBLIC_FEATURE_PHASE_124=1
 # FEATURE_PHASE_124=1
+
+# ─── Spikes 603–606 (phase-170..173) — default off ────────────────────────
+# phase-170 log pipeline (https only). Secrets/seeds are redacted before shipping.
+# LOG_SINK_URL=https://logs.example.com/ingest
+# LOG_SINK_TOKEN=
+# LOG_SAMPLE_RATE_INFO=0.1
+# phase-172 enters dual-write only when this is set
+# PHASE_DATABASE_URL=
+# phase-173 production identity fingerprints (public/hashed only) for preview checks
+# PHASE_PROD_ISSUER=G...
+# PHASE_PROD_CRON_SECRET_SHA256=
+# PHASE_PROD_CONTRACT_IDS=C...,C...
+# PHASE_PREVIEW_CRON_ENABLED=1
diff --git a/PROJECT_ARCHITECTURE.md b/PROJECT_ARCHITECTURE.md
index db18ba82..a1ac2992 100644
--- a/PROJECT_ARCHITECTURE.md
+++ b/PROJECT_ARCHITECTURE.md
@@ -211,6 +211,10 @@ is a fixed-size digest of the canonical payload, keeping it under wallet
 | `phase-167` | `NEXT_PUBLIC_FEATURE_PHASE_167` / `FEATURE_PHASE_167` | WalletConnect v2 mobile session resilience (`lib/walletconnect-resilience.ts`): configurable relay URL, multi-chain support, session persistence, chain-switch guard, and mobile deep-link return. Spike 623 (issue #486) | off | Unset var, restart — WalletConnect banner and reconnect actions disabled, kit falls back to default single-session behavior |
 | `phase-168` | `NEXT_PUBLIC_FEATURE_PHASE_168` / `FEATURE_PHASE_168` | xBull wallet full support and parity (`lib/xbull-adapter.ts`): Freighter-difference matrix, xBull module detection, sign result normalization, trustline error translation, and badge surfacing. Spike 624 (issue #487) | off | Unset var, restart — xBull transactions fall through generic kit handlers, badge hidden |
 | `phase-169` | `NEXT_PUBLIC_FEATURE_PHASE_169` / `FEATURE_PHASE_169` | Ledger blind-signing to clear-signing review for Soroban (`lib/ledger-soroban-decoder.ts`): invokeHostFunction pre-image decode, verification code, on-device review modal, Gate #13 preemptive validation, and hardware error humanizer. Spike 625 (issue #488) | off | Unset var, restart — Ledger transactions fall back to default blind-signing flow without preflight review modal |
+| `phase-170` | `NEXT_PUBLIC_FEATURE_PHASE_170` / `FEATURE_PHASE_170` | Structured log pipeline (`lib/log-pipeline.ts`): deep secret redaction, deterministic per-correlation-id info sampling (warn/error always kept), closed index-label set (`level`, `event`, `route`, `method`, `status_class`), and a bounded non-blocking HTTPS batcher (`LOG_SINK_URL`, https only). `api-observability` routes every line through it. Spike 606 (issue #470) — partial slice: no `waitUntil` flush on freeze, no counter export, ad-hoc `console.*` callers not migrated | off | Unset var, restart — `emitLog` is the identity and console lines are written exactly as before; in-memory buffer dropped |
+| `phase-171` | `NEXT_PUBLIC_FEATURE_PHASE_171` / `FEATURE_PHASE_171` | Backup and DR snapshots (`lib/backup-snapshot.ts`): lock-consistent sidecar set + `VACUUM INTO` SQLite image, sha256/row-count manifest, `integrity_check`/`foreign_key_check` verify, tmp+rename restore with a restore marker that makes `/api/cron/faucet-health` return 503, and an AES-256-GCM sealed secrets envelope. `scripts/backup-drill.ts` runs snapshot→verify→restore→parity. Spike 605 (issue #469) — partial slice: no off-box shipping or scheduled snapshots | off | Unset var, restart — entry points throw `FLAG_DISABLED`, restore marker ignored; snapshots on disk are inert |
+| `phase-172` | `NEXT_PUBLIC_FEATURE_PHASE_172` / `FEATURE_PHASE_172` | SQLite→Postgres rehearsal (`lib/db-parity.ts`): dialect helpers (`?`→`$n`, type map, explicit upsert), typed per-table parity (value, `type_coercion`, checksum), per-test schema names, and a `DualWriter` (primary first, bounded/timeout-guarded secondary, drift log, `primary-only` rollback). Needs `PHASE_DATABASE_URL` to enter `dual` mode. Spike 604 (issue #468) — partial slice: no Postgres driver, async CAS re-plumbing or pool storm test | off | Unset var, restart — `resolveDualWriteMode` returns `primary-only`; SQLite stays authoritative, nothing in production paths calls the module |
+| `phase-173` | `NEXT_PUBLIC_FEATURE_PHASE_173` / `FEATURE_PHASE_173` | Vercel preview isolation (`lib/preview-isolation.ts`): per-preview data root (`<root>/previews/<pr-N or br-slug-hash>`), `validatePhaseEnv` fails a preview that shares the production issuer, cron secret, contract IDs, or runs mainnet (declared via `PHASE_PROD_ISSUER`, `PHASE_PROD_CRON_SECRET_SHA256`, `PHASE_PROD_CONTRACT_IDS`), preview cron skipped unless `PHASE_PREVIEW_CRON_ENABLED=1`, and `x-vercel-cron` no longer trusted in previews. Spike 603 (issue #467) — partial slice: no per-PR contract deploy, preview CI job, or branch-DB seeding | off | Unset var, restart — data root, cron auth and env validation revert; per-preview directories stay inert |
 
 
 
diff --git a/app/api/cron/faucet-health/route.ts b/app/api/cron/faucet-health/route.ts
index f856f4a5..28ff575f 100644
--- a/app/api/cron/faucet-health/route.ts
+++ b/app/api/cron/faucet-health/route.ts
@@ -3,6 +3,9 @@ import { Keypair, StrKey } from "@stellar/stellar-sdk"
 import { fetchDistributorBalance, distributorNeedsTopup } from "@/lib/distributor-topup"
 import { sendWebhookAlert, WebhookAlertType } from "@/lib/webhook-alerts"
 import { executeDistributorRefill } from "@/lib/distributor-refill"
+import { isRestoreInProgress } from "@/lib/backup-snapshot"
+import { isCronBlockedInPreview, previewTrustsVercelCronHeader } from "@/lib/preview-isolation"
+import { incSecurityCounter } from "@/lib/security-counters"
 import { getDistributorHealthStatus, recordHealthCheck } from "@/lib/distributor-health-store"
 
 /**
@@ -44,7 +47,7 @@ function validateCronAuth(req: NextRequest): boolean {
   
   // Vercel Cron sends this header
   const vercelCronHeader = req.headers.get("x-vercel-cron")
-  if (vercelCronHeader) {
+  if (vercelCronHeader && previewTrustsVercelCronHeader()) {
     return true
   }
   
@@ -84,6 +87,20 @@ export async function GET(req: NextRequest) {
     )
   }
 
+  // phase-173: previews do not run scheduled refills against the shared issuer.
+  if (isCronBlockedInPreview()) {
+    incSecurityCounter("preview_cron_blocked")
+    return NextResponse.json({ ok: true, skipped: "preview_cron_disabled" })
+  }
+
+  // phase-171: a refill mid-restore would act on half-restored distributor state.
+  if (await isRestoreInProgress()) {
+    return NextResponse.json(
+      { error: "Restore in progress" },
+      { status: 503, headers: { "Retry-After": "60" } },
+    )
+  }
+
   // Only run health checks in transfer mode
   if (!faucetUsesDistributorTransfer()) {
     return NextResponse.json({
diff --git a/docs/TECHNICAL.md b/docs/TECHNICAL.md
index 15cc590a..a995d80c 100644
--- a/docs/TECHNICAL.md
+++ b/docs/TECHNICAL.md
@@ -419,6 +419,11 @@ NEXT_PUBLIC_FEATURE_PHASE_158=1
 NEXT_PUBLIC_FEATURE_PHASE_159=1
 NEXT_PUBLIC_FEATURE_PHASE_160=1
 NEXT_PUBLIC_FEATURE_PHASE_161=1
+# Spikes 603/604/605/606 (issues #467/#468/#469/#470)
+NEXT_PUBLIC_FEATURE_PHASE_170=1
+NEXT_PUBLIC_FEATURE_PHASE_171=1
+NEXT_PUBLIC_FEATURE_PHASE_172=1
+NEXT_PUBLIC_FEATURE_PHASE_173=1
 # Server-only aliases also accepted: FEATURE_PHASE_104, etc.
 ```
 
@@ -695,5 +700,46 @@ Flag: `phase-169` (`NEXT_PUBLIC_FEATURE_PHASE_169` / `FEATURE_PHASE_169`). Defau
 - Audit counters: `ledger_clear_signing_reviewed`, `ledger_clear_signing_approved`, `ledger_clear_signing_rejected`, `ledger_blind_signing_blocked`, `ledger_gate13_preempted` in `lib/security-counters.ts`.
 - Routes: `POST /api/wallet/ledger` for clear-signing lifecycle telemetry.
 
+## 20. Log pipeline: redaction, sampling and external sink (spike 606)
 
+Flag: `phase-170` (`NEXT_PUBLIC_FEATURE_PHASE_170` / `FEATURE_PHASE_170`). Default off. See `PROJECT_ARCHITECTURE.md` §10 and `docs/spikes/606-log-pipeline-findings.md`.
 
+- `lib/log-pipeline.ts`: `redactLogValue` (masks `S...` seeds and secret-named keys such as `rawPayload`, `errorResult`, `authorization`; cycle/depth safe), `shouldSampleLog` (warn/error always kept, info sampled by FNV hash of the correlation id so a request is kept or dropped whole), `toPipelineRecord` (only `level`, `event`, `route`, `method`, `status_class` are labels; wallets, CIDs, token ids, hashes stay unindexed in `fields`), and `LogPipeline` (bounded buffer, drop-oldest, single-flight flush, abort timeout, at-most-once delivery).
+- `lib/api-observability.ts`: every `log()` goes through `emitLog`, so the console line is redacted too when the flag is on.
+- Env: `LOG_SINK_URL` (https only, validated in `serverEnvSchema`), `LOG_SINK_TOKEN`, `LOG_SAMPLE_RATE_INFO` (0..1, default 0.1), `LOG_SINK_BATCH_SIZE`, `LOG_SINK_MAX_BUFFER`, `LOG_SINK_FLUSH_MS`, `LOG_SINK_TIMEOUT_MS`.
+- Counters (`lib/security-counters.ts`): `log_pipeline_enqueued`, `_sampled_out`, `_dropped`, `_redacted`, `_flush_ok`, `_flush_failed`.
+- Not covered: `waitUntil`-style flush on serverless freeze, counter export, migrating ad-hoc `console.*` callers.
+
+## 21. Backup and disaster recovery (spike 605)
+
+Flag: `phase-171` (`NEXT_PUBLIC_FEATURE_PHASE_171` / `FEATURE_PHASE_171`). Default off. See `docs/spikes/605-backup-dr-findings.md`.
+
+- `lib/backup-snapshot.ts`: `createSnapshot` locks every sidecar path (sorted, via `withFileLock`) and takes the SQLite image with `VACUUM INTO`, so the set and the database share one logical instant and no WAL copy is involved; a corrupt sidecar fails the snapshot. `verifySnapshot` re-hashes, parses JSON, runs `integrity_check` and `foreign_key_check`, and compares per-table row counts. `restoreSnapshot` verifies first, removes stale `-wal`/`-shm`, writes tmp+rename, and holds `.restore-in-progress` (stale after 10 min).
+- `app/api/cron/faucet-health/route.ts`: returns `503` + `Retry-After: 60` while the marker is fresh, so a refill cannot run against half-restored distributor state.
+- Secrets: never written to a snapshot. `sealSecrets`/`openSecrets` use AES-256-GCM with a scrypt key (passphrase ≥ 16 chars); failures never echo plaintext; `diffSecretsMap` reports drift by name only.
+- `lib/server-data-paths.ts`: `serverDataRoot`, `serverDataFileKeys`, `serverDataFileName` exported so backups cannot drift from the file table.
+- Drill: `npm run backup:drill` (snapshot → verify → restore to scratch → SQLite parity).
+- Restore replaces the database file: close the connection (`resetDbForTests`) and run offline for the live root.
+- Counters: `backup_snapshot_created|verified|corrupt`, `backup_restore_completed|blocked`.
+
+## 22. SQLite to Postgres migration rehearsal (spike 604)
+
+Flag: `phase-172` (`NEXT_PUBLIC_FEATURE_PHASE_172` / `FEATURE_PHASE_172`). Default off. See `docs/spikes/604-postgres-rehearsal-findings.md`.
+
+- `lib/db-parity.ts`: `toPositionalParams`, `sqliteTypeToPostgres` (integer flags stay `BIGINT` so parity compares like with like), `buildPostgresCreateTable`, `buildPostgresUpsert` (explicit conflict target replaces `INSERT OR REPLACE`), `perTestSchemaName`, `compareSources`/`compareTable` (value diffs, `type_coercion`, order-independent checksum), `DualWriter`.
+- Dual-write ordering: primary (SQLite) first and authoritative; secondary is bounded (`maxSecondaryInflight`, default 5) and timed out (1.5 s). Failures, timeouts and shed writes land in `driftLog()` and never throw. `setMode("primary-only")` is the rollback.
+- `PHASE_DATABASE_URL` + flag → `dual`; otherwise `primary-only`.
+- Check two databases with `npm run db:parity -- <left> <right>`.
+- Counters: `db_parity_checked`, `db_parity_mismatch`, `db_dual_write_ok|drift|shed`.
+- Not covered: a `pg` driver, async re-plumbing of the `MAX_CAS_ATTEMPTS` loops, advisory locks replacing `fileQueues`, pool storm testing.
+
+## 23. Preview environment isolation (spike 603)
+
+Flag: `phase-173` (`NEXT_PUBLIC_FEATURE_PHASE_173` / `FEATURE_PHASE_173`). Default off. See `docs/spikes/603-preview-isolation-findings.md`.
+
+- `lib/preview-isolation.ts`: `resolvePreviewId` (`pr-<n>` from `VERCEL_GIT_PULL_REQUEST_ID`, else `br-<slug>-<hash>`; always one safe path segment), `checkPreviewIsolation`, `isCronBlockedInPreview`, `previewTrustsVercelCronHeader`. The module has no `node:*` imports because `lib/env-validation.ts` reaches client bundles through `phase-protocol.ts`.
+- `lib/server-data-paths.ts`: in a preview the root becomes `<root>/previews/<id>`.
+- `lib/env-validation.ts`: `validatePhaseEnv` reports `isolation_violation` for a preview that resolves to the production issuer (public key or derived from `CLASSIC_LIQ_ISSUER_SECRET`), reuses the production cron secret, uses a production contract ID, or sets `STELLAR_NETWORK=mainnet`. Production identity is declared with `PHASE_PROD_ISSUER`, `PHASE_PROD_CRON_SECRET_SHA256`, `PHASE_PROD_CONTRACT_IDS`; messages never contain secret values.
+- `app/api/cron/faucet-health/route.ts`: preview runs return `{ skipped: "preview_cron_disabled" }` unless `PHASE_PREVIEW_CRON_ENABLED=1`; the unauthenticated-header shortcut is disabled in previews.
+- Caveat: Vercel `/tmp` is per-instance, so namespacing prevents cross-preview bleed, not loss between instances.
+- Counters: `preview_isolation_violation`, `preview_cron_blocked`.
diff --git a/docs/spikes/603-preview-isolation-findings.md b/docs/spikes/603-preview-isolation-findings.md
new file mode 100644
index 00000000..e3f18941
--- /dev/null
+++ b/docs/spikes/603-preview-isolation-findings.md
@@ -0,0 +1,39 @@
+# Spike 603 — Vercel Preview-Environment Isolation with Per-PR Contracts, Sidecars and Data Findings (Issue #467)
+
+## What the issue claimed / what the code did
+
+- One staging lane (`deploy-staging` on push to main); PRs run contracts and a web build only.
+- The data root is ambient (`cwd/.data`, `os.tmpdir()` on Vercel, `PHASE_SERVER_DATA_DIR`) and nothing set it per preview.
+- Issuer, distributor and cron secrets are global; validation checked shape, not isolation.
+- `x-vercel-cron` was trusted as authentication, and cron fires per project: N previews could drain one issuer.
+
+## Delivered slice (phase-173)
+
+- **Identity**: `resolvePreviewId` yields `pr-<n>` or `br-<slug>-<hash>` (hash so `feat/a-b` and `feat/a/b` differ); always a single safe path segment, `null` outside previews.
+- **Data**: `serverDataRoot()` appends `previews/<id>` when the flag is on.
+- **Fencing**: `checkPreviewIsolation` (surfaced by `validatePhaseEnv` as `isolation_violation`) flags a production issuer (public key or derived from the secret), the production cron secret (compared by sha256 only), production contract IDs, and mainnet. Production identity is supplied as public or hashed values, so no production secret is ever needed in a preview.
+- **Cron**: previews skip scheduled work unless `PHASE_PREVIEW_CRON_ENABLED=1` and no longer trust `x-vercel-cron`.
+- `lib/preview-isolation.ts` has no `node:*` imports because `env-validation` is bundled for the client via `phase-protocol.ts`.
+
+## Affected files: changed or explicitly ruled out
+
+| File | Status |
+|---|---|
+| `lib/server-data-paths.ts` | Changed: per-preview root. |
+| `lib/env-validation.ts` | Changed: isolation violations. |
+| `app/api/cron/faucet-health/route.ts` | Changed: preview cron gate, header trust. |
+| `.env.local.example` | Changed: new variables. |
+| `lib/sqlite-db.ts`, `lib/json-store.ts` | Used unchanged: both resolve paths via `serverDataJsonPath`, so they inherit the namespace. |
+| `.github/workflows/ci.yml`, `vercel.json` | Ruled out: a preview deploy job needs Vercel credentials that cannot be validated here. |
+| `lib/phase-contract-defaults.ts`, `lib/phase-protocol.ts`, `next.config.mjs` | Ruled out: per-PR contract IDs are baked at build time and need a deploy pipeline. |
+| `diagnose-env.ts`, `scripts/diagnose-env.ts` | Ruled out: they call `validatePhaseEnv`, so they report violations without changes. |
+| `scripts/deploy-phase-sep50.ts`, `scripts/deploy-phase.ts`, `scripts/seed-phaser-prueba.ts`, `scripts/migrate-json-to-sqlite.ts` | Ruled out: per-PR deploy and seeding automation. |
+| `app/api/faucet/route.ts` | Ruled out: it inherits the namespaced data root. |
+
+## Rollback
+
+Unset `FEATURE_PHASE_173` and restart: data root, cron auth and validation revert; per-preview directories stay inert.
+
+## Follow-ups
+
+This is a partial slice. The issue's acceptance criteria that need a live testnet or hosted environment (end-to-end transaction proof, kill-the-primary resilience demo) are not claimed here; see `docs/TECHNICAL.md` for what the unit tests do cover.
diff --git a/docs/spikes/604-postgres-rehearsal-findings.md b/docs/spikes/604-postgres-rehearsal-findings.md
new file mode 100644
index 00000000..667cd453
--- /dev/null
+++ b/docs/spikes/604-postgres-rehearsal-findings.md
@@ -0,0 +1,35 @@
+# Spike 604 — SQLite to Postgres Migration Rehearsal with Dual-Write, Parity and Rollback Findings (Issue #468)
+
+## What the issue claimed / what the code did
+
+- `lib/sqlite-db.ts` owns the schema with additive `ALTER TABLE`; no Postgres dialect and no `DATABASE_URL`.
+- The only migration precedent is a one-shot JSON→SQLite `INSERT OR REPLACE` with a count-parity exit and no checksum or down-migration.
+- Callers rely on synchronous `DatabaseSync`, `BEGIN IMMEDIATE`, and `WHERE id=? AND version=?` CAS.
+- Count parity cannot see REAL vs NUMERIC, integer flags vs BOOLEAN, or BIGINT returned as a string.
+
+## Delivered slice (phase-172)
+
+- **Dialect** (`lib/db-parity.ts`): `?`→`$n`, type map (integer flags stay `BIGINT`), generated `CREATE TABLE`, and an explicit `ON CONFLICT` upsert replacing `INSERT OR REPLACE`; identifiers are validated.
+- **Parity**: per declared column type. Postgres-style string numerics and booleans are accepted for numeric columns; a TEXT column returning a number is reported as `type_coercion`; per-table order-independent checksum; missing/extra rows by primary key.
+- **Dual-write**: primary (SQLite) first and the crash winner; a primary failure never reaches the secondary. The secondary is bounded (in-flight cap 5) and timed out (1.5 s); failures, timeouts and shed writes go to a drift log and never throw, so pool exhaustion degrades to SQLite-only. Rollback is `setMode("primary-only")`.
+- **Tests**: `perTestSchemaName` gives each test its own schema on a shared Postgres.
+
+## Affected files: changed or explicitly ruled out
+
+| File | Status |
+|---|---|
+| `lib/sqlite-db.ts` | Used unchanged (`getDb`, schema); read through `sqliteParitySource`. |
+| `package.json` | Changed: `db:parity`. |
+| `PROJECT_ARCHITECTURE.md`, `docs/TECHNICAL.md` | Changed. |
+| `lib/signal-store.ts`, `lib/market-store.ts`, `lib/signal-crdt-store.ts`, `lib/json-store.ts`, `lib/server-data-paths.ts`, `lib/wallet-nft-index-cache.ts`, `lib/explore-owners-cache.ts` | Ruled out: moving their CAS loops to an async driver is the dominant risk and is a separate change; this slice must not touch production write paths. |
+| `scripts/migrate-json-to-sqlite.ts` | Ruled out: the Postgres loader reuses the same parity check in a follow-up. |
+| `lib/__tests__/{signal-store-concurrency,json-store-concurrency,signal-crdt,collection-offer-book,royalty-split}.test.ts` | Ruled out: they run against SQLite and keep doing so; per-test schemas apply once a Postgres adapter exists. |
+| `tsconfig.json` | Ruled out: `scripts/` stays excluded; the new scripts are exercised by the drill. |
+
+## Rollback
+
+Unset `FEATURE_PHASE_172` and restart: `primary-only`; SQLite stays authoritative.
+
+## Follow-ups
+
+This is a partial slice. The issue's acceptance criteria that need a live testnet or hosted environment (end-to-end transaction proof, kill-the-primary resilience demo) are not claimed here; see `docs/TECHNICAL.md` for what the unit tests do cover.
diff --git a/docs/spikes/605-backup-dr-findings.md b/docs/spikes/605-backup-dr-findings.md
new file mode 100644
index 00000000..5ad80d18
--- /dev/null
+++ b/docs/spikes/605-backup-dr-findings.md
@@ -0,0 +1,38 @@
+# Spike 605 — Backup and Disaster-Recovery Drills for Sidecars, SQLite and the Secrets Map Findings (Issue #469)
+
+## What the issue claimed / what the code did
+
+- ~40 sidecar keys and `phase.sqlite3` had no snapshot or export code.
+- Each sidecar write is atomic (tmp+rename) but the set is not; a copy of a WAL database is not a consistent image.
+- On Vercel the data root is `os.tmpdir()`, so a cold start can wipe claims, follows, jobs and distributor health.
+- A restore running while `executeDistributorRefill` fires could double-spend.
+
+## Delivered slice (phase-171)
+
+- **Consistency**: `createSnapshot` holds every sidecar lock in sorted order (no lock-order deadlock) and takes the database with `VACUUM INTO`, a transactionally consistent copy with no WAL/SHM. A corrupt sidecar fails the snapshot instead of shipping.
+- **Verification**: manifest with sha256, size and per-table row counts; `verifySnapshot` also parses JSON and runs `integrity_check` and `foreign_key_check` (the migration script only skips dangling FKs with a warning).
+- **Restore**: verified first; stale WAL/SHM removed; tmp+rename; `.restore-in-progress` marker makes `/api/cron/faucet-health` answer `503` + `Retry-After: 60` (marker stale after 10 min so a crashed restore cannot wedge the cron).
+- **Secrets**: never in a snapshot. AES-256-GCM envelope with a scrypt key; wrong passphrase and tampering are indistinguishable and never echo plaintext; `diffSecretsMap` reports rotation drift by name.
+- **Drill**: `npm run backup:drill` snapshots, verifies, restores into a scratch directory and proves SQLite parity.
+
+## Affected files: changed or explicitly ruled out
+
+| File | Status |
+|---|---|
+| `lib/server-data-paths.ts` | Changed: root and key table exported. |
+| `lib/sqlite-db.ts`, `lib/json-store.ts` | Used unchanged: `getDb` and `withFileLock` provide the consistency points. |
+| `app/api/cron/faucet-health/route.ts` | Changed: restore-marker gate. |
+| `lib/signal-store.ts`, `lib/market-store.ts`, `lib/follow-store.ts`, `lib/notification-store.ts`, `lib/quest-snapshot.ts`, `lib/distributor-health-store.ts`, `lib/x402-dead-letter.ts`, `lib/generation-job-store.ts` | Ruled out: they are covered through the shared file table and database; per-store export formats are not needed for a byte-level snapshot. |
+| `scripts/migrate-json-to-sqlite.ts` | Ruled out: it only reads legacy JSON. |
+| `scripts/check-secret-rotation.ts` | Ruled out: rotation metadata is separate; drift is checked by `diffSecretsMap`. |
+| `app/api/world/[collection_id]/export/route.ts` | Ruled out: lore export is a user feature, not ops-state backup. |
+| `lib/api-observability.ts` | Ruled out: snapshot code logs nothing, so no secret can reach the log sink. |
+| `vercel.json` | Ruled out deliberately: a cron writing into `os.tmpdir()` would be false safety. A scheduled snapshot needs off-box storage first. |
+
+## Rollback
+
+Unset `FEATURE_PHASE_171` and restart: entry points throw `FLAG_DISABLED`; snapshots on disk are inert.
+
+## Follow-ups
+
+This is a partial slice. The issue's acceptance criteria that need a live testnet or hosted environment (end-to-end transaction proof, kill-the-primary resilience demo) are not claimed here; see `docs/TECHNICAL.md` for what the unit tests do cover.
diff --git a/docs/spikes/606-log-pipeline-findings.md b/docs/spikes/606-log-pipeline-findings.md
new file mode 100644
index 00000000..3b115c62
--- /dev/null
+++ b/docs/spikes/606-log-pipeline-findings.md
@@ -0,0 +1,34 @@
+# Spike 606 — Log Pipeline: Structured Logs to an External Sink with Sampling and Redaction Findings (Issue #470)
+
+## What the issue claimed / what the code did
+
+- `lib/api-observability.ts` wrote one JSON line per event to `console.*`; no batching, sampling, or external transport.
+- Ad-hoc `console.*` callers (faucet, Horizon, proxy, stellar, market, i18n) bypass it entirely.
+- Counters (`security-counters`, `signal-version-metrics`, bulkhead) are process-local and never exported.
+- Nothing stopped a `rawPayload`/`errorResult` serialization from carrying an `S...` seed into Vercel logs.
+
+## Delivered slice (phase-170)
+
+- **Redaction** (`redactLogValue`): `S...` seeds masked in every string; keys matching secret/seed/private/passphrase/password/authorization/api key/token/rawPayload/errorResult/signature are replaced wholesale (`signature_verified` and `token_id` are deliberately kept). Applied before sampling and before the console line is written.
+- **Sampling** (`shouldSampleLog`): warn/error always kept; info kept by FNV hash of the correlation id (default 10%), so a request is kept or dropped whole.
+- **Cardinality** (`toPipelineRecord`): only `level`, `event`, `route`, `method`, `status_class` are labels; numeric and long-hex route segments collapse to `:id`; wallets, CIDs, token ids, hashes and `lastFailure` stay in unindexed `fields`.
+- **Transport** (`LogPipeline`): `emit` is synchronous and never throws; bounded buffer (drop-oldest, counted); single-flight flush with an abort timeout; at-most-once (a failed batch is counted and dropped); the flush timer is `unref`'d. The sink must be `https://` (`validateServerEnv`).
+
+## Affected files: changed or explicitly ruled out
+
+| File | Status |
+|---|---|
+| `lib/api-observability.ts` | Changed: all lines go through `emitLog`. |
+| `lib/security-counters.ts` | Changed: six closed `log_pipeline_*` counters. |
+| `lib/env-validation.ts` | Changed: `LOG_SINK_URL` must be https. |
+| `app/api/faucet/route.ts`, `app/api/soroban-rpc/route.ts`, `app/api/x402/settle/route.ts`, `app/api/forge-agent/route.ts`, `app/api/cron/faucet-health/route.ts`, `lib/stellar.ts`, `lib/market-store.ts`, `lib/i18n-loader.ts`, `lib/generation-job-store.ts` | Ruled out of this slice: their raw `console.*` calls are migrated to the context logger in a follow-up so each route's behavior change is reviewable on its own. |
+| `lib/signal-version-metrics.ts`, `lib/horizon-bulkhead.ts`, `lib/webhook-alerts.ts` | Ruled out: counter export needs a decision on the sink's metrics shape. |
+| `scripts/check-secret-rotation.ts`, `middleware.ts`, `vercel.json` | Ruled out: no log-shaped change is required. A `waitUntil` flush belongs in `middleware.ts` once the runtime is confirmed. |
+
+## Rollback
+
+Unset `FEATURE_PHASE_170` and restart: console lines are written unchanged and the buffer is dropped.
+
+## Follow-ups
+
+This is a partial slice. The issue's acceptance criteria that need a live testnet or hosted environment (end-to-end transaction proof, kill-the-primary resilience demo) are not claimed here; see `docs/TECHNICAL.md` for what the unit tests do cover.
diff --git a/lib/__tests__/backup-snapshot.test.ts b/lib/__tests__/backup-snapshot.test.ts
new file mode 100644
index 00000000..2b3886f3
--- /dev/null
+++ b/lib/__tests__/backup-snapshot.test.ts
@@ -0,0 +1,170 @@
+import { mkdtemp, readFile, rm, utimes, writeFile } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import path from "node:path"
+import { DatabaseSync } from "node:sqlite"
+import { afterEach, beforeEach, describe, expect, it } from "vitest"
+import {
+  BackupError,
+  RESTORE_MARKER_NAME,
+  collectSecretsMap,
+  createSnapshot,
+  diffSecretsMap,
+  isRestoreInProgress,
+  openSecrets,
+  restoreSnapshot,
+  sealSecrets,
+  verifySnapshot,
+} from "@/lib/backup-snapshot"
+import { getSecurityCounter, resetSecurityCounters } from "@/lib/security-counters"
+import { serverDataJsonPath } from "@/lib/server-data-paths"
+import { getDb, resetDbForTests } from "@/lib/sqlite-db"
+
+const PASSPHRASE = "correct horse battery staple"
+const SECRET = "SDQXX5YXNQJ6X4Y3MXBXL4SVGSJPBIEUT3LOEBJTAI7XTMSEFAPKKMEJ"
+
+let dataDir: string
+let snapDir: string
+let restoreDir: string
+let previousDataDir: string | undefined
+
+function enable(on: boolean) {
+  if (on) {
+    process.env.NEXT_PUBLIC_FEATURE_PHASE_171 = "1"
+    process.env.FEATURE_PHASE_171 = "1"
+  } else {
+    delete process.env.NEXT_PUBLIC_FEATURE_PHASE_171
+    delete process.env.FEATURE_PHASE_171
+  }
+}
+
+function seedSignal(id: string) {
+  getDb()
+    .prepare(
+      "INSERT INTO signals (id, author_wallet, author_display, channel, title, body, created_at, signature) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+    )
+    .run(id, "GAAA", "a", "general", `title ${id}`, "body", Date.now(), "sig")
+}
+
+beforeEach(async () => {
+  enable(true)
+  resetSecurityCounters()
+  resetDbForTests()
+  dataDir = await mkdtemp(path.join(tmpdir(), "phase-backup-data-"))
+  snapDir = await mkdtemp(path.join(tmpdir(), "phase-backup-snap-"))
+  restoreDir = await mkdtemp(path.join(tmpdir(), "phase-backup-restore-"))
+  previousDataDir = process.env.PHASE_SERVER_DATA_DIR
+  process.env.PHASE_SERVER_DATA_DIR = dataDir
+})
+
+afterEach(async () => {
+  resetDbForTests()
+  if (previousDataDir === undefined) delete process.env.PHASE_SERVER_DATA_DIR
+  else process.env.PHASE_SERVER_DATA_DIR = previousDataDir
+  enable(false)
+  await Promise.all([dataDir, snapDir, restoreDir].map((d) => rm(d, { recursive: true, force: true })))
+})
+
+describe("phase-171 backup and DR snapshots", () => {
+  it("throws FLAG_DISABLED when the flag is off", async () => {
+    enable(false)
+    await expect(createSnapshot(snapDir)).rejects.toMatchObject({ code: "FLAG_DISABLED" })
+    expect(await isRestoreInProgress(dataDir)).toBe(false)
+  })
+
+  it("snapshots sidecars and sqlite, verifies, and restores with parity", async () => {
+    seedSignal("s1")
+    seedSignal("s2")
+    await writeFile(serverDataJsonPath("profileFollows"), JSON.stringify({ a: ["b"] }))
+    await writeFile(serverDataJsonPath("notifications"), JSON.stringify({ n: 1 }))
+
+    const manifest = await createSnapshot(snapDir, () => new Date("2026-01-01T00:00:00Z"))
+    expect(manifest.createdAt).toBe("2026-01-01T00:00:00.000Z")
+    expect(manifest.files.map((f) => f.name).sort()).toEqual(["notifications.json", "profile-follows.json"])
+    expect(manifest.sqlite?.tables.signals).toBe(2)
+
+    const verification = await verifySnapshot(snapDir)
+    expect(verification.problems).toEqual([])
+    expect(verification.ok).toBe(true)
+
+    await restoreSnapshot(snapDir, restoreDir)
+    expect(JSON.parse(await readFile(path.join(restoreDir, "profile-follows.json"), "utf8"))).toEqual({ a: ["b"] })
+    const restored = new DatabaseSync(path.join(restoreDir, "phase.sqlite3"), { readOnly: true })
+    expect((restored.prepare("SELECT COUNT(*) AS n FROM signals").get() as { n: number }).n).toBe(2)
+    restored.close()
+    expect(getSecurityCounter("backup_restore_completed")).toBe(1)
+    expect(await isRestoreInProgress(restoreDir)).toBe(false)
+  })
+
+  it("captures writes made while the snapshot runs only as a whole (no torn set)", async () => {
+    await writeFile(serverDataJsonPath("profileFollows"), JSON.stringify({ v: 1 }))
+    await writeFile(serverDataJsonPath("notifications"), JSON.stringify({ v: 1 }))
+    const manifest = await createSnapshot(snapDir)
+    for (const f of manifest.files) {
+      expect(JSON.parse(await readFile(path.join(snapDir, f.name), "utf8"))).toEqual({ v: 1 })
+    }
+  })
+
+  it("refuses to snapshot a corrupt sidecar", async () => {
+    await writeFile(serverDataJsonPath("profileFollows"), "{ torn")
+    await expect(createSnapshot(snapDir)).rejects.toThrow()
+  })
+
+  it("detects tampered and missing files and refuses to restore them", async () => {
+    await writeFile(serverDataJsonPath("profileFollows"), JSON.stringify({ a: 1 }))
+    await createSnapshot(snapDir)
+    await writeFile(path.join(snapDir, "profile-follows.json"), JSON.stringify({ a: 2 }))
+    const verification = await verifySnapshot(snapDir)
+    expect(verification.ok).toBe(false)
+    expect(verification.problems).toContain("profile-follows.json: checksum mismatch")
+    await expect(restoreSnapshot(snapDir, restoreDir)).rejects.toMatchObject({ code: "SNAPSHOT_CORRUPT" })
+    expect(getSecurityCounter("backup_snapshot_corrupt")).toBeGreaterThan(0)
+
+    await rm(path.join(snapDir, "manifest.json"))
+    expect((await verifySnapshot(snapDir)).problems).toContain("manifest missing or unreadable")
+  })
+
+  it("blocks a second restore while the marker is fresh and ignores a stale marker", async () => {
+    await createSnapshot(snapDir)
+    const marker = path.join(restoreDir, RESTORE_MARKER_NAME)
+    await writeFile(marker, "{}")
+    expect(await isRestoreInProgress(restoreDir)).toBe(true)
+    await expect(restoreSnapshot(snapDir, restoreDir)).rejects.toMatchObject({ code: "RESTORE_IN_PROGRESS" })
+    expect(getSecurityCounter("backup_restore_blocked")).toBe(1)
+
+    const old = new Date(Date.now() - 60 * 60 * 1000)
+    await utimes(marker, old, old)
+    expect(await isRestoreInProgress(restoreDir)).toBe(false)
+    await expect(restoreSnapshot(snapDir, restoreDir)).resolves.toBeDefined()
+  })
+
+  it("seals and opens secrets without leaking plaintext", async () => {
+    const map = collectSecretsMap({ ADMIN_SECRET_KEY: SECRET, CRON_SECRET: "cron-1", UNRELATED: "x" })
+    expect(Object.keys(map).sort()).toEqual(["ADMIN_SECRET_KEY", "CRON_SECRET"])
+    const sealed = sealSecrets(map, PASSPHRASE)
+    expect(JSON.stringify(sealed)).not.toContain(SECRET)
+    expect(sealed.names).toEqual(["ADMIN_SECRET_KEY", "CRON_SECRET"])
+    expect(openSecrets(sealed, PASSPHRASE)).toEqual(map)
+
+    let thrown: unknown
+    try {
+      openSecrets(sealed, "wrong passphrase value")
+    } catch (e) {
+      thrown = e
+    }
+    expect(thrown).toBeInstanceOf(BackupError)
+    expect((thrown as BackupError).code).toBe("SECRETS_UNSEAL_FAILED")
+    expect(String((thrown as BackupError).message)).not.toContain(SECRET)
+
+    const tampered = { ...sealed, ct: Buffer.from("zzzz").toString("base64") }
+    expect(() => openSecrets(tampered, PASSPHRASE)).toThrow(/could not be opened/)
+    expect(() => sealSecrets(map, "short")).toThrow(/at least 16/)
+  })
+
+  it("reports drift by secret name only", () => {
+    const diff = diffSecretsMap(
+      { ADMIN_SECRET_KEY: "old", CRON_SECRET: "same", PINATA_JWT: "gone" },
+      { ADMIN_SECRET_KEY: "new", CRON_SECRET: "same", PHASE_ADMIN_KEY: "extra" },
+    )
+    expect(diff).toEqual({ changed: ["ADMIN_SECRET_KEY"], missingInEnv: ["PINATA_JWT"], missingInBackup: ["PHASE_ADMIN_KEY"] })
+  })
+})
diff --git a/lib/__tests__/db-parity.test.ts b/lib/__tests__/db-parity.test.ts
new file mode 100644
index 00000000..587ed469
--- /dev/null
+++ b/lib/__tests__/db-parity.test.ts
@@ -0,0 +1,178 @@
+import { describe, it, beforeEach, expect } from "vitest"
+import { DatabaseSync } from "node:sqlite"
+import {
+  DualWriter,
+  assertSafeIdentifier,
+  buildPostgresCreateTable,
+  buildPostgresUpsert,
+  canonicalize,
+  compareSources,
+  perTestSchemaName,
+  resolveDualWriteMode,
+  sqliteParitySource,
+  sqliteTypeToPostgres,
+  toPositionalParams,
+  type ParitySource,
+  type WriteOp,
+} from "@/lib/db-parity"
+import { getSecurityCounter, resetSecurityCounters } from "@/lib/security-counters"
+
+const DDL = `CREATE TABLE listings (id TEXT PRIMARY KEY, price REAL NOT NULL, accepts_offers INTEGER NOT NULL DEFAULT 0, name TEXT);`
+
+function seeded(rows: Array<[string, number, number, string | null]>): DatabaseSync {
+  const db = new DatabaseSync(":memory:")
+  db.exec(DDL)
+  const stmt = db.prepare("INSERT INTO listings VALUES (?, ?, ?, ?)")
+  for (const r of rows) stmt.run(...r)
+  return db
+}
+
+function enable(on: boolean) {
+  if (on) {
+    process.env.NEXT_PUBLIC_FEATURE_PHASE_172 = "1"
+    process.env.FEATURE_PHASE_172 = "1"
+  } else {
+    delete process.env.NEXT_PUBLIC_FEATURE_PHASE_172
+    delete process.env.FEATURE_PHASE_172
+  }
+}
+
+const op = (table = "listings"): WriteOp => ({ table, sql: "x", params: [] })
+
+describe("phase-172 sqlite→postgres rehearsal", () => {
+  beforeEach(() => {
+    enable(true)
+    resetSecurityCounters()
+  })
+
+  it("translates placeholders, types and upserts", () => {
+    expect(toPositionalParams("SELECT * FROM t WHERE a = ? AND b = ?")).toBe("SELECT * FROM t WHERE a = $1 AND b = $2")
+    expect(sqliteTypeToPostgres("INTEGER")).toBe("BIGINT")
+    expect(sqliteTypeToPostgres("REAL")).toBe("DOUBLE PRECISION")
+    expect(sqliteTypeToPostgres("TEXT")).toBe("TEXT")
+    expect(buildPostgresUpsert("listings", ["id", "price", "name"], ["id"])).toBe(
+      'INSERT INTO "listings" ("id", "price", "name") VALUES ($1, $2, $3) ON CONFLICT ("id") DO UPDATE SET "price" = EXCLUDED."price", "name" = EXCLUDED."name"',
+    )
+    expect(buildPostgresUpsert("k", ["a"], ["a"])).toContain("DO NOTHING")
+    expect(() => buildPostgresUpsert("t", ["a"], [])).toThrow(/no primary key/)
+    const source = sqliteParitySource(seeded([]))
+    expect(buildPostgresCreateTable("listings", source.read("listings").columns, perTestSchemaName("a"))).toContain('PRIMARY KEY ("id")')
+  })
+
+  it("rejects unsafe identifiers and isolates per-test schemas", () => {
+    expect(() => assertSafeIdentifier('x"; DROP TABLE t;--')).toThrow()
+    expect(perTestSchemaName("a")).toMatch(/^t_[0-9a-f]{12}$/)
+    expect(perTestSchemaName("a")).not.toBe(perTestSchemaName("b"))
+    expect(perTestSchemaName("a")).toBe(perTestSchemaName("a"))
+  })
+
+  it("accepts Postgres-style string numerics and booleans but flags text coercion", () => {
+    expect(canonicalize("10.50", "REAL")).toEqual({ ok: true, value: 10.5 })
+    expect(canonicalize("7", "INTEGER")).toEqual({ ok: true, value: 7 })
+    expect(canonicalize(true, "INTEGER")).toEqual({ ok: true, value: 1 })
+    expect(canonicalize(7n, "INTEGER")).toEqual({ ok: true, value: 7 })
+    expect(canonicalize(7, "TEXT")).toEqual({ ok: false })
+    expect(canonicalize("abc", "INTEGER")).toEqual({ ok: false })
+  })
+
+  it("reports parity for identical databases regardless of row order", () => {
+    const a = seeded([["1", 1.5, 1, "x"], ["2", 2, 0, null]])
+    const b = seeded([["2", 2, 0, null], ["1", 1.5, 1, "x"]])
+    const report = compareSources(sqliteParitySource(a), sqliteParitySource(b))
+    expect(report.ok).toBe(true)
+    expect(getSecurityCounter("db_parity_mismatch")).toBe(0)
+  })
+
+  it("catches missing, extra, drifted rows that a count check passes", () => {
+    const a = seeded([["1", 1.5, 1, "x"], ["2", 2, 0, "y"]])
+    const b = seeded([["1", 1.5, 1, "CHANGED"], ["3", 2, 0, "y"]])
+    const report = compareSources(sqliteParitySource(a), sqliteParitySource(b))
+    expect(report.ok).toBe(false)
+    const t = report.tables[0]
+    expect(t.leftCount).toBe(t.rightCount)
+    expect(t.missingInRight).toEqual(['["2"]'])
+    expect(t.extraInRight).toEqual(['["3"]'])
+    expect(t.diffs).toEqual([{ pk: '["1"]', column: "name", kind: "value" }])
+    expect(getSecurityCounter("db_parity_mismatch")).toBe(1)
+  })
+
+  it("detects silent type coercion and tolerates NUMERIC string rendering", () => {
+    const left = sqliteParitySource(seeded([["1", 10.5, 1, "42"]]))
+    const base = left.read("listings")
+    const coerced: ParitySource = {
+      tables: () => ["listings"],
+      read: () => ({ columns: base.columns, rows: [{ id: "1", price: "10.50", accepts_offers: "1", name: 42 }] }),
+    }
+    const report = compareSources(left, coerced)
+    expect(report.tables[0].diffs).toEqual([{ pk: '["1"]', column: "name", kind: "type_coercion" }])
+    expect(report.ok).toBe(false)
+  })
+
+  it("flags tables missing on the target", () => {
+    const empty: ParitySource = { tables: () => [], read: () => ({ columns: [], rows: [] }) }
+    const report = compareSources(sqliteParitySource(seeded([])), empty)
+    expect(report.missingTables).toEqual(["listings"])
+    expect(report.ok).toBe(false)
+  })
+
+  it("resolves dual mode only with flag and URL", () => {
+    expect(resolveDualWriteMode({ PHASE_DATABASE_URL: "postgres://x" })).toBe("dual")
+    expect(resolveDualWriteMode({})).toBe("primary-only")
+    enable(false)
+    expect(resolveDualWriteMode({ PHASE_DATABASE_URL: "postgres://x" })).toBe("primary-only")
+  })
+
+  it("writes primary first and records secondary failure as drift without throwing", async () => {
+    const order: string[] = []
+    const writer = new DualWriter({
+      mode: "dual",
+      primary: async () => void order.push("primary"),
+      secondary: async () => {
+        order.push("secondary")
+        throw new Error("pg down")
+      },
+    })
+    await expect(writer.write(op())).resolves.toBeUndefined()
+    expect(order).toEqual(["primary", "secondary"])
+    expect(writer.driftLog().map((d) => d.reason)).toEqual(["error"])
+    expect(getSecurityCounter("db_dual_write_drift")).toBe(1)
+  })
+
+  it("never attempts the secondary when the primary fails", async () => {
+    let secondary = 0
+    const writer = new DualWriter({
+      mode: "dual",
+      primary: async () => {
+        throw new Error("sqlite busy")
+      },
+      secondary: async () => void (secondary += 1),
+    })
+    await expect(writer.write(op())).rejects.toThrow("sqlite busy")
+    expect(secondary).toBe(0)
+  })
+
+  it("times out and sheds under pool exhaustion instead of blocking", async () => {
+    const writer = new DualWriter({
+      mode: "dual",
+      maxSecondaryInflight: 1,
+      secondaryTimeoutMs: 20,
+      primary: async () => {},
+      secondary: () => new Promise(() => {}),
+    })
+    const first = writer.write(op("a"))
+    await writer.write(op("b"))
+    await first
+    expect(writer.driftLog().map((d) => [d.table, d.reason])).toEqual([["b", "shed"], ["a", "timeout"]])
+    expect(getSecurityCounter("db_dual_write_shed")).toBe(1)
+  })
+
+  it("rolls back to primary-only", async () => {
+    let secondary = 0
+    const writer = new DualWriter({ mode: "dual", primary: async () => {}, secondary: async () => void (secondary += 1) })
+    await writer.write(op())
+    writer.setMode("primary-only")
+    await writer.write(op())
+    expect(secondary).toBe(1)
+    expect(getSecurityCounter("db_dual_write_ok")).toBe(1)
+  })
+})
diff --git a/lib/__tests__/log-pipeline.test.ts b/lib/__tests__/log-pipeline.test.ts
new file mode 100644
index 00000000..10d81f93
--- /dev/null
+++ b/lib/__tests__/log-pipeline.test.ts
@@ -0,0 +1,204 @@
+import { describe, it, beforeEach } from "vitest"
+import * as assert from "node:assert/strict"
+import {
+  LogPipeline,
+  emitLog,
+  isValidLogSinkUrl,
+  normalizeRouteLabel,
+  redactLogValue,
+  resetLogPipelineForTests,
+  resolveLogPipelineConfig,
+  shouldSampleLog,
+  toPipelineRecord,
+  type LogPipelineConfig,
+  type PipelineLogPayload,
+} from "@/lib/log-pipeline"
+import { getSecurityCounter, resetSecurityCounters } from "@/lib/security-counters"
+import { validateServerEnv } from "@/lib/env-validation"
+
+const SECRET = "SDQXX5YXNQJ6X4Y3MXBXL4SVGSJPBIEUT3LOEBJTAI7XTMSEFAPKKMEJ"
+
+function enable(on: boolean) {
+  if (on) {
+    process.env.NEXT_PUBLIC_FEATURE_PHASE_170 = "1"
+    process.env.FEATURE_PHASE_170 = "1"
+  } else {
+    delete process.env.NEXT_PUBLIC_FEATURE_PHASE_170
+    delete process.env.FEATURE_PHASE_170
+  }
+}
+
+function payload(over: Partial<PipelineLogPayload> = {}): PipelineLogPayload {
+  return {
+    timestamp: "2026-01-01T00:00:00.000Z",
+    level: "info",
+    event: "request_completed",
+    route: "/api/faucet",
+    method: "POST",
+    correlationId: "c-1",
+    ...over,
+  }
+}
+
+function config(over: Partial<LogPipelineConfig> = {}): LogPipelineConfig {
+  return {
+    sinkUrl: "https://logs.example.com/ingest",
+    sinkToken: "sink-token",
+    infoSampleRate: 1,
+    batchSize: 3,
+    maxBuffer: 10,
+    flushIntervalMs: 100,
+    timeoutMs: 500,
+    ...over,
+  }
+}
+
+describe("phase-170 log pipeline", () => {
+  beforeEach(() => {
+    enable(true)
+    resetSecurityCounters()
+    resetLogPipelineForTests()
+    delete process.env.LOG_SINK_URL
+  })
+
+  it("redacts secret keys, sensitive key names, cycles and bigint", () => {
+    const cyclic: Record<string, unknown> = { a: 1 }
+    cyclic.self = cyclic
+    const out = redactLogValue({
+      note: `failed with ${SECRET}`,
+      rawPayload: { xdr: "AAAA" },
+      errorResult: "tx_failed",
+      authorization: "Bearer abc",
+      signature_verified: true,
+      token_id: 7,
+      nested: { adminSecret: "x", ok: "fine" },
+      cyclic,
+      big: 5n,
+    }) as Record<string, any>
+    assert.equal(out.note, "failed with [REDACTED_SECRET_KEY]")
+    assert.equal(out.rawPayload, "[REDACTED]")
+    assert.equal(out.errorResult, "[REDACTED]")
+    assert.equal(out.authorization, "[REDACTED]")
+    assert.equal(out.signature_verified, true)
+    assert.equal(out.token_id, 7)
+    assert.equal(out.nested.adminSecret, "[REDACTED]")
+    assert.equal(out.nested.ok, "fine")
+    assert.equal(out.cyclic.self, "[CIRCULAR]")
+    assert.equal(out.big, "5")
+    assert.ok(!JSON.stringify(out).includes(SECRET))
+  })
+
+  it("keeps wallets out of labels and collapses id segments", () => {
+    const record = toPipelineRecord(
+      payload({ route: "/api/world/123/export", status: 404, wallet: "GABC", hash: "deadbeef" }),
+    )
+    assert.deepEqual(Object.keys(record.labels).sort(), ["event", "level", "method", "route", "status_class"])
+    assert.equal(record.labels.route, "/api/world/:id/export")
+    assert.equal(record.labels.status_class, "4xx")
+    assert.equal(record.fields.wallet, "GABC")
+    assert.equal(normalizeRouteLabel("/api/ipfs/" + "a".repeat(32)), "/api/ipfs/:id")
+  })
+
+  it("always keeps warn/error and samples info deterministically per correlation id", () => {
+    assert.equal(shouldSampleLog("error", "x", 0), true)
+    assert.equal(shouldSampleLog("warn", "x", 0), true)
+    assert.equal(shouldSampleLog("info", "x", 1), true)
+    assert.equal(shouldSampleLog("info", "x", 0), false)
+    const first = shouldSampleLog("info", "req-42", 0.5)
+    for (let i = 0; i < 5; i += 1) assert.equal(shouldSampleLog("info", "req-42", 0.5), first)
+    const kept = Array.from({ length: 1000 }, (_, i) => shouldSampleLog("info", `r-${i}`, 0.1)).filter(Boolean).length
+    assert.ok(kept > 40 && kept < 200, `kept=${kept}`)
+  })
+
+  it("batches to the sink with bearer auth and flushes at batch size", async () => {
+    const calls: Array<{ url: string; init: RequestInit }> = []
+    const pipeline = new LogPipeline(config(), async (url, init) => {
+      calls.push({ url, init })
+      return { ok: true, status: 200 }
+    })
+    pipeline.emit(payload({ correlationId: "a" }))
+    pipeline.emit(payload({ correlationId: "b" }))
+    assert.equal(calls.length, 0)
+    pipeline.emit(payload({ correlationId: "c" }))
+    await pipeline.drain()
+    assert.equal(calls.length, 1)
+    assert.equal(calls[0].url, "https://logs.example.com/ingest")
+    assert.equal((calls[0].init.headers as Record<string, string>).authorization, "Bearer sink-token")
+    assert.equal(JSON.parse(calls[0].init.body as string).records.length, 3)
+    assert.equal(getSecurityCounter("log_pipeline_flush_ok"), 1)
+  })
+
+  it("never throws or blocks when the sink fails, and counts the loss", async () => {
+    const pipeline = new LogPipeline(config({ batchSize: 2 }), async () => {
+      throw new Error("sink down")
+    })
+    assert.doesNotThrow(() => {
+      pipeline.emit(payload({ correlationId: "a" }))
+      pipeline.emit(payload({ correlationId: "b" }))
+    })
+    await pipeline.drain()
+    assert.equal(getSecurityCounter("log_pipeline_flush_failed"), 1)
+    assert.equal(getSecurityCounter("log_pipeline_dropped"), 2)
+    assert.equal(pipeline.pending, 0)
+  })
+
+  it("treats a non-2xx sink response as a failed flush", async () => {
+    const pipeline = new LogPipeline(config({ batchSize: 1 }), async () => ({ ok: false, status: 503 }))
+    pipeline.emit(payload())
+    await pipeline.drain()
+    assert.equal(getSecurityCounter("log_pipeline_flush_failed"), 1)
+  })
+
+  it("bounds the buffer and drops oldest first", async () => {
+    const sent: string[] = []
+    const pipeline = new LogPipeline(config({ batchSize: 500, maxBuffer: 10, flushIntervalMs: 30_000 }), async (_u, init) => {
+      for (const r of JSON.parse(init.body as string).records) sent.push(r.correlationId)
+      return { ok: true, status: 200 }
+    })
+    for (let i = 0; i < 15; i += 1) pipeline.emit(payload({ correlationId: `r-${i}` }))
+    assert.equal(pipeline.pending, 10)
+    assert.equal(getSecurityCounter("log_pipeline_dropped"), 5)
+    await pipeline.drain()
+    assert.equal(sent[0], "r-5")
+  })
+
+  it("is a no-op without a sink URL and counts sampled-out info", () => {
+    const noSink = new LogPipeline(config({ sinkUrl: null }))
+    assert.equal(noSink.emit(payload()), false)
+    const sampled = new LogPipeline(config({ infoSampleRate: 0 }))
+    assert.equal(sampled.emit(payload()), false)
+    assert.equal(sampled.emit(payload({ level: "error" })), true)
+    assert.equal(getSecurityCounter("log_pipeline_sampled_out"), 1)
+  })
+
+  it("emitLog is the identity when the flag is off", () => {
+    enable(false)
+    const input = payload({ note: SECRET })
+    assert.equal(emitLog(input), input)
+    assert.equal(getSecurityCounter("log_pipeline_enqueued"), 0)
+  })
+
+  it("emitLog masks the console payload when the flag is on", () => {
+    const out = emitLog(payload({ note: SECRET, rawPayload: "x" }))
+    assert.equal(out.note, "[REDACTED_SECRET_KEY]")
+    assert.equal(out.rawPayload, "[REDACTED]")
+    assert.equal(out.event, "request_completed")
+    assert.equal(getSecurityCounter("log_pipeline_redacted"), 1)
+  })
+
+  it("resolves bounded config and validates the sink URL", () => {
+    const cfg = resolveLogPipelineConfig({
+      LOG_SINK_URL: " https://x.test/in ",
+      LOG_SAMPLE_RATE_INFO: "7",
+      LOG_SINK_BATCH_SIZE: "99999",
+    })
+    assert.equal(cfg.sinkUrl, "https://x.test/in")
+    assert.equal(cfg.infoSampleRate, 1)
+    assert.equal(cfg.batchSize, 500)
+    assert.equal(resolveLogPipelineConfig({}).infoSampleRate, 0.1)
+    assert.equal(isValidLogSinkUrl("http://x.test"), false)
+    assert.equal(isValidLogSinkUrl("not a url"), false)
+    assert.equal(validateServerEnv({ LOG_SINK_URL: "http://x.test" }).valid, false)
+    assert.equal(validateServerEnv({ LOG_SINK_URL: "https://x.test" }).valid, true)
+  })
+})
diff --git a/lib/__tests__/preview-isolation.test.ts b/lib/__tests__/preview-isolation.test.ts
new file mode 100644
index 00000000..717dc68e
--- /dev/null
+++ b/lib/__tests__/preview-isolation.test.ts
@@ -0,0 +1,121 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest"
+import path from "node:path"
+import { Keypair } from "@stellar/stellar-sdk"
+import { createHash } from "node:crypto"
+import {
+  checkPreviewIsolation,
+  isCronBlockedInPreview,
+  previewDataSubdir,
+  previewTrustsVercelCronHeader,
+  resolvePreviewId,
+} from "@/lib/preview-isolation"
+import { validatePhaseEnv } from "@/lib/env-validation"
+import { serverDataJsonPath } from "@/lib/server-data-paths"
+
+const KEEP = ["VERCEL", "VERCEL_ENV", "VERCEL_GIT_PULL_REQUEST_ID", "VERCEL_GIT_COMMIT_REF", "PHASE_SERVER_DATA_DIR", "CRON_SECRET", "PHASE_PROD_ISSUER", "PHASE_PROD_CRON_SECRET_SHA256", "PHASE_PROD_CONTRACT_IDS", "CLASSIC_LIQ_ISSUER_SECRET", "NEXT_PUBLIC_PHASE_PROTOCOL_ID"]
+let saved: Record<string, string | undefined> = {}
+
+function enable(on: boolean) {
+  if (on) {
+    process.env.NEXT_PUBLIC_FEATURE_PHASE_173 = "1"
+    process.env.FEATURE_PHASE_173 = "1"
+  } else {
+    delete process.env.NEXT_PUBLIC_FEATURE_PHASE_173
+    delete process.env.FEATURE_PHASE_173
+  }
+}
+
+beforeEach(() => {
+  saved = Object.fromEntries(KEEP.map((k) => [k, process.env[k]]))
+  for (const k of KEEP) delete process.env[k]
+  enable(true)
+})
+afterEach(() => {
+  for (const k of KEEP) {
+    if (saved[k] === undefined) delete process.env[k]
+    else process.env[k] = saved[k]
+  }
+  enable(false)
+})
+
+describe("phase-173 preview isolation", () => {
+  it("resolves per-preview ids only in previews, as one safe path segment", () => {
+    expect(resolvePreviewId({ VERCEL_ENV: "production", VERCEL_GIT_PULL_REQUEST_ID: "5" })).toBeNull()
+    expect(resolvePreviewId({})).toBeNull()
+    expect(resolvePreviewId({ VERCEL_ENV: "preview", VERCEL_GIT_PULL_REQUEST_ID: "467" })).toBe("pr-467")
+    const a = resolvePreviewId({ VERCEL_ENV: "preview", VERCEL_GIT_COMMIT_REF: "feat/a-b" })!
+    const b = resolvePreviewId({ VERCEL_ENV: "preview", VERCEL_GIT_COMMIT_REF: "feat/a/b" })!
+    expect(a).not.toBe(b)
+    const evil = resolvePreviewId({ VERCEL_ENV: "preview", VERCEL_GIT_COMMIT_REF: "../../etc/passwd" })!
+    expect(evil).toMatch(/^br-[a-z0-9-]+$/)
+    expect(evil).not.toContain("/")
+    expect(resolvePreviewId({ VERCEL_ENV: "preview", VERCEL_GIT_PULL_REQUEST_ID: "../x" })).toBe("preview-unknown")
+  })
+
+  it("namespaces the data root per preview and is inert when the flag is off", () => {
+    process.env.PHASE_SERVER_DATA_DIR = "/data"
+    process.env.VERCEL_ENV = "preview"
+    process.env.VERCEL_GIT_PULL_REQUEST_ID = "12"
+    expect(previewDataSubdir()).toBe("previews/pr-12")
+    expect(serverDataJsonPath("faucetClaims")).toBe(path.join("/data", "previews", "pr-12", "faucet-claims.json"))
+    process.env.VERCEL_GIT_PULL_REQUEST_ID = "13"
+    expect(serverDataJsonPath("faucetClaims")).not.toBe(path.join("/data", "previews", "pr-12", "faucet-claims.json"))
+
+    enable(false)
+    expect(serverDataJsonPath("faucetClaims")).toBe(path.join("/data", "faucet-claims.json"))
+    process.env.VERCEL_ENV = "production"
+    enable(true)
+    expect(serverDataJsonPath("faucetClaims")).toBe(path.join("/data", "faucet-claims.json"))
+  })
+
+  it("flags a preview that reuses the production issuer, cron secret, contract or mainnet", () => {
+    const issuer = Keypair.random()
+    const env = {
+      VERCEL_ENV: "preview",
+      PHASE_PROD_ISSUER: issuer.publicKey(),
+      CLASSIC_LIQ_ISSUER_SECRET: issuer.secret(),
+      PHASE_PROD_CRON_SECRET_SHA256: createHash("sha256").update("prod-cron").digest("hex"),
+      CRON_SECRET: "prod-cron",
+      PHASE_PROD_CONTRACT_IDS: "CBBCEDZ747IC3V4TWWCVGP33V74QM5T3MCPZV7DQ7CB6MA66YNWAC3DX",
+      NEXT_PUBLIC_PHASE_PROTOCOL_ID: "CBBCEDZ747IC3V4TWWCVGP33V74QM5T3MCPZV7DQ7CB6MA66YNWAC3DX",
+      STELLAR_NETWORK: "mainnet",
+    }
+    const codes = checkPreviewIsolation(env).map((v) => v.code).sort()
+    expect(codes).toEqual(["contract_matches_production", "cron_secret_matches_production", "issuer_matches_production", "mainnet_in_preview"])
+    const serialized = JSON.stringify(checkPreviewIsolation(env))
+    expect(serialized).not.toContain(issuer.secret())
+    expect(serialized).not.toContain("prod-cron")
+  })
+
+  it("passes an isolated preview and ignores production and flag-off", () => {
+    const prod = Keypair.random()
+    const isolated = { VERCEL_ENV: "preview", PHASE_PROD_ISSUER: prod.publicKey(), CLASSIC_LIQ_ISSUER_SECRET: Keypair.random().secret(), CRON_SECRET: "preview-cron" }
+    expect(checkPreviewIsolation(isolated)).toEqual([])
+    const shared = { ...isolated, CLASSIC_LIQ_ISSUER_SECRET: prod.secret() }
+    expect(checkPreviewIsolation({ ...shared, VERCEL_ENV: "production" })).toEqual([])
+    enable(false)
+    expect(checkPreviewIsolation(shared)).toEqual([])
+  })
+
+  it("blocks preview cron unless enabled and stops trusting the vercel-cron header", () => {
+    expect(isCronBlockedInPreview({ VERCEL_ENV: "preview" })).toBe(true)
+    expect(isCronBlockedInPreview({ VERCEL_ENV: "preview", PHASE_PREVIEW_CRON_ENABLED: "1" })).toBe(false)
+    expect(isCronBlockedInPreview({ VERCEL_ENV: "production" })).toBe(false)
+    expect(previewTrustsVercelCronHeader({ VERCEL_ENV: "preview" })).toBe(false)
+    expect(previewTrustsVercelCronHeader({ VERCEL_ENV: "production" })).toBe(true)
+    enable(false)
+    expect(isCronBlockedInPreview({ VERCEL_ENV: "preview" })).toBe(false)
+    expect(previewTrustsVercelCronHeader({ VERCEL_ENV: "preview" })).toBe(true)
+  })
+
+  it("surfaces violations through validatePhaseEnv", () => {
+    const prod = Keypair.random()
+    process.env.VERCEL_ENV = "preview"
+    process.env.PHASE_PROD_ISSUER = prod.publicKey()
+    process.env.CLASSIC_LIQ_ISSUER_SECRET = prod.secret()
+    const result = validatePhaseEnv()
+    expect(result.errors.some((e) => e.issue === "isolation_violation" && e.variable === "CLASSIC_LIQ_ISSUER_SECRET")).toBe(true)
+    enable(false)
+    expect(validatePhaseEnv().errors.some((e) => e.issue === "isolation_violation")).toBe(false)
+  })
+})
diff --git a/lib/api-observability.ts b/lib/api-observability.ts
index 6535f5e8..25d810d0 100644
--- a/lib/api-observability.ts
+++ b/lib/api-observability.ts
@@ -1,5 +1,6 @@
 import { randomUUID } from "node:crypto"
 import { NextRequest, NextResponse } from "next/server"
+import { emitLog } from "@/lib/log-pipeline"
 
 type ApiLogLevel = "info" | "warn" | "error"
 type ApiMetadata = Record<string, unknown>
@@ -29,7 +30,7 @@ export function createApiRequestContext(request: NextRequest, route: string) {
   const correlationId = getIncomingCorrelationId(request)
 
   function log(level: ApiLogLevel, event: string, metadata: ApiMetadata = {}) {
-    const payload = {
+    const payload = emitLog({
       timestamp: new Date().toISOString(),
       level,
       event,
@@ -37,7 +38,7 @@ export function createApiRequestContext(request: NextRequest, route: string) {
       method: request.method,
       correlationId,
       ...metadata,
-    }
+    })
     const line = JSON.stringify(payload)
 
     if (level === "error") console.error(line)
diff --git a/lib/backup-snapshot.ts b/lib/backup-snapshot.ts
new file mode 100644
index 00000000..93c2f483
--- /dev/null
+++ b/lib/backup-snapshot.ts
@@ -0,0 +1,367 @@
+/**
+ * Backup and disaster-recovery snapshots — partial slice of spike 605 (issue #469).
+ *
+ * Nothing used to snapshot the ~40 JSON sidecars or `phase.sqlite3`. Each
+ * sidecar write is atomic, but the *set* is not, and a raw copy of a WAL
+ * database is not a consistent image. This module captures all of them at one
+ * logical instant, proves the image is restorable, and restores it without
+ * racing the distributor refill cron.
+ *
+ * - Consistency: every sidecar path is locked (sorted order, so lockers cannot
+ *   deadlock) and the SQLite image is produced with `VACUUM INTO`, which is a
+ *   transactionally consistent copy that needs no WAL/SHM files.
+ * - Integrity: a manifest carries sha256 and size per file plus per-table row
+ *   counts. `verifySnapshot` re-hashes, parses JSON, and runs SQLite
+ *   `integrity_check` and `foreign_key_check`.
+ * - Restore safety: `restoreSnapshot` verifies first, writes through tmp+rename,
+ *   and holds a marker file for the duration. `isRestoreInProgress` lets the
+ *   faucet-health cron refuse to run mid-restore instead of double-spending.
+ * - Secrets: never written into a snapshot. `sealSecrets` produces an
+ *   AES-256-GCM envelope (scrypt-derived key) that is only ever decrypted by
+ *   `openSecrets`; errors never echo plaintext.
+ *
+ * Scope: library + drill script (`scripts/backup-drill.ts`). Off-box
+ * shipping to object storage, scheduled snapshots, and `age`/SOPS interop are
+ * spike 605 follow-ups; a cron that writes into Vercel's `os.tmpdir()` would be
+ * false safety, so none is added.
+ *
+ * Feature flag: phase-171. Rollback: unset → every entry point throws
+ * `FLAG_DISABLED` and `isRestoreInProgress` returns false; snapshots already on
+ * disk are inert.
+ */
+
+import { createCipheriv, createDecipheriv, createHash, randomBytes, scryptSync } from "node:crypto"
+import { mkdir, readFile, rename, rm, stat, writeFile } from "node:fs/promises"
+import path from "node:path"
+import { DatabaseSync } from "node:sqlite"
+import { isFeatureEnabled } from "@/lib/feature-flags"
+import { withFileLock } from "@/lib/json-store"
+import { incSecurityCounter } from "@/lib/security-counters"
+import { getDb } from "@/lib/sqlite-db"
+import {
+  serverDataFileKeys,
+  serverDataFileName,
+  serverDataJsonPath,
+  serverDataRoot,
+} from "@/lib/server-data-paths"
+
+export const SNAPSHOT_MANIFEST_NAME = "manifest.json"
+export const RESTORE_MARKER_NAME = ".restore-in-progress"
+/** A marker older than this is a crashed restore, not a live one. */
+export const RESTORE_MARKER_TTL_MS = 10 * 60 * 1000
+
+export type SnapshotFileEntry = { name: string; bytes: number; sha256: string }
+export type SnapshotManifest = {
+  version: 1
+  createdAt: string
+  files: SnapshotFileEntry[]
+  sqlite: (SnapshotFileEntry & { tables: Record<string, number> }) | null
+}
+
+export type BackupErrorCode =
+  | "FLAG_DISABLED"
+  | "SNAPSHOT_CORRUPT"
+  | "RESTORE_IN_PROGRESS"
+  | "SECRETS_UNSEAL_FAILED"
+  | "WEAK_PASSPHRASE"
+
+export class BackupError extends Error {
+  constructor(
+    readonly code: BackupErrorCode,
+    message: string,
+    readonly problems: string[] = [],
+  ) {
+    super(message)
+    this.name = "BackupError"
+  }
+}
+
+export function isBackupEnabled(): boolean {
+  return isFeatureEnabled("phase-171")
+}
+
+export function backupRollbackNote(): string {
+  return "Rollback phase-171: unset NEXT_PUBLIC_FEATURE_PHASE_171 / FEATURE_PHASE_171 or set to 0/false and restart. Snapshots on disk are inert; no migration to revert."
+}
+
+function assertEnabled(): void {
+  if (!isBackupEnabled()) throw new BackupError("FLAG_DISABLED", "Backup snapshots disabled (phase-171 flag off)")
+}
+
+function sha256(bytes: Buffer): string {
+  return createHash("sha256").update(bytes).digest("hex")
+}
+
+async function writeAtomic(target: string, bytes: Buffer | string): Promise<void> {
+  const tmp = `${target}.${process.pid}.${Date.now()}.tmp`
+  await writeFile(tmp, bytes)
+  await rename(tmp, target)
+}
+
+/** Holds every lock in sorted order, then runs `task` inside all of them. */
+function withAllLocks<T>(paths: string[], task: () => Promise<T>): Promise<T> {
+  const [head, ...rest] = [...paths].sort()
+  if (!head) return task()
+  return withFileLock(head, () => withAllLocks(rest, task))
+}
+
+function tableCounts(db: DatabaseSync): Record<string, number> {
+  const names = db
+    .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name")
+    .all() as Array<{ name: string }>
+  const out: Record<string, number> = {}
+  for (const { name } of names) {
+    const row = db.prepare(`SELECT COUNT(*) AS n FROM "${name.replace(/"/g, '""')}"`).get() as { n: number }
+    out[name] = Number(row.n)
+  }
+  return out
+}
+
+async function exists(file: string): Promise<boolean> {
+  try {
+    await stat(file)
+    return true
+  } catch {
+    return false
+  }
+}
+
+/** Captures all sidecars and the SQLite database at one logical instant. */
+export async function createSnapshot(outDir: string, now: () => Date = () => new Date()): Promise<SnapshotManifest> {
+  assertEnabled()
+  const sidecars = serverDataFileKeys().filter((key) => key !== "sqliteDb")
+  const lockPaths = sidecars.map((key) => serverDataJsonPath(key))
+  await mkdir(outDir, { recursive: true })
+
+  const manifest = await withAllLocks(lockPaths, async () => {
+    const files: SnapshotFileEntry[] = []
+    for (const key of sidecars) {
+      const source = serverDataJsonPath(key)
+      if (!(await exists(source))) continue
+      const bytes = await readFile(source)
+      JSON.parse(bytes.toString("utf8")) // a torn or corrupt sidecar must fail the snapshot, not ship
+      const name = serverDataFileName(key)
+      await writeAtomic(path.join(outDir, name), bytes)
+      files.push({ name, bytes: bytes.length, sha256: sha256(bytes) })
+    }
+
+    const sqliteName = serverDataFileName("sqliteDb")
+    const target = path.join(outDir, sqliteName)
+    await rm(target, { force: true })
+    const db = getDb()
+    db.exec(`VACUUM INTO '${target.replace(/'/g, "''")}'`)
+    const sqliteBytes = await readFile(target)
+    const sqlite = {
+      name: sqliteName,
+      bytes: sqliteBytes.length,
+      sha256: sha256(sqliteBytes),
+      tables: tableCounts(db),
+    }
+    return { version: 1 as const, createdAt: now().toISOString(), files, sqlite }
+  })
+
+  await writeAtomic(path.join(outDir, SNAPSHOT_MANIFEST_NAME), JSON.stringify(manifest, null, 2))
+  incSecurityCounter("backup_snapshot_created")
+  return manifest
+}
+
+export type SnapshotVerification = { ok: boolean; problems: string[]; manifest: SnapshotManifest | null }
+
+export async function verifySnapshot(dir: string): Promise<SnapshotVerification> {
+  assertEnabled()
+  const problems: string[] = []
+  let manifest: SnapshotManifest | null = null
+  try {
+    manifest = JSON.parse(await readFile(path.join(dir, SNAPSHOT_MANIFEST_NAME), "utf8")) as SnapshotManifest
+    if (manifest.version !== 1) problems.push(`unsupported manifest version ${String(manifest.version)}`)
+  } catch {
+    problems.push("manifest missing or unreadable")
+    incSecurityCounter("backup_snapshot_corrupt")
+    return { ok: false, problems, manifest: null }
+  }
+
+  const entries = manifest.sqlite ? [...manifest.files, manifest.sqlite] : manifest.files
+  for (const entry of entries) {
+    let bytes: Buffer
+    try {
+      bytes = await readFile(path.join(dir, entry.name))
+    } catch {
+      problems.push(`${entry.name}: missing`)
+      continue
+    }
+    if (bytes.length !== entry.bytes || sha256(bytes) !== entry.sha256) {
+      problems.push(`${entry.name}: checksum mismatch`)
+      continue
+    }
+    if (entry.name.endsWith(".json")) {
+      try {
+        JSON.parse(bytes.toString("utf8"))
+      } catch {
+        problems.push(`${entry.name}: invalid JSON`)
+      }
+    }
+  }
+
+  if (manifest.sqlite && !problems.some((p) => p.startsWith(manifest!.sqlite!.name))) {
+    let db: DatabaseSync | null = null
+    try {
+      db = new DatabaseSync(path.join(dir, manifest.sqlite.name), { readOnly: true })
+      const integrity = db.prepare("PRAGMA integrity_check").all() as Array<{ integrity_check: string }>
+      if (integrity.length !== 1 || integrity[0].integrity_check !== "ok") problems.push("sqlite: integrity_check failed")
+      const dangling = db.prepare("PRAGMA foreign_key_check").all()
+      if (dangling.length > 0) problems.push(`sqlite: ${dangling.length} dangling foreign key row(s)`)
+      const counts = tableCounts(db)
+      for (const [table, expected] of Object.entries(manifest.sqlite.tables)) {
+        if (counts[table] !== expected) problems.push(`sqlite: table ${table} has ${counts[table] ?? "no"} rows, expected ${expected}`)
+      }
+    } catch (error) {
+      problems.push(`sqlite: unreadable (${error instanceof Error ? error.message : "unknown"})`)
+    } finally {
+      db?.close()
+    }
+  }
+
+  incSecurityCounter(problems.length === 0 ? "backup_snapshot_verified" : "backup_snapshot_corrupt")
+  return { ok: problems.length === 0, problems, manifest }
+}
+
+/** True while a restore holds the marker in `root`; stale markers are ignored. */
+export async function isRestoreInProgress(root: string = serverDataRoot(), now: number = Date.now()): Promise<boolean> {
+  if (!isBackupEnabled()) return false
+  try {
+    const info = await stat(path.join(root, RESTORE_MARKER_NAME))
+    return now - info.mtimeMs < RESTORE_MARKER_TTL_MS
+  } catch {
+    return false
+  }
+}
+
+/**
+ * Restores a verified snapshot into `targetDir` (default: the live data root).
+ * The caller must have closed any open SQLite connection on the target first —
+ * restore replaces the file underneath it.
+ */
+export async function restoreSnapshot(dir: string, targetDir: string = serverDataRoot()): Promise<SnapshotManifest> {
+  assertEnabled()
+  const verification = await verifySnapshot(dir)
+  if (!verification.ok || !verification.manifest) {
+    throw new BackupError("SNAPSHOT_CORRUPT", "Refusing to restore an unverified snapshot", verification.problems)
+  }
+  if (await isRestoreInProgress(targetDir)) {
+    incSecurityCounter("backup_restore_blocked")
+    throw new BackupError("RESTORE_IN_PROGRESS", "Another restore holds the marker for this data root")
+  }
+
+  const manifest = verification.manifest
+  await mkdir(targetDir, { recursive: true })
+  const marker = path.join(targetDir, RESTORE_MARKER_NAME)
+  await writeFile(marker, JSON.stringify({ startedAt: new Date().toISOString(), pid: process.pid }))
+  try {
+    for (const entry of manifest.files) {
+      await writeAtomic(path.join(targetDir, entry.name), await readFile(path.join(dir, entry.name)))
+    }
+    if (manifest.sqlite) {
+      const target = path.join(targetDir, manifest.sqlite.name)
+      // A stale WAL/SHM pair would be replayed over the restored image.
+      await rm(`${target}-wal`, { force: true })
+      await rm(`${target}-shm`, { force: true })
+      await writeAtomic(target, await readFile(path.join(dir, manifest.sqlite.name)))
+    }
+  } finally {
+    await rm(marker, { force: true })
+  }
+  incSecurityCounter("backup_restore_completed")
+  return manifest
+}
+
+// ── Sealed secrets envelope ────────────────────────────────────────────────
+
+/** Secret env names covered by the secrets map; values are never logged. */
+export const BACKUP_SECRET_ENV_NAMES = [
+  "ADMIN_SECRET_KEY",
+  "FAUCET_DISTRIBUTOR_SECRET_KEY",
+  "CLASSIC_LIQ_ISSUER_SECRET",
+  "PINATA_JWT",
+  "PINATA_API_JWT",
+  "NANOBANANA_API_KEY",
+  "NANOBANANA_WEBHOOK_SECRET",
+  "CRON_SECRET",
+  "PHASE_ADMIN_KEY",
+  "SEP10_SERVER_SECRET",
+] as const
+
+export type SealedSecrets = {
+  v: 1
+  kdf: "scrypt"
+  names: string[]
+  salt: string
+  iv: string
+  tag: string
+  ct: string
+}
+
+const MIN_PASSPHRASE = 16
+
+function deriveKey(passphrase: string, salt: Buffer): Buffer {
+  if (passphrase.length < MIN_PASSPHRASE) {
+    throw new BackupError("WEAK_PASSPHRASE", `Backup passphrase must be at least ${MIN_PASSPHRASE} characters`)
+  }
+  return scryptSync(passphrase, salt, 32)
+}
+
+export function collectSecretsMap(env: Record<string, string | undefined> = process.env): Record<string, string> {
+  const out: Record<string, string> = {}
+  for (const name of BACKUP_SECRET_ENV_NAMES) {
+    const value = env[name]?.trim()
+    if (value) out[name] = value
+  }
+  return out
+}
+
+export function sealSecrets(secrets: Record<string, string>, passphrase: string): SealedSecrets {
+  assertEnabled()
+  const salt = randomBytes(16)
+  const iv = randomBytes(12)
+  const cipher = createCipheriv("aes-256-gcm", deriveKey(passphrase, salt), iv)
+  const ct = Buffer.concat([cipher.update(JSON.stringify(secrets), "utf8"), cipher.final()])
+  return {
+    v: 1,
+    kdf: "scrypt",
+    names: Object.keys(secrets).sort(),
+    salt: salt.toString("base64"),
+    iv: iv.toString("base64"),
+    tag: cipher.getAuthTag().toString("base64"),
+    ct: ct.toString("base64"),
+  }
+}
+
+export function openSecrets(sealed: SealedSecrets, passphrase: string): Record<string, string> {
+  assertEnabled()
+  try {
+    const decipher = createDecipheriv(
+      "aes-256-gcm",
+      deriveKey(passphrase, Buffer.from(sealed.salt, "base64")),
+      Buffer.from(sealed.iv, "base64"),
+    )
+    decipher.setAuthTag(Buffer.from(sealed.tag, "base64"))
+    const plain = Buffer.concat([decipher.update(Buffer.from(sealed.ct, "base64")), decipher.final()])
+    return JSON.parse(plain.toString("utf8")) as Record<string, string>
+  } catch (error) {
+    if (error instanceof BackupError) throw error
+    // Wrong passphrase and tampering are indistinguishable by design.
+    throw new BackupError("SECRETS_UNSEAL_FAILED", "Sealed secrets could not be opened")
+  }
+}
+
+/** Names only: which secrets differ between a restored map and the live env. */
+export function diffSecretsMap(
+  restored: Record<string, string>,
+  env: Record<string, string | undefined> = process.env,
+): { changed: string[]; missingInEnv: string[]; missingInBackup: string[] } {
+  const live = collectSecretsMap(env)
+  return {
+    changed: Object.keys(restored).filter((name) => name in live && live[name] !== restored[name]).sort(),
+    missingInEnv: Object.keys(restored).filter((name) => !(name in live)).sort(),
+    missingInBackup: Object.keys(live).filter((name) => !(name in restored)).sort(),
+  }
+}
diff --git a/lib/db-parity.ts b/lib/db-parity.ts
new file mode 100644
index 00000000..247d15ea
--- /dev/null
+++ b/lib/db-parity.ts
@@ -0,0 +1,335 @@
+/**
+ * SQLite → Postgres migration rehearsal — partial slice of spike 604 (issue #468).
+ *
+ * There is no Postgres dialect, no checksum and no down-migration; the only
+ * precedent compares row counts, which cannot see silent type coercion (REAL vs
+ * NUMERIC, INTEGER vs BOOLEAN, BIGINT returned as a string). This module is the
+ * driver-agnostic half of a rehearsal: dialect translation, a typed per-table
+ * parity comparison, and a dual-writer whose ordering and failure semantics are
+ * explicit. No Postgres driver is added; a follow-up implements `ParitySource`
+ * and `DualWriteSink` for `pg`.
+ *
+ * - Ordering: the primary (SQLite) is written first and is the crash winner. A
+ *   primary failure never reaches the secondary.
+ * - The secondary is best-effort and bounded: failures, timeouts, and pool
+ *   exhaustion (in-flight cap) are recorded as drift and shed, never thrown, so
+ *   a slow Postgres degrades to SQLite-only instead of failing requests.
+ * - Rollback is `setMode("primary-only")`; SQLite stays authoritative.
+ * - Parity is by declared column type, so a TEXT column that comes back numeric
+ *   is reported as `type_coercion` rather than passing on count alone.
+ *
+ * Scope: dialect helpers, parity, dual-writer, per-test schema naming. Async
+ * re-plumbing of the CAS retry loops, advisory-lock replacement of `fileQueues`,
+ * and pool storm testing are spike 604 follow-ups.
+ *
+ * Feature flag: phase-172. Rollback: unset → `resolveDualWriteMode` returns
+ * `primary-only`; nothing here is called by production paths.
+ */
+
+import { createHash } from "node:crypto"
+import type { DatabaseSync } from "node:sqlite"
+import { isFeatureEnabled } from "@/lib/feature-flags"
+import { incSecurityCounter } from "@/lib/security-counters"
+
+type Env = Record<string, string | undefined>
+export type Row = Record<string, unknown>
+
+export function isDbParityEnabled(): boolean {
+  return isFeatureEnabled("phase-172")
+}
+
+export function dbParityRollbackNote(): string {
+  return "Rollback phase-172: unset NEXT_PUBLIC_FEATURE_PHASE_172 / FEATURE_PHASE_172 or set to 0/false and restart. SQLite remains authoritative; no data migration to revert."
+}
+
+export type DualWriteMode = "primary-only" | "dual"
+
+/** Dual-write needs both the flag and an explicit Postgres URL. */
+export function resolveDualWriteMode(env: Env = process.env): DualWriteMode {
+  return isDbParityEnabled() && env.PHASE_DATABASE_URL?.trim() ? "dual" : "primary-only"
+}
+
+// ── Dialect ────────────────────────────────────────────────────────────────
+
+const IDENT_RE = /^[A-Za-z_][A-Za-z0-9_]{0,62}$/
+
+export function assertSafeIdentifier(name: string): string {
+  if (!IDENT_RE.test(name)) throw new Error(`Unsafe SQL identifier: ${name}`)
+  return name
+}
+
+const quote = (name: string) => `"${assertSafeIdentifier(name)}"`
+
+/** `?` → `$1..$n`. Strings are not scanned; the stores never inline `?` in literals. */
+export function toPositionalParams(sql: string): string {
+  let n = 0
+  return sql.replace(/\?/g, () => `$${++n}`)
+}
+
+/**
+ * Integer flag columns stay BIGINT rather than becoming BOOLEAN so a parity run
+ * compares like with like; a BOOLEAN conversion is its own migration.
+ */
+export function sqliteTypeToPostgres(declared: string): string {
+  const t = declared.trim().toUpperCase()
+  if (t.startsWith("INT")) return "BIGINT"
+  if (t === "REAL" || t.startsWith("FLOAT") || t.startsWith("DOUBLE")) return "DOUBLE PRECISION"
+  if (t === "BLOB") return "BYTEA"
+  return "TEXT"
+}
+
+export type ColumnInfo = { name: string; type: string; pk: boolean; notNull: boolean }
+
+export function buildPostgresCreateTable(table: string, columns: ColumnInfo[], schema?: string): string {
+  const target = schema ? `${quote(schema)}.${quote(table)}` : quote(table)
+  const defs = columns.map(
+    (c) => `${quote(c.name)} ${sqliteTypeToPostgres(c.type)}${c.notNull || c.pk ? " NOT NULL" : ""}`,
+  )
+  const pk = columns.filter((c) => c.pk).map((c) => quote(c.name))
+  if (pk.length > 0) defs.push(`PRIMARY KEY (${pk.join(", ")})`)
+  return `CREATE TABLE IF NOT EXISTS ${target} (${defs.join(", ")})`
+}
+
+/** Replacement for `INSERT OR REPLACE`: explicit conflict target, all non-key columns updated. */
+export function buildPostgresUpsert(table: string, columns: string[], pk: string[]): string {
+  if (pk.length === 0) throw new Error(`Table ${table} has no primary key; upsert is undefined`)
+  const cols = columns.map(quote).join(", ")
+  const params = columns.map((_, i) => `$${i + 1}`).join(", ")
+  const updates = columns.filter((c) => !pk.includes(c)).map((c) => `${quote(c)} = EXCLUDED.${quote(c)}`)
+  const conflict = pk.map(quote).join(", ")
+  const action = updates.length > 0 ? `DO UPDATE SET ${updates.join(", ")}` : "DO NOTHING"
+  return `INSERT INTO ${quote(table)} (${cols}) VALUES (${params}) ON CONFLICT (${conflict}) ${action}`
+}
+
+/** One isolated schema per test run, so a shared Postgres has no cross-test pollution. */
+export function perTestSchemaName(testId: string): string {
+  return `t_${createHash("sha256").update(testId).digest("hex").slice(0, 12)}`
+}
+
+// ── Parity ─────────────────────────────────────────────────────────────────
+
+export type TableData = { columns: ColumnInfo[]; rows: Row[] }
+export type ParitySource = { tables(): string[]; read(table: string): TableData }
+
+export function sqliteParitySource(db: DatabaseSync): ParitySource {
+  return {
+    tables: () =>
+      (
+        db
+          .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name")
+          .all() as Array<{ name: string }>
+      ).map((r) => r.name),
+    read: (table) => {
+      const info = db.prepare(`PRAGMA table_info(${quote(table)})`).all() as Array<{
+        name: string
+        type: string
+        pk: number
+        notnull: number
+      }>
+      return {
+        columns: info.map((c) => ({ name: c.name, type: c.type, pk: c.pk > 0, notNull: c.notnull === 1 })),
+        rows: db.prepare(`SELECT * FROM ${quote(table)}`).all() as Row[],
+      }
+    },
+  }
+}
+
+type Kind = "numeric" | "text" | "other"
+const kindOf = (declared: string): Kind => {
+  const t = sqliteTypeToPostgres(declared)
+  return t === "BIGINT" || t === "DOUBLE PRECISION" ? "numeric" : t === "TEXT" ? "text" : "other"
+}
+
+export type Canonical = { ok: true; value: string | number | null } | { ok: false }
+
+/** Canonical form by declared type; `ok: false` means the value is the wrong JS type for the column. */
+export function canonicalize(value: unknown, declared: string): Canonical {
+  if (value === null || value === undefined) return { ok: true, value: null }
+  const kind = kindOf(declared)
+  if (kind === "numeric") {
+    if (typeof value === "boolean") return { ok: true, value: value ? 1 : 0 }
+    if (typeof value === "bigint") return { ok: true, value: Number(value) }
+    if (typeof value === "number") return { ok: true, value }
+    // Postgres drivers return BIGINT and NUMERIC as decimal strings.
+    if (typeof value === "string" && /^-?\d+(\.\d+)?$/.test(value)) return { ok: true, value: Number(value) }
+    return { ok: false }
+  }
+  if (kind === "text") return typeof value === "string" ? { ok: true, value } : { ok: false }
+  return { ok: true, value: String(value) }
+}
+
+const approxEqual = (a: number, b: number) => a === b || Math.abs(a - b) <= 1e-9 * Math.max(1, Math.abs(a), Math.abs(b))
+
+const pkKey = (row: Row, pk: string[]) => JSON.stringify(pk.map((c) => row[c] ?? null))
+
+export type ParityDiff = { pk: string; column: string; kind: "value" | "type_coercion" }
+export type TableParity = {
+  table: string
+  leftCount: number
+  rightCount: number
+  missingInRight: string[]
+  extraInRight: string[]
+  diffs: ParityDiff[]
+  diffsTruncated: boolean
+  leftChecksum: string
+  rightChecksum: string
+  match: boolean
+}
+
+const MAX_REPORTED = 50
+
+export function tableChecksum(data: TableData): string {
+  const pk = data.columns.filter((c) => c.pk).map((c) => c.name)
+  const keyed = data.rows
+    .map((row) => ({
+      key: pk.length > 0 ? pkKey(row, pk) : JSON.stringify(row),
+      cells: data.columns.map((c) => {
+        const canon = canonicalize(row[c.name], c.type)
+        return canon.ok ? canon.value : `__type_error__:${typeof row[c.name]}`
+      }),
+    }))
+    .sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
+  return createHash("sha256").update(JSON.stringify(keyed)).digest("hex")
+}
+
+export function compareTable(table: string, left: TableData, right: TableData): TableParity {
+  const columns = left.columns
+  const pk = columns.filter((c) => c.pk).map((c) => c.name)
+  const rightByKey = new Map(right.rows.map((row) => [pkKey(row, pk), row]))
+  const leftKeys = new Set<string>()
+  const diffs: ParityDiff[] = []
+  const missingInRight: string[] = []
+
+  for (const lrow of left.rows) {
+    const key = pkKey(lrow, pk)
+    leftKeys.add(key)
+    const rrow = rightByKey.get(key)
+    if (!rrow) {
+      missingInRight.push(key)
+      continue
+    }
+    for (const col of columns) {
+      const l = canonicalize(lrow[col.name], col.type)
+      const r = canonicalize(rrow[col.name], col.type)
+      if (!l.ok || !r.ok) {
+        if (l.ok !== r.ok) diffs.push({ pk: key, column: col.name, kind: "type_coercion" })
+        else if (!l.ok && lrow[col.name] !== rrow[col.name]) diffs.push({ pk: key, column: col.name, kind: "value" })
+        continue
+      }
+      const same =
+        typeof l.value === "number" && typeof r.value === "number" ? approxEqual(l.value, r.value) : l.value === r.value
+      if (!same) diffs.push({ pk: key, column: col.name, kind: "value" })
+    }
+  }
+  const extraInRight = [...rightByKey.keys()].filter((k) => !leftKeys.has(k))
+  const leftChecksum = tableChecksum(left)
+  const rightChecksum = tableChecksum({ columns, rows: right.rows })
+  return {
+    table,
+    leftCount: left.rows.length,
+    rightCount: right.rows.length,
+    missingInRight: missingInRight.slice(0, MAX_REPORTED),
+    extraInRight: extraInRight.slice(0, MAX_REPORTED),
+    diffs: diffs.slice(0, MAX_REPORTED),
+    diffsTruncated: diffs.length > MAX_REPORTED || missingInRight.length > MAX_REPORTED || extraInRight.length > MAX_REPORTED,
+    leftChecksum,
+    rightChecksum,
+    match: missingInRight.length === 0 && extraInRight.length === 0 && diffs.length === 0 && leftChecksum === rightChecksum,
+  }
+}
+
+export type ParityReport = { ok: boolean; tables: TableParity[]; missingTables: string[] }
+
+export function compareSources(left: ParitySource, right: ParitySource): ParityReport {
+  const rightTables = new Set(right.tables())
+  const tables: TableParity[] = []
+  const missingTables: string[] = []
+  for (const table of left.tables()) {
+    if (!rightTables.has(table)) {
+      missingTables.push(table)
+      continue
+    }
+    tables.push(compareTable(table, left.read(table), right.read(table)))
+  }
+  const ok = missingTables.length === 0 && tables.every((t) => t.match)
+  incSecurityCounter("db_parity_checked")
+  if (!ok) incSecurityCounter("db_parity_mismatch")
+  return { ok, tables, missingTables }
+}
+
+// ── Dual-write ─────────────────────────────────────────────────────────────
+
+export type WriteOp = { table: string; sql: string; params: unknown[] }
+export type DualWriteSink = (op: WriteOp) => Promise<void>
+export type DriftEntry = { table: string; reason: "error" | "timeout" | "shed"; at: number }
+
+export type DualWriterOptions = {
+  primary: DualWriteSink
+  secondary: DualWriteSink
+  mode?: DualWriteMode
+  /** Pool-exhaustion guard: beyond this many in-flight secondary writes, shed. */
+  maxSecondaryInflight?: number
+  secondaryTimeoutMs?: number
+  maxDriftEntries?: number
+}
+
+export class DualWriter {
+  private mode: DualWriteMode
+  private inflight = 0
+  private readonly drift: DriftEntry[] = []
+  private readonly maxInflight: number
+  private readonly timeoutMs: number
+  private readonly maxDrift: number
+
+  constructor(private readonly options: DualWriterOptions) {
+    this.mode = options.mode ?? resolveDualWriteMode()
+    this.maxInflight = options.maxSecondaryInflight ?? 5
+    this.timeoutMs = options.secondaryTimeoutMs ?? 1500
+    this.maxDrift = options.maxDriftEntries ?? 1000
+  }
+
+  setMode(mode: DualWriteMode): void {
+    this.mode = mode
+  }
+
+  getMode(): DualWriteMode {
+    return this.mode
+  }
+
+  /** Tables that diverged since start; the input to a re-sync before cutover. */
+  driftLog(): readonly DriftEntry[] {
+    return this.drift
+  }
+
+  private recordDrift(table: string, reason: DriftEntry["reason"]): void {
+    incSecurityCounter(reason === "shed" ? "db_dual_write_shed" : "db_dual_write_drift")
+    this.drift.push({ table, reason, at: Date.now() })
+    if (this.drift.length > this.maxDrift) this.drift.shift()
+  }
+
+  /** Throws only if the primary fails — and then the secondary is never attempted. */
+  async write(op: WriteOp): Promise<void> {
+    await this.options.primary(op)
+    if (this.mode !== "dual") return
+    if (this.inflight >= this.maxInflight) {
+      this.recordDrift(op.table, "shed")
+      return
+    }
+    this.inflight += 1
+    let timer: ReturnType<typeof setTimeout> | undefined
+    try {
+      await Promise.race([
+        this.options.secondary(op),
+        new Promise<never>((_, reject) => {
+          timer = setTimeout(() => reject(new Error("timeout")), this.timeoutMs)
+        }),
+      ])
+      incSecurityCounter("db_dual_write_ok")
+    } catch (error) {
+      this.recordDrift(op.table, error instanceof Error && error.message === "timeout" ? "timeout" : "error")
+    } finally {
+      if (timer) clearTimeout(timer)
+      this.inflight -= 1
+    }
+  }
+}
diff --git a/lib/env-validation.ts b/lib/env-validation.ts
index 1dec9b14..df0364db 100644
--- a/lib/env-validation.ts
+++ b/lib/env-validation.ts
@@ -5,10 +5,12 @@
 
 import { StrKey } from "@stellar/stellar-sdk"
 import { z } from "zod"
+import { checkPreviewIsolation } from "@/lib/preview-isolation"
+import { incSecurityCounter } from "@/lib/security-counters"
 
 export type EnvValidationError = {
   variable: string
-  issue: "missing" | "invalid_format" | "wrong_key_type"
+  issue: "missing" | "invalid_format" | "wrong_key_type" | "isolation_violation"
   message: string
   hint: string
 }
@@ -18,6 +20,15 @@ export type EnvValidationResult = {
   errors: EnvValidationError[]
 }
 
+/** The log sink must be https; plain http would ship records in the clear. */
+export function isValidLogSinkUrl(value: string): boolean {
+  try {
+    return new URL(value.trim()).protocol === "https:"
+  } catch {
+    return false
+  }
+}
+
 function isValidContractId(value: string): boolean {
   return StrKey.isValidContract(value)
 }
@@ -171,6 +182,10 @@ export const serverEnvSchema = z.object({
     { message: "STELLAR_RPC_URL must be a valid HTTP/HTTPS URL" },
   ),
   STELLAR_RPC_FALLBACK_URLS: z.string().optional(),
+  LOG_SINK_URL: z.string().optional().refine(
+    (v) => !v || isValidLogSinkUrl(v),
+    { message: "LOG_SINK_URL must be a valid https:// URL" },
+  ),
   SOROBAN_PROXY_FETCH_TIMEOUT_MS: z.string().optional().refine(
     (v) => {
       if (!v) return true
@@ -264,6 +279,16 @@ export function validatePhaseEnv(): EnvValidationResult {
   )
   if (phaseProtocol) errors.push(phaseProtocol)
 
+  for (const violation of checkPreviewIsolation(env)) {
+    errors.push({
+      variable: violation.variable,
+      issue: "isolation_violation",
+      message: violation.message,
+      hint: violation.hint,
+    })
+    incSecurityCounter("preview_isolation_violation")
+  }
+
   // Validar secret keys del faucet si están configuradas
   const adminSecret = validateSecretKey(env.ADMIN_SECRET_KEY, "ADMIN_SECRET_KEY")
   if (adminSecret) errors.push(adminSecret)
diff --git a/lib/feature-flags.ts b/lib/feature-flags.ts
index 7e26b07c..f4562f5b 100644
--- a/lib/feature-flags.ts
+++ b/lib/feature-flags.ts
@@ -62,6 +62,10 @@
  * - phase-167: WalletConnect v2 mobile session resilience with reconnect and chain-switch (spike 623 / issue #486)
  * - phase-168: xBull full support with sign, trustline and Freighter-difference matrix (spike 624 / issue #487)
  * - phase-169: Ledger blind-signing to clear-signing review for Soroban transactions (spike 625 / issue #488)
+ * - phase-170: structured log pipeline with redaction, sampling and a batched external sink (spike 606 / issue #470)
+ * - phase-171: backup and disaster-recovery snapshots for sidecars, SQLite and the secrets map (spike 605 / issue #469)
+ * - phase-172: SQLite to Postgres migration rehearsal with dual-write, parity and rollback (spike 604 / issue #468)
+ * - phase-173: Vercel preview-environment isolation for data roots, secrets, contracts and cron (spike 603 / issue #467)
  */
 
 export type PhaseFeatureFlag =
@@ -137,7 +141,11 @@ export type PhaseFeatureFlag =
   | "phase-166"
   | "phase-167"
   | "phase-168"
-  | "phase-169";
+  | "phase-169"
+  | "phase-170"
+  | "phase-171"
+  | "phase-172"
+  | "phase-173";
 
 
 const FLAG_ENV_MAP: Record<PhaseFeatureFlag, string[]> = {
@@ -212,6 +220,10 @@ const FLAG_ENV_MAP: Record<PhaseFeatureFlag, string[]> = {
   "phase-167": ["NEXT_PUBLIC_FEATURE_PHASE_167", "FEATURE_PHASE_167"],
   "phase-168": ["NEXT_PUBLIC_FEATURE_PHASE_168", "FEATURE_PHASE_168"],
   "phase-169": ["NEXT_PUBLIC_FEATURE_PHASE_169", "FEATURE_PHASE_169"],
+  "phase-170": ["NEXT_PUBLIC_FEATURE_PHASE_170", "FEATURE_PHASE_170"],
+  "phase-171": ["NEXT_PUBLIC_FEATURE_PHASE_171", "FEATURE_PHASE_171"],
+  "phase-172": ["NEXT_PUBLIC_FEATURE_PHASE_172", "FEATURE_PHASE_172"],
+  "phase-173": ["NEXT_PUBLIC_FEATURE_PHASE_173", "FEATURE_PHASE_173"],
 };
 
 
@@ -319,6 +331,10 @@ export function getEnabledFeatureFlags(): PhaseFeatureFlag[] {
     "phase-167",
     "phase-168",
     "phase-169",
+    "phase-170",
+    "phase-171",
+    "phase-172",
+    "phase-173",
   ];
 
   // Filtering is evaluated at call time, allowing runtime test overrides.
diff --git a/lib/log-pipeline.ts b/lib/log-pipeline.ts
new file mode 100644
index 00000000..57a242ae
--- /dev/null
+++ b/lib/log-pipeline.ts
@@ -0,0 +1,294 @@
+/**
+ * Structured log pipeline — partial slice of spike 606 (issue #470).
+ *
+ * `lib/api-observability.ts` was the single sink and its only transport was
+ * `console.*`. This module adds the pieces a hosted log store needs without
+ * changing what the console sees on the happy path: secret redaction, a
+ * deterministic sampler, a closed set of indexable labels, and a bounded,
+ * non-blocking batcher that ships records to an HTTPS sink.
+ *
+ * Design constraints (see docs/spikes/606-log-pipeline-findings.md):
+ * - `emit` never awaits the network and never throws, so a slow or dead sink
+ *   cannot extend request P50 or surface an unhandled rejection.
+ * - The buffer is bounded and drops oldest first; drops are counted, never
+ *   silent. Delivery is at-most-once: a failed batch is dropped, not retried.
+ * - Redaction runs before sampling and before the console line is written, so
+ *   a secret that reaches `metadata` is masked for both transports.
+ * - Only `LOG_LABEL_FIELDS` are safe to index. Wallets, CIDs, token ids, hashes
+ *   and free text stay in the unindexed `fields` bag.
+ *
+ * Scope: redaction, sampling, labels, batching, env validation. Wiring every
+ * ad-hoc `console.*` caller, counter export, and `waitUntil` flush on freeze
+ * are spike 606 follow-ups.
+ *
+ * Feature flag: phase-170. Rollback: unset → `emitLog` is a no-op and the
+ * console line is written exactly as before; buffered records are dropped.
+ */
+
+import { isFeatureEnabled } from "@/lib/feature-flags"
+import { redactSecretKeys } from "@/lib/redaction"
+import { incSecurityCounter } from "@/lib/security-counters"
+
+export type PipelineLogLevel = "info" | "warn" | "error"
+
+export type PipelineLogPayload = {
+  timestamp: string
+  level: PipelineLogLevel
+  event: string
+  route: string
+  method: string
+  correlationId: string
+  [key: string]: unknown
+}
+
+export type PipelineRecord = {
+  labels: {
+    level: PipelineLogLevel
+    event: string
+    route: string
+    method: string
+    status_class: string
+  }
+  timestamp: string
+  correlationId: string
+  fields: Record<string, unknown>
+}
+
+/** Fields that may become index labels. Everything else is stored unindexed. */
+export const LOG_LABEL_FIELDS = ["level", "event", "route", "method", "status_class"] as const
+
+const REDACTED = "[REDACTED]"
+const MAX_STRING = 1000
+const MAX_DEPTH = 6
+
+/** Keys whose values are never shipped, whatever they contain. */
+const SENSITIVE_KEY_RE =
+  /secret|seed|private|passphrase|password|authorization|api[_-]?key|token(?!_?id)|rawpayload|errorresult|signature(?!_?verified)/i
+
+export function isLogPipelineEnabled(): boolean {
+  return isFeatureEnabled("phase-170")
+}
+
+export function logPipelineRollbackNote(): string {
+  return "Rollback phase-170: unset NEXT_PUBLIC_FEATURE_PHASE_170 / FEATURE_PHASE_170 or set to 0/false and restart. Console logging is unchanged; in-memory buffer is dropped."
+}
+
+/** Deep-masks secret-bearing keys and Stellar `S...` seeds; cycle- and depth-safe. */
+export function redactLogValue(value: unknown, depth = 0, seen = new WeakSet<object>()): unknown {
+  if (typeof value === "string") {
+    const masked = redactSecretKeys(value)
+    return masked.length > MAX_STRING ? `${masked.slice(0, MAX_STRING)}…[truncated]` : masked
+  }
+  if (typeof value === "bigint") return value.toString()
+  if (value === null || typeof value !== "object") return value
+  if (depth >= MAX_DEPTH) return "[MAX_DEPTH]"
+  if (seen.has(value)) return "[CIRCULAR]"
+  seen.add(value)
+
+  if (value instanceof Error) {
+    return { name: value.name, message: redactLogValue(value.message, depth + 1, seen) }
+  }
+  if (Array.isArray(value)) return value.map((item) => redactLogValue(item, depth + 1, seen))
+
+  const out: Record<string, unknown> = {}
+  for (const [key, inner] of Object.entries(value)) {
+    out[key] = SENSITIVE_KEY_RE.test(key) ? REDACTED : redactLogValue(inner, depth + 1, seen)
+  }
+  return out
+}
+
+/** Collapses ids so a route label cannot grow with traffic (`/api/x/123` → `/api/x/:id`). */
+export function normalizeRouteLabel(route: string): string {
+  return route
+    .split("/")
+    .map((segment) => (/^\d+$/.test(segment) || /^[0-9a-f]{16,}$/i.test(segment) ? ":id" : segment))
+    .join("/")
+}
+
+function statusClass(status: unknown): string {
+  if (typeof status !== "number" || !Number.isFinite(status)) return "none"
+  return `${Math.floor(status / 100)}xx`
+}
+
+/** FNV-1a, so one correlation id is always kept or dropped together. */
+function hashUnit(input: string): number {
+  let hash = 0x811c9dc5
+  for (let i = 0; i < input.length; i += 1) {
+    hash ^= input.charCodeAt(i)
+    hash = Math.imul(hash, 0x01000193) >>> 0
+  }
+  return hash / 0x100000000
+}
+
+/** Warn and error are always kept; info is sampled per correlation id. */
+export function shouldSampleLog(level: PipelineLogLevel, correlationId: string, infoRate: number): boolean {
+  if (level !== "info") return true
+  if (infoRate >= 1) return true
+  if (infoRate <= 0) return false
+  return hashUnit(correlationId) < infoRate
+}
+
+export function toPipelineRecord(payload: PipelineLogPayload): PipelineRecord {
+  const { timestamp, level, event, route, method, correlationId, ...rest } = payload
+  const redacted = redactLogValue(rest) as Record<string, unknown>
+  return {
+    labels: {
+      level,
+      event,
+      route: normalizeRouteLabel(route),
+      method,
+      status_class: statusClass(rest.status),
+    },
+    timestamp,
+    correlationId,
+    fields: redacted,
+  }
+}
+
+export type LogPipelineConfig = {
+  sinkUrl: string | null
+  sinkToken: string | null
+  infoSampleRate: number
+  batchSize: number
+  maxBuffer: number
+  flushIntervalMs: number
+  timeoutMs: number
+}
+
+function boundedInt(raw: string | undefined, fallback: number, min: number, max: number): number {
+  const n = Number.parseInt(raw ?? "", 10)
+  return Number.isFinite(n) ? Math.min(Math.max(n, min), max) : fallback
+}
+
+export function resolveLogPipelineConfig(env: Record<string, string | undefined> = process.env): LogPipelineConfig {
+  const rate = Number.parseFloat(env.LOG_SAMPLE_RATE_INFO ?? "")
+  return {
+    sinkUrl: env.LOG_SINK_URL?.trim() || null,
+    sinkToken: env.LOG_SINK_TOKEN?.trim() || null,
+    infoSampleRate: Number.isFinite(rate) ? Math.min(Math.max(rate, 0), 1) : 0.1,
+    batchSize: boundedInt(env.LOG_SINK_BATCH_SIZE, 50, 1, 500),
+    maxBuffer: boundedInt(env.LOG_SINK_MAX_BUFFER, 500, 10, 5000),
+    flushIntervalMs: boundedInt(env.LOG_SINK_FLUSH_MS, 2000, 100, 30_000),
+    timeoutMs: boundedInt(env.LOG_SINK_TIMEOUT_MS, 2000, 200, 10_000),
+  }
+}
+
+export { isValidLogSinkUrl } from "@/lib/env-validation"
+
+type FetchLike = (input: string, init: RequestInit) => Promise<{ ok: boolean; status: number }>
+
+export class LogPipeline {
+  private buffer: PipelineRecord[] = []
+  private inflight: Promise<void> | null = null
+  private timer: ReturnType<typeof setTimeout> | null = null
+
+  constructor(
+    private readonly config: LogPipelineConfig,
+    private readonly fetchImpl: FetchLike = fetch as unknown as FetchLike,
+  ) {}
+
+  get pending(): number {
+    return this.buffer.length
+  }
+
+  /** Non-blocking and non-throwing. Returns whether the record was queued. */
+  emit(payload: PipelineLogPayload): boolean {
+    try {
+      if (!this.config.sinkUrl) return false
+      if (!shouldSampleLog(payload.level, payload.correlationId, this.config.infoSampleRate)) {
+        incSecurityCounter("log_pipeline_sampled_out")
+        return false
+      }
+      this.buffer.push(toPipelineRecord(payload))
+      incSecurityCounter("log_pipeline_enqueued")
+      while (this.buffer.length > this.config.maxBuffer) {
+        this.buffer.shift()
+        incSecurityCounter("log_pipeline_dropped")
+      }
+      if (this.buffer.length >= this.config.batchSize) void this.flush()
+      else this.scheduleFlush()
+      return true
+    } catch {
+      return false
+    }
+  }
+
+  private scheduleFlush(): void {
+    if (this.timer) return
+    this.timer = setTimeout(() => {
+      this.timer = null
+      void this.flush()
+    }, this.config.flushIntervalMs)
+    // A pending flush must never keep a process alive.
+    this.timer.unref?.()
+  }
+
+  /** Single-flight; resolves once the in-flight batch settles. Never rejects. */
+  flush(): Promise<void> {
+    if (this.inflight) return this.inflight
+    if (!this.config.sinkUrl || this.buffer.length === 0) return Promise.resolve()
+    const batch = this.buffer.splice(0, this.config.batchSize)
+    this.inflight = this.send(batch).finally(() => {
+      this.inflight = null
+      if (this.buffer.length > 0) this.scheduleFlush()
+    })
+    return this.inflight
+  }
+
+  private async send(batch: PipelineRecord[]): Promise<void> {
+    try {
+      const headers: Record<string, string> = { "content-type": "application/json" }
+      if (this.config.sinkToken) headers.authorization = `Bearer ${this.config.sinkToken}`
+      const response = await this.fetchImpl(this.config.sinkUrl as string, {
+        method: "POST",
+        headers,
+        body: JSON.stringify({ records: batch }),
+        signal: AbortSignal.timeout(this.config.timeoutMs),
+      })
+      if (response.ok) incSecurityCounter("log_pipeline_flush_ok")
+      else throw new Error(`sink status ${response.status}`)
+    } catch {
+      // At-most-once: the batch is gone. Count it so the loss is visible.
+      incSecurityCounter("log_pipeline_flush_failed")
+      incSecurityCounter("log_pipeline_dropped", batch.length)
+    }
+  }
+
+  /** Test and shutdown hook. */
+  async drain(): Promise<void> {
+    if (this.timer) {
+      clearTimeout(this.timer)
+      this.timer = null
+    }
+    while (this.buffer.length > 0 || this.inflight) await this.flush()
+  }
+}
+
+let singleton: LogPipeline | null = null
+
+function getPipeline(): LogPipeline {
+  if (!singleton) singleton = new LogPipeline(resolveLogPipelineConfig())
+  return singleton
+}
+
+export function resetLogPipelineForTests(): void {
+  singleton = null
+}
+
+/**
+ * Entry point used by `api-observability`. Returns the redacted payload that the
+ * console line should carry, or the original payload when the flag is off.
+ */
+export function emitLog(payload: PipelineLogPayload): PipelineLogPayload {
+  if (!isLogPipelineEnabled()) return payload
+  const { timestamp, level, event, route, method, correlationId, ...rest } = payload
+  const safe = redactLogValue(rest) as Record<string, unknown>
+  if (JSON.stringify(safe) !== JSON.stringify(rest)) incSecurityCounter("log_pipeline_redacted")
+  getPipeline().emit(payload)
+  return { timestamp, level, event, route, method, correlationId, ...safe }
+}
+
+/** Awaitable flush for cron/route handlers that can afford it. */
+export function flushLogPipeline(): Promise<void> {
+  return isLogPipelineEnabled() ? getPipeline().flush() : Promise.resolve()
+}
diff --git a/lib/preview-isolation.ts b/lib/preview-isolation.ts
new file mode 100644
index 00000000..23d4cb95
--- /dev/null
+++ b/lib/preview-isolation.ts
@@ -0,0 +1,187 @@
+/**
+ * Vercel preview-environment isolation — partial slice of spike 603 (issue #467).
+ *
+ * The only staging lane was `deploy-staging` on push to main. A preview that
+ * simply inherits project env would share the data root, the issuer, the cron
+ * secret and the contract IDs with production: it could mint real liability,
+ * and N previews running the hourly refill cron would drain one issuer.
+ *
+ * This module turns "is this deploy a preview, and which one" into a pure
+ * function of env, namespaces the data root per preview, and fails validation
+ * when a preview points at production credentials or contracts.
+ *
+ * Production identity is declared by public/hashed fingerprints only, so the
+ * check never needs a production secret in a preview environment:
+ * - `PHASE_PROD_ISSUER`                 public key (G...) of the production issuer
+ * - `PHASE_PROD_CRON_SECRET_SHA256`     sha256 hex of the production `CRON_SECRET`
+ * - `PHASE_PROD_CONTRACT_IDS`           comma-separated production contract IDs
+ *
+ * Scope: id resolution, data-root namespacing, isolation checks, cron gating.
+ * Per-PR contract deploy, a preview deploy CI job and branch-database seeding
+ * are spike 603 follow-ups (see docs/spikes/603-preview-isolation-findings.md).
+ * `/tmp` on Vercel is still ephemeral per instance, so a namespaced root
+ * prevents cross-preview bleed, not loss between instances.
+ *
+ * Feature flag: phase-173. Rollback: unset → data root, cron auth and env
+ * validation behave exactly as before; per-preview directories stay inert.
+ */
+
+import { Keypair, StrKey, hash } from "@stellar/stellar-sdk"
+import { isFeatureEnabled } from "@/lib/feature-flags"
+
+type Env = Record<string, string | undefined>
+
+export type PreviewIsolationViolation = {
+  variable: string
+  code:
+    | "issuer_matches_production"
+    | "cron_secret_matches_production"
+    | "contract_matches_production"
+    | "mainnet_in_preview"
+  message: string
+  hint: string
+}
+
+export function isPreviewIsolationEnabled(): boolean {
+  return isFeatureEnabled("phase-173")
+}
+
+export function previewIsolationRollbackNote(): string {
+  return "Rollback phase-173: unset NEXT_PUBLIC_FEATURE_PHASE_173 / FEATURE_PHASE_173 or set to 0/false and restart. Per-preview data directories stay on disk, inert."
+}
+
+export function isPreviewDeployment(env: Env = process.env): boolean {
+  return env.VERCEL_ENV?.trim().toLowerCase() === "preview"
+}
+
+// `lib/env-validation.ts` reaches client bundles through `phase-protocol.ts`, so
+// this module must stay free of `node:*` imports; the SDK's sha256 is isomorphic.
+function sha256Hex(input: string): string {
+  return hash(Buffer.from(input, "utf8")).toString("hex")
+}
+
+function slug(raw: string): string {
+  return raw
+    .toLowerCase()
+    .replace(/[^a-z0-9]+/g, "-")
+    .replace(/^-+|-+$/g, "")
+    .slice(0, 40)
+}
+
+/**
+ * Stable per-preview id: `pr-<number>` when Vercel provides the PR id, else a
+ * slug of the branch plus a short hash (so `feat/a-b` and `feat/a/b` differ).
+ * `null` outside previews. Always a safe single path segment.
+ */
+export function resolvePreviewId(env: Env = process.env): string | null {
+  if (!isPreviewDeployment(env)) return null
+  const pr = env.VERCEL_GIT_PULL_REQUEST_ID?.trim()
+  if (pr && /^\d{1,9}$/.test(pr)) return `pr-${pr}`
+  const ref = env.VERCEL_GIT_COMMIT_REF?.trim()
+  if (ref) {
+    return `br-${slug(ref) || "branch"}-${sha256Hex(ref).slice(0, 8)}`
+  }
+  return "preview-unknown"
+}
+
+/** Subpath appended to the data root, or `null` when isolation does not apply. */
+export function previewDataSubdir(env: Env = process.env): string | null {
+  if (!isPreviewIsolationEnabled()) return null
+  const id = resolvePreviewId(env)
+  return id ? `previews/${id}` : null
+}
+
+function listFrom(raw: string | undefined): string[] {
+  return (raw ?? "")
+    .split(",")
+    .map((value) => value.trim())
+    .filter(Boolean)
+}
+
+function issuerFromEnv(env: Env): string | null {
+  const publicKey = (env.NEXT_PUBLIC_CLASSIC_LIQ_ISSUER ?? "").trim()
+  if (publicKey && StrKey.isValidEd25519PublicKey(publicKey)) return publicKey
+  const secret = (env.CLASSIC_LIQ_ISSUER_SECRET ?? "").trim()
+  if (secret && StrKey.isValidEd25519SecretSeed(secret)) {
+    try {
+      return Keypair.fromSecret(secret).publicKey()
+    } catch {
+      return null
+    }
+  }
+  return null
+}
+
+/**
+ * Violations that make a preview unsafe. Empty outside previews and when the
+ * flag is off. Never includes a secret value in a message.
+ */
+export function checkPreviewIsolation(env: Env = process.env): PreviewIsolationViolation[] {
+  if (!isPreviewIsolationEnabled() || !isPreviewDeployment(env)) return []
+  const violations: PreviewIsolationViolation[] = []
+
+  const prodIssuer = env.PHASE_PROD_ISSUER?.trim()
+  const issuer = issuerFromEnv(env)
+  if (prodIssuer && issuer && issuer === prodIssuer) {
+    violations.push({
+      variable: "CLASSIC_LIQ_ISSUER_SECRET",
+      code: "issuer_matches_production",
+      message: "Preview resolves to the production issuer account.",
+      hint: "Give previews a dedicated testnet issuer; a preview minting from the production issuer creates real liability.",
+    })
+  }
+
+  const prodCronHash = env.PHASE_PROD_CRON_SECRET_SHA256?.trim().toLowerCase()
+  const cronSecret = env.CRON_SECRET?.trim()
+  if (prodCronHash && cronSecret && sha256Hex(cronSecret) === prodCronHash) {
+    violations.push({
+      variable: "CRON_SECRET",
+      code: "cron_secret_matches_production",
+      message: "Preview CRON_SECRET equals the production cron secret.",
+      hint: "Use a preview-scoped CRON_SECRET so a preview cannot be driven with the production credential.",
+    })
+  }
+
+  const prodContracts = new Set(listFrom(env.PHASE_PROD_CONTRACT_IDS))
+  if (prodContracts.size > 0) {
+    for (const [variable, value] of Object.entries(env)) {
+      if (variable.startsWith("PHASE_PROD_") || !/CONTRACT|PROTOCOL_ID|TOKEN_ID/.test(variable) || !value) continue
+      if (prodContracts.has(value.trim())) {
+        violations.push({
+          variable,
+          code: "contract_matches_production",
+          message: `${variable} points at a production contract.`,
+          hint: "Deploy a per-preview contract (scripts/deploy-phase-sep50.ts) and set the ID for this preview only.",
+        })
+      }
+    }
+  }
+
+  if (env.STELLAR_NETWORK?.trim().toLowerCase() === "mainnet") {
+    violations.push({
+      variable: "STELLAR_NETWORK",
+      code: "mainnet_in_preview",
+      message: "Previews must run on testnet.",
+      hint: "Unset STELLAR_NETWORK or set it to testnet for the preview environment.",
+    })
+  }
+
+  return violations
+}
+
+/**
+ * Previews do not run scheduled jobs unless explicitly enabled, so N previews
+ * cannot each run the hourly refill against one issuer.
+ */
+export function isCronBlockedInPreview(env: Env = process.env): boolean {
+  if (!isPreviewIsolationEnabled() || !isPreviewDeployment(env)) return false
+  return env.PHASE_PREVIEW_CRON_ENABLED?.trim() !== "1"
+}
+
+/**
+ * `x-vercel-cron` is a plain header any caller can send. In a preview, a bearer
+ * `CRON_SECRET` is always required.
+ */
+export function previewTrustsVercelCronHeader(env: Env = process.env): boolean {
+  return !(isPreviewIsolationEnabled() && isPreviewDeployment(env))
+}
diff --git a/lib/security-counters.ts b/lib/security-counters.ts
index 04fb427b..24ff5ca4 100644
--- a/lib/security-counters.ts
+++ b/lib/security-counters.ts
@@ -67,7 +67,25 @@ export type SecurityCounterName =
   | "ledger_clear_signing_approved"
   | "ledger_clear_signing_rejected"
   | "ledger_blind_signing_blocked"
-  | "ledger_gate13_preempted";
+  | "ledger_gate13_preempted"
+  | "log_pipeline_enqueued"
+  | "log_pipeline_sampled_out"
+  | "log_pipeline_dropped"
+  | "log_pipeline_redacted"
+  | "log_pipeline_flush_ok"
+  | "log_pipeline_flush_failed"
+  | "backup_snapshot_created"
+  | "backup_snapshot_verified"
+  | "backup_snapshot_corrupt"
+  | "backup_restore_completed"
+  | "backup_restore_blocked"
+  | "db_parity_checked"
+  | "db_parity_mismatch"
+  | "db_dual_write_ok"
+  | "db_dual_write_drift"
+  | "db_dual_write_shed"
+  | "preview_isolation_violation"
+  | "preview_cron_blocked";
 
 
 const counters = new Map<SecurityCounterName, number>()
diff --git a/lib/server-data-paths.ts b/lib/server-data-paths.ts
index 49f551c1..0cc2e7dc 100644
--- a/lib/server-data-paths.ts
+++ b/lib/server-data-paths.ts
@@ -1,5 +1,6 @@
 import os from "node:os"
 import path from "node:path"
+import { previewDataSubdir } from "@/lib/preview-isolation"
 
 /** Local filenames only — Next file tracing stays predictable. */
 const FILES = {
@@ -56,8 +57,9 @@ export type ServerDataFile = keyof typeof FILES
  * - Local: `<cwd>/.data`
  * - Vercel: `<os.tmpdir()>/phase-server-data` (project dir is read-only)
  * - Override: `PHASE_SERVER_DATA_DIR` (e.g. mounted volume)
+ * - Preview (phase-173): `serverDataRoot()` appends `previews/<id>` to the above
  */
-function serverDataRoot(): string {
+function baseDataRoot(): string {
   const fromEnv = process.env.PHASE_SERVER_DATA_DIR?.trim()
   if (fromEnv) return fromEnv
   if (process.env.VERCEL) {
@@ -66,6 +68,21 @@ function serverDataRoot(): string {
   return path.join(process.cwd(), ".data")
 }
 
+export function serverDataRoot(): string {
+  const base = baseDataRoot()
+  const preview = previewDataSubdir()
+  return preview ? path.join(base, preview) : base
+}
+
 export function serverDataJsonPath(key: ServerDataFile): string {
   return path.join(serverDataRoot(), FILES[key])
 }
+
+/** Every sidecar key, so backup tooling never drifts from the FILES table. */
+export function serverDataFileKeys(): ServerDataFile[] {
+  return Object.keys(FILES) as ServerDataFile[]
+}
+
+export function serverDataFileName(key: ServerDataFile): string {
+  return FILES[key]
+}
diff --git a/package.json b/package.json
index 956dea8c..88068d1f 100644
--- a/package.json
+++ b/package.json
@@ -20,7 +20,9 @@
     "start": "next start",
     "security:audit": "npm audit --audit-level=high",
     "security:secret-rotation": "tsx scripts/check-secret-rotation.ts",
-    "migrate:sqlite": "tsx scripts/migrate-json-to-sqlite.ts"
+    "migrate:sqlite": "tsx scripts/migrate-json-to-sqlite.ts",
+    "backup:drill": "tsx scripts/backup-drill.ts",
+    "db:parity": "tsx scripts/db-parity.ts"
   },
   "dependencies": {
     "@albedo-link/intent": "0.12.0",
diff --git a/scripts/backup-drill.ts b/scripts/backup-drill.ts
new file mode 100644
index 00000000..b4bd4fb2
--- /dev/null
+++ b/scripts/backup-drill.ts
@@ -0,0 +1,51 @@
+/**
+ * Spike 605 (issue #469) disaster-recovery drill.
+ *
+ * Snapshots the live data root, verifies it, restores it into a scratch
+ * directory, and proves SQLite parity between source and restore. Never writes
+ * into the live data root and never prints secrets.
+ *
+ * Usage:
+ *   npx tsx scripts/backup-drill.ts [snapshot-dir]
+ */
+import { mkdtemp } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { DatabaseSync } from "node:sqlite";
+
+process.env.FEATURE_PHASE_171 = "1";
+process.env.FEATURE_PHASE_172 = "1";
+
+async function main() {
+  const { createSnapshot, restoreSnapshot, verifySnapshot } = await import("../lib/backup-snapshot");
+  const { compareSources, sqliteParitySource } = await import("../lib/db-parity");
+  const { getDb } = await import("../lib/sqlite-db");
+
+  const snapDir = process.argv[2] ?? (await mkdtemp(path.join(tmpdir(), "phase-drill-snap-")));
+  const restoreDir = await mkdtemp(path.join(tmpdir(), "phase-drill-restore-"));
+
+  const manifest = await createSnapshot(snapDir);
+  console.log(`snapshot: ${manifest.files.length} sidecar(s), ${Object.keys(manifest.sqlite?.tables ?? {}).length} table(s) -> ${snapDir}`);
+
+  const verification = await verifySnapshot(snapDir);
+  if (!verification.ok) {
+    console.error("verify FAILED:", verification.problems);
+    process.exit(1);
+  }
+  console.log("verify: ok");
+
+  await restoreSnapshot(snapDir, restoreDir);
+  const restored = new DatabaseSync(path.join(restoreDir, "phase.sqlite3"), { readOnly: true });
+  const parity = compareSources(sqliteParitySource(getDb()), sqliteParitySource(restored));
+  restored.close();
+  if (!parity.ok) {
+    console.error("parity FAILED:", JSON.stringify(parity.tables.filter((t) => !t.match)));
+    process.exit(1);
+  }
+  console.log(`parity: ok across ${parity.tables.length} table(s); restored into ${restoreDir}`);
+}
+
+main().catch((error) => {
+  console.error(error instanceof Error ? error.message : error);
+  process.exit(1);
+});
diff --git a/scripts/db-parity.ts b/scripts/db-parity.ts
new file mode 100644
index 00000000..0d945852
--- /dev/null
+++ b/scripts/db-parity.ts
@@ -0,0 +1,34 @@
+/**
+ * Spike 604 (issue #468) parity check between two SQLite files.
+ *
+ * Exercises the same `compareSources` the Postgres rehearsal will use, so a
+ * restored backup or a re-created database can be checked table by table
+ * (count, per-column value, type coercion, checksum).
+ *
+ * Usage:
+ *   npx tsx scripts/db-parity.ts <left.sqlite3> <right.sqlite3>
+ */
+import { DatabaseSync } from "node:sqlite";
+
+process.env.FEATURE_PHASE_172 = "1";
+
+async function main() {
+  const [leftPath, rightPath] = process.argv.slice(2);
+  if (!leftPath || !rightPath) {
+    console.error("usage: db-parity.ts <left.sqlite3> <right.sqlite3>");
+    process.exit(2);
+  }
+  const { compareSources, sqliteParitySource } = await import("../lib/db-parity");
+  const left = new DatabaseSync(leftPath, { readOnly: true });
+  const right = new DatabaseSync(rightPath, { readOnly: true });
+  const report = compareSources(sqliteParitySource(left), sqliteParitySource(right));
+  left.close();
+  right.close();
+  for (const t of report.tables) {
+    console.log(`${t.match ? "ok  " : "DIFF"} ${t.table} (${t.leftCount} vs ${t.rightCount})`);
+  }
+  for (const t of report.missingTables) console.log(`MISS ${t}`);
+  process.exit(report.ok ? 0 : 1);
+}
+
+void main();