diff --git a/go/core/cli/cmd/kagent/main.go b/go/core/cli/cmd/kagent/main.go
index 78b6c062e..2ed5e2e6a 100644
--- a/go/core/cli/cmd/kagent/main.go
+++ b/go/core/cli/cmd/kagent/main.go
@@ -5,7 +5,9 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
+	"time"
 
 	cli "github.com/kagent-dev/kagent/go/core/cli/internal/cli/agent"
 	"github.com/kagent-dev/kagent/go/core/cli/internal/cli/envdoc"
@@ -13,7 +15,12 @@ import (
 	"github.com/kagent-dev/kagent/go/core/cli/internal/config"
 	"github.com/kagent-dev/kagent/go/core/cli/internal/profiles"
 	"github.com/kagent-dev/kagent/go/core/cli/internal/tui"
+	dbcli "github.com/kagent-dev/kagent/go/core/pkg/cli/db"
+	dbmigrate "github.com/kagent-dev/kagent/go/core/pkg/cli/db/migrate"
+	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
 	"github.com/spf13/cobra"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/clientcmd"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -449,11 +456,82 @@ Examples:
 	runCmd.Flags().StringVar(&runCfg.ProjectDir, "project-dir", "", "Project directory (default: current directory)")
 	runCmd.Flags().BoolVar(&runCfg.Build, "build", false, "Rebuild the Docker image before running")
 
-	rootCmd.AddCommand(installCmd, uninstallCmd, invokeCmd, bugReportCmd, versionCmd, dashboardCmd, getCmd, initCmd, buildCmd, deployCmd, addMcpCmd, runCmd, mcp.NewMCPCmd(), envdoc.NewEnvCmd())
+	rootCmd.AddCommand(installCmd, uninstallCmd, invokeCmd, bugReportCmd, versionCmd, dashboardCmd, getCmd, initCmd, buildCmd, deployCmd, addMcpCmd, runCmd, mcp.NewMCPCmd(), envdoc.NewEnvCmd(), dbcli.NewCommandFromFunc(migrationSources(cfg)))
 
 	return rootCmd
 }
 
+// vectorEnabledKey names two lookups that deliberately share it: the CLI's
+// own DATABASE_VECTOR_ENABLED env var (a local operator override), and the
+// controller-configmap key the chart renders — the value the controller pod
+// itself consumes via envFrom. Same name, two different places.
+const vectorEnabledKey = "DATABASE_VECTOR_ENABLED"
+
+// migrationSources resolves the built-in migration tracks when a db
+// subcommand runs (never during command construction, so unrelated commands
+// do no work and print no warnings). The vector track is gated, in order of
+// precedence, on: the DATABASE_VECTOR_ENABLED env var in the CLI's own
+// environment (explicit operator intent, works without a cluster), the
+// controller's configmap on the live cluster (the same value the server
+// reads), and finally the controller's default (enabled).
+func migrationSources(cfg *config.Config) dbmigrate.SourcesFunc {
+	return func(ctx context.Context) ([]migrations.Source, error) {
+		vectorEnabled := true
+		if v := os.Getenv(vectorEnabledKey); v != "" {
+			b, err := strconv.ParseBool(v)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "warning: invalid %s=%q; assuming true\n", vectorEnabledKey, v)
+			} else {
+				vectorEnabled = b
+			}
+		} else if b, ok := clusterVectorEnabled(ctx, cfg.Namespace); ok {
+			vectorEnabled = b
+		}
+		return migrations.BuiltinSources(vectorEnabled), nil
+	}
+}
+
+// clusterVectorEnabled reads the vectorEnabledKey entry from the controller
+// configmap in the given namespace (the same "kagent-controller" default
+// naming the rest of the CLI assumes) — the cluster-side counterpart of the
+// env-var override in migrationSources. When the value is used it says so on
+// stderr, naming the kubeconfig context it was read from — the lookup follows
+// the *current* context, so this is the operator's cue that the cluster and
+// their --db-url had better be the same install. Best-effort: reports
+// ok=false when no cluster is reachable, the configmap is absent, or the
+// value doesn't parse — callers fall back to the default.
+func clusterVectorEnabled(ctx context.Context, namespace string) (enabled, ok bool) {
+	k8sClient, err := cli.CreateKubernetesClient()
+	if err != nil {
+		return false, false
+	}
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+	var cm corev1.ConfigMap
+	if err := k8sClient.Get(ctx, client.ObjectKey{Namespace: namespace, Name: "kagent-controller"}, &cm); err != nil {
+		return false, false
+	}
+	b, err := strconv.ParseBool(cm.Data[vectorEnabledKey])
+	if err != nil {
+		return false, false
+	}
+	// Trailing blank line separates the notice from the command's stdout
+	// when both land on a terminal; piped stdout is unaffected.
+	fmt.Fprintf(os.Stderr, "resolved vector track from cluster context %q: configmap %s/kagent-controller has %s=%t (set %s to override)\n\n",
+		currentKubeContext(), namespace, vectorEnabledKey, b, vectorEnabledKey)
+	return b, true
+}
+
+// currentKubeContext names the kubeconfig context the CLI's Kubernetes client
+// dials, for operator-facing messages. Best-effort.
+func currentKubeContext() string {
+	raw, err := clientcmd.NewDefaultClientConfigLoadingRules().Load()
+	if err != nil || raw.CurrentContext == "" {
+		return "(current kubeconfig context)"
+	}
+	return raw.CurrentContext
+}
+
 func runInteractive(cmd *cobra.Command, args []string, cfg *config.Config) {
 	client := cfg.Client()
 
diff --git a/go/core/pkg/cli/db/db.go b/go/core/pkg/cli/db/db.go
new file mode 100644
index 000000000..55f53e37f
--- /dev/null
+++ b/go/core/pkg/cli/db/db.go
@@ -0,0 +1,54 @@
+// Package db hosts the `kagent db` parent command and its subcommands.
+// Currently only `migrate` is wired; future siblings attach here.
+package db
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+
+	"github.com/kagent-dev/kagent/go/core/pkg/cli/db/migrate"
+	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
+)
+
+// NewCommand returns the `db` parent command with `migrate` attached. The
+// given sources define the migration tracks the subcommands operate on, in
+// orchestrator registration order.
+func NewCommand(sources ...migrations.Source) *cobra.Command {
+	return newCommand(migrate.NewCommand(sources...))
+}
+
+// NewCommandFromFunc is NewCommand with deferred source resolution: fn runs
+// when a db subcommand executes, not while the command tree is constructed.
+// See migrate.NewCommandFromFunc.
+func NewCommandFromFunc(fn migrate.SourcesFunc) *cobra.Command {
+	return newCommand(migrate.NewCommandFromFunc(fn))
+}
+
+func newCommand(migrateCmd *cobra.Command) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "db",
+		Short: "Database operations (migrations, inspection)",
+	}
+	cmd.AddCommand(migrateCmd)
+
+	// Hide the root's API-oriented persistent flags from help across the
+	// entire `db` subtree. They target the kagent server / Kubernetes,
+	// but db commands talk to Postgres directly via --db-url.
+	//
+	// Hidden is a property of the flag itself (shared across the whole
+	// tree), so we can't flip it permanently. The HelpFunc override
+	// toggles it for the duration of the help render and restores it
+	// after. Children of `db` that don't set their own HelpFunc walk the
+	// parent chain and pick this one up.
+	cmd.SetHelpFunc(func(c *cobra.Command, args []string) {
+		for _, name := range []string{"config", "kagent-url", "namespace", "output-format", "timeout", "verbose"} {
+			if f := c.InheritedFlags().Lookup(name); f != nil {
+				f.Hidden = true
+				defer func(f *pflag.Flag) { f.Hidden = false }(f)
+			}
+		}
+		c.Root().HelpFunc()(c, args)
+	})
+
+	return cmd
+}
diff --git a/go/core/pkg/cli/db/migrate/migrate.go b/go/core/pkg/cli/db/migrate/migrate.go
new file mode 100644
index 000000000..57c355134
--- /dev/null
+++ b/go/core/pkg/cli/db/migrate/migrate.go
@@ -0,0 +1,792 @@
+// Package migrate exposes the `kagent db migrate` subcommand: out-of-band
+// application, rollback, and inspection of the database migration tracks.
+//
+// The command operates on the same migrations.Source values app.Start
+// registers and opens migrators through migrations.WithMigrator, so every
+// operation shares the orchestrator's schema handling, tracking tables, and
+// advisory-lock identity — a CLI invocation racing a booting server
+// serializes instead of corrupting state.
+//
+// Unlike the in-app startup path, the CLI never auto-recovers a dirty
+// tracking table: up, down, and goto refuse a dirty source and report the
+// `force` invocation that clears it. Deliberate recovery is the operator's
+// call.
+package migrate
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"regexp"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/golang-migrate/migrate/v4"
+	"github.com/spf13/cobra"
+
+	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
+)
+
+const (
+	// dbURLEnv is the controller's env var for --postgres-database-url
+	// (see app.LoadFromEnv); the CLI falls back to it so an operator with
+	// the controller's environment needs no extra flags.
+	dbURLEnv   = "POSTGRES_DATABASE_URL"
+	sourceFlag = "source"
+)
+
+// sourceNameRE constrains Source.Name to lowercase identifiers with
+// underscores or hyphens (e.g. "core", "vector", and hyphenated names from
+// downstream-registered sources). Name flows into `--source <name>` handling
+// and operator-facing messages; the regex keeps those strings predictable.
+var sourceNameRE = regexp.MustCompile(`^[a-z][a-z0-9_-]*$`)
+
+// SourcesFunc resolves the migration sources for a command invocation. It is
+// called when a subcommand actually executes — never at command construction —
+// so implementations may do real work (consult the environment, query a live
+// cluster) without taxing unrelated CLI commands. It is called at most once
+// per command execution; the result is memoized.
+type SourcesFunc func(ctx context.Context) ([]migrations.Source, error)
+
+type commandState struct {
+	dbURL     string
+	source    string
+	resolveFn SourcesFunc
+
+	// memoized getSources result
+	resolved   bool
+	sources    []migrations.Source
+	sourcesErr error
+}
+
+// getSources resolves and validates the source list on first use, memoizing
+// the result for the rest of the command execution.
+func (s *commandState) getSources(ctx context.Context) ([]migrations.Source, error) {
+	if s.resolved {
+		return s.sources, s.sourcesErr
+	}
+	s.resolved = true
+	srcs, err := s.resolveFn(ctx)
+	if err != nil {
+		s.sourcesErr = fmt.Errorf("resolve migration sources: %w", err)
+		return nil, s.sourcesErr
+	}
+	if err := validateSourceNames(srcs); err != nil {
+		s.sourcesErr = err
+		return nil, s.sourcesErr
+	}
+	s.sources = append([]migrations.Source(nil), srcs...)
+	return s.sources, nil
+}
+
+func validateSourceNames(sources []migrations.Source) error {
+	seen := map[string]bool{}
+	for _, source := range sources {
+		if !sourceNameRE.MatchString(source.Name) {
+			return fmt.Errorf("migration source Name=%q must match %s", source.Name, sourceNameRE.String())
+		}
+		if seen[source.Name] {
+			return fmt.Errorf("migration source %q configured twice; each source must have a unique Name", source.Name)
+		}
+		seen[source.Name] = true
+	}
+	return nil
+}
+
+// NewCommand returns the `migrate` parent command with all subcommands
+// attached, operating on the given sources in orchestrator registration
+// order. Panics on an invalid or duplicate source Name so misconfiguration
+// fails at wiring time, not mid-operation.
+func NewCommand(sources ...migrations.Source) *cobra.Command {
+	if err := validateSourceNames(sources); err != nil {
+		panic("migrate.NewCommand: " + err.Error())
+	}
+	static := append([]migrations.Source(nil), sources...)
+	return NewCommandFromFunc(func(context.Context) ([]migrations.Source, error) {
+		return static, nil
+	})
+}
+
+// NewCommandFromFunc is NewCommand with deferred source resolution: fn runs
+// when a subcommand executes, and an invalid source set surfaces as a command
+// error instead of a wiring-time panic. Use this when the source list depends
+// on state that shouldn't be consulted while merely constructing the command
+// tree (environment variables, a live cluster).
+func NewCommandFromFunc(fn SourcesFunc) *cobra.Command {
+	state := &commandState{resolveFn: fn}
+
+	cmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "Apply, roll back, and inspect database migrations",
+		Long: `Apply, roll back, and inspect database migrations independently
+of server startup. Reads ` + dbURLEnv + ` from the environment when
+--db-url is omitted.`,
+	}
+	cmd.PersistentFlags().StringVar(&state.dbURL, "db-url", "",
+		"PostgreSQL connection URL (defaults to value of "+dbURLEnv+" env var)")
+	cmd.PersistentFlags().StringVar(&state.source, sourceFlag, "",
+		"Migration source name for per-source ops (down/goto/force/version); inferred when only one source is registered. Not applicable to up or status — those aggregate across every registered source.")
+
+	cmd.AddCommand(newUpCmd(state))
+	cmd.AddCommand(newDownCmd(state))
+	cmd.AddCommand(newStatusCmd(state))
+	cmd.AddCommand(newVersionCmd(state))
+	cmd.AddCommand(newGotoCmd(state))
+	cmd.AddCommand(newForceCmd(state))
+	return cmd
+}
+
+func (s *commandState) resolveDSN() (string, error) {
+	dsn := strings.TrimSpace(s.dbURL)
+	if dsn == "" {
+		dsn = os.Getenv(dbURLEnv)
+	}
+	if dsn == "" {
+		return "", fmt.Errorf("database URL not set; pass --db-url or set %s", dbURLEnv)
+	}
+	return dsn, nil
+}
+
+// resolveSource picks the source for a per-source operation. With one source
+// registered it's returned directly; with more than one the operator must
+// pass --source and we report the registered set when they don't.
+func (s *commandState) resolveSource(ctx context.Context) (migrations.Source, error) {
+	srcs, err := s.getSources(ctx)
+	if err != nil {
+		return migrations.Source{}, err
+	}
+	if len(srcs) == 0 {
+		return migrations.Source{}, errors.New("no migration sources registered")
+	}
+	if len(srcs) == 1 {
+		if s.source != "" && s.source != srcs[0].Name {
+			return migrations.Source{}, fmt.Errorf("--source %q not registered; registered source: %s", s.source, srcs[0].Name)
+		}
+		return srcs[0], nil
+	}
+	if s.source == "" {
+		return migrations.Source{}, fmt.Errorf("registered sources: %s; pass --source", sourceNames(srcs))
+	}
+	for _, src := range srcs {
+		if src.Name == s.source {
+			return src, nil
+		}
+	}
+	return migrations.Source{}, fmt.Errorf("--source %q not registered; registered sources: %s", s.source, sourceNames(srcs))
+}
+
+func sourceNames(srcs []migrations.Source) string {
+	names := make([]string, len(srcs))
+	for i, s := range srcs {
+		names[i] = s.Name
+	}
+	return strings.Join(names, ", ")
+}
+
+// readVersion returns mg's highest applied version and whether the tracking
+// row is dirty (mid-failed-migration). ErrNilVersion (nothing applied) is
+// normalized to (0, false, nil).
+func readVersion(mg *migrate.Migrate) (uint, bool, error) {
+	v, dirty, err := mg.Version()
+	if err != nil {
+		if errors.Is(err, migrate.ErrNilVersion) {
+			return 0, false, nil
+		}
+		return 0, false, fmt.Errorf("read version: %w", err)
+	}
+	return v, dirty, nil
+}
+
+// ensureClean returns an actionable error when src's tracking table is dirty.
+// The CLI refuses to operate on a dirty source rather than auto-recovering
+// (the in-app startup path is the automatic tier; the CLI is the manual one).
+func ensureClean(src migrations.Source, mg *migrate.Migrate) error {
+	v, dirty, err := readVersion(mg)
+	if err != nil {
+		return err
+	}
+	if dirty {
+		return fmt.Errorf("source %q is dirty at version %d (a previous migration attempt failed and must be resolved manually); inspect the schema, then clear the flag with \"kagent db migrate force <version> --source %s\" where <version> is the version the schema actually reflects",
+			src.Name, v, src.Name)
+	}
+	return nil
+}
+
+// sourceFileVersions returns the (ascending-sorted) NNN versions parsed from
+// every NNN_name.up.sql file in src.FS/src.Dir.
+//
+// The set isn't required to be contiguous — gaps (e.g. a deleted 005) are
+// real and the missing numbers are treated as not-applicable-to-this-binary.
+// status/desync math goes via this list so the count of files and the highest
+// version stay distinct.
+func sourceFileVersions(src migrations.Source) ([]int, error) {
+	entries, err := fs.ReadDir(src.FS, src.Dir)
+	if err != nil {
+		return nil, fmt.Errorf("read migration dir %s: %w", src.Dir, err)
+	}
+	var versions []int
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+		name := e.Name()
+		if !strings.HasSuffix(name, ".up.sql") {
+			continue
+		}
+		parts := strings.SplitN(name, "_", 2)
+		if len(parts) != 2 {
+			continue
+		}
+		v, err := strconv.Atoi(parts[0])
+		if err != nil {
+			continue
+		}
+		versions = append(versions, v)
+	}
+	slices.Sort(versions)
+	return versions, nil
+}
+
+// lineRow carries per-source status data through the status command's text
+// and JSON output paths.
+type lineRow struct {
+	src        migrations.Source
+	applied    int
+	pending    int
+	dbVersion  int  // raw DB version for desync reporting
+	downgraded bool // dbVersion > highest shipped version
+	dirty      bool // mid-failed-migration; surfaced as a (dirty) annotation
+}
+
+func newUpCmd(state *commandState) *cobra.Command {
+	return &cobra.Command{
+		Use:   "up",
+		Short: "Apply all pending migrations across every registered source",
+		Long: `Applies pending migrations for every registered source in
+registration order, through the same orchestrator the server runs at
+startup: per-source advisory locking, pre-run version snapshots, and
+compensating rollback of earlier sources when a later one fails.
+
+Refuses to run while any source's tracking table is dirty; clear it
+with 'force' first.
+
+The --source flag is intentionally not applicable to up; pass it only
+on the per-source subcommands (down/goto/force).`,
+		Args: cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			if state.source != "" {
+				return errors.New("up aggregates across all registered sources; --source is not applicable")
+			}
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			ctx := cmd.Context()
+			srcs, err := state.getSources(ctx)
+			if err != nil {
+				return err
+			}
+			if len(srcs) == 0 {
+				return errors.New("no migration sources registered")
+			}
+
+			// One pre-pass over the sources: refuse dirty state, and snapshot
+			// pending counts so we can report "applied N migration(s)" after
+			// the orchestrator succeeds.
+			prePending := 0
+			for _, src := range srcs {
+				p, err := pendingCount(ctx, src, dsn)
+				if err != nil {
+					return err
+				}
+				prePending += p
+			}
+
+			if err := migrations.RunUp(ctx, dsn, srcs); err != nil {
+				return err
+			}
+
+			if prePending == 0 {
+				fmt.Fprintln(cmd.OutOrStdout(), "no pending migrations; schema is up to date")
+				return nil
+			}
+			fmt.Fprintf(cmd.OutOrStdout(), "applied %d migration(s); schema is up to date\n", prePending)
+			return nil
+		},
+	}
+}
+
+// pendingCount counts NNN_*.up.sql files whose version is greater than the
+// source's current applied version, refusing a dirty source. Uses the same
+// `sourceFileVersions` primitive as `status` so the two paths can't drift.
+func pendingCount(ctx context.Context, src migrations.Source, dsn string) (int, error) {
+	versions, err := sourceFileVersions(src)
+	if err != nil {
+		return 0, err
+	}
+	var pending int
+	err = migrations.WithMigrator(ctx, dsn, src, func(mg *migrate.Migrate) error {
+		if err := ensureClean(src, mg); err != nil {
+			return err
+		}
+		v, _, verr := readVersion(mg)
+		if verr != nil {
+			return verr
+		}
+		for _, fv := range versions {
+			if uint(fv) > v {
+				pending++
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return 0, err
+	}
+	return pending, nil
+}
+
+func newDownCmd(state *commandState) *cobra.Command {
+	return &cobra.Command{
+		Use:   "down N",
+		Short: "Roll back the N most-recent applied migrations for the selected source",
+		Long: `Roll back the N most-recent applied migrations for the selected source.
+
+Down migrations can lose data by design — a rolled-back column loses
+its contents. Refuses to run while the source's tracking table is
+dirty; clear it with 'force' first.`,
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			n, err := strconv.Atoi(args[0])
+			if err != nil || n < 1 {
+				return fmt.Errorf("expected a positive integer for N, got %q", args[0])
+			}
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			src, err := state.resolveSource(cmd.Context())
+			if err != nil {
+				return err
+			}
+			return migrations.WithMigrator(cmd.Context(), dsn, src, func(mg *migrate.Migrate) error {
+				if err := ensureClean(src, mg); err != nil {
+					return err
+				}
+				preV, _, err := readVersion(mg)
+				if err != nil {
+					return err
+				}
+				// Guarded here rather than relying on golang-migrate: Steps
+				// on an empty track reports a confusing "file does not
+				// exist" instead of ErrNoChange.
+				if preV == 0 {
+					fmt.Fprintln(cmd.OutOrStdout(), "no migrations to roll back")
+					return nil
+				}
+				if err := mg.Steps(-n); err != nil {
+					if errors.Is(err, migrate.ErrNoChange) {
+						fmt.Fprintln(cmd.OutOrStdout(), "no migrations to roll back")
+						return nil
+					}
+					return err
+				}
+				postV, _, verr := readVersion(mg)
+				if verr != nil {
+					return fmt.Errorf("read version after rollback: %w", verr)
+				}
+				rolled := countVersionsBetween(src, postV, preV)
+				fmt.Fprintf(cmd.OutOrStdout(), "rolled back %d migration(s)\n", rolled)
+				return nil
+			})
+		},
+	}
+}
+
+func newStatusCmd(state *commandState) *cobra.Command {
+	var output string
+	cmd := &cobra.Command{
+		Use:   "status",
+		Short: "Show how many migrations are applied vs pending across all sources",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			if state.source != "" {
+				return errors.New("status aggregates across all registered sources; --source is not applicable")
+			}
+			if output != "text" && output != "json" {
+				return fmt.Errorf("invalid --output %q; supported: text, json", output)
+			}
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			srcs, err := state.getSources(cmd.Context())
+			if err != nil {
+				return err
+			}
+			if len(srcs) == 0 {
+				return errors.New("no migration sources registered")
+			}
+
+			lines := make([]lineRow, 0, len(srcs))
+			appliedTotal, pendingTotal := 0, 0
+			// Single-source builds print no per-source breakdown, so the
+			// stderr desync warning below is gated to them as their only
+			// signal; multi-source builds carry it in the stdout breakdown.
+			multiSource := len(srcs) > 1
+			for _, src := range srcs {
+				versions, err := sourceFileVersions(src)
+				if err != nil {
+					return err
+				}
+				maxFileVersion := 0
+				if len(versions) > 0 {
+					maxFileVersion = versions[len(versions)-1]
+				}
+				var applied, dbVersion int
+				var downgraded, dirty bool
+				if rerr := migrations.WithMigrator(cmd.Context(), dsn, src, func(mg *migrate.Migrate) error {
+					v, d, err := readVersion(mg)
+					if err != nil {
+						return err
+					}
+					dbVersion = int(v)
+					dirty = d
+					// Count files at/below the DB version as applied
+					// (counting by version, not file count, survives gaps
+					// like a deleted v5).
+					for _, fv := range versions {
+						if fv <= dbVersion {
+							applied++
+						}
+					}
+					if dbVersion > maxFileVersion {
+						// Older binary against a DB migrated by a newer
+						// build. Warn, don't fail.
+						downgraded = true
+						if !multiSource {
+							fmt.Fprintf(cmd.ErrOrStderr(),
+								"warning: %s reports version %d but this binary's highest shipped migration is v%d (older binary against newer DB?)\n",
+								src.Name, v, maxFileVersion)
+						}
+					}
+					return nil
+				}); rerr != nil {
+					return rerr
+				}
+				pending := len(versions) - applied
+				lines = append(lines, lineRow{src: src, applied: applied, pending: pending, dbVersion: dbVersion, downgraded: downgraded, dirty: dirty})
+				appliedTotal += applied
+				pendingTotal += pending
+			}
+
+			out := cmd.OutOrStdout()
+			if output == "json" {
+				return writeStatusJSON(out, lines, appliedTotal, pendingTotal)
+			}
+			if multiSource {
+				fmt.Fprintf(out, "%d migration(s) applied, %d pending\n", appliedTotal, pendingTotal)
+				// Reuses the same `multiSource` gate as the stderr desync
+				// warning above on purpose: a per-source skip branch must
+				// not let the two diverge (a desync gets a source warned
+				// twice or not at all).
+				for _, l := range lines {
+					if l.downgraded {
+						fmt.Fprintf(out, "  %s: %d applied, %d pending (db reports v%d%s — binary out of date)\n",
+							l.src.Name, l.applied, l.pending, l.dbVersion, dirtyTag(l.dirty))
+					} else {
+						fmt.Fprintf(out, "  %s: %d applied (at v%d%s), %d pending\n",
+							l.src.Name, l.applied, l.dbVersion, dirtyTag(l.dirty), l.pending)
+					}
+				}
+			} else {
+				// Single-source: fold the version into the headline so
+				// operators needn't run `version` separately. dbVersion is
+				// the raw tracking-table value (matches `force V`).
+				l := lines[0]
+				fmt.Fprintf(out, "%d migration(s) applied (at v%d%s), %d pending\n",
+					l.applied, l.dbVersion, dirtyTag(l.dirty), l.pending)
+			}
+			return nil
+		},
+	}
+	// No -o shorthand: the kagent root command already owns -o
+	// (--output-format) as a persistent flag, and cobra panics on a
+	// shorthand redefinition.
+	cmd.Flags().StringVar(&output, "output", "text",
+		`Output format: "text" (default) or "json"`)
+	return cmd
+}
+
+// statusJSON is the wire format for `kagent db migrate status --output json`.
+// Operators consume it via `jq`, so the field names and types are a frozen
+// contract; TestStatusJSONShape locks them and fails CI on a rename or
+// retype.
+type statusJSON struct {
+	Applied int                `json:"applied"`
+	Pending int                `json:"pending"`
+	Sources []statusSourceJSON `json:"sources"`
+}
+
+// statusSourceJSON is the per-source object inside statusJSON; same
+// frozen-shape contract, also locked by TestStatusJSONShape.
+type statusSourceJSON struct {
+	Name       string `json:"name"`
+	Applied    int    `json:"applied"`
+	Pending    int    `json:"pending"`
+	Version    int    `json:"version"`
+	Downgraded bool   `json:"downgraded"`
+	Dirty      bool   `json:"dirty"`
+}
+
+func writeStatusJSON(out io.Writer, lines []lineRow, appliedTotal, pendingTotal int) error {
+	payload := statusJSON{
+		Applied: appliedTotal,
+		Pending: pendingTotal,
+		Sources: make([]statusSourceJSON, 0, len(lines)),
+	}
+	for _, l := range lines {
+		payload.Sources = append(payload.Sources, statusSourceJSON{
+			Name:       l.src.Name,
+			Applied:    l.applied,
+			Pending:    l.pending,
+			Version:    l.dbVersion,
+			Downgraded: l.downgraded,
+			Dirty:      l.dirty,
+		})
+	}
+	enc := json.NewEncoder(out)
+	enc.SetIndent("", "  ")
+	return enc.Encode(payload)
+}
+
+// dirtyTag returns " (dirty)" when the source is mid-failed-migration, ""
+// otherwise. Used to annotate the version in status/version text output
+// without adding a separate line.
+func dirtyTag(dirty bool) string {
+	if dirty {
+		return " (dirty)"
+	}
+	return ""
+}
+
+// versionAnnotation renders the trailing annotation for `version` output.
+// Disambiguates an unapplied-migrations state (v=0, !dirty) from a versioned
+// state by tagging the former; dirty wins over the "no migrations applied"
+// tag because it's the more actionable signal.
+func versionAnnotation(v uint, dirty bool) string {
+	if dirty {
+		return " (dirty)"
+	}
+	if v == 0 {
+		return " (no migrations applied)"
+	}
+	return ""
+}
+
+func newVersionCmd(state *commandState) *cobra.Command {
+	return &cobra.Command{
+		Use:   "version",
+		Short: "Print the highest applied migration version",
+		Long: `Print the highest applied migration version.
+For a single registered source the value is on one line; multi-source
+binaries print one line per source. When multiple sources are
+registered, --source filters to a single track.`,
+		Args: cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			srcs, err := state.getSources(cmd.Context())
+			if err != nil {
+				return err
+			}
+			if len(srcs) == 0 {
+				return errors.New("no migration sources registered")
+			}
+			// --source filters the output even though version is otherwise
+			// an aggregate op. Empty flag = print all.
+			if state.source != "" {
+				picked := -1
+				for i, s := range srcs {
+					if s.Name == state.source {
+						picked = i
+						break
+					}
+				}
+				if picked < 0 {
+					return fmt.Errorf("--source %q not registered; registered sources: %s", state.source, sourceNames(srcs))
+				}
+				srcs = []migrations.Source{srcs[picked]}
+			}
+			out := cmd.OutOrStdout()
+			if len(srcs) == 1 {
+				return migrations.WithMigrator(cmd.Context(), dsn, srcs[0], func(mg *migrate.Migrate) error {
+					v, dirty, err := readVersion(mg)
+					if err != nil {
+						return err
+					}
+					fmt.Fprintf(out, "%d%s\n", v, versionAnnotation(v, dirty))
+					return nil
+				})
+			}
+			for _, src := range srcs {
+				if err := migrations.WithMigrator(cmd.Context(), dsn, src, func(mg *migrate.Migrate) error {
+					v, dirty, err := readVersion(mg)
+					if err != nil {
+						return err
+					}
+					fmt.Fprintf(out, "%s: %d%s\n", src.Name, v, versionAnnotation(v, dirty))
+					return nil
+				}); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	}
+}
+
+func newGotoCmd(state *commandState) *cobra.Command {
+	return &cobra.Command{
+		Use:   "goto V",
+		Short: "Move the selected source's schema to version V",
+		Long: `Move the selected source's schema to version V (forward or backward).
+V=0 is the special "empty schema" target: every applied migration in
+the source is rolled back.
+
+Refuses to run while the source's tracking table is dirty; clear it
+with 'force' first.`,
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			v, err := strconv.Atoi(args[0])
+			if err != nil || v < 0 {
+				return fmt.Errorf("expected a non-negative integer for V, got %q", args[0])
+			}
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			src, err := state.resolveSource(cmd.Context())
+			if err != nil {
+				return err
+			}
+			return migrations.WithMigrator(cmd.Context(), dsn, src, func(mg *migrate.Migrate) error {
+				if err := ensureClean(src, mg); err != nil {
+					return err
+				}
+				if v == 0 {
+					if err := mg.Down(); err != nil && !errors.Is(err, migrate.ErrNoChange) {
+						return err
+					}
+					fmt.Fprintln(cmd.OutOrStdout(), "schema is at version 0 (empty)")
+					return nil
+				}
+				if err := mg.Migrate(uint(v)); err != nil && !errors.Is(err, migrate.ErrNoChange) {
+					return err
+				}
+				actual, dirty, aerr := readVersion(mg)
+				if aerr != nil {
+					return aerr
+				}
+				fmt.Fprintf(cmd.OutOrStdout(), "schema is at version %d%s\n", actual, versionAnnotation(actual, dirty))
+				return nil
+			})
+		},
+	}
+}
+
+func newForceCmd(state *commandState) *cobra.Command {
+	return &cobra.Command{
+		Use:   "force V",
+		Short: "Mark version V as applied without running its SQL",
+		Long: `Used to reconcile the selected source's tracking table after manual
+remediation, e.g. to clear a dirty flag left by a failed migration.
+V=0 clears the version record entirely (the "no migrations applied"
+state). Any other V must correspond to a shipped migration file in
+the selected source — otherwise the tracking row would point at a
+version the binary cannot apply or roll back to, wedging the DB.`,
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			v, err := strconv.Atoi(args[0])
+			if err != nil || v < 0 {
+				return fmt.Errorf("expected a non-negative integer for V, got %q", args[0])
+			}
+			dsn, err := state.resolveDSN()
+			if err != nil {
+				return err
+			}
+			src, err := state.resolveSource(cmd.Context())
+			if err != nil {
+				return err
+			}
+			if v > 0 {
+				versions, err := sourceFileVersions(src)
+				if err != nil {
+					return err
+				}
+				if !slices.Contains(versions, v) {
+					return fmt.Errorf(
+						"version %d is not a shipped migration for source %q; valid versions are %s",
+						v, src.Name, formatVersionList(versions))
+				}
+			}
+			return migrations.WithMigrator(cmd.Context(), dsn, src, func(mg *migrate.Migrate) error {
+				if v == 0 {
+					// golang-migrate's Force takes -1 to delete the version
+					// record; 0 is not a valid stored version.
+					if err := mg.Force(-1); err != nil {
+						return err
+					}
+					fmt.Fprintln(cmd.OutOrStdout(), "version record cleared (no migrations applied)")
+					return nil
+				}
+				if err := mg.Force(v); err != nil {
+					return err
+				}
+				fmt.Fprintf(cmd.OutOrStdout(), "version %d marked as applied\n", v)
+				return nil
+			})
+		},
+	}
+}
+
+// countVersionsBetween returns the count of shipped source migrations in the
+// half-open interval (low, high]. Used by `down N` to report the actual
+// number of migrations rolled back regardless of whether the user-supplied N
+// exceeded the applied count. Returns 0 on a source-enumeration error; the
+// call site already surfaces success/failure via Steps(-N) and the count is
+// operator-facing display only.
+func countVersionsBetween(src migrations.Source, low, high uint) int {
+	versions, err := sourceFileVersions(src)
+	if err != nil {
+		return 0
+	}
+	count := 0
+	for _, v := range versions {
+		uv := uint(v)
+		if uv > low && uv <= high {
+			count++
+		}
+	}
+	return count
+}
+
+// formatVersionList renders a small []int as a human-readable list for error
+// messages: "1, 2, 5" or "(none)" if empty.
+func formatVersionList(versions []int) string {
+	if len(versions) == 0 {
+		return "(none)"
+	}
+	parts := make([]string, len(versions))
+	for i, v := range versions {
+		parts[i] = strconv.Itoa(v)
+	}
+	return strings.Join(parts, ", ")
+}
diff --git a/go/core/pkg/cli/db/migrate/migrate_test.go b/go/core/pkg/cli/db/migrate/migrate_test.go
new file mode 100644
index 000000000..96f802296
--- /dev/null
+++ b/go/core/pkg/cli/db/migrate/migrate_test.go
@@ -0,0 +1,364 @@
+package migrate
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+	"testing/fstest"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+
+	"github.com/kagent-dev/kagent/go/core/internal/dbtest"
+	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
+)
+
+// --- fixtures ---
+
+var alphaFS = fstest.MapFS{
+	"alpha/000001_create.up.sql":   {Data: []byte(`CREATE TABLE IF NOT EXISTS cli_alpha (id SERIAL PRIMARY KEY);`)},
+	"alpha/000001_create.down.sql": {Data: []byte(`DROP TABLE IF EXISTS cli_alpha;`)},
+	"alpha/000002_alter.up.sql":    {Data: []byte(`ALTER TABLE cli_alpha ADD COLUMN IF NOT EXISTS name TEXT;`)},
+	"alpha/000002_alter.down.sql":  {Data: []byte(`ALTER TABLE cli_alpha DROP COLUMN IF EXISTS name;`)},
+}
+
+var betaFS = fstest.MapFS{
+	"beta/000001_create.up.sql":   {Data: []byte(`CREATE TABLE IF NOT EXISTS cli_beta (id SERIAL PRIMARY KEY);`)},
+	"beta/000001_create.down.sql": {Data: []byte(`DROP TABLE IF EXISTS cli_beta;`)},
+}
+
+func testSources() []migrations.Source {
+	return []migrations.Source{
+		{Name: "alpha", TrackingTable: "alpha_schema_migrations", FS: alphaFS, Dir: "alpha"},
+		{Name: "beta", TrackingTable: "beta_schema_migrations", FS: betaFS, Dir: "beta"},
+	}
+}
+
+// runCLI executes `db migrate <args>` against a fresh command tree and
+// returns combined stdout and the error, plus stderr separately.
+func runCLI(t *testing.T, sources []migrations.Source, args ...string) (string, string, error) {
+	t.Helper()
+	cmd := NewCommand(sources...)
+	var out, errOut bytes.Buffer
+	cmd.SetOut(&out)
+	cmd.SetErr(&errOut)
+	cmd.SetArgs(args)
+	err := cmd.ExecuteContext(context.Background())
+	return out.String(), errOut.String(), err
+}
+
+// --- unit tests (no database) ---
+
+func TestNewCommandRejectsBadSources(t *testing.T) {
+	tests := []struct {
+		name    string
+		sources []migrations.Source
+	}{
+		{name: "invalid name", sources: []migrations.Source{{Name: "Bad Name", TrackingTable: "t", FS: alphaFS, Dir: "alpha"}}},
+		{name: "duplicate name", sources: []migrations.Source{
+			{Name: "core", TrackingTable: "a", FS: alphaFS, Dir: "alpha"},
+			{Name: "core", TrackingTable: "b", FS: betaFS, Dir: "beta"},
+		}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			defer func() {
+				if recover() == nil {
+					t.Fatal("expected NewCommand to panic")
+				}
+			}()
+			NewCommand(tt.sources...)
+		})
+	}
+}
+
+func TestNewCommandAcceptsHyphenatedNames(t *testing.T) {
+	// Downstream-registered sources may use hyphenated names.
+	NewCommand(migrations.Source{Name: "extra-track", TrackingTable: "t", FS: alphaFS, Dir: "alpha"})
+}
+
+func TestResolveDSN(t *testing.T) {
+	tests := []struct {
+		name    string
+		flag    string
+		env     string
+		want    string
+		wantErr bool
+	}{
+		{name: "flag wins over env", flag: "postgres://flag", env: "postgres://env", want: "postgres://flag"},
+		{name: "env fallback", env: "postgres://env", want: "postgres://env"},
+		{name: "neither set", wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv(dbURLEnv, tt.env)
+			s := &commandState{dbURL: tt.flag}
+			got, err := s.resolveDSN()
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("resolveDSN() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if got != tt.want {
+				t.Errorf("resolveDSN() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestResolveSource(t *testing.T) {
+	multi := testSources()
+	single := multi[:1]
+	tests := []struct {
+		name    string
+		sources []migrations.Source
+		flag    string
+		want    string
+		wantErr string
+	}{
+		{name: "single source inferred", sources: single, want: "alpha"},
+		{name: "single source explicit match", sources: single, flag: "alpha", want: "alpha"},
+		{name: "single source mismatch", sources: single, flag: "beta", wantErr: "not registered"},
+		{name: "multi requires flag", sources: multi, wantErr: "pass --source"},
+		{name: "multi explicit", sources: multi, flag: "beta", want: "beta"},
+		{name: "multi unknown", sources: multi, flag: "nope", wantErr: "not registered"},
+		{name: "none registered", sources: nil, wantErr: "no migration sources"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srcs := tt.sources
+			s := &commandState{
+				source:    tt.flag,
+				resolveFn: func(context.Context) ([]migrations.Source, error) { return srcs, nil },
+			}
+			got, err := s.resolveSource(context.Background())
+			if tt.wantErr != "" {
+				if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+					t.Fatalf("resolveSource() error = %v, want containing %q", err, tt.wantErr)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("resolveSource() unexpected error: %v", err)
+			}
+			if got.Name != tt.want {
+				t.Errorf("resolveSource() = %q, want %q", got.Name, tt.want)
+			}
+		})
+	}
+}
+
+// TestArgValidation covers rejections that fire before any database access.
+func TestArgValidation(t *testing.T) {
+	tests := []struct {
+		name    string
+		args    []string
+		wantErr string
+	}{
+		{name: "down non-integer", args: []string{"down", "abc"}, wantErr: "positive integer"},
+		{name: "down zero", args: []string{"down", "0"}, wantErr: "positive integer"},
+		{name: "goto non-integer", args: []string{"goto", "abc"}, wantErr: "non-negative integer"},
+		{name: "goto negative", args: []string{"goto", "--", "-1"}, wantErr: "non-negative integer"},
+		{name: "force non-integer", args: []string{"force", "abc"}, wantErr: "non-negative integer"},
+		{name: "up rejects source flag", args: []string{"up", "--source", "alpha"}, wantErr: "--source is not applicable"},
+		{name: "status rejects source flag", args: []string{"status", "--source", "alpha"}, wantErr: "--source is not applicable"},
+		{name: "status invalid output", args: []string{"status", "--output", "yaml"}, wantErr: "invalid --output"},
+		{name: "no dsn", args: []string{"version"}, wantErr: "database URL not set"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv(dbURLEnv, "")
+			_, _, err := runCLI(t, testSources(), tt.args...)
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("error = %v, want containing %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+// TestNewCommandFromFunc covers deferred source resolution: the resolver must
+// not run at construction, and its failures surface as command errors rather
+// than wiring-time panics.
+func TestNewCommandFromFunc(t *testing.T) {
+	t.Run("resolver is not called at construction", func(t *testing.T) {
+		called := false
+		NewCommandFromFunc(func(context.Context) ([]migrations.Source, error) {
+			called = true
+			return testSources(), nil
+		})
+		if called {
+			t.Fatal("resolver ran during command construction")
+		}
+	})
+
+	t.Run("resolver error surfaces as command error", func(t *testing.T) {
+		cmd := NewCommandFromFunc(func(context.Context) ([]migrations.Source, error) {
+			return nil, errors.New("cluster unreachable")
+		})
+		cmd.SetOut(io.Discard)
+		cmd.SetErr(io.Discard)
+		cmd.SetArgs([]string{"version", "--db-url", "postgres://unused"})
+		err := cmd.ExecuteContext(context.Background())
+		if err == nil || !strings.Contains(err.Error(), "cluster unreachable") {
+			t.Fatalf("error = %v, want resolver error", err)
+		}
+	})
+
+	t.Run("duplicate names from resolver error instead of panicking", func(t *testing.T) {
+		dup := []migrations.Source{
+			{Name: "core", TrackingTable: "a", FS: alphaFS, Dir: "alpha"},
+			{Name: "core", TrackingTable: "b", FS: betaFS, Dir: "beta"},
+		}
+		cmd := NewCommandFromFunc(func(context.Context) ([]migrations.Source, error) { return dup, nil })
+		cmd.SetOut(io.Discard)
+		cmd.SetErr(io.Discard)
+		cmd.SetArgs([]string{"version", "--db-url", "postgres://unused"})
+		err := cmd.ExecuteContext(context.Background())
+		if err == nil || !strings.Contains(err.Error(), "configured twice") {
+			t.Fatalf("error = %v, want duplicate-name error", err)
+		}
+	})
+}
+
+// TestStatusJSONShape freezes the `status -o json` wire format. Operators
+// consume it via jq, so a rename or retype is a breaking change; update this
+// test only with a deliberate contract change.
+func TestStatusJSONShape(t *testing.T) {
+	payload := statusJSON{
+		Applied: 3,
+		Pending: 1,
+		Sources: []statusSourceJSON{{Name: "alpha", Applied: 2, Pending: 1, Version: 2, Downgraded: false, Dirty: true}},
+	}
+	got, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := `{"applied":3,"pending":1,"sources":[{"name":"alpha","applied":2,"pending":1,"version":2,"downgraded":false,"dirty":true}]}`
+	if string(got) != want {
+		t.Errorf("status JSON shape changed:\n got: %s\nwant: %s", got, want)
+	}
+}
+
+// --- database-backed tests ---
+
+// TestCLIAgainstPostgres walks the operator surface end to end against a real
+// Postgres: up, status, version, down, goto, dirty refusal, and force. The
+// subtests share one container and run in order — each builds on the schema
+// state the previous one left.
+func TestCLIAgainstPostgres(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping container-backed test in -short mode")
+	}
+	ctx := context.Background()
+	dsn := dbtest.StartT(ctx, t)
+	sources := testSources()
+
+	mustContain := func(t *testing.T, out, want string) {
+		t.Helper()
+		if !strings.Contains(out, want) {
+			t.Fatalf("output %q does not contain %q", out, want)
+		}
+	}
+	mustRun := func(t *testing.T, args ...string) string {
+		t.Helper()
+		out, _, err := runCLI(t, sources, append(args, "--db-url", dsn)...)
+		if err != nil {
+			t.Fatalf("%v: %v", args, err)
+		}
+		return out
+	}
+
+	t.Run("up applies all sources", func(t *testing.T) {
+		mustContain(t, mustRun(t, "up"), "applied 3 migration(s)")
+	})
+
+	t.Run("up is idempotent", func(t *testing.T) {
+		mustContain(t, mustRun(t, "up"), "no pending migrations")
+	})
+
+	t.Run("status text", func(t *testing.T) {
+		out := mustRun(t, "status")
+		mustContain(t, out, "3 migration(s) applied, 0 pending")
+		mustContain(t, out, "alpha: 2 applied (at v2), 0 pending")
+		mustContain(t, out, "beta: 1 applied (at v1), 0 pending")
+	})
+
+	t.Run("status json", func(t *testing.T) {
+		var got statusJSON
+		if err := json.Unmarshal([]byte(mustRun(t, "status", "--output", "json")), &got); err != nil {
+			t.Fatal(err)
+		}
+		if got.Applied != 3 || got.Pending != 0 || len(got.Sources) != 2 {
+			t.Fatalf("unexpected status: %+v", got)
+		}
+	})
+
+	t.Run("version per source and filtered", func(t *testing.T) {
+		out := mustRun(t, "version")
+		mustContain(t, out, "alpha: 2")
+		mustContain(t, out, "beta: 1")
+		mustContain(t, mustRun(t, "version", "--source", "alpha"), "2")
+	})
+
+	t.Run("down rolls back one step", func(t *testing.T) {
+		mustContain(t, mustRun(t, "down", "1", "--source", "alpha"), "rolled back 1 migration(s)")
+		mustContain(t, mustRun(t, "version", "--source", "alpha"), "1")
+	})
+
+	t.Run("goto moves forward", func(t *testing.T) {
+		mustContain(t, mustRun(t, "goto", "2", "--source", "alpha"), "schema is at version 2")
+	})
+
+	t.Run("goto zero empties the source", func(t *testing.T) {
+		mustContain(t, mustRun(t, "goto", "0", "--source", "beta"), "version 0 (empty)")
+		mustContain(t, mustRun(t, "version", "--source", "beta"), "no migrations applied")
+	})
+
+	t.Run("down with nothing to roll back", func(t *testing.T) {
+		mustContain(t, mustRun(t, "down", "1", "--source", "beta"), "no migrations to roll back")
+	})
+
+	t.Run("dirty source is refused and force recovers", func(t *testing.T) {
+		markDirty(t, dsn, "alpha_schema_migrations")
+		for _, args := range [][]string{{"up"}, {"down", "1", "--source", "alpha"}, {"goto", "1", "--source", "alpha"}} {
+			_, _, err := runCLI(t, sources, append(args, "--db-url", dsn)...)
+			if err == nil || !strings.Contains(err.Error(), "dirty") {
+				t.Fatalf("%v: error = %v, want dirty refusal", args, err)
+			}
+		}
+		// status still reports rather than refusing, and annotates the row.
+		mustContain(t, mustRun(t, "status"), "(dirty)")
+
+		mustContain(t, mustRun(t, "force", "2", "--source", "alpha"), "version 2 marked as applied")
+		mustContain(t, mustRun(t, "up"), "applied 1 migration(s)") // beta was left at 0 by the goto above
+	})
+
+	t.Run("force rejects unshipped version", func(t *testing.T) {
+		_, _, err := runCLI(t, sources, "force", "99", "--source", "alpha", "--db-url", dsn)
+		if err == nil || !strings.Contains(err.Error(), "not a shipped migration") {
+			t.Fatalf("error = %v, want unshipped-version rejection", err)
+		}
+	})
+
+	t.Run("force zero clears the version record", func(t *testing.T) {
+		mustContain(t, mustRun(t, "force", "0", "--source", "beta"), "version record cleared")
+		mustContain(t, mustRun(t, "version", "--source", "beta"), "no migrations applied")
+		mustContain(t, mustRun(t, "up"), "applied 1 migration(s)")
+	})
+}
+
+// markDirty flips the dirty flag on a tracking table, simulating a process
+// that died mid-migration.
+func markDirty(t *testing.T, dsn, table string) {
+	t.Helper()
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec("UPDATE " + table + " SET dirty = true"); err != nil {
+		t.Fatalf("mark %s dirty: %v", table, err)
+	}
+}
diff --git a/go/core/pkg/migrations/runner.go b/go/core/pkg/migrations/runner.go
index 2b697430b..a90dd946a 100644
--- a/go/core/pkg/migrations/runner.go
+++ b/go/core/pkg/migrations/runner.go
@@ -350,6 +350,23 @@ func applySource(ctx context.Context, url string, src Source) (prevVersion uint,
 	return prevVersion, nil
 }
 
+// WithMigrator opens a migrator for src against url, runs fn against it, and
+// closes it. The migrator carries the same schema handling, tracking-table
+// configuration, and advisory-lock identity as the orchestrator's own runs, so
+// out-of-band tooling (the `kagent db migrate` CLI) built on this serializes
+// correctly against a concurrently booting server and cannot drift from the
+// startup path. fn's migration operations (Up/Down/Steps/Migrate/Force) each
+// take golang-migrate's per-(database, schema) advisory lock; Version reads do
+// not.
+func WithMigrator(ctx context.Context, url string, src Source, fn func(*migrate.Migrate) error) error {
+	mg, err := newMigrate(ctx, url, src)
+	if err != nil {
+		return err
+	}
+	defer closeMigrate(src.Name, mg)
+	return fn(mg)
+}
+
 // rollbackSource opens a fresh migrate instance and rolls a source back to
 // targetVersion. Used to compensate a previously-succeeded source when a later
 // source fails. It returns an error (also logged) when the rollback fails, so