diff --git a/pkg/cmd/commands/storage.go b/pkg/cmd/commands/storage.go
index 669a6cac3..84109b69c 100644
--- a/pkg/cmd/commands/storage.go
+++ b/pkg/cmd/commands/storage.go
@@ -15,15 +15,40 @@ import (
 	"github.com/block/schemabot/pkg/schema"
 )
 
-// StorageCmd groups operator commands that act directly on SchemaBot's own
-// storage database. Unlike the API-client commands, these connect to storage
-// themselves and work while the server is down — they exist for maintenance
-// windows such as a cross-dialect data move or a restore from a dump.
+// StorageCmd groups operator commands that act on SchemaBot's own storage
+// database rather than on a user's database.
+//
+// The maintenance commands connect to storage themselves and work while the
+// server is down, for windows such as a cross-dialect data move or a restore
+// from a dump. The schema commands do both: they read a storage database
+// through the API by default, so an operator can reach a data plane's storage
+// that no workstation can dial, and connect directly when the server is down —
+// including when it is down because its own schema bootstrap is failing.
 type StorageCmd struct {
+	Plan                     StoragePlanCmd              `cmd:"" help:"Show the storage DDL outstanding between a live storage database and the release you name; read-only, safe at any time. Exits 0 when converged and 2 when statements are outstanding."`
+	Apply                    StorageApplyCmd             `cmd:"" help:"Converge a SchemaBot instance's storage database by running the schema bootstrap it would run on its next boot, under the same advisory lock and the same destructive-statement refusal."`
 	ResyncIdentitySequences  ResyncIdentitySequencesCmd  `cmd:"" name:"resync-identity-sequences" help:"Advance PostgreSQL identity sequences on storage tables past their columns' stored maxima after an explicit-id bulk load; run after the load has fully committed and before default inserts resume — advance-only and safe to rerun."`
 	CanonicalizeIdentityKeys CanonicalizeIdentityKeysCmd `cmd:"" name:"canonicalize-identity-keys" help:"Fold stored identity strings (repository, database, environment, deployment, lock owner) on PostgreSQL storage tables to canonical lowercase; run once, in a quiesced maintenance window, after every writer runs a release that folds identity strings at the write boundaries. The rewrite is one-way — original spellings are not recorded — so the command prompts unless --auto-approve is set; it only rewrites non-canonical rows, safe to rerun."`
 }
 
+// UsesAPI reports whether the named storage subcommand will reach its target
+// through the API, and so needs an endpoint resolved before it runs.
+//
+// Only the schema commands have both paths, and the flags say which is in use.
+// The maintenance commands open the storage database themselves and never call
+// the API, so an unknown subcommand answering false is the safe default: the
+// cost is a command that resolves no endpoint it was not going to use.
+func (c *StorageCmd) UsesAPI(subcommand string) bool {
+	switch subcommand {
+	case "plan":
+		return !c.Plan.direct()
+	case "apply":
+		return !c.Apply.direct()
+	default:
+		return false
+	}
+}
+
 // ResyncIdentitySequencesCmd resyncs the identity sequences of SchemaBot's
 // PostgreSQL storage tables after an explicit-id bulk load — a data move
 // that preserves ids, or a restore from a dump without sequence state —
@@ -159,74 +184,45 @@ func (cmd *CanonicalizeIdentityKeysCmd) Run(ctx context.Context, g *Globals) err
 }
 
 // resolveStorageDSN returns the storage DSN and a loggable description of
-// where it came from. A direct --dsn is used after verifying it parses as a
-// PostgreSQL DSN; otherwise the server config (--config, then
-// $SCHEMABOT_CONFIG_FILE) is loaded and its resolved storage DSN is used,
-// failing closed when the configured storage dialect is not postgres —
-// purpose names the operation in these errors. The source never contains
-// the DSN itself, which may embed credentials.
+// where it came from, for the commands that only apply to PostgreSQL storage —
+// purpose names the operation in the refusal. The source never contains the
+// DSN itself, which may embed credentials.
 func resolveStorageDSN(dsnFlag, configFlag, purpose string) (string, string, error) {
-	directDSN := strings.TrimSpace(dsnFlag)
-	if directDSN != "" && configFlag != "" {
+	if dsnFlag != "" && configFlag != "" {
 		return "", "", fmt.Errorf("--dsn and --config are mutually exclusive; pass the storage DSN directly or resolve it from a server config, not both")
 	}
+	// A direct DSN is checked against the PostgreSQL grammar rather than handed
+	// to the generic family inference: these commands take no --dialect, so
+	// there is nothing for an ambiguous DSN to be resolved by, and the refusal
+	// that helps is the one naming the operation that does not apply to it.
+	//
+	// Presence routes, not content: a --dsn of whitespace is a malformed direct
+	// connection and is refused as one, because reading it as "no DSN" would
+	// resolve the config instead and operate on a database the operator never
+	// named (AZ-5).
 	if dsnFlag != "" {
+		directDSN := strings.TrimSpace(dsnFlag)
 		if directDSN == "" {
 			return "", "", fmt.Errorf("storage DSN not configured: --dsn contains only whitespace")
 		}
-		// The config path refuses non-postgres storage via the configured
-		// dialect; the direct path has no dialect field, so refuse any DSN
-		// that does not parse as PostgreSQL instead of failing later with an
-		// opaque connection error — or worse, against the wrong server.
 		if _, err := postgresconn.ConnectionDSN(directDSN); err != nil {
 			return "", "", fmt.Errorf("storage DSN from --dsn is not a PostgreSQL DSN; %s only applies to %q storage: %w", purpose, schema.DialectPostgres, err)
 		}
 		return directDSN, "--dsn flag", nil
 	}
-
-	configPath := configFlag
-	source := fmt.Sprintf("server config %s", configPath)
-	if configPath == "" {
-		configPath = os.Getenv("SCHEMABOT_CONFIG_FILE")
-		if configPath == "" {
-			return "", "", fmt.Errorf("no storage DSN source: set --dsn, --config, or the SCHEMABOT_CONFIG_FILE environment variable")
-		}
-		source = fmt.Sprintf("server config %s ($SCHEMABOT_CONFIG_FILE)", configPath)
-	}
-
-	var cfg *api.ServerConfig
-	var err error
-	if configFlag == "" {
-		cfg, err = api.LoadServerConfig()
-	} else {
-		cfg, err = api.LoadServerConfigFromFile(configPath)
-	}
-	if err != nil {
-		return "", "", fmt.Errorf("load %s: %w", source, err)
-	}
-
-	dialect, err := cfg.Storage.ResolveDialect()
+	// The family is read from the config before the DSN is fetched, so a
+	// command that does not apply to it refuses on the family — naming the
+	// reason, and reading no secret for a connection it will never open.
+	configured, err := resolveStorageConfig(configFlag)
 	if err != nil {
-		return "", "", fmt.Errorf("resolve storage dialect from %s: %w", source, err)
+		return "", "", err
 	}
-	if dialect != schema.DialectPostgres {
-		return "", "", fmt.Errorf("storage dialect in %s is %q; %s only applies to %q storage", source, dialect, purpose, schema.DialectPostgres)
+	if configured.dialect != schema.DialectPostgres {
+		return "", "", fmt.Errorf("storage dialect in %s is %q; %s only applies to %q storage", configured.source, configured.dialect, purpose, schema.DialectPostgres)
 	}
-
-	dsn, err := cfg.StorageDSN()
+	target, err := configured.target()
 	if err != nil {
-		return "", "", fmt.Errorf("resolve storage DSN from %s: %w", source, err)
-	}
-	dsn = strings.TrimSpace(dsn)
-	if dsn == "" {
-		return "", "", fmt.Errorf("storage DSN not configured (set --dsn, config storage.dsn or storage.dsn_from, STORAGE_DSN, or MYSQL_DSN)")
-	}
-	if cfg.Storage.DSN == "" && cfg.Storage.DSNFrom == nil {
-		if strings.TrimSpace(os.Getenv("STORAGE_DSN")) != "" {
-			source = "STORAGE_DSN environment variable"
-		} else if strings.TrimSpace(os.Getenv("MYSQL_DSN")) != "" {
-			source = "MYSQL_DSN environment variable"
-		}
+		return "", "", err
 	}
-	return dsn, source, nil
+	return target.dsn, target.source, nil
 }
diff --git a/pkg/cmd/commands/storage_schema.go b/pkg/cmd/commands/storage_schema.go
new file mode 100644
index 000000000..104fc7bf5
--- /dev/null
+++ b/pkg/cmd/commands/storage_schema.go
@@ -0,0 +1,546 @@
+package commands
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"os"
+	"strings"
+
+	"github.com/block/schemabot/pkg/api"
+	"github.com/block/schemabot/pkg/apitypes"
+	cmdclient "github.com/block/schemabot/pkg/cmd/client"
+	"github.com/block/schemabot/pkg/schema"
+)
+
+// Storage schema commands answer, and then close, the one question a deploy
+// that did not converge leaves open: which storage DDL is still outstanding on
+// this instance's own storage database.
+//
+// Both read the live database, always. A release names the schema to compare
+// against (--release, --schema-dir), and it never stands in for the live side:
+// what a release would converge to and what the storage actually converged to
+// differ exactly when a deploy has failed, which is the only time anyone runs
+// these. So the desired side is a release's files — a published tag's or a
+// checkout's — and the report names which.
+//
+// There are two ways to reach a storage database, and which one applies is the
+// operator's to state rather than the CLI's to discover:
+//
+//	through the API      the server reads its own storage, or asks a data
+//	                     plane to read its own over the connection that
+//	                     already exists between them
+//	directly, by DSN     this workstation opens the storage database itself,
+//	                     for when the server is down — including when it is
+//	                     down because its schema bootstrap is failing
+//
+// Nothing falls back from one to the other. A deployment that cannot be
+// reached through the API is an error naming the deployment, never a report
+// about a different database that happened to be reachable.
+
+// storageSchemaTargetFlags selects which storage database the command acts on.
+// The two paths are mutually exclusive and the flags say which is in use, so a
+// command never has to infer the operator's intent from what happened to
+// resolve.
+type storageSchemaTargetFlags struct {
+	Deployment  string `help:"Data plane whose storage to read, as named under tern_deployments in the server config; omit for the storage of the server the CLI is pointed at"`
+	Environment string `short:"e" help:"Environment of the deployment's endpoint; required with --deployment"`
+	DSN         string `help:"Connect to the storage database directly with this DSN instead of going through the API; for when the server is down"`
+	Config      string `help:"Server config file to resolve the storage DSN from, connecting directly instead of going through the API"`
+	Dialect     string `help:"Storage database family of --dsn (mysql or postgres) when its form does not say"`
+}
+
+// direct reports whether the operator asked for a direct connection. Passing
+// either flag is the whole signal: both exist only on that path.
+//
+// Presence is what routes, not content. A --dsn of whitespace is a malformed
+// direct connection, and resolveStorageTarget refuses it by name; reading it
+// as "no DSN" would send the command through the API instead and report a
+// different database than the operator addressed (AZ-5).
+//
+// Only a flag routes, never the environment. $SCHEMABOT_CONFIG_FILE is the
+// fallback --config source for the storage maintenance commands, which are
+// direct-only and have no other path to take; on these two it is never read,
+// because a direct connection is already chosen by then or not at all. An
+// operator who runs these with no target flags asked for the server's storage,
+// and an exported variable in their shell must not turn that into a direct
+// connection to whatever database the config names.
+func (f *storageSchemaTargetFlags) direct() bool {
+	return f.DSN != "" || f.Config != ""
+}
+
+// validate refuses flag combinations that mix the two paths, instead of
+// silently honoring one and dropping the other. Dropping a --deployment would
+// report the wrong database under the right name, which is worse than any
+// error message.
+func (f *storageSchemaTargetFlags) validate() error {
+	if !f.direct() {
+		if strings.TrimSpace(f.Dialect) != "" {
+			return fmt.Errorf("--dialect only applies to a direct connection: through the API the server reports its own storage dialect; pass --dsn or --config to connect directly")
+		}
+		if f.Deployment == "" && f.Environment != "" {
+			return fmt.Errorf("-e %s was given without --deployment: an environment selects which of a deployment's endpoints to reach, so name the deployment too, or omit both to read the storage of the server the CLI is pointed at", f.Environment)
+		}
+		if f.Deployment != "" && f.Environment == "" {
+			return fmt.Errorf("--deployment %s needs -e <environment>: a deployment serves one endpoint per environment, so there is no single storage to read without one", f.Deployment)
+		}
+		return nil
+	}
+	if f.Deployment != "" {
+		return fmt.Errorf("--deployment cannot be combined with a direct connection: a DSN addresses one storage database, and reaching a data plane's storage goes through its own endpoint; drop --dsn/--config to route through the API, or drop --deployment to read the database the DSN names")
+	}
+	if f.Environment != "" {
+		return fmt.Errorf("-e cannot be combined with a direct connection: a DSN already names one database, so there is no endpoint to select")
+	}
+	return nil
+}
+
+// StoragePlanCmd reports the storage DDL outstanding between a live storage
+// database and the schema files of the release the operator names.
+//
+// It is strictly read-only — it reads the catalog, computes a diff, and takes
+// no lock — so it is safe to run at any time, including against production
+// while an apply is in flight or an incident is open.
+//
+// The exit status is the machine-readable half of the answer: 0 when the
+// storage needs nothing, 2 when statements are outstanding, and 1 when the
+// read itself failed. A pre-deploy gate needs those three apart, because
+// "converged" and "unreachable" call for opposite decisions.
+//
+// The release it compares against is always named — --release or --schema-dir —
+// because the question a deploy asks is whether the storage is ready for the
+// release about to roll, and that is a different question from whether it
+// matches the release now running.
+type StoragePlanCmd struct {
+	storageSchemaTargetFlags `embed:""`
+	storageSchemaSourceFlags `embed:""`
+	JSON                     bool `help:"Output as JSON"`
+	// AllowUnsafe is deliberately not a flag on the plan. The normal
+	// plan/apply flow has no way to preview an apply's --allow-unsafe either:
+	// the plan discloses the destructive statements and names the flag, and the
+	// way to see them as statements that will run is to run the apply and
+	// decline its prompt. The apply's own preview is that path, and sets this.
+	//
+	// A plan can still report them as running without it, because a target's
+	// standing storage policy can allow destructive changes with no flag on the
+	// line at all.
+	AllowUnsafe bool `kong:"-"`
+}
+
+func (cmd *StoragePlanCmd) Run(ctx context.Context, g *Globals) error {
+	if err := cmd.validate(); err != nil {
+		return err
+	}
+	if err := cmd.validateSource(); err != nil {
+		return err
+	}
+	report, err := cmd.read(ctx, g)
+	if err != nil {
+		return err
+	}
+	if cmd.JSON {
+		encoder := json.NewEncoder(os.Stdout)
+		encoder.SetIndent("", "  ")
+		if err := encoder.Encode(apitypes.StorageSchemaPlanResponse{Report: report}); err != nil {
+			return fmt.Errorf("encode storage schema report: %w", err)
+		}
+	} else if err := outputStorageSchemaPlan(report, false, "", nil); err != nil {
+		return err
+	}
+	if report.Converged {
+		return nil
+	}
+	return exitStorageSchemaOutstanding()
+}
+
+// read fetches the report over whichever path the flags selected.
+func (cmd *StoragePlanCmd) read(ctx context.Context, g *Globals) (*apitypes.StorageSchemaReport, error) {
+	if cmd.direct() {
+		return cmd.readDirect(ctx, g)
+	}
+	return cmd.readThroughAPI(ctx, g)
+}
+
+// readDirect opens the storage database from this workstation and diffs it
+// here, for when the server is down — including when it is down because its own
+// schema bootstrap is failing.
+func (cmd *StoragePlanCmd) readDirect(ctx context.Context, g *Globals) (*apitypes.StorageSchemaReport, error) {
+	target, err := resolveStorageTarget(cmd.DSN, cmd.Config, cmd.Dialect)
+	if err != nil {
+		return nil, err
+	}
+	// The dialect is already resolved on this path, so a release fetch costs no
+	// extra round trip.
+	desired, err := cmd.resolve(ctx, func() (schema.Dialect, error) { return target.dialect, nil })
+	if err != nil {
+		return nil, err
+	}
+	logger := storageSchemaLogger(g)
+	logger.Info("reading storage schema directly",
+		"source", target.source, "dialect", target.dialect, "schema_source", desired.Describe())
+	report, err := api.PlanStorageSchema(ctx, target.dsn, desired, logger,
+		target.ensureSchemaOptions(cmd.AllowUnsafe)...)
+	if err != nil {
+		return nil, fmt.Errorf("diff storage schema on the database from %s: %w", target.source, err)
+	}
+	// This CLI answered, so the version is this binary's — and so is the
+	// schema, unless the operator named another release's.
+	report.AttributeTo(g.Version)
+	return report.APIType(), nil
+}
+
+// readThroughAPI asks the server, which reads its own storage or has a data
+// plane read its own. A desired schema the operator named travels with the
+// request, because the answering binary does not carry another release's files.
+func (cmd *StoragePlanCmd) readThroughAPI(ctx context.Context, g *Globals) (*apitypes.StorageSchemaReport, error) {
+	endpoint, err := g.Resolve()
+	if err != nil {
+		return nil, err
+	}
+	request := apitypes.StorageSchemaPlanRequest{
+		Deployment:       cmd.Deployment,
+		Environment:      cmd.Environment,
+		AllowDestructive: cmd.AllowUnsafe,
+	}
+	desired, err := cmd.resolve(ctx, func() (schema.Dialect, error) {
+		return cmd.dialectThroughAPI(ctx, endpoint)
+	})
+	if err != nil {
+		return nil, err
+	}
+	if desired != nil {
+		request.SchemaFiles = desired.Files
+		request.SchemaSource = desired.Description
+	}
+
+	response, err := cmdclient.StorageSchemaPlan(ctx, endpoint, request)
+	if err != nil {
+		return nil, fmt.Errorf("diff storage schema%s: %w", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment), err)
+	}
+	if response.Report == nil {
+		return nil, fmt.Errorf("storage schema diff%s returned no report", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment))
+	}
+	return response.Report, nil
+}
+
+// dialectThroughAPI asks the target which storage family it runs, so a release
+// fetch reads the right one of its schema directories.
+//
+// It is the diff itself, asked with no desired schema: a read-only call that
+// the target answers about its own storage, which is the only authority on the
+// question. Asking costs a round trip and is why only --release pays for it —
+// but asking beats making the operator state a dialect their control plane
+// already knows, and beats guessing one and fetching DDL of the wrong family.
+func (cmd *StoragePlanCmd) dialectThroughAPI(ctx context.Context, endpoint string) (schema.Dialect, error) {
+	response, err := cmdclient.StorageSchemaPlan(ctx, endpoint, apitypes.StorageSchemaPlanRequest{
+		Deployment:  cmd.Deployment,
+		Environment: cmd.Environment,
+	})
+	if err != nil {
+		return "", fmt.Errorf("ask which storage family%s runs: %w", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment), err)
+	}
+	if response.Report == nil || response.Report.Dialect == "" {
+		return "", fmt.Errorf("the report for the storage%s named no dialect, so there is no way to tell which of a release's schema files apply to it", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment))
+	}
+	return schema.Dialect(response.Report.Dialect), nil
+}
+
+// StorageApplyCmd converges a SchemaBot instance's storage database by running
+// the schema bootstrap that instance would run on its next boot.
+//
+// It is the bootstrap, not a second implementation of it: the same differ, the
+// same refusal of destructive statements, and the same advisory lock — so two
+// operators running this at once serialize exactly the way two booting pods
+// do, and a pre-deploy convergence step is this command with nothing added.
+// It converges to the schema of the binary that runs it, and there is no flag
+// to point it at another release's — see storageSchemaSourceRefusal.
+type StorageApplyCmd struct {
+	storageSchemaTargetFlags `embed:""`
+	AllowUnsafe              bool `help:"Permit the destructive statements the convergence would otherwise refuse; it widens the target's standing storage policy and never narrows it" name:"allow-unsafe"`
+	AutoApprove              bool `short:"y" help:"Skip confirmation prompt" name:"auto-approve"`
+	JSON                     bool `help:"Output as JSON"`
+	// The diff's file selectors are accepted here only to be refused with the
+	// reason and the alternative. An operator who has just run the diff against
+	// a release reaches for the same flags on the apply, and Kong's bare
+	// "unknown flag" would leave them guessing at whether the convergence
+	// silently used a different schema. --release-repo is accepted for the same
+	// reason: it is the flag most likely to be left on the line after the one
+	// it modifies has been dropped.
+	SchemaDir string `hidden:"" name:"schema-dir"`
+	Release   string `hidden:""`
+	Repo      string `hidden:"" name:"release-repo"`
+}
+
+func (cmd *StorageApplyCmd) Run(ctx context.Context, g *Globals) error {
+	if err := cmd.validate(); err != nil {
+		return err
+	}
+	if err := storageSchemaSourceRefusal(cmd.SchemaDir, cmd.Release, cmd.Repo); err != nil {
+		return err
+	}
+
+	// Every convergence plans first, attended or not, the way `apply` does.
+	// The plan is a fresh read rather than a description of the command: an
+	// operator approving DDL on SchemaBot's own storage should see the
+	// statements, and the read is free of side effects. Unattended, it is what
+	// the destructive gate below decides from.
+	//
+	// No selector on the preview asks the target about its own embedded schema,
+	// which is the schema this convergence is about to run.
+	preview := &StoragePlanCmd{
+		storageSchemaTargetFlags: cmd.storageSchemaTargetFlags,
+		AllowUnsafe:              cmd.AllowUnsafe,
+	}
+	report, err := preview.read(ctx, g)
+	if err != nil {
+		return err
+	}
+	// With --auto-approve the convergence's own planned report says what ran,
+	// so the plan is printed after the fact instead — unless the gate stops the
+	// run, which prints it itself.
+	if !cmd.AutoApprove {
+		if err := outputStorageSchemaPlan(report, true, cmd.rerunWithAllowUnsafe(), nil); err != nil {
+			return err
+		}
+	}
+
+	// Both refusals run on the attended and the unattended path alike, so an
+	// operator and a pre-deploy job are told the same thing. Manual first: it
+	// gates the whole drift set, which makes a destructive statement behind it
+	// unreachable rather than merely refused.
+	if err := blockManualStorageApply(report, cmd.AutoApprove, cmd.rerunWithAllowUnsafe()); err != nil {
+		return err
+	}
+	if err := blockDestructiveStorageApply(report, cmd.AutoApprove, cmd.rerunWithAllowUnsafe()); err != nil {
+		return err
+	}
+
+	if !cmd.AutoApprove {
+		confirmed, err := confirmAction(
+			storageSchemaConfirmation(report),
+			"\nApply cancelled.",
+		)
+		if err != nil {
+			return err
+		}
+		if !confirmed {
+			return nil
+		}
+	}
+
+	planned, remaining, err := cmd.converge(ctx, g)
+	if err != nil {
+		return err
+	}
+	if cmd.JSON {
+		encoder := json.NewEncoder(os.Stdout)
+		encoder.SetIndent("", "  ")
+		if err := encoder.Encode(apitypes.StorageSchemaApplyResponse{Planned: planned, Remaining: remaining}); err != nil {
+			return fmt.Errorf("encode storage schema convergence: %w", err)
+		}
+	} else if err := outputStorageSchemaConvergence(planned, remaining, cmd.AutoApprove, cmd.rerunWithAllowUnsafe()); err != nil {
+		return err
+	}
+	return storageSchemaConvergenceOutcome(remaining)
+}
+
+// rerunWithAllowUnsafe is the command that permits what this one refused, for
+// an operator to copy off a refusal.
+//
+// It carries the target flags forward because the command has to address the
+// same storage database the refusal is about. Dropping them would suggest a
+// convergence of the storage of the server the CLI happens to point at, which
+// during a rollback is a different database than the one being looked at.
+//
+// A DSN is named rather than repeated: it carries the storage database's
+// credentials, and this is printed to a terminal and scrolled back through.
+func (cmd *StorageApplyCmd) rerunWithAllowUnsafe() string {
+	parts := []string{"storage", "apply"}
+	switch {
+	case strings.TrimSpace(cmd.DSN) != "":
+		parts = append(parts, "--dsn <the same DSN>")
+	case cmd.Config != "":
+		parts = append(parts, "--config", cmd.Config)
+	case cmd.Deployment != "":
+		parts = append(parts, "--deployment", cmd.Deployment, "-e", cmd.Environment)
+	}
+	if cmd.Dialect != "" {
+		parts = append(parts, "--dialect", cmd.Dialect)
+	}
+	return strings.Join(append(parts, "--allow-unsafe"), " ")
+}
+
+// blockDestructiveStorageApply stops a convergence that would have to destroy
+// storage state nothing has permitted, before it runs anything.
+//
+// This is the gate `apply` puts in front of a destructive schema change, in the
+// same place and behind the same flag: the plan is on screen, the statements
+// are named, and --allow-unsafe is the way through. It is in front of the
+// confirmation rather than after it because --auto-approve skips a prompt, and
+// consenting to a convergence is not consenting to destroy state.
+//
+// A deployment that already allows destructive storage changes has permitted
+// them, so there is nothing here to ask: the report says so and the statements
+// run. The gate narrows no standing policy (AV-9), and where it does stop a run
+// it runs strictly less than the convergence would have — the bootstrap refuses
+// the same statements on its own and converges the safe remainder. What changes
+// is when the operator finds out: before a DROP against SchemaBot's own storage
+// is decided, rather than in a report of what was already done.
+//
+// withPlan prints the plan for an unattended run, which has not printed one
+// yet. Naming refused statements without showing them would send the operator
+// back to `storage plan` to find out what was refused.
+// blockManualStorageApply stops a convergence that cannot run at all.
+//
+// A manual entry gates the whole drift set: the bootstrap refuses every
+// statement in the report until an operator resolves it by hand. So it is the
+// refusal to report even when destructive statements are present too — those
+// are not reachable until this one is resolved — which is why it runs before
+// the destructive gate rather than after it.
+//
+// withPlan prints the plan for an unattended run, which has not printed one
+// yet. The error says the entries are listed above, and on that path nothing
+// has listed them.
+func blockManualStorageApply(report *apitypes.StorageSchemaReport, withPlan bool, rerun string) error {
+	if len(report.Manual) == 0 {
+		return nil
+	}
+	if withPlan {
+		if err := outputStorageSchemaPlan(report, true, rerun, nil); err != nil {
+			return err
+		}
+	}
+	return fmt.Errorf("refusing to converge storage schema on %s: %d change(s) need manual remediation first (listed above)",
+		storageSchemaDatabaseLabel(report), len(report.Manual))
+}
+
+func blockDestructiveStorageApply(report *apitypes.StorageSchemaReport, withPlan bool, rerun string) error {
+	if report.DestructiveAllowed || len(report.Destructive) == 0 {
+		return nil
+	}
+	if withPlan {
+		if err := outputStorageSchemaPlan(report, true, rerun, nil); err != nil {
+			return err
+		}
+	}
+	// The plan above carries the refusal, the statements, and the flag, so this
+	// asks only for the exit status — an "Error:" line restating it would be
+	// the third time the same refusal is on screen.
+	return ErrSilent
+}
+
+// storageSchemaConfirmation asks for the convergence the operator is about to
+// run, which is not always a set of statements.
+//
+// A preview that found nothing outstanding does not end the command. The
+// bootstrap converges more than the catalog: it also clears the schema change
+// engine's leftover tables, which outlive an interrupted convergence and are
+// invisible to a diff of the catalog (see api.ApplyStorageSchema). Stopping
+// here would leave them on the database and make an interactive apply do less
+// than the same command with --auto-approve — and the one an operator reaches
+// for mid-incident is the interactive one.
+func storageSchemaConfirmation(report *apitypes.StorageSchemaReport) string {
+	label := storageSchemaDatabaseLabel(report)
+	if report.Converged {
+		return fmt.Sprintf("\nThe catalog of %s already matches. Run the bootstrap anyway, to clear any engine state a catalog diff cannot see? Only 'yes' will be accepted: ", label)
+	}
+	return fmt.Sprintf("\nDo you want to apply these changes to %s? Only 'yes' will be accepted: ", label)
+}
+
+// storageSchemaConvergenceOutcome is whether a convergence counts as having
+// happened, decided from what it left behind.
+//
+// A manual-remediation entry means nothing ran at all: the bootstrap refuses
+// the whole drift set while one is outstanding. The report has already listed
+// them, and this is what makes the command fail — an unattended run must not
+// exit 0 having converged nothing, whichever output mode it was asked for.
+//
+// Refused destructive statements are the other case, and they are not a
+// failure: the surplus state is left in place on purpose (AV-9), everything
+// else converged, and a rollback is expected to sit there. An operator reaching
+// this having permitted none of them is what blockDestructiveStorageApply
+// stops, so the only way here is a target that gained surplus state between the
+// plan and the convergence — where leaving it in place is still the answer.
+func storageSchemaConvergenceOutcome(remaining *apitypes.StorageSchemaReport) error {
+	if len(remaining.Manual) == 0 {
+		return nil
+	}
+	return fmt.Errorf("storage schema on %s was not converged: %d change(s) need manual remediation before anything else runs",
+		storageSchemaDatabaseLabel(remaining), len(remaining.Manual))
+}
+
+// converge runs the convergence over whichever path the flags selected.
+func (cmd *StorageApplyCmd) converge(ctx context.Context, g *Globals) (planned, remaining *apitypes.StorageSchemaReport, err error) {
+	if cmd.direct() {
+		target, err := resolveStorageTarget(cmd.DSN, cmd.Config, cmd.Dialect)
+		if err != nil {
+			return nil, nil, err
+		}
+		logger := storageSchemaLogger(g)
+		logger.Info("converging storage schema directly",
+			"source", target.source,
+			"dialect", target.dialect,
+			"allow_destructive", target.allowDestructive || cmd.AllowUnsafe,
+			"config_allows_destructive", target.allowDestructive)
+		plannedReport, remainingReport, err := api.ApplyStorageSchema(ctx, target.dsn, logger,
+			target.ensureSchemaOptions(cmd.AllowUnsafe)...)
+		if err != nil {
+			return nil, nil, fmt.Errorf("converge storage schema on the database from %s: %w", target.source, err)
+		}
+		attributeStorageSchemaConvergence(g.Version, plannedReport, remainingReport)
+		return plannedReport.APIType(), remainingReport.APIType(), nil
+	}
+
+	endpoint, err := g.Resolve()
+	if err != nil {
+		return nil, nil, err
+	}
+	response, err := cmdclient.StorageSchemaApply(ctx, endpoint, apitypes.StorageSchemaApplyRequest{
+		Deployment:       cmd.Deployment,
+		Environment:      cmd.Environment,
+		AllowDestructive: cmd.AllowUnsafe,
+	})
+	if err != nil {
+		return nil, nil, fmt.Errorf("converge storage schema%s: %w", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment), err)
+	}
+	if response.Planned == nil || response.Remaining == nil {
+		// Both halves are required to say what happened; without the pair there
+		// is no way to tell a convergence that finished from one that left
+		// statements behind.
+		return nil, nil, fmt.Errorf("storage schema convergence%s returned an incomplete result; check the target's logs for whether it converged", storageSchemaTargetSuffix(cmd.Deployment, cmd.Environment))
+	}
+	return response.Planned, response.Remaining, nil
+}
+
+// attributeStorageSchemaConvergence says which release a convergence ran, on
+// both halves of it.
+//
+// A convergence names no schema source of its own — that is the invariant, not
+// an omission (AV-9) — so the attribution is what turns "the embedded schema"
+// into this binary's release. Both halves take it, and by the same call the
+// preview took: a run whose plan header named a release and whose result header
+// named a placeholder would read as two runs against two schemas.
+func attributeStorageSchemaConvergence(version string, planned, remaining *api.StorageSchemaReport) {
+	planned.AttributeTo(version)
+	remaining.AttributeTo(version)
+}
+
+// storageSchemaLogger builds the diagnostics logger for a direct connection. A
+// text handler on stderr, deliberately: this is a one-shot command read at a
+// terminal, and keeping diagnostics off stdout leaves the report itself
+// pipeable.
+func storageSchemaLogger(g *Globals) *slog.Logger {
+	return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
+		Level: logLevel(),
+	})).With("schemabot_version", g.Version)
+}
+
+// storageSchemaTargetSuffix names the target in an error, so a failure says
+// which storage was being read rather than only that a read failed.
+func storageSchemaTargetSuffix(deployment, environment string) string {
+	if deployment == "" {
+		return ""
+	}
+	return fmt.Sprintf(" for deployment %s in %s", deployment, environment)
+}
diff --git a/pkg/cmd/commands/storage_schema_integration_test.go b/pkg/cmd/commands/storage_schema_integration_test.go
new file mode 100644
index 000000000..9f2502164
--- /dev/null
+++ b/pkg/cmd/commands/storage_schema_integration_test.go
@@ -0,0 +1,107 @@
+//go:build integration
+
+package commands
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/testutil"
+)
+
+// The direct path is the one these commands exist for: the server is down —
+// possibly down because its own schema bootstrap is failing — so the operator
+// points the CLI at the storage database itself. It resolves the target, infers
+// the dialect from the DSN's form, runs this binary's own bootstrap and
+// attributes the report to this build, none of which the API path exercises.
+//
+// So it is driven here against a real, empty storage database: the convergence
+// has to create the whole schema from nothing, and the plan that follows has to
+// agree that it did.
+func TestStorageSchemaDirect_ConvergesAnEmptyDatabaseAndReportsItConverged(t *testing.T) {
+	dsn, db := testutil.StartPostgres(t, "schemabot")
+	globals := &Globals{Version: "v1.4.0"}
+	require.False(t, testutil.PostgresTableExists(t, db, "public", "applies"),
+		"the database starts with no storage schema, so the convergence has to create it")
+
+	apply := &StorageApplyCmd{
+		storageSchemaTargetFlags: storageSchemaTargetFlags{DSN: dsn},
+		AutoApprove:              true,
+	}
+	applyOut := captureStdout(func() {
+		require.NoError(t, apply.Run(t.Context(), globals))
+	})
+
+	assert.True(t, testutil.PostgresTableExists(t, db, "public", "applies"),
+		"the convergence ran this binary's own bootstrap against the database the DSN names")
+	assert.True(t, testutil.PostgresTableExists(t, db, "public", "settings"))
+	assert.Contains(t, applyOut, "Nothing is outstanding.",
+		"a convergence that created the whole schema leaves nothing behind")
+	assert.Contains(t, applyOut, "the schema embedded in v1.4.0",
+		"the CLI answered, so the report names this build as the schema it converged to")
+
+	// The plan is the other half of the direct path, and it takes a named
+	// source: the release the deploy is about to roll. The embedded files of
+	// this build are what the convergence just ran, so the same files have to
+	// report no outstanding statements.
+	plan := &StoragePlanCmd{
+		storageSchemaTargetFlags: storageSchemaTargetFlags{DSN: dsn},
+		storageSchemaSourceFlags: storageSchemaSourceFlags{SchemaDir: filepath.Join("..", "..", "schema", "postgres")},
+	}
+	planOut := captureStdout(func() {
+		require.NoError(t, plan.Run(t.Context(), globals), "exit 0 is how a pre-deploy gate reads 'this storage is ready'")
+	})
+	assert.Contains(t, planOut, "No schema changes detected.")
+	assert.Contains(t, planOut, "Database: schemabot", "the plan names the database the DSN addressed")
+	assert.NotContains(t, planOut, "+ applies", "nothing is outstanding against the schema that was just applied")
+}
+
+// An attended direct apply previews the convergence before it asks: the plan on
+// screen is a real read of the database the DSN names, and the schema it names
+// is this build's own, because a convergence has no flag to point it at another
+// release's.
+func TestStorageSchemaDirect_AttendedApplyPreviewsThisBuildsOwnSchema(t *testing.T) {
+	dsn, db := testutil.StartPostgres(t, "schemabot")
+	answerPrompt(t, "yes")
+
+	apply := &StorageApplyCmd{storageSchemaTargetFlags: storageSchemaTargetFlags{DSN: dsn}}
+	out := captureStdout(func() {
+		require.NoError(t, apply.Run(t.Context(), &Globals{Version: "v1.4.0"}))
+	})
+
+	assert.Contains(t, out, "the schema embedded in v1.4.0",
+		"the preview names the release whose schema the convergence is about to run")
+	assert.Contains(t, out, "Do you want to apply these changes to schemabot",
+		"an attended run asks before it converges, and names what it would converge")
+	assert.True(t, testutil.PostgresTableExists(t, db, "public", "applies"),
+		"the answered prompt converged the database")
+}
+
+// A direct connection reports the database it was pointed at, and the dialect
+// comes from the DSN's own form with no flag and no server to ask. A plan is
+// read-only, so this runs against a database with no storage schema at all and
+// the whole schema reports as outstanding — the state an operator is in when
+// the bootstrap that would create it is what is failing.
+func TestStorageSchemaDirect_PlanReportsTheWholeSchemaOutstanding(t *testing.T) {
+	dsn, _ := testutil.StartPostgres(t, "schemabot")
+
+	plan := &StoragePlanCmd{
+		storageSchemaTargetFlags: storageSchemaTargetFlags{DSN: dsn},
+		storageSchemaSourceFlags: storageSchemaSourceFlags{SchemaDir: filepath.Join("..", "..", "schema", "postgres")},
+	}
+	var err error
+	out := captureStdout(func() {
+		err = plan.Run(t.Context(), &Globals{Version: "v1.4.0"})
+	})
+
+	require.Error(t, err, "outstanding statements exit non-zero so a pre-deploy gate can tell them from a converged database")
+	assert.Equal(t, ExitStorageSchemaOutstanding, ExitCodeFor(err),
+		"outstanding work has its own status, distinct from a failed read")
+	assert.Contains(t, out, "+ applies", "the whole schema is outstanding, table by table")
+	assert.Contains(t, out, "Database: schemabot", "the report names the database the DSN addressed")
+	assert.Contains(t, out, "PostgreSQL Schema Change Plan",
+		"the dialect came from the DSN's own form, with no --dialect and no server to ask")
+}
diff --git a/pkg/cmd/commands/storage_schema_render.go b/pkg/cmd/commands/storage_schema_render.go
index 250c4bb6f..05eee9750 100644
--- a/pkg/cmd/commands/storage_schema_render.go
+++ b/pkg/cmd/commands/storage_schema_render.go
@@ -16,11 +16,11 @@ import (
 // per-table sections with their change symbols, the disclosure of changes that
 // will not run, and the plan summary line.
 //
-// The database is unusual — SchemaBot's own bookkeeping storage rather than one
-// an operator asked to change — but nothing an operator does with the output is
-// unusual, so nothing about the output should be. Whoever can read `plan` can
-// read `storage plan`, which matters because the second one is read during an
-// incident and the first one is read every day.
+// The question being asked is the same one either way — here is a live
+// database, here is a desired schema, here is the DDL between them — and only
+// the target differs. So anyone who can read `plan` can read `storage plan`,
+// which matters because `plan` is read every day and `storage plan` is read
+// during an incident.
 
 // writeStorageSchemaHeader writes the header box and the environment heading.
 //
diff --git a/pkg/cmd/commands/storage_schema_source.go b/pkg/cmd/commands/storage_schema_source.go
index f86c9ae1e..9151002e3 100644
--- a/pkg/cmd/commands/storage_schema_source.go
+++ b/pkg/cmd/commands/storage_schema_source.go
@@ -92,13 +92,15 @@ func (f *storageSchemaSourceFlags) validateSource() error {
 // converge storage ahead of a deploy without the fleet's own boots then
 // disagreeing with it (AV-9). A convergence to files named on the command line
 // would be a second implementation of the one path that must not have two.
-func storageSchemaSourceRefusal(schemaDir, release string) error {
+func storageSchemaSourceRefusal(schemaDir, release, repo string) error {
 	selector := ""
 	switch {
 	case strings.TrimSpace(release) != "":
 		selector = "--release"
 	case strings.TrimSpace(schemaDir) != "":
 		selector = "--schema-dir"
+	case strings.TrimSpace(repo) != "":
+		selector = "--release-repo"
 	default:
 		return nil
 	}
diff --git a/pkg/cmd/commands/storage_schema_source_test.go b/pkg/cmd/commands/storage_schema_source_test.go
index 78536d1cc..17a146727 100644
--- a/pkg/cmd/commands/storage_schema_source_test.go
+++ b/pkg/cmd/commands/storage_schema_source_test.go
@@ -444,15 +444,22 @@ func TestEscapePathSegments(t *testing.T) {
 // to converge a release named, since that is what the operator is reaching
 // for.
 func TestStorageSchemaSourceRefusal(t *testing.T) {
-	require.NoError(t, storageSchemaSourceRefusal("", ""))
+	require.NoError(t, storageSchemaSourceRefusal("", "", ""))
 
-	release := storageSchemaSourceRefusal("", "v1.4.0")
+	release := storageSchemaSourceRefusal("", "v1.4.0", "")
 	require.Error(t, release)
 	assert.Contains(t, release.Error(), "--release cannot be used with a convergence")
 	assert.Contains(t, release.Error(), "run that release's binary")
 	assert.Contains(t, release.Error(), "storage plan")
 
-	dir := storageSchemaSourceRefusal("./schema/mysql", "")
+	dir := storageSchemaSourceRefusal("./schema/mysql", "", "")
 	require.Error(t, dir)
 	assert.Contains(t, dir.Error(), "--schema-dir cannot be used with a convergence")
+
+	// --release-repo is the flag most likely to be left on the line after the
+	// one it modifies has been dropped, so it earns the same refusal rather
+	// than Kong's bare "unknown flag".
+	repo := storageSchemaSourceRefusal("", "", "block/schemabot")
+	require.Error(t, repo)
+	assert.Contains(t, repo.Error(), "--release-repo cannot be used with a convergence")
 }
diff --git a/pkg/cmd/commands/storage_schema_test.go b/pkg/cmd/commands/storage_schema_test.go
new file mode 100644
index 000000000..1cb3e6ac0
--- /dev/null
+++ b/pkg/cmd/commands/storage_schema_test.go
@@ -0,0 +1,562 @@
+package commands
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/alecthomas/kong"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/api"
+	"github.com/block/schemabot/pkg/apitypes"
+)
+
+// A target is either the API path or a direct connection, never a blend of the
+// two. Every refusal below is a combination that would otherwise have read a
+// database the operator did not name — the failure mode these flags exist to
+// prevent.
+func TestStorageSchemaTargetFlags_Validate(t *testing.T) {
+	tests := []struct {
+		name    string
+		flags   storageSchemaTargetFlags
+		wantErr string
+	}{
+		{
+			name:  "no flags reads the storage of the server the CLI is pointed at",
+			flags: storageSchemaTargetFlags{},
+		},
+		{
+			name:  "deployment with its environment",
+			flags: storageSchemaTargetFlags{Deployment: "west", Environment: "production"},
+		},
+		{
+			name:  "a DSN on its own",
+			flags: storageSchemaTargetFlags{DSN: "root@tcp(127.0.0.1:3306)/schemabot"},
+		},
+		{
+			name:  "a config file on its own",
+			flags: storageSchemaTargetFlags{Config: "/etc/schemabot/config.yaml"},
+		},
+		{
+			name:    "a deployment without an environment names no single endpoint",
+			flags:   storageSchemaTargetFlags{Deployment: "west"},
+			wantErr: "needs -e <environment>",
+		},
+		{
+			name:    "an environment without a deployment selects nothing",
+			flags:   storageSchemaTargetFlags{Environment: "production"},
+			wantErr: "without --deployment",
+		},
+		{
+			name:    "a deployment cannot be read through a direct DSN",
+			flags:   storageSchemaTargetFlags{Deployment: "west", Environment: "production", DSN: "root@tcp(127.0.0.1:3306)/schemabot"},
+			wantErr: "--deployment cannot be combined with a direct connection",
+		},
+		{
+			name:    "an environment cannot be combined with a direct DSN",
+			flags:   storageSchemaTargetFlags{Environment: "production", Config: "/etc/schemabot/config.yaml"},
+			wantErr: "-e cannot be combined with a direct connection",
+		},
+		{
+			name:    "a dialect assertion has nothing to apply to through the API",
+			flags:   storageSchemaTargetFlags{Dialect: "postgres"},
+			wantErr: "--dialect only applies to a direct connection",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.flags.validate()
+			if tc.wantErr == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tc.wantErr)
+		})
+	}
+}
+
+// A plan names the schema it compares the database against: one selector is
+// required, and naming two is refused by the parser. The plan would say which
+// schema it used either way; requiring the flag is what makes the operator
+// decide before they read the plan rather than after.
+func TestStoragePlanCmd_RequiresADesiredSchema(t *testing.T) {
+	cmd := &StoragePlanCmd{
+		storageSchemaTargetFlags: storageSchemaTargetFlags{DSN: "root@tcp(127.0.0.1:3306)/schemabot"},
+	}
+	err := cmd.Run(t.Context(), &Globals{})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing flags: --release=STRING or --schema-dir=STRING")
+
+	parse := func(args ...string) error {
+		var cli struct {
+			Plan StoragePlanCmd `cmd:"" name:"plan"`
+		}
+		parser, err := kong.New(&cli, kong.Name("schemabot"))
+		require.NoError(t, err)
+		_, err = parser.Parse(append([]string{"plan"}, args...))
+		return err
+	}
+	require.NoError(t, parse("--release", "v1.4.0"))
+	require.NoError(t, parse("--schema-dir", "."))
+
+	err = parse("--release", "v1.4.0", "--schema-dir", ".")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "--release and --schema-dir can't be used together")
+}
+
+// Destructive storage statements are permitted with --allow-unsafe, the flag
+// the rest of the CLI already uses for destructive changes, and the convergence
+// is the only command that takes it. The plan discloses those statements and
+// names the flag; seeing them as statements that will run means running the
+// apply, which is where the normal plan/apply flow puts the same question.
+func TestStorageSchemaCommands_PermitDestructiveStatementsWithAllowUnsafe(t *testing.T) {
+	parse := func(command string, target any, args ...string) error {
+		parser, err := kong.New(target, kong.Name("schemabot"))
+		require.NoError(t, err)
+		_, err = parser.Parse(append([]string{command}, args...))
+		return err
+	}
+
+	var applyCLI struct {
+		Apply StorageApplyCmd `cmd:"" name:"apply"`
+	}
+	require.NoError(t, parse("apply", &applyCLI, "--dsn", "postgres://localhost/schemabot", "--allow-unsafe"))
+	assert.True(t, applyCLI.Apply.AllowUnsafe, "the flag has to reach the convergence that acts on it")
+
+	var planCLI struct {
+		Plan StoragePlanCmd `cmd:"" name:"plan"`
+	}
+	err := parse("plan", &planCLI, "--schema-dir", ".", "--allow-unsafe")
+	require.Error(t, err, "the plan runs nothing, so there is no consent for it to take")
+	assert.Contains(t, err.Error(), "unknown flag --allow-unsafe")
+}
+
+// Passing a DSN source is what selects the direct path, so a command can tell
+// which path it is on without inspecting what happened to resolve.
+func TestStorageSchemaTargetFlags_Direct(t *testing.T) {
+	assert.False(t, (&storageSchemaTargetFlags{}).direct())
+	assert.False(t, (&storageSchemaTargetFlags{Deployment: "west", Environment: "production"}).direct())
+	assert.True(t, (&storageSchemaTargetFlags{DSN: "root@tcp(127.0.0.1:3306)/schemabot"}).direct())
+	assert.True(t, (&storageSchemaTargetFlags{Config: "/etc/schemabot/config.yaml"}).direct())
+
+	// A malformed direct source stays on the direct path, where it is refused
+	// by name. Reading it as "no DSN" would quietly report the storage of the
+	// server the CLI happens to point at instead of the one addressed.
+	assert.True(t, (&storageSchemaTargetFlags{DSN: "   "}).direct())
+	_, err := resolveStorageTarget("   ", "", "")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "--dsn contains only whitespace")
+}
+
+// A refused destructive statement is not counted as applied, and is not a
+// failure either: the surplus state stays in place on purpose.
+func TestStorageSchemaConvergenceOutcome(t *testing.T) {
+	require.NoError(t, storageSchemaConvergenceOutcome(
+		&apitypes.StorageSchemaReport{Database: "schemabot", Dialect: "mysql", Converged: true}))
+	require.NoError(t, storageSchemaConvergenceOutcome(&apitypes.StorageSchemaReport{
+		Database:    "schemabot",
+		Dialect:     "mysql",
+		Destructive: []apitypes.StorageSchemaStatement{{Table: "checks", DDL: "DROP TABLE `checks_old`", Reason: "drops a table"}},
+	}))
+
+	// Manual remediation aborts the whole convergence, so an unattended run has
+	// to fail rather than exit 0 having converged nothing.
+	err := storageSchemaConvergenceOutcome(&apitypes.StorageSchemaReport{
+		Database: "schemabot",
+		Host:     "db-1.example",
+		Dialect:  "postgres",
+		Manual: []apitypes.StorageSchemaStatement{{
+			Table:  "checks",
+			DDL:    `ALTER TABLE "checks" ADD COLUMN "head_sha" varchar(64) NOT NULL`,
+			Reason: "column is NOT NULL without a DEFAULT",
+		}},
+	})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "schemabot on db-1.example (postgres) was not converged")
+	assert.Contains(t, err.Error(), "1 change(s) need manual remediation")
+}
+
+// storageSchemaTestServer answers the two storage schema routes with the
+// reports a test hands it, and records the routes the command called. What was
+// called is half of what these tests assert: a command that skips the apply is
+// indistinguishable from one that ran it, if only the output is read.
+func storageSchemaTestServer(t *testing.T, plan *apitypes.StorageSchemaReport, applied, remaining *apitypes.StorageSchemaReport) (endpoint string, routes *[]string) {
+	t.Helper()
+	called := make([]string, 0, 4)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called = append(called, r.Method+" "+r.URL.Path)
+		w.Header().Set("Content-Type", "application/json")
+		switch r.URL.Path {
+		case "/api/storage/schema/plan":
+			assert.NoError(t, json.NewEncoder(w).Encode(apitypes.StorageSchemaPlanResponse{Report: plan}))
+		case "/api/storage/schema/apply":
+			assert.NoError(t, json.NewEncoder(w).Encode(apitypes.StorageSchemaApplyResponse{Planned: applied, Remaining: remaining}))
+		default:
+			http.Error(w, "unexpected request", http.StatusBadRequest)
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server.URL, &called
+}
+
+// answerPrompt feeds a confirmation prompt its answer, so an interactive run
+// can be driven from a test.
+func answerPrompt(t *testing.T, answer string) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "confirmation")
+	require.NoError(t, os.WriteFile(path, []byte(answer+"\n"), 0o600))
+	input, err := os.Open(path)
+	require.NoError(t, err)
+	original := os.Stdin
+	os.Stdin = input
+	t.Cleanup(func() {
+		os.Stdin = original
+		require.NoError(t, input.Close())
+	})
+}
+
+// storageSchemaCheckoutDir writes a checkout of a release's schema files, which
+// is what --schema-dir reads, so a diff needs no network to name its desired
+// side.
+func storageSchemaCheckoutDir(t *testing.T) string {
+	t.Helper()
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "applies.sql"),
+		[]byte("CREATE TABLE `applies` (`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY)"), 0o600))
+	return dir
+}
+
+// The exit status is the machine-readable half of a diff's answer. A converged
+// storage exits 0 and an outstanding one exits 2, and nothing is printed under
+// the report to explain the status: 2 is not a failure, so an "Error:" line
+// below a report that reads correctly would misrepresent it.
+func TestStoragePlanCmd_ExitStatusSaysWhetherWorkIsOutstanding(t *testing.T) {
+	dir := storageSchemaCheckoutDir(t)
+
+	outstanding := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema files in " + dir,
+		Outstanding: []apitypes.StorageSchemaStatement{
+			{Table: "applies", Operation: "alter_table", DDL: "ALTER TABLE `applies` ADD COLUMN `caller` varchar(255) NOT NULL DEFAULT ''"},
+		},
+	}
+	endpoint, routes := storageSchemaTestServer(t, outstanding, nil, nil)
+
+	var err error
+	out := captureStdout(func() {
+		cmd := StoragePlanCmd{storageSchemaSourceFlags: storageSchemaSourceFlags{SchemaDir: dir}}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+	})
+	require.Error(t, err)
+	assert.Equal(t, ExitStorageSchemaOutstanding, ExitCodeFor(err))
+	assert.ErrorIs(t, err, ErrSilent, "the report is the answer; an error line under it would read as a failed read")
+	assert.Contains(t, stripAnsi(out), "~ applies")
+	assert.Equal(t, []string{"POST /api/storage/schema/plan"}, *routes)
+
+	converged := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema files in " + dir,
+		Converged:    true,
+	}
+	cleanEndpoint, _ := storageSchemaTestServer(t, converged, nil, nil)
+	out = captureStdout(func() {
+		cmd := StoragePlanCmd{storageSchemaSourceFlags: storageSchemaSourceFlags{SchemaDir: dir}}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: cleanEndpoint})
+	})
+	require.NoError(t, err, "a converged storage is status 0, which is what a pre-deploy gate reads")
+	assert.Contains(t, out, "No schema changes detected")
+}
+
+// An interactive apply converges what --auto-approve would, including when the
+// preview found the catalog already matching. The bootstrap also clears the
+// schema change engine's leftover tables, which a catalog diff cannot see, so
+// stopping at a converged preview would leave them on the database and make the
+// command an operator reaches for mid-incident do less than the unattended one.
+func TestStorageApplyCmd_ConvergesAConvergedCatalogToo(t *testing.T) {
+	converged := &apitypes.StorageSchemaReport{
+		Dialect:      "postgres",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Converged:    true,
+	}
+	endpoint, routes := storageSchemaTestServer(t, converged, converged, converged)
+	answerPrompt(t, "yes")
+
+	var err error
+	out := captureStdout(func() {
+		cmd := StorageApplyCmd{}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+	})
+	require.NoError(t, err)
+	assert.Contains(t, out, "already matches", "the prompt asks for the run it is about to do, not for statements there are none of")
+	assert.Equal(t, []string{"POST /api/storage/schema/plan", "POST /api/storage/schema/apply"}, *routes,
+		"the convergence runs; a converged catalog is not a reason to skip the bootstrap")
+	assert.Contains(t, out, "Nothing is outstanding.")
+}
+
+// A declined confirmation converges nothing at all. The preview is read-only,
+// so the command has to leave the database exactly as it found it.
+func TestStorageApplyCmd_DeclinedConfirmationRunsNothing(t *testing.T) {
+	plan := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Outstanding: []apitypes.StorageSchemaStatement{
+			{Table: "applies", Operation: "alter_table", DDL: "ALTER TABLE `applies` ADD COLUMN `caller` varchar(255) NOT NULL DEFAULT ''"},
+		},
+	}
+	endpoint, routes := storageSchemaTestServer(t, plan, nil, nil)
+	answerPrompt(t, "no")
+
+	var err error
+	out := captureStdout(func() {
+		cmd := StorageApplyCmd{}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+	})
+	require.NoError(t, err)
+	assert.Contains(t, out, "Apply cancelled.")
+	assert.Equal(t, []string{"POST /api/storage/schema/plan"}, *routes)
+}
+
+// A preview carrying a manual-remediation entry refuses before the prompt.
+// Nothing would converge anyway — the bootstrap refuses the whole drift set
+// while one is outstanding — so asking for consent to a run that cannot happen
+// would be asking the operator to approve nothing.
+func TestStorageApplyCmd_ManualRemediationRefusesBeforeThePrompt(t *testing.T) {
+	plan := &apitypes.StorageSchemaReport{
+		Dialect:      "postgres",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Manual: []apitypes.StorageSchemaStatement{{
+			Table:     "checks",
+			Operation: "add_column",
+			DDL:       `ALTER TABLE "checks" ADD COLUMN "head_sha" varchar(64) NOT NULL`,
+			Reason:    "definition is NOT NULL without a DEFAULT",
+		}},
+	}
+	endpoint, routes := storageSchemaTestServer(t, plan, nil, nil)
+
+	var err error
+	out := captureStdout(func() {
+		cmd := StorageApplyCmd{}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+	})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "need manual remediation first")
+	assert.NotContains(t, out, "Only 'yes' will be accepted")
+	assert.Equal(t, []string{"POST /api/storage/schema/plan"}, *routes)
+}
+
+// A destructive statement nothing has permitted stops the convergence before it
+// runs, the way `apply` stops a destructive schema change: the plan is shown,
+// the statement is named, --allow-unsafe is named as the way through, and
+// nothing converges. The gate is in front of --auto-approve too, because
+// skipping the prompt is not consenting to destroy storage state.
+func TestStorageApplyCmd_DestructiveStatementsBlockTheConvergence(t *testing.T) {
+	plan := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Outstanding: []apitypes.StorageSchemaStatement{
+			{Table: "applies", Operation: "alter_table", DDL: "ALTER TABLE `applies` ADD COLUMN `caller` varchar(255) NOT NULL DEFAULT ''"},
+		},
+		Destructive: []apitypes.StorageSchemaStatement{
+			{Table: "stale_state", Operation: "drop_table", DDL: "DROP TABLE `stale_state`", Reason: "DROP TABLE destroys data"},
+		},
+	}
+
+	for _, tc := range []struct {
+		name        string
+		autoApprove bool
+	}{
+		{name: "attended"},
+		{name: "unattended", autoApprove: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			endpoint, routes := storageSchemaTestServer(t, plan, nil, nil)
+			// An answer is staged for the attended run to prove the prompt is
+			// never reached: a gate that asked first and blocked afterwards
+			// would consume this and still pass an output assertion.
+			answerPrompt(t, "yes")
+
+			var err error
+			out := captureStdout(func() {
+				cmd := StorageApplyCmd{AutoApprove: tc.autoApprove}
+				err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+			})
+
+			require.Error(t, err)
+			assert.ErrorIs(t, err, ErrSilent, "the plan carries the refusal, so an error line under it would repeat it")
+			assert.Equal(t, []string{"POST /api/storage/schema/plan"}, *routes,
+				"the convergence must not run; the safe remainder is not a reason to proceed past a refused DROP")
+			assert.Contains(t, out, "Apply blocked: 1 unsafe change(s) detected",
+				"the same heading a blocked schema change apply prints")
+			assert.Contains(t, out, "schemabot storage apply --allow-unsafe",
+				"the refusal names the command that permits what it refused")
+			assert.Contains(t, out, "1. stale_state: DROP TABLE destroys data")
+			assert.NotContains(t, out, "Only 'yes' will be accepted",
+				"consent to a convergence is not consent to destroy state, so the gate is in front of the prompt")
+		})
+	}
+}
+
+// A manual-remediation entry is the refusal an operator is told about, even
+// when destructive statements are outstanding too. It blocks the whole drift
+// set, so the destructive ones are not yet reachable, and offering consent for
+// them would name a flag that permits a DROP and converges nothing.
+func TestStorageApplyCmd_ManualRemediationOutranksTheDestructiveRefusal(t *testing.T) {
+	plan := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Destructive: []apitypes.StorageSchemaStatement{
+			{Table: "stale_state", Operation: "drop_table", DDL: "DROP TABLE `stale_state`", Reason: "DROP TABLE destroys data"},
+		},
+		Manual: []apitypes.StorageSchemaStatement{{
+			Table:     "checks",
+			Operation: "add_column",
+			DDL:       "ALTER TABLE `checks` ADD COLUMN `head_sha` varchar(64) NOT NULL",
+			Reason:    "definition is NOT NULL without a DEFAULT",
+		}},
+	}
+	for _, unattended := range []bool{false, true} {
+		name := "attended"
+		if unattended {
+			name = "unattended"
+		}
+		t.Run(name, func(t *testing.T) {
+			endpoint, routes := storageSchemaTestServer(t, plan, nil, nil)
+
+			var err error
+			out := captureStdout(func() {
+				cmd := StorageApplyCmd{AutoApprove: unattended}
+				err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+			})
+
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "need manual remediation first",
+				"the refusal an operator can act on is named, not left to an exit status")
+			assert.Contains(t, out, "Needs manual remediation",
+				"the error says these are listed above, so they are listed on both paths")
+			assert.Contains(t, out, "checks: definition is NOT NULL without a DEFAULT",
+				"the entry names the table and what has to be resolved by hand")
+			assert.NotContains(t, out, "Apply blocked",
+				"a manual entry makes the destructive statement unreachable, so refusing it separately would name the wrong remedy")
+			assert.Equal(t, []string{"POST /api/storage/schema/plan"}, *routes, "nothing converges either way")
+		})
+	}
+}
+
+// The command a refusal offers addresses the same storage database the refusal
+// is about, so it carries the target flags forward. It never carries the DSN:
+// that is a credential, and the refusal is printed to a terminal.
+func TestStorageApplyCmd_RerunCommandAddressesTheSameTarget(t *testing.T) {
+	tests := []struct {
+		name  string
+		cmd   StorageApplyCmd
+		rerun string
+	}{
+		{
+			name:  "the server's own storage",
+			rerun: "storage apply --allow-unsafe",
+		},
+		{
+			name:  "a data plane's storage",
+			cmd:   StorageApplyCmd{storageSchemaTargetFlags: storageSchemaTargetFlags{Deployment: "shard-a", Environment: "production"}},
+			rerun: "storage apply --deployment shard-a -e production --allow-unsafe",
+		},
+		{
+			name:  "resolved from a config file",
+			cmd:   StorageApplyCmd{storageSchemaTargetFlags: storageSchemaTargetFlags{Config: "/etc/schemabot/config.yaml"}},
+			rerun: "storage apply --config /etc/schemabot/config.yaml --allow-unsafe",
+		},
+		{
+			name: "a DSN is named, not repeated",
+			cmd: StorageApplyCmd{storageSchemaTargetFlags: storageSchemaTargetFlags{
+				DSN:     "postgres://schemabot:hunter2@db-1.example:5432/schemabot",
+				Dialect: "postgres",
+			}},
+			rerun: "storage apply --dsn <the same DSN> --dialect postgres --allow-unsafe",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			rerun := tc.cmd.rerunWithAllowUnsafe()
+			assert.Equal(t, tc.rerun, rerun)
+			assert.NotContains(t, rerun, "hunter2", "a suggested command must not print the storage credentials")
+		})
+	}
+}
+
+// Destructive statements the target has permitted are not gated: the report
+// says they will run, and blocking them would narrow a standing storage policy
+// the CLI has no business narrowing (AV-9).
+func TestStorageApplyCmd_PermittedDestructiveStatementsConverge(t *testing.T) {
+	permitted := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Destructive: []apitypes.StorageSchemaStatement{
+			{Table: "stale_state", Operation: "drop_table", DDL: "DROP TABLE `stale_state`", Reason: "DROP TABLE destroys data"},
+		},
+		DestructiveAllowed: true,
+	}
+	converged := &apitypes.StorageSchemaReport{
+		Dialect:      "mysql",
+		Database:     "schemabot",
+		Host:         "db-1.example",
+		SchemaSource: "the schema embedded in v1.4.0",
+		Converged:    true,
+	}
+	endpoint, routes := storageSchemaTestServer(t, permitted, permitted, converged)
+
+	var err error
+	out := captureStdout(func() {
+		cmd := StorageApplyCmd{AllowUnsafe: true, AutoApprove: true}
+		err = cmd.Run(t.Context(), &Globals{Endpoint: endpoint})
+	})
+	require.NoError(t, err)
+	assert.Equal(t, []string{"POST /api/storage/schema/plan", "POST /api/storage/schema/apply"}, *routes)
+	assert.Contains(t, out, "Nothing is outstanding.")
+}
+
+// Both halves of a direct convergence name the release that ran it, the same
+// way the preview named it. Stamping only the version would leave the result
+// header carrying the placeholder the preview had already expanded, so one run
+// would describe its own schema two ways.
+func TestAttributeStorageSchemaConvergence(t *testing.T) {
+	planned := &api.StorageSchemaReport{Outstanding: []api.StorageSchemaStatement{{Table: "applies"}}}
+	remaining := &api.StorageSchemaReport{}
+
+	attributeStorageSchemaConvergence("v1.4.0", planned, remaining)
+
+	assert.Equal(t, "v1.4.0", planned.Version)
+	assert.Equal(t, "the schema embedded in v1.4.0", planned.SchemaSource)
+	assert.Equal(t, "v1.4.0", remaining.Version)
+	assert.Equal(t, "the schema embedded in v1.4.0", remaining.SchemaSource,
+		"the half an operator reads last describes the same schema as the half they read first")
+}
+
+// An error names the storage that was being read. A failure that said only
+// that a read failed would leave an operator unable to tell whether they had
+// reached the deployment they asked for.
+func TestStorageSchemaTargetSuffix(t *testing.T) {
+	assert.Empty(t, storageSchemaTargetSuffix("", ""))
+	assert.Equal(t, " for deployment west in production", storageSchemaTargetSuffix("west", "production"))
+}
diff --git a/pkg/cmd/commands/storage_target.go b/pkg/cmd/commands/storage_target.go
index eb3100ec9..172ace1a0 100644
--- a/pkg/cmd/commands/storage_target.go
+++ b/pkg/cmd/commands/storage_target.go
@@ -101,6 +101,35 @@ func resolveStorageTarget(dsnFlag, configFlag, dialectFlag string) (*storageTarg
 		return &storageTarget{dsn: directDSN, dialect: dialect, source: "--dsn flag"}, nil
 	}
 
+	configured, err := resolveStorageConfig(configFlag)
+	if err != nil {
+		return nil, err
+	}
+	if asserted := strings.TrimSpace(strings.ToLower(dialectFlag)); asserted != "" && schema.Dialect(asserted) != configured.dialect {
+		return nil, fmt.Errorf("--dialect says %q but %s configures %q storage; drop --dialect, which only applies to a DSN passed with --dsn", asserted, configured.source, configured.dialect)
+	}
+	return configured.target()
+}
+
+// storageConfig is a server config resolved as far as its storage family,
+// before the DSN itself is fetched.
+//
+// The two steps are separate because a caller can be done after the first one.
+// Fetching the DSN is not free of consequence — storage.dsn_from reads a
+// secret, which is a call to someone else's system with its own audit trail —
+// and a command that only applies to one family has already reached its answer
+// once the family is known. Refusing there names the family as the reason and
+// fetches nothing; refusing after the fetch would report whatever went wrong
+// resolving a DSN the command was never going to use.
+type storageConfig struct {
+	cfg     *api.ServerConfig
+	dialect schema.Dialect
+	source  string
+}
+
+// resolveStorageConfig loads the server config named by --config, or by
+// $SCHEMABOT_CONFIG_FILE, and reads the storage dialect it states.
+func resolveStorageConfig(configFlag string) (*storageConfig, error) {
 	configPath := configFlag
 	source := fmt.Sprintf("server config %s", configPath)
 	if configPath == "" {
@@ -126,31 +155,34 @@ func resolveStorageTarget(dsnFlag, configFlag, dialectFlag string) (*storageTarg
 	if err != nil {
 		return nil, fmt.Errorf("resolve storage dialect from %s: %w", source, err)
 	}
-	if asserted := strings.TrimSpace(strings.ToLower(dialectFlag)); asserted != "" && schema.Dialect(asserted) != dialect {
-		return nil, fmt.Errorf("--dialect says %q but %s configures %q storage; drop --dialect, which only applies to a DSN passed with --dsn", asserted, source, dialect)
-	}
+	return &storageConfig{cfg: cfg, dialect: dialect, source: source}, nil
+}
 
-	dsn, err := cfg.StorageDSN()
+// target fetches the configured DSN and returns the storage database it names,
+// under the policy the config states for it.
+func (c *storageConfig) target() (*storageTarget, error) {
+	dsn, err := c.cfg.StorageDSN()
 	if err != nil {
-		return nil, fmt.Errorf("resolve storage DSN from %s: %w", source, err)
+		return nil, fmt.Errorf("resolve storage DSN from %s: %w", c.source, err)
 	}
 	dsn = strings.TrimSpace(dsn)
 	if dsn == "" {
 		return nil, fmt.Errorf("storage DSN not configured (set --dsn, config storage.dsn or storage.dsn_from, STORAGE_DSN, or MYSQL_DSN)")
 	}
-	if cfg.Storage.DSN == "" && cfg.Storage.DSNFrom == nil {
+	source := c.source
+	if c.cfg.Storage.DSN == "" && c.cfg.Storage.DSNFrom == nil {
 		if strings.TrimSpace(os.Getenv("STORAGE_DSN")) != "" {
 			source = "STORAGE_DSN environment variable"
 		} else if strings.TrimSpace(os.Getenv("MYSQL_DSN")) != "" {
 			source = "MYSQL_DSN environment variable"
 		}
 	}
-	statementTimeout := cfg.Postgres.StatementTimeoutOrDefault()
+	statementTimeout := c.cfg.Postgres.StatementTimeoutOrDefault()
 	return &storageTarget{
 		dsn:                      dsn,
-		dialect:                  dialect,
+		dialect:                  c.dialect,
 		source:                   source,
-		allowDestructive:         cfg.Storage.AllowDestructiveSchemaChanges,
+		allowDestructive:         c.cfg.Storage.AllowDestructiveSchemaChanges,
 		postgresStatementTimeout: &statementTimeout,
 	}, nil
 }
diff --git a/pkg/cmd/commands/storage_test.go b/pkg/cmd/commands/storage_test.go
index d0bb9063c..a974a8803 100644
--- a/pkg/cmd/commands/storage_test.go
+++ b/pkg/cmd/commands/storage_test.go
@@ -84,6 +84,26 @@ storage:
 	require.ErrorContains(t, err, `only applies to "postgres" storage`)
 }
 
+// A command that only applies to one storage family refuses on the family,
+// before the DSN is fetched. Fetching is not free of consequence — storage
+// dsn_from reads a secret, a call to another system with its own audit trail —
+// and the refusal that helps names the family rather than whatever went wrong
+// resolving a connection this command was never going to open.
+func TestResolveStorageDSN_RefusesTheDialectBeforeFetchingTheDSN(t *testing.T) {
+	path := writeStorageTestConfig(t, `
+storage:
+  dialect: mysql
+  dsn_from:
+    config_ref: file:/nonexistent/storage-connection.yaml
+    username: schemabot
+    password_ref: file:/nonexistent/storage-password
+`)
+	_, _, err := resolveStorageDSN("", path, "the identity sequence resync")
+	require.ErrorContains(t, err, `the identity sequence resync only applies to "postgres" storage`)
+	assert.NotContains(t, err.Error(), "nonexistent",
+		"the secret is never read, so nothing about resolving it can be the reported cause")
+}
+
 func TestResolveStorageDSN_EmptyConfigDSN(t *testing.T) {
 	t.Setenv("STORAGE_DSN", "")
 	t.Setenv("MYSQL_DSN", "")
diff --git a/pkg/cmd/main.go b/pkg/cmd/main.go
index d8ba58a56..8ee6d2f30 100644
--- a/pkg/cmd/main.go
+++ b/pkg/cmd/main.go
@@ -58,7 +58,7 @@ type CLI struct {
 	Settings   commands.SettingsCmd   `cmd:"" help:"View or update schema change settings"`
 	Webhooks   commands.WebhooksCmd   `cmd:"" help:"Manage GitHub App webhook deliveries"`
 	Checks     commands.ChecksCmd     `cmd:"" help:"Manage SchemaBot Check Runs on PRs"`
-	Storage    commands.StorageCmd    `cmd:"" help:"Operate directly on SchemaBot's storage database"`
+	Storage    commands.StorageCmd    `cmd:"" help:"Inspect and maintain SchemaBot's own storage database"`
 	Local      commands.LocalCmd      `cmd:"" hidden:"" help:"Internal local runtime host"`
 	Serve      commands.ServeCmd      `cmd:"" help:"Start the SchemaBot HTTP API server"`
 }
@@ -94,7 +94,7 @@ func main() {
 	// Hosting a local runtime does not use a remote profile or its credentials.
 	var localEndpoint string
 	if !strings.HasPrefix(ctx.Command(), "local ") && ctx.Command() != "init" {
-		if usesLocalRuntime(ctx.Command()) {
+		if usesLocalRuntime(ctx.Command(), &cli) {
 			localCtx, cancelLocal := context.WithTimeout(context.Background(), 30*time.Second)
 			connection, err := client.ResolveLocalConnection(localCtx, cli.Endpoint, cli.Profile, cli.Token, version)
 			cancelLocal()
@@ -165,8 +165,9 @@ func main() {
 	cancelRun()
 	if err != nil {
 		// ErrSilent means the error was already displayed, so only the status
-		// is left to report. A command that asked for a particular status gets
-		// it (see commands.ExitCodeFor); everything else is the usual 1.
+		// is left to report. A command may ask for its own status when a caller
+		// scripting it needs to tell two successful-but-different outcomes
+		// apart (see commands.ExitCodeFor); anything else exits 1.
 		if !errors.Is(err, commands.ErrSilent) {
 			fmt.Fprintf(os.Stderr, "\033[31mError: %v\033[0m\n", err)
 		}
@@ -175,7 +176,7 @@ func main() {
 }
 
 // Only commands that use the schema API may implicitly start a selected runtime.
-func usesLocalRuntime(command string) bool {
+func usesLocalRuntime(command string, cli *CLI) bool {
 	parts := strings.Fields(command)
 	if len(parts) == 0 {
 		return false
@@ -183,6 +184,12 @@ func usesLocalRuntime(command string) bool {
 	switch parts[0] {
 	case "plan", "onboard", "pull", "apply", "progress", "cutover", "stop", "cancel", "start", "release", "revert", "skip-revert", "rollback", "databases", "unlock", "locks", "logs", "status", "list-plans":
 		return true
+	case "storage":
+		// A storage subcommand reaches its database either through the API or
+		// by opening it directly, and only the first needs an endpoint. The
+		// direct path exists for when no server is up, so starting a runtime
+		// for it would be work the operator asked this command to avoid.
+		return len(parts) > 1 && cli.Storage.UsesAPI(parts[1])
 	default:
 		return false
 	}
diff --git a/pkg/cmd/main_test.go b/pkg/cmd/main_test.go
index 922159295..12e475d70 100644
--- a/pkg/cmd/main_test.go
+++ b/pkg/cmd/main_test.go
@@ -36,3 +36,50 @@ func TestStorageResyncIdentitySequencesIsInvocable(t *testing.T) {
 	_, err = parser.Parse([]string{"storage", "resync-identity-sequences", "--dsn", "postgres://user@localhost:5432/db"})
 	require.NoError(t, err)
 }
+
+// A storage schema command routed through the API needs an endpoint, and with
+// a local runtime selected that endpoint is one the CLI has to start first.
+// The direct path needs none: it exists for when no server is up, so starting
+// a runtime for it would be exactly the work the operator reached for --dsn to
+// avoid. The maintenance commands never call the API at all.
+func TestUsesLocalRuntime_StorageSubcommands(t *testing.T) {
+	tests := []struct {
+		name string
+		args []string
+		want bool
+	}{
+		{name: "plan through the API", args: []string{"storage", "plan", "--release", "v1.4.0"}, want: true},
+		{name: "plan of a data plane's storage", args: []string{"storage", "plan", "--release", "v1.4.0", "--deployment", "west", "-e", "production"}, want: true},
+		{name: "apply through the API", args: []string{"storage", "apply"}, want: true},
+		{name: "plan against a DSN", args: []string{"storage", "plan", "--release", "v1.4.0", "--dsn", "postgres://user@localhost:5432/db"}, want: false},
+		{name: "apply against a server config", args: []string{"storage", "apply", "--config", "/etc/schemabot/config.yaml"}, want: false},
+		{name: "a maintenance command", args: []string{"storage", "resync-identity-sequences", "--dsn", "postgres://user@localhost:5432/db"}, want: false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var cli CLI
+			parser, err := kong.New(&cli,
+				kong.Name("schemabot"),
+				kong.Writers(io.Discard, io.Discard),
+				kong.Vars{"cli_name": "schemabot"},
+			)
+			require.NoError(t, err)
+
+			ctx, err := parser.Parse(tc.args)
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, usesLocalRuntime(ctx.Command(), &cli))
+		})
+	}
+
+	// The commands that already resolved an endpoint keep doing so.
+	var cli CLI
+	parser, err := kong.New(&cli,
+		kong.Name("schemabot"),
+		kong.Writers(io.Discard, io.Discard),
+		kong.Vars{"cli_name": "schemabot"},
+	)
+	require.NoError(t, err)
+	ctx, err := parser.Parse([]string{"status", "apply_abc123"})
+	require.NoError(t, err)
+	assert.True(t, usesLocalRuntime(ctx.Command(), &cli))
+}