diff --git a/.agents/checks/review.md b/.agents/checks/review.md
index 4635d39..3968ce3 100644
--- a/.agents/checks/review.md
+++ b/.agents/checks/review.md
@@ -4,6 +4,12 @@ Checks for AI review agents. The authoritative rules live in [SAFETY.md](../../S
 [AGENTS.md](../../AGENTS.md), and [docs/tcb-model.md](../../docs/tcb-model.md) — this file is
 the reviewer's distillation.
 
+- Judge the change through both project lenses: (a) OSS-first — pg-sprite as the preferred
+  standalone PostgreSQL online-DDL tool (CLI usable without an orchestrator, external-user
+  docs and errors, no Block-internal assumptions); (b) clean SchemaBot integration — a stable
+  adapter-friendly seam (library API, verdict/plan JSON, error taxonomy) with the core never
+  depending on SchemaBot. Flag changes that serve one lens at the other's expense without a
+  recorded decision.
 - Look up every touched `pkg/` package in the SAFETY.md partition table first — the review bar
   differs between the safety-critical core and the periphery. Flag core changes with 🌶️ and
   state the blast radius (data corruption, lost writes, wrong-table swap, stranded slot).
@@ -21,8 +27,11 @@ the reviewer's distillation.
   module — ideas are ported with citations, not code.
 - Connections go through `pkg/dbconn` (bounded `lock_timeout` / `statement_timeout`) — flag
   raw `pgx` pools in production code.
-- SQL parsing goes through `pg_query_go`; flag `strings.Split(";")` or any hand-parsing. A
-  parse failure is an error surfaced to the caller.
+- SQL parsing goes through `wasilibs/go-pgquery` (Wasm `libpg_query`); flag
+  `strings.Split(";")`, any hand-parsing, and imports of the cgo `pg_query_go` (documented
+  escape hatch, not the default). A parse failure is an error surfaced to the caller.
+  Shadow-table DDL and checkpoint fingerprints come from execute-and-introspect on the scratch
+  database — flag AST surgery that constructs the shadow schema or fingerprints SQL text.
 - Generated SQL quotes every user-supplied or introspected identifier
   (`pgx.Identifier{...}.Sanitize()` / `quote_ident()`) — flag raw interpolation of names into
   SQL. Connection strings are parsed and re-serialized (`pgx.ParseConfig`), never
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0bbc1c7..fd6c64a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -76,6 +76,14 @@ jobs:
         with:
           go-version-file: go.mod
       - run: make build
+      # The no-cgo promise is a contract: the Wasm parser keeps `go install`
+      # toolchain-free. One accidental import of the cgo escape hatch
+      # (pg_query_go's parser) would silently start requiring a C toolchain
+      # on every contributor's machine.
+      - name: Build with CGO disabled
+        run: go build ./...
+        env:
+          CGO_ENABLED: "0"
 
   # The integration suite runs against every Aurora-supported PostgreSQL
   # major (see docs/postgresql-version-support.md): the version floor is a
diff --git a/.golangci.yml b/.golangci.yml
index dc76b27..026d2ff 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -20,7 +20,16 @@ linters:
     - gochecknoinits  # no init() functions
     - gochecknoglobals # no package-level mutable state (error sentinels exempt)
     - containedctx    # no context.Context stored in struct fields
+    - sloglint        # structured-logging hygiene: static messages, snake_case keys
+    - forbidigo       # no printing to process stdout; output goes to the injected writer
   settings:
+    sloglint:
+      static-msg: true
+      key-naming-case: snake
+    forbidigo:
+      forbid:
+        - pattern: ^(fmt\.Print(f|ln)?|print|println)$
+          msg: command output goes to the injected io.Writer and diagnostics to slog, never process stdout
     usetesting:
       context-background: true
       context-todo: true
@@ -32,6 +41,13 @@ linters:
         - name: package-comments
   exclusions:
     rules:
+      # The tracelog adapter forwards pgx's own message strings; the
+      # static-message rule applies to messages we author, not to a bridge
+      # for a foreign logger.
+      - path: pkg/dbconn/dbconn\.go
+        linters:
+          - sloglint
+        text: message should be a string literal
       # Test helpers intentionally keep uniform signatures and fixed arguments
       # for readability, so unused/constant params there are not worth churn.
       - path: _test\.go
diff --git a/AGENTS.md b/AGENTS.md
index 9bb974d..7aab437 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -7,6 +7,18 @@ This file is canonical. `CLAUDE.md`, `GEMINI.md`, `.cursorrules`, `.goosehints`,
 `.github/copilot-instructions.md` are symlinks to it — edit only this file. Review-agent
 checks live in [.agents/checks/review.md](.agents/checks/review.md).
 
+## Two lenses on every change
+
+Judge every PR, design, and review through both lenses — a change that serves one at the
+expense of the other needs an explicit decision, not a silent trade:
+
+1. **OSS-first.** pg-sprite aims to be the preferred PostgreSQL online-DDL tool in its own
+   right: the CLI works standalone with no orchestrator setup, docs and error messages are
+   written for external users, and nothing assumes a Block-internal environment.
+2. **Clean SchemaBot integration.** pg-sprite must slot into SchemaBot as an engine behind a
+   stable seam: keep the library API, verdict/plan JSON contracts, and error taxonomy
+   adapter-friendly, and never let the core depend on SchemaBot (or any orchestrator).
+
 ## Read SAFETY.md first
 
 This codebase is partitioned into a **safety-critical core** and a periphery.
@@ -51,8 +63,12 @@ make lint        # golangci-lint
   generated SQL goes through `pgx.Identifier{...}.Sanitize()` (or `quote_ident()` server-side).
 - Never string-manipulate connection strings/DSNs — parse (`pgx.ParseConfig`), modify fields,
   re-serialize; string ops break on passwords containing `/`, `@`, or `%`.
-- All SQL parsing goes through `pg_query_go` (once `pkg/statement` exists). No
-  `strings.Split(";")`, no hand-parsing; a parse failure is an error surfaced to the caller.
+- All SQL parsing goes through the real PostgreSQL grammar via `wasilibs/go-pgquery` (Wasm
+  `libpg_query`; the cgo `pg_query_go` is the API-compatible escape hatch, not the default),
+  with `pkg/statement` as the parse boundary. No `strings.Split(";")`, no hand-parsing; a parse failure is an
+  error surfaced to the caller. Shadow-table DDL and checkpoint fingerprints are derived by
+  execute-and-introspect on the engine-owned scratch database, never by AST transformation
+  (see [docs/low-level-design.md](docs/low-level-design.md#how-the-planner-understands-ddl-decided)).
 - Tests use testify (`require` for setup, `assert` for verification), `t.Context()` (in
   cleanups, which run after the context is cancelled, use
   `context.WithoutCancel(t.Context())`), and named polling deadlines — no bare `time.Sleep`
@@ -104,8 +120,36 @@ make lint        # golangci-lint
 - Never reference internal company details (cluster names, hostnames, org names) in code,
   comments, commits, or PRs — this is a public repo.
 
+## Logging and observability
+
+- **stdout is the product's output; diagnostics go to stderr.** Command results (verdicts,
+  status) are written to the injected writer only; everything diagnostic goes through
+  `log/slog`. `--debug` on DB commands enables statement-level tracing (pgx tracelog via
+  `pkg/dbconn`) plus lifecycle events; without it, diagnostics are discarded.
+- **Log decisions and state transitions, not progress noise.** Static messages; the
+  variability goes into attrs with stable snake_case keys, and the same key means the same
+  thing everywhere (`schema`, `table`, `total_bytes`, `elapsed`).
+- **Logs answer the triage question.** Error- and warn-path logs carry the identifiers an
+  operator needs to act — schema, table, database, the operation being attempted — as
+  attrs, not buried in prose.
+- **One error, one log.** Errors are wrapped and returned; only the entry point logs or
+  prints them. `pkg/` packages never log an error they also return.
+- **Never log credentials or connection strings** — a DSN/URL carries a password; log host,
+  database, and user as separate attrs when needed. Never log row data.
+- **Log output is never a test surface.** Tests assert typed outcomes — `errors.Is`/`As`,
+  verdict fields, exit codes, JSON output — never log text or human-facing wording. If a
+  behavioral difference is visible only in prose, make it machine-readable first (a typed
+  field or reason), then test that. The only exception is a renderer's own unit test.
+- **Operational quantities ride on logs until there is a metrics runtime.** Durations,
+  sizes, and retry counts are logged as attrs. When the long-running phases need real
+  metrics, they arrive as OpenTelemetry instruments behind one engine-owned `pkg/metrics`
+  with `Record*` helpers — dotted `pgsprite.` names with explicit units, low-cardinality
+  snake_case attributes, counters for rare or dangerous branches operators can act on —
+  never direct exporter imports in core (the dependency rule in SAFETY.md applies).
+
 Mechanical style rules (doc comments on exported symbols, no `init()`, no package-level
-mutable state, no `context.Context` in structs) are enforced by `.golangci.yml`, not prose.
+mutable state, no `context.Context` in structs, static slog messages with snake_case keys,
+no printing to process stdout) are enforced by `.golangci.yml`, not prose.
 
 ## Git and PRs
 
diff --git a/README.md b/README.md
index e1c9562..b17ccf7 100644
--- a/README.md
+++ b/README.md
@@ -17,9 +17,14 @@ when one exists (`CONCURRENTLY`, `NOT VALID` + `VALIDATE`, fast default,
 `USING INDEX`), and a log-based, checksum-gated, resumable copy-and-swap when
 a genuine table rewrite is unavoidable.
 
-**Status: Phase 0 (scaffold + test harness).** All subcommands are stubs. The
-design docs and the phased build plan live in [docs/](docs/) — start with
-[docs/README.md](docs/README.md).
+**Status: Phase 1 (optimistic front door).** `pg-sprite migrate --alter '…'`
+runs easy `ALTER TABLE` changes directly under tight lock/statement budgets
+and refuses everything else with a structured verdict (exit code 2): index
+maintenance gets a pointer to the `CONCURRENTLY` idiom, and changes that need
+a table rewrite — caught by the size guard or a cancelled bounded attempt —
+get an explicit **not native-safe** verdict. `diff`, `fmt`, and `lint` are
+still stubs. The design docs and the phased build plan live in
+[docs/](docs/) — start with [docs/README.md](docs/README.md).
 
 The codebase is partitioned into a small safety-critical core and a
 periphery — **[SAFETY.md](SAFETY.md)** says which packages are which and the
diff --git a/SAFETY.md b/SAFETY.md
index 1bd04f8..b78a1fb 100644
--- a/SAFETY.md
+++ b/SAFETY.md
@@ -18,23 +18,25 @@ The invariant registry (invariant IDs referenced below) lives in
 | Package | Core? | Status | Invariants enforced |
 | --- | --- | --- | --- |
 | `pkg/dbconn` — pool defaults, advisory lock, terminate-blockers, retries, RDS TLS | ✅ core | exists (Phase 0) | LK-1, LK-2 primitives |
-| `pkg/preflight` — precondition verifier, refusals | ✅ core | planned (Phase 1–2) | ST-6, RF-1..RF-5 |
+| `pkg/preflight` — precondition verifier, refusals | ✅ core | exists (Phase 1: table-size guard); grows through Phase 2 | ST-6, RF-1..RF-5 |
+| `pkg/executor` — bounded optimistic attempt; native executor later | ✅ core | exists (Phase 1: attempt-under-budget); Executor contract at Phase 2–3 | LK-2 (attempt bound) |
 | `pkg/checksum` — chunk verifier, continuous checker, repair | ✅ core | planned (Phase 5) | CO-1, CO-2, CO-3 |
 | `pkg/copier` — shadow-table chunked copy | ✅ core | planned (Phase 4) | CO-4, LK-3 |
 | `pkg/applier` — change apply, buffer, flush scheduling | ✅ core | planned (Phase 6) | CO-4, CO-5, CO-6, LK-3 |
 | `pkg/decode` — logical decoding, LSN/position accounting | ✅ core | planned (Phase 6) | ST-4, CO-4 |
 | `pkg/checkpoint` — durable resume state | ✅ core | planned (Phase 8) | ST-1, ST-2 |
 | slot lifecycle (in `pkg/decode`) — create, reap, lag ceiling | ✅ core | planned (Phase 8) | ST-3 |
-| `pkg/migration` — orchestrator, **cutover swap + fidelity gate** | ✅ core | planned (Phase 7) | LK-2, LK-4, ST-5 |
-| `pkg/statement`, `pkg/planner`, `pkg/schemadiff`, `pkg/lint` — classify/diff/route | ❌ periphery¹ | planned (Phase 1–2) | (CO-7 holds at the parse boundary) |
-| `internal/cli` — CLI, flags, help, prompts | ❌ periphery | exists (stubs) | — |
+| `pkg/schemachange` — orchestrator, **cutover swap + fidelity gate** | ✅ core | planned (Phase 7) | LK-2, LK-4, ST-5 |
+| `pkg/statement`, `pkg/planner`, `pkg/schemadiff`, `pkg/lint` — classify/diff/route | ❌ periphery¹ | `pkg/statement` exists (Phase 1: type gate); rest planned (Phase 2) | (CO-7 holds at the parse boundary) |
+| `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
+| `internal/cli` — CLI, flags, help, prompts | ❌ periphery | `migrate`/`status` exist (Phase 1); rest stubs | — |
 | status / progress / advisory rendering, metrics | ❌ periphery | planned | — |
 | orchestrator adapter | ❌ periphery | planned (Phase 11) | OC-* hold *at* the boundary |
 | `internal/testutil` | ❌ test-only | exists | — |
 
 ¹ **The planner is deliberately outside the core.** Its verdicts are *requests*, not
 permissions: a wrong "native-safe" verdict is capped by the executor's own `lock_timeout` bound;
-a wrong "copy" verdict produces a wasteful but *correct* migration (the checksum still gates).
+a wrong "copy" verdict produces a wasteful but *correct* schema change (the checksum still gates).
 The core executors re-verify their own preconditions and never trust that the planner checked.
 
 ## Rules inside the core
diff --git a/cmd/pg-sprite/main.go b/cmd/pg-sprite/main.go
index 0f1ad85..1c8ed7d 100644
--- a/cmd/pg-sprite/main.go
+++ b/cmd/pg-sprite/main.go
@@ -2,9 +2,13 @@
 package main
 
 import (
+	"errors"
+	"os"
+
 	"github.com/alecthomas/kong"
 
 	"github.com/block/pg-sprite/internal/cli"
+	"github.com/block/pg-sprite/pkg/verdict"
 )
 
 // version is stamped at release time via -ldflags "-X main.version=…".
@@ -17,5 +21,11 @@ func main() {
 		kong.UsageOnError(),
 		kong.Vars{"version": version},
 	)
-	k.FatalIfErrorf(k.Run())
+	err := k.Run()
+	// A refusal verdict was already printed; its exit code is distinct from
+	// operational errors so automation can branch on the difference.
+	if errors.Is(err, verdict.ErrRefused) {
+		os.Exit(verdict.ExitCodeRefused)
+	}
+	k.FatalIfErrorf(err)
 }
diff --git a/docs/architecture.md b/docs/architecture.md
index 672bfa4..ff3b784 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -48,20 +48,21 @@ boundary) is defined in [../SAFETY.md](../SAFETY.md).
 
 | Package | Role | Status |
 | --- | --- | --- |
-| `cmd/pg-sprite` | CLI entry point (Kong): `migrate` · `diff` · `fmt` · `lint` · `status` | exists (stubs) |
-| `internal/cli` | Command tree and flag handling | exists (stubs) |
+| `cmd/pg-sprite` | CLI entry point (Kong): `migrate` · `diff` · `fmt` · `lint` · `status` | `migrate`/`status` exist; rest stubs |
+| `internal/cli` | Command tree and flag handling | `migrate`/`status` exist; rest stubs |
 | `internal/testutil` | Test harness: containerized PostgreSQL, throwaway schemas | exists |
 | `pkg/dbconn` | Pool with bounded session timeouts, retries, RDS/Aurora auto-TLS (embedded CA bundle), terminate-blockers; advisory-lock mutual exclusion lands here | exists |
-| `pkg/statement` | `pg_query_go` parsing + classification (never hand-parse SQL) | Phase 1–2 |
-| `pkg/preflight` | Precondition verification and refusals before any write | Phase 1–2 |
+| `pkg/statement` | `go-pgquery` (Wasm `libpg_query`) parsing + classification (never hand-parse SQL); shadow DDL + fingerprints come from scratch-DB execute-and-introspect | exists (Phase 1: type gate); classification at Phase 2 |
+| `pkg/preflight` | Precondition verification and refusals before any write | exists (Phase 1: table-size guard); grows through Phase 2 |
+| `pkg/verdict` | Structured outcome contract (executed / refused + reason + safer idiom), rendering, exit codes | exists (Phase 1) |
 | `pkg/planner` / `pkg/schemadiff` / `pkg/lint` | Shared front-end: introspect, declarative diff (may wrap [stripe/pg-schema-diff](https://github.com/stripe/pg-schema-diff) — see the low-level design's open decisions), classify, lint | Phase 2 |
-| `pkg/executor` | The `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) + native executor | Phase 2–3 |
+| `pkg/executor` | The `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) + native executor | exists (Phase 1: bounded optimistic attempt); contract at Phase 2–3 |
 | `pkg/table` | PK-range chunkers (single-column fast path, composite), dynamic time-based sizing | Phase 4 |
 | `pkg/copier` | Parallel chunked copy into the shadow table (never overwrites) | Phase 4 |
 | `pkg/checksum` | The mandatory correctness gate; continuous checker; repair primitive | Phase 5 |
 | `pkg/decode` | Logical-decoding change capture, LSN accounting, slot lifecycle | Phase 6, 8 |
 | `pkg/applier` | Change apply onto the shadow (always wins), buffer/dedup, flush scheduling | Phase 6 |
-| `pkg/migration` | Orchestrator: lifecycle, cutover swap + fidelity gate, checkpoint/resume | Phase 7–8 |
+| `pkg/schemachange` | Orchestrator: lifecycle, cutover swap + fidelity gate, checkpoint/resume | Phase 7–8 |
 | `pkg/checkpoint` | Durable single-row resume state | Phase 8 |
 | `pkg/throttler` | Aurora reader-lag / slot-lag / WAL throttling | Phase 8 |
 
diff --git a/docs/invariants.md b/docs/invariants.md
index 75d4154..955a912 100644
--- a/docs/invariants.md
+++ b/docs/invariants.md
@@ -104,8 +104,12 @@ is incomplete without this.
 
 ### CO-7 — Every statement parses, or it is an error
 
-All SQL the engine processes must parse with `pg_query_go`. No `strings.Split(";")` fallback, no
-silently skipping unparseable statements — a parse failure is surfaced to the caller as an error.
+All SQL the engine processes must parse with the real PostgreSQL grammar — `wasilibs/go-pgquery`,
+the Wasm build of `libpg_query` (the cgo `pg_query_go` is the API-compatible escape hatch). No
+`strings.Split(";")` fallback, no silently skipping unparseable statements — a parse failure is
+surfaced to the caller as an error. The invariant pins the *capability* (classified or refused);
+the parser choice is an implementation decision of the understanding layer (see
+[low-level-design](low-level-design.md#how-the-planner-understands-ddl-decided)).
 *Enforced:* `pkg/statement` boundary. *Source:* SchemaBot AGENTS.md (TiDB-parser hard
 requirement, rewritten for our parser); carried in the repo's [AGENTS.md](../AGENTS.md).
 
@@ -179,7 +183,9 @@ Resume must tell apart: (a) a readable, matching checkpoint → resume; (b) a ch
 an incompatible engine version or for a **different statement** → refuse to resume, start fresh
 (never mix state across versions/statements); (c) a *transient* read failure → retry, and never
 trigger fresh-start recovery on a blip. *Enforced:* checkpoint read/validation path (version +
-statement fingerprint stored with the watermark). *Source:* Spirit `checkpoint.IsIncompatible` +
+statement fingerprint stored with the watermark; the fingerprint hashes the scratch-introspected
+after-schema model, not SQL text, so textually-different-but-identical statements match and
+cosmetic edits don't force a fresh start). *Source:* Spirit `checkpoint.IsIncompatible` +
 "resume requires the identical ALTER".
 
 ### ST-3 — Slot cleanup is guaranteed on success, failure, and crash
@@ -212,10 +218,25 @@ risks-and-mitigations.
 
 Every knowable prerequisite is validated before the engine writes anything: logical-replication
 enablement and role, PK usability, `REPLICA IDENTITY`, slot/WAL-sender headroom, disk headroom
-(~2× the table), lock LK-1 acquired, and the RF-* refusals below. Failing hours into a copy on
-something knowable up front is a bug. *Enforced:* preflight stage. *Source:*
+(~2× the table), the [scratch database](low-level-design.md#plan-time-prerequisite-the-scratch-database)
+(pre-provisioned `pg_sprite_scratch`, or `CREATEDB` so preflight can self-provision it), lock
+LK-1 acquired, and the RF-* refusals below. Failing hours into a copy on something knowable up
+front is a bug. **Sub-obligation — server-authoritative validation:** every statement is
+validated by a PostgreSQL server (executed in a rolled-back transaction on the scratch database)
+before the first write to the target; the server is the semantic authority and client-side
+parsing is advisory. *Enforced:* preflight stage. *Source:*
 [design-principles](design-principles.md#correctness-and-safety).
 
+### ST-7 — The executor runs exactly the statement that was gated
+
+The executor accepts only a parsed `statement.Statement` — constructible solely by `ParseOne`,
+which enforces exactly one statement through the real grammar — and refuses, before anything
+executes, any statement whose target table does not match the preflight proof it was handed.
+A proof for one table can never smuggle SQL against another, and a multi-statement string can
+never reach the database through the executor (pgx's simple protocol would happily run all of
+it). *Enforced:* `pkg/executor` (`AttemptNative`), `pkg/statement` (proof construction).
+*Source:* adversarial review of the optimistic front door.
+
 ## Refusals and preflight (RF)
 
 Each refusal is a preflight **error with a stated reason** — never a warning, never attempted.
@@ -307,4 +328,5 @@ about **how we write and review the code**.
 | LK-4, ST-5 | 7 | dropped-connection cutover, fidelity checklist |
 | ST-1, ST-2, ST-3, ST-4 | 8 | kill/resume, cross-version refuse, orphan-slot reap, failover reconcile |
 | ST-6 | 1 onward, complete by 8 | preflight matrix |
+| ST-7 | 1 | target-mismatch refusal + single-statement-by-construction tests |
 | OC-1..OC-6 | shape APIs from 2; bind at 11 | engine-contract tests |
diff --git a/docs/low-level-design.md b/docs/low-level-design.md
index 26d15ed..c89c47d 100644
--- a/docs/low-level-design.md
+++ b/docs/low-level-design.md
@@ -21,6 +21,7 @@ for how the Spirit original works and tool-pgroll.md for pgroll.
 - [Architecture: decoupled planner, router, and executors](#architecture-decoupled-planner-router-and-executors)
   - [Proposed architecture (end-to-end)](#proposed-architecture-end-to-end)
   - [Routing view (which executor handles what)](#routing-view-which-executor-handles-what)
+  - [How the planner understands DDL (decided)](#how-the-planner-understands-ddl-decided)
   - [Why this is the right shape](#why-this-is-the-right-shape)
   - [The honest tradeoffs (why this is an *option*, not a free win)](#the-honest-tradeoffs-why-this-is-an-option-not-a-free-win)
   - [v1 stance](#v1-stance)
@@ -40,7 +41,7 @@ for how the Spirit original works and tool-pgroll.md for pgroll.
   - [2. Scope of v1](#2-scope-of-v1)
   - [3. Repo location / language](#3-repo-location--language)
   - [4. Expand/contract (pgroll) as a second execution backend](#4-expandcontract-pgroll-as-a-second-execution-backend)
-  - [5. Declarative diff engine — build on pg_query_go vs wrap pg-schema-diff](#5-declarative-diff-engine--build-on-pg_query_go-vs-wrap-pg-schema-diff)
+  - [5. Declarative diff engine — build on go-pgquery vs wrap pg-schema-diff](#5-declarative-diff-engine--build-on-go-pgquery-vs-wrap-pg-schema-diff)
 - [Next step](#next-step)
 
 ## Architecture: decoupled planner, router, and executors
@@ -76,7 +77,7 @@ seam inside the copy-and-swap executor is the same idea applied one level down.
         ╰────────────────┬─────────────────────────────────────────────────────╯
                          │
    ┌─────────────────────▼──────────────────── PLANNER / front-end (shared) ────┐
-   │  pkg/statement   parse ALTER/CREATE (pg_query_go)                          │
+   │  pkg/statement   parse ALTER/CREATE (go-pgquery)                           │
    │  pkg/schemadiff  introspect live schema → diff vs desired → ordered ALTERs │
    │  classifier      per op: native-safe | needs-rewrite | refuse              │
    │  pkg/lint        reject unsafe/unsupported up front                        │
@@ -136,6 +137,35 @@ seam inside the copy-and-swap executor is the same idea applied one level down.
    NOT VALID …    (Pattern A)
 ```
 
+### How the planner understands DDL (decided)
+
+The DDL-understanding mechanism — the equivalent of Spirit's TiDB parser — is a **layered
+hybrid**, decided deliberately rather than inherited, mapping each function to the mechanism
+that serves it best:
+
+- **Classification** parses with the real PostgreSQL grammar via
+  [`wasilibs/go-pgquery`](https://github.com/wasilibs/go-pgquery) — `libpg_query` compiled to
+  WebAssembly, executed in-process by wazero. Same grammar as the server, pure-Go builds (no
+  cgo toolchain), and a parser crash is a recoverable Go error rather than a process-wide
+  segfault — containment that matters in a shared process owning in-flight migrations. The Wasm
+  module is embedded at build time (`go:embed`); nothing is downloaded at runtime. The cgo
+  [`pg_query_go`](https://github.com/pganalyze/pg_query_go) is the documented escape hatch —
+  API-compatible, a one-file swap.
+- **Shadow-table DDL and checkpoint fingerprints** come from **execute-and-introspect**: apply
+  the change inside a rolled-back transaction on the engine-owned
+  [scratch database](#plan-time-prerequisite-the-scratch-database) hydrated to the
+  before-schema, then read the canonical after-state back from the catalogs (`pg_get_*def`).
+  No AST surgery — correctness of the after-schema is delegated to PostgreSQL itself, and the
+  resume fingerprint (ST-2) hashes the introspected model, not SQL text.
+- **Refusals (RF-1..5)** use both layers: parse-level lint for what the AST shows (dangerous
+  literals, ambiguous renames), scratch execution for semantic truth (syntax *and* semantics,
+  with the server's own SQLSTATEs). Refusal messages quote the server error where one exists.
+
+Classification still *predicts* lock/rewrite behaviour — an empty scratch table reveals nothing
+about a 2 TB rewrite — so execute-and-introspect complements the classifier, never replaces it.
+A structured operations DSL (pgroll-style) and catalog-snapshot diffing are noted as possible
+future *additional* front doors for declarative mode, not the primary.
+
 ### Why this is the right shape
 
 This is exactly the answer to *"why build only copy-and-swap when pgroll already wins some
@@ -208,7 +238,7 @@ live schema (introspected)  ─┘                          │
 
 ### How the diff is derived
 
-1. **Parse desired state** with `pg_query_go` into a normalized table model (columns, types,
+1. **Parse desired state** with `go-pgquery` into a normalized table model (columns, types,
    defaults, nullability, identity/sequences, constraints, indexes).
 2. **Introspect live state** from the catalogs (`pg_attribute`, `pg_constraint`, `pg_index`,
    `pg_attrdef`, …) into the same normalized model.
@@ -251,9 +281,9 @@ remains the primitive that everything ultimately runs through.
 this front-end: introspection, canonicalization (by applying the desired DDL to a **temp
 database** and letting the server itself canonicalize), dependency-ordered emission of the same
 safe idioms, per-statement timeouts, typed **hazard annotations**, and **plan validation**
-against the temp database. Whether `pkg/schemadiff` wraps it or builds on `pg_query_go`
+against the temp database. Whether `pkg/schemadiff` wraps it or builds on `go-pgquery`
 directly is
-[open decision #5](#5-declarative-diff-engine--build-on-pg_query_go-vs-wrap-pg-schema-diff).
+[open decision #5](#5-declarative-diff-engine--build-on-go-pgquery-vs-wrap-pg-schema-diff).
 Either way its output flows through our classifier and executors unchanged — planner output is
 a request, not a permission.
 
@@ -460,7 +490,9 @@ matrix is part of the "decisions, not options" philosophy.
   of which comes along by creating a table with the right columns. Miss the grants and
   application roles **lose access at the instant of cutover**. The cutover refuses to swap until
   this fidelity checklist passes; OID-bound dependents (views, publications) are refused up
-  front in v1 (see the schema-shape matrix above).
+  front in v1 (see the schema-shape matrix above). The shadow's column definition itself comes
+  from [execute-and-introspect](#how-the-planner-understands-ddl-decided) on the scratch
+  database, not from AST transformation of the user's `ALTER`.
 
 ### What "tuned for Aurora" actually means here
 
@@ -497,6 +529,21 @@ this section states *why* and pins the analog to the underlying primitive.
 | **Lossy conversions** (shorten `VARCHAR` below longest value, add `NOT NULL` w/o default, add `UNIQUE` on non-unique data) | Refuse; require the data be fixed first | These can fail or truncate *during the copy or the constraint validation*, after work is spent. PG surfaces them as `VALIDATE CONSTRAINT` / cast failures; better to reject up front. |
 | Read-replica `<10s` lag fidelity | Not a goal | Like Spirit, the engine prioritizes copy throughput; it observes Aurora reader/slot lag only to throttle and protect DR, not to guarantee replica freshness. |
 
+### Plan-time prerequisite: the scratch database
+
+[Execute-and-introspect](#how-the-planner-understands-ddl-decided) (semantic validation,
+shadow-DDL derivation, checkpoint fingerprints) needs a scratch database **on the target
+cluster** — server version and extension parity hold by construction, and the storage cost is
+schema-only (no data ever lands in scratch). Preflight (ST-6) verifies one of two acceptable
+states and refuses with a stated reason otherwise:
+
+1. **`pg_sprite_scratch` is pre-provisioned** (engine-role-owned), or
+2. the engine role holds **`CREATEDB`**, so preflight can self-provision it.
+
+The scratch database is engine-owned and disposable: preflight may reset it (drop/recreate
+contents) at any time. Restricted environments that won't grant `CREATEDB` pre-provision
+instead.
+
 ### Postgres-only preconditions Spirit has no analog for
 
 These have **no MySQL counterpart** but are hard requirements for the logical-decoding path:
@@ -536,7 +583,7 @@ pkg/applier/          -> ON CONFLICT upsert + delete apply
 pkg/table/            -> PK-range chunkers (optimistic + composite), dynamic sizing
 pkg/checksum/         -> md5/row-text chunked verification
 pkg/dbconn/           -> pgx pool, retries, lock_timeout, RDS CA, pg_terminate_backend
-pkg/statement/        -> pg_query_go parsing + "is this natively safe?" classifier
+pkg/statement/        -> go-pgquery parsing + "is this natively safe?" classifier
 pkg/schemadiff/       -> declarative mode: introspect live schema, diff vs desired
                          CREATE TABLE, derive ordered ALTER/CREATE statements (+ fmt)
 pkg/lint/             -> unsafe-DDL linters (PG flavored)
@@ -549,12 +596,15 @@ pkg/throttler/        -> Aurora PG replica-lag / slot-lag throttle
   `pgconn`/`pglogrepl` building blocks for logical replication.
 - **`github.com/jackc/pglogrepl`** — start replication, parse `pgoutput`/`wal2json`
   messages, send standby status (LSN flush) updates. This is the binlog-syncer analog.
-- **`github.com/pganalyze/pg_query_go/v5`** — parse `ALTER`/`CREATE TABLE` (libpg_query,
-  the actual Postgres grammar). Analog of Spirit's TiDB parser.
+- **`github.com/wasilibs/go-pgquery`** — parse `ALTER`/`CREATE TABLE` with the actual Postgres
+  grammar (`libpg_query` compiled to Wasm, executed by wazero: pure-Go builds, parser crashes
+  contained). Analog of Spirit's TiDB parser. The cgo `github.com/pganalyze/pg_query_go/v5` is
+  the API-compatible escape hatch (see
+  [How the planner understands DDL](#how-the-planner-understands-ddl-decided)).
 - **`github.com/stripe/pg-schema-diff`** *(candidate — open decision #5)* — declarative diff
   engine: introspection + dependency-ordered plan emission with hazard annotations and
   temp-database plan validation; would power `pkg/schemadiff` instead of building the diff on
-  `pg_query_go` directly.
+  `go-pgquery` directly.
 - **`github.com/alecthomas/kong`** — CLI, same as Spirit.
 
 ## Design decisions inherited from Spirit (safety over speed)
@@ -665,7 +715,7 @@ why building declarative first costs nothing on the execution side.
 
 ### 3. Repo location / language
 
-Go (reuse `pgx` + `pglogrepl` + `pg_query_go`; matches Spirit's language and idioms). Fresh
+Go (reuse `pgx` + `pglogrepl` + `go-pgquery`; matches Spirit's language and idioms). Fresh
 standalone repo — this repository.
 
 ### 4. Expand/contract (pgroll) as a second execution backend
@@ -684,9 +734,9 @@ rewrites use copy-and-swap.
   one-shot vs start/complete/rollback lifecycles under one `status`; and the default routing
   policy (auto-route vs explicit `--strategy`) given "decisions, not options".
 
-### 5. Declarative diff engine — build on pg_query_go vs wrap pg-schema-diff
+### 5. Declarative diff engine — build on go-pgquery vs wrap pg-schema-diff
 
-Whether `pkg/schemadiff` builds the desired-vs-live diff on `pg_query_go` + our own schema
+Whether `pkg/schemadiff` builds the desired-vs-live diff on `go-pgquery` + our own schema
 model, or wraps [stripe/pg-schema-diff](https://github.com/stripe/pg-schema-diff) (MIT, Go,
 PG 14–17, actively maintained) as the diff engine.
 
@@ -697,9 +747,10 @@ PG 14–17, actively maintained) as the diff engine.
   validation against the temp database. Plan generation is cleanly separated from application,
   so our executors keep our own timeout/lock/retry discipline. It is a **periphery** dependency
   (plan generation), so the TCB bar does not apply — pinned like any load-bearing dep.
-- **Costs of wrapping:** the temp-database factory is an operational precondition
-  (`CREATE DATABASE` on the target or a scratch instance — needs a deliberate answer for
-  locked-down production clusters); renames surface as drop+add and **must** sit behind our
+- **Costs of wrapping:** the temp-database factory is an operational precondition — answered:
+  the engine-owned [scratch database](#plan-time-prerequisite-the-scratch-database) is already
+  a preflight-verified prerequisite for execute-and-introspect, so wrapping adds no new
+  operational demand; renames surface as drop+add and **must** sit behind our
   destructive-diff gate and never-guess-renames refusals; type support beyond enums is missing;
   its embedded timeout policy is replaced by ours at execution.
 - **Build:** full control and no temp-database precondition — at the cost of the hardest code
diff --git a/docs/tcb-model.md b/docs/tcb-model.md
index ba12912..69a5f72 100644
--- a/docs/tcb-model.md
+++ b/docs/tcb-model.md
@@ -161,14 +161,14 @@ short:
 | --- | --- | --- |
 | `pgx/v5` / `pgconn` | TCB (unavoidable — the wire) | pin, review upgrades like TCB changes, changelog read before bump |
 | `pglogrepl` | TCB (decode path) | same |
-| `pg_query_go` | boundary (parses untrusted input into `Classified`) | fuzz at our boundary; parse failure is an error (CO-7), never a fallback |
+| `go-pgquery` (Wasm `libpg_query`) | boundary (parses untrusted input into `Classified`) | fuzz at our boundary; parse failure is an error (CO-7), never a fallback; a parser crash is a Wasm trap surfaced as a Go error, not a process crash; verify wasilibs' reproducible-build provenance on every bump (cgo `pg_query_go` is the API-compatible escape hatch) |
 | `kong`, `testcontainers`, testify | periphery / test-only | normal hygiene |
 
 Rule: **no new dependency inside TCB packages without an explicit recorded decision.** CI
 enforces the import boundary (below), so a periphery-only dep physically cannot creep into the
 core. The decision rubric, in order:
 
-1. **Is it load-bearing expertise?** A real SQL grammar (`pg_query_go`), the wire protocol
+1. **Is it load-bearing expertise?** A real SQL grammar (`go-pgquery`), the wire protocol
    (`pgx`/`pglogrepl`), crypto — take the dependency, pin it, treat it as TCB. Hand-rolling a
    SQL parser to avoid a dependency would be the *opposite* of safety (CO-7 exists because
    string-splitting SQL is how tools corrupt data).
diff --git a/docs/testing.md b/docs/testing.md
index 91fb224..721d08d 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -116,8 +116,9 @@ rows that the operation demonstrably spans the observation or injection
 point — otherwise the operation can finish before the fault lands and the
 test passes without testing anything. Vacuous runs are a failure: the test
 asserts the interruption actually hit mid-operation (e.g. the checkpoint
-shows partial progress), not just the final state. *Binds:* Phase 4
-(copy-and-swap kill/resume) onward. *Source:* SchemaBot's in-flight
+shows partial progress), not just the final state. *Binds:* Phase 1
+(budget-cancellation fixtures, which seed enough rows that a rewrite cannot
+finish inside its statement budget) onward. *Source:* SchemaBot's in-flight
 progress tests, which seed large row counts so an operation spans a poll
 interval.
 
diff --git a/go.mod b/go.mod
index 1f75851..c9120f1 100644
--- a/go.mod
+++ b/go.mod
@@ -5,9 +5,11 @@ go 1.26
 require (
 	github.com/alecthomas/kong v1.15.0
 	github.com/jackc/pgx/v5 v5.10.0
+	github.com/pganalyze/pg_query_go/v6 v6.2.2
 	github.com/stretchr/testify v1.11.1
 	github.com/testcontainers/testcontainers-go v0.43.0
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.43.0
+	github.com/wasilibs/go-pgquery v0.0.0-20260728010200-155ebad2880e
 )
 
 require (
@@ -52,8 +54,10 @@ require (
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/shirou/gopsutil/v4 v4.26.5 // indirect
 	github.com/sirupsen/logrus v1.9.4 // indirect
+	github.com/tetratelabs/wazero v1.12.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.16 // indirect
 	github.com/tklauser/numcpus v0.11.0 // indirect
+	github.com/wasilibs/wazero-helpers v0.0.0-20250123031827-cd30c44769bb // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
@@ -64,5 +68,6 @@ require (
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/text v0.37.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index 3f1ddff..34d264f 100644
--- a/go.sum
+++ b/go.sum
@@ -99,6 +99,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040=
 github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M=
+github.com/pganalyze/pg_query_go/v6 v6.2.2 h1:O0L6zMC226R82RF3X5n0Ki6HjytDsoAzuzp4ATVAHNo=
+github.com/pganalyze/pg_query_go/v6 v6.2.2/go.mod h1:Cn6+j4870kJz3iYNsb0VsNG04vpSWgEvBwc590J4qD0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
@@ -120,10 +122,16 @@ github.com/testcontainers/testcontainers-go v0.43.0 h1:oEQx5MW2DGd9z3AeEQfB2lPM0
 github.com/testcontainers/testcontainers-go v0.43.0/go.mod h1:+VxkT2NQnKOZPKi6praMuMKYHYyOGXr0XSBSlSMCzFo=
 github.com/testcontainers/testcontainers-go/modules/postgres v0.43.0 h1:ShNOFYAF4lKHvdIG258hi69bSxC88uXnxJkJvNs/IVs=
 github.com/testcontainers/testcontainers-go/modules/postgres v0.43.0/go.mod h1:vdq5/RqmGfWeefzyfcVI/pID1rzmc1TDvqXa15bPJks=
+github.com/tetratelabs/wazero v1.12.0 h1:DuWcpNu/FzgEXgGBDp8J1Spc+CWOvvtvVyjKlaZopYU=
+github.com/tetratelabs/wazero v1.12.0/go.mod h1:LvKtzl2RqO4gyF27BiXU+nKAjcV8f38U+kP/q2vgxh0=
 github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA=
 github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI=
 github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw=
 github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ=
+github.com/wasilibs/go-pgquery v0.0.0-20260728010200-155ebad2880e h1:yWIo9Ibxg0qNScjPcdaH99BfetgmYepCxs9a6TFC2LM=
+github.com/wasilibs/go-pgquery v0.0.0-20260728010200-155ebad2880e/go.mod h1:ZSyYLCRbk2xPqu7lgfrDSSHm+g/7Rxk6JK4KE2cxJ3s=
+github.com/wasilibs/wazero-helpers v0.0.0-20250123031827-cd30c44769bb h1:gQ+ZV4wJke/EBKYciZ2MshEouEHFuinB85dY3f5s1q8=
+github.com/wasilibs/wazero-helpers v0.0.0-20250123031827-cd30c44769bb/go.mod h1:jMeV4Vpbi8osrE/pKUxRZkVaA0EX7NZN0A9/oRzgpgY=
 github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
@@ -154,6 +162,8 @@ golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
 golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
 golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
diff --git a/internal/cli/bytesize.go b/internal/cli/bytesize.go
new file mode 100644
index 0000000..0e3a501
--- /dev/null
+++ b/internal/cli/bytesize.go
@@ -0,0 +1,56 @@
+package cli
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+)
+
+// byteSize is a flag type for human-readable sizes ("512MiB", "1GiB", plain
+// bytes). All suffixes are binary multiples — PostgreSQL's own convention for
+// size units.
+type byteSize int64
+
+// suffixMultiplier maps an accepted (lowercased) size suffix to its
+// multiplier; ok is false for anything unrecognized.
+func suffixMultiplier(suffix string) (mult int64, ok bool) {
+	switch suffix {
+	case "", "b":
+		return 1, true
+	case "kb", "kib":
+		return 1 << 10, true
+	case "mb", "mib":
+		return 1 << 20, true
+	case "gb", "gib":
+		return 1 << 30, true
+	case "tb", "tib":
+		return 1 << 40, true
+	default:
+		return 0, false
+	}
+}
+
+// UnmarshalText implements encoding.TextUnmarshaler so kong can parse the
+// flag directly.
+func (b *byteSize) UnmarshalText(text []byte) error {
+	s := strings.TrimSpace(strings.ToLower(string(text)))
+	digits := strings.TrimRight(s, "bkmgit ")
+	suffix := strings.TrimSpace(s[len(digits):])
+	mult, ok := suffixMultiplier(suffix)
+	if !ok {
+		return fmt.Errorf("unknown size suffix %q in %q (use B, KiB, MiB, GiB, or TiB)", suffix, string(text))
+	}
+	n, err := strconv.ParseInt(strings.TrimSpace(digits), 10, 64)
+	if err != nil {
+		return fmt.Errorf("parse size %q: %w", string(text), err)
+	}
+	if n <= 0 {
+		return fmt.Errorf("size must be positive, got %q", string(text))
+	}
+	if n > math.MaxInt64/mult {
+		return fmt.Errorf("size %q overflows int64 bytes", string(text))
+	}
+	*b = byteSize(n * mult)
+	return nil
+}
diff --git a/internal/cli/bytesize_test.go b/internal/cli/bytesize_test.go
new file mode 100644
index 0000000..0b39eda
--- /dev/null
+++ b/internal/cli/bytesize_test.go
@@ -0,0 +1,43 @@
+package cli
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestByteSizeUnmarshal(t *testing.T) {
+	tests := []struct {
+		in   string
+		want int64
+	}{
+		{"1024", 1024},
+		{"512B", 512},
+		{"4KiB", 4 << 10},
+		{"4kb", 4 << 10},
+		{"100MiB", 100 << 20},
+		{"100MB", 100 << 20},
+		{"1GiB", 1 << 30},
+		{"1gb", 1 << 30},
+		{"2TiB", 2 << 40},
+		{" 8 MiB ", 8 << 20},
+	}
+	for _, tt := range tests {
+		t.Run(tt.in, func(t *testing.T) {
+			var b byteSize
+			require.NoError(t, b.UnmarshalText([]byte(tt.in)))
+			assert.Equal(t, byteSize(tt.want), b)
+		})
+	}
+}
+
+func TestByteSizeUnmarshalRejectsInvalid(t *testing.T) {
+	// The last two would overflow int64 bytes after unit multiplication.
+	for _, in := range []string{"", "GiB", "1XB", "-5MiB", "0", "1.5GiB", "9999999999GiB", "9223372036854775807KiB"} {
+		t.Run(in, func(t *testing.T) {
+			var b byteSize
+			assert.Error(t, b.UnmarshalText([]byte(in)))
+		})
+	}
+}
diff --git a/internal/cli/cli.go b/internal/cli/cli.go
index 3b3d805..b718b7c 100644
--- a/internal/cli/cli.go
+++ b/internal/cli/cli.go
@@ -1,9 +1,14 @@
-// Package cli defines the pg-sprite command tree (Kong). Subcommand Run
-// methods are stubs; each build-plan phase fills one in.
+// Package cli defines the pg-sprite command tree (Kong). migrate and status
+// are implemented (the Phase 1 optimistic front door); the remaining
+// subcommand Run methods are stubs each build-plan phase fills in.
 package cli
 
 import (
+	"context"
 	"fmt"
+	"io"
+	"log/slog"
+	"os"
 	"time"
 
 	"github.com/alecthomas/kong"
@@ -19,7 +24,7 @@ type CLI struct {
 	Diff    DiffCmd    `cmd:"" help:"Diff a desired-state schema file against the live schema."`
 	Fmt     FmtCmd     `cmd:"" help:"Canonicalize a schema file."`
 	Lint    LintCmd    `cmd:"" help:"Lint DDL for unsafe patterns."`
-	Status  StatusCmd  `cmd:"" help:"Report the status of a running migration."`
+	Status  StatusCmd  `cmd:"" help:"Report the status of a running schema change."`
 }
 
 // New returns an empty command tree for kong.Parse.
@@ -36,27 +41,56 @@ type DBFlags struct {
 	CACert           string        `help:"CA bundle path for verify-full TLS. RDS/Aurora endpoints verify with the embedded bundle automatically." env:"PGSPRITE_CA_CERT" type:"existingfile"`
 	LockTimeout      time.Duration `help:"Session lock_timeout applied to every statement." default:"3s"`
 	StatementTimeout time.Duration `help:"Session statement_timeout applied to every statement." default:"30s"`
+	Debug            bool          `help:"Log statement-level tracing and lifecycle diagnostics to stderr."`
+
+	// diagOut overrides the diagnostics destination (stderr) in tests. Kong
+	// ignores unexported fields.
+	diagOut io.Writer
 }
 
 // Config translates the flags into the connectivity layer's configuration.
+// The tracer is wired only under --debug: dbconn skips statement tracing
+// entirely for a nil logger.
 func (f DBFlags) Config() dbconn.Config {
-	return dbconn.Config{
+	cfg := dbconn.Config{
 		URL:              f.URL,
 		CACertPath:       f.CACert,
 		LockTimeout:      f.LockTimeout,
 		StatementTimeout: f.StatementTimeout,
 	}
+	if f.Debug {
+		cfg.Logger = f.diag()
+	}
+	return cfg
 }
 
-// MigrateCmd runs a schema change (imperative front-end).
+// diag returns the diagnostics logger: debug-level text on stderr (or the
+// test override) under --debug, a discarding logger otherwise. Diagnostics
+// never share stdout with command output.
+func (f DBFlags) diag() *slog.Logger {
+	if !f.Debug {
+		return slog.New(slog.DiscardHandler)
+	}
+	out := f.diagOut
+	if out == nil {
+		out = os.Stderr
+	}
+	return slog.New(slog.NewTextHandler(out, &slog.HandlerOptions{Level: slog.LevelDebug}))
+}
+
+// MigrateCmd runs a schema change (imperative front-end): the Phase 1
+// optimistic front door. Easy changes execute directly under tight budgets;
+// everything else is refused with an explicit verdict.
 type MigrateCmd struct {
 	DBFlags `embed:""`
 
-	Alter string `help:"Imperative ALTER statement to run." name:"alter"`
+	Alter        string   `help:"Imperative ALTER statement to run." name:"alter" required:""`
+	MaxTableSize byteSize `help:"Size threshold above which the optimistic attempt is skipped, measured as the table's full on-disk footprint: heap, indexes, and TOAST, all partitions (binary units: B, KiB, MiB, GiB, TiB)." default:"1GiB"`
+	JSON         bool     `help:"Emit the verdict as JSON."`
 }
 
 // Run implements the migrate subcommand.
-func (c *MigrateCmd) Run() error { return notImplemented("migrate") }
+func (c *MigrateCmd) Run() error { return c.run(context.Background(), os.Stdout) }
 
 // DiffCmd derives statements from a desired-state schema (declarative front-end).
 type DiffCmd struct {
@@ -82,10 +116,12 @@ type LintCmd struct{}
 // Run implements the lint subcommand.
 func (c *LintCmd) Run() error { return notImplemented("lint") }
 
-// StatusCmd reports migration progress.
+// StatusCmd reports schema-change progress.
 type StatusCmd struct {
 	DBFlags `embed:""`
+
+	JSON bool `help:"Emit the session listing as JSON."`
 }
 
 // Run implements the status subcommand.
-func (c *StatusCmd) Run() error { return notImplemented("status") }
+func (c *StatusCmd) Run() error { return c.run(context.Background(), os.Stdout) }
diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go
new file mode 100644
index 0000000..6566ec7
--- /dev/null
+++ b/internal/cli/migrate.go
@@ -0,0 +1,180 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/block/pg-sprite/pkg/dbconn"
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/block/pg-sprite/pkg/verdict"
+)
+
+// run is the migrate flow: gate the statement type, size-guard the table,
+// attempt the change under budget, and end in exactly one verdict. Refusal
+// verdicts are printed to out and returned as verdict.ErrRefused so the entry
+// point maps them to the refusal exit code.
+func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error {
+	logger := c.diag()
+	st, err := statement.ParseOne(c.Alter)
+	if err != nil {
+		return err
+	}
+	logger.Debug("statement parsed", "kind", st.Kind(), "schema", st.Schema(), "table", st.Table())
+	if v, refused := gateVerdict(st); refused {
+		return c.emit(out, v)
+	}
+
+	pool, err := dbconn.NewPool(ctx, c.Config())
+	if err != nil {
+		return err
+	}
+	defer pool.Close()
+
+	pt, err := preflight.CheckTable(ctx, pool, st.Schema(), st.Table(), int64(c.MaxTableSize))
+	var sizeErr *preflight.SizeError
+	if errors.As(err, &sizeErr) {
+		return c.emit(out, sizeGuardVerdict(st, sizeErr))
+	}
+	if err != nil {
+		return err
+	}
+	logger.Debug("preflight passed",
+		"table", qualified(st), "total_bytes", pt.TotalBytes(), "limit_bytes", int64(c.MaxTableSize))
+
+	budget := executor.Budget{LockTimeout: c.LockTimeout, StatementTimeout: c.StatementTimeout}
+	start := time.Now()
+	err = executor.AttemptNative(ctx, pool, pt, st, budget)
+	elapsed := time.Since(start)
+	var budgetErr *executor.BudgetError
+	if errors.As(err, &budgetErr) {
+		logger.Debug("optimistic attempt cancelled",
+			"cause", budgetErr.Cause, "budget", budgetErr.Budget, "elapsed", elapsed)
+		return c.emit(out, budgetVerdict(st, budgetErr))
+	}
+	if err != nil {
+		return err
+	}
+	logger.Debug("optimistic attempt committed", "table", qualified(st), "elapsed", elapsed)
+	return c.emit(out, verdict.Verdict{
+		Outcome:   verdict.OutcomeExecuted,
+		Statement: st.SQL(),
+		Table:     qualified(st),
+		Detail: fmt.Sprintf("committed within budgets (lock %s, statement %s): the change was effectively instant",
+			budget.LockTimeout, budget.StatementTimeout),
+	})
+}
+
+// emit prints the verdict in the selected format and returns ErrRefused for
+// refusals so the exit code distinguishes them from operational errors.
+func (c *MigrateCmd) emit(out io.Writer, v verdict.Verdict) error {
+	text := v.String()
+	if c.JSON {
+		var err error
+		if text, err = v.JSON(); err != nil {
+			return err
+		}
+	}
+	if _, err := fmt.Fprintln(out, text); err != nil {
+		return fmt.Errorf("write verdict: %w", err)
+	}
+	if v.Outcome == verdict.OutcomeRefused {
+		return verdict.ErrRefused
+	}
+	return nil
+}
+
+// gateVerdict is the Phase 1 statement-type gate: only ALTER TABLE proceeds;
+// index maintenance is pointed at its concurrent idiom, everything else is
+// unsupported. Refused statements are never executed.
+func gateVerdict(st statement.Statement) (verdict.Verdict, bool) {
+	v := verdict.Verdict{Outcome: verdict.OutcomeRefused, Statement: st.SQL()}
+	switch st.Kind() {
+	case statement.KindAlterTable:
+		return verdict.Verdict{}, false
+	case statement.KindCreateIndex, statement.KindDropIndex, statement.KindReindex:
+		v.Reason = verdict.ReasonIndexStatement
+		v.Detail, v.SaferIdiom = indexAdvice(st)
+	case statement.KindOther:
+		v.Reason = verdict.ReasonUnsupportedStatement
+		v.Detail = "only ALTER TABLE statements are supported by the optimistic front door"
+	}
+	return v, true
+}
+
+// indexAdvice explains an index-statement refusal. The already-concurrent
+// forms carry no safer idiom: suggesting the statement the user submitted
+// would confuse a human once and send a resubmitting automation into a loop.
+func indexAdvice(st statement.Statement) (detail, saferIdiom string) {
+	if st.Concurrent() {
+		return "this is already the safe concurrent idiom; pg-sprite does not drive index maintenance yet — run it directly against the database", ""
+	}
+	switch st.Kind() {
+	case statement.KindCreateIndex:
+		return "a plain CREATE INDEX blocks writes for the whole build; the concurrent build does not", "CREATE INDEX CONCURRENTLY"
+	case statement.KindDropIndex:
+		return "a plain DROP INDEX takes ACCESS EXCLUSIVE on the table; the concurrent drop does not", "DROP INDEX CONCURRENTLY"
+	case statement.KindReindex:
+		return "a plain REINDEX blocks writes; the concurrent rebuild does not", "REINDEX ... CONCURRENTLY"
+	default:
+		return "", ""
+	}
+}
+
+// sizeGuardVerdict is the refusal for tables above the size threshold, where
+// even a budget-bounded attempt would visibly stall the table.
+func sizeGuardVerdict(st statement.Statement, sizeErr *preflight.SizeError) verdict.Verdict {
+	return verdict.Verdict{
+		Outcome:   verdict.OutcomeRefused,
+		Reason:    verdict.ReasonTableTooLarge,
+		Statement: st.SQL(),
+		Table:     qualified(st),
+		Detail: fmt.Sprintf("table is %d bytes on disk (heap, indexes, and TOAST), above the %d-byte "+
+			"--max-table-size threshold. pg-sprite cannot yet prove this change is instant on a table this "+
+			"size; if it requires a rewrite, a cancelled attempt is not a free probe — it would hold "+
+			"ACCESS EXCLUSIVE doing rewrite work for the whole budget",
+			sizeErr.TotalBytes, sizeErr.LimitBytes),
+	}
+}
+
+// budgetVerdict is the refusal for an attempt that exceeded its lock or
+// statement budget and was cancelled without executing.
+func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError) verdict.Verdict {
+	v := verdict.Verdict{
+		Outcome:   verdict.OutcomeRefused,
+		Reason:    verdict.ReasonBudgetExceeded,
+		Statement: st.SQL(),
+		Table:     qualified(st),
+	}
+	switch budgetErr.Cause {
+	case executor.CauseLock:
+		v.Cause = verdict.CauseLockBudget
+		v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget: the table is too "+
+			"contended for a blind attempt; nothing was executed", budgetErr.Budget)
+	case executor.CauseStatement:
+		v.Cause = verdict.CauseStatementBudget
+		v.Detail = fmt.Sprintf("cancelled after the %s statement budget: the change does real rewrite work, "+
+			"not an in-place catalog change, and needs a copy-and-swap rewrite that pg-sprite does not perform yet. "+
+			"If it adds a constraint, ADD CONSTRAINT ... NOT VALID followed by VALIDATE CONSTRAINT avoids the long lock",
+			budgetErr.Budget)
+	default:
+		v.Detail = budgetErr.Error()
+	}
+	return v
+}
+
+// qualified renders the statement's target table for the verdict, empty when
+// the statement has none.
+func qualified(st statement.Statement) string {
+	if st.Table() == "" {
+		return ""
+	}
+	if st.Schema() == "" {
+		return st.Table()
+	}
+	return st.Schema() + "." + st.Table()
+}
diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go
new file mode 100644
index 0000000..edde1b6
--- /dev/null
+++ b/internal/cli/migrate_integration_test.go
@@ -0,0 +1,359 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+	neturl "net/url"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"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/verdict"
+)
+
+// newMigrateCmd builds a MigrateCmd with the flag defaults kong would apply.
+func newMigrateCmd(url, alter string) *MigrateCmd {
+	return &MigrateCmd{
+		DBFlags: DBFlags{
+			URL:              url,
+			LockTimeout:      3 * time.Second,
+			StatementTimeout: 30 * time.Second,
+		},
+		Alter:        alter,
+		MaxTableSize: 1 << 30,
+	}
+}
+
+// Acceptance (i): an instant-eligible change runs and commits within budget.
+func TestMigrateExecutesInstantChange(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+	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)
+
+	cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+	cmd.JSON = true
+	var out strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &out))
+
+	var v verdict.Verdict
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &v))
+	assert.Equal(t, verdict.OutcomeExecuted, v.Outcome)
+	assert.Equal(t, schema+".t", v.Table)
+
+	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 = 'age'`, schema).Scan(&typ))
+	assert.Equal(t, "integer", typ)
+}
+
+// ALTER TABLE ... RENAME COLUMN parses as a RenameStmt, not an
+// AlterTableStmt, but is a table-targeted instant catalog change: the front
+// door must route it through, not refuse it as unsupported.
+func TestMigrateExecutesRenameColumn(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, a int)", schema))
+	require.NoError(t, err)
+
+	cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t RENAME COLUMN a TO b", schema))
+	cmd.JSON = true
+	var out strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &out))
+
+	var v verdict.Verdict
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &v))
+	assert.Equal(t, verdict.OutcomeExecuted, v.Outcome)
+
+	var n int
+	require.NoError(t, pool.QueryRow(t.Context(),
+		`SELECT count(*) FROM information_schema.columns
+		 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'b'`, schema).Scan(&n))
+	assert.Equal(t, 1, n, "the rename must have committed")
+}
+
+// Acceptance (ii): a rewrite-requiring change is cancelled, leaves schema and
+// data unchanged, and returns the not-native-safe verdict with its reason.
+func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+	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)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(
+		"INSERT INTO %s.t SELECT g, repeat('x', 100) FROM generate_series(1, 300000) g", schema))
+	require.NoError(t, err)
+
+	cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema))
+	cmd.StatementTimeout = 50 * time.Millisecond
+	cmd.JSON = true
+	var out strings.Builder
+	err = cmd.run(t.Context(), &out)
+	require.ErrorIs(t, err, verdict.ErrRefused)
+
+	var v verdict.Verdict
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &v))
+	assert.Equal(t, verdict.OutcomeRefused, v.Outcome)
+	assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason)
+	assert.Equal(t, verdict.CauseStatementBudget, v.Cause)
+	assert.Equal(t, schema+".t", v.Table)
+
+	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 = 'id'`, schema).Scan(&typ))
+	assert.Equal(t, "integer", typ, "the cancelled attempt must not change the schema")
+	var count int
+	require.NoError(t, pool.QueryRow(t.Context(),
+		fmt.Sprintf("SELECT count(*) FROM %s.t", schema)).Scan(&count))
+	assert.Equal(t, 300000, count, "the cancelled attempt must not change the data")
+}
+
+// Acceptance (iii): a table above the size threshold skips the attempt and
+// returns the same verdict class.
+func TestMigrateSizeGuardSkipsAttempt(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+	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)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(
+		"INSERT INTO %s.t SELECT g, 'v' FROM generate_series(1, 1000) g", schema))
+	require.NoError(t, err)
+
+	cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+	cmd.MaxTableSize = 1 // guarantees the guard fires without a big fixture
+	cmd.JSON = true
+	var out strings.Builder
+	err = cmd.run(t.Context(), &out)
+	require.ErrorIs(t, err, verdict.ErrRefused)
+
+	var v verdict.Verdict
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &v))
+	assert.Equal(t, verdict.ReasonTableTooLarge, v.Reason)
+
+	// The attempt was skipped, so the (instant-eligible) change must not
+	// have been applied.
+	var n int
+	require.NoError(t, pool.QueryRow(t.Context(),
+		`SELECT count(*) FROM information_schema.columns
+		 WHERE table_schema = $1 AND table_name = 't' AND column_name = 'age'`, schema).Scan(&n))
+	assert.Zero(t, n, "the size guard must skip the attempt entirely")
+}
+
+// Acceptance (iv): non-ALTER TABLE statements are refused with the safe-idiom
+// pointer and never executed. The gate needs no database at all.
+func TestMigrateGateRefusesWithoutDatabase(t *testing.T) {
+	tests := []struct {
+		name       string
+		alter      string
+		reason     verdict.Reason
+		saferIdiom string
+	}{
+		{"create index", "CREATE INDEX i ON t (c)", verdict.ReasonIndexStatement, "CREATE INDEX CONCURRENTLY"},
+		{"drop index", "DROP INDEX i", verdict.ReasonIndexStatement, "DROP INDEX CONCURRENTLY"},
+		{"reindex", "REINDEX TABLE t", verdict.ReasonIndexStatement, "REINDEX ... CONCURRENTLY"},
+		// The already-concurrent forms carry no safer idiom: suggesting the
+		// statement the user submitted would loop a resubmitting automation.
+		{"create index concurrently", "CREATE INDEX CONCURRENTLY i ON t (c)", verdict.ReasonIndexStatement, ""},
+		{"drop index concurrently", "DROP INDEX CONCURRENTLY i", verdict.ReasonIndexStatement, ""},
+		{"reindex concurrently", "REINDEX TABLE CONCURRENTLY t", verdict.ReasonIndexStatement, ""},
+		{"alter index", "ALTER INDEX i SET (fillfactor = 90)", verdict.ReasonUnsupportedStatement, ""},
+		{"create table", "CREATE TABLE t (id int)", verdict.ReasonUnsupportedStatement, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// An unroutable URL proves the gate refuses before connecting.
+			cmd := newMigrateCmd("postgres://nobody@localhost:1/nope", tt.alter)
+			cmd.JSON = true
+			var out strings.Builder
+			err := cmd.run(t.Context(), &out)
+			require.ErrorIs(t, err, verdict.ErrRefused)
+
+			var v verdict.Verdict
+			require.NoError(t, json.Unmarshal([]byte(out.String()), &v))
+			assert.Equal(t, verdict.OutcomeRefused, v.Outcome)
+			assert.Equal(t, tt.reason, v.Reason)
+			assert.Equal(t, tt.saferIdiom, v.SaferIdiom)
+		})
+	}
+}
+
+func TestMigrateSurfacesParseErrors(t *testing.T) {
+	cmd := newMigrateCmd("postgres://nobody@localhost:1/nope", "ALTER TABEL t ADD COLUMN x int")
+	var out strings.Builder
+	err := cmd.run(t.Context(), &out)
+	require.Error(t, err)
+	assert.NotErrorIs(t, err, verdict.ErrRefused, "a parse failure is an operational error, not a refusal")
+}
+
+// syncWriter guards the diagnostics buffer: pgx tracelog can write from pool
+// housekeeping goroutines concurrently with the command's own lifecycle logs.
+type syncWriter struct {
+	mu sync.Mutex
+	b  strings.Builder
+}
+
+func (w *syncWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.b.Write(p)
+}
+
+func (w *syncWriter) String() string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.b.String()
+}
+
+// --debug emits lifecycle events and pgx statement tracing on the diagnostics
+// stream, and never leaks them into the command's stdout output; without the
+// flag diagnostics are discarded entirely.
+func TestMigrateDebugDiagnostics(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+	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)
+
+	t.Run("debug on", func(t *testing.T) {
+		cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+		cmd.Debug = true
+		cmd.JSON = true
+		var diag syncWriter
+		cmd.diagOut = &diag
+		var out strings.Builder
+		require.NoError(t, cmd.run(t.Context(), &out))
+
+		// A strict decode of stdout proves diagnostics did not leak into the
+		// command's output stream: any interleaved log line would break it.
+		var v verdict.Verdict
+		require.NoError(t, json.Unmarshal([]byte(out.String()), &v),
+			"stdout must carry exactly the verdict, nothing else")
+		assert.Equal(t, verdict.OutcomeExecuted, v.Outcome)
+
+		assert.NotEmpty(t, diag.String(),
+			"--debug must emit diagnostics on the diagnostics stream")
+	})
+
+	t.Run("debug off discards diagnostics", func(t *testing.T) {
+		cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age2 int", schema))
+		var diag syncWriter
+		cmd.diagOut = &diag
+		var out strings.Builder
+		require.NoError(t, cmd.run(t.Context(), &out))
+		assert.Empty(t, diag.String())
+	})
+}
+
+// Both output modes report the empty case: an empty JSON list, and a
+// non-empty human explanation.
+func TestStatusReportsNoSessions(t *testing.T) {
+	url := testutil.StartPostgres(t)
+
+	cmd := &StatusCmd{DBFlags: DBFlags{URL: url}, JSON: true}
+	var out strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &out))
+	var sessions []session
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &sessions))
+	assert.Empty(t, sessions)
+
+	cmd = &StatusCmd{DBFlags: DBFlags{URL: url}}
+	var text strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &text))
+	assert.NotEmpty(t, text.String())
+}
+
+// pg_stat_activity nulls out state and query for other roles' backends when
+// the viewer lacks pg_read_all_stats — the read-only-operator shape. status
+// must render those sessions, not crash the scan.
+func TestStatusHandlesOtherRolesSessions(t *testing.T) {
+	superURL := testutil.StartPostgres(t)
+	superPool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: superURL})
+	require.NoError(t, err)
+	defer superPool.Close()
+
+	_, err = superPool.Exec(t.Context(), "CREATE ROLE limited LOGIN PASSWORD 'limited-test-only'")
+	require.NoError(t, err)
+
+	// Hold a superuser pg-sprite session open on a pinned connection so
+	// pg_stat_activity is guaranteed to contain a foreign-role row while
+	// status runs.
+	conn, err := superPool.Acquire(t.Context())
+	require.NoError(t, err)
+	defer conn.Release()
+	var pid int
+	require.NoError(t, conn.QueryRow(t.Context(), "SELECT pg_backend_pid()").Scan(&pid))
+
+	u, err := neturl.Parse(superURL)
+	require.NoError(t, err)
+	u.User = neturl.UserPassword("limited", "limited-test-only")
+
+	cmd := &StatusCmd{DBFlags: DBFlags{URL: u.String()}, JSON: true}
+	var out strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &out))
+
+	var sessions []session
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &sessions))
+	i := slices.IndexFunc(sessions, func(s session) bool { return s.PID == pid })
+	require.GreaterOrEqual(t, i, 0, "the other role's session must be listed, not crash the scan")
+}
+
+// A live pg-sprite session (any connection made through pkg/dbconn) is
+// reported with its pid and per-session fields. The session is held open on a
+// pinned connection so pg_stat_activity is guaranteed to contain it while
+// status runs.
+func TestStatusReportsActiveSession(t *testing.T) {
+	url := testutil.StartPostgres(t)
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url})
+	require.NoError(t, err)
+	defer pool.Close()
+
+	conn, err := pool.Acquire(t.Context())
+	require.NoError(t, err)
+	defer conn.Release()
+	var pid int
+	// The marker alias makes the session's last-query text deterministic
+	// for the field assertion below.
+	require.NoError(t, conn.QueryRow(t.Context(),
+		"SELECT pg_backend_pid() AS pgsprite_status_marker").Scan(&pid))
+
+	cmd := &StatusCmd{DBFlags: DBFlags{URL: url}, JSON: true}
+	var out strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &out))
+
+	var sessions []session
+	require.NoError(t, json.Unmarshal([]byte(out.String()), &sessions))
+	i := slices.IndexFunc(sessions, func(s session) bool { return s.PID == pid })
+	require.GreaterOrEqual(t, i, 0, "the held session must be listed")
+	assert.Equal(t, "idle", sessions[i].State)
+	assert.Contains(t, sessions[i].Query, "pgsprite_status_marker",
+		"the session's last query must be reported")
+
+	// The human rendering carries the same session.
+	cmd = &StatusCmd{DBFlags: DBFlags{URL: url}}
+	var text strings.Builder
+	require.NoError(t, cmd.run(t.Context(), &text))
+	assert.Contains(t, text.String(), strconv.Itoa(pid))
+}
diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go
new file mode 100644
index 0000000..bd0df77
--- /dev/null
+++ b/internal/cli/migrate_test.go
@@ -0,0 +1,42 @@
+package cli
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/executor"
+	"github.com/block/pg-sprite/pkg/statement"
+	"github.com/block/pg-sprite/pkg/verdict"
+)
+
+func TestBudgetVerdict(t *testing.T) {
+	st, err := statement.ParseOne("ALTER TABLE billing.invoices ALTER COLUMN id TYPE bigint")
+	require.NoError(t, err)
+
+	t.Run("lock budget", func(t *testing.T) {
+		v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second})
+		assert.Equal(t, verdict.OutcomeRefused, v.Outcome)
+		assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason)
+		assert.Equal(t, verdict.CauseLockBudget, v.Cause)
+		assert.Equal(t, "billing.invoices", v.Table)
+		assert.NotEmpty(t, v.Detail)
+	})
+
+	t.Run("statement budget", func(t *testing.T) {
+		v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: 30 * time.Second})
+		assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason)
+		assert.Equal(t, verdict.CauseStatementBudget, v.Cause)
+		assert.NotEmpty(t, v.Detail)
+	})
+
+	t.Run("unknown cause falls back to the error text", func(t *testing.T) {
+		budgetErr := &executor.BudgetError{Budget: time.Second}
+		v := budgetVerdict(st, budgetErr)
+		assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason)
+		assert.Equal(t, verdict.CauseNone, v.Cause)
+		assert.Equal(t, budgetErr.Error(), v.Detail)
+	})
+}
diff --git a/internal/cli/status.go b/internal/cli/status.go
new file mode 100644
index 0000000..21b8106
--- /dev/null
+++ b/internal/cli/status.go
@@ -0,0 +1,102 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/block/pg-sprite/pkg/dbconn"
+)
+
+// session is one live pg-sprite backend from pg_stat_activity.
+type session struct {
+	PID        int    `json:"pid"`
+	State      string `json:"state"`
+	WaitEvent  string `json:"wait_event"`
+	RunningFor string `json:"running_for"`
+	Query      string `json:"query"`
+}
+
+// run reports the engine's live database sessions. Phase 1 has no durable
+// schema-change state — a change either committed within its budgets or was
+// refused — so status is a view over pg_stat_activity for pg-sprite sessions.
+func (c *StatusCmd) run(ctx context.Context, out io.Writer) error {
+	pool, err := dbconn.NewPool(ctx, c.Config())
+	if err != nil {
+		return err
+	}
+	defer pool.Close()
+
+	sessions, err := querySessions(ctx, pool)
+	if err != nil {
+		return err
+	}
+	if c.JSON {
+		b, err := json.MarshalIndent(sessions, "", "  ")
+		if err != nil {
+			return fmt.Errorf("encode status: %w", err)
+		}
+		if _, err := fmt.Fprintln(out, string(b)); err != nil {
+			return fmt.Errorf("write status: %w", err)
+		}
+		return nil
+	}
+	return renderSessions(out, sessions)
+}
+
+// querySessions lists the live pg-sprite backends other than the one running
+// the status query itself. pg_stat_activity nulls out state and query for
+// other roles' backends unless the viewer has pg_read_all_stats — exactly
+// the read-only-operator-checking-on-the-engine-role shape — so those
+// columns are coalesced instead of crashing the scan.
+func querySessions(ctx context.Context, pool *pgxpool.Pool) ([]session, error) {
+	rows, err := pool.Query(ctx, `
+		SELECT pid,
+		       COALESCE(state, '-'),
+		       COALESCE(wait_event_type || '/' || wait_event, '-'),
+		       COALESCE(now() - query_start, '0'::interval)::text,
+		       COALESCE(left(query, 80), '<insufficient privilege>')
+		FROM pg_stat_activity
+		WHERE application_name = 'pg-sprite' AND pid <> pg_backend_pid()
+		ORDER BY query_start`)
+	if err != nil {
+		return nil, fmt.Errorf("query pg_stat_activity: %w", err)
+	}
+	defer rows.Close()
+
+	sessions := []session{}
+	for rows.Next() {
+		var s session
+		if err := rows.Scan(&s.PID, &s.State, &s.WaitEvent, &s.RunningFor, &s.Query); err != nil {
+			return nil, fmt.Errorf("scan pg_stat_activity row: %w", err)
+		}
+		sessions = append(sessions, s)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("read pg_stat_activity: %w", err)
+	}
+	return sessions, nil
+}
+
+// renderSessions writes the human-readable session listing.
+func renderSessions(out io.Writer, sessions []session) error {
+	if len(sessions) == 0 {
+		if _, err := fmt.Fprintln(out, "no active pg-sprite sessions (Phase 1 keeps no durable schema-change state: a change either committed within its budgets or was refused)"); err != nil {
+			return fmt.Errorf("write status: %w", err)
+		}
+		return nil
+	}
+	if _, err := fmt.Fprintln(out, "active pg-sprite sessions:"); err != nil {
+		return fmt.Errorf("write status: %w", err)
+	}
+	for _, s := range sessions {
+		if _, err := fmt.Fprintf(out, "  pid=%d state=%s wait=%s running_for=%s query=%q\n",
+			s.PID, s.State, s.WaitEvent, s.RunningFor, s.Query); err != nil {
+			return fmt.Errorf("write status: %w", err)
+		}
+	}
+	return nil
+}
diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go
new file mode 100644
index 0000000..911b350
--- /dev/null
+++ b/pkg/executor/optimistic.go
@@ -0,0 +1,190 @@
+// Package executor runs schema changes against the database. In Phase 1 it
+// holds the optimistic front door: attempt the change directly under a tight
+// lock_timeout and statement_timeout so it can only succeed if it is
+// effectively an instant / in-place change. A budget overrun cancels the
+// statement cleanly — nothing is executed — and surfaces as a typed
+// BudgetError the caller turns into a not-native-safe verdict.
+//
+// This is a safety-critical core package: see SAFETY.md. It never trusts the
+// caller's classification — its own protections are the budget, applied with
+// SET LOCAL inside the attempt's transaction regardless of session defaults,
+// and the statement binding: it accepts only a parsed statement.Statement
+// (exactly one statement by construction) whose target matches the
+// preflighted table (invariant ST-7).
+package executor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/statement"
+)
+
+// ErrInvariantViolation is the fail-closed error class for breaches of the
+// registry in docs/invariants.md (see SAFETY.md); the message names the
+// invariant ID. It is never a warning and never retried.
+var ErrInvariantViolation = errors.New("invariant violation")
+
+// SQLSTATE codes the attempt maps to budget outcomes. Postgres errors are
+// matched by SQLSTATE, never by message text.
+const (
+	sqlstateLockNotAvailable = "55P03" // lock_timeout expired
+	sqlstateQueryCanceled    = "57014" // statement_timeout expired
+)
+
+// BudgetCause says which budget the optimistic attempt exceeded.
+type BudgetCause int
+
+// The two budgets an attempt runs under.
+const (
+	// CauseLock means the lock was not granted within lock_timeout: the
+	// table is too contended for a blind attempt right now.
+	CauseLock BudgetCause = iota + 1
+	// CauseStatement means the statement ran past statement_timeout: the
+	// change is doing real work (a rewrite), not an in-place catalog change.
+	CauseStatement
+)
+
+// String returns the human-readable budget name.
+func (c BudgetCause) String() string {
+	switch c {
+	case CauseLock:
+		return "lock budget"
+	case CauseStatement:
+		return "statement budget"
+	default:
+		return "unknown budget"
+	}
+}
+
+// BudgetError reports that the optimistic attempt exceeded one of its budgets
+// and was cancelled cleanly: the statement did not execute and the
+// transaction rolled back. It is a refusal input, not an operational failure.
+type BudgetError struct {
+	// Cause is the budget that was exceeded.
+	Cause BudgetCause
+	// Budget is the configured limit for that cause.
+	Budget time.Duration
+}
+
+// Error implements the error interface.
+func (e *BudgetError) Error() string {
+	return fmt.Sprintf("optimistic attempt exceeded its %s (%s) and was cancelled", e.Cause, e.Budget)
+}
+
+// Budget bounds one optimistic attempt. Both limits must be at least
+// minBudget: an unbounded attempt is exactly the stall the front door exists
+// to prevent.
+type Budget struct {
+	// LockTimeout bounds how long the attempt may wait in the lock queue.
+	LockTimeout time.Duration
+	// StatementTimeout bounds how long the statement may run once started.
+	StatementTimeout time.Duration
+}
+
+// minBudget is the smallest expressible budget: budgets are applied to
+// PostgreSQL in whole milliseconds, and a sub-millisecond value would
+// truncate to 0 — which disables the corresponding limit entirely.
+const minBudget = time.Millisecond
+
+// validate rejects budgets that would leave the attempt unbounded. Rejecting
+// a sub-millisecond budget is more fail-closed than silently rounding it up.
+func (b Budget) validate() error {
+	// INV: LK-2 — the attempt is bounded by construction; a zero, negative,
+	// or sub-millisecond timeout would disable the corresponding PostgreSQL
+	// limit after millisecond truncation.
+	if b.LockTimeout < minBudget {
+		return fmt.Errorf("lock budget must be at least %s, got %s", minBudget, b.LockTimeout)
+	}
+	if b.StatementTimeout < minBudget {
+		return fmt.Errorf("statement budget must be at least %s, got %s", minBudget, b.StatementTimeout)
+	}
+	return nil
+}
+
+// AttemptNative runs st once, directly, inside a transaction bounded by b.
+// The table must have passed preflight and the statement must target it —
+// both proofs make the unsafe call unrepresentable: a statement.Statement
+// can only come from ParseOne (exactly one statement, parsed by the real
+// grammar), and a target mismatch is refused before anything executes, so a
+// proof for one table cannot smuggle SQL against another. On success the
+// change is committed: it was effectively instant. If a budget is exceeded
+// the statement is cancelled by the server, the transaction rolls back, and
+// a *BudgetError is returned. Any other failure is surfaced as an
+// operational error.
+func AttemptNative(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, st statement.Statement, b Budget) error {
+	if err := b.validate(); err != nil {
+		return err
+	}
+	// INV: ST-7 — the executor runs exactly the statement that was gated,
+	// and only against the table the preflight proof verified.
+	if st.Table() == "" || st.Schema() != pt.Schema() || st.Table() != pt.Table() {
+		return fmt.Errorf("%w: ST-7: statement targets %q but preflight verified %q",
+			ErrInvariantViolation, qualifiedName(st.Schema(), st.Table()), qualifiedName(pt.Schema(), pt.Table()))
+	}
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin optimistic attempt: %w", err)
+	}
+	defer func() {
+		// Redundant safety closer: after a successful Commit this returns
+		// the guaranteed ErrTxClosed; on a failure path a rollback error
+		// only means the connection died, and the server aborts the
+		// transaction with its session either way.
+		_ = tx.Rollback(context.WithoutCancel(ctx))
+	}()
+
+	// INV: LK-2 — budgets are applied inside this transaction regardless of
+	// the session defaults, so the attempt cannot outlive them even on a
+	// misconfigured pool. A bare integer is milliseconds to PostgreSQL;
+	// SET LOCAL cannot use bind parameters.
+	setBudgets := "SET LOCAL lock_timeout = " + strconv.FormatInt(b.LockTimeout.Milliseconds(), 10) +
+		"; SET LOCAL statement_timeout = " + strconv.FormatInt(b.StatementTimeout.Milliseconds(), 10)
+	if _, err := tx.Exec(ctx, setBudgets); err != nil {
+		return fmt.Errorf("set attempt budgets: %w", err)
+	}
+
+	if _, err := tx.Exec(ctx, st.SQL()); err != nil {
+		if budgetErr := asBudgetError(err, b); budgetErr != nil {
+			return budgetErr
+		}
+		return fmt.Errorf("optimistic attempt: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit optimistic attempt: %w", err)
+	}
+	return nil
+}
+
+// qualifiedName renders schema.table for error messages, omitting the dot
+// when the name is unqualified.
+func qualifiedName(schema, table string) string {
+	if schema == "" {
+		return table
+	}
+	return schema + "." + table
+}
+
+// asBudgetError maps a PostgreSQL error to the budget it exceeded, or nil
+// when the error is not a budget overrun.
+func asBudgetError(err error, b Budget) *BudgetError {
+	var pgErr *pgconn.PgError
+	if !errors.As(err, &pgErr) {
+		return nil
+	}
+	switch pgErr.Code {
+	case sqlstateLockNotAvailable:
+		return &BudgetError{Cause: CauseLock, Budget: b.LockTimeout}
+	case sqlstateQueryCanceled:
+		return &BudgetError{Cause: CauseStatement, Budget: b.StatementTimeout}
+	default:
+		return nil
+	}
+}
diff --git a/pkg/executor/optimistic_integration_test.go b/pkg/executor/optimistic_integration_test.go
new file mode 100644
index 0000000..e277b80
--- /dev/null
+++ b/pkg/executor/optimistic_integration_test.go
@@ -0,0 +1,212 @@
+package executor_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"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/executor"
+	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/statement"
+)
+
+// budget is generous enough that an instant catalog change always fits and
+// tight enough that a blocked or rewriting attempt is cancelled quickly.
+var budget = executor.Budget{LockTimeout: 500 * time.Millisecond, StatementTimeout: 2 * time.Second}
+
+func newPool(t *testing.T) (*pgxpool.Pool, string) {
+	t.Helper()
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	t.Cleanup(pool.Close)
+	return pool, testutil.NewSchema(t, pool)
+}
+
+func mustPreflight(t *testing.T, pool *pgxpool.Pool, schema, table string) preflight.PreflightedTable {
+	t.Helper()
+	pt, err := preflight.CheckTable(t.Context(), pool, schema, table, 1<<30)
+	require.NoError(t, err)
+	return pt
+}
+
+func mustParse(t *testing.T, sql string) statement.Statement {
+	t.Helper()
+	st, err := statement.ParseOne(sql)
+	require.NoError(t, err)
+	return st
+}
+
+func columnType(t *testing.T, pool *pgxpool.Pool, schema, table, column string) string {
+	t.Helper()
+	var typ string
+	err := pool.QueryRow(t.Context(),
+		`SELECT data_type FROM information_schema.columns
+		 WHERE table_schema = $1 AND table_name = $2 AND column_name = $3`,
+		schema, table, column).Scan(&typ)
+	require.NoError(t, err)
+	return typ
+}
+
+func TestAttemptNativeCommitsInstantChange(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int NOT NULL DEFAULT 0", schema))
+	require.NoError(t, executor.AttemptNative(t.Context(), pool, pt, st, budget))
+
+	assert.Equal(t, "integer", columnType(t, pool, schema, "t", "age"), "the committed change must be visible")
+}
+
+func TestAttemptNativeCancelsWhenLockBlocked(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+
+	// A second session holds ACCESS EXCLUSIVE for the whole test, so the
+	// attempt can never be granted its lock.
+	blocker, err := pool.Begin(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context())))
+	})
+	_, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema))
+	require.NoError(t, err)
+
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+	err = executor.AttemptNative(t.Context(), pool, pt, st, budget)
+
+	var budgetErr *executor.BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, executor.CauseLock, budgetErr.Cause)
+	assert.Equal(t, budget.LockTimeout, budgetErr.Budget)
+}
+
+func TestAttemptNativeCancelsRewriteAndLeavesTableUnchanged(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema))
+	require.NoError(t, err)
+	// Enough rows that a full table rewrite cannot finish inside a
+	// millisecond-scale statement budget.
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(
+		"INSERT INTO %s.t SELECT g, repeat('x', 100) FROM generate_series(1, 300000) g", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+
+	// int -> bigint forces a full table rewrite under ACCESS EXCLUSIVE.
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema))
+	tight := executor.Budget{LockTimeout: budget.LockTimeout, StatementTimeout: 50 * time.Millisecond}
+	err = executor.AttemptNative(t.Context(), pool, pt, st, tight)
+
+	var budgetErr *executor.BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, executor.CauseStatement, budgetErr.Cause)
+
+	// The cancelled attempt must leave schema and data untouched.
+	assert.Equal(t, "integer", columnType(t, pool, schema, "t", "id"))
+	var count int
+	require.NoError(t, pool.QueryRow(t.Context(),
+		fmt.Sprintf("SELECT count(*) FROM %s.t", schema)).Scan(&count))
+	assert.Equal(t, 300000, count)
+}
+
+func TestAttemptNativeSurfacesOperationalErrors(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+
+	// Dropping a column that does not exist is a plain SQL error, not a
+	// budget overrun.
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t DROP COLUMN nope", schema))
+	err = executor.AttemptNative(t.Context(), pool, pt, st, budget)
+	require.Error(t, err)
+	var budgetErr *executor.BudgetError
+	assert.NotErrorAs(t, err, &budgetErr)
+}
+
+// Sub-millisecond budgets are as unbounded as zero ones: they truncate to
+// PostgreSQL's 0ms, which disables the corresponding limit entirely.
+func TestAttemptNativeRejectsUnboundedBudgets(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema))
+	require.NoError(t, err)
+	pt := mustPreflight(t, pool, schema, "t")
+
+	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
+	unbounded := map[string]executor.Budget{
+		"zero lock":                 {LockTimeout: 0, StatementTimeout: time.Second},
+		"zero statement":            {LockTimeout: time.Second, StatementTimeout: 0},
+		"sub-millisecond lock":      {LockTimeout: 500 * time.Microsecond, StatementTimeout: time.Second},
+		"sub-millisecond statement": {LockTimeout: time.Second, StatementTimeout: 999 * time.Microsecond},
+	}
+	for name, b := range unbounded {
+		t.Run(name, func(t *testing.T) {
+			require.Error(t, executor.AttemptNative(t.Context(), pool, pt, st, b))
+		})
+	}
+
+	// The smallest representable budget is valid and does not serialize to 0:
+	// a 1ms lock timeout must still cancel a blocked attempt rather than
+	// disabling the limit.
+	blocker, err := pool.Begin(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context())))
+	})
+	_, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema))
+	require.NoError(t, err)
+	err = executor.AttemptNative(t.Context(), pool, pt, st, executor.Budget{LockTimeout: time.Millisecond, StatementTimeout: time.Second})
+	var budgetErr *executor.BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, executor.CauseLock, budgetErr.Cause)
+}
+
+// INV: ST-7 — a preflight proof for one table can never execute a statement
+// against another, and a statement without a table target never executes.
+func TestAttemptNativeRefusesTargetMismatch(t *testing.T) {
+	pool, schema := newPool(t)
+	for _, ddl := range []string{
+		fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema),
+		fmt.Sprintf("CREATE TABLE %s.victim (id int PRIMARY KEY)", schema),
+	} {
+		_, err := pool.Exec(t.Context(), ddl)
+		require.NoError(t, err)
+	}
+	pt := mustPreflight(t, pool, schema, "t")
+
+	t.Run("statement targets a different table", func(t *testing.T) {
+		st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.victim ADD COLUMN a int", schema))
+		err := executor.AttemptNative(t.Context(), pool, pt, st, budget)
+		require.ErrorIs(t, err, executor.ErrInvariantViolation)
+
+		var n int
+		require.NoError(t, pool.QueryRow(t.Context(),
+			`SELECT count(*) FROM information_schema.columns
+			 WHERE table_schema = $1 AND table_name = 'victim' AND column_name = 'a'`, schema).Scan(&n))
+		assert.Zero(t, n, "the refused statement must never reach the database")
+	})
+
+	t.Run("unqualified statement does not match a qualified proof", func(t *testing.T) {
+		// Fail-closed: the proof verified schema.t, the statement names a
+		// bare t that search_path could resolve elsewhere.
+		st := mustParse(t, "ALTER TABLE t ADD COLUMN a int")
+		err := executor.AttemptNative(t.Context(), pool, pt, st, budget)
+		require.ErrorIs(t, err, executor.ErrInvariantViolation)
+	})
+
+	t.Run("statement without a table target", func(t *testing.T) {
+		st := mustParse(t, "CREATE TABLE elsewhere (id int)")
+		err := executor.AttemptNative(t.Context(), pool, pt, st, budget)
+		require.ErrorIs(t, err, executor.ErrInvariantViolation)
+	})
+}
diff --git a/pkg/preflight/preflight.go b/pkg/preflight/preflight.go
new file mode 100644
index 0000000..59bed7c
--- /dev/null
+++ b/pkg/preflight/preflight.go
@@ -0,0 +1,130 @@
+// Package preflight verifies preconditions before the engine writes anything
+// (invariant ST-6). In Phase 1 that is the table-size guard in front of the
+// optimistic attempt: a cancelled rewrite attempt is not a free probe — it
+// holds ACCESS EXCLUSIVE and does real rewrite work for the full statement
+// budget — so above a size threshold the attempt is skipped entirely.
+//
+// This is a safety-critical core package: see SAFETY.md. It returns proof
+// types with package-private constructors; dangerous downstream APIs accept
+// only the proof.
+package preflight
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// ErrTableNotFound is returned when the target table does not exist (or is
+// not visible with the session's search_path).
+var ErrTableNotFound = errors.New("table not found")
+
+// ErrNotTable is returned when the target exists but is not an ordinary or
+// partitioned table (e.g. a view or foreign table).
+var ErrNotTable = errors.New("not an ordinary or partitioned table")
+
+// SizeError reports that the table exceeds the configured size threshold, so
+// the optimistic attempt must be skipped. It is a refusal input, not an
+// operational failure.
+type SizeError struct {
+	// TotalBytes is the table's measured on-disk size (all partitions,
+	// including indexes and TOAST).
+	TotalBytes int64
+	// LimitBytes is the threshold that was exceeded.
+	LimitBytes int64
+}
+
+// Error implements the error interface.
+func (e *SizeError) Error() string {
+	return fmt.Sprintf("table size %d bytes exceeds the %d-byte threshold for an optimistic attempt", e.TotalBytes, e.LimitBytes)
+}
+
+// PreflightedTable proves the target table exists, is a table, and is under
+// the size threshold for an optimistic attempt. It can only be constructed by
+// CheckTable in this package.
+type PreflightedTable struct {
+	schema     string
+	table      string
+	totalBytes int64
+	relTuples  float64
+}
+
+// Schema returns the schema qualification the check ran with (empty when the
+// lookup used the session search_path).
+func (t PreflightedTable) Schema() string { return t.schema }
+
+// Table returns the verified table name.
+func (t PreflightedTable) Table() string { return t.table }
+
+// TotalBytes returns the measured on-disk size across all partitions,
+// including indexes and TOAST.
+func (t PreflightedTable) TotalBytes() int64 { return t.totalBytes }
+
+// RelTuples returns the planner's row estimate (-1 when the table has never
+// been vacuumed or analyzed). Reporting only — the size guard's authority is
+// bytes on disk.
+func (t PreflightedTable) RelTuples() float64 { return t.relTuples }
+
+// CheckTable verifies that schema.table (search_path when schema is empty)
+// exists, is an ordinary or partitioned table, and is at most limitBytes on
+// disk. Above the limit it returns a *SizeError; on success it returns the
+// PreflightedTable proof.
+func CheckTable(ctx context.Context, pool *pgxpool.Pool, schema, table string, limitBytes int64) (PreflightedTable, error) {
+	if limitBytes <= 0 {
+		return PreflightedTable{}, fmt.Errorf("size limit must be positive, got %d", limitBytes)
+	}
+	// INV: ST-6 — size facts are measured on-disk bytes
+	// (pg_total_relation_size: heap, indexes, and TOAST — the rewrite the
+	// guard fears rebuilds every index under the same ACCESS EXCLUSIVE
+	// lock, so an index-heavy table must not sail under the threshold),
+	// summed over pg_partition_tree so a partitioned parent (whose own
+	// relation is 0 bytes) cannot fail open. Stale planner statistics
+	// (relpages) are never the guard's authority.
+	// The table's own size plus every descendant in its partition tree:
+	// pg_partition_tree returns no rows for a plain table (its own
+	// pg_total_relation_size carries the total) and the parent's own
+	// relation is 0 bytes for a partitioned table (the descendants carry
+	// the total).
+	const q = `
+		SELECT c.relkind::text,
+		       pg_total_relation_size(c.oid) +
+		       (SELECT COALESCE(sum(pg_total_relation_size(p.relid)), 0)
+		          FROM pg_partition_tree(c.oid) p
+		         WHERE p.relid <> c.oid),
+		       c.reltuples
+		FROM pg_class c
+		WHERE c.oid = to_regclass(
+			CASE WHEN $1 = '' THEN quote_ident($2)
+			     ELSE quote_ident($1) || '.' || quote_ident($2) END)`
+	var relkind string
+	var totalBytes int64
+	var relTuples float64
+	err := pool.QueryRow(ctx, q, schema, table).Scan(&relkind, &totalBytes, &relTuples)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return PreflightedTable{}, fmt.Errorf("%w: %s", ErrTableNotFound, qualifiedName(schema, table))
+	}
+	if err != nil {
+		return PreflightedTable{}, fmt.Errorf("look up table %s: %w", qualifiedName(schema, table), err)
+	}
+	// relkind 'r' is an ordinary table, 'p' a partitioned parent; anything
+	// else (view, matview, foreign table, sequence) is refused fail-closed.
+	if relkind != "r" && relkind != "p" {
+		return PreflightedTable{}, fmt.Errorf("%w: %s has relkind %q", ErrNotTable, qualifiedName(schema, table), relkind)
+	}
+	if totalBytes > limitBytes {
+		return PreflightedTable{}, &SizeError{TotalBytes: totalBytes, LimitBytes: limitBytes}
+	}
+	return PreflightedTable{schema: schema, table: table, totalBytes: totalBytes, relTuples: relTuples}, nil
+}
+
+// qualifiedName renders schema.table for error messages, omitting the dot
+// when the name is unqualified.
+func qualifiedName(schema, table string) string {
+	if schema == "" {
+		return table
+	}
+	return schema + "." + table
+}
diff --git a/pkg/preflight/preflight_integration_test.go b/pkg/preflight/preflight_integration_test.go
new file mode 100644
index 0000000..8e444e2
--- /dev/null
+++ b/pkg/preflight/preflight_integration_test.go
@@ -0,0 +1,158 @@
+package preflight_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/preflight"
+)
+
+func TestCheckTableUnderLimit(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.small (id int PRIMARY KEY, v text)", schema))
+	require.NoError(t, err)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("INSERT INTO %s.small SELECT g, 'v' FROM generate_series(1, 100) g", schema))
+	require.NoError(t, err)
+
+	pt, err := preflight.CheckTable(t.Context(), pool, schema, "small", 1<<30)
+	require.NoError(t, err)
+	assert.Equal(t, schema, pt.Schema())
+	assert.Equal(t, "small", pt.Table())
+	assert.Positive(t, pt.TotalBytes(), "a populated table must report a nonzero on-disk size")
+}
+
+func TestCheckTableOverLimit(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.big (id int PRIMARY KEY, v text)", schema))
+	require.NoError(t, err)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("INSERT INTO %s.big SELECT g, repeat('x', 100) FROM generate_series(1, 10000) g", schema))
+	require.NoError(t, err)
+
+	_, err = preflight.CheckTable(t.Context(), pool, schema, "big", 1)
+	var sizeErr *preflight.SizeError
+	require.ErrorAs(t, err, &sizeErr)
+	assert.Positive(t, sizeErr.TotalBytes)
+	assert.Equal(t, int64(1), sizeErr.LimitBytes)
+}
+
+// A partitioned parent's own relation is 0 bytes on disk; the guard must sum
+// the partitions so a huge partitioned table cannot slip under the limit.
+func TestCheckTableSumsPartitions(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	for _, ddl := range []string{
+		fmt.Sprintf("CREATE TABLE %s.parted (id int, v text) PARTITION BY RANGE (id)", schema),
+		fmt.Sprintf("CREATE TABLE %s.parted_lo PARTITION OF %s.parted FOR VALUES FROM (0) TO (5000)", schema, schema),
+		fmt.Sprintf("CREATE TABLE %s.parted_hi PARTITION OF %s.parted FOR VALUES FROM (5000) TO (10001)", schema, schema),
+		fmt.Sprintf("INSERT INTO %s.parted SELECT g, repeat('x', 100) FROM generate_series(0, 10000) g", schema),
+	} {
+		_, err = pool.Exec(t.Context(), ddl)
+		require.NoError(t, err)
+	}
+
+	pt, err := preflight.CheckTable(t.Context(), pool, schema, "parted", 1<<30)
+	require.NoError(t, err)
+	assert.Positive(t, pt.TotalBytes(), "the guard must see the partitions' bytes, not the parent's zero")
+
+	_, err = preflight.CheckTable(t.Context(), pool, schema, "parted", 1)
+	var sizeErr *preflight.SizeError
+	require.ErrorAs(t, err, &sizeErr, "a populated partitioned table must exceed a 1-byte limit")
+}
+
+// The rewrite the guard fears rebuilds every index under the same ACCESS
+// EXCLUSIVE lock, so the measured footprint must include index bytes: a
+// heavily indexed table exceeds a threshold its heap alone would fit under.
+func TestCheckTableCountsIndexBytes(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, a text, b text)", schema))
+	require.NoError(t, err)
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(
+		"INSERT INTO %s.t SELECT g, md5(g::text), md5((g+1)::text) FROM generate_series(1, 20000) g", schema))
+	require.NoError(t, err)
+	for _, idx := range []string{
+		fmt.Sprintf("CREATE INDEX ON %s.t (a, b)", schema),
+		fmt.Sprintf("CREATE INDEX ON %s.t (b, a)", schema),
+	} {
+		_, err = pool.Exec(t.Context(), idx)
+		require.NoError(t, err)
+	}
+
+	var heap, total int64
+	require.NoError(t, pool.QueryRow(t.Context(),
+		"SELECT pg_table_size($1::regclass), pg_total_relation_size($1::regclass)",
+		schema+".t").Scan(&heap, &total))
+	require.Greater(t, total, heap, "the fixture's indexes must add measurable bytes")
+
+	// A limit the heap alone would fit under must still refuse, and the
+	// reported size must be the full footprint.
+	_, err = preflight.CheckTable(t.Context(), pool, schema, "t", heap)
+	var sizeErr *preflight.SizeError
+	require.ErrorAs(t, err, &sizeErr)
+	assert.Equal(t, total, sizeErr.TotalBytes)
+}
+
+func TestCheckTableMissingTable(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = preflight.CheckTable(t.Context(), pool, schema, "nope", 1<<30)
+	require.ErrorIs(t, err, preflight.ErrTableNotFound)
+}
+
+func TestCheckTableRefusesNonTable(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE VIEW %s.v AS SELECT 1 AS one", schema))
+	require.NoError(t, err)
+
+	_, err = preflight.CheckTable(t.Context(), pool, schema, "v", 1<<30)
+	require.ErrorIs(t, err, preflight.ErrNotTable)
+}
+
+func TestCheckTableQuotedIdentifiers(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+	schema := testutil.NewSchema(t, pool)
+
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(`CREATE TABLE %s."Order Items" (id int)`, schema))
+	require.NoError(t, err)
+
+	pt, err := preflight.CheckTable(t.Context(), pool, schema, "Order Items", 1<<30)
+	require.NoError(t, err)
+	assert.Equal(t, "Order Items", pt.Table())
+}
+
+func TestCheckTableRejectsNonPositiveLimit(t *testing.T) {
+	pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t)})
+	require.NoError(t, err)
+	defer pool.Close()
+
+	_, err = preflight.CheckTable(t.Context(), pool, "", "whatever", 0)
+	require.Error(t, err)
+}
diff --git a/pkg/statement/statement.go b/pkg/statement/statement.go
new file mode 100644
index 0000000..6e1fed4
--- /dev/null
+++ b/pkg/statement/statement.go
@@ -0,0 +1,156 @@
+// Package statement parses SQL through the real PostgreSQL grammar
+// (wasilibs/go-pgquery, Wasm libpg_query) and reports the facts the engine's
+// front door needs. In Phase 1 that is a statement-type gate only: which kind
+// of statement this is and, for ALTER TABLE, which table it targets. No
+// schema model, no classification.
+package statement
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	pganalyze "github.com/pganalyze/pg_query_go/v6"
+	pgquery "github.com/wasilibs/go-pgquery"
+)
+
+// Kind is the statement-type bucket the Phase 1 gate branches on.
+type Kind int
+
+// The kinds the gate distinguishes. Everything the engine does not recognize
+// as one of the named kinds is KindOther and is refused by the front door.
+const (
+	KindOther Kind = iota
+	KindAlterTable
+	KindCreateIndex
+	KindDropIndex
+	KindReindex
+)
+
+// String returns the human-readable name of the kind.
+func (k Kind) String() string {
+	switch k {
+	case KindAlterTable:
+		return "ALTER TABLE"
+	case KindCreateIndex:
+		return "CREATE INDEX"
+	case KindDropIndex:
+		return "DROP INDEX"
+	case KindReindex:
+		return "REINDEX"
+	default:
+		return "other"
+	}
+}
+
+// Statement is one parsed SQL statement plus the facts the gate needs. It can
+// only be constructed by ParseOne, so holding one proves the SQL parsed as
+// exactly one statement through the PostgreSQL grammar — the proof the
+// executor requires before running anything (invariant ST-7).
+type Statement struct {
+	sql        string
+	kind       Kind
+	schema     string
+	table      string
+	concurrent bool
+}
+
+// SQL returns the original statement text as submitted.
+func (s Statement) SQL() string { return s.sql }
+
+// Kind returns the statement-type bucket.
+func (s Statement) Kind() Kind { return s.kind }
+
+// Schema returns the target table's schema qualification for ALTER TABLE
+// statements; empty when the statement was unqualified (search_path resolves
+// it) or when the kind has no single table target.
+func (s Statement) Schema() string { return s.schema }
+
+// Table returns the target table name for ALTER TABLE statements; empty for
+// other kinds.
+func (s Statement) Table() string { return s.table }
+
+// Concurrent reports whether an index statement used its CONCURRENTLY form.
+// It is always false for non-index kinds.
+func (s Statement) Concurrent() bool { return s.concurrent }
+
+// ErrNotOneStatement is returned by ParseOne when the input does not contain
+// exactly one SQL statement.
+var ErrNotOneStatement = errors.New("input must contain exactly one SQL statement")
+
+// ParseOne parses sql with the PostgreSQL grammar and requires exactly one
+// statement. A parse failure is surfaced to the caller, never guessed around.
+func ParseOne(sql string) (Statement, error) {
+	tree, err := pgquery.Parse(sql)
+	if err != nil {
+		return Statement{}, fmt.Errorf("parse statement: %w", err)
+	}
+	if n := len(tree.GetStmts()); n != 1 {
+		return Statement{}, fmt.Errorf("%w: got %d", ErrNotOneStatement, n)
+	}
+	st := Statement{sql: sql}
+	node := tree.GetStmts()[0].GetStmt()
+	switch {
+	case node.GetAlterTableStmt() != nil:
+		alter := node.GetAlterTableStmt()
+		// ALTER INDEX (and ALTER VIEW etc.) also parse as AlterTableStmt;
+		// only a true table target is KindAlterTable.
+		if alter.GetObjtype() != pganalyze.ObjectType_OBJECT_TABLE {
+			return st, nil
+		}
+		st.kind = KindAlterTable
+		st.schema = alter.GetRelation().GetSchemaname()
+		st.table = alter.GetRelation().GetRelname()
+	case node.GetRenameStmt() != nil:
+		// ALTER TABLE ... RENAME TO / RENAME COLUMN / RENAME CONSTRAINT
+		// parse as RenameStmt, not AlterTableStmt. Only table-targeted
+		// renames are KindAlterTable: RENAME TO carries OBJECT_TABLE as the
+		// rename type, RENAME COLUMN carries it as the relation type, and
+		// RENAME CONSTRAINT carries the table-specific OBJECT_TABCONSTRAINT.
+		// ALTER INDEX/VIEW ... RENAME carry their own object types and stay
+		// KindOther.
+		ren := node.GetRenameStmt()
+		if ren.GetRenameType() != pganalyze.ObjectType_OBJECT_TABLE &&
+			ren.GetRenameType() != pganalyze.ObjectType_OBJECT_TABCONSTRAINT &&
+			ren.GetRelationType() != pganalyze.ObjectType_OBJECT_TABLE {
+			return st, nil
+		}
+		st.kind = KindAlterTable
+		st.schema = ren.GetRelation().GetSchemaname()
+		st.table = ren.GetRelation().GetRelname()
+	case node.GetAlterObjectSchemaStmt() != nil:
+		// ALTER TABLE ... SET SCHEMA parses as AlterObjectSchemaStmt; only
+		// the table-targeted form is KindAlterTable.
+		move := node.GetAlterObjectSchemaStmt()
+		if move.GetObjectType() != pganalyze.ObjectType_OBJECT_TABLE {
+			return st, nil
+		}
+		st.kind = KindAlterTable
+		st.schema = move.GetRelation().GetSchemaname()
+		st.table = move.GetRelation().GetRelname()
+	case node.GetIndexStmt() != nil:
+		st.kind = KindCreateIndex
+		st.concurrent = node.GetIndexStmt().GetConcurrent()
+	case node.GetDropStmt() != nil:
+		if node.GetDropStmt().GetRemoveType() == pganalyze.ObjectType_OBJECT_INDEX {
+			st.kind = KindDropIndex
+			st.concurrent = node.GetDropStmt().GetConcurrent()
+		}
+	case node.GetReindexStmt() != nil:
+		st.kind = KindReindex
+		st.concurrent = reindexConcurrently(node.GetReindexStmt())
+	}
+	return st, nil
+}
+
+// reindexConcurrently reports whether a REINDEX statement used its
+// CONCURRENTLY form, which the grammar carries as a generic option rather
+// than a dedicated field.
+func reindexConcurrently(stmt *pganalyze.ReindexStmt) bool {
+	for _, p := range stmt.GetParams() {
+		if strings.EqualFold(p.GetDefElem().GetDefname(), "concurrently") {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/statement/statement_test.go b/pkg/statement/statement_test.go
new file mode 100644
index 0000000..f55f826
--- /dev/null
+++ b/pkg/statement/statement_test.go
@@ -0,0 +1,177 @@
+package statement
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseOneKinds(t *testing.T) {
+	tests := []struct {
+		name string
+		sql  string
+		want Statement
+	}{
+		{
+			name: "alter table unqualified",
+			sql:  "ALTER TABLE users ADD COLUMN age int",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table schema-qualified",
+			sql:  "ALTER TABLE billing.invoices DROP COLUMN note",
+			want: Statement{kind: KindAlterTable, schema: "billing", table: "invoices"},
+		},
+		{
+			name: "alter table quoted mixed-case identifier",
+			sql:  `ALTER TABLE "Order Items" ADD COLUMN qty int`,
+			want: Statement{kind: KindAlterTable, table: "Order Items"},
+		},
+		{
+			name: "alter table if exists",
+			sql:  "ALTER TABLE IF EXISTS users ADD COLUMN age int",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table rename to parses as RenameStmt but is a table target",
+			sql:  "ALTER TABLE users RENAME TO users_old",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table rename column",
+			sql:  "ALTER TABLE users RENAME COLUMN a TO b",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table rename constraint",
+			sql:  "ALTER TABLE users RENAME CONSTRAINT users_pk TO users_pkey",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table rename schema-qualified",
+			sql:  "ALTER TABLE billing.users RENAME COLUMN a TO b",
+			want: Statement{kind: KindAlterTable, schema: "billing", table: "users"},
+		},
+		{
+			name: "alter table set schema parses as AlterObjectSchemaStmt but is a table target",
+			sql:  "ALTER TABLE users SET SCHEMA archive",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter table owner to",
+			sql:  "ALTER TABLE users OWNER TO app_owner",
+			want: Statement{kind: KindAlterTable, table: "users"},
+		},
+		{
+			name: "alter view rename is not a table target",
+			sql:  "ALTER VIEW v RENAME TO w",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "alter index rename is not a table target",
+			sql:  "ALTER INDEX i RENAME TO j",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "alter sequence set schema is not a table target",
+			sql:  "ALTER SEQUENCE s SET SCHEMA archive",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "create index",
+			sql:  "CREATE INDEX idx_users_email ON users (email)",
+			want: Statement{kind: KindCreateIndex},
+		},
+		{
+			name: "create unique index concurrently is a concurrent index statement",
+			sql:  "CREATE UNIQUE INDEX CONCURRENTLY idx ON users (email)",
+			want: Statement{kind: KindCreateIndex, concurrent: true},
+		},
+		{
+			name: "drop index",
+			sql:  "DROP INDEX idx_users_email",
+			want: Statement{kind: KindDropIndex},
+		},
+		{
+			name: "drop index concurrently",
+			sql:  "DROP INDEX CONCURRENTLY idx_users_email",
+			want: Statement{kind: KindDropIndex, concurrent: true},
+		},
+		{
+			name: "reindex table",
+			sql:  "REINDEX TABLE users",
+			want: Statement{kind: KindReindex},
+		},
+		{
+			name: "reindex index",
+			sql:  "REINDEX INDEX idx_users_email",
+			want: Statement{kind: KindReindex},
+		},
+		{
+			name: "reindex table concurrently",
+			sql:  "REINDEX TABLE CONCURRENTLY users",
+			want: Statement{kind: KindReindex, concurrent: true},
+		},
+		{
+			name: "alter index parses as AlterTableStmt but is not a table target",
+			sql:  "ALTER INDEX idx_users_email SET (fillfactor = 90)",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "drop table is not a drop-index",
+			sql:  "DROP TABLE users",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "create table",
+			sql:  "CREATE TABLE t (id int)",
+			want: Statement{kind: KindOther},
+		},
+		{
+			name: "dml",
+			sql:  "UPDATE users SET age = 1",
+			want: Statement{kind: KindOther},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ParseOne(tt.sql)
+			require.NoError(t, err)
+			tt.want.sql = tt.sql
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestParseOneRejectsInvalidSQL(t *testing.T) {
+	_, err := ParseOne("ALTER TABEL users ADD COLUMN age int")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "parse statement")
+}
+
+func TestParseOneRejectsMultipleStatements(t *testing.T) {
+	_, err := ParseOne("ALTER TABLE a ADD COLUMN x int; ALTER TABLE b ADD COLUMN y int")
+	require.ErrorIs(t, err, ErrNotOneStatement)
+}
+
+// A second statement smuggled behind a legitimate ALTER never yields a
+// Statement at all — the executor only accepts what ParseOne constructs, so
+// multi-statement SQL is unrepresentable downstream (invariant ST-7).
+func TestParseOneRejectsSmuggledStatement(t *testing.T) {
+	_, err := ParseOne("ALTER TABLE t ADD COLUMN a int; DROP TABLE victim")
+	require.ErrorIs(t, err, ErrNotOneStatement)
+}
+
+func TestParseOneRejectsEmptyInput(t *testing.T) {
+	_, err := ParseOne("")
+	require.ErrorIs(t, err, ErrNotOneStatement)
+}
+
+func TestKindString(t *testing.T) {
+	assert.Equal(t, "ALTER TABLE", KindAlterTable.String())
+	assert.Equal(t, "CREATE INDEX", KindCreateIndex.String())
+	assert.Equal(t, "DROP INDEX", KindDropIndex.String())
+	assert.Equal(t, "REINDEX", KindReindex.String())
+	assert.Equal(t, "other", KindOther.String())
+}
diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go
new file mode 100644
index 0000000..676093e
--- /dev/null
+++ b/pkg/verdict/verdict.go
@@ -0,0 +1,124 @@
+// Package verdict is the engine's structured outcome contract: every migrate
+// invocation ends in exactly one verdict — executed natively, or refused with
+// a typed reason and, where one exists, a safer native idiom. Refusals use a
+// distinct exit code from operational errors. This type is the seam a future
+// orchestrator adapter maps onto SchemaBot's ExecutionModeBlocked.
+package verdict
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// ExitCodeRefused is the process exit code for a refusal verdict — distinct
+// from 1, which means an operational error (could not connect, bad flag, SQL
+// error). Automation branches on the difference.
+const ExitCodeRefused = 2
+
+// ErrRefused is the sentinel the CLI returns after printing a refusal
+// verdict, so the entry point can map it to ExitCodeRefused.
+var ErrRefused = errors.New("refused")
+
+// Outcome is what happened to the submitted change.
+type Outcome string
+
+// The two outcomes a migrate run can end in.
+const (
+	// OutcomeExecuted means the change ran and committed natively within
+	// its budgets.
+	OutcomeExecuted Outcome = "executed-natively"
+	// OutcomeRefused means the change was not executed; Reason says why.
+	OutcomeRefused Outcome = "refused"
+)
+
+// Reason is the typed cause of a refusal. Reasons are flat kebab-case
+// tokens — they are what automation switches on; prose belongs in Detail.
+type Reason string
+
+// The refusal reasons Phase 1 can emit.
+const (
+	// ReasonNone is the zero reason carried by an executed verdict.
+	ReasonNone Reason = ""
+	// ReasonUnsupportedStatement: only ALTER TABLE is supported.
+	ReasonUnsupportedStatement Reason = "unsupported-statement"
+	// ReasonIndexStatement: index maintenance has a native safe idiom
+	// (CONCURRENTLY) and is never attempted here.
+	ReasonIndexStatement Reason = "index-statement"
+	// ReasonTableTooLarge: the size guard skipped the optimistic attempt.
+	ReasonTableTooLarge Reason = "not-native-safe-table-too-large"
+	// ReasonBudgetExceeded: the optimistic attempt exceeded its lock or
+	// statement budget and was cancelled.
+	ReasonBudgetExceeded Reason = "not-native-safe-budget-exceeded"
+)
+
+// Cause narrows ReasonBudgetExceeded to the budget that was exceeded, so
+// automation can branch on which limit fired without parsing prose.
+type Cause string
+
+// The budget causes a refusal can carry.
+const (
+	// CauseNone is the zero cause for verdicts that are not budget refusals.
+	CauseNone Cause = ""
+	// CauseLockBudget: the lock was not granted within lock_timeout; nothing
+	// was executed.
+	CauseLockBudget Cause = "lock-budget"
+	// CauseStatementBudget: the statement ran past statement_timeout and was
+	// cancelled; the change needs a rewrite.
+	CauseStatementBudget Cause = "statement-budget"
+)
+
+// Verdict is the structured outcome of one migrate invocation.
+type Verdict struct {
+	// Outcome is what happened.
+	Outcome Outcome `json:"outcome"`
+	// Reason is the typed refusal cause; empty when executed.
+	Reason Reason `json:"reason,omitempty"`
+	// Cause narrows a budget refusal to the budget that fired; empty
+	// otherwise.
+	Cause Cause `json:"cause,omitempty"`
+	// Statement is the submitted SQL.
+	Statement string `json:"statement"`
+	// Table is the target table (schema-qualified when the statement was),
+	// when the statement has one.
+	Table string `json:"table,omitempty"`
+	// Detail is the human explanation: why refused, or what committed.
+	Detail string `json:"detail,omitempty"`
+	// SaferIdiom is a native alternative to the refused statement, when one
+	// exists (e.g. CREATE INDEX CONCURRENTLY, ADD CONSTRAINT ... NOT VALID).
+	SaferIdiom string `json:"safer_idiom,omitempty"`
+}
+
+// JSON renders the verdict as a single JSON object.
+func (v Verdict) JSON() (string, error) {
+	b, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		return "", fmt.Errorf("encode verdict: %w", err)
+	}
+	return string(b), nil
+}
+
+// String renders the verdict for humans.
+func (v Verdict) String() string {
+	var b strings.Builder
+	switch v.Outcome {
+	case OutcomeExecuted:
+		b.WriteString("executed natively")
+	case OutcomeRefused:
+		fmt.Fprintf(&b, "refused (%s)", v.Reason)
+	default:
+		fmt.Fprintf(&b, "unknown outcome %q", string(v.Outcome))
+	}
+	if v.Table != "" {
+		fmt.Fprintf(&b, "\n  table:     %s", v.Table)
+	}
+	fmt.Fprintf(&b, "\n  statement: %s", v.Statement)
+	if v.Detail != "" {
+		fmt.Fprintf(&b, "\n  detail:    %s", v.Detail)
+	}
+	if v.SaferIdiom != "" {
+		fmt.Fprintf(&b, "\n  safer:     %s", v.SaferIdiom)
+	}
+	return b.String()
+}
diff --git a/pkg/verdict/verdict_test.go b/pkg/verdict/verdict_test.go
new file mode 100644
index 0000000..73dacae
--- /dev/null
+++ b/pkg/verdict/verdict_test.go
@@ -0,0 +1,72 @@
+package verdict
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestJSONRoundTrip(t *testing.T) {
+	v := Verdict{
+		Outcome:    OutcomeRefused,
+		Reason:     ReasonBudgetExceeded,
+		Statement:  "ALTER TABLE t ALTER COLUMN id TYPE bigint",
+		Table:      "t",
+		Detail:     "the optimistic attempt exceeded its statement budget",
+		SaferIdiom: "ADD CONSTRAINT ... NOT VALID; VALIDATE CONSTRAINT",
+	}
+	s, err := v.JSON()
+	require.NoError(t, err)
+
+	var got Verdict
+	require.NoError(t, json.Unmarshal([]byte(s), &got))
+	assert.Equal(t, v, got)
+}
+
+func TestJSONOmitsEmptyOptionalFields(t *testing.T) {
+	s, err := Verdict{Outcome: OutcomeExecuted, Statement: "ALTER TABLE t ADD COLUMN x int"}.JSON()
+	require.NoError(t, err)
+	assert.NotContains(t, s, "reason")
+	assert.NotContains(t, s, "table")
+	assert.NotContains(t, s, "safer_idiom")
+}
+
+// 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(CauseLockBudget),
+		string(CauseStatementBudget),
+	} {
+		assert.Regexp(t, `^[a-z0-9]+(-[a-z0-9]+)*$`, tok)
+	}
+}
+
+func TestStringExecuted(t *testing.T) {
+	s := Verdict{
+		Outcome:   OutcomeExecuted,
+		Statement: "ALTER TABLE t ADD COLUMN x int",
+		Table:     "t",
+		Detail:    "committed within budget",
+	}.String()
+	assert.Contains(t, s, "executed natively")
+	assert.Contains(t, s, "table:     t")
+	assert.Contains(t, s, "ALTER TABLE t ADD COLUMN x int")
+}
+
+func TestStringRefusedIncludesReasonAndIdiom(t *testing.T) {
+	s := Verdict{
+		Outcome:    OutcomeRefused,
+		Reason:     ReasonIndexStatement,
+		Statement:  "CREATE INDEX i ON t (c)",
+		SaferIdiom: "CREATE INDEX CONCURRENTLY i ON t (c)",
+	}.String()
+	assert.Contains(t, s, "refused (index-statement)")
+	assert.Contains(t, s, "CREATE INDEX CONCURRENTLY")
+}