diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0eb849e..5512461 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -37,6 +37,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   carry a `guidance` field naming the typed manual path, drawn from the
   suggest report's Guidance vocabulary. The fingerprint definition is
   unchanged.
+- **`ALTER COLUMN ... DROP NOT NULL` now classifies as destructive** in plan
+  reports (`destructive: true`): dropping `NOT NULL` discards the same
+  guarantee as dropping the equivalent constraint. The full destructive set
+  is a dropped column, constraint, index, or `NOT NULL`; `DROP DEFAULT` is
+  deliberately not destructive — a default guarantees nothing about existing
+  rows and is recreated by a metadata-only statement. A consumer gating on
+  `.statements[].destructive` now sees `DROP NOT NULL` flagged, and
+  desired-state execution refuses it like any other drop.
 - **The suggest report is format version 2**: the Guidance vocabulary gains
   `name-constraint-then-validate`, emitted for an unnamed `ADD CHECK` /
   `ADD FOREIGN KEY`, and `unique-index-then-constraint`, covering an
@@ -102,6 +110,21 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **Library-level desired-state execution: `migrate.RunDesired`** converges
+  one live table onto its parsed desired schema — derive the convergence
+  plan, admit it as a whole (table existence, destructive guard, routed
+  dispositions, optional `ExpectedFingerprint` pin), then run each planned
+  statement back through the same `migrate.Run` pipeline with fresh
+  introspection and classification, stopping at the first refusal or
+  failure. The result carries the plan, per-statement verdicts, and an
+  aggregate outcome with committed-prefix detail
+  ([docs/execution-model.md](docs/execution-model.md)). Two new refusal
+  reasons enter the vocabulary: `destructive-change` (the plan discards
+  live structure — a dropped column, constraint, index, or `NOT NULL`;
+  desired-state execution never runs it) and
+  `plan-fingerprint-mismatch` (the plan derived at execution time is not
+  the pinned reviewed plan). Library-only for now — the `migrate --desired`
+  CLI flag follows separately.
 - **A third verdict outcome, `failed`,** for execution failures (still exit
   1 — refusals remain exit 2). The verdict carries the executor's stable
   outcome code in `code`, and for a mid-sequence failure the 1-based
diff --git a/SAFETY.md b/SAFETY.md
index 467202d..83d5f2d 100644
--- a/SAFETY.md
+++ b/SAFETY.md
@@ -30,7 +30,7 @@ The invariant registry (invariant IDs referenced below) lives in
 | `pkg/statement`, `pkg/planner`, `pkg/schemadiff`, `pkg/router`, `pkg/plan`, `pkg/lint`, `pkg/suggest` — classify/diff/route/report | ❌ periphery¹ | `pkg/statement` (parse boundary), `pkg/schemadiff` (introspect/diff via scratch execute-and-introspect), `pkg/planner` (classifier), `pkg/router` (backend assignment + availability policy), `pkg/plan` (versioned dry-run plan report), `pkg/lint` (offline typed findings), and `pkg/suggest` (advisory rewrites with typed caveats) exist (Phases 2.1–2.5) | (CO-7 holds at the parse boundary) |
 | `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
 | `pkg/diffplan` — desired schema → routed convergence plan, the declarative front door as a library (the CLI `diff` and embedding orchestrators share it) | ❌ periphery | exists | — |
-| `pkg/migrate` — one gated statement → resolve, classify, route, execute → one verdict; the imperative front door as a library (the CLI `migrate` and embedding orchestrators share it) | ❌ periphery² | exists | — |
+| `pkg/migrate` — one gated statement → resolve, classify, route, execute → one verdict; the imperative front door as a library (the CLI `migrate` and embedding orchestrators share it), plus the desired-state execution loop (`RunDesired`: derive the convergence plan, admit it as a whole, run each planned statement back through the same pipeline) | ❌ periphery² | exists | — |
 | `internal/cli` — CLI, flags, help, prompts | ❌ periphery | `migrate`, `status`, `diff`, `fmt`, `lint`, and `suggest` exist | — |
 | `pkg/progress` — strategy-wide progress snapshots; the executors' observation seam (core imports it, so its locking discipline is core-critical); copy counters reserved for later | ✅ core | native progress exists | — |
 | orchestrator adapter | ❌ periphery | planned (Phase 11) | OC-* hold *at* the boundary |
@@ -46,7 +46,19 @@ The core executors re-verify their own preconditions and never trust that the pl
 pipeline — gate, resolve, preflight, execute — but every dangerous step it requests is enforced
 by the core packages it calls: the executors re-verify admission and run under their own bounded
 budgets, and preflight's proof types gate what may execute. A wrong sequencing decision in
-`pkg/migrate` yields a refusal or a bounded failed attempt, never an unbounded lock.
+`pkg/migrate` yields a refusal or a bounded failed attempt, never an unbounded lock. The
+desired-state loop inherits that argument for every *execution-time* property: it executes
+nothing itself — every planned statement goes back through `Run`, so each one is
+re-introspected, re-classified, re-routed, and re-preflighted at execution time, and a plan
+the loop wrongly admits still cannot make the core exceed a lock budget or skip a preflight.
+**One admission check has no core backstop: the destructive guard.** The core has no concept
+of destructiveness — `pkg/executor` and `pkg/preflight` never check it — so refusing a
+destructive desired-state plan rests entirely on `RunDesired`'s admission gate and on the
+classifier's `Destructive` derivation in `pkg/planner`, and its failure mode is data loss (a
+falsely-admitted `DROP COLUMN` commits), not a refusal or a bounded failed attempt. Those two
+sites are the exception to the periphery posture: treat `destructiveOp` and the desired-state
+admission gate with the core's review bar — spec-first, test-first, small diffs — even though
+their packages stay periphery for everything else they do.
 
 ## Rules inside the core
 
diff --git a/docs/architecture.md b/docs/architecture.md
index 111c292..2440b56 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -190,7 +190,7 @@ different levels of commitment:
 | `pkg/suggest` | Offline advisory surface: maps risky DDL to the safer native form the engine would run, with typed caveats and manual-path guidance; emits the versioned suggest report ([suggest-report.md](suggest-report.md)) | exists |
 | `pkg/plan` | Versioned machine-readable dry-run plan report — the one JSON contract both front doors emit and an orchestrator consumes | exists (Phase 2.5) |
 | `pkg/diffplan` | The declarative front door as a library: desired schema in, routed `plan.Report` out — the CLI `diff` and embedding orchestrators share this one pipeline | exists |
-| `pkg/migrate` | The imperative front door as a library: one parsed statement in — gate, resolve, classify, route, execute — one `verdict.Verdict` out; the CLI `migrate` and embedding orchestrators share this one pipeline | exists |
+| `pkg/migrate` | The imperative front door as a library: one parsed statement in — gate, resolve, classify, route, execute — one `verdict.Verdict` out; the CLI `migrate` and embedding orchestrators share this one pipeline. Also the desired-state execution loop: `RunDesired` derives the convergence plan (`diffplan.Plan`), admits it as a whole (existence, destructive guard, dispositions, optional fingerprint pin), and runs each planned statement back through `Run` — per-statement verdicts, committed-prefix semantics | exists |
 | `pkg/router` | Route classified statements to native / copy-and-swap / refuse dispositions; copy-and-swap reports unavailable until that backend lands | exists (Phase 2.4) |
 | `pkg/executor` | Bounded optimistic native attempt, the concurrent index build, and the autocommit safer-sequence runner, with stable outcome codes; the full `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) arrives with the copy-and-swap backend | native execution exists |
 | `pkg/progress` | Strategy-wide, pollable progress snapshots: native phase/elapsed time, sequence position, retry attempt, and server-reported concurrent-index work; optional copy counters are reserved for copy-and-swap | native progress exists |
diff --git a/docs/cli-output-examples.md b/docs/cli-output-examples.md
index 0e4499a..2e4f0b5 100644
--- a/docs/cli-output-examples.md
+++ b/docs/cli-output-examples.md
@@ -37,6 +37,7 @@ a verdict, not a plan report — and exit 2. The JSON report schema is
 [plan-report.md](plan-report.md).
 
 - [Codes used in these examples](#codes-used-in-these-examples)
+- [Refusal reasons](#refusal-reasons)
 - [Migrate](#migrate)
   - [Runs as written (`metadata-only`) — exit 0](#runs-as-written-metadata-only--exit-0)
   - [Safer-sequence substitution (`safer-idiom`) — exit 0](#safer-sequence-substitution-safer-idiom--exit-0)
@@ -68,6 +69,25 @@ is a one-line summary; the linked reference entry is authoritative.
 | [`destructive`](postgres-online-ddl-reference.md#destructive) | The change discards live data or structure (`DROP COLUMN`, `DROP TABLE`, truncating conversions). A warning alongside the routing decision, not a refusal. |
 | [`blocking-idiom`](lint-report.md#codes-code) | Lint-only code: the submitted form blocks readers or writers and a safer native form exists; the finding's `suggestion` carries the safer SQL when the linter can construct it. |
 
+## Refusal reasons
+
+Every refusal verdict (`"outcome": "refused"`, exit 2) carries exactly one of
+these typed `reason` tokens — the value automation switches on; prose belongs
+in `detail`. The set is closed and pinned by test (`verdict.Reasons()`).
+
+| Reason | Meaning |
+|---|---|
+| `unsupported-statement` | No safe path is known for the statement — only `ALTER TABLE` and `CREATE INDEX` reach classification — or a desired-state plan needs a table that does not exist yet. |
+| `index-statement` | Index maintenance (`DROP INDEX`, `REINDEX`) has a native safe idiom (`CONCURRENTLY`) and is never attempted; the verdict's `safer_idiom` names it. |
+| `not-native-safe-table-too-large` | The size guard skipped the optimistic attempt: the table exceeds the configured bound and the change is not provably metadata-only. |
+| `insufficient-privileges` | The connected role lacks the access the change needs; `detail` names the exact missing GRANT (see [engine-role.md](engine-role.md)). |
+| `unsupported-partitioned-parent` | The routed plan builds an index on a partitioned parent, where PostgreSQL cannot `CREATE INDEX CONCURRENTLY`. |
+| `not-native-safe-budget-exceeded` | The optimistic attempt exceeded its lock or statement budget and was cancelled; the verdict's `cause` narrows which budget fired. |
+| `not-native-safe-rewrite-required` | The submitted form blocks and must run as a safer native sequence, but none could be constructed. |
+| `backend-unavailable` | The change routes to an execution strategy this build does not implement (copy-and-swap). |
+| `destructive-change` | The desired-state plan discards live structure — a dropped column, constraint, index, or `NOT NULL` — and desired-state execution runs no destructive statement; run the drop deliberately instead ([execution model](execution-model.md)). |
+| `plan-fingerprint-mismatch` | The plan recomputed at execution time does not carry the pinned fingerprint: the plan a reviewer approved is not the plan that would execute, so nothing runs ([execution model](execution-model.md)). |
+
 ## Migrate
 
 The imperative front door: submit one DDL statement; pg-sprite classifies
diff --git a/docs/execution-model.md b/docs/execution-model.md
index 2c889ee..b4a4d1d 100644
--- a/docs/execution-model.md
+++ b/docs/execution-model.md
@@ -201,3 +201,12 @@ desired state — must report the statements already committed, the one that
 failed, and the ones never attempted. That is the strongest guarantee
 available once PostgreSQL rules out atomicity for online DDL, and it is the
 contract embedders should expose rather than paper over.
+
+The engine implements that contract itself: `migrate.RunDesired` (in
+[`pkg/migrate`](../pkg/migrate/desired.go)) derives the convergence plan for
+a desired-state schema and executes each planned statement back through the
+same pipeline — fresh introspection, classification, and routing per
+statement — stopping at the first refusal or failure. Its result carries the
+plan, one verdict per attempted statement, and a detail naming exactly which
+planned statements committed and remain in effect: the committed prefix at
+the plan level, statements instead of steps.
diff --git a/docs/limitations.md b/docs/limitations.md
index d0105be..acce085 100644
--- a/docs/limitations.md
+++ b/docs/limitations.md
@@ -29,3 +29,34 @@ with a typed refusal — never a silently wrong or incomplete result:
 | Column collations | An explicit `COLLATE` on a column is not managed: converging a collation delta rewrites the column and its indexes. Export refuses a collated column — a baseline without the clause would silently change sort order and index semantics — and a collation delta (including on an added column) is a typed `diff` refusal. |
 | Non-table objects | Views, materialized views, standalone sequences, enums, domains, extensions, functions, and triggers are outside the model. A serial column's owned sequence is the one exception: it round-trips through the `serial` pseudo-types — and ownership is verified through the catalog (`pg_depend`), so a hand-written `nextval` default on a standalone sequence that merely carries the serial-style name refuses rather than exporting as `serial` and silently privatizing a shared sequence. A column may *use* an unmanaged type (an enum, a domain) — the type text round-trips — but the type's definition is not managed. |
 | Multiple tables per file | A desired file is single-table scoped: exactly one `CREATE TABLE` plus `CREATE INDEX` statements on it. Multi-table schemas are managed as one file per table. |
+| Changed index or constraint definition | A redefinition diffs to drop-and-recreate, the drop is destructive, and desired-state execution refuses any plan containing a destructive statement — the whole plan, including the harmless recreate. Run the drop deliberately first (`DROP INDEX CONCURRENTLY` directly against the database; `ALTER TABLE ... DROP CONSTRAINT` through the imperative front door), then rerun — the remaining plan converges the recreate. |
+
+## What desired-state execution converges today
+
+Desired-state execution (`migrate.RunDesired`, library-only today) feeds
+every planned statement back through the same gates as the imperative
+front door, so the outcome of an ordinary desired-file edit is the
+composition of the model boundaries above with those gates. At a glance:
+
+| Desired-file edit | Outcome today |
+| --- | --- |
+| Add a column | Converges. Runs as a bounded attempt of the submitted form, so the table-size guard applies (below). |
+| Widen a column type (`varchar(50)` → `varchar(255)`) | Converges — the same bounded attempt, under the same size guard. |
+| Add an index | Converges via `CREATE INDEX CONCURRENTLY`. Not size-guarded: long online work on a large table is the pattern's purpose. |
+| Add a constraint (`UNIQUE`, `CHECK`); `SET NOT NULL` | Converges via the safer online sequence; not size-guarded either. |
+| Relax a `NOT NULL` | Refused, whole plan: dropping `NOT NULL` discards the same guarantee its constraint form would, so it is destructive. Run it deliberately through the imperative front door — it executes natively there — then rerun. |
+| Change an index or constraint definition | Refused, whole plan — the drop-and-recreate row above. |
+| Narrow a column type (`varchar(255)` → `varchar(50)`) | Refused: the change routes to copy-and-swap, which is not yet available. |
+| A bounded-attempt edit on a table above the size threshold | Refused at that statement (`not-native-safe-table-too-large`): with the default 1 GiB threshold, adding a column to a larger table refuses until the threshold is raised. |
+
+A destructive refusal is all-or-nothing: one destructive statement refuses
+the whole plan, and the non-destructive statements beside it do not run —
+the refusal detail says how many were skipped.
+
+The size threshold is policy, not capability: `Options.MaxTableSizeBytes`
+(the CLI's `--max-table-size`) defaults to 1 GiB, and the guard covers only
+the blind bounded attempt of a submitted form — planner-proven online
+sequences are exempt. On a table you operate deliberately, raising the
+threshold is the sanctioned way to converge the bounded-attempt edits; the
+refusal means pg-sprite cannot prove the change is instant at that size,
+not that the change is unsafe.
diff --git a/docs/plan-report.md b/docs/plan-report.md
index 3d90e17..31c0f5d 100644
--- a/docs/plan-report.md
+++ b/docs/plan-report.md
@@ -58,7 +58,7 @@ consumer rendering either into a shared surface must clamp and escape them.
 |---|---|---|---|
 | `sql` | string | always | The statement in the engine's **canonical rendering**: parsed and reprinted through the PostgreSQL deparser, whichever front door derived it. Never a verbatim echo of submitted text — the same change carries the same string through either door. Commented input is refused rather than silently stripped; optional noise words follow the grammar's canonical spelling. |
 | `kind` | string | diff source only | Classifies a diff-derived statement (see Kinds) so a consumer can gate whole classes of change. Absent for the alter source: a submitted statement may carry several operations and has no single kind. |
-| `destructive` | bool | always | Marks statements that discard live structure — a dropped column, constraint, or index. Derived from the classifier's decisions, so both sources report it identically by construction. Always emitted, never omitted: a safety flag a consumer gates on must be explicit even when false. |
+| `destructive` | bool | always | Marks statements that discard live structure — a dropped column, constraint, index, or `NOT NULL`. Derived from the classifier's decisions, so both sources report it identically by construction. Always emitted, never omitted: a safety flag a consumer gates on must be explicit even when false. |
 | `route` | string | always | The planner's aggregate route for the statement (see Routes). |
 | `backend` | string | except refusals | The assigned execution strategy (see Backends); absent for refusals. |
 | `disposition` | string | always | What execution would do with this statement now (see Dispositions). |
diff --git a/docs/postgres-online-ddl-reference.md b/docs/postgres-online-ddl-reference.md
index 020aba1..7310c25 100644
--- a/docs/postgres-online-ddl-reference.md
+++ b/docs/postgres-online-ddl-reference.md
@@ -408,9 +408,13 @@ an unclassified statement is never executed.
 |---|---|---|---|
 | warning — does not change the routing decision | per the routing decision | per the routing decision | unchanged by this code |
 
-The change discards live data or structure (`DROP COLUMN`, `DROP TABLE`,
-truncating conversions). Emitted alongside the routing decision as a warning so
-destructive intent is always visible in review.
+The change discards live data or structure: `DROP COLUMN`, `DROP TABLE`,
+truncating conversions, a dropped constraint or index, or `DROP NOT NULL`
+(which discards the same guarantee as dropping the equivalent constraint;
+`DROP DEFAULT` is deliberately not destructive — a default guarantees nothing
+about existing rows and is recreated by a metadata-only statement). Emitted
+alongside the routing decision as a warning so destructive intent is always
+visible in review.
 
 ### `rewrite-required`
 
diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md
index 9144ee1..a97bcfe 100644
--- a/docs/schemabot-integration.md
+++ b/docs/schemabot-integration.md
@@ -77,7 +77,7 @@ the integration phase starts; they drift.)
 | --- | --- |
 | `Name()` | a stable identifier, e.g. `"pg-sprite"` |
 | `Plan` | run parse → declarative diff when applicable → classify → route — exported as `diffplan.Plan` in [`pkg/diffplan`](../pkg/diffplan/diffplan.go) (parse via `statement.ParseDesired`, connect via `dbconn.NewPool`, inputs named by `diffplan.Request{Schema, Desired}`); return a `PlanResult` whose `SchemaChange.TableChanges` are `engine.TableChange{Table, Operation (statement.StatementType), DDL, IsUnsafe, UnsafeReason, ExecutionMode, ModeReason}`; map a **not native-safe** refusal to `engine.ExecutionModeBlocked` with the refusal reason as `ModeReason` (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)) |
-| `Apply` | start the native executor asynchronously and return immediately — the synchronous core is exported as `migrate.Run` in [`pkg/migrate`](../pkg/migrate/migrate.go) (parse via `statement.ParseOne`, gate via `migrate.Gate` before dialing, connect via `dbconn.NewPool`, policy via `migrate.DefaultOptions` tuned per table): one statement in, one `verdict.Verdict` out, with a three-shape contract — refusal (verdict, nil error), execution failure (failed verdict carrying the stable code and the committed prefix, plus the operational error), or an error with a zero verdict (stopped before executing). `Run` re-resolves the routing decision at execution time, so the adapter never trusts the stored plan-time verdict (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)) |
+| `Apply` | start the native executor asynchronously and return immediately — the synchronous core is exported as `migrate.Run` in [`pkg/migrate`](../pkg/migrate/migrate.go) (parse via `statement.ParseOne`, gate via `migrate.Gate` before dialing, connect via `dbconn.NewPool`, policy via `migrate.DefaultOptions` tuned per table): one statement in, one `verdict.Verdict` out, with a three-shape contract — refusal (verdict, nil error), execution failure (failed verdict carrying the stable code and the committed prefix, plus the operational error), or an error with a zero verdict (stopped before executing). `Run` re-resolves the routing decision at execution time, so the adapter never trusts the stored plan-time verdict (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)). For the declarative flow the adapter does not iterate the plan itself: `migrate.RunDesired` takes the parsed desired schema, re-derives the convergence plan, and runs every planned statement back through `Run` — returning per-statement verdicts with committed-prefix semantics ([execution-model.md](execution-model.md)) — so plan-vs-execute drift and per-statement re-gating stay inside the engine, and the adapter can pin the reviewed plan with `ExpectedFingerprint` |
 | `Progress` | per-table rows-copied / total / percent / ETA / checksum state |
 | `Stop` / `Start` | checkpoint and resume (slot + copy + applier watermark) |
 | `Cutover` | the deferred, operator-gated atomic swap |
diff --git a/pkg/migrate/desired.go b/pkg/migrate/desired.go
new file mode 100644
index 0000000..ce8ba2f
--- /dev/null
+++ b/pkg/migrate/desired.go
@@ -0,0 +1,294 @@
+package migrate
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/block/pg-sprite/pkg/diffplan"
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/plan"
+	"github.com/block/pg-sprite/pkg/router"
+	"github.com/block/pg-sprite/pkg/schemadiff"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/block/pg-sprite/pkg/verdict"
+)
+
+// stoppedBeforeVerdict is the committed-prefix wording for a statement the
+// pipeline stopped on without reaching a verdict: unlike a failed verdict,
+// nothing about the statement was executed.
+const stoppedBeforeVerdict = "stopped before a verdict; nothing about it was executed"
+
+// ErrForceNotSupported is returned by [RunDesired] when Options.Force is
+// set: the force acknowledgement applies to the imperative front door
+// only — desired-state execution never runs a submitted form blind. It is
+// a sentinel so an embedder can tell the unsupported option apart from an
+// operational failure with [errors.Is].
+var ErrForceNotSupported = errors.New(
+	"the force acknowledgement applies to the imperative front door only; desired-state execution never runs a submitted form blind")
+
+// DesiredRequest names the inputs to [RunDesired]. Zero-value fields are
+// invalid: the schema must be set, and the desired state must come from
+// [statement.ParseDesired] — the zero DesiredSchema is refused.
+type DesiredRequest struct {
+	// Schema is the target schema the desired table lives in.
+	Schema string
+	// Desired is the parsed desired-state schema for the table.
+	Desired statement.DesiredSchema
+	// ExpectedFingerprint optionally pins the plan: when set, the plan
+	// derived at execution time must carry exactly this fingerprint
+	// (plan.Report.Fingerprint) or nothing runs. It is how a caller that
+	// had a plan reviewed enforces that the plan the reviewer approved is
+	// the plan that executes; empty skips the check.
+	ExpectedFingerprint string
+}
+
+// DesiredResult is the aggregate report for one desired-state execution:
+// the plan that was derived, the verdict of every statement that was
+// attempted, and the overall outcome.
+//
+// Verdicts[i] is the verdict of Plan.Statements[i]; fewer verdicts than
+// planned statements means execution stopped and the remaining statements
+// were never attempted. The committed prefix is read from the verdicts:
+// every executed verdict committed in full, and a failed verdict's own
+// ExecutedSQL discloses the committed steps inside the statement that
+// failed. Whether anything changed is read from the plan: an executed
+// outcome with empty Plan.Statements is the no-op signal — the live table
+// already matched the desired schema and nothing ran.
+type DesiredResult struct {
+	// Plan is the convergence plan derived at execution time; empty
+	// Plan.Statements means the live table already matched the desired
+	// schema.
+	Plan plan.Report `json:"plan"`
+	// Verdicts are the per-statement verdicts, in plan order, one per
+	// attempted statement.
+	Verdicts []verdict.Verdict `json:"verdicts,omitempty"`
+	// Outcome is what happened overall: executed when every planned
+	// statement committed (or there was nothing to run), refused when the
+	// plan or one of its statements was refused and execution stopped,
+	// failed when execution stopped on an operational error. A failed
+	// result whose stopping statement has no verdict means the pipeline
+	// stopped before reaching one — nothing about that statement was
+	// executed; a failed verdict means the statement was attempted and
+	// failed. Detail says which.
+	Outcome verdict.Outcome `json:"outcome"`
+	// Reason is the typed refusal cause; empty unless Outcome is refused.
+	Reason verdict.Reason `json:"reason,omitempty"`
+	// Detail is the human explanation: why refused, what committed, or
+	// that there was nothing to do.
+	Detail string `json:"detail,omitempty"`
+}
+
+// RunDesired converges one live table onto its desired-state schema:
+// derive the convergence plan with [diffplan.Plan], admit the plan as a
+// whole, then execute each planned statement back through the full [Run]
+// pipeline — fresh introspection, classification, and routing per
+// statement, so a statement that became unsafe after planning refuses
+// instead of running — stopping at the first refusal or failure.
+//
+// Plan-time admission is all-or-nothing: a plan that needs a table that
+// does not exist yet, contains a destructive statement, routes any
+// statement away from execution, or does not match the pinned fingerprint
+// is refused before anything runs. Execution-time semantics are
+// committed-prefix: once statements start running, an executed statement
+// stays committed even when a later one refuses or fails, and the result's
+// verdicts disclose exactly how far convergence got.
+//
+// The result-and-error contract mirrors [Run]'s three shapes. A refusal —
+// at plan admission or on a mid-plan statement — returns the result with a
+// nil error. An execution failure returns the failed result together with
+// the operational error. An error with a zero result means the pipeline
+// stopped before planning or executing anything.
+//
+// Options.Force is rejected: the declarative front door never runs a
+// submitted form blind. A destructive or force-worthy change belongs on
+// the imperative front door where the operator states it explicitly.
+//
+// RunDesired does not close the pool; one pool serves any number of calls.
+func RunDesired(ctx context.Context, pool *pgxpool.Pool, req DesiredRequest, opts Options) (DesiredResult, error) {
+	if err := opts.validate(); err != nil {
+		return DesiredResult{}, err
+	}
+	if opts.Force != "" {
+		return DesiredResult{}, ErrForceNotSupported
+	}
+	report, err := diffplan.Plan(ctx, pool, diffplan.Request{Schema: req.Schema, Desired: req.Desired})
+	if err != nil {
+		return DesiredResult{}, err
+	}
+	opts.logger().Debug("desired plan derived",
+		"schema", report.Schema, "table", report.Table,
+		"statements", len(report.Statements), "fingerprint", report.Fingerprint)
+
+	// An already-converged table resolves before admission: an empty plan
+	// carries no plan identity for a pinned fingerprint to verify (every
+	// empty plan hashes alike), and nothing runs either way — which is all
+	// a pin protects. Checking the pin first would refuse a legitimate
+	// re-run of an approved plan that already converged.
+	if len(report.Statements) == 0 {
+		return DesiredResult{
+			Plan:    report,
+			Outcome: verdict.OutcomeExecuted,
+			Detail:  "already converged: the live table matches the desired schema; nothing to run",
+		}, nil
+	}
+	if refused, ok := admitPlan(req, report); !ok {
+		return refused, nil
+	}
+
+	result := DesiredResult{Plan: report}
+	for i, ps := range report.Statements {
+		st, err := statement.ParseOne(ps.SQL)
+		if err != nil {
+			// The planned SQL is engine-generated: a reparse failure is a
+			// breach of the engine's own contract, not a database problem.
+			result.Outcome = verdict.OutcomeFailed
+			result.Detail = committedPrefixDetail(i, len(report.Statements), stoppedBeforeVerdict)
+			return result, fmt.Errorf("%w: planned statement %d is engine-generated SQL its own parser rejects: %w",
+				executor.ErrInvariantViolation, i+1, err)
+		}
+		v, runErr := Run(ctx, pool, st, opts)
+		if runErr != nil {
+			result.Outcome = verdict.OutcomeFailed
+			// A zero verdict is Run's "stopped before reaching a verdict"
+			// shape: nothing about the statement was executed, so the
+			// detail must not call the statement failed.
+			what := stoppedBeforeVerdict
+			if v.Outcome != "" {
+				result.Verdicts = append(result.Verdicts, v)
+				what = "failed"
+			}
+			result.Detail = committedPrefixDetail(i, len(report.Statements), what)
+			return result, fmt.Errorf("planned statement %d: %w", i+1, runErr)
+		}
+		result.Verdicts = append(result.Verdicts, v)
+		if v.Outcome == verdict.OutcomeRefused {
+			result.Outcome = verdict.OutcomeRefused
+			result.Reason = v.Reason
+			result.Detail = committedPrefixDetail(i, len(report.Statements), "was refused at execution time")
+			return result, nil
+		}
+	}
+	result.Outcome = verdict.OutcomeExecuted
+	result.Detail = fmt.Sprintf("converged: all %d planned statements committed", len(report.Statements))
+	return result, nil
+}
+
+// admitPlan is the all-or-nothing plan-time admission: it refuses the whole
+// plan — before anything runs — when the plan cannot converge the table as
+// a unit. The checks run from the caller's contract outward: the pinned
+// fingerprint first (the caller's approval is void whatever else holds),
+// then the table's existence, then the destructive guard, then the routed
+// dispositions. An empty (already-converged) plan never reaches admission:
+// the caller resolves it first, because it carries no plan identity for
+// the pin to verify and nothing would run anyway.
+func admitPlan(req DesiredRequest, report plan.Report) (DesiredResult, bool) {
+	refused := DesiredResult{Plan: report, Outcome: verdict.OutcomeRefused}
+	if req.ExpectedFingerprint != "" && req.ExpectedFingerprint != report.Fingerprint {
+		refused.Reason = verdict.ReasonPlanFingerprintMismatch
+		refused.Detail = fmt.Sprintf(
+			"the plan derived at execution time (fingerprint %s) is not the pinned plan (fingerprint %s); "+
+				"the live table or the desired schema changed since the plan was reviewed — re-review the new plan",
+			report.Fingerprint, req.ExpectedFingerprint)
+		return refused, false
+	}
+	if report.TableExists != nil && !*report.TableExists {
+		refused.Reason = verdict.ReasonUnsupportedStatement
+		refused.Detail = fmt.Sprintf(
+			"table %s.%s does not exist; desired-state execution converges an existing table — "+
+				"create the table from the plan's SQL script first",
+			report.Schema, report.Table)
+		return refused, false
+	}
+	for i, ps := range report.Statements {
+		if !ps.Destructive {
+			continue
+		}
+		refused.Reason = verdict.ReasonDestructiveChange
+		// The deliberate path differs by shape: the imperative front door
+		// runs an ALTER TABLE drop when the operator states it, but it
+		// refuses a plain DROP INDEX in favor of the concurrent idiom — so
+		// an index drop is pointed straight at that idiom instead of at a
+		// door that would bounce it.
+		deliberatePath := "run it deliberately through the imperative front door"
+		if ps.Kind == schemadiff.ChangeDropIndex {
+			deliberatePath = "drop it deliberately with DROP INDEX CONCURRENTLY, then rerun"
+		}
+		refused.Detail = fmt.Sprintf(
+			"planned statement %d discards live structure (%s); desired-state execution runs no "+
+				"destructive statement — %s",
+			i+1, ps.SQL, deliberatePath)
+		refused.Detail += skippedRestDetail(len(report.Statements) - 1)
+		return refused, false
+	}
+	if report.Disposition != router.DispositionExecute {
+		refused.Reason, refused.Detail = planRefusal(report)
+		return refused, false
+	}
+	return DesiredResult{}, true
+}
+
+// skippedRestDetail names what else a whole-plan destructive refusal
+// blocked, so the operator knows the size of what is stopped before
+// reading the plan: admission is all-or-nothing, and a desired file
+// usually carries several edits at once. Zero skipped statements say
+// nothing — there is nothing else in the plan to disclose.
+func skippedRestDetail(n int) string {
+	switch {
+	case n == 1:
+		return "; admission is all-or-nothing, so the plan's other statement, even if non-destructive, was not run"
+	case n > 1:
+		return fmt.Sprintf(
+			"; admission is all-or-nothing, so the plan's %d other statements, non-destructive ones included, were not run", n)
+	default:
+		return ""
+	}
+}
+
+// planRefusal maps the first non-executable planned statement to the typed
+// refusal the aggregate result carries, mirroring how [Run] refuses the
+// same dispositions at execution time.
+func planRefusal(report plan.Report) (verdict.Reason, string) {
+	for i, ps := range report.Statements {
+		detail := func(why string) string {
+			return fmt.Sprintf("planned statement %d (%s) %s; nothing was executed", i+1, ps.SQL, why)
+		}
+		switch ps.Disposition {
+		case router.DispositionExecute:
+			continue
+		case router.DispositionRewriteRequired:
+			return verdict.ReasonRewriteRequired, detail("blocks and has no safer native sequence")
+		case router.DispositionUnavailable:
+			return verdict.ReasonBackendUnavailable, detail("routes to an execution strategy this build does not implement")
+		case router.DispositionRefuse:
+			reason := ps.Reason
+			if reason == verdict.ReasonNone {
+				reason = verdict.ReasonUnsupportedStatement
+			}
+			return reason, detail("has no safe path")
+		default:
+			return verdict.ReasonUnsupportedStatement, detail("carries a disposition this build does not know")
+		}
+	}
+	// The aggregate disposition is non-executable but every statement is:
+	// a report this build cannot have produced. Refuse rather than guess.
+	return verdict.ReasonUnsupportedStatement,
+		fmt.Sprintf("the plan's aggregate disposition is %q but no statement carries it; nothing was executed",
+			report.Disposition)
+}
+
+// committedPrefixDetail states how far convergence got when execution
+// stopped at statement i (0-based) of n: the statements before it are
+// committed and stay committed. A stop on the first statement says plainly
+// that nothing committed before it — it does not claim the table is
+// untouched, because a failed statement's own committed steps are
+// disclosed by that statement's verdict, not here.
+func committedPrefixDetail(i, n int, what string) string {
+	if i == 0 {
+		return fmt.Sprintf("planned statement 1 of %d %s; nothing was committed before it", n, what)
+	}
+	return fmt.Sprintf("planned statement %d of %d %s; the %d preceding statements committed and remain in effect",
+		i+1, n, what, i)
+}
diff --git a/pkg/migrate/desired_integration_test.go b/pkg/migrate/desired_integration_test.go
new file mode 100644
index 0000000..65f78ad
--- /dev/null
+++ b/pkg/migrate/desired_integration_test.go
@@ -0,0 +1,244 @@
+package migrate_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/internal/testutil"
+	"github.com/block/pg-sprite/pkg/dbconn"
+	"github.com/block/pg-sprite/pkg/diffplan"
+	"github.com/block/pg-sprite/pkg/migrate"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/block/pg-sprite/pkg/verdict"
+)
+
+func parseDesired(t *testing.T, sql string) statement.DesiredSchema {
+	t.Helper()
+	ds, err := statement.ParseDesired(sql)
+	require.NoError(t, err)
+	return ds
+}
+
+// RunDesired is the declarative execution loop: these tests drive the full
+// plan-then-execute flow against a live database — convergence and its
+// no-op re-run, the plan-time admission refusals, the fingerprint pin, and
+// the committed-prefix shapes when execution stops partway.
+func TestRunDesired(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+
+	desiredSQL := `CREATE TABLE t (id int PRIMARY KEY, v text);
+CREATE INDEX t_v_idx ON t (v);`
+
+	t.Run("converges the live table and re-runs as a no-op", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+		require.NoError(t, err)
+
+		req := migrate.DesiredRequest{Schema: schema, Desired: parseDesired(t, desiredSQL)}
+		res, err := migrate.RunDesired(t.Context(), pool, req, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeExecuted, res.Outcome)
+		require.Len(t, res.Verdicts, len(res.Plan.Statements),
+			"every planned statement carries a verdict")
+		for _, v := range res.Verdicts {
+			assert.Equal(t, verdict.OutcomeExecuted, v.Outcome)
+		}
+
+		var typ string
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT data_type FROM information_schema.columns
+			 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'v'`, schema).Scan(&typ))
+		assert.Equal(t, "text", typ)
+		var indexValid bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT i.indisvalid FROM pg_index i
+			 WHERE i.indexrelid = ($1 || '.t_v_idx')::regclass`, schema).Scan(&indexValid))
+		assert.True(t, indexValid, "the index build must have completed and validated")
+
+		// The convergence oracle: a second run derives an empty plan and
+		// runs nothing.
+		res, err = migrate.RunDesired(t.Context(), pool, req, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeExecuted, res.Outcome)
+		assert.Empty(t, res.Plan.Statements, "the converged table plans no statements")
+		assert.Empty(t, res.Verdicts)
+	})
+
+	t.Run("refuses a greenfield plan and creates nothing", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+
+		res, err := migrate.RunDesired(t.Context(), pool,
+			migrate.DesiredRequest{Schema: schema, Desired: parseDesired(t, desiredSQL)}, runOptions())
+		require.NoError(t, err, "a plan-time refusal is a result, not an error")
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonUnsupportedStatement, res.Reason)
+		assert.Empty(t, res.Verdicts, "nothing was attempted")
+
+		var exists bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT EXISTS (SELECT 1 FROM information_schema.tables
+			 WHERE table_schema = $1 AND table_name = 't')`, schema).Scan(&exists))
+		assert.False(t, exists, "the refused plan must not create the table")
+	})
+
+	t.Run("refuses a destructive plan and drops nothing", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf(
+			"CREATE TABLE %s.t (id int PRIMARY KEY, v text, extra int)", schema))
+		require.NoError(t, err)
+
+		res, err := migrate.RunDesired(t.Context(), pool,
+			migrate.DesiredRequest{Schema: schema, Desired: parseDesired(t, desiredSQL)}, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.Empty(t, res.Verdicts, "a destructive plan refuses before any statement runs")
+
+		var exists bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT EXISTS (SELECT 1 FROM information_schema.columns
+			 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'extra')`, schema).Scan(&exists))
+		assert.True(t, exists, "the live column the desired schema lacks must survive")
+	})
+
+	t.Run("refuses a plan that drops NOT NULL and keeps the guarantee", func(t *testing.T) {
+		// Dropping NOT NULL discards the same guarantee as dropping the
+		// equivalent constraint: the destructive guard must stop the plan
+		// before anything runs, and the live guarantee must survive.
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf(
+			"CREATE TABLE %s.t (id int PRIMARY KEY, v text NOT NULL)", schema))
+		require.NoError(t, err)
+
+		res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{
+			Schema:  schema,
+			Desired: parseDesired(t, "CREATE TABLE t (id int PRIMARY KEY, v text)"),
+		}, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.Empty(t, res.Verdicts, "the guard refuses before any statement runs")
+
+		var notNull bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT is_nullable = 'NO' FROM information_schema.columns
+			 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'v'`, schema).Scan(&notNull))
+		assert.True(t, notNull, "the NOT NULL the desired schema dropped must survive")
+	})
+
+	t.Run("refuses a pinned fingerprint mismatch and runs nothing", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+		require.NoError(t, err)
+
+		res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{
+			Schema:              schema,
+			Desired:             parseDesired(t, desiredSQL),
+			ExpectedFingerprint: "not-the-reviewed-plan",
+		}, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonPlanFingerprintMismatch, res.Reason)
+		assert.Empty(t, res.Verdicts)
+
+		var exists bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT EXISTS (SELECT 1 FROM information_schema.columns
+			 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'v')`, schema).Scan(&exists))
+		assert.False(t, exists, "a fingerprint mismatch must execute nothing")
+	})
+
+	t.Run("executes under a matching pinned fingerprint", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+		require.NoError(t, err)
+
+		desired := parseDesired(t, desiredSQL)
+		reviewed, err := diffplan.Plan(t.Context(), pool, diffplan.Request{Schema: schema, Desired: desired})
+		require.NoError(t, err)
+
+		res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{
+			Schema:              schema,
+			Desired:             desired,
+			ExpectedFingerprint: reviewed.Fingerprint,
+		}, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeExecuted, res.Outcome)
+		assert.Equal(t, reviewed.Fingerprint, res.Plan.Fingerprint,
+			"the executed plan is the reviewed plan")
+
+		// A retry of the same pinned request after convergence is a
+		// no-op, not a fingerprint mismatch: the empty plan resolves
+		// before the pin is checked, so an idempotent re-run of an
+		// approved plan stays safe to issue.
+		res, err = migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{
+			Schema:              schema,
+			Desired:             desired,
+			ExpectedFingerprint: reviewed.Fingerprint,
+		}, runOptions())
+		require.NoError(t, err)
+		assert.Equal(t, verdict.OutcomeExecuted, res.Outcome,
+			"a pinned re-run of a converged table is a no-op, not a refusal")
+		assert.Empty(t, res.Plan.Statements, "the converged table plans no statements")
+		assert.Empty(t, res.Verdicts)
+	})
+
+	t.Run("stops at an execution-time refusal", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+		require.NoError(t, err)
+
+		// The plan admits at plan time (diffplan applies no size guard),
+		// but Run size-guards the blind ADD COLUMN attempt: with the
+		// threshold below one heap page the first statement refuses at
+		// execution time and everything after it is never attempted.
+		opts := runOptions()
+		opts.MaxTableSizeBytes = 1
+		res, err := migrate.RunDesired(t.Context(), pool,
+			migrate.DesiredRequest{Schema: schema, Desired: parseDesired(t, desiredSQL)}, opts)
+		require.NoError(t, err, "an execution-time refusal is a result, not an error")
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonTableTooLarge, res.Reason,
+			"the aggregate carries the refusing statement's reason")
+		require.NotEmpty(t, res.Verdicts)
+		last := res.Verdicts[len(res.Verdicts)-1]
+		assert.Equal(t, verdict.OutcomeRefused, last.Outcome)
+		assert.Less(t, len(res.Verdicts), len(res.Plan.Statements),
+			"the statements after the refusal were never attempted")
+
+		var exists bool
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT EXISTS (SELECT 1 FROM pg_indexes
+			 WHERE schemaname = $1 AND indexname = 't_v_idx')`, schema).Scan(&exists))
+		assert.False(t, exists, "the planned index after the refusal must not exist")
+	})
+
+	t.Run("returns the failed result together with the operational error", func(t *testing.T) {
+		schema := testutil.NewSchema(t, pool)
+		_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema))
+		require.NoError(t, err)
+		// The NULL row makes the substituted safer sequence's VALIDATE
+		// CONSTRAINT step fail after the scaffold CHECK ... NOT VALID
+		// committed.
+		_, err = pool.Exec(t.Context(), fmt.Sprintf("INSERT INTO %s.t VALUES (1, NULL)", schema))
+		require.NoError(t, err)
+
+		res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{
+			Schema:  schema,
+			Desired: parseDesired(t, "CREATE TABLE t (id int PRIMARY KEY, v text NOT NULL)"),
+		}, runOptions())
+		require.Error(t, err, "an execution failure is an operational error")
+		assert.Equal(t, verdict.OutcomeFailed, res.Outcome)
+		require.Len(t, res.Verdicts, 1)
+		assert.Equal(t, verdict.OutcomeFailed, res.Verdicts[0].Outcome,
+			"the failed statement's verdict is the error's machine-readable twin")
+		assert.NotEmpty(t, res.Verdicts[0].ExecutedSQL,
+			"the failed verdict discloses the committed prefix inside the statement")
+	})
+}
diff --git a/pkg/migrate/desired_test.go b/pkg/migrate/desired_test.go
new file mode 100644
index 0000000..cc63e57
--- /dev/null
+++ b/pkg/migrate/desired_test.go
@@ -0,0 +1,202 @@
+package migrate
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/plan"
+	"github.com/block/pg-sprite/pkg/router"
+	"github.com/block/pg-sprite/pkg/schemadiff"
+	"github.com/block/pg-sprite/pkg/verdict"
+)
+
+func TestRunDesiredRejectsUnrunnableOptions(t *testing.T) {
+	// Options validation happens before any database work, so no pool is
+	// needed: desired-state execution enforces the same "the zero value is
+	// not a runnable policy" contract Run does.
+	res, err := RunDesired(t.Context(), nil, DesiredRequest{}, Options{})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "MaxTableSizeBytes",
+		"the rejection must name the Options field")
+	assert.Equal(t, DesiredResult{}, res, "an error before a plan carries a zero result")
+}
+
+func TestRunDesiredRejectsForce(t *testing.T) {
+	opts := DefaultOptions()
+	opts.Force = "public.t"
+	res, err := RunDesired(t.Context(), nil, DesiredRequest{}, opts)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrForceNotSupported,
+		"the rejection is the typed sentinel an embedder can branch on")
+	assert.Contains(t, err.Error(), "imperative front door",
+		"the rejection points the caller at the front door where force applies")
+	assert.Equal(t, DesiredResult{}, res)
+}
+
+func TestAdmitPlan(t *testing.T) {
+	executable := func() plan.Report {
+		exists := true
+		return plan.Report{
+			Fingerprint: "fp-live",
+			Schema:      "app",
+			Table:       "t",
+			TableExists: &exists,
+			Disposition: router.DispositionExecute,
+			Statements: []plan.Statement{
+				{SQL: "ALTER TABLE app.t ADD COLUMN v text", Disposition: router.DispositionExecute},
+			},
+		}
+	}
+
+	t.Run("admits an executable plan", func(t *testing.T) {
+		_, ok := admitPlan(DesiredRequest{}, executable())
+		assert.True(t, ok)
+	})
+
+	t.Run("admits a matching pinned fingerprint", func(t *testing.T) {
+		_, ok := admitPlan(DesiredRequest{ExpectedFingerprint: "fp-live"}, executable())
+		assert.True(t, ok)
+	})
+
+	t.Run("refuses a pinned fingerprint mismatch before every other check", func(t *testing.T) {
+		// The plan is also greenfield and destructive: the mismatch must
+		// win, because the caller's approval is void whatever else holds.
+		report := executable()
+		exists := false
+		report.TableExists = &exists
+		report.Statements[0].Destructive = true
+		res, ok := admitPlan(DesiredRequest{ExpectedFingerprint: "fp-reviewed"}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.OutcomeRefused, res.Outcome)
+		assert.Equal(t, verdict.ReasonPlanFingerprintMismatch, res.Reason)
+		assert.Contains(t, res.Detail, "fp-reviewed")
+		assert.Contains(t, res.Detail, "fp-live")
+		assert.Equal(t, report, res.Plan, "a refusal carries the plan it refused")
+	})
+
+	t.Run("refuses a greenfield plan", func(t *testing.T) {
+		report := executable()
+		exists := false
+		report.TableExists = &exists
+		res, ok := admitPlan(DesiredRequest{}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.ReasonUnsupportedStatement, res.Reason)
+		assert.Contains(t, res.Detail, "app.t")
+		assert.Equal(t, report, res.Plan, "a refusal carries the plan it refused")
+	})
+
+	t.Run("refuses a destructive statement anywhere in the plan", func(t *testing.T) {
+		report := executable()
+		report.Statements = append(report.Statements, plan.Statement{
+			SQL:         "ALTER TABLE app.t DROP COLUMN old",
+			Destructive: true,
+			Disposition: router.DispositionExecute,
+		})
+		res, ok := admitPlan(DesiredRequest{}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.Contains(t, res.Detail, "statement 2")
+		assert.Contains(t, res.Detail, "DROP COLUMN old")
+		assert.Contains(t, res.Detail, "imperative front door",
+			"an ALTER TABLE drop is pointed at the door that runs it deliberately")
+		assert.Contains(t, res.Detail, "the plan's other statement, even if non-destructive, was not run",
+			"a multi-statement refusal discloses that the rest of the plan was skipped too")
+		assert.Equal(t, report, res.Plan, "a refusal carries the plan it refused")
+	})
+
+	t.Run("counts the skipped statements when more than one is blocked", func(t *testing.T) {
+		report := executable()
+		report.Statements = append(report.Statements,
+			plan.Statement{
+				SQL:         "ALTER TABLE app.t DROP COLUMN old",
+				Destructive: true,
+				Disposition: router.DispositionExecute,
+			},
+			plan.Statement{
+				SQL:         "CREATE INDEX t_v_idx ON app.t (v)",
+				Disposition: router.DispositionExecute,
+			})
+		res, ok := admitPlan(DesiredRequest{}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.Contains(t, res.Detail, "the plan's 2 other statements, non-destructive ones included, were not run",
+			"the disclosure counts every skipped statement, before and after the destructive one")
+	})
+
+	t.Run("a single-statement destructive refusal claims no skipped statements", func(t *testing.T) {
+		report := executable()
+		report.Statements = []plan.Statement{{
+			SQL:         "ALTER TABLE app.t DROP COLUMN old",
+			Destructive: true,
+			Disposition: router.DispositionExecute,
+		}}
+		res, ok := admitPlan(DesiredRequest{}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.NotContains(t, res.Detail, "all-or-nothing",
+			"there is nothing else in the plan to disclose as skipped")
+	})
+
+	t.Run("points a destructive index drop at its concurrent idiom", func(t *testing.T) {
+		// The imperative front door refuses a plain DROP INDEX, so the
+		// refusal must not send an index drop there — it names the
+		// concurrent idiom the operator can run directly.
+		report := executable()
+		report.Statements = append(report.Statements, plan.Statement{
+			SQL:         `DROP INDEX "app"."t_v_idx"`,
+			Kind:        schemadiff.ChangeDropIndex,
+			Destructive: true,
+			Disposition: router.DispositionExecute,
+		})
+		res, ok := admitPlan(DesiredRequest{}, report)
+		require.False(t, ok)
+		assert.Equal(t, verdict.ReasonDestructiveChange, res.Reason)
+		assert.Contains(t, res.Detail, "DROP INDEX CONCURRENTLY")
+		assert.NotContains(t, res.Detail, "imperative front door",
+			"the front door would refuse the drop; the detail must not point there")
+	})
+
+	t.Run("maps the first non-executable disposition to its refusal", func(t *testing.T) {
+		cases := []struct {
+			disposition router.Disposition
+			stReason    verdict.Reason
+			want        verdict.Reason
+		}{
+			{router.DispositionRewriteRequired, verdict.ReasonNone, verdict.ReasonRewriteRequired},
+			{router.DispositionUnavailable, verdict.ReasonNone, verdict.ReasonBackendUnavailable},
+			{router.DispositionRefuse, verdict.ReasonUnsupportedPartitionedParent, verdict.ReasonUnsupportedPartitionedParent},
+			{router.DispositionRefuse, verdict.ReasonNone, verdict.ReasonUnsupportedStatement},
+		}
+		for _, tc := range cases {
+			report := executable()
+			report.Disposition = tc.disposition
+			report.Statements = append(report.Statements, plan.Statement{
+				SQL:         "ALTER TABLE app.t ALTER COLUMN v TYPE bigint",
+				Disposition: tc.disposition,
+				Reason:      tc.stReason,
+			})
+			res, ok := admitPlan(DesiredRequest{}, report)
+			require.False(t, ok, "disposition %s", tc.disposition)
+			assert.Equal(t, tc.want, res.Reason, "disposition %s", tc.disposition)
+			assert.Contains(t, res.Detail, "statement 2", "the detail names the non-executable statement")
+			assert.Contains(t, res.Detail, "nothing was executed")
+			assert.Equal(t, report, res.Plan, "a refusal carries the plan it refused")
+		}
+	})
+}
+
+// committedPrefixDetail is the disclosure of how far convergence got; its
+// arithmetic must not drift — the stopping statement is 1-based, the
+// committed prefix count is the 0-based index. As a renderer helper its
+// exact wording is pinned here, in its own unit test.
+func TestCommittedPrefixDetail(t *testing.T) {
+	assert.Equal(t,
+		"planned statement 3 of 5 failed; the 2 preceding statements committed and remain in effect",
+		committedPrefixDetail(2, 5, "failed"))
+	assert.Equal(t,
+		"planned statement 1 of 2 stopped before a verdict; nothing about it was executed; "+
+			"nothing was committed before it",
+		committedPrefixDetail(0, 2, stoppedBeforeVerdict))
+}
diff --git a/pkg/migrate/example_test.go b/pkg/migrate/example_test.go
index 6a2c67b..4e3fa97 100644
--- a/pkg/migrate/example_test.go
+++ b/pkg/migrate/example_test.go
@@ -66,3 +66,54 @@ func Example_run() {
 		fmt.Println(v.Code, v.ExecutedSQL)
 	}
 }
+
+// Example_runDesired is the declarative execution flow: parse the
+// desired-state schema, connect, and converge the live table onto it — the
+// engine derives the plan and drives every planned statement through the
+// same pipeline Run uses. It is compile-checked but not executed —
+// RunDesired needs a live PostgreSQL database.
+func Example_runDesired() {
+	ctx := context.Background()
+
+	// One desired file describes one table: exactly one CREATE TABLE plus
+	// its indexes. Parse failures surface here, at the boundary where the
+	// embedder can render them.
+	desired, err := statement.ParseDesired(`CREATE TABLE users (id bigint PRIMARY KEY, email text);
+CREATE INDEX users_email_idx ON users (email);`)
+	if err != nil {
+		log.Print(err)
+		return
+	}
+
+	pool, err := dbconn.NewPool(ctx, dbconn.Config{URL: "postgres://engine@localhost:5432/app"})
+	if err != nil {
+		log.Print(err)
+		return
+	}
+	defer pool.Close()
+
+	// The result-and-error contract mirrors Run's three shapes: a refusal
+	// (at plan admission or on a mid-plan statement) returns the result
+	// with a nil error; an execution failure returns the failed result
+	// together with the operational error; an error with a zero result
+	// means nothing was planned or executed. Verdicts[i] is the verdict of
+	// Plan.Statements[i] — fewer verdicts than planned statements means
+	// execution stopped there.
+	res, err := migrate.RunDesired(ctx, pool, migrate.DesiredRequest{
+		Schema:  "public",
+		Desired: desired,
+		// ExpectedFingerprint pins a reviewed plan: leave it empty to run
+		// whatever plan the live table needs now.
+	}, migrate.DefaultOptions())
+	if err != nil {
+		log.Print(err)
+	}
+	switch res.Outcome {
+	case verdict.OutcomeExecuted:
+		fmt.Println(len(res.Plan.Statements), "statements converged")
+	case verdict.OutcomeRefused:
+		fmt.Println(res.Reason, res.Detail)
+	case verdict.OutcomeFailed:
+		fmt.Println(res.Detail)
+	}
+}
diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go
index fcaa4dc..bd49753 100644
--- a/pkg/migrate/migrate.go
+++ b/pkg/migrate/migrate.go
@@ -4,12 +4,21 @@
 // embedding pg-sprite share this one pipeline, so a verdict means the same
 // thing no matter which caller produced it.
 //
+// [RunDesired] is the declarative execution loop on the same pipeline: one
+// parsed desired-state schema in, the convergence plan derived through
+// diffplan.Plan, and every planned statement executed back through [Run] —
+// per-statement verdicts, committed-prefix semantics, stop at the first
+// refusal or failure. Both entry points share [Options], the executors,
+// and the verdict contract; the desired loop adds only plan admission and
+// sequencing.
+//
 // Callers own the boundary concerns: parse the statement through
-// [statement.ParseOne] (a parse failure surfaces at the caller) and build
-// the connection through [dbconn.NewPool]. [Gate] is exported so a caller
-// can refuse an unsupported statement kind before dialing; [Run] re-checks
-// it, so a caller that skips the early gate still cannot execute a gated
-// kind.
+// [statement.ParseOne] (or the desired schema through
+// [statement.ParseDesired]; a parse failure surfaces at the caller) and
+// build the connection through [dbconn.NewPool]. [Gate] is exported so a
+// caller can refuse an unsupported statement kind before dialing; [Run]
+// re-checks it, so a caller that skips the early gate still cannot execute
+// a gated kind.
 //
 // [Run] takes the concrete [pgxpool.Pool] that [dbconn.NewPool] returns —
 // a deliberate concrete dependency, not an oversight: the execution paths
diff --git a/pkg/plan/plan.go b/pkg/plan/plan.go
index 4ec1690..148237d 100644
--- a/pkg/plan/plan.go
+++ b/pkg/plan/plan.go
@@ -61,7 +61,7 @@ type Statement struct {
 	// has no single kind.
 	Kind schemadiff.ChangeKind `json:"kind,omitempty"`
 	// Destructive marks statements that discard live structure — a dropped
-	// column, constraint, or index. It is derived from the classifier's
+	// column, constraint, index, or NOT NULL. It is derived from the classifier's
 	// decisions, so both sources report it identically; it is always
 	// emitted, never omitted, because a safety flag a consumer gates on
 	// must be explicit even when false.
diff --git a/pkg/planner/planner.go b/pkg/planner/planner.go
index b87cd81..446c81e 100644
--- a/pkg/planner/planner.go
+++ b/pkg/planner/planner.go
@@ -178,7 +178,7 @@ type Decision struct {
 	// Operation is the operator-facing label (display only).
 	Operation string `json:"operation"`
 	// Destructive marks operations that discard live structure — a dropped
-	// column, constraint, or index. It is derived from the operation shape
+	// column, constraint, index, or NOT NULL. It is derived from the operation shape
 	// here, in the one place every front door shares, so a plan reports the
 	// same statement as destructive no matter how it was submitted. It is
 	// always emitted, never omitted: a safety flag a consumer gates on must
@@ -384,11 +384,16 @@ func classifyOp(op statement.Op, st statement.Statement, facts Facts, sql string
 // destructiveOp reports whether an operation shape discards live
 // structure. A drop is destructive regardless of how it routes: a dropped
 // column discards data, a dropped constraint discards a guarantee the
-// schema was providing, and a dropped index discards a structure that is
-// expensive to rebuild (and, for a unique index, the uniqueness guarantee).
+// schema was providing — and dropping NOT NULL discards exactly the same
+// guarantee its constraint form would, so it is marked identically — and a
+// dropped index discards a structure that is expensive to rebuild (and,
+// for a unique index, the uniqueness guarantee). Dropping a DEFAULT is
+// deliberately not destructive: a default guarantees nothing about
+// existing rows and is recreated by a metadata-only statement.
 func destructiveOp(kind statement.OpKind) bool {
 	switch kind {
-	case statement.OpDropColumn, statement.OpDropConstraint, statement.OpDropIndex:
+	case statement.OpDropColumn, statement.OpDropConstraint, statement.OpDropIndex,
+		statement.OpDropNotNull:
 		return true
 	default:
 		return false
diff --git a/pkg/planner/planner_test.go b/pkg/planner/planner_test.go
index 5e45cfc..a964241 100644
--- a/pkg/planner/planner_test.go
+++ b/pkg/planner/planner_test.go
@@ -47,16 +47,20 @@ func classifyOne(t *testing.T, sql string) planner.Decision {
 }
 
 // Destructive is a decision-level fact derived from the operation shape:
-// drops of columns, constraints, and indexes discard live structure, and
-// every front door that routes through the classifier inherits the same
-// marking — including DROP INDEX, whose drop discards the index's
-// guarantee (uniqueness, for a unique index) however it is submitted.
+// drops of columns, constraints, indexes, and NOT NULL discard live
+// structure or a guarantee, and every front door that routes through the
+// classifier inherits the same marking — including DROP INDEX, whose drop
+// discards the index's guarantee (uniqueness, for a unique index) however
+// it is submitted. DROP DEFAULT stays non-destructive: a default
+// guarantees nothing about existing rows and is recreated by a
+// metadata-only statement.
 func TestClassifyMarksDropsDestructive(t *testing.T) {
 	destructive := []string{
 		"ALTER TABLE t DROP COLUMN age",
 		"ALTER TABLE t DROP CONSTRAINT t_age_check",
 		"DROP INDEX t_v_idx",
 		"DROP INDEX CONCURRENTLY t_v_idx",
+		"ALTER TABLE t ALTER COLUMN age DROP NOT NULL",
 	}
 	for _, sql := range destructive {
 		assert.True(t, classifyOne(t, sql).Destructive, sql)
@@ -65,7 +69,6 @@ func TestClassifyMarksDropsDestructive(t *testing.T) {
 		"ALTER TABLE t ADD COLUMN age int",
 		"ALTER TABLE t ALTER COLUMN v50 TYPE varchar(100)",
 		"ALTER TABLE t ALTER COLUMN age DROP DEFAULT",
-		"ALTER TABLE t ALTER COLUMN age DROP NOT NULL",
 		"CREATE INDEX t_v_idx ON t (v50)",
 	}
 	for _, sql := range nonDestructive {
diff --git a/pkg/verdict/docs_test.go b/pkg/verdict/docs_test.go
new file mode 100644
index 0000000..77c7b39
--- /dev/null
+++ b/pkg/verdict/docs_test.go
@@ -0,0 +1,27 @@
+package verdict
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// cliOutputExamplesDoc is the human-facing page that documents every
+// refusal-reason token; this test keeps it honest the same way
+// pkg/plan/docs_test.go keeps docs/plan-report.md honest.
+const cliOutputExamplesDoc = "../../docs/cli-output-examples.md"
+
+// Every refusal reason automation can meet must be documented: a Reason
+// constant added without a row in the doc's refusal-reason table fails here.
+func TestDocListsEveryRefusalReason(t *testing.T) {
+	raw, err := os.ReadFile(cliOutputExamplesDoc)
+	require.NoError(t, err)
+	doc := string(raw)
+	for _, r := range Reasons() {
+		assert.Contains(t, doc, fmt.Sprintf("| `%s` |", string(r)),
+			"docs/cli-output-examples.md is missing a refusal-reason row for %q", r)
+	}
+}
diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go
index 3d49761..a63a144 100644
--- a/pkg/verdict/verdict.go
+++ b/pkg/verdict/verdict.go
@@ -77,8 +77,37 @@ const (
 	// ReasonBackendUnavailable: the change routes to an execution strategy
 	// this build does not implement (copy-and-swap).
 	ReasonBackendUnavailable Reason = "backend-unavailable"
+	// ReasonDestructiveChange: the desired-state plan discards live
+	// structure — a dropped column, constraint, index, or NOT NULL — and
+	// desired-state execution runs no destructive statement without an
+	// explicit path for it. The imperative front door remains the way to
+	// run a reviewed destructive statement deliberately.
+	ReasonDestructiveChange Reason = "destructive-change"
+	// ReasonPlanFingerprintMismatch: the plan recomputed at execution time
+	// does not carry the fingerprint the caller pinned, so the plan a
+	// reviewer approved is not the plan that would execute; nothing runs.
+	ReasonPlanFingerprintMismatch Reason = "plan-fingerprint-mismatch"
 )
 
+// Reasons returns the closed set of non-zero Reason values. It is part of
+// the verdict contract: the tokens are what automation switches on, so the
+// set changes only deliberately, every token is pinned by test, and every
+// token has a row in docs/cli-output-examples.md's refusal-reason table.
+func Reasons() []Reason {
+	return []Reason{
+		ReasonUnsupportedStatement,
+		ReasonIndexStatement,
+		ReasonTableTooLarge,
+		ReasonInsufficientPrivileges,
+		ReasonUnsupportedPartitionedParent,
+		ReasonBudgetExceeded,
+		ReasonRewriteRequired,
+		ReasonBackendUnavailable,
+		ReasonDestructiveChange,
+		ReasonPlanFingerprintMismatch,
+	}
+}
+
 // Cause narrows ReasonBudgetExceeded to the budget that was exceeded, so
 // automation can branch on which limit fired without parsing prose.
 type Cause string
diff --git a/pkg/verdict/verdict_test.go b/pkg/verdict/verdict_test.go
index 76d8f0b..81eefac 100644
--- a/pkg/verdict/verdict_test.go
+++ b/pkg/verdict/verdict_test.go
@@ -78,20 +78,37 @@ func TestJSONOmitsEmptyOptionalFields(t *testing.T) {
 // Reason and Cause values are the machine contract automation switches on:
 // flat kebab-case tokens, no spaces or colons — prose belongs in Detail.
 func TestReasonAndCauseTokensAreFlat(t *testing.T) {
-	for _, tok := range []string{
-		string(ReasonUnsupportedStatement),
-		string(ReasonIndexStatement),
-		string(ReasonTableTooLarge),
-		string(ReasonBudgetExceeded),
-		string(ReasonInsufficientPrivileges),
-		string(ReasonUnsupportedPartitionedParent),
-		string(CauseLockBudget),
-		string(CauseStatementBudget),
-	} {
+	toks := []string{string(CauseLockBudget), string(CauseStatementBudget)}
+	for _, r := range Reasons() {
+		toks = append(toks, string(r))
+	}
+	for _, tok := range toks {
 		assert.Regexp(t, `^[a-z0-9]+(-[a-z0-9]+)*$`, tok)
 	}
 }
 
+// The wire tokens themselves are the contract, not just their shape: a
+// renamed token ships a breaking change to every consumer switching on it,
+// so the exact strings are pinned here.
+func TestReasonsPinsWireTokens(t *testing.T) {
+	var got []string
+	for _, r := range Reasons() {
+		got = append(got, string(r))
+	}
+	assert.Equal(t, []string{
+		"unsupported-statement",
+		"index-statement",
+		"not-native-safe-table-too-large",
+		"insufficient-privileges",
+		"unsupported-partitioned-parent",
+		"not-native-safe-budget-exceeded",
+		"not-native-safe-rewrite-required",
+		"backend-unavailable",
+		"destructive-change",
+		"plan-fingerprint-mismatch",
+	}, got)
+}
+
 func TestStringExecuted(t *testing.T) {
 	s := Verdict{
 		Outcome:   OutcomeExecuted,