diff --git a/.changeset/schema-apply-statement-debugging.md b/.changeset/schema-apply-statement-debugging.md
new file mode 100644
index 00000000..3287ada4
--- /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 a portable SQL apply script (including transaction framing, full per-segment preambles, and cleanup) to stdout without applying anything or running the fingerprint gate. The script records its execution contract in comments: dispatch statements in order on one session, stop at the first error, and preserve autocommit outside explicit transaction blocks; do not send it as one multi-statement request or wrap it in one global transaction. `--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. When secret redaction is disabled by either the flag or export manifest, every output surface that may reveal credentials emits an explicit warning. 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. Apply failures identify the exact action or executor control statement that failed, keep setup failures unapplied, preserve successful non-transactional actions when session cleanup fails, and discard connections whose cleanup could not be verified.
diff --git a/docs/getting-started.md b/docs/getting-started.md
index 2e8153a2..2b19504e 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -139,6 +139,27 @@ pgdelta schema apply \
   --target "$TARGET_URL"     # the database to migrate
 ```
 
+Use `--dry-run` to write the portable apply script to stdout without changing
+the target, `--verbose` to stream the real apply's segment and statement trace
+to stderr, and `--out-plan <plan.json>` to archive the exact plan immediately
+after planning. Status, warnings, and diagnostics stay on stderr, so stdout from
+`--dry-run` can be redirected verbatim:
+
+```bash
+pgdelta schema apply --dir ./schema --target "$TARGET_URL" --dry-run > apply.sql
+```
+
+The script must run statement by statement, in order, on one database session.
+The runner must stop on the first error and preserve autocommit outside the
+script's explicit `BEGIN`/`COMMIT` blocks; do not submit the whole file as one
+multi-statement request or wrap it in one global transaction. For `psql`, use
+secure libpq connection configuration (environment, service, or passfile
+settings) and run:
+
+```bash
+psql -X -v ON_ERROR_STOP=1 -f apply.sql
+```
+
 Add `--shadow "$SHADOW_URL"` to use an explicit fresh database; otherwise,
 database scope creates and later drops a co-located shadow automatically.
 
@@ -201,7 +222,7 @@ has drifted (and prints the deltas) — handy in CI.
 | `snapshot` | Save a database's fact base to a file | `--source` `--out` `[--strict-coverage]` |
 | `drift` | Compare a live DB against a saved snapshot | `--env` `--snapshot` `[--strict-coverage]` |
 | `schema export` | Export a live DB to `.sql` files | `--source` `--out-dir` `[--layout]` `[--profile]` `[--strict-coverage]` |
-| `schema apply` | Load `.sql` files via a shadow DB and migrate a target | `--dir` `[--shadow]` `--target` `[--scope]` `[--isolated-shadow]` `[--renames]` `[--accept-rename]` `[--force]` `[--allow-data-loss]` `[--trusted-local-host]` `[--allow-remote-shadow]` `[--profile]` `[--restrict-to-applier]` `[--strict-coverage]` |
+| `schema apply` | Load `.sql` files via a shadow DB and migrate a target | `--dir` `[--shadow]` `--target` `[--scope]` `[--isolated-shadow]` `[--renames]` `[--accept-rename]` `[--force]` `[--allow-data-loss]` `[--trusted-local-host]` `[--allow-remote-shadow]` `[--profile]` `[--restrict-to-applier]` `[--strict-coverage]` `[--dry-run]` `[--verbose]` `[--out-plan]` |
 
 Common flags, explained:
 
diff --git a/packages/pg-delta/API-REVIEW.md b/packages/pg-delta/API-REVIEW.md
index f55a33a4..c3967bf3 100644
--- a/packages/pg-delta/API-REVIEW.md
+++ b/packages/pg-delta/API-REVIEW.md
@@ -79,7 +79,9 @@ the documented vocabulary.
 | Name | Kind | Contract | facts | deltas | actions | proof |
 |------|------|----------|:-----:|:------:|:-------:|:-----:|
 | `ActionStatus` | type | `"applied" \| "unapplied" \| "inDoubt"` — per-action outcome after execution. | — | — | ✓ | ✓ |
-| `ApplyOptions` | type | `{ fingerprintGate?, lockTimeoutMs?, statementTimeoutMs? }` — executor configuration. | — | — | ✓ | — |
+| `ApplyEvent` | type | Statement-level observer event emitted at segment/action boundaries and for executor control SQL. | — | — | ✓ | — |
+| `ApplyError` | type | `{ actionIndex, statementKind?, sql, message }` — attributed action/control failure; an absent legacy `statementKind` means `"action"`. | — | — | ✓ | ✓ |
+| `ApplyOptions` | type | Executor configuration including optional `onEvent(event: ApplyEvent)` statement-level observation. | — | — | ✓ | — |
 | `ApplyReport` | type | `{ status, appliedActions, actionStatuses, error? }` — execution outcome with per-action status and a structured error entry. | — | — | ✓ | ✓ |
 | `apply` | function | `(Plan, Pool, options?) → Promise<ApplyReport>` — sequential, segmented, lock-aware execution. | — | — | ✓ | ✓ |
 
diff --git a/packages/pg-delta/src/apply/apply-preamble.ts b/packages/pg-delta/src/apply/apply-preamble.ts
new file mode 100644
index 00000000..63275c62
--- /dev/null
+++ b/packages/pg-delta/src/apply/apply-preamble.ts
@@ -0,0 +1,26 @@
+import type { Plan } from "../plan/plan.ts";
+
+export interface ApplyTimeoutOptions {
+  lockTimeoutMs?: number;
+  statementTimeoutMs?: number;
+}
+
+/** Build the session settings sent before each apply segment. */
+export function buildApplyPreamble(
+  plan: Pick<Plan, "preamble">,
+  options: ApplyTimeoutOptions | undefined,
+  local: boolean,
+): string[] {
+  const scope = local ? "LOCAL " : "";
+  return [
+    ...(options?.lockTimeoutMs !== undefined
+      ? [`SET ${scope}lock_timeout = ${options.lockTimeoutMs}`]
+      : []),
+    ...(options?.statementTimeoutMs !== undefined
+      ? [`SET ${scope}statement_timeout = ${options.statementTimeoutMs}`]
+      : []),
+    ...plan.preamble.map(
+      (setting) => `SET ${scope}${setting.name} = ${setting.value}`,
+    ),
+  ];
+}
diff --git a/packages/pg-delta/src/apply/apply.test.ts b/packages/pg-delta/src/apply/apply.test.ts
index 02f839a5..c114b493 100644
--- a/packages/pg-delta/src/apply/apply.test.ts
+++ b/packages/pg-delta/src/apply/apply.test.ts
@@ -3,10 +3,11 @@
  * lists exercise maximal transactional runs, lone nonTransactional
  * actions, and commitBoundaryAfter boundaries.
  */
-import { describe, expect, test } from "bun:test";
+import { describe, expect, spyOn, test } from "bun:test";
 import type { Pool } from "pg";
-import { ENGINE_VERSION, type Plan } from "../plan/plan.ts";
-import { apply, segmentActions } from "./apply.ts";
+import type { Action, Plan } from "../plan/plan.ts";
+import { ENGINE_VERSION } from "../plan/plan.ts";
+import { apply, type ApplyEvent, segmentActions } from "./apply.ts";
 
 const txn = (newSegmentBefore = false) => ({
   transactionality: "transactional" as const,
@@ -67,6 +68,373 @@ describe("segmentActions", () => {
   });
 });
 
+function planWithAction(transactionality: Action["transactionality"]): Plan {
+  return {
+    formatVersion: 1,
+    engineVersion: ENGINE_VERSION,
+    source: { fingerprint: "source" },
+    target: { fingerprint: "target" },
+    preamble: [],
+    deltas: [],
+    filteredDeltas: [],
+    renameCandidates: [],
+    actions: [
+      {
+        sql: "SELECT 42",
+        verb: "create",
+        produces: [],
+        consumes: [],
+        destroys: [],
+        releases: [],
+        transactionality,
+        lockClass: "none",
+        newSegmentBefore: false,
+        dataLoss: "none",
+        rewriteRisk: false,
+      } as Action,
+    ],
+    safetyReport: {
+      destructiveActions: 0,
+      rewriteRiskActions: 0,
+      nonTransactionalActions: transactionality === "nonTransactional" ? 1 : 0,
+      lockClasses: {},
+    },
+  } as Plan;
+}
+
+async function expectObserverLatencyExcluded(
+  transactionality: Action["transactionality"],
+): Promise<void> {
+  let monotonicNow = 1_000;
+  let wallNow = 10_000;
+  const monotonicNowSpy = spyOn(performance, "now").mockImplementation(
+    () => monotonicNow,
+  );
+  const wallNowSpy = spyOn(Date, "now").mockImplementation(() => wallNow);
+  const events: ApplyEvent[] = [];
+  const client = {
+    query: (sql: string) => {
+      if (sql === "SELECT 42") {
+        monotonicNow += 7;
+        wallNow -= 50;
+      }
+      return Promise.resolve({ rows: [] });
+    },
+    release: () => {},
+  };
+  const pool = {
+    connect: () => Promise.resolve(client),
+  } as unknown as Pool;
+
+  try {
+    const report = await apply(planWithAction(transactionality), pool, {
+      fingerprintGate: false,
+      onEvent: (event) => {
+        events.push(event);
+        if (event.kind === "actionStart") {
+          monotonicNow += 100;
+          wallNow += 100;
+        }
+      },
+    });
+
+    expect(report.status).toBe("applied");
+    const actionEnd = events.find(
+      (event): event is Extract<ApplyEvent, { kind: "actionEnd" }> =>
+        event.kind === "actionEnd",
+    );
+    expect(actionEnd?.ms).toBeGreaterThanOrEqual(0);
+    expect(actionEnd?.ms).toBe(7);
+  } finally {
+    monotonicNowSpy.mockRestore();
+    wallNowSpy.mockRestore();
+  }
+}
+
+describe("apply action timing", () => {
+  test("transactional actionEnd is monotonic and excludes synchronous actionStart observer latency", async () => {
+    await expectObserverLatencyExcluded("transactional");
+  });
+
+  test("non-transactional actionEnd is monotonic and excludes synchronous actionStart observer latency", async () => {
+    await expectObserverLatencyExcluded("nonTransactional");
+  });
+});
+
+interface ScriptedApply {
+  pool: Pool;
+  queries: string[];
+  releases: Array<Error | boolean | undefined>;
+}
+
+function scriptedApplyClient(failingSql: ReadonlySet<string>): ScriptedApply {
+  const queries: string[] = [];
+  const releases: Array<Error | boolean | undefined> = [];
+  const client = {
+    query: (sql: string) => {
+      queries.push(sql);
+      return failingSql.has(sql)
+        ? Promise.reject(new Error(`scripted failure: ${sql}`))
+        : Promise.resolve({ rows: [] });
+    },
+    release: (error?: Error | boolean) => releases.push(error),
+  };
+  return {
+    pool: {
+      connect: () => Promise.resolve(client),
+    } as unknown as Pool,
+    queries,
+    releases,
+  };
+}
+
+function planWithPreamble(
+  transactionality: Action["transactionality"],
+  preamble: Plan["preamble"],
+): Plan {
+  return { ...planWithAction(transactionality), preamble };
+}
+
+function segmentOutcomes(events: ApplyEvent[]): string[] {
+  return events
+    .filter(
+      (event): event is Extract<ApplyEvent, { kind: "segmentEnd" }> =>
+        event.kind === "segmentEnd",
+    )
+    .map((event) => event.outcome);
+}
+
+describe("apply control-error attribution", () => {
+  test("BEGIN failure reports the exact control and leaves the action unapplied", async () => {
+    const scripted = scriptedApplyClient(new Set(["BEGIN"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(planWithAction("transactional"), scripted.pool, {
+      fingerprintGate: false,
+      onEvent: (event) => events.push(event),
+    });
+
+    expect(scripted.queries).toEqual(["BEGIN", "ROLLBACK"]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["unapplied"],
+      error: {
+        actionIndex: 0,
+        statementKind: "control",
+        sql: "BEGIN",
+        message: "scripted failure: BEGIN",
+      },
+    });
+    expect(events.some((event) => event.kind === "actionStart")).toBe(false);
+    expect(events.some((event) => event.kind === "actionEnd")).toBe(false);
+    expect(segmentOutcomes(events)).toEqual(["failed"]);
+    expect(scripted.releases).toEqual([undefined]);
+  });
+
+  test("later transactional preamble failure reports that SET LOCAL and rolls back", async () => {
+    const failingSet = "SET LOCAL check_function_bodies = off";
+    const scripted = scriptedApplyClient(new Set([failingSet]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(
+      planWithPreamble("transactional", [
+        { name: "check_function_bodies", value: "off" },
+      ]),
+      scripted.pool,
+      {
+        fingerprintGate: false,
+        lockTimeoutMs: 5000,
+        onEvent: (event) => events.push(event),
+      },
+    );
+
+    expect(scripted.queries).toEqual([
+      "BEGIN",
+      "SET LOCAL lock_timeout = 5000",
+      failingSet,
+      "ROLLBACK",
+    ]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["unapplied"],
+      error: {
+        actionIndex: 0,
+        statementKind: "control",
+        sql: failingSet,
+      },
+    });
+    expect(events.some((event) => event.kind === "actionStart")).toBe(false);
+    expect(events.some((event) => event.kind === "actionEnd")).toBe(false);
+    expect(segmentOutcomes(events)).toEqual(["failed"]);
+  });
+
+  test("transactional action failure remains an action error after successful rollback", async () => {
+    const scripted = scriptedApplyClient(new Set(["SELECT 42"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(planWithAction("transactional"), scripted.pool, {
+      fingerprintGate: false,
+      onEvent: (event) => events.push(event),
+    });
+
+    expect(scripted.queries).toEqual(["BEGIN", "SELECT 42", "ROLLBACK"]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["unapplied"],
+      error: {
+        actionIndex: 0,
+        statementKind: "action",
+        sql: "SELECT 42",
+      },
+    });
+    expect(segmentOutcomes(events)).toEqual(["rolledBack"]);
+    expect(scripted.releases).toEqual([undefined]);
+  });
+
+  test("COMMIT failure is an in-doubt control failure even when ROLLBACK succeeds", async () => {
+    const scripted = scriptedApplyClient(new Set(["COMMIT"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(planWithAction("transactional"), scripted.pool, {
+      fingerprintGate: false,
+      onEvent: (event) => events.push(event),
+    });
+
+    expect(scripted.queries).toEqual([
+      "BEGIN",
+      "SELECT 42",
+      "COMMIT",
+      "ROLLBACK",
+    ]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["inDoubt"],
+      error: {
+        actionIndex: 0,
+        statementKind: "control",
+        sql: "COMMIT",
+      },
+    });
+    expect(segmentOutcomes(events)).toEqual(["inDoubt"]);
+  });
+
+  test("non-transactional preamble failure is failed and never marks the action in doubt", async () => {
+    const failingSet = "SET check_function_bodies = invalid";
+    const scripted = scriptedApplyClient(new Set([failingSet]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(
+      planWithPreamble("nonTransactional", [
+        { name: "check_function_bodies", value: "invalid" },
+      ]),
+      scripted.pool,
+      {
+        fingerprintGate: false,
+        onEvent: (event) => events.push(event),
+      },
+    );
+
+    expect(scripted.queries).toEqual([failingSet, "RESET ALL"]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["unapplied"],
+      error: {
+        actionIndex: 0,
+        statementKind: "control",
+        sql: failingSet,
+      },
+    });
+    expect(events.some((event) => event.kind === "actionStart")).toBe(false);
+    expect(events.some((event) => event.kind === "actionEnd")).toBe(false);
+    expect(segmentOutcomes(events)).toEqual(["failed"]);
+  });
+
+  test("non-transactional action failure stays primary and in doubt when RESET ALL also fails", async () => {
+    const scripted = scriptedApplyClient(new Set(["SELECT 42", "RESET ALL"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(
+      planWithAction("nonTransactional"),
+      scripted.pool,
+      {
+        fingerprintGate: false,
+        onEvent: (event) => events.push(event),
+      },
+    );
+
+    expect(scripted.queries).toEqual(["SELECT 42", "RESET ALL"]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 0,
+      actionStatuses: ["inDoubt"],
+      error: {
+        actionIndex: 0,
+        statementKind: "action",
+        sql: "SELECT 42",
+      },
+    });
+    expect(segmentOutcomes(events)).toEqual(["inDoubt"]);
+    expect(scripted.releases).toEqual([true]);
+  });
+
+  test("RESET ALL failure after a successful action preserves the applied action", async () => {
+    const scripted = scriptedApplyClient(new Set(["RESET ALL"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(
+      planWithAction("nonTransactional"),
+      scripted.pool,
+      {
+        fingerprintGate: false,
+        onEvent: (event) => events.push(event),
+      },
+    );
+
+    expect(scripted.queries).toEqual(["SELECT 42", "RESET ALL"]);
+    expect(report).toMatchObject({
+      status: "failed",
+      appliedActions: 1,
+      actionStatuses: ["applied"],
+      error: {
+        actionIndex: 0,
+        statementKind: "control",
+        sql: "RESET ALL",
+      },
+    });
+    expect(segmentOutcomes(events)).toEqual(["failed"]);
+    expect(scripted.releases).toEqual([true]);
+  });
+
+  test("failed ROLLBACK destroys the client without replacing the primary preamble failure", async () => {
+    const failingSet = "SET LOCAL check_function_bodies = invalid";
+    const scripted = scriptedApplyClient(new Set([failingSet, "ROLLBACK"]));
+    const events: ApplyEvent[] = [];
+
+    const report = await apply(
+      planWithPreamble("transactional", [
+        { name: "check_function_bodies", value: "invalid" },
+      ]),
+      scripted.pool,
+      {
+        fingerprintGate: false,
+        onEvent: (event) => events.push(event),
+      },
+    );
+
+    expect(report.error).toMatchObject({
+      statementKind: "control",
+      sql: failingSet,
+    });
+    expect(segmentOutcomes(events)).toEqual(["failed"]);
+    expect(scripted.releases).toEqual([true]);
+  });
+});
+
 describe("apply plan integrity", () => {
   test("rejects contradictory destructive metadata before connecting or mutating", async () => {
     let connected = false;
diff --git a/packages/pg-delta/src/apply/apply.ts b/packages/pg-delta/src/apply/apply.ts
index 6c0da96d..877efa71 100644
--- a/packages/pg-delta/src/apply/apply.ts
+++ b/packages/pg-delta/src/apply/apply.ts
@@ -6,10 +6,11 @@
  * Segmentation changes transaction boundaries only, never order.
  *
  * Mid-plan failure semantics are explicit: every action is reported
- * applied / unapplied / inDoubt. A failure inside a transaction segment
- * rolls that segment back (its actions report unapplied); earlier
- * segments are committed (applied); a failure AT commit reports the
- * segment inDoubt.
+ * applied / unapplied / inDoubt, and the error identifies whether an action
+ * or an executor control statement failed. A failure inside a transaction
+ * segment rolls that segment back (its actions report unapplied); earlier
+ * segments are committed (applied); a failure AT commit reports the segment
+ * inDoubt.
  */
 import type { Pool } from "pg";
 import type { FactBase } from "../core/fact.ts";
@@ -17,18 +18,60 @@ import { extract } from "../extract/extract.ts";
 import { ENGINE_VERSION, type Plan } from "../plan/plan.ts";
 import { assertDestructionMetadataIntegrity } from "../plan/safety.ts";
 import { reconstructManagedView } from "../policy/reconstruct.ts";
+import { buildApplyPreamble } from "./apply-preamble.ts";
 
 export type ActionStatus = "applied" | "unapplied" | "inDoubt";
 
+export interface ApplyError {
+  /** First affected action for controls; exact failing action otherwise. */
+  actionIndex: number;
+  /** Absent on legacy consumer-authored errors; interpreted as `"action"`. */
+  statementKind?: "action" | "control";
+  sql: string;
+  message: string;
+}
+
+type ApplyStatementKind = NonNullable<ApplyError["statementKind"]>;
+type ProducedApplyError = ApplyError & { statementKind: ApplyStatementKind };
+
 export interface ApplyReport {
   status: "applied" | "failed";
   /** count of actions in committed segments */
   appliedActions: number;
   /** one entry per plan action, in plan order */
   actionStatuses: ActionStatus[];
-  error?: { actionIndex: number; sql: string; message: string };
+  error?: ApplyError;
 }
 
+/** 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" | "failed";
+    }
+  | { 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 +93,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 {
@@ -99,16 +146,46 @@ export function segmentActions(
 
 function errorEntry(
   actionIndex: number,
+  statementKind: ApplyStatementKind,
   sql: string,
   error: unknown,
-): NonNullable<ApplyReport["error"]> {
+): ProducedApplyError {
   return {
     actionIndex,
+    statementKind,
     sql,
     message: error instanceof Error ? error.message : String(error),
   };
 }
 
+/** 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 {
+    // 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
+  }
+}
+
 export async function apply(
   thePlan: Plan,
   target: Pool,
@@ -189,99 +266,238 @@ export async function apply(
   let appliedActions = 0;
 
   const client = await target.connect();
+  let destroyClient = false;
   try {
-    const preamble = (local: boolean): string[] => [
-      ...(options?.lockTimeoutMs !== undefined
-        ? [
-            `SET ${local ? "LOCAL " : ""}lock_timeout = ${options.lockTimeoutMs}`,
-          ]
-        : []),
-      ...(options?.statementTimeoutMs !== undefined
-        ? [
-            `SET ${local ? "LOCAL " : ""}statement_timeout = ${options.statementTimeoutMs}`,
-          ]
-        : []),
-      ...thePlan.preamble.map(
-        (s) => `SET ${local ? "LOCAL " : ""}${s.name} = ${s.value}`,
-      ),
-    ];
+    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,
+      });
 
-    for (const segment of segments) {
       if (!segment.transactional) {
         // a lone non-transactional action; session-level settings, reset after
         const index = segment.start;
         const action = thePlan.actions[index]!;
+        // 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: ProducedApplyError | undefined;
+        let currentKind: ApplyStatementKind = "control";
+        let currentSql = "";
         try {
-          for (const sql of preamble(false)) await client.query(sql);
-          await client.query(action.sql);
+          // 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 buildApplyPreamble(thePlan, options, false)) {
+            currentKind = "control";
+            currentSql = sql;
+            emit(onEvent, { kind: "control", sql });
+            await client.query(sql);
+          }
+          currentKind = "action";
+          currentSql = action.sql;
+          emit(onEvent, {
+            kind: "actionStart",
+            actionIndex: index,
+            sql: action.sql,
+          });
+          const actionStartedAt = performance.now();
+          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).
+            const actionElapsedMs = performance.now() - actionStartedAt;
+            emit(onEvent, {
+              kind: "actionEnd",
+              actionIndex: index,
+              ok: true,
+              ms: actionElapsedMs,
+            });
+          } catch (error) {
+            const actionElapsedMs = performance.now() - actionStartedAt;
+            emit(onEvent, {
+              kind: "actionEnd",
+              actionIndex: index,
+              ok: false,
+              ms: actionElapsedMs,
+            });
+            throw error;
+          }
+        } catch (error) {
+          if (currentKind === "action") {
+            // 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).
+            statuses[index] = "inDoubt";
+          }
+          failure = errorEntry(index, currentKind, currentSql, error);
+        }
+
+        // ALWAYS restore session state before the client returns to the pool,
+        // on success or failure. A cleanup failure never replaces the primary
+        // failure, and its connection is destroyed instead of being reused.
+        emit(onEvent, { kind: "control", sql: "RESET ALL" });
+        let resetFailed = false;
+        let resetError: unknown;
+        try {
+          await client.query("RESET ALL");
         } 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";
+          resetFailed = true;
+          resetError = error;
+          destroyClient = true;
+        }
+
+        if (failure !== undefined) {
+          emit(onEvent, {
+            kind: "segmentEnd",
+            segmentIndex: segIdx,
+            outcome: failure.statementKind === "action" ? "inDoubt" : "failed",
+          });
           return {
             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.
-          await client.query("RESET ALL").catch(() => {});
         }
+        // The action completed in autocommit before RESET ALL ran, so a RESET
+        // failure cannot relabel that durable action inDoubt or unapplied.
         statuses[index] = "applied";
         appliedActions += 1;
+        if (resetFailed) {
+          emit(onEvent, {
+            kind: "segmentEnd",
+            segmentIndex: segIdx,
+            outcome: "failed",
+          });
+          return {
+            status: "failed",
+            appliedActions,
+            actionStatuses: statuses,
+            error: errorEntry(index, "control", "RESET ALL", resetError),
+          };
+        }
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "committed",
+        });
         continue;
       }
 
+      let setupSql = "BEGIN";
       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 buildApplyPreamble(thePlan, options, true)) {
+          setupSql = sql;
+          emit(onEvent, { kind: "control", sql });
+          await client.query(sql);
+        }
       } catch (error) {
-        await client.query("ROLLBACK").catch(() => {});
+        emit(onEvent, { kind: "control", sql: "ROLLBACK" });
+        try {
+          await client.query("ROLLBACK");
+        } catch {
+          destroyClient = true;
+        }
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "failed",
+        });
         return {
           status: "failed",
           appliedActions,
           actionStatuses: statuses,
-          error: errorEntry(segment.start, "BEGIN", error),
+          error: errorEntry(segment.start, "control", setupSql, error),
         };
       }
       for (let i = segment.start; i < segment.end; i++) {
         const action = thePlan.actions[i]!;
+        emit(onEvent, { kind: "actionStart", actionIndex: i, sql: action.sql });
+        const actionStartedAt = performance.now();
         try {
           await client.query(action.sql);
         } catch (error) {
-          await client.query("ROLLBACK").catch(() => {});
+          const actionElapsedMs = performance.now() - actionStartedAt;
+          emit(onEvent, {
+            kind: "actionEnd",
+            actionIndex: i,
+            ok: false,
+            ms: actionElapsedMs,
+          });
+          emit(onEvent, { kind: "control", sql: "ROLLBACK" });
+          let rollbackSucceeded = true;
+          try {
+            await client.query("ROLLBACK");
+          } catch {
+            rollbackSucceeded = false;
+            destroyClient = true;
+          }
+          emit(onEvent, {
+            kind: "segmentEnd",
+            segmentIndex: segIdx,
+            outcome: rollbackSucceeded ? "rolledBack" : "failed",
+          });
           return {
             status: "failed",
             appliedActions,
             actionStatuses: statuses,
-            error: errorEntry(i, action.sql, error),
+            error: errorEntry(i, "action", action.sql, error),
           };
         }
+        const actionElapsedMs = performance.now() - actionStartedAt;
+        emit(onEvent, {
+          kind: "actionEnd",
+          actionIndex: i,
+          ok: true,
+          ms: actionElapsedMs,
+        });
       }
       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";
-        await client.query("ROLLBACK").catch(() => {});
+        emit(onEvent, { kind: "control", sql: "ROLLBACK" });
+        try {
+          await client.query("ROLLBACK");
+        } catch {
+          destroyClient = true;
+        }
+        emit(onEvent, {
+          kind: "segmentEnd",
+          segmentIndex: segIdx,
+          outcome: "inDoubt",
+        });
         return {
           status: "failed",
           appliedActions,
           actionStatuses: statuses,
-          error: errorEntry(segment.start, "COMMIT", error),
+          error: errorEntry(segment.start, "control", "COMMIT", error),
         };
       }
       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();
+    if (destroyClient) client.release(true);
+    else client.release();
   }
   return {
     status: "applied",
diff --git a/packages/pg-delta/src/cli/commands/apply.ts b/packages/pg-delta/src/cli/commands/apply.ts
index 60104f49..9a7e5b06 100644
--- a/packages/pg-delta/src/cli/commands/apply.ts
+++ b/packages/pg-delta/src/cli/commands/apply.ts
@@ -3,7 +3,7 @@
  *
  * Parse the plan artifact and apply it to the target database.
  * --force disables the fingerprint gate.
- * On failure, print the per-action failure report.
+ * On failure, print the attributed action or control failure report.
  */
 import { readFileSync } from "node:fs";
 import { parsePlan } from "../../plan/artifact.ts";
@@ -109,9 +109,11 @@ export async function cmdApply(args: string[]): Promise<void> {
     } else {
       process.stderr.write(`Apply failed!\n`);
       if (report.error) {
-        process.stderr.write(
-          `  action[${report.error.actionIndex}]: ${report.error.message}\n`,
-        );
+        const subject =
+          (report.error.statementKind ?? "action") === "action"
+            ? `action[${report.error.actionIndex}]`
+            : "control";
+        process.stderr.write(`  ${subject}: ${report.error.message}\n`);
         process.stderr.write(`  sql: ${report.error.sql}\n`);
       }
       const applied = report.actionStatuses.filter(
diff --git a/packages/pg-delta/src/cli/commands/prove.test.ts b/packages/pg-delta/src/cli/commands/prove.test.ts
index dfe1a64c..3cd9b58a 100644
--- a/packages/pg-delta/src/cli/commands/prove.test.ts
+++ b/packages/pg-delta/src/cli/commands/prove.test.ts
@@ -32,6 +32,7 @@ import type {
   ProjectionAuditEntry,
   ProjectionAuditStage,
 } from "../../plan/plan.ts";
+import type { ApplyError } from "../../apply/apply.ts";
 
 const baseVerdict = (): ProofVerdict => ({
   ok: false,
@@ -680,6 +681,22 @@ describe("formatProjectionAudit", () => {
 });
 
 describe("formatProofFailure (review P2)", () => {
+  test("treats a legacy ApplyError without statementKind as an action", () => {
+    const legacyApplyError: ApplyError = {
+      actionIndex: 2,
+      sql: "ALTER TABLE app.t ADD COLUMN value text",
+      message: "legacy failure",
+    };
+    const verdict: ProofVerdict = {
+      ...baseVerdict(),
+      applyError: legacyApplyError,
+    };
+
+    expect(formatProofFailure(verdict)).toContain(
+      "apply error at action[2]: legacy failure",
+    );
+  });
+
   test("renders an intrinsically destructive subobject metadata failure", () => {
     const verdict: ProofVerdict = {
       ...baseVerdict(),
diff --git a/packages/pg-delta/src/cli/commands/prove.ts b/packages/pg-delta/src/cli/commands/prove.ts
index 34a6e5a0..aa790d6d 100644
--- a/packages/pg-delta/src/cli/commands/prove.ts
+++ b/packages/pg-delta/src/cli/commands/prove.ts
@@ -87,9 +87,11 @@ export function formatProofFailure(verdict: ProofVerdict): string {
     );
   }
   if (verdict.applyError) {
-    lines.push(
-      `  apply error at action[${verdict.applyError.actionIndex}]: ${verdict.applyError.message}`,
-    );
+    const subject =
+      (verdict.applyError.statementKind ?? "action") === "action"
+        ? `action[${verdict.applyError.actionIndex}]`
+        : "control";
+    lines.push(`  apply error at ${subject}: ${verdict.applyError.message}`);
   }
   if (verdict.driftDeltas.length > 0) {
     lines.push(`  drift deltas (${verdict.driftDeltas.length}):`);
diff --git a/packages/pg-delta/src/cli/commands/schema.ts b/packages/pg-delta/src/cli/commands/schema.ts
index 7f924123..2549607d 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,41 @@
  *   --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 a portable executable script containing
+ *   the same successful-path statements and segment boundaries, without
+ *   applying them. It must be dispatched statement by statement on one session,
+ *   stopping at the first error, with autocommit outside its explicit
+ *   transaction blocks; never submit it as one multi-statement request or wrap
+ *   the whole script in one transaction.
+ *
+ *   --dry-run
+ *     Plan as usual, then print the executable portable SQL script to STDOUT
+ *     instead of calling apply() — no fingerprint gate runs, nothing is
+ *     applied. Execute it statement by statement on one session and stop on
+ *     the first error; preserve autocommit outside its explicit transaction
+ *     blocks. 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
+ *     (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 type { Pool } from "pg";
 import {
@@ -59,7 +95,10 @@ import {
   readExportManifest,
   writeExportManifest,
 } from "../../frontends/export-manifest.ts";
-import { type SqlFile } from "../../frontends/load-sql-files.ts";
+import {
+  ShadowLoadError,
+  type SqlFile,
+} from "../../frontends/load-sql-files.ts";
 import { analyzeForShadow } from "../../frontends/sql-order.ts";
 import { buildSchemaExport } from "../../frontends/schema-export.ts";
 import {
@@ -71,8 +110,11 @@ import {
   formatLintReport,
   rewriteReorderedShadowError,
 } from "../reorder-display.ts";
+import { serializePlan } from "../../plan/artifact.ts";
 import type { ManagementScope } from "../../policy/view.ts";
-import { apply } from "../../apply/apply.ts";
+import { apply, type ApplyEvent } from "../../apply/apply.ts";
+import { renderApplyScript } from "../../frontends/render-apply-script.ts";
+import { isDestructiveAction } from "../render.ts";
 import { encodeId, parseId, type StableId } from "../../core/stable-id.ts";
 import { exitIfBlocking, printDiagnostics } from "../diagnostics.ts";
 import { makePool } from "../pool.ts";
@@ -454,6 +496,41 @@ export function prepareApplyFiles(
   return prepared;
 }
 
+/** Warn whenever a schema-apply output surface can expose cleartext secrets.
+ *  The caller passes the EFFECTIVE redaction mode, already reconciled with the
+ *  export manifest, so flag- and manifest-driven unsafe modes cannot drift. */
+function warnIfUnredactedOutput(
+  redactSecrets: boolean,
+  output:
+    | "plan artifact"
+    | "dry-run script"
+    | "verbose output"
+    | "planning failure diagnostic"
+    | "failure diagnostic",
+): void {
+  if (!redactSecrets) {
+    process.stderr.write(
+      `  WARNING: secrets are unredacted (--unsafe-show-secrets or the export manifest) — the ${output} may contain unredacted credentials.\n`,
+    );
+  }
+}
+
+/** True only when a shadow-load failure can surface an authored statement.
+ *  Early policy/connection/empty-shadow failures do not carry statement text
+ *  and therefore do not need the unredacted-output warning. */
+function hasShadowStatementDiagnostic(
+  error: unknown,
+): error is ShadowLoadError {
+  return (
+    error instanceof ShadowLoadError &&
+    error.details.some(
+      (detail) =>
+        detail.code === "stuck_statement" ||
+        detail.code === "max_rounds_exceeded",
+    )
+  );
+}
+
 export async function cmdSchemaApply(args: string[]): Promise<void> {
   let parsed;
   try {
@@ -474,6 +551,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" },
       "allow-data-loss": { type: "boolean" },
       "trusted-local-host": { type: "multi" },
       "allow-remote-shadow": { type: "boolean" },
@@ -485,6 +565,7 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
           `[--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] [--allow-data-loss] ` +
           `[--trusted-local-host <hostname>]... [--allow-remote-shadow]\n` +
+          `  [--dry-run] (print the portable apply script to stdout; apply nothing; see pgdelta --help for execution requirements) [--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.`,
       );
     }
@@ -497,6 +578,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.
@@ -775,6 +859,11 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
         }
         return enriched;
       },
+    }).catch((error: unknown) => {
+      if (hasShadowStatementDiagnostic(error)) {
+        warnIfUnredactedOutput(redactSecrets, "planning failure diagnostic");
+      }
+      throw error;
     });
 
     printDiagnostics(planned.loadDiagnostics, { label: "shadow" });
@@ -808,11 +897,45 @@ 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) {
+      warnIfUnredactedOutput(redactSecrets, "plan artifact");
+      writeFileSync(outPlanPath, serializePlan(thePlan), "utf8");
+      process.stderr.write(`Plan artifact written to ${outPlanPath}\n`);
+    }
+
     if (thePlan.actions.length === 0) {
       process.stderr.write("Target is already up to date.\n");
       return;
     }
 
+    if (dryRun) {
+      const script = renderApplyScript(thePlan, {
+        ...(planned.applyOptions.lockTimeoutMs !== undefined
+          ? { lockTimeoutMs: planned.applyOptions.lockTimeoutMs }
+          : {}),
+        ...(planned.applyOptions.statementTimeoutMs !== undefined
+          ? { statementTimeoutMs: planned.applyOptions.statementTimeoutMs }
+          : {}),
+      });
+      // The script is routinely redirected to a file, making it as persistent
+      // as an archived plan artifact. Warn before stdout can expose it.
+      warnIfUnredactedOutput(redactSecrets, "dry-run script");
+      process.stdout.write(script);
+      process.stderr.write(
+        `Dry run: ${thePlan.actions.length} action(s) planned; nothing applied.\n`,
+      );
+      const destructiveCount =
+        thePlan.actions.filter(isDestructiveAction).length;
+      if (destructiveCount > 0) {
+        process.stderr.write(
+          `WARNING: plan contains ${destructiveCount} destructive action(s).\n`,
+        );
+      }
+      return;
+    }
+
     const destructive = assertDataLossAllowed(
       thePlan.actions,
       flags["allow-data-loss"],
@@ -824,15 +947,67 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
       );
     }
 
+    // A real verbose apply writes every action SQL to stderr. Warn after the
+    // dry-run early return (where --verbose has no effect) and before the first
+    // action can expose a credential.
+    if (verbose) {
+      warnIfUnredactedOutput(redactSecrets, "verbose output");
+    }
+
     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, {
       ...planned.applyOptions,
       reextract: (p) =>
         planned.extract(p, { redactSecrets: planned.redactSecrets }),
       fingerprintGate: !force,
+      ...(onEvent !== undefined ? { onEvent } : {}),
     });
 
     if (report.status === "applied") {
@@ -842,9 +1017,17 @@ export async function cmdSchemaApply(args: string[]): Promise<void> {
     } else {
       process.stderr.write("Apply failed!\n");
       if (report.error) {
-        process.stderr.write(
-          `  action[${report.error.actionIndex}]: ${report.error.message}\n`,
-        );
+        // Non-verbose applies have not exposed action SQL yet. PostgreSQL's
+        // error message can echo that SQL, so warn before the entire failure
+        // diagnostic. Verbose mode already warned before its actionStart trace.
+        if (!verbose) {
+          warnIfUnredactedOutput(redactSecrets, "failure diagnostic");
+        }
+        const subject =
+          (report.error.statementKind ?? "action") === "action"
+            ? `action[${report.error.actionIndex}]`
+            : "control";
+        process.stderr.write(`  ${subject}: ${report.error.message}\n`);
         process.stderr.write(`  sql: ${report.error.sql}\n`);
       }
       // The finally below releases resources (drops any co-located shadow);
diff --git a/packages/pg-delta/src/cli/main.test.ts b/packages/pg-delta/src/cli/main.test.ts
new file mode 100644
index 00000000..84b1b197
--- /dev/null
+++ b/packages/pg-delta/src/cli/main.test.ts
@@ -0,0 +1,44 @@
+import { describe, expect, test } from "bun:test";
+
+const CLI = new URL("./main.ts", import.meta.url).pathname;
+
+async function runHelp(...args: string[]): Promise<{
+  stdout: string;
+  stderr: string;
+  exitCode: number;
+}> {
+  const proc = Bun.spawn(["bun", CLI, ...args], {
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([
+    new Response(proc.stdout).text(),
+    new Response(proc.stderr).text(),
+    proc.exited,
+  ]);
+  return { stdout, stderr, exitCode };
+}
+
+describe("schema help", () => {
+  test("lists the schema apply debugging flags", async () => {
+    const { stdout, stderr, exitCode } = await runHelp("schema", "--help");
+
+    expect({ exitCode, stderr }).toMatchObject({ exitCode: 0, stderr: "" });
+    expect(stdout).toContain("--dry-run");
+    expect(stdout).toContain("--verbose");
+    expect(stdout).toContain("--out-plan");
+  });
+});
+
+describe("top-level help", () => {
+  test("qualifies how to execute a dry-run script", async () => {
+    const { stdout, stderr, exitCode } = await runHelp("--help");
+
+    expect({ exitCode, stderr }).toMatchObject({ exitCode: 0, stderr: "" });
+    expect(stdout).toContain("statement by statement on one session");
+    expect(stdout).toMatch(/Do\s+not submit it as one multi-statement request/);
+    expect(stdout).toContain("psql -X -v ON_ERROR_STOP=1 -f apply.sql");
+    expect(stdout).toContain("secure libpq configuration");
+    expect(stdout).not.toContain('"$DATABASE_URL"');
+  });
+});
diff --git a/packages/pg-delta/src/cli/main.ts b/packages/pg-delta/src/cli/main.ts
index bed52ef8..9769d228 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.
@@ -81,6 +82,7 @@ Commands:
                  [--scope database|cluster] [--isolated-shadow]
                  [--trusted-local-host <hostname>]... [--allow-remote-shadow]
                  [--accept-rename <from>=<to>] ... [--no-reorder]
+                 [--dry-run] [--verbose] [--out-plan <plan.json>]
   schema lint    --dir <dir>
 
 Notes:
@@ -135,6 +137,35 @@ 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 the same
+    successful-path statements and transaction framing, split on the same
+    segment boundaries, without applying them.
+  --dry-run (schema apply): plan as usual, then print a portable apply script
+    to stdout instead of applying — no fingerprint gate runs, nothing is
+    applied. Execute it statement by statement on one session, stopping on the
+    first error, with autocommit outside its explicit BEGIN/COMMIT blocks. Do
+    not submit it as one multi-statement request or wrap it in one global
+    transaction. Safe psql execution behavior:
+      psql -X -v ON_ERROR_STOP=1 -f apply.sql
+    Supply the connection through secure libpq configuration, environment,
+    service, or passfile settings rather than a password-bearing URI in argv.
+    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,
+    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.
@@ -165,6 +196,7 @@ Subcommands:
                  [--scope database|cluster] [--isolated-shadow]
                  [--trusted-local-host <hostname>]... [--allow-remote-shadow]
                  [--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.
diff --git a/packages/pg-delta/src/cli/render.ts b/packages/pg-delta/src/cli/render.ts
index 143bcdd9..0c3a95d9 100644
--- a/packages/pg-delta/src/cli/render.ts
+++ b/packages/pg-delta/src/cli/render.ts
@@ -3,4 +3,7 @@
  * CLI imports (`./render.ts`) keep working. New callers should prefer
  * `renderPlanFiles` from `@supabase/pg-delta/frontends`.
  */
-export { renderPlanFiles as renderPlan } from "../frontends/render-plan-files.ts";
+export {
+  renderPlanFiles as renderPlan,
+  isDestructiveAction,
+} from "../frontends/render-plan-files.ts";
diff --git a/packages/pg-delta/src/frontends/index.ts b/packages/pg-delta/src/frontends/index.ts
index 3af63af8..db84669e 100644
--- a/packages/pg-delta/src/frontends/index.ts
+++ b/packages/pg-delta/src/frontends/index.ts
@@ -42,6 +42,7 @@ export {
 
 export {
   renderPlanFiles,
+  isDestructiveAction,
   type RenderPlanFilesOptions,
   type RenderPlanFilesResult,
   type RenderedPlanFile,
diff --git a/packages/pg-delta/src/frontends/render-apply-script.test.ts b/packages/pg-delta/src/frontends/render-apply-script.test.ts
new file mode 100644
index 00000000..1d968f69
--- /dev/null
+++ b/packages/pg-delta/src/frontends/render-apply-script.test.ts
@@ -0,0 +1,121 @@
+import { describe, expect, test } from "bun:test";
+import type { Action, Plan } from "../plan/plan.ts";
+import { renderApplyScript } from "./render-apply-script.ts";
+
+function action(
+  sql: string,
+  transactionality: Action["transactionality"] = "transactional",
+): Action {
+  return {
+    sql,
+    verb: "create",
+    produces: [],
+    consumes: [],
+    destroys: [],
+    releases: [],
+    transactionality,
+    lockClass: "none",
+    newSegmentBefore: false,
+    dataLoss: "none",
+    rewriteRisk: false,
+  } as Action;
+}
+
+function planWithMixedSegments(): Plan {
+  return {
+    formatVersion: 1,
+    engineVersion: "test",
+    source: { fingerprint: "source" },
+    target: { fingerprint: "target" },
+    preamble: [
+      { name: "search_path", value: "pg_catalog" },
+      { name: "check_function_bodies", value: "off" },
+    ],
+    deltas: [],
+    filteredDeltas: [],
+    renameCandidates: [],
+    actions: [
+      action("CREATE TABLE public.widgets (id integer)"),
+      action(
+        "CREATE INDEX CONCURRENTLY widgets_id_idx ON public.widgets (id)",
+        "nonTransactional",
+      ),
+      action("ALTER TABLE public.widgets ADD COLUMN label text"),
+    ],
+    safetyReport: {
+      destructiveActions: 0,
+      rewriteRiskActions: 0,
+      nonTransactionalActions: 1,
+      lockClasses: {},
+    },
+  } as Plan;
+}
+
+describe("renderApplyScript", () => {
+  test("renders the exact apply transaction framing and full preamble", () => {
+    expect(
+      renderApplyScript(planWithMixedSegments(), {
+        lockTimeoutMs: 1_000,
+        statementTimeoutMs: 2_000,
+      }),
+    ).toMatchInlineSnapshot(`
+      "-- pg-delta schema apply --dry-run
+      -- Execute statements one at a time, in order, on one database session.
+      -- Stop on the first error and preserve autocommit outside BEGIN/COMMIT blocks.
+      -- Do not submit this as one multi-statement request or wrap it in one transaction.
+
+      BEGIN;
+      SET LOCAL lock_timeout = 1000;
+      SET LOCAL statement_timeout = 2000;
+      SET LOCAL search_path = pg_catalog;
+      SET LOCAL check_function_bodies = off;
+      CREATE TABLE public.widgets (id integer);
+      COMMIT;
+
+      SET lock_timeout = 1000;
+      SET statement_timeout = 2000;
+      SET search_path = pg_catalog;
+      SET check_function_bodies = off;
+      CREATE INDEX CONCURRENTLY widgets_id_idx ON public.widgets (id);
+      RESET ALL;
+
+      BEGIN;
+      SET LOCAL lock_timeout = 1000;
+      SET LOCAL statement_timeout = 2000;
+      SET LOCAL search_path = pg_catalog;
+      SET LOCAL check_function_bodies = off;
+      ALTER TABLE public.widgets ADD COLUMN label text;
+      COMMIT;
+      "
+    `);
+  });
+
+  test("keeps commitBoundaryAfter actions behind a committed boundary", () => {
+    const plan = planWithMixedSegments();
+    plan.actions = [
+      action("ALTER TYPE public.mood ADD VALUE 'ok'", "commitBoundaryAfter"),
+      action("ALTER TABLE public.widgets ADD COLUMN mood public.mood"),
+    ];
+
+    expect(renderApplyScript(plan)).toContain(
+      "ALTER TYPE public.mood ADD VALUE 'ok';\nCOMMIT;\n\nBEGIN;\nSET LOCAL search_path",
+    );
+  });
+
+  test("resets a non-transactional segment even without a preamble", () => {
+    const plan = planWithMixedSegments();
+    plan.preamble = [];
+    plan.actions = [action("VACUUM public.widgets", "nonTransactional")];
+
+    expect(renderApplyScript(plan)).toEndWith(
+      "\n\nVACUUM public.widgets;\nRESET ALL;\n",
+    );
+  });
+
+  test("renders an empty plan as an empty script", () => {
+    const plan = planWithMixedSegments();
+    plan.actions = [];
+
+    expect(renderApplyScript(plan)).toBe("");
+  });
+});
diff --git a/packages/pg-delta/src/frontends/render-apply-script.ts b/packages/pg-delta/src/frontends/render-apply-script.ts
new file mode 100644
index 00000000..476dc815
--- /dev/null
+++ b/packages/pg-delta/src/frontends/render-apply-script.ts
@@ -0,0 +1,42 @@
+import type { Plan } from "../plan/plan.ts";
+import {
+  buildApplyPreamble,
+  type ApplyTimeoutOptions,
+} from "../apply/apply-preamble.ts";
+import { segmentActions } from "../apply/apply.ts";
+
+function terminate(sql: string): string {
+  const trimmed = sql.trimEnd();
+  return trimmed.endsWith(";") ? trimmed : `${trimmed};`;
+}
+
+const EXECUTION_CONTRACT = `-- pg-delta schema apply --dry-run
+-- Execute statements one at a time, in order, on one database session.
+-- Stop on the first error and preserve autocommit outside BEGIN/COMMIT blocks.
+-- Do not submit this as one multi-statement request or wrap it in one transaction.`;
+
+/** Render the portable SQL statement sequence that apply() sends on success. */
+export function renderApplyScript(
+  plan: Plan,
+  timeoutOptions?: ApplyTimeoutOptions,
+): string {
+  if (plan.actions.length === 0) return "";
+
+  const segments = segmentActions(plan.actions);
+  const rendered = segments.map((segment) => {
+    const statements: string[] = [];
+    if (segment.transactional) statements.push("BEGIN;");
+    statements.push(
+      ...buildApplyPreamble(plan, timeoutOptions, segment.transactional).map(
+        terminate,
+      ),
+    );
+    for (let index = segment.start; index < segment.end; index++) {
+      statements.push(terminate(plan.actions[index]!.sql));
+    }
+    statements.push(segment.transactional ? "COMMIT;" : "RESET ALL;");
+    return statements.join("\n");
+  });
+
+  return `${EXECUTION_CONTRACT}\n\n${rendered.join("\n\n")}\n`;
+}
diff --git a/packages/pg-delta/src/frontends/render-plan-files.ts b/packages/pg-delta/src/frontends/render-plan-files.ts
index 6dd2d0be..6e029002 100644
--- a/packages/pg-delta/src/frontends/render-plan-files.ts
+++ b/packages/pg-delta/src/frontends/render-plan-files.ts
@@ -35,6 +35,16 @@ export interface RenderPlanFilesResult {
   files: RenderedPlanFile[];
 }
 
+/** 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 {
@@ -56,9 +66,7 @@ export function renderPlanFiles(
   }
 
   if (!opts.allowDrops) {
-    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/src/frontends/schema-plan.ts b/packages/pg-delta/src/frontends/schema-plan.ts
index 84204167..043fcdc1 100644
--- a/packages/pg-delta/src/frontends/schema-plan.ts
+++ b/packages/pg-delta/src/frontends/schema-plan.ts
@@ -585,6 +585,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);
diff --git a/packages/pg-delta/src/index.ts b/packages/pg-delta/src/index.ts
index f111c823..7437a176 100644
--- a/packages/pg-delta/src/index.ts
+++ b/packages/pg-delta/src/index.ts
@@ -57,8 +57,10 @@ export { type LockClass } from "./plan/locks.ts";
 // ── apply ────────────────────────────────────────────────────────────────────
 export {
   apply,
+  type ApplyError,
   type ApplyReport,
   type ApplyOptions,
+  type ApplyEvent,
   type ActionStatus,
 } from "./apply/apply.ts";
 
@@ -105,6 +107,7 @@ export {
 } from "./frontends/schema-plan.ts";
 export {
   renderPlanFiles,
+  isDestructiveAction,
   type RenderPlanFilesOptions,
   type RenderPlanFilesResult,
   type RenderedPlanFile,
diff --git a/packages/pg-delta/src/proof/prove.ts b/packages/pg-delta/src/proof/prove.ts
index efdfeb97..94cca4e8 100644
--- a/packages/pg-delta/src/proof/prove.ts
+++ b/packages/pg-delta/src/proof/prove.ts
@@ -10,7 +10,7 @@
  *      risk is observed on the clone, not certified by the rule)
  */
 import type { Pool } from "pg";
-import { apply } from "../apply/apply.ts";
+import { apply, type ApplyError } from "../apply/apply.ts";
 import { diff, type Delta } from "../core/diff.ts";
 import type { FactBase } from "../core/fact.ts";
 import type { StableId } from "../core/stable-id.ts";
@@ -93,7 +93,7 @@ export interface ProofVerdict {
   /** Why opt-in strict projection-audit enforcement failed. Absent when strict
    * audit was not requested or its requirements passed. */
   strictAuditFailure?: "unavailable" | "suspicious";
-  applyError?: { actionIndex: number; sql: string; message: string };
+  applyError?: ApplyError;
   /** Clone state did not match the state the plan was produced from. The proof
    *  refuses before auto-seeding or applying any action. */
   sourceStateViolation?: {
diff --git a/packages/pg-delta/tests/apply-nontransactional.test.ts b/packages/pg-delta/tests/apply-nontransactional.test.ts
index 83f635c1..88a346ba 100644
--- a/packages/pg-delta/tests/apply-nontransactional.test.ts
+++ b/packages/pg-delta/tests/apply-nontransactional.test.ts
@@ -66,6 +66,11 @@ describe("non-transactional apply: reset + inDoubt (P1)", () => {
       expect(report.status).toBe("failed");
       // inDoubt, NOT unapplied — the action may have durable side effects
       expect(report.actionStatuses[0]).toBe("inDoubt");
+      expect(report.error).toMatchObject({
+        actionIndex: 0,
+        statementKind: "action",
+        sql: "SELECT 1 / 0",
+      });
       // RESET ALL ran in finally despite the failure → back to the default
       const { rows } = await probe.query("SHOW lock_timeout");
       expect((rows[0] as { lock_timeout: string }).lock_timeout).toBe("0");
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 00000000..2d4bb21b
--- /dev/null
+++ b/packages/pg-delta/tests/apply-observer.test.ts
@@ -0,0 +1,479 @@
+/**
+ * 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 } 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";
+
+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: 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()]);
+    }
+  }, 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);
+
+  // 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");
+
+      const failingSet = "SET lock_timeout = -1";
+      expect(report).toMatchObject({
+        appliedActions: 0,
+        actionStatuses: ["unapplied"],
+        error: {
+          actionIndex: 0,
+          statementKind: "control",
+          sql: failingSet,
+        },
+      });
+
+      // 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("failed");
+    } finally {
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 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");
+    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 c67b26a0..c855d70b 100644
--- a/packages/pg-delta/tests/cli.test.ts
+++ b/packages/pg-delta/tests/cli.test.ts
@@ -17,7 +17,7 @@ import {
 import { loadSnapshot } from "../src/frontends/snapshot-file.ts";
 import { serializeSnapshot } from "../src/core/snapshot.ts";
 import { extract } from "../src/extract/extract.ts";
-import { serializePlan } from "../src/plan/artifact.ts";
+import { parsePlan, serializePlan } from "../src/plan/artifact.ts";
 import { plan } from "../src/plan/plan.ts";
 import type { Policy } from "../src/policy/policy.ts";
 import { isolatedClusterPair, sharedCluster } from "./containers.ts";
@@ -31,6 +31,17 @@ interface SpawnResult {
   exitCode: number;
 }
 
+function expectDryRunStdoutIsScript(stdout: string): void {
+  for (const diagnostic of [
+    "Dry run:",
+    "WARNING:",
+    "Plan artifact written",
+    "UNREDACTED",
+  ]) {
+    expect(stdout).not.toContain(diagnostic);
+  }
+}
+
 async function runCli(args: string[]): Promise<SpawnResult> {
   const proc = Bun.spawn(["bun", CLI, ...args], {
     cwd: PKG_DIR,
@@ -91,6 +102,44 @@ describe("CLI: error → exit-code mapping (main is the sole exiter)", () => {
   });
 });
 
+describe("CLI: apply failure attribution", () => {
+  test("renders a failing preamble statement as a control, not an action", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("cli_apply_control_tgt");
+    const desired = await cluster.createDb("cli_apply_control_desired");
+    const artifactDir = mkdtempSync(join(tmpdir(), "pgdn-apply-control-"));
+    try {
+      await desired.pool.query(`CREATE SCHEMA app`);
+      const [sourceState, desiredState] = await Promise.all([
+        extract(target.pool),
+        extract(desired.pool),
+      ]);
+      const thePlan = plan(sourceState.factBase, desiredState.factBase);
+      thePlan.preamble = [{ name: "lock_timeout", value: "-1" }];
+      const planFile = join(artifactDir, "plan.json");
+      writeFileSync(planFile, serializePlan(thePlan), "utf8");
+
+      const result = await runCli([
+        "apply",
+        "--plan",
+        planFile,
+        "--target",
+        target.uri,
+        "--force",
+      ]);
+
+      expect(result.exitCode).toBe(1);
+      expect(result.stdout).toBe("");
+      expect(result.stderr).toMatch(/  control: .*lock_timeout/i);
+      expect(result.stderr).toContain("  sql: SET LOCAL lock_timeout = -1");
+      expect(result.stderr).not.toContain("  action[0]:");
+    } finally {
+      rmSync(artifactDir, { recursive: true, force: true });
+      await Promise.all([target.drop(), desired.drop()]);
+    }
+  }, 60_000);
+});
+
 describe("CLI: --help", () => {
   test("does not recommend apply --force for unsafe plans", async () => {
     const res = await runCli(["--help"]);
@@ -1799,6 +1848,102 @@ describe("CLI: secret redaction surface", () => {
       await Promise.all([shadow.drop(), target.drop()]);
     }
   }, 120_000);
+
+  test("schema apply warns before verbose output exposes manifest-unredacted secrets", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_verbose_secret_shadow");
+    const target = await cluster.createDb("cli_verbose_secret_tgt");
+    const secret = "verbose-secret-xyz";
+    try {
+      const dir = join(tmpdir(), `pg-delta-next-verbose-secret-${Date.now()}`);
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(
+        join(dir, "01_fdw.sql"),
+        `CREATE FOREIGN DATA WRAPPER cli_verbose_fdw;\n` +
+          `CREATE SERVER cli_verbose_srv FOREIGN DATA WRAPPER cli_verbose_fdw\n` +
+          `  OPTIONS (host 'h.example.com', password '${secret}');\n`,
+      );
+      writeFileSync(
+        join(dir, ".pgdelta-export.json"),
+        JSON.stringify({ formatVersion: 1, redactSecrets: false }),
+        "utf8",
+      );
+
+      // No --unsafe-show-secrets flag: the export manifest disables redaction.
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--verbose",
+      ]);
+
+      expect(res.exitCode).toBe(0);
+      const warningText =
+        "WARNING: secrets are unredacted (--unsafe-show-secrets or the export manifest) — the verbose output may contain unredacted credentials.";
+      expect(res.stderr).toContain(warningText);
+      const warningLine =
+        res.stderr.split("\n").find((line) => line.includes(warningText)) ?? "";
+      expect(warningLine).not.toContain(secret);
+      expect(res.stderr.indexOf(warningText)).toBeLessThan(
+        res.stderr.indexOf(secret),
+      );
+    } finally {
+      await Promise.all([shadow.drop(), target.drop()]);
+    }
+  }, 120_000);
+
+  test("schema apply qualifies the verbose warning for a secret-free unredacted plan", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_verbose_no_secret_shadow");
+    const target = await cluster.createDb("cli_verbose_no_secret_tgt");
+    try {
+      const dir = join(
+        tmpdir(),
+        `pg-delta-next-verbose-no-secret-${Date.now()}`,
+      );
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `CREATE SCHEMA app;\nCREATE TABLE app.items (id integer PRIMARY KEY);\n`,
+      );
+      writeFileSync(
+        join(dir, ".pgdelta-export.json"),
+        JSON.stringify({ formatVersion: 1, redactSecrets: false }),
+        "utf8",
+      );
+
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--verbose",
+      ]);
+
+      expect(res.exitCode).toBe(0);
+      expect(res.stderr).toContain(
+        "the verbose output may contain unredacted credentials.",
+      );
+      expect(res.stderr).not.toContain(
+        "the verbose output contains UNREDACTED credentials.",
+      );
+    } finally {
+      await Promise.all([shadow.drop(), target.drop()]);
+    }
+  }, 120_000);
 });
 
 describe("CLI: schema lint", () => {
@@ -1945,3 +2090,648 @@ 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).toStartWith(
+        "-- pg-delta schema apply --dry-run\n" +
+          "-- Execute statements one at a time, in order, on one database session.\n",
+      );
+      expect(res.stdout).toContain("CREATE TABLE");
+      expect(res.stdout).toContain('"app"."t"');
+      const beginIndex = res.stdout.indexOf("BEGIN;");
+      const searchPathIndex = res.stdout.indexOf(
+        "SET LOCAL search_path = pg_catalog;",
+      );
+      const actionIndex = res.stdout.indexOf("CREATE TABLE");
+      const commitIndex = res.stdout.indexOf("COMMIT;", actionIndex);
+      expect(beginIndex).toBeGreaterThan(-1);
+      expect(searchPathIndex).toBeGreaterThan(beginIndex);
+      expect(actionIndex).toBeGreaterThan(searchPathIndex);
+      expect(commitIndex).toBeGreaterThan(actionIndex);
+      expect(res.stderr).toMatch(
+        /Dry run: \d+ action\(s\) planned; nothing applied\./,
+      );
+      expectDryRunStdoutIsScript(res.stdout);
+
+      // 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 stdout executes through psql with every transactionality boundary intact", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("cli_apply_dryrun_psql_tgt");
+    const dir = mkdtempSync(join(tmpdir(), "pg-delta-next-dryrun-psql-"));
+    try {
+      await target.pool.query(`
+        CREATE SCHEMA app;
+        CREATE TYPE app.mood AS ENUM ('sad');
+        CREATE TABLE app.items (
+          id integer PRIMARY KEY,
+          mood app.mood NOT NULL DEFAULT 'sad',
+          label text NOT NULL
+        );
+        INSERT INTO app.items (id, label) VALUES (1, 'kept');
+      `);
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `
+          CREATE SCHEMA app;
+          CREATE TYPE app.mood AS ENUM ('sad', 'ok');
+          CREATE TABLE app.items (
+            id integer PRIMARY KEY,
+            mood app.mood NOT NULL DEFAULT 'sad',
+            label text NOT NULL
+          );
+          CREATE INDEX items_label_idx ON app.items (label);
+        `,
+      );
+      const profilePath = join(dir, "concurrent-indexes.json");
+      writeFileSync(
+        profilePath,
+        JSON.stringify({
+          id: "cli-dryrun-concurrent-indexes",
+          handlers: [],
+          policy: {
+            id: "cli-dryrun-concurrent-indexes-policy",
+            serialize: [
+              {
+                match: { all: [] },
+                params: { concurrentIndexes: true },
+              },
+            ],
+          },
+        }),
+      );
+
+      const dryRun = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--target",
+        target.uri,
+        "--profile",
+        profilePath,
+        "--renames",
+        "off",
+        "--dry-run",
+      ]);
+      expect({ code: dryRun.exitCode, stderr: dryRun.stderr }).toMatchObject({
+        code: 0,
+      });
+      expectDryRunStdoutIsScript(dryRun.stdout);
+      expect(dryRun.stdout).toContain(`ALTER TYPE "app"."mood" ADD VALUE 'ok'`);
+      expect(dryRun.stdout).toContain("CREATE INDEX CONCURRENTLY");
+
+      const psql = Bun.spawn(
+        [
+          "docker",
+          "exec",
+          "-i",
+          cluster.container.getId(),
+          "psql",
+          "-X",
+          "-v",
+          "ON_ERROR_STOP=1",
+          "-U",
+          "test",
+          "-d",
+          target.name,
+          "-f",
+          "-",
+        ],
+        { stdin: "pipe", stdout: "pipe", stderr: "pipe" },
+      );
+      await psql.stdin.write(dryRun.stdout);
+      await psql.stdin.end();
+      const [psqlStdout, psqlStderr, psqlExitCode] = await Promise.all([
+        new Response(psql.stdout).text(),
+        new Response(psql.stderr).text(),
+        psql.exited,
+      ]);
+      expect({ psqlExitCode, psqlStdout, psqlStderr }).toMatchObject({
+        psqlExitCode: 0,
+      });
+
+      const state = await target.pool.query<{
+        default_expression: string;
+        enum_values: string[];
+        index_valid: boolean;
+        kept_rows: number;
+      }>(`
+        SELECT
+          pg_get_expr(d.adbin, d.adrelid) AS default_expression,
+          enum_range(NULL::app.mood)::text[] AS enum_values,
+          (SELECT i.indisvalid
+             FROM pg_index i
+             JOIN pg_class c ON c.oid = i.indexrelid
+            WHERE c.oid = 'app.items_label_idx'::regclass) AS index_valid,
+          (SELECT count(*)::int FROM app.items WHERE id = 1 AND label = 'kept') AS kept_rows
+        FROM pg_attrdef d
+        JOIN pg_attribute a
+          ON a.attrelid = d.adrelid AND a.attnum = d.adnum
+        WHERE d.adrelid = 'app.items'::regclass AND a.attname = 'mood'
+      `);
+      expect(state.rows[0]).toMatchObject({
+        default_expression: "'sad'::app.mood",
+        enum_values: ["sad", "ok"],
+        index_valid: true,
+        kept_rows: 1,
+      });
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+      await target.drop();
+    }
+  }, 90_000);
+
+  test("--dry-run reports destructive actions without applying them", async () => {
+    const cluster = await sharedCluster();
+    const target = await cluster.createDb("cli_apply_dryrun_drop_tgt");
+    try {
+      await target.pool.query(
+        `CREATE SCHEMA app; CREATE TABLE app.obsolete (id integer PRIMARY KEY);`,
+      );
+      const dir = join(tmpdir(), `pg-delta-next-dryrun-drop-${Date.now()}`);
+      mkdirSync(dir, { recursive: true });
+      writeFileSync(join(dir, "01_schema.sql"), `CREATE SCHEMA app;\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,
+        stdout: res.stdout,
+        stderr: res.stderr,
+      }).toMatchObject({ code: 0 });
+      expect(res.stdout).toContain("DROP TABLE");
+      expect(res.stdout).toContain('"app"."obsolete"');
+      expectDryRunStdoutIsScript(res.stdout);
+      const destructiveWarning = res.stderr.match(
+        /WARNING: plan contains (\d+) destructive action\(s\)\./,
+      );
+      expect(destructiveWarning).not.toBeNull();
+      const warningCount = Number(destructiveWarning?.[1]);
+      const parsed = parsePlan(readFileSync(planPath, "utf8"));
+      expect(warningCount).toBeGreaterThan(0);
+      expect(warningCount).toBe(parsed.safetyReport.destructiveActions);
+      const { rows } = await target.pool.query<{ exists: boolean }>(
+        `SELECT to_regclass('app.obsolete') IS NOT NULL AS exists`,
+      );
+      expect(rows[0]?.exists).toBe(true);
+    } 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}`);
+      expectDryRunStdoutIsScript(res.stdout);
+
+      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 }>(
+        `SELECT count(*)::int AS n FROM pg_tables WHERE schemaname = 'app'`,
+      );
+      expect(rows[0]?.n).toBe(0);
+    } finally {
+      await target.drop();
+    }
+  }, 90_000);
+
+  test("warns before attempting to write an unredacted plan artifact", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_apply_warn_order_shadow");
+    const target = await cluster.createDb("cli_apply_warn_order_tgt");
+    const dir = mkdtempSync(join(tmpdir(), "pg-delta-next-warn-order-"));
+    const unwritablePlanPath = join(dir, "plan.json");
+    try {
+      writeFileSync(
+        join(dir, "01_schema.sql"),
+        `CREATE SCHEMA app;\nCREATE TABLE app.t (id integer PRIMARY KEY);\n`,
+      );
+      writeFileSync(
+        join(dir, ".pgdelta-export.json"),
+        JSON.stringify({ formatVersion: 1, redactSecrets: false }),
+        "utf8",
+      );
+      // A directory at the output path makes writeFileSync fail. The warning
+      // must already be on stderr before that attempted exposure can fail or
+      // partially write an artifact.
+      mkdirSync(unwritablePlanPath);
+
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+        "--dry-run",
+        "--out-plan",
+        unwritablePlanPath,
+      ]);
+
+      expect(res.exitCode).toBe(1);
+      expect(res.stderr).toContain(
+        "the plan artifact may contain unredacted credentials.",
+      );
+      expect(res.stderr).not.toContain(
+        "the dry-run script may contain unredacted credentials.",
+      );
+      expect(res.stdout).toBe("");
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+      await Promise.all([shadow.drop(), 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) — 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");
+      expect(res.stdout).toBe("");
+
+      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);
+
+  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 {
+      const secret = "cli-dryrun-secret-xyz";
+      await source.pool.query(`
+        CREATE FOREIGN DATA WRAPPER cli_dryrun_fdw;
+        CREATE SERVER cli_dryrun_srv FOREIGN DATA WRAPPER cli_dryrun_fdw
+          OPTIONS (host 'h.example.com', password '${secret}');
+      `);
+      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("may contain unredacted credentials"));
+      expect(warnings.length).toBe(2);
+      expectDryRunStdoutIsScript(res.stdout);
+      expect(res.stdout).toContain(secret);
+      expect(res.stdout).not.toContain("__OPTION_PASSWORD__");
+      expect(res.stderr).not.toContain(secret);
+
+      // 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 serializedPlan = readFileSync(planPath, "utf8");
+      expect(serializedPlan).toContain(secret);
+      expect(serializedPlan).not.toContain("__OPTION_PASSWORD__");
+      const parsed = parsePlan(serializedPlan);
+      expect(parsed.redactSecrets).toBe(false);
+    } finally {
+      await Promise.all([source.drop(), target.drop()]);
+    }
+  }, 90_000);
+
+  test("warns before a manifest-unredacted failed action diagnostic without --verbose", async () => {
+    const cluster = await sharedCluster();
+    const source = await cluster.createDb("cli_apply_failure_unred_src");
+    const shadow = await cluster.createDb("cli_apply_failure_unred_shadow");
+    const target = await cluster.createDb("cli_apply_failure_unred_tgt");
+    const secret = "cli-apply-failure-secret-xyz";
+    const dir = join(tmpdir(), `pg-delta-next-failure-unred-${Date.now()}`);
+    const rejectTargetCreateServer = `
+      CREATE FUNCTION public.cli_fail_secret_ddl() RETURNS event_trigger
+      LANGUAGE plpgsql AS $$
+      BEGIN
+        IF current_database() = '${target.name}' AND TG_TAG = 'CREATE SERVER' THEN
+          RAISE EXCEPTION 'blocked secret-bearing DDL: %', current_query();
+        END IF;
+      END
+      $$;
+      CREATE EVENT TRIGGER cli_fail_secret_ddl
+        ON ddl_command_start
+        EXECUTE FUNCTION public.cli_fail_secret_ddl();
+    `;
+    try {
+      await Promise.all([
+        source.pool.query(`
+          CREATE FOREIGN DATA WRAPPER cli_failure_fdw;
+          CREATE SERVER cli_failure_srv FOREIGN DATA WRAPPER cli_failure_fdw
+            OPTIONS (host 'h.example.com', password '${secret}');
+          ${rejectTargetCreateServer}
+        `),
+        target.pool.query(`
+          CREATE FOREIGN DATA WRAPPER cli_failure_fdw;
+          ${rejectTargetCreateServer}
+        `),
+      ]);
+
+      const exported = await runCli([
+        "schema",
+        "export",
+        "--source",
+        source.uri,
+        "--out-dir",
+        dir,
+        "--unsafe-show-secrets",
+      ]);
+      expect(exported.exitCode).toBe(0);
+
+      // No --unsafe-show-secrets and no --verbose: the export manifest alone
+      // selects unredacted extraction, and only the final failure diagnostic
+      // exposes the action SQL.
+      // The desired function/trigger already exists identically on source and
+      // target. It is inert while exporting and loading the shadow, but rejects
+      // CREATE SERVER deterministically when the plan runs on this target.
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+      ]);
+
+      expect(res.exitCode).toBe(1);
+      expect(res.stdout).toBe("");
+      const warning = res.stderr
+        .split("\n")
+        .find((line) => line.includes("may contain unredacted credentials"));
+      expect(warning).toBeDefined();
+      expect(warning).not.toContain(secret);
+      const warningIndex = res.stderr.indexOf(warning!);
+      const failedSqlIndex = res.stderr.indexOf(`  sql: `);
+      const firstSecretIndex = res.stderr.indexOf(secret);
+      expect(firstSecretIndex).toBeGreaterThanOrEqual(0);
+      expect(warningIndex).toBeLessThan(firstSecretIndex);
+      expect(failedSqlIndex).toBeGreaterThan(warningIndex);
+      expect(res.stderr.indexOf(secret, firstSecretIndex + 1)).toBeGreaterThan(
+        failedSqlIndex,
+      );
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+      await Promise.all([source.drop(), shadow.drop(), target.drop()]);
+    }
+  }, 90_000);
+
+  test("warns before a manifest-unredacted planning failure can expose a secret", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_plan_failure_unred_shadow");
+    const target = await cluster.createDb("cli_plan_failure_unred_tgt");
+    const secret = "cli-planning-failure-secret-xyz";
+    const secretStatement =
+      `CREATE SERVER cli_planning_secret_srv ` +
+      `FOREIGN DATA WRAPPER cli_planning_secret_fdw ` +
+      `OPTIONS (password '${secret}')`;
+    const dir = mkdtempSync(
+      join(tmpdir(), "pg-delta-next-planning-failure-unred-"),
+    );
+    try {
+      // PostgreSQL itself puts the credential-bearing statement in
+      // DatabaseError.message, exercising the generic planning-error path
+      // rather than a later action report assembled by the CLI.
+      writeFileSync(
+        join(dir, "01_rejected.sql"),
+        `DO $body$\n` +
+          `BEGIN\n` +
+          `  RAISE EXCEPTION 'shadow rejected statement: ${secretStatement.replaceAll("'", "''")}';\n` +
+          `END\n` +
+          `$body$;\n`,
+      );
+      writeFileSync(
+        join(dir, ".pgdelta-export.json"),
+        JSON.stringify({ formatVersion: 1, redactSecrets: false }),
+        "utf8",
+      );
+
+      // No --unsafe-show-secrets: the manifest selects unredacted planning.
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+      ]);
+
+      expect(res.exitCode).toBe(1);
+      expect(res.stdout).toBe("");
+      const warningNeedle = "may contain unredacted credentials";
+      const warning = res.stderr
+        .split("\n")
+        .find((line) => line.includes(warningNeedle));
+      const warningIndex = res.stderr.indexOf(warningNeedle);
+      const firstSecretIndex = res.stderr.indexOf(secret);
+      expect(firstSecretIndex).toBeGreaterThanOrEqual(0);
+      expect(warningIndex).toBeGreaterThanOrEqual(0);
+      expect(warning).toBeDefined();
+      expect(warning).not.toContain(secret);
+      expect(warningIndex).toBeLessThan(firstSecretIndex);
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+      await Promise.all([shadow.drop(), target.drop()]);
+    }
+  }, 90_000);
+
+  test("does not warn for an early manifest-unredacted empty-shadow guard", async () => {
+    const cluster = await sharedCluster();
+    const shadow = await cluster.createDb("cli_plan_early_unred_shadow");
+    const target = await cluster.createDb("cli_plan_early_unred_tgt");
+    const dir = mkdtempSync(
+      join(tmpdir(), "pg-delta-next-planning-early-unred-"),
+    );
+    try {
+      await shadow.pool.query(`CREATE TABLE public.already_here (id integer)`);
+      writeFileSync(
+        join(dir, "01_table.sql"),
+        `CREATE TABLE public.wanted (id integer);\n`,
+      );
+      writeFileSync(
+        join(dir, ".pgdelta-export.json"),
+        JSON.stringify({ formatVersion: 1, redactSecrets: false }),
+        "utf8",
+      );
+
+      const res = await runCli([
+        "schema",
+        "apply",
+        "--dir",
+        dir,
+        "--shadow",
+        shadow.uri,
+        "--target",
+        target.uri,
+        "--renames",
+        "off",
+      ]);
+
+      expect(res.exitCode).toBe(1);
+      expect(res.stderr).toContain("shadow database is not empty");
+      expect(res.stderr).not.toContain("may contain unredacted credentials");
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+      await Promise.all([shadow.drop(), target.drop()]);
+    }
+  }, 90_000);
+});