extract migrate's pipeline into pkg/migrate, a library front door

SAFETY.md

@@ -30,6 +30,7 @@ The invariant registry (invariant IDs referenced below) lives in
 | `pkg/statement`, `pkg/planner`, `pkg/schemadiff`, `pkg/router`, `pkg/plan`, `pkg/lint`, `pkg/suggest` — classify/diff/route/report | ❌ periphery¹ | `pkg/statement` (parse boundary), `pkg/schemadiff` (introspect/diff via scratch execute-and-introspect), `pkg/planner` (classifier), `pkg/router` (backend assignment + availability policy), `pkg/plan` (versioned dry-run plan report), `pkg/lint` (offline typed findings), and `pkg/suggest` (advisory rewrites with typed caveats) exist (Phases 2.1–2.5) | (CO-7 holds at the parse boundary) |
 | `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
 | `pkg/diffplan` — desired schema → routed convergence plan, the declarative front door as a library (the CLI `diff` and embedding orchestrators share it) | ❌ periphery | exists | — |
+| `pkg/migrate` — one gated statement → resolve, classify, route, execute → one verdict; the imperative front door as a library (the CLI `migrate` and embedding orchestrators share it) | ❌ periphery² | exists | — |
 | `internal/cli` — CLI, flags, help, prompts | ❌ periphery | `migrate`, `status`, `diff`, `fmt`, `lint`, and `suggest` exist | — |
 | `pkg/progress` — strategy-wide progress snapshots; the executors' observation seam (core imports it, so its locking discipline is core-critical); copy counters reserved for later | ✅ core | native progress exists | — |
 | orchestrator adapter | ❌ periphery | planned (Phase 11) | OC-* hold *at* the boundary |
@@ -41,6 +42,12 @@ Today a "copy" route reports unavailable; once copy-and-swap exists, a wrong "co
 produce a wasteful but *correct* schema change because the checksum will still gate it.
 The core executors re-verify their own preconditions and never trust that the planner checked.
 
+² **The imperative front door is periphery for the same reason.** `pkg/migrate` sequences the
+pipeline — gate, resolve, preflight, execute — but every dangerous step it requests is enforced
+by the core packages it calls: the executors re-verify admission and run under their own bounded
+budgets, and preflight's proof types gate what may execute. A wrong sequencing decision in
+`pkg/migrate` yields a refusal or a bounded failed attempt, never an unbounded lock.
+
 ## Rules inside the core
 
 The short version — the full rules live in [docs/tcb-model.md](docs/tcb-model.md):

docs/architecture.md

@@ -190,6 +190,7 @@ different levels of commitment:
 | `pkg/suggest` | Offline advisory surface: maps risky DDL to the safer native form the engine would run, with typed caveats and manual-path guidance; emits the versioned suggest report ([suggest-report.md](suggest-report.md)) | exists |
 | `pkg/plan` | Versioned machine-readable dry-run plan report — the one JSON contract both front doors emit and an orchestrator consumes | exists (Phase 2.5) |
 | `pkg/diffplan` | The declarative front door as a library: desired schema in, routed `plan.Report` out — the CLI `diff` and embedding orchestrators share this one pipeline | exists |
+| `pkg/migrate` | The imperative front door as a library: one parsed statement in — gate, resolve, classify, route, execute — one `verdict.Verdict` out; the CLI `migrate` and embedding orchestrators share this one pipeline | exists |
 | `pkg/router` | Route classified statements to native / copy-and-swap / refuse dispositions; copy-and-swap reports unavailable until that backend lands | exists (Phase 2.4) |
 | `pkg/executor` | Bounded optimistic native attempt, the concurrent index build, and the autocommit safer-sequence runner, with stable outcome codes; the full `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) arrives with the copy-and-swap backend | native execution exists |
 | `pkg/progress` | Strategy-wide, pollable progress snapshots: native phase/elapsed time, sequence position, retry attempt, and server-reported concurrent-index work; optional copy counters are reserved for copy-and-swap | native progress exists |

docs/schemabot-integration.md

@@ -77,7 +77,7 @@ the integration phase starts; they drift.)
 | --- | --- |
 | `Name()` | a stable identifier, e.g. `"pg-sprite"` |
 | `Plan` | run parse → declarative diff when applicable → classify → route — exported as `diffplan.Plan` in [`pkg/diffplan`](../pkg/diffplan/diffplan.go) (parse via `statement.ParseDesired`, connect via `dbconn.NewPool`, inputs named by `diffplan.Request{Schema, Desired}`); return a `PlanResult` whose `SchemaChange.TableChanges` are `engine.TableChange{Table, Operation (statement.StatementType), DDL, IsUnsafe, UnsafeReason, ExecutionMode, ModeReason}`; map a **not native-safe** refusal to `engine.ExecutionModeBlocked` with the refusal reason as `ModeReason` (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)) |
-| `Apply` | start the native executor asynchronously and return immediately; re-resolve the routing decision at execution time rather than trusting the stored plan-time verdict (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)) |
+| `Apply` | start the native executor asynchronously and return immediately — the synchronous core is exported as `migrate.Run` in [`pkg/migrate`](../pkg/migrate/migrate.go) (parse via `statement.ParseOne`, gate via `migrate.Gate` before dialing, connect via `dbconn.NewPool`, policy via `migrate.DefaultOptions` tuned per table): one statement in, one `verdict.Verdict` out, with a three-shape contract — refusal (verdict, nil error), execution failure (failed verdict carrying the stable code and the committed prefix, plus the operational error), or an error with a zero verdict (stopped before executing). `Run` re-resolves the routing decision at execution time, so the adapter never trusts the stored plan-time verdict (see [execution-mode verdicts](#execution-mode-verdicts-and-direct-execution)) |
 | `Progress` | per-table rows-copied / total / percent / ETA / checksum state |
 | `Stop` / `Start` | checkpoint and resume (slot + copy + applier watermark) |
 | `Cutover` | the deferred, operator-gated atomic swap |

internal/cli/dryrun.go

@@ -2,18 +2,15 @@ package cli
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 
-	"github.com/jackc/pgx/v5/pgxpool"
-
 	"github.com/block/pg-sprite/pkg/dbconn"
+	"github.com/block/pg-sprite/pkg/migrate"
 	"github.com/block/pg-sprite/pkg/plan"
 	"github.com/block/pg-sprite/pkg/planner"
 	"github.com/block/pg-sprite/pkg/preflight"
 	"github.com/block/pg-sprite/pkg/router"
-	"github.com/block/pg-sprite/pkg/schemadiff"
 	"github.com/block/pg-sprite/pkg/statement"
 	"github.com/block/pg-sprite/pkg/verdict"
 )
@@ -33,7 +30,7 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 		return err
 	}
 	logger.Debug("statement parsed", "kind", st.Kind(), "schema", st.Schema(), "table", st.Table())
-	if v, refused := gateVerdict(st); refused {
+	if v, refused := migrate.Gate(st); refused {
 		return c.emit(out, v)
 	}
 
@@ -43,7 +40,7 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 	}
 	defer pool.Close()
 
-	facts, targetFacts, tableExists, err := dryRunFacts(ctx, pool, st)
+	facts, err := migrate.LiveFacts(ctx, pool, st)
 	if err != nil {
 		return err
 	}
@@ -54,7 +51,7 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 	if err != nil {
 		return err
 	}
-	classified, err := planner.Classify(canonical, facts)
+	classified, err := planner.Classify(canonical, facts.Classifier)
 	if err != nil {
 		return err
 	}
@@ -63,9 +60,9 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 		"route", string(classified.Route), "disposition", string(routed.Disposition))
 
 	report := plan.NewReport(plan.SourceAlter)
-	report.Schema = resolvedSchema(st)
+	report.Schema = migrate.ResolvedSchema(st)
 	report.Table = st.Table()
-	report.TableExists = tableExists
+	report.TableExists = facts.TableExists
 	if report.ServerVersion, err = dbconn.ServerVersion(ctx, pool); err != nil {
 		return err
 	}
@@ -77,11 +74,11 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 		}
 		report.Statements = append(report.Statements, ps)
 	}
-	if targetFacts.Partitioned() {
+	if facts.Target.Partitioned() {
 		refused := make([]bool, len(report.Statements))
 		for i := range report.Statements {
 			var cause preflight.PartitionRefusalCause
-			cause, err = preflight.RefusesPartitionedParent(targetFacts.ServerMajor(), report.Statements[i].ExecSQL)
+			cause, err = preflight.RefusesPartitionedParent(facts.Target.ServerMajor(), report.Statements[i].ExecSQL)
 			if err != nil {
 				return err
 			}
@@ -108,41 +105,3 @@ func (c *MigrateCmd) runDryRun(ctx context.Context, out io.Writer) error {
 	}
 	return nil
 }
-
-// resolvedSchema is the schema the engine plans against: the statement's
-// qualification, or public — the default the engine introspects — when a
-// table-targeted statement leaves it unqualified. The report carries the
-// resolved name, never the submitted one: a stored plan must not depend on
-// the reader's search_path to say which table it describes.
-func resolvedSchema(st statement.Statement) string {
-	if st.Schema() == "" && st.Table() != "" {
-		return "public"
-	}
-	return st.Schema()
-}
-
-// dryRunFacts introspects the statement's target table for classifier
-// facts. Statements without a single table target (index drops, REINDEX)
-// and missing tables classify with zero facts. The returned tableExists
-// mirrors the introspection outcome for the report: true when the table
-// was found, false when it was looked up and missing, nil when the
-// statement has no single table target to introspect.
-func dryRunFacts(ctx context.Context, pool *pgxpool.Pool, st statement.Statement) (planner.Facts, preflight.TargetFacts, *bool, error) {
-	if st.Table() == "" {
-		return planner.Facts{}, preflight.TargetFacts{}, nil, nil
-	}
-	live, err := schemadiff.Introspect(ctx, pool, resolvedSchema(st), st.Table())
-	switch {
-	case errors.Is(err, schemadiff.ErrTableNotFound):
-		exists := false
-		return planner.Facts{}, preflight.TargetFacts{}, &exists, nil
-	case err != nil:
-		return planner.Facts{}, preflight.TargetFacts{}, nil, err
-	}
-	targetFacts, err := preflight.LookupTargetFacts(ctx, pool, resolvedSchema(st), st.Table())
-	if err != nil {
-		return planner.Facts{}, preflight.TargetFacts{}, nil, err
-	}
-	exists := true
-	return planner.FactsFrom(live), targetFacts, &exists, nil
-}