From 38ca9c2a84fcd989134bdb8a5fd7cd621aceb50b Mon Sep 17 00:00:00 2001
From: Jeremy Alvis <jeremy.alvis@solo.io>
Date: Fri, 18 Sep 2026 13:56:53 -0700
Subject: [PATCH] Separate PostgresQL runtime and schema ownership

---
 cmd/ate-setup/differences.md                  |   2 +-
 cmd/ate-setup/internal/config/config.go       |  25 ++--
 cmd/ate-setup/internal/config/config_test.go  |  35 +++++-
 cmd/ate-setup/internal/steps/create.go        |  63 +++++++---
 cmd/ate-setup/internal/steps/create_test.go   |  76 ++++++++++--
 cmd/ate-setup/internal/steps/deploy.go        |   8 ++
 cmd/ate-setup/internal/steps/postgres.go      |  57 +++++++++
 cmd/ate-setup/internal/steps/postgres_test.go |  52 ++++++++
 cmd/ateapi/internal/store/atepg/atepg.go      |  70 ++++++++---
 cmd/ateapi/internal/store/atepg/atepg_test.go | 115 ++++++++++++++++--
 cmd/ateapi/internal/store/atepg/outbox.go     |   6 +-
 .../internal/store/atepg/outbox_test.go       |  21 ++--
 cmd/ateapi/internal/store/atepg/schema.go     |  36 ++++++
 cmd/ateapi/main.go                            |  18 ++-
 docs/dev/postgresql-schema-evolution.md       |   1 +
 hack/install-ate.sh                           |  60 +++++++--
 manifests/ate-install/ate-api-server.yaml     |   1 +
 manifests/ate-install/postgres/postgres.yaml  |  59 ++++++++-
 tools/setup-gcp/cloud-sql.md                  |  17 ++-
 19 files changed, 618 insertions(+), 104 deletions(-)

diff --git a/cmd/ate-setup/differences.md b/cmd/ate-setup/differences.md
index ff7fc16d84..dc80c0509f 100644
--- a/cmd/ate-setup/differences.md
+++ b/cmd/ate-setup/differences.md
@@ -277,5 +277,5 @@ to the in-cluster database, leaving behind an orphaned proxy. Use
 
 The shell installer had no tests. `cmd/ate-setup` has unit tests for template
 rendering, overlay selection, config resolution, the authentication config, the
-apiserver environment ConfigMap, delegated script arguments, manifest deletion,
+apiserver environment ConfigMap and Secret, delegated script arguments, manifest deletion,
 per-demo rendering, and image reference rewriting.
diff --git a/cmd/ate-setup/internal/config/config.go b/cmd/ate-setup/internal/config/config.go
index a7312a90f1..f0761de28c 100644
--- a/cmd/ate-setup/internal/config/config.go
+++ b/cmd/ate-setup/internal/config/config.go
@@ -43,12 +43,6 @@ const (
 // DefaultRolloutTimeout is the default wait timeout for workload rollouts.
 const DefaultRolloutTimeout = 60 * time.Second
 
-// DefaultPostgresConnectionString mirrors default_postgres_connection_string in
-// the shell installer: the apiserver reaches PostgreSQL over mTLS using the
-// podcertificate controller's projected servicedns trust bundle and its own
-// podidentity credential bundle.
-const DefaultPostgresConnectionString = "postgresql://postgres@postgres.ate-system.svc:5432/atepg?sslmode=verify-full&sslrootcert=/run/servicedns.podcert.ate.dev/trust-bundle.pem&sslcert=/run/podidentity.podcert.ate.dev/credential-bundle.pem&sslkey=/run/podidentity.podcert.ate.dev/credential-bundle.pem"
-
 // DefaultPostgresSchema mirrors the shell installer's default for
 // ATE_API_POSTGRES_SCHEMA, the PostgreSQL schema holding the Substrate tables.
 const DefaultPostgresSchema = "public"
@@ -98,8 +92,12 @@ type Config struct {
 	// Router selects the atenet router dataplane.
 	Router string
 	// PostgresConnectionString is the apiserver's store connection string.
-	// Empty means use DefaultPostgresConnectionString.
+	// Empty means use the bundled PostgreSQL runtime role.
 	PostgresConnectionString string
+	// PostgresDDLConnectionString is the optional schema-owner connection
+	// string. Empty means use the runtime string for an external database; a
+	// non-empty value requires PostgresConnectionString.
+	PostgresDDLConnectionString string
 	// PostgresSchema is the PostgreSQL schema for the Substrate tables
 	// (ATE_API_POSTGRES_SCHEMA). Empty means DefaultPostgresSchema.
 	PostgresSchema string
@@ -232,6 +230,7 @@ func Load(opts Options) (*Config, error) {
 		KODefaultPlatforms:                    env["KO_DEFAULTPLATFORMS"],
 		Images:                                loadImageSource(opts, env),
 		PostgresConnectionString:              env["ATE_API_POSTGRES_CONNECTION_STRING"],
+		PostgresDDLConnectionString:           env["ATE_API_POSTGRES_DDL_CONNECTION_STRING"],
 		PostgresSchema:                        env["ATE_API_POSTGRES_SCHEMA"],
 		RolloutTimeout:                        rolloutTimeout,
 		rolloutTimeoutSet:                     timeoutStr != "",
@@ -281,6 +280,9 @@ func validate(cfg *Config) error {
 	if err := cfg.Images.Validate(); err != nil {
 		return err
 	}
+	if cfg.PostgresDDLConnectionString != "" && cfg.PostgresConnectionString == "" {
+		return fmt.Errorf("ATE_API_POSTGRES_DDL_CONNECTION_STRING requires ATE_API_POSTGRES_CONNECTION_STRING")
+	}
 	switch cfg.Router {
 	case RouterEnvoy, RouterAgentgateway:
 	default:
@@ -333,15 +335,6 @@ func validateExtprocService(spec string) error {
 	return nil
 }
 
-// PostgresConnString returns the configured connection string, falling back to
-// the in-cluster default.
-func (c *Config) PostgresConnString() string {
-	if c.PostgresConnectionString != "" {
-		return c.PostgresConnectionString
-	}
-	return DefaultPostgresConnectionString
-}
-
 // PostgresSchemaName returns the configured schema, falling back to the
 // shell installer's default. ate-api-server rejects an empty value.
 func (c *Config) PostgresSchemaName() string {
diff --git a/cmd/ate-setup/internal/config/config_test.go b/cmd/ate-setup/internal/config/config_test.go
index e035295755..4e361f9872 100644
--- a/cmd/ate-setup/internal/config/config_test.go
+++ b/cmd/ate-setup/internal/config/config_test.go
@@ -44,6 +44,7 @@ func loadEnv(t *testing.T) {
 		"ATE_CREDENTIAL_PROVIDER_ADDRESS",
 		"ATE_CREDENTIAL_PROVIDER_NAME",
 		"ATE_API_POSTGRES_CONNECTION_STRING",
+		"ATE_API_POSTGRES_DDL_CONNECTION_STRING",
 		"ATE_API_POSTGRES_SCHEMA",
 		"ATE_ATENET_DATAPLANE",
 		"ATE_EXPERIMENTAL_USE_SDSMINT",
@@ -78,8 +79,8 @@ func TestLoadDefaults(t *testing.T) {
 	if cfg.Router != RouterEnvoy {
 		t.Errorf("Router = %q, want %q", cfg.Router, RouterEnvoy)
 	}
-	if cfg.PostgresConnString() != DefaultPostgresConnectionString {
-		t.Errorf("PostgresConnString() = %q, want %q", cfg.PostgresConnString(), DefaultPostgresConnectionString)
+	if cfg.PostgresConnectionString != "" {
+		t.Errorf("PostgresConnectionString = %q, want bundled PostgreSQL", cfg.PostgresConnectionString)
 	}
 	if cfg.RolloutTimeout != DefaultRolloutTimeout {
 		t.Errorf("RolloutTimeout = %v, want %v", cfg.RolloutTimeout, DefaultRolloutTimeout)
@@ -114,8 +115,34 @@ func TestLoadPostgresConnectionStringOverride(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Load() error = %v", err)
 	}
-	if cfg.PostgresConnString() != dsn {
-		t.Errorf("PostgresConnString() = %q, want %q", cfg.PostgresConnString(), dsn)
+	if cfg.PostgresConnectionString != dsn {
+		t.Errorf("PostgresConnectionString = %q, want %q", cfg.PostgresConnectionString, dsn)
+	}
+}
+
+func TestLoadPostgresDDLConnectionString(t *testing.T) {
+	loadEnv(t)
+	const runtimeDSN = "postgresql://runtime@db.example:5432/atepg?sslmode=disable"
+	const dsn = "postgresql://owner@db.example:5432/atepg?sslmode=disable"
+	t.Setenv("ATE_API_POSTGRES_CONNECTION_STRING", runtimeDSN)
+	t.Setenv("ATE_API_POSTGRES_DDL_CONNECTION_STRING", dsn)
+
+	cfg, err := Load(Options{})
+	if err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+	if cfg.PostgresDDLConnectionString != dsn {
+		t.Errorf("PostgresDDLConnectionString = %q, want %q", cfg.PostgresDDLConnectionString, dsn)
+	}
+}
+
+func TestLoadRejectsPostgresDDLConnectionStringWithoutRuntime(t *testing.T) {
+	loadEnv(t)
+	t.Setenv("ATE_API_POSTGRES_DDL_CONNECTION_STRING", "postgresql://owner@db.example:5432/atepg")
+
+	_, err := Load(Options{})
+	if err == nil || !strings.Contains(err.Error(), "requires ATE_API_POSTGRES_CONNECTION_STRING") {
+		t.Fatalf("Load() error = %v, want missing runtime DSN error", err)
 	}
 }
 
diff --git a/cmd/ate-setup/internal/steps/create.go b/cmd/ate-setup/internal/steps/create.go
index cf7e3388f0..cfa708a5ba 100644
--- a/cmd/ate-setup/internal/steps/create.go
+++ b/cmd/ate-setup/internal/steps/create.go
@@ -16,12 +16,14 @@ package steps
 
 import (
 	"context"
+	"crypto/sha256"
 	"fmt"
 	"strings"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/config"
 	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/log"
@@ -37,8 +39,11 @@ const (
 	SecretServiceDNSCA     = "service-dns-ca-pool"
 	SecretPodIdentityCA    = "pod-identity-ca-pool"
 	SecretEgressMITMCAPool = "egress-mitm-ca-pool"
+	SecretAPIServerEnvVars = "ate-api-server-secret-envvars"
+	SecretPostgresRoles    = "postgres-role-passwords"
 	ConfigMapAPIEnvVars    = "ate-api-server-envvars"
 	ConfigMapAPIAuthn      = "ate-api-authentication"
+	apiServerEnvHashKey    = "ate.dev/env-hash"
 	// poolKeyID is the identifier given to the first CA and JWT key in a new
 	// pool, matching the --ca-id/--key-id the shell scripts passed.
 	poolKeyID = "1"
@@ -119,35 +124,61 @@ func (e *Env) CreateActorIDCACertsSecret(ctx context.Context) error {
 	})
 }
 
-// CreateAPIServerEnvVars writes the ConfigMap that tells ate-api-server how to
-// reach its PostgreSQL store. ate-api-server.yaml pulls it in via an optional
-// envFrom and resolves --postgres-connection-string=@env and
-// --postgres-schema=@env from it.
+// CreateAPIServerEnvVars writes the PostgreSQL schema to a ConfigMap and the
+// credential-bearing connection strings to a Secret.
 func (e *Env) CreateAPIServerEnvVars(ctx context.Context) error {
 	log.Step("create_api_server_env_vars")
 	if err := e.Kube.EnsureNamespace(ctx, NamespaceAteSystem); err != nil {
 		return err
 	}
 
-	connString := e.Cfg.PostgresConnString()
-	log.Infof("POSTGRES_CONNECTION_STRING: %s", connString)
+	runtimeDSN, ddlDSN, err := e.postgresConnectionStrings(ctx)
+	if err != nil {
+		return err
+	}
+	log.Infof("POSTGRES_CONNECTION_STRING: configured")
+	log.Infof("POSTGRES_DDL_CONNECTION_STRING: configured")
 
-	return e.Kube.ApplyConfigMap(ctx, NamespaceAteSystem, ConfigMapAPIEnvVars,
-		buildAPIServerEnvVars(connString, e.Cfg.PostgresSchemaName()))
+	configVars := map[string]string{
+		"ATE_API_POSTGRES_SCHEMA": e.Cfg.PostgresSchemaName(),
+	}
+	secretVars := buildAPIServerSecretEnvVars(runtimeDSN, ddlDSN)
+	if err := e.Kube.ApplyConfigMap(ctx, NamespaceAteSystem, ConfigMapAPIEnvVars, configVars); err != nil {
+		return err
+	}
+	if err := e.Kube.ApplySecret(ctx, NamespaceAteSystem, SecretAPIServerEnvVars, secretVars); err != nil {
+		return err
+	}
+	return e.annotateAPIServerEnvHash(ctx, apiServerEnvHash(configVars, secretVars))
 }
 
-// buildAPIServerEnvVars is the ConfigMap payload. ate-api-server takes the
-// connection string and the schema from it, and exits on an empty schema; an
-// unrecognized key here reaches the container as a stray environment variable,
-// so the set stays exactly what the shell installer's
-// create_api_server_env_vars writes.
-func buildAPIServerEnvVars(connString, schema string) map[string]string {
+func buildAPIServerSecretEnvVars(runtimeDSN, ddlDSN string) map[string]string {
 	return map[string]string{
-		"ATE_API_POSTGRES_CONNECTION_STRING": connString,
-		"ATE_API_POSTGRES_SCHEMA":            schema,
+		"ATE_API_POSTGRES_CONNECTION_STRING":     runtimeDSN,
+		"ATE_API_POSTGRES_DDL_CONNECTION_STRING": ddlDSN,
 	}
 }
 
+func apiServerEnvHash(configVars, secretVars map[string]string) string {
+	payload := configVars["ATE_API_POSTGRES_SCHEMA"] + "\x00" +
+		secretVars["ATE_API_POSTGRES_CONNECTION_STRING"] + "\x00" +
+		secretVars["ATE_API_POSTGRES_DDL_CONNECTION_STRING"]
+	return fmt.Sprintf("%x", sha256.Sum256([]byte(payload)))
+}
+
+func (e *Env) annotateAPIServerEnvHash(ctx context.Context, hash string) error {
+	exists, err := e.Kube.DeploymentExists(ctx, NamespaceAteSystem, "ate-api-server")
+	if err != nil || !exists {
+		return err
+	}
+	patch := fmt.Sprintf(`{"spec":{"template":{"metadata":{"annotations":{"%s":%q}}}}}`, apiServerEnvHashKey, hash)
+	if _, err := e.Kube.Typed.AppsV1().Deployments(NamespaceAteSystem).Patch(
+		ctx, "ate-api-server", types.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}); err != nil {
+		return fmt.Errorf("while annotating ate-api-server environment hash: %w", err)
+	}
+	return nil
+}
+
 // CreateAPIAuthenticationConfig writes the default ate-api-server
 // authentication config, pointing it at the cluster's service account issuer.
 func (e *Env) CreateAPIAuthenticationConfig(ctx context.Context) error {
diff --git a/cmd/ate-setup/internal/steps/create_test.go b/cmd/ate-setup/internal/steps/create_test.go
index e212004831..48fac55981 100644
--- a/cmd/ate-setup/internal/steps/create_test.go
+++ b/cmd/ate-setup/internal/steps/create_test.go
@@ -15,33 +15,83 @@
 package steps
 
 import (
+	"context"
 	"crypto/x509"
 	"maps"
 	"slices"
 	"testing"
 
+	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/kube"
 	"github.com/agent-substrate/substrate/internal/localca"
 )
 
-// ate-api-server resolves --postgres-connection-string=@env and
-// --postgres-schema=@env from this ConfigMap. These are the keys the shell
-// installer writes, and an empty value for either makes the apiserver exit
-// ("--postgres-connection-string is required", "PostgreSQL schema must not be
-// empty"), so both the key set and the values are pinned here.
-func TestBuildAPIServerEnvVars(t *testing.T) {
-	const dsn = "postgresql://postgres@postgres.ate-system.svc:5432/atepg?sslmode=verify-full"
+func TestBuildAPIServerSecretEnvVars(t *testing.T) {
+	const runtimeDSN = "postgresql://runtime@postgres:5432/atepg"
+	const ddlDSN = "postgresql://ddl@postgres:5432/atepg"
 
-	got := buildAPIServerEnvVars(dsn, "public")
+	got := buildAPIServerSecretEnvVars(runtimeDSN, ddlDSN)
 
-	want := []string{"ATE_API_POSTGRES_CONNECTION_STRING", "ATE_API_POSTGRES_SCHEMA"}
+	want := []string{"ATE_API_POSTGRES_CONNECTION_STRING", "ATE_API_POSTGRES_DDL_CONNECTION_STRING"}
 	if keys := slices.Sorted(maps.Keys(got)); !slices.Equal(keys, want) {
 		t.Errorf("keys = %v, want %v", keys, want)
 	}
-	if got["ATE_API_POSTGRES_CONNECTION_STRING"] != dsn {
-		t.Errorf("ATE_API_POSTGRES_CONNECTION_STRING = %q, want %q", got["ATE_API_POSTGRES_CONNECTION_STRING"], dsn)
+	if got["ATE_API_POSTGRES_CONNECTION_STRING"] != runtimeDSN {
+		t.Errorf("ATE_API_POSTGRES_CONNECTION_STRING = %q, want %q", got["ATE_API_POSTGRES_CONNECTION_STRING"], runtimeDSN)
+	}
+	if got["ATE_API_POSTGRES_DDL_CONNECTION_STRING"] != ddlDSN {
+		t.Errorf("ATE_API_POSTGRES_DDL_CONNECTION_STRING = %q, want %q", got["ATE_API_POSTGRES_DDL_CONNECTION_STRING"], ddlDSN)
+	}
+}
+
+func TestAPIServerEnvHash(t *testing.T) {
+	configVars := map[string]string{"ATE_API_POSTGRES_SCHEMA": "substrate"}
+	secretVars := buildAPIServerSecretEnvVars("runtime", "ddl")
+
+	want := apiServerEnvHash(configVars, secretVars)
+	if got := apiServerEnvHash(configVars, secretVars); got != want {
+		t.Fatalf("stable inputs produced hashes %q and %q", want, got)
+	}
+	for name, values := range map[string][2]map[string]string{
+		"schema": {
+			{"ATE_API_POSTGRES_SCHEMA": "other"},
+			secretVars,
+		},
+		"connection string": {
+			configVars,
+			buildAPIServerSecretEnvVars("other", "ddl"),
+		},
+		"DDL connection string": {
+			configVars,
+			buildAPIServerSecretEnvVars("runtime", "other"),
+		},
+	} {
+		if got := apiServerEnvHash(values[0], values[1]); got == want {
+			t.Errorf("changing %s did not change the hash", name)
+		}
+	}
+}
+
+func TestAnnotateAPIServerEnvHash(t *testing.T) {
+	deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{
+		Name:      "ate-api-server",
+		Namespace: NamespaceAteSystem,
+	}}
+	e := &Env{Kube: &kube.Client{Typed: fake.NewSimpleClientset(deployment)}}
+
+	if err := e.annotateAPIServerEnvHash(context.Background(), "new-hash"); err != nil {
+		t.Fatal(err)
+	}
+	got, err := e.Kube.Typed.AppsV1().Deployments(NamespaceAteSystem).Get(
+		context.Background(), "ate-api-server", metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
 	}
-	if got["ATE_API_POSTGRES_SCHEMA"] != "public" {
-		t.Errorf("ATE_API_POSTGRES_SCHEMA = %q, want %q", got["ATE_API_POSTGRES_SCHEMA"], "public")
+	if got.Spec.Template.Annotations[apiServerEnvHashKey] != "new-hash" {
+		t.Errorf("environment hash annotation = %q, want new-hash", got.Spec.Template.Annotations[apiServerEnvHashKey])
 	}
 }
 
diff --git a/cmd/ate-setup/internal/steps/deploy.go b/cmd/ate-setup/internal/steps/deploy.go
index e7ea78b212..fc40d57828 100644
--- a/cmd/ate-setup/internal/steps/deploy.go
+++ b/cmd/ate-setup/internal/steps/deploy.go
@@ -234,6 +234,14 @@ func (e *Env) DeployAteAPIServer(ctx context.Context) error {
 	if err := e.applyOtelConfig(ctx); err != nil {
 		return err
 	}
+	if e.useBundledPostgres() {
+		if err := e.applyBundledPostgres(ctx); err != nil {
+			return err
+		}
+		if err := e.Kube.RolloutStatus(ctx, kube.KindStatefulSet, NamespaceAteSystem, "postgres", e.Cfg.RolloutTimeout); err != nil {
+			return err
+		}
+	}
 	if err := e.ResolveAndApply(ctx, e.Cfg.Manifest("ate-api-server.yaml")); err != nil {
 		return err
 	}
diff --git a/cmd/ate-setup/internal/steps/postgres.go b/cmd/ate-setup/internal/steps/postgres.go
index 5669ff3f80..820504c953 100644
--- a/cmd/ate-setup/internal/steps/postgres.go
+++ b/cmd/ate-setup/internal/steps/postgres.go
@@ -16,11 +16,60 @@ package steps
 
 import (
 	"context"
+	"crypto/rand"
+	"fmt"
 
 	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/kube"
 	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/log"
 )
 
+// pgx cannot derive tls-server-end-point channel-binding data from the
+// Ed25519-signed service certificate. TLS verification, client certificates,
+// and SCRAM authentication remain required independently.
+const postgresTLSParams = "sslmode=verify-full&sslrootcert=/run/servicedns.podcert.ate.dev/trust-bundle.pem&sslcert=/run/podidentity.podcert.ate.dev/credential-bundle.pem&sslkey=/run/podidentity.podcert.ate.dev/credential-bundle.pem&channel_binding=disable"
+
+func bundledPostgresDSN(role, password string) string {
+	return fmt.Sprintf("postgresql://%s:%s@postgres.ate-system.svc:5432/atepg?%s", role, password, postgresTLSParams)
+}
+
+func (e *Env) postgresConnectionStrings(ctx context.Context) (string, string, error) {
+	if runtimeDSN := e.Cfg.PostgresConnectionString; runtimeDSN != "" {
+		ddlDSN := e.Cfg.PostgresDDLConnectionString
+		if ddlDSN == "" {
+			ddlDSN = runtimeDSN
+		}
+		return runtimeDSN, ddlDSN, nil
+	}
+	runtimePassword, ddlPassword, err := e.ensureBundledPostgresCredentials(ctx)
+	if err != nil {
+		return "", "", err
+	}
+	ddlDSN := e.Cfg.PostgresDDLConnectionString
+	if ddlDSN == "" {
+		ddlDSN = bundledPostgresDSN("ateapi_ddl", ddlPassword)
+	}
+	return bundledPostgresDSN("ateapi_runtime", runtimePassword), ddlDSN, nil
+}
+
+func (e *Env) ensureBundledPostgresCredentials(ctx context.Context) (string, string, error) {
+	secret, err := e.Kube.GetSecret(ctx, NamespaceAteSystem, SecretPostgresRoles)
+	if err != nil {
+		return "", "", err
+	}
+	if secret == nil {
+		data := map[string]string{"runtime-password": rand.Text(), "ddl-password": rand.Text()}
+		if err := e.Kube.ApplySecret(ctx, NamespaceAteSystem, SecretPostgresRoles, data); err != nil {
+			return "", "", err
+		}
+		return data["runtime-password"], data["ddl-password"], nil
+	}
+	runtimePassword, ddlPassword := string(secret.Data["runtime-password"]), string(secret.Data["ddl-password"])
+	if runtimePassword == "" || ddlPassword == "" {
+		return "", "", fmt.Errorf("secret %s/%s must contain runtime-password and ddl-password", NamespaceAteSystem, SecretPostgresRoles)
+	}
+	return runtimePassword, ddlPassword, nil
+}
+
 // useBundledPostgres reports whether ateapi uses the in-cluster database
 // (when no external DSN is configured). Gates applying the bundled StatefulSet
 // and waiting on its rollout in DeployAteSystem.
@@ -46,6 +95,14 @@ func (e *Env) DeployPostgres(ctx context.Context) error {
 	if err := e.EnsureAteSystemNamespace(ctx); err != nil {
 		return err
 	}
+	if _, _, err := e.ensureBundledPostgresCredentials(ctx); err != nil {
+		return err
+	}
+	if err := e.Kube.ApplyConfigMap(ctx, NamespaceAteSystem, ConfigMapAPIEnvVars, map[string]string{
+		"ATE_API_POSTGRES_SCHEMA": e.Cfg.PostgresSchemaName(),
+	}); err != nil {
+		return err
+	}
 	if err := e.EnsurePodCertificateCAs(ctx); err != nil {
 		return err
 	}
diff --git a/cmd/ate-setup/internal/steps/postgres_test.go b/cmd/ate-setup/internal/steps/postgres_test.go
index e8c1cac39c..a43fcc1717 100644
--- a/cmd/ate-setup/internal/steps/postgres_test.go
+++ b/cmd/ate-setup/internal/steps/postgres_test.go
@@ -15,11 +15,18 @@
 package steps
 
 import (
+	"context"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
 	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/config"
+	"github.com/agent-substrate/substrate/cmd/ate-setup/internal/kube"
 )
 
 func TestUseBundledPostgres(t *testing.T) {
@@ -44,6 +51,51 @@ func TestUseBundledPostgres(t *testing.T) {
 	}
 }
 
+func TestPostgresConnectionStrings(t *testing.T) {
+	t.Run("external DDL defaults to runtime", func(t *testing.T) {
+		const dsn = "postgresql://runtime@db.example/atepg"
+		e := &Env{Cfg: &config.Config{PostgresConnectionString: dsn}}
+		runtimeDSN, ddlDSN, err := e.postgresConnectionStrings(context.Background())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if runtimeDSN != dsn || ddlDSN != dsn {
+			t.Fatalf("connection strings = %q, %q; want %q twice", runtimeDSN, ddlDSN, dsn)
+		}
+	})
+
+	t.Run("bundled credentials are reused", func(t *testing.T) {
+		secret := &corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{Name: SecretPostgresRoles, Namespace: NamespaceAteSystem},
+			Data: map[string][]byte{
+				"runtime-password": []byte("runtime-secret"),
+				"ddl-password":     []byte("ddl-secret"),
+			},
+		}
+		e := &Env{
+			Cfg:  &config.Config{},
+			Kube: &kube.Client{Typed: fake.NewSimpleClientset(secret)},
+		}
+		runtimeDSN, ddlDSN, err := e.postgresConnectionStrings(context.Background())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !strings.Contains(runtimeDSN, "ateapi_runtime:runtime-secret") || !strings.Contains(ddlDSN, "ateapi_ddl:ddl-secret") || runtimeDSN == ddlDSN {
+			t.Fatalf("unexpected bundled connection strings: %q, %q", runtimeDSN, ddlDSN)
+		}
+		if !strings.Contains(runtimeDSN, "channel_binding=disable") || !strings.Contains(ddlDSN, "channel_binding=disable") {
+			t.Fatalf("bundled connection strings do not disable unsupported channel binding: %q, %q", runtimeDSN, ddlDSN)
+		}
+		runtimeAgain, ddlAgain, err := e.postgresConnectionStrings(context.Background())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if runtimeAgain != runtimeDSN || ddlAgain != ddlDSN {
+			t.Fatal("bundled PostgreSQL credentials changed on the second read")
+		}
+	})
+}
+
 // The StatefulSet lives in a subdirectory that the bundle render does not
 // descend into, so DeployAteSystem has to apply it by name. A rename that
 // misses Manifest("postgres", "postgres.yaml") would otherwise only surface
diff --git a/cmd/ateapi/internal/store/atepg/atepg.go b/cmd/ateapi/internal/store/atepg/atepg.go
index c383a6bef5..af5fbfc92a 100644
--- a/cmd/ateapi/internal/store/atepg/atepg.go
+++ b/cmd/ateapi/internal/store/atepg/atepg.go
@@ -41,20 +41,25 @@ import (
 )
 
 // Persistence is a service that stores ate state in PostgreSQL.
-// watchPoolMaxConns sizes the dedicated outbox watch pool: one connection
-// for the WatchWorkers poller, one for the maintenance loop, and one of headroom
-// so a transiently slow poll can never gate a maintenance pass.
+// watchPoolMaxConns sizes the dedicated outbox watch pool, leaving headroom
+// so a transiently slow poll can never gate another watcher.
 const (
 	watchPoolMaxConns = 3
 	watchPoolMinConns = 1
+	// Migrations need one connection for Goose's session lock and one for
+	// migration work. Outbox maintenance is serial after startup.
+	ownerPoolMaxConns = 2
 )
 
 type Persistence struct {
 	pool *pgxpool.Pool
-	// watchPool serves the outbox side only: the WatchWorkers pollers
-	// and the partition-maintenance loop.
+	// watchPool serves the runtime-only WatchWorkers pollers. ownerPool is
+	// borrowed during startup for schema migrations, then serves outbox
+	// partition maintenance for the life of the process.
 	watchPool             *pgxpool.Pool
+	ownerPool             *pgxpool.Pool
 	ownsWatchPool         bool
+	ownsOwnerPool         bool
 	leaseTTL              time.Duration
 	pollFailureCloseAfter time.Duration
 	stopMaintenance       context.CancelFunc
@@ -107,10 +112,9 @@ var _ store.Interface = (*Persistence)(nil)
 // PostgreSQL connection. Callers can retry this error before startup.
 var ErrUnavailable = errors.New("PostgreSQL is unavailable")
 
-// Connect opens a pgxpool against dsn, creates schema if necessary, and
-// applies pending schema migrations. A dedicated watch pool isolates outbox
-// polling and maintenance from writes.
-func Connect(ctx context.Context, dsn, schema string) (*Persistence, error) {
+// Connect opens runtime and DDL pools, creates schema if necessary, and
+// applies pending schema migrations. An empty ddlDSN uses dsn for both roles.
+func Connect(ctx context.Context, dsn, ddlDSN, schema string) (*Persistence, error) {
 	if schema == "" {
 		return nil, fmt.Errorf("PostgreSQL schema must not be empty")
 	}
@@ -127,7 +131,31 @@ func Connect(ctx context.Context, dsn, schema string) (*Persistence, error) {
 		pool.Close()
 		return nil, fmt.Errorf("%w: pinging PostgreSQL: %w", ErrUnavailable, err)
 	}
-	if err := createSchema(ctx, pool, schema); err != nil {
+
+	if ddlDSN == "" {
+		ddlDSN = dsn
+	}
+	ownerCfg, err := poolConfig(ddlDSN)
+	if err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("parsing PostgreSQL DDL connection string: %w", err)
+	}
+	ownerCfg.ConnConfig.RuntimeParams["search_path"] = pgx.Identifier{schema}.Sanitize()
+	ownerCfg.MaxConns = ownerPoolMaxConns
+	ownerCfg.MinConns = 0
+	ownerCfg.MinIdleConns = 0
+	ownerPool, err := pgxpool.NewWithConfig(ctx, ownerCfg)
+	if err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("opening PostgreSQL DDL pool: %w", err)
+	}
+	if err := ownerPool.Ping(ctx); err != nil {
+		ownerPool.Close()
+		pool.Close()
+		return nil, fmt.Errorf("%w: pinging PostgreSQL DDL connection: %w", ErrUnavailable, err)
+	}
+	if err := createSchema(ctx, ownerPool, schema); err != nil {
+		ownerPool.Close()
 		pool.Close()
 		return nil, err
 	}
@@ -137,17 +165,20 @@ func Connect(ctx context.Context, dsn, schema string) (*Persistence, error) {
 	watchCfg.MinConns = watchPoolMinConns
 	watchPool, err := pgxpool.NewWithConfig(ctx, watchCfg)
 	if err != nil {
+		ownerPool.Close()
 		pool.Close()
 		return nil, fmt.Errorf("opening PostgreSQL watch pool: %w", err)
 	}
 
-	p, err := newPersistence(ctx, pool, watchPool)
+	p, err := newPersistence(ctx, pool, watchPool, ownerPool)
 	if err != nil {
 		watchPool.Close()
+		ownerPool.Close()
 		pool.Close()
 		return nil, err
 	}
 	p.ownsWatchPool = true
+	p.ownsOwnerPool = true
 	return p, nil
 }
 
@@ -208,17 +239,21 @@ func poolConfig(dsn string) (*pgxpool.Config, error) {
 // Callers that already hold a pool (e.g. tests using testcontainers) use
 // this directly instead of Connect; outbox watch traffic shares the given pool.
 func NewPersistence(ctx context.Context, pool *pgxpool.Pool) (*Persistence, error) {
-	return newPersistence(ctx, pool, pool)
+	return newPersistence(ctx, pool, pool, pool)
 }
 
-func newPersistence(ctx context.Context, pool, watchPool *pgxpool.Pool) (*Persistence, error) {
-	if err := applyMigrations(ctx, pool); err != nil {
+func newPersistence(ctx context.Context, pool, watchPool, ownerPool *pgxpool.Pool) (*Persistence, error) {
+	if err := applyMigrations(ctx, ownerPool); err != nil {
+		return nil, err
+	}
+	if err := grantRuntimePrivileges(ctx, ownerPool, pool.Config().ConnConfig.User); err != nil {
 		return nil, err
 	}
 	maintenanceCtx, stopMaintenance := context.WithCancel(context.Background())
 	p := &Persistence{
 		pool:                  pool,
 		watchPool:             watchPool,
+		ownerPool:             ownerPool,
 		leaseTTL:              defaultLeaseTTL,
 		pollFailureCloseAfter: outboxPollFailureCloseAfter,
 		stopMaintenance:       stopMaintenance,
@@ -245,14 +280,17 @@ func newPersistence(ctx context.Context, pool, watchPool *pgxpool.Pool) (*Persis
 }
 
 // Close stops the outbox maintenance loop and waits for it to exit,
-// then closes the watch pool if Connect created one. It does not close the
-// main pool, which the caller owns.
+// then closes the auxiliary pools if Connect created them. It does not close
+// the main pool, which the caller owns.
 func (p *Persistence) Close() {
 	p.stopMaintenance()
 	<-p.maintenanceDone
 	if p.ownsWatchPool {
 		p.watchPool.Close()
 	}
+	if p.ownsOwnerPool {
+		p.ownerPool.Close()
+	}
 }
 
 // querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting read helpers
diff --git a/cmd/ateapi/internal/store/atepg/atepg_test.go b/cmd/ateapi/internal/store/atepg/atepg_test.go
index d2ed623991..a6222983e1 100644
--- a/cmd/ateapi/internal/store/atepg/atepg_test.go
+++ b/cmd/ateapi/internal/store/atepg/atepg_test.go
@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/pressly/goose/v3"
 	"github.com/testcontainers/testcontainers-go/modules/postgres"
@@ -136,7 +137,7 @@ func TestMigrationsConcurrentStartup(t *testing.T) {
 	errs := make(chan error, 2)
 	for range 2 {
 		go func() {
-			p, err := Connect(ctx, containerDSN, "concurrent-startup")
+			p, err := Connect(ctx, containerDSN, "", "concurrent-startup")
 			if p != nil {
 				p.Close()
 				p.pool.Close()
@@ -252,7 +253,7 @@ func TestConnectUsesConfiguredSchema(t *testing.T) {
 	if err != nil {
 		t.Fatalf("getting PostgreSQL connection string: %v", err)
 	}
-	persistence, err := Connect(ctx, dsn+"&search_path=public", schema)
+	persistence, err := Connect(ctx, dsn+"&search_path=public", "", schema)
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
@@ -285,7 +286,7 @@ func TestConnectUsesConfiguredSchema(t *testing.T) {
 		t.Error("migration removed an unrelated table")
 	}
 
-	if err := persistence.dropExpiredWorkerOutboxPartitions(ctx, persistence.watchPool, time.Now()); err != nil {
+	if err := persistence.dropExpiredWorkerOutboxPartitions(ctx, persistence.ownerPool, time.Now()); err != nil {
 		t.Fatalf("dropping expired partitions in the configured schema: %v", err)
 	}
 	var unrelatedPartitionExists bool
@@ -297,6 +298,104 @@ func TestConnectUsesConfiguredSchema(t *testing.T) {
 	}
 }
 
+func TestConnectSeparatesRuntimeAndDDLPrivileges(t *testing.T) {
+	admin := requirePool(t)
+	ctx := t.Context()
+	const (
+		schema      = "separate-role-test"
+		runtimeRole = "atepg_runtime_test"
+		ddlRole     = "atepg_ddl_test"
+		password    = "test-password"
+	)
+	if _, err := admin.Exec(ctx, fmt.Sprintf(`
+		DROP SCHEMA IF EXISTS %s CASCADE;
+		DROP ROLE IF EXISTS %s;
+		DROP ROLE IF EXISTS %s;
+		CREATE ROLE %s LOGIN PASSWORD '%s';
+		CREATE ROLE %s LOGIN PASSWORD '%s';
+		GRANT CREATE ON DATABASE atepg TO %s`,
+		pgx.Identifier{schema}.Sanitize(), pgx.Identifier{runtimeRole}.Sanitize(),
+		pgx.Identifier{ddlRole}.Sanitize(), pgx.Identifier{runtimeRole}.Sanitize(), password,
+		pgx.Identifier{ddlRole}.Sanitize(), password, pgx.Identifier{ddlRole}.Sanitize())); err != nil {
+		t.Fatalf("creating PostgreSQL test roles: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = admin.Exec(context.Background(), fmt.Sprintf(`
+			DROP SCHEMA IF EXISTS %s CASCADE;
+			REVOKE ALL ON DATABASE atepg FROM %s;
+			REVOKE ALL ON DATABASE atepg FROM %s;
+			DROP ROLE IF EXISTS %s;
+			DROP ROLE IF EXISTS %s`, pgx.Identifier{schema}.Sanitize(),
+			pgx.Identifier{runtimeRole}.Sanitize(), pgx.Identifier{ddlRole}.Sanitize(),
+			pgx.Identifier{runtimeRole}.Sanitize(), pgx.Identifier{ddlRole}.Sanitize()))
+	})
+
+	runtimeDSN := strings.Replace(containerDSN, "://atepg:atepg@", "://"+runtimeRole+":"+password+"@", 1)
+	ddlDSN := strings.Replace(containerDSN, "://atepg:atepg@", "://"+ddlRole+":"+password+"@", 1)
+	if runtimeDSN == containerDSN || ddlDSN == containerDSN {
+		t.Fatalf("unexpected test DSN format: %q", containerDSN)
+	}
+	p, err := Connect(ctx, runtimeDSN, ddlDSN, schema)
+	if err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	defer p.pool.Close()
+	defer p.Close()
+
+	if _, err := p.CreateAtespace(ctx, newTestAtespace("runtime-write")); err != nil {
+		t.Fatalf("runtime DML failed: %v", err)
+	}
+	if _, err := p.pool.Exec(ctx, `CREATE TABLE forbidden (id integer)`); err == nil {
+		t.Fatal("runtime role created a table")
+	}
+	if _, err := p.pool.Exec(ctx, `UPDATE schema_migrations SET is_applied = false`); err == nil {
+		t.Fatal("runtime role modified the migration ledger")
+	}
+	if err := p.createWorkerOutboxPartitions(ctx, time.Now().Add(24*time.Hour)); err != nil {
+		t.Fatalf("DDL maintenance failed: %v", err)
+	}
+}
+
+func TestConnectSingleRoleDoesNotRequireSchemaOwnership(t *testing.T) {
+	admin := requirePool(t)
+	ctx := t.Context()
+	const (
+		schema   = "single-role-test"
+		role     = "atepg_single_role_test"
+		password = "test-password"
+	)
+	if _, err := admin.Exec(ctx, fmt.Sprintf(`
+		DROP SCHEMA IF EXISTS %s CASCADE;
+		DROP ROLE IF EXISTS %s;
+		CREATE ROLE %s LOGIN PASSWORD '%s';
+		CREATE SCHEMA %s;
+		GRANT CREATE ON DATABASE atepg TO %s;
+		GRANT USAGE, CREATE ON SCHEMA %s TO %s`,
+		pgx.Identifier{schema}.Sanitize(), pgx.Identifier{role}.Sanitize(),
+		pgx.Identifier{role}.Sanitize(), password, pgx.Identifier{schema}.Sanitize(),
+		pgx.Identifier{role}.Sanitize(), pgx.Identifier{schema}.Sanitize(), pgx.Identifier{role}.Sanitize())); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = admin.Exec(context.Background(), fmt.Sprintf(`
+			DROP SCHEMA IF EXISTS %s CASCADE;
+			REVOKE ALL ON DATABASE atepg FROM %s;
+			DROP ROLE IF EXISTS %s`, pgx.Identifier{schema}.Sanitize(),
+			pgx.Identifier{role}.Sanitize(), pgx.Identifier{role}.Sanitize()))
+	})
+
+	dsn := strings.Replace(containerDSN, "://atepg:atepg@", "://"+role+":"+password+"@", 1)
+	p, err := Connect(ctx, dsn, "", schema)
+	if err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	defer p.pool.Close()
+	defer p.Close()
+	if _, err := p.CreateAtespace(ctx, newTestAtespace("single-role-write")); err != nil {
+		t.Fatalf("runtime DML failed: %v", err)
+	}
+}
+
 func TestMigrationSchemaStates(t *testing.T) {
 	pool := requirePool(t)
 	ctx := t.Context()
@@ -309,7 +408,7 @@ func TestMigrationSchemaStates(t *testing.T) {
 			_, _ = pool.Exec(context.Background(), `DROP SCHEMA IF EXISTS "migration-ahead" CASCADE`)
 		})
 
-		p, err := Connect(ctx, containerDSN, "migration-ahead")
+		p, err := Connect(ctx, containerDSN, "", "migration-ahead")
 		if err != nil {
 			t.Fatalf("creating current schema: %v", err)
 		}
@@ -319,7 +418,7 @@ func TestMigrationSchemaStates(t *testing.T) {
 			t.Fatalf("setting ahead migration state: %v", err)
 		}
 
-		p, err = Connect(ctx, containerDSN, "migration-ahead")
+		p, err = Connect(ctx, containerDSN, "", "migration-ahead")
 		if err != nil {
 			t.Fatalf("Connect with an ahead clean schema failed: %v", err)
 		}
@@ -338,7 +437,7 @@ func TestMigrationSchemaStates(t *testing.T) {
 			_, _ = pool.Exec(context.Background(), `DROP SCHEMA IF EXISTS "migration-legacy" CASCADE`)
 		})
 
-		_, err := Connect(ctx, containerDSN, "migration-legacy")
+		_, err := Connect(ctx, containerDSN, "", "migration-legacy")
 		if err == nil || !strings.Contains(err.Error(), "Substrate tables exist without a migration ledger") {
 			t.Fatalf("Connect error = %v, want unsupported schema error", err)
 		}
@@ -873,7 +972,7 @@ func TestSaveWorker_RejectsAStaleWrite(t *testing.T) {
 	requirePool(t)
 	ctx := context.Background()
 
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
@@ -915,7 +1014,7 @@ func TestSaveWorker_RejectsAVanishedWorker(t *testing.T) {
 	requirePool(t)
 	ctx := context.Background()
 
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
diff --git a/cmd/ateapi/internal/store/atepg/outbox.go b/cmd/ateapi/internal/store/atepg/outbox.go
index c20e8646af..fd1745cd93 100644
--- a/cmd/ateapi/internal/store/atepg/outbox.go
+++ b/cmd/ateapi/internal/store/atepg/outbox.go
@@ -213,7 +213,7 @@ func (p *Persistence) maintainWorkerOutboxPartitions(ctx context.Context) error
 // own elected transaction. Locks touched: DEFAULT child only — never the
 // parent (see maintainWorkerOutboxPartitions on deadlock ordering).
 func (p *Persistence) retireStrayedOutboxDefault(ctx context.Context) error {
-	tx, err := p.watchPool.Begin(ctx)
+	tx, err := p.ownerPool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("beginning outbox stray-cleanup transaction: %w", err)
 	}
@@ -255,7 +255,7 @@ func (p *Persistence) retireStrayedOutboxDefault(ctx context.Context) error {
 // worker write's outbox append) plus the dropped children — never the
 // DEFAULT while waiting on the parent.
 func (p *Persistence) dropExpiredOutboxRetention(ctx context.Context, now time.Time) error {
-	tx, err := p.watchPool.Begin(ctx)
+	tx, err := p.ownerPool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("beginning outbox retention transaction: %w", err)
 	}
@@ -313,7 +313,7 @@ func (p *Persistence) createWorkerOutboxPartitions(ctx context.Context, instants
 func isCheckViolation(err error) bool { return pgErrCode(err) == "23514" }
 
 func (p *Persistence) tryCreateWorkerOutboxPartitions(ctx context.Context, truncateStrays bool, instants ...time.Time) error {
-	tx, err := p.watchPool.Begin(ctx)
+	tx, err := p.ownerPool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("beginning outbox partition transaction: %w", err)
 	}
diff --git a/cmd/ateapi/internal/store/atepg/outbox_test.go b/cmd/ateapi/internal/store/atepg/outbox_test.go
index 22a9f9f245..5def1be9a4 100644
--- a/cmd/ateapi/internal/store/atepg/outbox_test.go
+++ b/cmd/ateapi/internal/store/atepg/outbox_test.go
@@ -34,15 +34,14 @@ import (
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 )
 
-// TestConnect_DedicatedWatchPool covers the dual-pool path only Connect
-// takes (the rest of the suite uses NewPersistence, where feed traffic
-// shares the caller's pool): the watch pool must be distinct and owned, and
-// the watch must deliver through it end to end.
+// TestConnect_DedicatedWatchPool covers the auxiliary-pool path only Connect
+// takes (the rest of the suite uses NewPersistence, where every operation
+// shares the caller's pool).
 func TestConnect_DedicatedWatchPool(t *testing.T) {
 	requirePool(t) // ensures the container is up and containerDSN is set
 	ctx := context.Background()
 
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
@@ -55,9 +54,15 @@ func TestConnect_DedicatedWatchPool(t *testing.T) {
 	if !p.ownsWatchPool {
 		t.Fatal("Connect must own the watch pool so Close releases it")
 	}
+	if p.ownerPool == p.pool || p.ownerPool == p.watchPool || !p.ownsOwnerPool {
+		t.Fatal("Connect must own a dedicated owner pool")
+	}
 	if got := p.watchPool.Config().MaxConns; got != watchPoolMaxConns {
 		t.Fatalf("watch pool MaxConns = %d, want %d", got, watchPoolMaxConns)
 	}
+	if got := p.ownerPool.Config().MaxConns; got != ownerPoolMaxConns {
+		t.Fatalf("owner pool MaxConns = %d, want %d", got, ownerPoolMaxConns)
+	}
 
 	clearAll(t, p)
 	watch, err := p.WatchWorkers(ctx)
@@ -859,7 +864,7 @@ func TestWatchWorkers_ClosesAfterPersistentPollFailure(t *testing.T) {
 
 	// Connect so the watcher has its own pool: killing it simulates a
 	// persistent outage without touching the shared container pool.
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
@@ -1036,7 +1041,7 @@ func TestLocalPublishReachesWatchers(t *testing.T) {
 	requirePool(t)
 	ctx := context.Background()
 
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
@@ -1120,7 +1125,7 @@ func TestLocalPublishSurvivesWatchClose(t *testing.T) {
 	requirePool(t)
 	ctx := context.Background()
 
-	p, err := Connect(ctx, containerDSN, "public")
+	p, err := Connect(ctx, containerDSN, "", "public")
 	if err != nil {
 		t.Fatalf("Connect failed: %v", err)
 	}
diff --git a/cmd/ateapi/internal/store/atepg/schema.go b/cmd/ateapi/internal/store/atepg/schema.go
index 0ca7ac80a9..f4525d4677 100644
--- a/cmd/ateapi/internal/store/atepg/schema.go
+++ b/cmd/ateapi/internal/store/atepg/schema.go
@@ -23,6 +23,7 @@ import (
 	"log/slog"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jackc/pgx/v5/stdlib"
 	"github.com/pressly/goose/v3"
@@ -67,6 +68,41 @@ func applyMigrations(ctx context.Context, pool *pgxpool.Pool) error {
 	return errors.Join(migrateToLatest(ctx, provider), provider.Close())
 }
 
+// grantRuntimePrivileges gives the runtime role DML access to schema objects
+// while keeping the migration ledger private to the DDL role.
+func grantRuntimePrivileges(ctx context.Context, pool *pgxpool.Pool, runtimeRole string) error {
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("starting PostgreSQL runtime grant transaction: %w", err)
+	}
+	defer tx.Rollback(ctx) //nolint:errcheck // Commit or the returned error decides the outcome.
+
+	var schema, ddlRole string
+	if err := tx.QueryRow(ctx, `SELECT current_schema(), current_user`).Scan(&schema, &ddlRole); err != nil {
+		return fmt.Errorf("get PostgreSQL schema for runtime grants: %w", err)
+	}
+	if ddlRole == runtimeRole {
+		return nil
+	}
+	role := pgx.Identifier{runtimeRole}.Sanitize()
+	schemaName := pgx.Identifier{schema}.Sanitize()
+	migrationTable := pgx.Identifier{schema, migrationTableName}.Sanitize()
+
+	_, err = tx.Exec(ctx, fmt.Sprintf(`
+		GRANT USAGE ON SCHEMA %[1]s TO %[2]s;
+		GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA %[1]s TO %[2]s;
+		GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA %[1]s TO %[2]s;
+		REVOKE ALL PRIVILEGES ON TABLE %[3]s FROM %[2]s`,
+		schemaName, role, migrationTable))
+	if err != nil {
+		return fmt.Errorf("granting PostgreSQL runtime privileges to %q: %w", runtimeRole, err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("committing PostgreSQL runtime privileges for %q: %w", runtimeRole, err)
+	}
+	return nil
+}
+
 func openMigrationProvider(ctx context.Context, pool *pgxpool.Pool, migrations fs.FS) (*goose.Provider, error) {
 	lockID, err := migrationLockID(ctx, pool)
 	if err != nil {
diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go
index 4ac421cf63..b158b3c732 100644
--- a/cmd/ateapi/main.go
+++ b/cmd/ateapi/main.go
@@ -71,9 +71,10 @@ var (
 	metricsListenAddr    = pflag.String("metrics-listen-addr", ":9090", "Address and port the prometheus metrics server should listen on.")
 	grpcServerCredBundle = pflag.String("grpc-server-cred-bundle", "", "File with the server TLS credential bundle.")
 
-	authenticationConfigFile = pflag.String("authentication-config", "", "YAML file configuring trusted JWT providers.")
-	postgresConnectionString = pflag.String("postgres-connection-string", "", "PostgreSQL connection string (libpq DSN or URI).")
-	postgresSchema           = pflag.String("postgres-schema", "public", "PostgreSQL schema for Substrate tables. This overrides a search_path connection parameter.")
+	authenticationConfigFile    = pflag.String("authentication-config", "", "YAML file configuring trusted JWT providers.")
+	postgresConnectionString    = pflag.String("postgres-connection-string", "", "PostgreSQL connection string (libpq DSN or URI).")
+	postgresDDLConnectionString = pflag.String("postgres-ddl-connection-string", "", "PostgreSQL DDL and maintenance connection string. Defaults to --postgres-connection-string.")
+	postgresSchema              = pflag.String("postgres-schema", "public", "PostgreSQL schema for Substrate tables. This overrides a search_path connection parameter.")
 
 	actorIDJWTPoolFile   = pflag.String("actor-id-jwt-pool", "", "The file that contains the serialized JWT authority pool for signing actor JWTs")
 	egressGatewayAddress = pflag.String("egress-gateway-address", "", "Address of the egress PEP. Empty disables tunneled egress.")
@@ -330,6 +331,7 @@ func loadFlagsFromEnv() {
 		env  string
 	}{
 		{postgresConnectionString, "ATE_API_POSTGRES_CONNECTION_STRING"},
+		{postgresDDLConnectionString, "ATE_API_POSTGRES_DDL_CONNECTION_STRING"},
 		{postgresSchema, "ATE_API_POSTGRES_SCHEMA"},
 	}
 	for _, o := range overrides {
@@ -344,7 +346,8 @@ func logFlagValues(ctx context.Context) {
 		slog.String("grpc-listen-addr", *listenAddr),
 		slog.String("grpc-server-cred-bundle", *grpcServerCredBundle),
 		slog.String("authentication-config", *authenticationConfigFile),
-		slog.String("postgres-connection-string", *postgresConnectionString),
+		slog.Bool("postgres-connection-string-set", *postgresConnectionString != ""),
+		slog.Bool("postgres-ddl-connection-string-set", *postgresDDLConnectionString != ""),
 		slog.String("postgres-schema", *postgresSchema),
 		slog.String("actor-id-jwt-pool", *actorIDJWTPoolFile),
 		slog.String("actor-id-ca-pool", *actorIDCAPoolFile),
@@ -393,6 +396,11 @@ func connectStore(ctx context.Context) (store.Interface, error) {
 	if _, err := pgxpool.ParseConfig(*postgresConnectionString); err != nil {
 		return nil, fmt.Errorf("parsing PostgreSQL connection string: %w", err)
 	}
+	if *postgresDDLConnectionString != "" {
+		if _, err := pgxpool.ParseConfig(*postgresDDLConnectionString); err != nil {
+			return nil, fmt.Errorf("parsing PostgreSQL DDL connection string: %w", err)
+		}
+	}
 	persistence, err := connectPostgresWithRetries(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("setting up PostgreSQL: %w", err)
@@ -408,7 +416,7 @@ var (
 func connectPostgresWithRetries(ctx context.Context) (*atepg.Persistence, error) {
 	var connectErr error
 	for attempt := 1; attempt <= postgresConnectTries; attempt++ {
-		persistence, err := atepg.Connect(ctx, *postgresConnectionString, *postgresSchema)
+		persistence, err := atepg.Connect(ctx, *postgresConnectionString, *postgresDDLConnectionString, *postgresSchema)
 		if err == nil {
 			return persistence, nil
 		}
diff --git a/docs/dev/postgresql-schema-evolution.md b/docs/dev/postgresql-schema-evolution.md
index 4b60280fc8..4e436f02c8 100644
--- a/docs/dev/postgresql-schema-evolution.md
+++ b/docs/dev/postgresql-schema-evolution.md
@@ -57,6 +57,7 @@ Store migration files in `cmd/ateapi/internal/store/atepg/migrations`.
 - Do not add SQL transaction control statements.
 - Do not use `IF NOT EXISTS` for a schema change.
 - Keep each startup migration short.
+- Runtime DML privileges are granted to all tables and sequences after migrations. Add an explicit exception for any new DDL-only table, like the Goose migration ledger.
 
 Before the first stable v1 release, developers can change or squash migration files. Recreate a development database after its migration history changes.
 
diff --git a/hack/install-ate.sh b/hack/install-ate.sh
index 278467d1cf..800338324e 100755
--- a/hack/install-ate.sh
+++ b/hack/install-ate.sh
@@ -123,11 +123,12 @@ function usage() {
   echo "  --create-api-server-env-vars           Create ate-api-server env vars"
   echo "  --create-api-authentication-config     Create the default ate-api-server authentication config"
   echo ""
-  echo "PostgreSQL configuration (either of the first two selects an external"
-  echo "database and skips the bundled instance):"
+  echo "PostgreSQL configuration (a runtime DSN or Cloud SQL instance selects an"
+  echo "external database and skips the bundled instance):"
   echo ""
-  echo "  ATE_API_POSTGRES_CONNECTION_STRING     DSN for any external PostgreSQL (stored in a Secret;"
+  echo "  ATE_API_POSTGRES_CONNECTION_STRING     Runtime/DML DSN for any external PostgreSQL (stored in a Secret;"
   echo "                                         pair with ATE_API_POSTGRES_SERVER_CA_FILE for sslmode=verify-ca)"
+  echo "  ATE_API_POSTGRES_DDL_CONNECTION_STRING DDL/migration DSN (requires a runtime DSN; defaults to it)"
   echo "  ATE_API_POSTGRES_CLOUDSQL_INSTANCE     Cloud SQL instance connection name (project:region:instance)."
   echo "                                         Deploys the Cloud SQL Auth Proxy sidecar: connector-managed TLS"
   echo "                                         and automatic IAM database auth, no passwords (see tools/setup-gcp/cloud-sql.md)."
@@ -137,7 +138,8 @@ function usage() {
   echo "  ATE_API_POSTGRES_CLOUDSQL_IAM_AUTH     true (default) | false (password-over-proxy escape hatch)"
   echo "  ATE_API_POSTGRES_POOL_MAX_CONNS        pgxpool max connections per ateapi replica (default: max(4, NumCPU))"
   echo "  ATE_API_POSTGRES_SERVER_CA_FILE        PEM file to mount for verify-ca DSNs (non-Cloud-SQL databases)"
-  echo "  ATE_API_POSTGRES_SCHEMA                Select the Substrate schema (default: public)"
+  echo "  ATE_API_POSTGRES_SCHEMA                Select the Substrate schema (default: public; use a dedicated schema"
+  echo "                                         when configuring separate runtime and DDL roles)"
   echo ""
   echo "Authentication configuration:"
   echo ""
@@ -265,7 +267,10 @@ rollout_timeout() {
 }
 
 default_postgres_connection_string() {
-  echo "postgresql://postgres@postgres.ate-system.svc:5432/atepg?sslmode=verify-full&sslrootcert=/run/servicedns.podcert.ate.dev/trust-bundle.pem&sslcert=/run/podidentity.podcert.ate.dev/credential-bundle.pem&sslkey=/run/podidentity.podcert.ate.dev/credential-bundle.pem"
+  local role="$1" password="$2"
+  # pgx cannot derive tls-server-end-point channel-binding data from the
+  # Ed25519-signed service certificate. TLS, mTLS, and SCRAM remain required.
+  echo "postgresql://${role}:${password}@postgres.ate-system.svc:5432/atepg?sslmode=verify-full&sslrootcert=/run/servicedns.podcert.ate.dev/trust-bundle.pem&sslcert=/run/podidentity.podcert.ate.dev/credential-bundle.pem&sslkey=/run/podidentity.podcert.ate.dev/credential-bundle.pem&channel_binding=disable"
 }
 
 # True if deploying the bundled in-cluster PostgreSQL. Returns false if an
@@ -275,6 +280,15 @@ use_bundled_postgres() {
   [[ -z "${ATE_API_POSTGRES_CONNECTION_STRING:-}" && -z "$(resolve_cloudsql_instance)" ]]
 }
 
+ensure_bundled_postgres_credentials() {
+  if run_kubectl get secret -n ate-system postgres-role-passwords >/dev/null 2>&1; then
+    return
+  fi
+  run_kubectl create secret generic -n ate-system postgres-role-passwords \
+    --from-literal=runtime-password="$(openssl rand -hex 32)" \
+    --from-literal=ddl-password="$(openssl rand -hex 32)"
+}
+
 # --- Versioned dataplane rendering ---
 #
 # The atelet DaemonSet is keyed by substrate version (name suffix + nodeSelector on
@@ -671,7 +685,12 @@ create_api_server_env_vars() {
     | run_kubectl apply -f -
 
   local postgres_connection_string="${ATE_API_POSTGRES_CONNECTION_STRING:-}"
+  local postgres_ddl_connection_string="${ATE_API_POSTGRES_DDL_CONNECTION_STRING:-}"
   local postgres_schema="${ATE_API_POSTGRES_SCHEMA:-public}"
+  if [[ -n "${postgres_ddl_connection_string}" && -z "${postgres_connection_string}" ]]; then
+    echo "Error: ATE_API_POSTGRES_DDL_CONNECTION_STRING requires ATE_API_POSTGRES_CONNECTION_STRING" >&2
+    exit 1
+  fi
   # Distinguishes a DSN the operator supplied on this run from one
   # synthesized, defaulted, or adopted back from the Secret: only the former
   # outranks ATE_API_POSTGRES_POOL_MAX_CONNS below.
@@ -704,6 +723,10 @@ create_api_server_env_vars() {
         postgres_connection_string="$(run_kubectl get secret -n ate-system ate-api-server-secret-envvars \
           -o jsonpath='{.data.ATE_API_POSTGRES_CONNECTION_STRING}' 2>/dev/null | base64 --decode || true)"
       fi
+      if [[ -z "${postgres_ddl_connection_string}" ]]; then
+        postgres_ddl_connection_string="$(run_kubectl get secret -n ate-system ate-api-server-secret-envvars \
+          -o jsonpath='{.data.ATE_API_POSTGRES_DDL_CONNECTION_STRING}' 2>/dev/null | base64 --decode || true)"
+      fi
     fi
   fi
   if [[ -z "${postgres_connection_string}" ]]; then
@@ -730,7 +753,16 @@ create_api_server_env_vars() {
       fi
       postgres_connection_string="user=${cloudsql_gsa%.gserviceaccount.com} host=127.0.0.1 port=5432 dbname=atepg sslmode=disable"
     else
-      postgres_connection_string="$(default_postgres_connection_string)"
+      ensure_bundled_postgres_credentials
+      local runtime_password ddl_password
+      runtime_password="$(run_kubectl get secret -n ate-system postgres-role-passwords \
+        -o jsonpath='{.data.runtime-password}' | base64 --decode)"
+      ddl_password="$(run_kubectl get secret -n ate-system postgres-role-passwords \
+        -o jsonpath='{.data.ddl-password}' | base64 --decode)"
+      postgres_connection_string="$(default_postgres_connection_string ateapi_runtime "${runtime_password}")"
+      if [[ -z "${postgres_ddl_connection_string}" ]]; then
+        postgres_ddl_connection_string="$(default_postgres_connection_string ateapi_ddl "${ddl_password}")"
+      fi
     fi
   fi
   # Appends pgxpool sizing (pool_max_conns) to the DSN to prevent silent client
@@ -752,9 +784,17 @@ create_api_server_env_vars() {
     fi
   fi
 
+  # A separate DDL credential is optional for external databases so existing
+  # installs retain their single-role behavior.
+  if [[ -z "${postgres_ddl_connection_string}" ]]; then
+    postgres_ddl_connection_string="${postgres_connection_string}"
+  fi
+
   # Redact any password before logging (URI user:pw@host and keyword password=).
   echo "POSTGRES_CONNECTION_STRING: $(printf '%s' "${postgres_connection_string}" \
     | sed -E 's#(://[^:/@]*):[^@]*@#\1:***@#; s/(password=)[^ &]*/\1***/g')"
+  echo "POSTGRES_DDL_CONNECTION_STRING: $(printf '%s' "${postgres_ddl_connection_string}" \
+    | sed -E 's#(://[^:/@]*):[^@]*@#\1:***@#; s/(password=)[^ &]*/\1***/g')"
 
   # Empty unless Cloud SQL is configured; expanded below with the
   # ${arr[@]+...} idiom because bash 3.2's nounset rejects "${arr[@]}" on an
@@ -787,6 +827,7 @@ create_api_server_env_vars() {
   fi
   run_kubectl create configmap -n ate-system ate-api-server-envvars \
     ${cm_args[@]+"${cm_args[@]}"} \
+    --from-literal=ATE_API_POSTGRES_SCHEMA="${postgres_schema}" \
     --dry-run=client -o yaml \
     | run_kubectl apply -f -
 
@@ -796,7 +837,7 @@ create_api_server_env_vars() {
   # define the key.
   run_kubectl create secret generic -n ate-system ate-api-server-secret-envvars \
     --from-literal=ATE_API_POSTGRES_CONNECTION_STRING="${postgres_connection_string}" \
-    --from-literal=ATE_API_POSTGRES_SCHEMA="${postgres_schema}" \
+    --from-literal=ATE_API_POSTGRES_DDL_CONNECTION_STRING="${postgres_ddl_connection_string}" \
     --dry-run=client -o yaml \
     | run_kubectl apply -f -
 
@@ -1041,6 +1082,11 @@ deploy_ate_apiserver() {
   apply_otel_config
   apply_otel_endpoint_override
 
+  if use_bundled_postgres; then
+    apply_postgres
+    run_kubectl rollout status statefulset/postgres -n ate-system --timeout="$(rollout_timeout)"
+  fi
+
   run_ko apply -f manifests/ate-install/ate-api-server.yaml
   reconcile_cloudsql_proxy_sidecar
   run_kubectl rollout status deployment/ate-api-server -n ate-system --timeout="$(rollout_timeout)"
diff --git a/manifests/ate-install/ate-api-server.yaml b/manifests/ate-install/ate-api-server.yaml
index cde0c42ab8..a4e6f7089a 100644
--- a/manifests/ate-install/ate-api-server.yaml
+++ b/manifests/ate-install/ate-api-server.yaml
@@ -90,6 +90,7 @@ spec:
           - "--grpc-server-cred-bundle=/run/servicedns.podcert.ate.dev/credential-bundle.pem"
           - --authentication-config=/etc/ateapi/authentication/authentication.yaml
           - --postgres-connection-string=@env
+          - --postgres-ddl-connection-string=@env
           - --postgres-schema=@env
           - --actor-id-jwt-pool=/run/actor-id-jwt-pool/pool.json
           - --actor-id-ca-pool=/run/actor-id-ca-pool/pool.json
diff --git a/manifests/ate-install/postgres/postgres.yaml b/manifests/ate-install/postgres/postgres.yaml
index 14a11ea473..ff6c11a8fd 100644
--- a/manifests/ate-install/postgres/postgres.yaml
+++ b/manifests/ate-install/postgres/postgres.yaml
@@ -30,9 +30,10 @@ data:
     # health checks, the workload's idempotent database bootstrap, and the
     # tls-reloader sidecar's configuration reloads.
     local all all trust
-    # PostgreSQL verifies client certificates against the pod-identity CA. It
-    # does not need its own serving CA because it never verifies its server certificate.
-    hostssl all all all trust clientcert=verify-ca
+    # Only ateapi's two roles may connect over TCP. PostgreSQL requires both a
+    # password and a client certificate signed by the pod-identity CA.
+    hostssl atepg ateapi_runtime all scram-sha-256 clientcert=verify-ca
+    hostssl atepg ateapi_ddl all scram-sha-256 clientcert=verify-ca
   reload-tls.sh: |
     # PostgreSQL opens ssl_cert_file, ssl_key_file and ssl_ca_file at startup
     # and on SIGHUP, and nowhere else. The kubelet replaces the projected pod
@@ -57,7 +58,7 @@ data:
 
     reloaded=""
     while true; do
-      current="$(sha256sum "${CERT}" "${CA}")"
+      current="$(sha256sum "${CERT}" "${CA}" /etc/postgresql/pg_hba.conf)"
       # Reloading fails until the server is accepting connections, which is
       # where every pod starts out, so only record a hash once it has worked.
       # Starting empty also means a restart of this container costs one
@@ -158,6 +159,41 @@ spec:
                   "SELECT 1 FROM pg_database WHERE datname = 'atepg'" | grep -qx 1; then
                   createdb -U postgres atepg
                 fi
+                psql -v ON_ERROR_STOP=1 -U postgres -d postgres \
+                  -v runtime_password="${ATEAPI_RUNTIME_PASSWORD}" \
+                  -v ddl_password="${ATEAPI_DDL_PASSWORD}" <<'SQL'
+                SET password_encryption = 'scram-sha-256';
+                SELECT 'CREATE ROLE ateapi_runtime LOGIN'
+                  WHERE NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'ateapi_runtime') \gexec
+                SELECT 'CREATE ROLE ateapi_ddl LOGIN'
+                  WHERE NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'ateapi_ddl') \gexec
+                ALTER ROLE ateapi_runtime NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :'runtime_password';
+                ALTER ROLE ateapi_ddl NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :'ddl_password';
+                REVOKE CONNECT, CREATE, TEMPORARY ON DATABASE atepg FROM PUBLIC;
+                GRANT CONNECT ON DATABASE atepg TO ateapi_runtime, ateapi_ddl;
+                GRANT CREATE ON DATABASE atepg TO ateapi_ddl;
+                SQL
+                psql -v ON_ERROR_STOP=1 -U postgres -d atepg \
+                  -v substrate_schema="${ATEAPI_POSTGRES_SCHEMA}" <<'SQL'
+                SELECT format('REVOKE ALL ON SCHEMA %I FROM PUBLIC', :'substrate_schema')
+                  WHERE EXISTS (SELECT FROM pg_namespace WHERE nspname = :'substrate_schema') \gexec
+                SELECT format('ALTER SCHEMA %I OWNER TO ateapi_ddl', :'substrate_schema')
+                  WHERE EXISTS (SELECT FROM pg_namespace WHERE nspname = :'substrate_schema') \gexec
+                SELECT format('ALTER TABLE %I.%I OWNER TO ateapi_ddl', n.nspname, c.relname)
+                  FROM pg_class c
+                  JOIN pg_namespace n ON n.oid = c.relnamespace
+                  WHERE n.nspname = :'substrate_schema'
+                    AND c.relkind IN ('r', 'p')
+                    AND c.relowner <> (SELECT oid FROM pg_roles WHERE rolname = 'ateapi_ddl')
+                  \gexec
+                SELECT format('ALTER SEQUENCE %I.%I OWNER TO ateapi_ddl', n.nspname, c.relname)
+                  FROM pg_class c
+                  JOIN pg_namespace n ON n.oid = c.relnamespace
+                  WHERE n.nspname = :'substrate_schema'
+                    AND c.relkind = 'S'
+                    AND c.relowner <> (SELECT oid FROM pg_roles WHERE rolname = 'ateapi_ddl')
+                  \gexec
+                SQL
         env:
         - name: POSTGRES_DB
           value: atepg
@@ -165,6 +201,21 @@ spec:
           value: trust
         - name: PGDATA
           value: /var/lib/postgresql/data/pgdata
+        - name: ATEAPI_RUNTIME_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: postgres-role-passwords
+              key: runtime-password
+        - name: ATEAPI_DDL_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: postgres-role-passwords
+              key: ddl-password
+        - name: ATEAPI_POSTGRES_SCHEMA
+          valueFrom:
+            configMapKeyRef:
+              name: ate-api-server-envvars
+              key: ATE_API_POSTGRES_SCHEMA
         ports:
         - name: postgres
           containerPort: 5432
diff --git a/tools/setup-gcp/cloud-sql.md b/tools/setup-gcp/cloud-sql.md
index 6d680ab0f0..ccc27205ff 100644
--- a/tools/setup-gcp/cloud-sql.md
+++ b/tools/setup-gcp/cloud-sql.md
@@ -148,10 +148,20 @@ Optional environment variables:
   password yourself (the install script rejects `false` without an explicit
   connection string — a synthesized passwordless DSN cannot log in once the
   proxy stops injecting IAM tokens).
+- `ATE_API_POSTGRES_DDL_CONNECTION_STRING` — a separate schema-owner DSN for
+  migrations and outbox partition maintenance. Setting it requires an explicit
+  `ATE_API_POSTGRES_CONNECTION_STRING`; otherwise it defaults to the runtime DSN.
+  Operators that provision separate database identities should give the
+  runtime role only `CONNECT` and let ateapi apply its table grants. The
+  DDL role must own the Substrate schema and its objects, plus `CREATE` on the
+  database when ateapi needs to create that schema. Use a schema dedicated to
+  Substrate when configuring separate identities: runtime grants cover all
+  tables and sequences in that schema.
 - `ATE_API_POSTGRES_SCHEMA` — the schema holding the store's tables
-  (default `public`). If using a custom schema, the one-time schema grant
-  in §2 (`GRANT USAGE, CREATE ON SCHEMA ...`) must target your custom schema
-  instead of `public`.
+  (default `public`). A dedicated schema such as `substrate` is recommended
+  when using separate runtime and DDL roles. If using a custom schema, the
+  one-time schema grant in §2 (`GRANT USAGE, CREATE ON SCHEMA ...`) must target
+  your custom schema instead of `public`.
 - `ATE_API_POSTGRES_POOL_MAX_CONNS` — pgxpool connections per ateapi replica
   (default: `max(4, NumCPU)`); appended as `pool_max_conns` to whichever DSN
   is in effect (synthesized, in-cluster default, or explicitly provided —
@@ -203,6 +213,7 @@ server CA and credentials yourself.)
 
 ```sh
 export ATE_API_POSTGRES_CONNECTION_STRING='postgresql://<user>:<pw>@<host>:5432/atepg?sslmode=verify-ca&sslrootcert=/run/postgres-server-ca/server-ca.pem'
+export ATE_API_POSTGRES_DDL_CONNECTION_STRING='postgresql://<schema-owner>:<pw>@<host>:5432/atepg?sslmode=verify-ca&sslrootcert=/run/postgres-server-ca/server-ca.pem'
 export ATE_API_POSTGRES_SERVER_CA_FILE=/path/to/server-ca.pem
 ./hack/install-ate.sh --deploy-ate-system
 ```