Add SKIP_MIGRATIONS to run migrations out-of-band

go/core/pkg/app/app.go

@@ -137,9 +137,10 @@ type Config struct {
 	// that originates TLS upstream. Off by default;
 	MCPEgressPlaintext bool
 	Database           struct {
-		Url           string
-		UrlFile       string
-		VectorEnabled bool
+		Url            string
+		UrlFile        string
+		VectorEnabled  bool
+		SkipMigrations bool
 	}
 	Substrate struct {
 		AteAPIEndpoint             string
@@ -181,6 +182,7 @@ func (cfg *Config) SetFlags(commandLine *flag.FlagSet) {
 	commandLine.StringVar(&cfg.Database.Url, "postgres-database-url", "postgres://postgres:kagent@kagent-postgresql.kagent.svc.cluster.local:5432/postgres", "The URL of the PostgreSQL database.")
 	commandLine.StringVar(&cfg.Database.UrlFile, "postgres-database-url-file", "", "Path to a file containing the PostgreSQL database URL. Takes precedence over --postgres-database-url.")
 	commandLine.BoolVar(&cfg.Database.VectorEnabled, "database-vector-enabled", true, "Enable pgvector extension and memory table. Requires pgvector to be installed on the PostgreSQL server.")
+	commandLine.BoolVar(&cfg.Database.SkipMigrations, "skip-migrations", false, "Do not run database migrations at startup; instead verify the database is already migrated and fail if it is not. Migrations must be applied out-of-band (e.g. from a pipeline or pre-upgrade hook). Settable via the SKIP_MIGRATIONS env var.")
 
 	commandLine.StringVar(&cfg.WatchNamespaces, "watch-namespaces", "", "The namespaces to watch for .")
 
@@ -470,13 +472,25 @@ func Start(getExtensionConfig GetExtensionConfig, extraSources []migrations.Sour
 
 	// Run migrations before connecting; schema must exist before queries.
 	// Built-in sources run first, then any downstream-registered extras.
-	setupLog.Info("running database migrations")
+	// With --skip-migrations (SKIP_MIGRATIONS) the server applies nothing and
+	// instead verifies the database is already migrated, so migrations can run
+	// out-of-band and this connection needs no DDL privileges.
 	sources := append(migrations.BuiltinSources(cfg.Database.VectorEnabled), extraSources...)
-	if err := migrations.RunUp(ctx, dbURL, sources); err != nil {
-		setupLog.Error(err, "database migration failed")
-		os.Exit(1)
+	if cfg.Database.SkipMigrations {
+		setupLog.Info("skipping database migrations; verifying schema is migrated")
+		if err := migrations.VerifyMigrated(ctx, dbURL, sources); err != nil {
+			setupLog.Error(err, "database migration verification failed")
+			os.Exit(1)
+		}
+		setupLog.Info("database schema verified")
+	} else {
+		setupLog.Info("running database migrations")
+		if err := migrations.RunUp(ctx, dbURL, sources); err != nil {
+			setupLog.Error(err, "database migration failed")
+			os.Exit(1)
+		}
+		setupLog.Info("database migrations complete")
 	}
-	setupLog.Info("database migrations complete")
 
 	// Connect to database
 	db, err := database.Connect(ctx, &database.PostgresConfig{

go/core/pkg/migrations/runner.go

@@ -179,6 +179,83 @@ func RunUp(ctx context.Context, url string, sources []Source) error {
 	return nil
 }
 
+// VerifyMigrated checks, without applying or reverting anything, that every
+// source's migrations have been applied to the database. It is the boot-time
+// guard for the SKIP_MIGRATIONS deployment mode, where migrations run
+// out-of-band (a pipeline or pre-upgrade hook) and the server must refuse to
+// serve a wrong-shaped schema. It issues only SELECTs — never golang-migrate,
+// which creates the tracking table on open — so it is safe on a connection
+// whose role has no DDL privileges.
+//
+// Per source: a missing tracking table or a version behind this binary's
+// embedded max is an error; a dirty tracking table is an error; a database
+// ahead of the binary is tolerated (compatibility mode), matching RunUp.
+func VerifyMigrated(ctx context.Context, url string, sources []Source) error {
+	if len(sources) == 0 {
+		return nil
+	}
+	if err := validateSources(sources); err != nil {
+		return err
+	}
+	// Reject the same resolved-schema collisions RunUp rejects: a colliding
+	// source set shares one tracking table, so verification would read the
+	// same row twice and "pass" an unsafe configuration.
+	if err := checkResolvedSchemaCollisions(ctx, url, sources); err != nil {
+		return err
+	}
+
+	db, err := sql.Open("pgx", url)
+	if err != nil {
+		return fmt.Errorf("open database to verify migrations: %w", err)
+	}
+	defer db.Close()
+
+	for _, src := range sources {
+		if err := ctx.Err(); err != nil {
+			return fmt.Errorf("migration verification cancelled before %s: %w", src.Name, err)
+		}
+		maxVer, err := maxEmbeddedVersion(src.FS, src.Dir)
+		if err != nil {
+			return fmt.Errorf("determine max embedded version for %s: %w", src.Name, err)
+		}
+
+		// For Schema == "" the unqualified name resolves via the connection's
+		// search_path — the same place RunUp put the table.
+		table := quoteIdentifier(src.TrackingTable)
+		if src.Schema != "" {
+			table = quoteIdentifier(src.Schema) + "." + table
+		}
+
+		var exists bool
+		if err := db.QueryRowContext(ctx, "SELECT to_regclass($1) IS NOT NULL", table).Scan(&exists); err != nil {
+			return fmt.Errorf("check tracking table for %s: %w", src.Name, err)
+		}
+		if !exists {
+			return fmt.Errorf("source %s: tracking table %s does not exist - the database has not been migrated; apply migrations out-of-band or unset SKIP_MIGRATIONS", src.Name, table)
+		}
+
+		var version int64
+		var dirty bool
+		err = db.QueryRowContext(ctx, "SELECT version, dirty FROM "+table+" LIMIT 1").Scan(&version, &dirty)
+		if errors.Is(err, sql.ErrNoRows) {
+			version, dirty = 0, false // table exists but nothing applied yet
+		} else if err != nil {
+			return fmt.Errorf("read tracking table for %s: %w", src.Name, err)
+		}
+
+		switch {
+		case dirty:
+			return fmt.Errorf("source %s is dirty at version %d: a previous migration attempt failed and must be resolved before starting with SKIP_MIGRATIONS", src.Name, version)
+		case version < int64(maxVer):
+			return fmt.Errorf("source %s is at version %d but this binary requires version %d: apply migrations out-of-band or unset SKIP_MIGRATIONS", src.Name, version, maxVer)
+		case version > int64(maxVer):
+			log.Info("database schema is ahead of this binary; running in compatibility mode",
+				"track", src.Name, "dbVersion", version, "binaryMax", maxVer)
+		}
+	}
+	return nil
+}
+
 // validateSources rejects a source set that cannot be run safely. It checks two
 // things.
 //

go/core/pkg/migrations/runner_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"net/url"
 	"strings"
 	"testing"
 	"testing/fstest"
@@ -972,3 +973,110 @@ func TestApplySource_DirtyStateRecoveryOnRestart(t *testing.T) {
 		t.Errorf("after restart: version = %d, want 1 (dirty cleared, rolled back)", got)
 	}
 }
+
+// TestVerifyMigrated covers the SKIP_MIGRATIONS boot guard: refuse an
+// un-migrated or behind or dirty database, tolerate exact-match and ahead
+// (compatibility mode), and work on a connection whose role has no DDL
+// privileges — the deployment mode the guard exists for.
+func TestVerifyMigrated(t *testing.T) {
+	ctx := context.Background()
+	connStr := startTestDB(t)
+	full := []Source{coreSource(goodCoreFS)} // max embedded version 2
+
+	t.Run("unmigrated database is refused", func(t *testing.T) {
+		err := VerifyMigrated(ctx, connStr, full)
+		if err == nil || !strings.Contains(err.Error(), "has not been migrated") {
+			t.Fatalf("error = %v, want missing-tracking-table refusal", err)
+		}
+	})
+
+	t.Run("pending migrations are refused", func(t *testing.T) {
+		if _, err := applySource(ctx, connStr, coreSource(oneCoreFS)); err != nil {
+			t.Fatalf("apply v1: %v", err)
+		}
+		err := VerifyMigrated(ctx, connStr, full)
+		if err == nil || !strings.Contains(err.Error(), "requires version 2") {
+			t.Fatalf("error = %v, want behind-binary refusal", err)
+		}
+	})
+
+	t.Run("fully migrated database passes", func(t *testing.T) {
+		if _, err := applySource(ctx, connStr, coreSource(goodCoreFS)); err != nil {
+			t.Fatalf("apply v2: %v", err)
+		}
+		if err := VerifyMigrated(ctx, connStr, full); err != nil {
+			t.Fatalf("VerifyMigrated() = %v, want nil", err)
+		}
+	})
+
+	t.Run("works without DDL privileges", func(t *testing.T) {
+		db, err := sql.Open("pgx", connStr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		for _, q := range []string{
+			`CREATE ROLE readonly LOGIN PASSWORD 'ro'`,
+			`GRANT USAGE ON SCHEMA public TO readonly`,
+			`GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly`,
+		} {
+			if _, err := db.ExecContext(ctx, q); err != nil {
+				t.Fatalf("%s: %v", q, err)
+			}
+		}
+		u, err := url.Parse(connStr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		u.User = url.UserPassword("readonly", "ro")
+		if err := VerifyMigrated(ctx, u.String(), full); err != nil {
+			t.Fatalf("VerifyMigrated() as readonly = %v, want nil", err)
+		}
+	})
+
+	t.Run("database ahead of binary passes", func(t *testing.T) {
+		if err := VerifyMigrated(ctx, connStr, []Source{coreSource(oneCoreFS)}); err != nil {
+			t.Fatalf("VerifyMigrated() with older binary = %v, want nil (compatibility mode)", err)
+		}
+	})
+
+	t.Run("dirty tracking table is refused", func(t *testing.T) {
+		db, err := sql.Open("pgx", connStr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+		if _, err := db.ExecContext(ctx, `UPDATE schema_migrations SET dirty = true`); err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if _, err := db.ExecContext(ctx, `UPDATE schema_migrations SET dirty = false`); err != nil {
+				t.Fatal(err)
+			}
+		}()
+		err = VerifyMigrated(ctx, connStr, full)
+		if err == nil || !strings.Contains(err.Error(), "dirty") {
+			t.Fatalf("error = %v, want dirty refusal", err)
+		}
+	})
+
+	t.Run("resolved schema collision is refused", func(t *testing.T) {
+		// Schema "" resolves to public here, colliding with the explicit
+		// "public" source on the same tracking table — the same source set
+		// RunUp rejects.
+		collide := []Source{
+			coreSource(goodCoreFS),
+			{Name: "explicit", Schema: "public", TrackingTable: "schema_migrations", FS: goodCoreFS, Dir: "core"},
+		}
+		err := VerifyMigrated(ctx, connStr, collide)
+		if err == nil || !strings.Contains(err.Error(), "resolve to the same tracking table") {
+			t.Fatalf("error = %v, want resolved-schema collision refusal", err)
+		}
+	})
+
+	t.Run("no sources is a no-op", func(t *testing.T) {
+		if err := VerifyMigrated(ctx, "postgres://unused", nil); err != nil {
+			t.Fatalf("VerifyMigrated() with no sources = %v, want nil", err)
+		}
+	})
+}

helm/kagent/templates/NOTES.txt

@@ -92,6 +92,16 @@ DOCUMENTATION:
     database.postgres.bundled.enabled=false
 {{- end }}
 {{- end }}
+{{- if .Values.database.postgres.skipMigrations }}
+################################################################################
+# NOTE: STARTUP MIGRATIONS ARE DISABLED                                        #
+################################################################################
+  database.postgres.skipMigrations is set: the controller will not run database
+  migrations at startup. It verifies the schema is already migrated and fails
+  to start if it is not.
+
+  Ensure migrations are applied out-of-band before installing or upgrading.
+{{- end }}
 {{- if .Values.substrate.enabled }}
 ################################################################################
 # WARNING: SUBSTRATE IS EXPERIMENTAL, USE AT OWN RISK                          #

helm/kagent/templates/controller-configmap.yaml

@@ -55,6 +55,7 @@ data:
   PROXY_URL: {{ .Values.proxy.url | quote }}
   {{- end }}
   DATABASE_VECTOR_ENABLED: {{ .Values.database.postgres.vectorEnabled | quote }}
+  SKIP_MIGRATIONS: {{ .Values.database.postgres.skipMigrations | default false | quote }}
   WATCH_NAMESPACES: {{ include "kagent.watchNamespaces" . | quote }}
   MCP_EGRESS_PLAINTEXT: {{ .Values.controller.mcpEgressPlaintext | default false | quote }}
   {{- if .Values.controller.a2aClientTimeout }}

helm/kagent/values.yaml

@@ -79,6 +79,10 @@ database:
     # Required to use features that depend on database vector capability. (e.g. long-term memory)
     # Set to true when using an external PostgreSQL that has the pgvector extension installed.
     vectorEnabled: false
+    # -- Skip running database migrations at controller startup.
+    # The controller instead verifies the database is already migrated and fails if it is not.
+    # Migrations must be applied out-of-band (e.g. from a CI/CD pipeline) before install/upgrade.
+    skipMigrations: false
     # -- Bundled PostgreSQL instance — for development and evaluation only.
     # Not suitable for production. Deployed when enabled is true and url/urlFile are not set.
     bundled: