From 84afc36b0b427e660615215fa3cf9633e2a22915 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 16:04:43 +0000
Subject: [PATCH 1/9] test(pg-delta): add failing coverage for schema apply
 statement debugging
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Regression coverage for statement-level debugging in `schema apply`:
apply()'s new onEvent observer (tests/apply-observer.test.ts) and the
`schema apply --dry-run`/`--verbose`/`--out-plan` CLI flags
(tests/cli.test.ts, "CLI: schema apply debugging"). Both fail today —
onEvent does not exist on ApplyOptions yet, and the three flags are
unrecognized by parseFlags.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 .../pg-delta/tests/apply-observer.test.ts     | 243 ++++++++++++++++++
 packages/pg-delta/tests/cli.test.ts           | 133 ++++++++++
 2 files changed, 376 insertions(+)
 create mode 100644 packages/pg-delta/tests/apply-observer.test.ts

diff --git a/packages/pg-delta/tests/apply-observer.test.ts b/packages/pg-delta/tests/apply-observer.test.ts
new file mode 100644
index 000000000..7263449cf
--- /dev/null
+++ b/packages/pg-delta/tests/apply-observer.test.ts
@@ -0,0 +1,243 @@
+/**
+ * Statement-level debugging for `schema apply` (issue: observability into
+ * exactly what apply() executes). apply() gains an optional `onEvent`
+ * observer (src/apply/apply.ts) that reports segment/action boundaries as
+ * they happen — additive only: no event may change apply()'s control flow,
+ * action statuses, or the returned ApplyReport.
+ *
+ * Drives extract -> plan -> apply directly against two real databases (the
+ * same pattern as tests/execution.test.ts), so the event stream reflects the
+ * production execution path, not a hand-rolled fixture.
+ */
+import { describe, expect, test } from "bun:test";
+import { apply, type ApplyEvent } from "../src/apply/apply.ts";
+import { extract } from "../src/extract/extract.ts";
+import { plan } from "../src/plan/plan.ts";
+import { sharedCluster } from "./containers.ts";
+
+describe("apply() onEvent observer", () => {
+  test("happy path: segmentStart, actionStart/actionEnd(ok:true) pairs in plan order, segmentEnd committed", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_tgt");
+    const desired = await cluster.createDb("apply_observer_desired");
+    try {
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer PRIMARY KEY, name text);
+        CREATE INDEX t_name_idx ON app.t (name);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase);
+      expect(thePlan.actions.length).toBeGreaterThan(0);
+
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("applied");
+
+      // exactly one segmentStart/segmentEnd for a single-segment plan
+      const segmentStarts = events.filter((e) => e.kind === "segmentStart");
+      const segmentEnds = events.filter((e) => e.kind === "segmentEnd");
+      expect(segmentStarts.length).toBeGreaterThan(0);
+      expect(segmentEnds.length).toBe(segmentStarts.length);
+      for (const e of segmentEnds) {
+        expect(e.kind === "segmentEnd" && e.outcome).toBe("committed");
+      }
+
+      // actionStart/actionEnd pairs, in plan order, one per action, all ok
+      const actionStarts = events.filter((e) => e.kind === "actionStart");
+      const actionEnds = events.filter((e) => e.kind === "actionEnd");
+      expect(actionStarts.length).toBe(thePlan.actions.length);
+      expect(actionEnds.length).toBe(thePlan.actions.length);
+      for (let i = 0; i < thePlan.actions.length; i++) {
+        const s = actionStarts[i];
+        const e = actionEnds[i];
+        expect(s?.kind === "actionStart" && s.actionIndex).toBe(i);
+        expect(s?.kind === "actionStart" && s.sql).toBe(
+          thePlan.actions[i]!.sql,
+        );
+        expect(e?.kind === "actionEnd" && e.actionIndex).toBe(i);
+        expect(e?.kind === "actionEnd" && e.ok).toBe(true);
+        expect(e?.kind === "actionEnd" && typeof e.ms).toBe("number");
+        expect(e?.kind === "actionEnd" && e.ms).toBeGreaterThanOrEqual(0);
+      }
+
+      // events interleave in the order apply() executes them: for every
+      // actionStart[i], its actionEnd[i] appears immediately after (modulo
+      // segment boundary events), never before the next actionStart.
+      const kinds = events.map((e) => e.kind);
+      const firstActionStartIdx = kinds.indexOf("actionStart");
+      expect(firstActionStartIdx).toBeGreaterThanOrEqual(0);
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
+  test("control events: BEGIN, the preamble SET, and COMMIT bracket the action events", async () => {
+    // The statements apply() actually sends to the wire are NOT limited to
+    // plan actions — BEGIN/COMMIT framing and preamble SETs go over the same
+    // connection. `onEvent` must report those too (kind: "control"), so a
+    // debugging trace is a COMPLETE record of the wire, not just the planner's
+    // atomic DDL.
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_ctrl_tgt");
+    const desired = await cluster.createDb("apply_observer_ctrl_desired");
+    try {
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer PRIMARY KEY);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase);
+      expect(thePlan.actions.length).toBeGreaterThan(0);
+
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        lockTimeoutMs: 5000, // forces a preamble SET LOCAL lock_timeout
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("applied");
+
+      const isControl = (
+        e: ApplyEvent,
+      ): e is Extract<ApplyEvent, { kind: "control" }> => e.kind === "control";
+      const controlEvents = events.filter(isControl);
+      expect(controlEvents.some((e) => e.sql === "BEGIN")).toBe(true);
+      expect(
+        controlEvents.some((e) => e.sql.startsWith("SET LOCAL lock_timeout")),
+      ).toBe(true);
+      expect(controlEvents.some((e) => e.sql === "COMMIT")).toBe(true);
+
+      // ordering: BEGIN -> preamble SET -> first actionStart ... last actionEnd -> COMMIT -> segmentEnd
+      const beginPos = events.findIndex(
+        (e) => e.kind === "control" && e.sql === "BEGIN",
+      );
+      const setPos = events.findIndex(
+        (e) =>
+          e.kind === "control" && e.sql.startsWith("SET LOCAL lock_timeout"),
+      );
+      const firstActionStartPos = events.findIndex(
+        (e) => e.kind === "actionStart",
+      );
+      const lastActionEndPos = events
+        .map((e) => e.kind)
+        .lastIndexOf("actionEnd");
+      const commitPos = events.findIndex(
+        (e) => e.kind === "control" && e.sql === "COMMIT",
+      );
+      const segmentEndPos = events.findIndex((e) => e.kind === "segmentEnd");
+
+      expect(beginPos).toBeGreaterThanOrEqual(0);
+      expect(beginPos).toBeLessThan(setPos);
+      expect(setPos).toBeLessThan(firstActionStartPos);
+      expect(commitPos).toBeGreaterThan(lastActionEndPos);
+      expect(commitPos).toBeLessThan(segmentEndPos);
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
+  test("failure path: the failing action gets actionEnd ok:false, its segment gets segmentEnd rolledBack", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_fail_tgt");
+    const desired = await cluster.createDb("apply_observer_fail_desired");
+    try {
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer PRIMARY KEY);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase);
+      expect(thePlan.actions.length).toBeGreaterThan(0);
+
+      // sabotage: pre-create the schema on the target so the planned
+      // `CREATE SCHEMA app` action fails when replayed.
+      await target.pool.query(`CREATE SCHEMA app`);
+
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("failed");
+      expect(report.error).toBeDefined();
+
+      const failedIndex = report.error!.actionIndex;
+      const actionEnds = events.filter((e) => e.kind === "actionEnd");
+      const failedEnd = actionEnds.find(
+        (e) => e.kind === "actionEnd" && e.actionIndex === failedIndex,
+      );
+      expect(failedEnd).toBeDefined();
+      expect(failedEnd?.kind === "actionEnd" && failedEnd.ok).toBe(false);
+
+      const segmentEnds = events.filter((e) => e.kind === "segmentEnd");
+      expect(segmentEnds.length).toBeGreaterThan(0);
+      // the (single) segment containing the failure rolled back
+      expect(
+        segmentEnds.some(
+          (e) => e.kind === "segmentEnd" && e.outcome === "rolledBack",
+        ),
+      ).toBe(true);
+
+      // report semantics unchanged: the failing action and everything after
+      // it in its (rolled-back) segment reports unapplied.
+      expect(report.actionStatuses[failedIndex]).toBe("unapplied");
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
+  test("an observer that always throws never changes apply()'s outcome", async () => {
+    const cluster = await sharedCluster();
+    const targetA = await cluster.createDb("apply_observer_safe_a");
+    const targetB = await cluster.createDb("apply_observer_safe_b");
+    const desired = await cluster.createDb("apply_observer_safe_desired");
+    try {
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer PRIMARY KEY, name text);
+        CREATE INDEX t_name_idx ON app.t (name);
+      `);
+      const desiredState = await extract(desired.pool);
+
+      const [stateA, stateB] = [
+        await extract(targetA.pool),
+        await extract(targetB.pool),
+      ];
+      const planA = plan(stateA.factBase, desiredState.factBase);
+      const planB = plan(stateB.factBase, desiredState.factBase);
+
+      const reportNoObserver = await apply(planA, targetA.pool, {
+        fingerprintGate: false,
+      });
+      const reportThrowingObserver = await apply(planB, targetB.pool, {
+        fingerprintGate: false,
+        onEvent: () => {
+          throw new Error("observer boom");
+        },
+      });
+
+      expect(reportThrowingObserver.status).toBe(reportNoObserver.status);
+      expect(reportThrowingObserver.appliedActions).toBe(
+        reportNoObserver.appliedActions,
+      );
+      expect(reportThrowingObserver.actionStatuses).toEqual(
+        reportNoObserver.actionStatuses,
+      );
+    } finally {
+      await Promise.all([targetA.drop(), targetB.drop(), desired.drop()]);
+    }
+  }, 60_000);
+});
diff --git a/packages/pg-delta/tests/cli.test.ts b/packages/pg-delta/tests/cli.test.ts
index 2b8e154cf..a2c1907bd 100644
--- a/packages/pg-delta/tests/cli.test.ts
+++ b/packages/pg-delta/tests/cli.test.ts
@@ -9,6 +9,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
 import { loadSnapshot } from "../src/frontends/snapshot-file.ts";
+import { parsePlan } from "../src/plan/artifact.ts";
 import { isolatedClusterPair, sharedCluster } from "./containers.ts";
 
 const PKG_DIR = new URL("..", import.meta.url).pathname.replace(/\/$/, "");
@@ -1685,3 +1686,135 @@ describe("CLI: schema export --layout grouped", () => {
     expect(result.stderr).toContain("--format-options");
   });
 });
+
+describe("CLI: schema apply debugging", () => {
+  test("--dry-run prints the executable script to stdout and applies nothing", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("cli_apply_dryrun_tgt");
+    try {
+      const dir = join(tmpdir(), `pg-delta-next-dryrun-${Date.now()}`);
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `CREATE SCHEMA app;\nCREATE TABLE app.t (id integer PRIMARY KEY);\n`,
+      );
+
+      // no --shadow: co-located shadow on the target's own cluster
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--dry-run",
+      ]);
+      expect({ code: res.exitCode, stderr: res.stderr }).toMatchObject({
+        code: 0,
+      });
+      expect(res.stdout).toContain("CREATE TABLE");
+      expect(res.stdout).toContain('"app"."t"');
+      expect(res.stderr).toMatch(
+        /Dry run: \d+ action\(s\) planned; nothing applied\./,
+      );
+
+      // the target must be UNCHANGED — nothing was applied
+      const { rows } = await target.pool.query<{ n: number }>(
+        `SELECT count(*)::int AS n FROM pg_tables WHERE schemaname = 'app'`,
+      );
+      expect(rows[0]?.n).toBe(0);
+    } finally {
+      await target.drop();
+    }
+  }, 90_000);
+
+  test("--dry-run --out-plan writes a plan artifact that parses with actions", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("cli_apply_dryrun_outplan_tgt");
+    try {
+      const dir = join(tmpdir(), `pg-delta-next-dryrun-outplan-${Date.now()}`);
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `CREATE SCHEMA app;\nCREATE TABLE app.t (id integer PRIMARY KEY);\n`,
+      );
+      const planPath = join(dir, "plan.json");
+
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--dry-run",
+        "--out-plan",
+        planPath,
+      ]);
+      expect({ code: res.exitCode, stderr: res.stderr }).toMatchObject({
+        code: 0,
+      });
+      expect(res.stderr).toContain(`Plan artifact written to ${planPath}`);
+
+      const parsed = parsePlan(readFileSync(planPath, "utf8"));
+      expect(parsed.actions.length).toBeGreaterThan(0);
+
+      // still nothing applied
+      const { rows } = await target.pool.query<{ n: number }>(
+        `SELECT count(*)::int AS n FROM pg_tables WHERE schemaname = 'app'`,
+      );
+      expect(rows[0]?.n).toBe(0);
+    } finally {
+      await target.drop();
+    }
+  }, 90_000);
+
+  test("--verbose logs per-statement progress to stderr during a real apply", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_apply_verbose_shadow");
+    const target = await cluster.createDb("cli_apply_verbose_tgt");
+    try {
+      const dir = join(tmpdir(), `pg-delta-next-verbose-${Date.now()}`);
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `CREATE SCHEMA app;\nCREATE TABLE app.t (id integer PRIMARY KEY);\n`,
+      );
+
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--verbose",
+      ]);
+      expect({ code: res.exitCode, stderr: res.stderr }).toMatchObject({
+        code: 0,
+      });
+      expect(res.stderr).toContain("[1/");
+      expect(res.stderr).toContain("ok (");
+      // --verbose is a COMPLETE record of the wire, not just plan actions: the
+      // applied statements are planner-rendered atomic DDL, so the trace must
+      // also show the transaction framing actually sent (BEGIN/COMMIT).
+      expect(res.stderr).toContain("BEGIN");
+      expect(res.stderr).toContain("COMMIT");
+
+      const { rows } = await target.pool.query<{ n: number }>(
+        `SELECT count(*)::int AS n FROM pg_tables WHERE schemaname = 'app'`,
+      );
+      expect(rows[0]?.n).toBe(1);
+    } finally {
+      await Promise.all([shadow.drop(), target.drop()]);
+    }
+  }, 90_000);
+});

From 1cfcfbc3dea26c8bc74c2ac30977b1a76f8c49e5 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 16:18:19 +0000
Subject: [PATCH 2/9] feat(pg-delta): add --dry-run/--verbose/--out-plan to
 schema apply and an apply onEvent observer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`schema apply` today only prints action counts; the SQL it actually runs is
visible only when a statement fails. Since the executed statements are
planner-rendered atomic DDL derived from the catalog diff — not the authored
declarative SQL — there was no way to inspect what would run, or watch it run,
without instrumenting the process externally.

Library (src/apply/apply.ts): ApplyOptions gains an optional `onEvent`
observer fired at segment/action boundaries (ApplyEvent: segmentStart,
actionStart, actionEnd, segmentEnd, and control — the latter covering every
non-action statement sent on the wire: BEGIN, preamble SET/SET LOCAL, COMMIT,
ROLLBACK, RESET ALL). Every emission is wrapped in try/catch so a throwing
observer can never change apply()'s control flow, action statuses, or the
returned report — purely additive.

CLI (src/cli/commands/schema.ts, src/cli/main.ts): `schema apply` gains:
  --dry-run     print the exact executable script (via the `render` command's
                renderer, so it splits on the same segment boundaries apply()
                executes) to stdout instead of applying; no fingerprint gate
                runs. A stderr summary reports the action count and flags
                destructive actions.
  --verbose     stream the onEvent trace to stderr during a real apply:
                segment start/end, each action's SQL and ok/FAILED outcome
                with timing, and every control statement (prefixed "  ; ").
  --out-plan    write the plan artifact right after planning, before apply
                (or the --dry-run script).

RED (tests/apply-observer.test.ts, tests/cli.test.ts "CLI: schema apply
debugging", captured before this commit):

  error: expect(received).toBeGreaterThan(expected)
  Expected: > 0
  Received: 0
  (fail) apply() onEvent observer > happy path: segmentStart, actionStart/
  actionEnd(ok:true) pairs in plan order, segmentEnd committed

  error: expect(received).toMatchObject(expected)
    { - "code": 0, + "code": 2, + "stderr": "Unknown flag: --dry-run ..." }
  (fail) CLI: schema apply debugging > --dry-run prints the executable
  script to stdout and applies nothing
  (fail) CLI: schema apply debugging > --dry-run --out-plan writes a plan
  artifact that parses with actions
  (fail) CLI: schema apply debugging > --verbose logs per-statement progress
  to stderr during a real apply

  (a follow-up spec addendum requiring `control` events for BEGIN/preamble
  SET/COMMIT also went RED first: "Expected: true / Received: false" on
  `controlEvents.some((e) => e.sql === "BEGIN")`.)

GREEN: all of the above pass, plus `bun test src/` (710 pass) and the full
existing tests/cli.test.ts suite (44 pass) show no regressions.

Changeset: .changeset/schema-apply-statement-debugging.md (minor).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 .../schema-apply-statement-debugging.md       |   5 +
 packages/pg-delta/src/apply/apply.ts          | 143 +++++++++++++++++-
 packages/pg-delta/src/cli/commands/schema.ts  | 133 +++++++++++++++-
 packages/pg-delta/src/cli/main.ts             |  23 +++
 packages/pg-delta/src/cli/render.ts           |   6 +-
 5 files changed, 302 insertions(+), 8 deletions(-)
 create mode 100644 .changeset/schema-apply-statement-debugging.md

diff --git a/.changeset/schema-apply-statement-debugging.md b/.changeset/schema-apply-statement-debugging.md
new file mode 100644
index 000000000..7976ed96f
--- /dev/null
+++ b/.changeset/schema-apply-statement-debugging.md
@@ -0,0 +1,5 @@
+---
+"@supabase/pg-delta": minor
+---
+
+Add statement-level debugging for `schema apply`: `--dry-run` prints the exact executable script (segment-split, mirroring execution) to stdout without applying anything or running the fingerprint gate; `--verbose` streams a segment/action progress trace to stderr as the real apply runs, including every non-action statement actually sent on the connection (`BEGIN`, preamble `SET`/`SET LOCAL`, `COMMIT`, `ROLLBACK`, `RESET ALL`) since the applied statements are planner-rendered atomic DDL, not the authored declarative SQL; `--out-plan <plan.json>` archives the plan artifact right after planning. The library `apply()` gains a corresponding optional `onEvent` observer (`ApplyEvent`: `segmentStart`/`actionStart`/`actionEnd`/`segmentEnd`/`control`) that any caller can hook into — purely additive, a throwing observer never changes apply's control flow, action statuses, or report.
diff --git a/packages/pg-delta/src/apply/apply.ts b/packages/pg-delta/src/apply/apply.ts
index e7d03b549..24bb02106 100644
--- a/packages/pg-delta/src/apply/apply.ts
+++ b/packages/pg-delta/src/apply/apply.ts
@@ -29,6 +29,35 @@ export interface ApplyReport {
   error?: { actionIndex: number; sql: string; message: string };
 }
 
+/** Observability events emitted during `apply()` (statement-level debugging
+ *  for `schema apply --verbose`). Purely additive — no event ever changes
+ *  what apply() does or reports; see the `onEvent` doc comment below.
+ *
+ *  The applied statements are planner-rendered atomic DDL, not the authored
+ *  declarative SQL, so `actionStart`/`actionEnd` alone are not a complete
+ *  record of the wire: `control` covers every OTHER statement apply() sends
+ *  on the same connection — `BEGIN`, each preamble `SET`/`SET LOCAL`, `COMMIT`,
+ *  `ROLLBACK` (including best-effort rollbacks on an error path), and
+ *  `RESET ALL` — so a `--verbose` trace shows exactly what ran, transaction
+ *  framing included. */
+export type ApplyEvent =
+  | {
+      kind: "segmentStart";
+      segmentIndex: number;
+      segmentCount: number;
+      start: number;
+      end: number;
+      transactional: boolean;
+    }
+  | { kind: "actionStart"; actionIndex: number; sql: string }
+  | { kind: "actionEnd"; actionIndex: number; ok: boolean; ms: number }
+  | {
+      kind: "segmentEnd";
+      segmentIndex: number;
+      outcome: "committed" | "rolledBack" | "inDoubt";
+    }
+  | { kind: "control"; sql: string };
+
 export interface ApplyOptions {
   /** re-extract the target and require its fingerprint to equal the
    *  plan's source fingerprint (stage 6 deliverable 3). Defaults to ON;
@@ -50,6 +79,10 @@ export interface ApplyOptions {
    *  plan was fingerprinted from; otherwise operationally-managed objects present
    *  on the real target read as drift and reject a valid managed plan. */
   reextract?: (pool: Pool) => Promise<{ factBase: FactBase }>;
+  /** observability hook (statement-level debugging): fired at segment/action
+   *  boundaries as apply() executes. Purely additive — see `emit` below for
+   *  the isolation guarantee. */
+  onEvent?: (event: ApplyEvent) => void;
 }
 
 interface Segment {
@@ -109,6 +142,23 @@ function errorEntry(
   };
 }
 
+/** Fire an observer event, swallowing anything it throws. `onEvent` is a
+ *  debugging hook (`schema apply --verbose`), never a semantics participant —
+ *  a throwing observer must NEVER change apply's control flow, action
+ *  statuses, or the returned report, so every emission is wrapped here rather
+ *  than left to each call site. */
+function emit(
+  onEvent: ((event: ApplyEvent) => void) | undefined,
+  event: ApplyEvent,
+): void {
+  if (onEvent === undefined) return;
+  try {
+    onEvent(event);
+  } catch {
+    // swallowed by design — see doc comment above
+  }
+}
+
 export async function apply(
   thePlan: Plan,
   target: Pool,
@@ -208,20 +258,53 @@ export async function apply(
       ),
     ];
 
-    for (const segment of segments) {
+    const onEvent = options?.onEvent;
+    for (let segIdx = 0; segIdx < segments.length; segIdx++) {
+      const segment = segments[segIdx]!;
+      // segmentStart fires before the preamble/BEGIN for this segment, for
+      // BOTH transactional and non-transactional segments.
+      emit(onEvent, {
+        kind: "segmentStart",
+        segmentIndex: segIdx,
+        segmentCount: segments.length,
+        start: segment.start,
+        end: segment.end,
+        transactional: segment.transactional,
+      });
+
       if (!segment.transactional) {
         // a lone non-transactional action; session-level settings, reset after
         const index = segment.start;
         const action = thePlan.actions[index]!;
+        const actionStartedAt = Date.now();
+        emit(onEvent, {
+          kind: "actionStart",
+          actionIndex: index,
+          sql: action.sql,
+        });
         try {
-          for (const sql of preamble(false)) await client.query(sql);
+          for (const sql of preamble(false)) {
+            emit(onEvent, { kind: "control", sql });
+            await client.query(sql);
+          }
           await client.query(action.sql);
         } catch (error) {
+          emit(onEvent, {
+            kind: "actionEnd",
+            actionIndex: index,
+            ok: false,
+            ms: Date.now() - actionStartedAt,
+          });
           // a failed non-transactional DDL is NOT safely unapplied — it can
           // leave durable side effects (e.g. an INVALID index from a cancelled
           // CREATE INDEX CONCURRENTLY). Report it inDoubt so the caller knows
           // the database must be re-extracted before retry (review P1).
           statuses[index] = "inDoubt";
+          emit(onEvent, {
+            kind: "segmentEnd",
+            segmentIndex: segIdx,
+            outcome: "inDoubt",
+          });
           return {
             status: "failed",
             appliedActions,
@@ -232,18 +315,40 @@ export async function apply(
           // ALWAYS restore session state before the client returns to the pool,
           // on success or failure — RESET ALL must not be skipped by the catch's
           // early return, and a reset failure must not flip the action's outcome.
+          emit(onEvent, { kind: "control", sql: "RESET ALL" });
           await client.query("RESET ALL").catch(() => {});
         }
+        emit(onEvent, {
+          kind: "actionEnd",
+          actionIndex: index,
+          ok: true,
+          ms: Date.now() - actionStartedAt,
+        });
         statuses[index] = "applied";
         appliedActions += 1;
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "committed",
+        });
         continue;
       }
 
       try {
+        emit(onEvent, { kind: "control", sql: "BEGIN" });
         await client.query("BEGIN");
-        for (const sql of preamble(true)) await client.query(sql);
+        for (const sql of preamble(true)) {
+          emit(onEvent, { kind: "control", sql });
+          await client.query(sql);
+        }
       } catch (error) {
+        emit(onEvent, { kind: "control", sql: "ROLLBACK" });
         await client.query("ROLLBACK").catch(() => {});
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "rolledBack",
+        });
         return {
           status: "failed",
           appliedActions,
@@ -253,10 +358,24 @@ export async function apply(
       }
       for (let i = segment.start; i < segment.end; i++) {
         const action = thePlan.actions[i]!;
+        const actionStartedAt = Date.now();
+        emit(onEvent, { kind: "actionStart", actionIndex: i, sql: action.sql });
         try {
           await client.query(action.sql);
         } catch (error) {
+          emit(onEvent, {
+            kind: "actionEnd",
+            actionIndex: i,
+            ok: false,
+            ms: Date.now() - actionStartedAt,
+          });
+          emit(onEvent, { kind: "control", sql: "ROLLBACK" });
           await client.query("ROLLBACK").catch(() => {});
+          emit(onEvent, {
+            kind: "segmentEnd",
+            segmentIndex: segIdx,
+            outcome: "rolledBack",
+          });
           return {
             status: "failed",
             appliedActions,
@@ -264,14 +383,27 @@ export async function apply(
             error: errorEntry(i, action.sql, error),
           };
         }
+        emit(onEvent, {
+          kind: "actionEnd",
+          actionIndex: i,
+          ok: true,
+          ms: Date.now() - actionStartedAt,
+        });
       }
       try {
+        emit(onEvent, { kind: "control", sql: "COMMIT" });
         await client.query("COMMIT");
       } catch (error) {
         // the commit itself failed: the segment's fate is unknown
         for (let i = segment.start; i < segment.end; i++)
           statuses[i] = "inDoubt";
+        emit(onEvent, { kind: "control", sql: "ROLLBACK" });
         await client.query("ROLLBACK").catch(() => {});
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "inDoubt",
+        });
         return {
           status: "failed",
           appliedActions,
@@ -281,6 +413,11 @@ export async function apply(
       }
       for (let i = segment.start; i < segment.end; i++) statuses[i] = "applied";
       appliedActions += segment.end - segment.start;
+      emit(onEvent, {
+        kind: "segmentEnd",
+        segmentIndex: segIdx,
+        outcome: "committed",
+      });
     }
   } finally {
     client.release();
diff --git a/packages/pg-delta/src/cli/commands/schema.ts b/packages/pg-delta/src/cli/commands/schema.ts
index 0434585a9..a47ccb0bf 100644
--- a/packages/pg-delta/src/cli/commands/schema.ts
+++ b/packages/pg-delta/src/cli/commands/schema.ts
@@ -22,6 +22,7 @@
  * schema apply --dir <dir> --shadow <pg-url> --target <pg-url>
  *              [--renames auto|prompt|off] [--force]
  *              [--accept-rename <from>=<to>] (repeatable) [--no-reorder]
+ *              [--dry-run] [--verbose] [--out-plan <plan.json>]
  *   Read .sql files recursively (lexicographic), load into shadow, extract
  *   target, plan, apply.  Maps to old `declarative-apply` / `sync`.
  *
@@ -39,6 +40,34 @@
  *   --accept-rename <from>=<to>
  *     Confirm one rename candidate by the encoded stable-ids shown in a prior
  *     --renames prompt run.  Repeatable; each flag names one confirmed rename.
+ *
+ *   The statements actually applied to the target are NOT the authored
+ *   declarative SQL: the planner re-derives atomic DDL from the catalog diff
+ *   between the shadow (desired) and target (current) states. --verbose shows
+ *   every statement actually executed on the target connection — including
+ *   transaction framing (BEGIN/COMMIT/ROLLBACK) and session SETs — never the
+ *   authored files; --dry-run prints that exact script without applying it.
+ *
+ *   --dry-run
+ *     Plan as usual, then print the exact executable script (reusing the
+ *     `render` command's renderer, so it mirrors execution segment-for-segment)
+ *     to STDOUT instead of calling apply() — no fingerprint gate runs, nothing
+ *     is applied. A stderr summary reports the action count and flags any
+ *     destructive actions. Composes with --out-plan; --force is a no-op since
+ *     the gate never runs.
+ *   --verbose
+ *     During the real apply, stream a segment/action-level progress trace to
+ *     STDERR: segment start/end (with outcome), every planner-rendered action
+ *     (the SQL about to run and whether it succeeded, with wall-clock timing),
+ *     AND every other statement sent on the same connection — BEGIN, preamble
+ *     SET/SET LOCAL, COMMIT, ROLLBACK, RESET ALL — prefixed `  ; ` to stay
+ *     visually distinct from action lines. Wraps apply()'s `onEvent` observer
+ *     hook (src/apply/apply.ts) — purely additive, never changes what gets
+ *     applied or the final report.
+ *   --out-plan <plan.json>
+ *     Write the plan artifact (the same format `plan --out` produces) to this
+ *     path right after planning, before apply (or the --dry-run script). Useful
+ *     for inspecting/archiving the exact plan a `schema apply` run executed.
  */
 import {
   mkdirSync,
@@ -82,12 +111,14 @@ import {
   rewriteReorderedShadowError,
 } from "../reorder-display.ts";
 import { plan } from "../../plan/plan.ts";
+import { serializePlan } from "../../plan/artifact.ts";
 import { flattenPolicy, resolveView } from "../../policy/policy.ts";
 import {
   type ManagementScope,
   projectManagementScope,
 } from "../../policy/view.ts";
-import { apply } from "../../apply/apply.ts";
+import { apply, type ApplyEvent } from "../../apply/apply.ts";
+import { renderPlan } from "../render.ts";
 import { encodeId, parseId, type StableId } from "../../core/stable-id.ts";
 import { exitIfBlocking, printDiagnostics } from "../diagnostics.ts";
 import { makePool } from "../pool.ts";
@@ -484,7 +515,7 @@ export async function cmdSchemaExport(args: string[]): Promise<void> {
 }
 
 /** Discriminated result of {@link prepareApplyFiles}. */
-export type PreparedApplyFiles =
+type PreparedApplyFiles =
   | { ok: true; files: SqlFile[]; skipped: { file: string; stmt: string }[] }
   | { ok: false; message: string };
 
@@ -575,6 +606,9 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       scope: { type: "value" },
       "skip-cluster-ddl": { type: "boolean" },
       "keep-shadow": { type: "boolean" },
+      "dry-run": { type: "boolean" },
+      verbose: { type: "boolean" },
+      "out-plan": { type: "value" },
     });
   } catch (err) {
     if (err instanceof UsageError) {
@@ -582,6 +616,7 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
         `${err.message}\nUsage: pgdelta schema apply --dir <dir> --target <pg-url> [--shadow <pg-url>] ` +
           `[--renames auto|prompt|off] [--force] [--accept-rename <from>=<to>] ... ` +
           `[--profile ${PROFILE_IDS}] [--restrict-to-applier] [--strict-coverage] [--strict-function-bodies] [--no-reorder] [--unsafe-show-secrets] [--isolated-shadow] [--scope database|cluster] [--skip-cluster-ddl] [--keep-shadow]\n` +
+          `  [--dry-run] (print the executable script to stdout; apply nothing) [--verbose] (stream per-statement progress to stderr) [--out-plan <plan.json>] (write the plan artifact)\n` +
           `  --shadow omitted: a co-located shadow database is created on the target's cluster (database scope only) and dropped after.\n`,
       );
       process.exit(2);
@@ -595,6 +630,9 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
   const targetUrl = flags["target"];
   const force = flags["force"];
   const acceptRenameRaw = flags["accept-rename"];
+  const dryRun = flags["dry-run"];
+  const verbose = flags["verbose"];
+  const outPlanPath = flags["out-plan"];
 
   // The export directory's manifest (redaction mode, profile, scope), consulted
   // once and reused. Absent for hand-authored dirs / older exports.
@@ -1250,15 +1288,105 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       process.stderr.write("\n");
     }
 
+    // --out-plan: archive the plan artifact right after planning, regardless
+    // of whether the run goes on to --dry-run's script or a real apply.
+    if (outPlanPath !== undefined) {
+      writeFileSync(outPlanPath, serializePlan(thePlan), "utf8");
+      process.stderr.write(`Plan artifact written to ${outPlanPath}\n`);
+      if (flags["unsafe-show-secrets"]) {
+        process.stderr.write(
+          `  WARNING: --unsafe-show-secrets is set — the plan artifact contains UNREDACTED credentials.\n`,
+        );
+      }
+    }
+
     if (thePlan.actions.length === 0) {
       process.stderr.write("Target is already up to date.\n");
       return;
     }
 
+    if (dryRun) {
+      // Reuse the `render` command's renderer so the printed script splits on
+      // the SAME segment boundaries apply() uses at execution time —
+      // faithfully mirroring what a real apply would run, statement for
+      // statement. allowDrops:true: --dry-run is inspection only, so the
+      // render-time destructive-action gate (meant for `render`'s file-writing
+      // path) does not apply here; destructiveness is instead surfaced in the
+      // summary below.
+      const rendered = renderPlan(thePlan, { allowDrops: true });
+      const multi = rendered.files.length > 1;
+      const script = rendered.files
+        .map((file, i) => {
+          if (!multi) return file.contents;
+          const header = `-- pg-delta segment ${i + 1}/${rendered.files.length} (${
+            file.transactional ? "transactional" : "non-transactional"
+          })\n`;
+          return `${header}${file.contents}`;
+        })
+        .join("\n");
+      process.stdout.write(script);
+      process.stderr.write(
+        `Dry run: ${thePlan.actions.length} action(s) planned; nothing applied.\n`,
+      );
+      const destructiveCount = thePlan.actions.filter(
+        (a) => a.verb === "drop" || a.dataLoss === "destructive",
+      ).length;
+      if (destructiveCount > 0) {
+        process.stderr.write(
+          `WARNING: plan contains ${destructiveCount} destructive action(s).\n`,
+        );
+      }
+      return;
+    }
+
     if (force) {
       process.stderr.write("WARNING: --force disables the fingerprint gate.\n");
     }
 
+    // --verbose: stream per-segment/per-action progress to stderr as apply()
+    // executes. Purely additive (apply.ts's `onEvent` is guarded against a
+    // throwing observer) — never changes what gets applied or the report.
+    // `segmentEnd` doesn't carry the segment count, so remember it from the
+    // most recent `segmentStart` (events for one apply() call arrive in order).
+    let lastSegmentCount = 1;
+    const onEvent = verbose
+      ? (event: ApplyEvent): void => {
+          switch (event.kind) {
+            case "segmentStart":
+              lastSegmentCount = event.segmentCount;
+              process.stderr.write(
+                `-- segment ${event.segmentIndex + 1}/${event.segmentCount} (${
+                  event.transactional ? "transactional" : "non-transactional"
+                }), ${event.end - event.start} action(s)\n`,
+              );
+              break;
+            case "actionStart":
+              process.stderr.write(
+                `[${event.actionIndex + 1}/${thePlan.actions.length}] ${event.sql}\n`,
+              );
+              break;
+            case "actionEnd":
+              process.stderr.write(
+                `[${event.actionIndex + 1}/${thePlan.actions.length}] ${
+                  event.ok ? `ok (${event.ms}ms)` : `FAILED (${event.ms}ms)`
+                }\n`,
+              );
+              break;
+            case "segmentEnd":
+              process.stderr.write(
+                `-- segment ${event.segmentIndex + 1}/${lastSegmentCount} ${event.outcome}\n`,
+              );
+              break;
+            case "control":
+              // every OTHER statement apply() sends on the wire — BEGIN,
+              // preamble SET/SET LOCAL, COMMIT, ROLLBACK, RESET ALL — prefixed
+              // to stay visually distinct from `[i/total] <action sql>` lines.
+              process.stderr.write(`  ; ${event.sql}\n`);
+              break;
+          }
+        }
+      : undefined;
+
     const report = await apply(thePlan, tgt.pool, {
       ...ctx.applyOptions, // baseline + handler-aware re-extract (from the profile)
       // the fingerprint gate re-extracts the target and compares to the plan
@@ -1271,6 +1399,7 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       // view than the plan fingerprinted (§scope).
       reextract: (p) => ctx.extract(p, { redactSecrets }),
       fingerprintGate: !force,
+      ...(onEvent !== undefined ? { onEvent } : {}),
     });
 
     if (report.status === "applied") {
diff --git a/packages/pg-delta/src/cli/main.ts b/packages/pg-delta/src/cli/main.ts
index 1efcd09f2..6093a94ee 100644
--- a/packages/pg-delta/src/cli/main.ts
+++ b/packages/pg-delta/src/cli/main.ts
@@ -29,6 +29,7 @@
  *   schema apply   --dir <dir> --shadow <pg-url> --target <pg-url>
  *                  [--renames auto|prompt|off] [--force]
  *                  [--accept-rename <from>=<to>] ... [--no-reorder]
+ *                  [--dry-run] [--verbose] [--out-plan <plan.json>]
  *   schema lint    --dir <dir>
  *                  Statically check the SQL files (pg-topo) for shadow-load
  *                  cycles and other issues, without touching a database.
@@ -73,6 +74,7 @@ Commands:
   schema apply   --dir <dir> --shadow <pg-url> --target <pg-url>
                  [--renames auto|prompt|off] [--force]
                  [--accept-rename <from>=<to>] ... [--no-reorder]
+                 [--dry-run] [--verbose] [--out-plan <plan.json>]
   schema lint    --dir <dir>
 
 Notes:
@@ -114,6 +116,27 @@ Notes:
     raw files at file granularity. Reorder is on by default — it splits files
     into one-statement units and topologically pre-sorts them so authoring
     order within a file no longer matters.
+  schema apply: the applied statements are planner-rendered atomic DDL, not
+    the authored declarative SQL (the planner re-derives them from the
+    catalog diff between the shadow and target states). --verbose shows every
+    statement actually executed on the target connection, including
+    transaction framing and session SETs; --dry-run prints that exact script
+    without applying it.
+  --dry-run (schema apply): plan as usual, then print the exact executable
+    script to stdout (reusing "render"'s segment splitter, so it mirrors
+    execution statement-for-statement) instead of applying — no fingerprint
+    gate runs, nothing is applied. A stderr summary reports the action count
+    and flags destructive actions. Composes with --out-plan.
+  --verbose (schema apply): during the real apply, stream a segment/action
+    progress trace to stderr — segment start/end (with outcome), every
+    planner-rendered action (the SQL about to run and whether it succeeded,
+    with timing), and every other statement sent on the same connection
+    (BEGIN, preamble SET/SET LOCAL, COMMIT, ROLLBACK, RESET ALL), prefixed
+    "  ; " to stay distinct from action lines. Purely additive: never changes
+    what gets applied or the final report.
+  --out-plan <plan.json> (schema apply): write the plan artifact (same format
+    as "plan --out") to this path right after planning, before apply (or the
+    --dry-run script).
   --unsafe-show-secrets (plan, diff, drift, snapshot, schema export, schema apply):
     emit REAL foreign-data option values and subscription conninfo instead of
     redacted placeholders. Off by default; raises a loud warning when set.
diff --git a/packages/pg-delta/src/cli/render.ts b/packages/pg-delta/src/cli/render.ts
index b72f4efd0..853d305ef 100644
--- a/packages/pg-delta/src/cli/render.ts
+++ b/packages/pg-delta/src/cli/render.ts
@@ -12,7 +12,7 @@
 import { segmentActions } from "../apply/apply.ts";
 import type { Plan } from "../plan/plan.ts";
 
-export interface RenderOptions {
+interface RenderOptions {
   /** allow destructive actions to be rendered. Off by default: rendering a
    *  plan that drops or rewrites data without acknowledging it is a common way
    *  to ship a destructive migration by accident. Gates BOTH `drop`-verb actions
@@ -22,7 +22,7 @@ export interface RenderOptions {
   allowDrops: boolean;
 }
 
-export interface RenderedFile {
+interface RenderedFile {
   /** null for a single-segment plan (`<base>.sql`); "_1", "_2", … in
    *  execution order when the plan splits into multiple segments. */
   suffix: string | null;
@@ -31,7 +31,7 @@ export interface RenderedFile {
   actionCount: number;
 }
 
-export interface RenderResult {
+interface RenderResult {
   /** false only when the plan has zero actions. */
   changes: boolean;
   files: RenderedFile[];

From 243cacfd8573e981e183005e8fe43509ff01d4e9 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 16:23:17 +0000
Subject: [PATCH 3/9] chore(pg-delta): sync bun.lock with branch package
 versions

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 bun.lock | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/bun.lock b/bun.lock
index f9940da56..e6181f885 100644
--- a/bun.lock
+++ b/bun.lock
@@ -32,7 +32,7 @@
     },
     "packages/pg-delta": {
       "name": "@supabase/pg-delta",
-      "version": "1.0.0-alpha.31",
+      "version": "1.0.0-alpha.32",
       "bin": {
         "pgdelta": "./dist/cli/main.js",
       },
@@ -42,7 +42,7 @@
       },
       "devDependencies": {
         "@supabase/bun-istanbul-coverage": "workspace:*",
-        "@supabase/pg-topo": "^1.0.0-alpha.2",
+        "@supabase/pg-topo": "^1.0.0-alpha.3",
         "@types/bun": "^1.3.9",
         "@types/debug": "^4.1.12",
         "@types/node": "^24.10.4",
@@ -52,7 +52,7 @@
         "typescript": "^5.9.3",
       },
       "peerDependencies": {
-        "@supabase/pg-topo": "^1.0.0-alpha.2",
+        "@supabase/pg-topo": "^1.0.0-alpha.3",
       },
       "optionalPeers": [
         "@supabase/pg-topo",

From f89204cf0b1fa8d39b2f04718cffdb26075e75a6 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 16:46:32 +0000
Subject: [PATCH 4/9] test(pg-delta): pin observer event ordering on the
 non-transactional path

Review findings 1-3 on the schema-apply debugging feature: the trace lies
exactly on the CREATE INDEX CONCURRENTLY path it exists to debug.

RED (against 1cfcfbc):
- success path: preamble SET is emitted AFTER actionStart
    expect(pos).toBeLessThan(actionStartPos)
    error: expect(received).toBeLessThan(expected)
- failure path: control(RESET ALL) is emitted AFTER segmentEnd
    Expected: "segmentEnd"  Received: "control"
- preamble failure: action events emitted for an action that never ran
    expect(events.some((e) => e.kind === "actionStart")).toBe(false)
    Expected: false  Received: true

Also replaces the dead interleaving assertion in the happy-path test with
real actionStart/actionEnd pairing checks (finding 3).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 .../pg-delta/tests/apply-observer.test.ts     | 191 +++++++++++++++++-
 1 file changed, 185 insertions(+), 6 deletions(-)

diff --git a/packages/pg-delta/tests/apply-observer.test.ts b/packages/pg-delta/tests/apply-observer.test.ts
index 7263449cf..8846b7203 100644
--- a/packages/pg-delta/tests/apply-observer.test.ts
+++ b/packages/pg-delta/tests/apply-observer.test.ts
@@ -67,12 +67,23 @@ describe("apply() onEvent observer", () => {
         expect(e?.kind === "actionEnd" && e.ms).toBeGreaterThanOrEqual(0);
       }
 
-      // events interleave in the order apply() executes them: for every
-      // actionStart[i], its actionEnd[i] appears immediately after (modulo
-      // segment boundary events), never before the next actionStart.
-      const kinds = events.map((e) => e.kind);
-      const firstActionStartIdx = kinds.indexOf("actionStart");
-      expect(firstActionStartIdx).toBeGreaterThanOrEqual(0);
+      // events interleave in the order apply() executes them: every
+      // actionStart[i] is followed by its actionEnd[i] BEFORE the next
+      // actionStart[i+1] begins — a refactor that batches starts or ends
+      // would break the trace's statement-by-statement story.
+      const startPositions: number[] = [];
+      const endPositions: number[] = [];
+      events.forEach((e, pos) => {
+        if (e.kind === "actionStart") startPositions.push(pos);
+        else if (e.kind === "actionEnd") endPositions.push(pos);
+      });
+      expect(startPositions.length).toBe(endPositions.length);
+      for (let i = 0; i < startPositions.length; i++) {
+        expect(endPositions[i]!).toBeGreaterThan(startPositions[i]!);
+        if (i + 1 < startPositions.length) {
+          expect(endPositions[i]!).toBeLessThan(startPositions[i + 1]!);
+        }
+      }
     } finally {
       await Promise.all([target.drop(), desired.drop()]);
     }
@@ -199,6 +210,174 @@ describe("apply() onEvent observer", () => {
     }
   }, 60_000);
 
+  // The non-transactional path (CREATE INDEX CONCURRENTLY under the
+  // `concurrentIndexes` param) is the one this trace exists to debug — a
+  // cancelled CIC leaves durable side effects — so its event ordering must
+  // mirror the wire exactly: preamble SETs before the action, actionEnd when
+  // the action settles (not after RESET ALL), segmentEnd terminal on every
+  // path.
+  test("non-transactional segment: preamble controls precede actionStart; actionEnd precedes RESET ALL; segmentEnd is terminal", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_ntx_tgt");
+    const desired = await cluster.createDb("apply_observer_ntx_desired");
+    try {
+      await target.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+      `);
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+        CREATE INDEX t_x_idx ON app.t (x);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase, {
+        params: { concurrentIndexes: true },
+      });
+      // the whole plan is the one CREATE INDEX CONCURRENTLY action, so the
+      // event stream is exactly one non-transactional segment.
+      expect(thePlan.actions.length).toBe(1);
+      expect(thePlan.actions[0]?.transactionality).toBe("nonTransactional");
+
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("applied");
+
+      const kinds = events.map((e) => e.kind);
+      const actionStartPos = kinds.indexOf("actionStart");
+      const actionEndPos = kinds.indexOf("actionEnd");
+      const resetPos = events.findIndex(
+        (e) => e.kind === "control" && e.sql === "RESET ALL",
+      );
+      expect(actionStartPos).toBeGreaterThanOrEqual(0);
+      expect(resetPos).toBeGreaterThanOrEqual(0);
+      // wire order: every session-level preamble SET goes out BEFORE the
+      // action statement …
+      let sawPreambleSet = false;
+      events.forEach((e, pos) => {
+        if (e.kind === "control" && e.sql.startsWith("SET ")) {
+          sawPreambleSet = true;
+          expect(pos).toBeLessThan(actionStartPos);
+        }
+      });
+      expect(sawPreambleSet).toBe(true); // plan.preamble is never empty
+      // … and actionEnd fires when the action settles, BEFORE the RESET ALL
+      // round-trip (so `ms` measures only the action itself) …
+      expect(actionEndPos).toBeGreaterThan(actionStartPos);
+      expect(resetPos).toBeGreaterThan(actionEndPos);
+      // … and segmentEnd is the segment's terminal event on the success path.
+      const last = events[events.length - 1]!;
+      expect(last.kind).toBe("segmentEnd");
+      expect(last.kind === "segmentEnd" && last.outcome).toBe("committed");
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
+  test("non-transactional failure: actionEnd(ok:false), then RESET ALL, then segmentEnd(inDoubt) as the terminal event", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_ntx_fail_tgt");
+    const desired = await cluster.createDb("apply_observer_ntx_fail_desired");
+    try {
+      await target.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+      `);
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+        CREATE INDEX t_x_idx ON app.t (x);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase, {
+        params: { concurrentIndexes: true },
+      });
+      expect(thePlan.actions.length).toBe(1);
+      expect(thePlan.actions[0]?.transactionality).toBe("nonTransactional");
+
+      // sabotage: pre-create the index name so CREATE INDEX CONCURRENTLY fails
+      await target.pool.query(`CREATE INDEX t_x_idx ON app.t (id)`);
+
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("failed");
+      expect(report.actionStatuses[0]).toBe("inDoubt");
+
+      const actionEndPos = events.findIndex((e) => e.kind === "actionEnd");
+      const resetPos = events.findIndex(
+        (e) => e.kind === "control" && e.sql === "RESET ALL",
+      );
+      const failedEnd = events[actionEndPos];
+      expect(failedEnd?.kind === "actionEnd" && failedEnd.ok).toBe(false);
+      // RESET ALL still hits the wire after the failure, and segmentEnd stays
+      // the segment's terminal event — a `--verbose` trace must never print
+      // wire traffic after the segment's outcome line.
+      expect(resetPos).toBeGreaterThan(actionEndPos);
+      const last = events[events.length - 1]!;
+      expect(last.kind).toBe("segmentEnd");
+      expect(last.kind === "segmentEnd" && last.outcome).toBe("inDoubt");
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
+  test("non-transactional preamble failure: no action events for an action that never reached the wire", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_ntx_pre_tgt");
+    const desired = await cluster.createDb("apply_observer_ntx_pre_desired");
+    try {
+      await target.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+      `);
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer, x text);
+        CREATE INDEX t_x_idx ON app.t (x);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase, {
+        params: { concurrentIndexes: true },
+      });
+      expect(thePlan.actions[0]?.transactionality).toBe("nonTransactional");
+
+      // lock_timeout = -1 is outside Postgres's valid range, so the segment's
+      // FIRST preamble SET fails — the action itself never reaches the wire.
+      const events: ApplyEvent[] = [];
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        lockTimeoutMs: -1,
+        onEvent: (e) => events.push(e),
+      });
+      expect(report.status).toBe("failed");
+
+      // the trace must not claim an action executed and failed when it was
+      // the preamble that failed: no actionStart, no actionEnd.
+      expect(events.some((e) => e.kind === "actionStart")).toBe(false);
+      expect(events.some((e) => e.kind === "actionEnd")).toBe(false);
+      const last = events[events.length - 1]!;
+      expect(last.kind).toBe("segmentEnd");
+      expect(last.kind === "segmentEnd" && last.outcome).toBe("inDoubt");
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
   test("an observer that always throws never changes apply()'s outcome", async () => {
     const cluster = await sharedCluster();
     const targetA = await cluster.createDb("apply_observer_safe_a");

From 43fac760952f26af08e22cefd3e68e2c0244953e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 16:46:32 +0000
Subject: [PATCH 5/9] fix(pg-delta): make the apply observer trace mirror wire
 order on the non-transactional path

Review follow-ups on the schema-apply debugging feature (findings 1-7):

- non-transactional segments now emit preamble control events BEFORE
  actionStart, emit actionEnd as the action settles (so ms measures only
  the action's round-trip, not RESET ALL), emit NO action events when the
  preamble itself fails, and always emit segmentEnd as the segment's
  terminal event (after the RESET ALL control) on success and failure.
  The ApplyReport is unchanged on every path.
- schema apply --dry-run now warns on stderr when --unsafe-show-secrets
  is set (the script is as persistent as the --out-plan artifact).
- destructive-action predicate shared via isDestructiveAction() in
  cli/render.ts instead of being duplicated in the --dry-run summary.
- docs no longer overstate --dry-run as 'that exact script' (transaction
  framing is shown by --verbose only) and note that --verbose has no
  effect under --dry-run.
- cli.test.ts pins the '  ; ' control-line prefix for BEGIN/COMMIT.

RED (captured before this fix, tests in the previous commit):
  expect(pos).toBeLessThan(actionStartPos)   // SET emitted after actionStart
  Expected: "segmentEnd"  Received: "control" // RESET ALL after segmentEnd
  Expected: false  Received: true            // action events without an action

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 packages/pg-delta/src/apply/apply.ts         | 63 ++++++++++++--------
 packages/pg-delta/src/cli/commands/schema.ts | 32 ++++++----
 packages/pg-delta/src/cli/main.ts            | 16 ++---
 packages/pg-delta/src/cli/render.ts          | 14 ++++-
 packages/pg-delta/tests/cli.test.ts          |  8 ++-
 5 files changed, 85 insertions(+), 48 deletions(-)

diff --git a/packages/pg-delta/src/apply/apply.ts b/packages/pg-delta/src/apply/apply.ts
index 24bb02106..b8e303eb3 100644
--- a/packages/pg-delta/src/apply/apply.ts
+++ b/packages/pg-delta/src/apply/apply.ts
@@ -276,30 +276,57 @@ export async function apply(
         // a lone non-transactional action; session-level settings, reset after
         const index = segment.start;
         const action = thePlan.actions[index]!;
-        const actionStartedAt = Date.now();
-        emit(onEvent, {
-          kind: "actionStart",
-          actionIndex: index,
-          sql: action.sql,
-        });
+        // the failure return is deferred past the finally so segmentEnd stays
+        // the segment's LAST event — after the RESET ALL control — on every
+        // path; a trace must never show wire traffic after the outcome line.
+        let failure: NonNullable<ApplyReport["error"]> | undefined;
         try {
+          // session-level preamble SETs hit the wire BEFORE the action; a
+          // preamble failure emits NO action events (the action never ran).
           for (const sql of preamble(false)) {
             emit(onEvent, { kind: "control", sql });
             await client.query(sql);
           }
-          await client.query(action.sql);
-        } catch (error) {
+          const actionStartedAt = Date.now();
           emit(onEvent, {
-            kind: "actionEnd",
+            kind: "actionStart",
             actionIndex: index,
-            ok: false,
-            ms: Date.now() - actionStartedAt,
+            sql: action.sql,
           });
+          try {
+            await client.query(action.sql);
+            // actionEnd fires as the action settles, so `ms` measures only the
+            // action's round-trip (not the RESET ALL below).
+            emit(onEvent, {
+              kind: "actionEnd",
+              actionIndex: index,
+              ok: true,
+              ms: Date.now() - actionStartedAt,
+            });
+          } catch (error) {
+            emit(onEvent, {
+              kind: "actionEnd",
+              actionIndex: index,
+              ok: false,
+              ms: Date.now() - actionStartedAt,
+            });
+            throw error;
+          }
+        } catch (error) {
           // a failed non-transactional DDL is NOT safely unapplied — it can
           // leave durable side effects (e.g. an INVALID index from a cancelled
           // CREATE INDEX CONCURRENTLY). Report it inDoubt so the caller knows
           // the database must be re-extracted before retry (review P1).
           statuses[index] = "inDoubt";
+          failure = errorEntry(index, action.sql, error);
+        } finally {
+          // ALWAYS restore session state before the client returns to the pool,
+          // on success or failure — RESET ALL must not be skipped on the
+          // failure path, and a reset failure must not flip the action's outcome.
+          emit(onEvent, { kind: "control", sql: "RESET ALL" });
+          await client.query("RESET ALL").catch(() => {});
+        }
+        if (failure !== undefined) {
           emit(onEvent, {
             kind: "segmentEnd",
             segmentIndex: segIdx,
@@ -309,21 +336,9 @@ export async function apply(
             status: "failed",
             appliedActions,
             actionStatuses: statuses,
-            error: errorEntry(index, action.sql, error),
+            error: failure,
           };
-        } finally {
-          // ALWAYS restore session state before the client returns to the pool,
-          // on success or failure — RESET ALL must not be skipped by the catch's
-          // early return, and a reset failure must not flip the action's outcome.
-          emit(onEvent, { kind: "control", sql: "RESET ALL" });
-          await client.query("RESET ALL").catch(() => {});
         }
-        emit(onEvent, {
-          kind: "actionEnd",
-          actionIndex: index,
-          ok: true,
-          ms: Date.now() - actionStartedAt,
-        });
         statuses[index] = "applied";
         appliedActions += 1;
         emit(onEvent, {
diff --git a/packages/pg-delta/src/cli/commands/schema.ts b/packages/pg-delta/src/cli/commands/schema.ts
index a47ccb0bf..f3b75a6ca 100644
--- a/packages/pg-delta/src/cli/commands/schema.ts
+++ b/packages/pg-delta/src/cli/commands/schema.ts
@@ -46,15 +46,19 @@
  *   between the shadow (desired) and target (current) states. --verbose shows
  *   every statement actually executed on the target connection — including
  *   transaction framing (BEGIN/COMMIT/ROLLBACK) and session SETs — never the
- *   authored files; --dry-run prints that exact script without applying it.
+ *   authored files; --dry-run prints the same statements, split on the same
+ *   segment boundaries, without applying them (transaction framing and
+ *   RESET ALL are shown by --verbose only; the preamble appears as plain
+ *   session `set`s repeated per segment).
  *
  *   --dry-run
- *     Plan as usual, then print the exact executable script (reusing the
- *     `render` command's renderer, so it mirrors execution segment-for-segment)
- *     to STDOUT instead of calling apply() — no fingerprint gate runs, nothing
- *     is applied. A stderr summary reports the action count and flags any
- *     destructive actions. Composes with --out-plan; --force is a no-op since
- *     the gate never runs.
+ *     Plan as usual, then print the executable script (reusing the `render`
+ *     command's renderer, so it splits on the same segment boundaries apply()
+ *     executes) to STDOUT instead of calling apply() — no fingerprint gate
+ *     runs, nothing is applied. A stderr summary reports the action count and
+ *     flags any destructive actions. Composes with --out-plan; --force is a
+ *     no-op since the gate never runs, and --verbose has no effect (nothing
+ *     executes, so there is no trace).
  *   --verbose
  *     During the real apply, stream a segment/action-level progress trace to
  *     STDERR: segment start/end (with outcome), every planner-rendered action
@@ -118,7 +122,7 @@ import {
   projectManagementScope,
 } from "../../policy/view.ts";
 import { apply, type ApplyEvent } from "../../apply/apply.ts";
-import { renderPlan } from "../render.ts";
+import { isDestructiveAction, renderPlan } from "../render.ts";
 import { encodeId, parseId, type StableId } from "../../core/stable-id.ts";
 import { exitIfBlocking, printDiagnostics } from "../diagnostics.ts";
 import { makePool } from "../pool.ts";
@@ -1328,9 +1332,15 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       process.stderr.write(
         `Dry run: ${thePlan.actions.length} action(s) planned; nothing applied.\n`,
       );
-      const destructiveCount = thePlan.actions.filter(
-        (a) => a.verb === "drop" || a.dataLoss === "destructive",
-      ).length;
+      if (flags["unsafe-show-secrets"]) {
+        // the dry-run script is routinely redirected to a file, making it as
+        // persistent as the --out-plan artifact warned about above.
+        process.stderr.write(
+          `  WARNING: --unsafe-show-secrets is set — the dry-run script contains UNREDACTED credentials.\n`,
+        );
+      }
+      const destructiveCount =
+        thePlan.actions.filter(isDestructiveAction).length;
       if (destructiveCount > 0) {
         process.stderr.write(
           `WARNING: plan contains ${destructiveCount} destructive action(s).\n`,
diff --git a/packages/pg-delta/src/cli/main.ts b/packages/pg-delta/src/cli/main.ts
index 6093a94ee..92c2f7819 100644
--- a/packages/pg-delta/src/cli/main.ts
+++ b/packages/pg-delta/src/cli/main.ts
@@ -120,13 +120,15 @@ Notes:
     the authored declarative SQL (the planner re-derives them from the
     catalog diff between the shadow and target states). --verbose shows every
     statement actually executed on the target connection, including
-    transaction framing and session SETs; --dry-run prints that exact script
-    without applying it.
-  --dry-run (schema apply): plan as usual, then print the exact executable
-    script to stdout (reusing "render"'s segment splitter, so it mirrors
-    execution statement-for-statement) instead of applying — no fingerprint
-    gate runs, nothing is applied. A stderr summary reports the action count
-    and flags destructive actions. Composes with --out-plan.
+    transaction framing and session SETs; --dry-run prints the same
+    statements, split on the same segment boundaries, without applying them
+    (transaction framing is shown by --verbose only).
+  --dry-run (schema apply): plan as usual, then print the executable script
+    to stdout (reusing "render"'s segment splitter, so it splits on the same
+    boundaries apply() executes) instead of applying — no fingerprint gate
+    runs, nothing is applied. A stderr summary reports the action count and
+    flags destructive actions. Composes with --out-plan; --verbose has no
+    effect under --dry-run (nothing executes, so there is no trace).
   --verbose (schema apply): during the real apply, stream a segment/action
     progress trace to stderr — segment start/end (with outcome), every
     planner-rendered action (the SQL about to run and whether it succeeded,
diff --git a/packages/pg-delta/src/cli/render.ts b/packages/pg-delta/src/cli/render.ts
index 853d305ef..1ed7a7635 100644
--- a/packages/pg-delta/src/cli/render.ts
+++ b/packages/pg-delta/src/cli/render.ts
@@ -37,6 +37,16 @@ interface RenderResult {
   files: RenderedFile[];
 }
 
+/** The planner marks destructive work two ways: a `drop` verb, or a non-drop
+ *  verb flagged `dataLoss: "destructive"` (e.g. an enum value-set migration
+ *  that rewrites dependent columns — verb `alter`). Shared by the render gate
+ *  below and `schema apply --dry-run`'s warning so the predicate cannot drift. */
+export function isDestructiveAction(
+  action: Pick<Plan["actions"][number], "verb" | "dataLoss">,
+): boolean {
+  return action.verb === "drop" || action.dataLoss === "destructive";
+}
+
 /** Normalize action SQL to end with exactly one semicolon (action.sql may or
  *  may not already have a trailing `;`, and may have trailing whitespace). */
 function terminate(sql: string): string {
@@ -53,9 +63,7 @@ export function renderPlan(plan: Plan, opts: RenderOptions): RenderResult {
     // Gate on the plan's own safety metadata (dataLoss), not just the verb: a
     // destructive action can be a non-`drop` verb (an enum value-set migration
     // rewriting columns is an `alter`), and those would otherwise slip through.
-    const offender = plan.actions.find(
-      (a) => a.verb === "drop" || a.dataLoss === "destructive",
-    );
+    const offender = plan.actions.find(isDestructiveAction);
     if (offender !== undefined) {
       const why =
         offender.verb === "drop" ? "a drop action" : "a destructive action";
diff --git a/packages/pg-delta/tests/cli.test.ts b/packages/pg-delta/tests/cli.test.ts
index a2c1907bd..142772d2a 100644
--- a/packages/pg-delta/tests/cli.test.ts
+++ b/packages/pg-delta/tests/cli.test.ts
@@ -1805,9 +1805,11 @@ describe("CLI: schema apply debugging", () => {
       expect(res.stderr).toContain("ok (");
       // --verbose is a COMPLETE record of the wire, not just plan actions: the
       // applied statements are planner-rendered atomic DDL, so the trace must
-      // also show the transaction framing actually sent (BEGIN/COMMIT).
-      expect(res.stderr).toContain("BEGIN");
-      expect(res.stderr).toContain("COMMIT");
+      // also show the transaction framing actually sent (BEGIN/COMMIT) — with
+      // the documented "  ; " control prefix that keeps those lines visually
+      // distinct from `[i/total] <action sql>` lines.
+      expect(res.stderr).toContain("  ; BEGIN");
+      expect(res.stderr).toContain("  ; COMMIT");
 
       const { rows } = await target.pool.query<{ n: number }>(
         `SELECT count(*)::int AS n FROM pg_tables WHERE schemaname = 'app'`,

From 971d5d6a0082ce19d6fe66d06f3218f35bdbc342 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 17:07:38 +0000
Subject: [PATCH 6/9] test(pg-delta): pin manifest-driven secrets warning and
 public ApplyEvent export

Final-review findings N1/N2 on the schema-apply debugging feature.

RED (against 43fac76):
- a dir exported with --unsafe-show-secrets (manifest redactSecrets:false)
  re-applied WITHOUT the flag gets no UNREDACTED warning on --dry-run /
  --out-plan:
    expect(warnings.length).toBe(2)
    Expected: 2  Received: 0
- ApplyEvent is advertised as library API by the changeset but not
  exported from the package entry:
    tests/apply-observer.test.ts(16,15): error TS2305: Module
    '"../src/index.ts"' has no exported member 'ApplyEvent'.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 .../pg-delta/tests/apply-observer.test.ts     |  5 +-
 packages/pg-delta/tests/cli.test.ts           | 54 +++++++++++++++++++
 2 files changed, 58 insertions(+), 1 deletion(-)

diff --git a/packages/pg-delta/tests/apply-observer.test.ts b/packages/pg-delta/tests/apply-observer.test.ts
index 8846b7203..ebf490e73 100644
--- a/packages/pg-delta/tests/apply-observer.test.ts
+++ b/packages/pg-delta/tests/apply-observer.test.ts
@@ -10,7 +10,10 @@
  * production execution path, not a hand-rolled fixture.
  */
 import { describe, expect, test } from "bun:test";
-import { apply, type ApplyEvent } from "../src/apply/apply.ts";
+import { apply } from "../src/apply/apply.ts";
+// ApplyEvent is advertised as public library API (see the changeset), so pin
+// that it is exported from the package entry, not just the internal module.
+import type { ApplyEvent } from "../src/index.ts";
 import { extract } from "../src/extract/extract.ts";
 import { plan } from "../src/plan/plan.ts";
 import { sharedCluster } from "./containers.ts";
diff --git a/packages/pg-delta/tests/cli.test.ts b/packages/pg-delta/tests/cli.test.ts
index 142772d2a..78f4e07d1 100644
--- a/packages/pg-delta/tests/cli.test.ts
+++ b/packages/pg-delta/tests/cli.test.ts
@@ -1819,4 +1819,58 @@ describe("CLI: schema apply debugging", () => {
       await Promise.all([shadow.drop(), target.drop()]);
     }
   }, 90_000);
+
+  test("--dry-run warns about unredacted credentials when the export MANIFEST recorded the unredacted mode (no flag re-passed)", async () => {
+    // The effective redaction mode is `manifest?.redactSecrets ?? !flag`: a
+    // dir exported with --unsafe-show-secrets stamps redactSecrets:false in
+    // its manifest and is re-applied unredacted WITHOUT the operator
+    // re-passing the flag. The dry-run script (and --out-plan artifact) then
+    // carry real credentials, so the warning must key on the effective mode,
+    // not on the flag.
+    const cluster = await sharedCluster();
+    const source = await cluster.createDb("cli_apply_dryrun_unred_src");
+    const target = await cluster.createDb("cli_apply_dryrun_unred_tgt");
+    try {
+      await source.pool.query(
+        `CREATE SCHEMA app; CREATE TABLE app.t (id integer PRIMARY KEY);`,
+      );
+      const dir = join(tmpdir(), `pg-delta-next-dryrun-unred-${Date.now()}`);
+      const exported = await runCli([
+        "schema",
+        "export",
+        "--source",
+        source.uri,
+        "--out-dir",
+        dir,
+        "--unsafe-show-secrets",
+      ]);
+      expect(exported.exitCode).toBe(0);
+
+      const planPath = join(dir, "plan.json");
+      // NOTE: no --unsafe-show-secrets here — the manifest supplies the mode.
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--dry-run",
+        "--out-plan",
+        planPath,
+      ]);
+      expect({ code: res.exitCode, stderr: res.stderr }).toMatchObject({
+        code: 0,
+      });
+      // both unredacted output channels warn: the plan artifact and the script
+      const warnings = res.stderr
+        .split("\n")
+        .filter((l) => l.includes("UNREDACTED"));
+      expect(warnings.length).toBe(2);
+    } finally {
+      await Promise.all([source.drop(), target.drop()]);
+    }
+  }, 90_000);
 });

From b42ee070cb27d25755796332bd1e0df40d338ca7 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 17:09:00 +0000
Subject: [PATCH 7/9] fix(pg-delta): key unredacted-secrets warnings on the
 effective mode and export ApplyEvent
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Final-review findings N1/N2:
- the --dry-run / --out-plan UNREDACTED warnings now gate on the effective
  redaction mode (manifest?.redactSecrets ?? !flag) instead of the flag
  alone, so a dir exported with --unsafe-show-secrets warns when re-applied
  without the flag, and a redacted-manifest dir no longer false-warns.
- ApplyEvent is exported from the package entry (src/index.ts), matching
  the changeset's advertised library API.

RED (captured before this fix, tests in the previous commit):
  expect(warnings.length).toBe(2) — Expected: 2  Received: 0
  TS2305: Module '"../src/index.ts"' has no exported member 'ApplyEvent'.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 packages/pg-delta/src/cli/commands/schema.ts | 14 +++++++++-----
 packages/pg-delta/src/index.ts               |  1 +
 2 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/packages/pg-delta/src/cli/commands/schema.ts b/packages/pg-delta/src/cli/commands/schema.ts
index f3b75a6ca..808b483e8 100644
--- a/packages/pg-delta/src/cli/commands/schema.ts
+++ b/packages/pg-delta/src/cli/commands/schema.ts
@@ -1297,9 +1297,12 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
     if (outPlanPath !== undefined) {
       writeFileSync(outPlanPath, serializePlan(thePlan), "utf8");
       process.stderr.write(`Plan artifact written to ${outPlanPath}\n`);
-      if (flags["unsafe-show-secrets"]) {
+      // key on the EFFECTIVE mode, not the flag: the export manifest can
+      // carry redactSecrets:false without the operator re-passing
+      // --unsafe-show-secrets (and a redacted manifest overrides the flag).
+      if (!redactSecrets) {
         process.stderr.write(
-          `  WARNING: --unsafe-show-secrets is set — the plan artifact contains UNREDACTED credentials.\n`,
+          `  WARNING: secrets are unredacted (--unsafe-show-secrets or the export manifest) — the plan artifact contains UNREDACTED credentials.\n`,
         );
       }
     }
@@ -1332,11 +1335,12 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       process.stderr.write(
         `Dry run: ${thePlan.actions.length} action(s) planned; nothing applied.\n`,
       );
-      if (flags["unsafe-show-secrets"]) {
+      if (!redactSecrets) {
         // the dry-run script is routinely redirected to a file, making it as
-        // persistent as the --out-plan artifact warned about above.
+        // persistent as the --out-plan artifact warned about above; same
+        // effective-mode (manifest-aware) gate as that warning.
         process.stderr.write(
-          `  WARNING: --unsafe-show-secrets is set — the dry-run script contains UNREDACTED credentials.\n`,
+          `  WARNING: secrets are unredacted (--unsafe-show-secrets or the export manifest) — the dry-run script contains UNREDACTED credentials.\n`,
         );
       }
       const destructiveCount =
diff --git a/packages/pg-delta/src/index.ts b/packages/pg-delta/src/index.ts
index 0472bf806..84266755b 100644
--- a/packages/pg-delta/src/index.ts
+++ b/packages/pg-delta/src/index.ts
@@ -53,6 +53,7 @@ export {
   apply,
   type ApplyReport,
   type ApplyOptions,
+  type ApplyEvent,
   type ActionStatus,
 } from "./apply/apply.ts";
 

From 292395c838b6264e058a52f7877b234a1087dfa2 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 21:00:15 +0000
Subject: [PATCH 8/9] test(pg-delta): pin out-plan redaction stamping and
 async-observer rejection safety

Codex review findings on PR #343 (both P2, verified real).

RED (against 2ac8de7):
- the --out-plan artifact omits the redaction mode it was fingerprinted
  under (apply --plan then re-extracts redacted and trips the gate):
    expect(parsed.redactSecrets).toBe(true)   Expected: true   Received: undefined
    expect(parsed.redactSecrets).toBe(false)  Expected: false  Received: undefined
- an async onEvent observer's rejection escapes emit()'s synchronous
  try/catch as an unhandled rejection:
    error: async observer boom
    # Unhandled error between tests

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 .../pg-delta/tests/apply-observer.test.ts     | 43 +++++++++++++++++++
 packages/pg-delta/tests/cli.test.ts           | 11 +++++
 2 files changed, 54 insertions(+)

diff --git a/packages/pg-delta/tests/apply-observer.test.ts b/packages/pg-delta/tests/apply-observer.test.ts
index ebf490e73..32270bf43 100644
--- a/packages/pg-delta/tests/apply-observer.test.ts
+++ b/packages/pg-delta/tests/apply-observer.test.ts
@@ -381,6 +381,49 @@ describe("apply() onEvent observer", () => {
     }
   }, 60_000);
 
+  test("an ASYNC observer that rejects never surfaces an unhandled rejection or changes the outcome", async () => {
+    // TypeScript allows an async function wherever a `void` callback is
+    // expected, so a library caller can pass `onEvent: async () => {...}`.
+    // A rejection from it is invisible to emit()'s synchronous try/catch —
+    // it must be consumed, or it escapes as an unhandled rejection and can
+    // take down the process mid-apply.
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("apply_observer_async_tgt");
+    const desired = await cluster.createDb("apply_observer_async_desired");
+    const rejections: unknown[] = [];
+    const onRejection = (reason: unknown): void => {
+      rejections.push(reason);
+    };
+    process.on("unhandledRejection", onRejection);
+    try {
+      await desired.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TABLE app.t (id integer PRIMARY KEY);
+      `);
+      const [targetState, desiredState] = [
+        await extract(target.pool),
+        await extract(desired.pool),
+      ];
+      const thePlan = plan(targetState.factBase, desiredState.factBase);
+      expect(thePlan.actions.length).toBeGreaterThan(0);
+
+      const report = await apply(thePlan, target.pool, {
+        fingerprintGate: false,
+        onEvent: async () => {
+          throw new Error("async observer boom");
+        },
+      });
+      expect(report.status).toBe("applied");
+
+      // give any escaped rejection a macrotask to reach the process handler
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      expect(rejections).toEqual([]);
+    } finally {
+      process.off("unhandledRejection", onRejection);
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+
   test("an observer that always throws never changes apply()'s outcome", async () => {
     const cluster = await sharedCluster();
     const targetA = await cluster.createDb("apply_observer_safe_a");
diff --git a/packages/pg-delta/tests/cli.test.ts b/packages/pg-delta/tests/cli.test.ts
index 8df5969fd..16bc518ca 100644
--- a/packages/pg-delta/tests/cli.test.ts
+++ b/packages/pg-delta/tests/cli.test.ts
@@ -1853,6 +1853,10 @@ describe("CLI: schema apply debugging", () => {
 
       const parsed = parsePlan(readFileSync(planPath, "utf8"));
       expect(parsed.actions.length).toBeGreaterThan(0);
+      // the artifact records the redaction mode it was fingerprinted under, so
+      // a later `pgdelta apply --plan` re-extracts in the SAME mode instead of
+      // defaulting to redacted and tripping the fingerprint gate.
+      expect(parsed.redactSecrets).toBe(true);
 
       // still nothing applied
       const { rows } = await target.pool.query<{ n: number }>(
@@ -1960,6 +1964,13 @@ describe("CLI: schema apply debugging", () => {
         .split("\n")
         .filter((l) => l.includes("UNREDACTED"));
       expect(warnings.length).toBe(2);
+
+      // and the artifact STAMPS the unredacted mode: its fingerprint was taken
+      // from unredacted extracts, so `pgdelta apply --plan` must re-extract
+      // unredacted too — an absent field reads as redacted and the gate would
+      // spuriously reject an unchanged target.
+      const parsed = parsePlan(readFileSync(planPath, "utf8"));
+      expect(parsed.redactSecrets).toBe(false);
     } finally {
       await Promise.all([source.drop(), target.drop()]);
     }

From c27dc3a5354e61a24e894e20511b54d8b7edfb32 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 17 Jul 2026 21:02:53 +0000
Subject: [PATCH 9/9] fix(pg-delta): stamp redaction mode on schema-apply plans
 and consume async observer rejections
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex review findings on PR #343:

- planSchemaFiles now passes the effective redactSecrets into plan(), so a
  plan archived via 'schema apply --out-plan' records the mode its
  fingerprint was taken under and 'apply --plan' re-extracts in that same
  mode — an unstamped unredacted plan read as redacted and the fingerprint
  gate spuriously rejected an unchanged target.
- emit() consumes a thenable returned by the onEvent observer (TypeScript
  allows async functions for void callbacks), so an async observer's
  rejection can no longer escape as an unhandled rejection and take down
  the process mid-apply.

RED (captured before this fix, tests in the previous commit):
  expect(parsed.redactSecrets).toBe(false)  Expected: false  Received: undefined
  error: async observer boom / # Unhandled error between tests

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JwpD8FPVBzVv2Qs8giVSBm
---
 packages/pg-delta/src/apply/apply.ts           | 13 ++++++++++++-
 packages/pg-delta/src/frontends/schema-plan.ts |  6 ++++++
 2 files changed, 18 insertions(+), 1 deletion(-)

diff --git a/packages/pg-delta/src/apply/apply.ts b/packages/pg-delta/src/apply/apply.ts
index b8e303eb3..937274dbe 100644
--- a/packages/pg-delta/src/apply/apply.ts
+++ b/packages/pg-delta/src/apply/apply.ts
@@ -153,7 +153,18 @@ function emit(
 ): void {
   if (onEvent === undefined) return;
   try {
-    onEvent(event);
+    // a `void`-typed callback may still be an async function (TypeScript
+    // allows it), whose rejection is invisible to this try/catch — consume a
+    // returned thenable so it cannot surface as an unhandled rejection and
+    // take down the process mid-apply.
+    const result = onEvent(event) as unknown;
+    if (
+      result !== null &&
+      (typeof result === "object" || typeof result === "function") &&
+      typeof (result as PromiseLike<unknown>).then === "function"
+    ) {
+      Promise.resolve(result as PromiseLike<unknown>).catch(() => {});
+    }
   } catch {
     // swallowed by design — see doc comment above
   }
diff --git a/packages/pg-delta/src/frontends/schema-plan.ts b/packages/pg-delta/src/frontends/schema-plan.ts
index ff10df97e..1c344b3e1 100644
--- a/packages/pg-delta/src/frontends/schema-plan.ts
+++ b/packages/pg-delta/src/frontends/schema-plan.ts
@@ -542,6 +542,12 @@ export async function planSchemaFiles(
     ...(applyDefaultOwner !== undefined
       ? { defaultOwner: applyDefaultOwner }
       : {}),
+    // stamp the redaction mode the extracts (and thus the fingerprint) were
+    // taken under, so a plan archived via `schema apply --out-plan` replays
+    // through `apply --plan` with the SAME re-extract mode — an unstamped
+    // unredacted plan would read as redacted and trip the fingerprint gate
+    // on an unchanged target.
+    redactSecrets,
   };
 
   const thePlan = plan(targetResult.factBase, loadResult.factBase, planOptions);