db: refuse a pending migration older than the newest applied one

applyMigrations now fails closed, before any write, when a pending
migration has a version below the highest version recorded in
schema_migrations: migrations must be appended, never inserted.

db/commands.go

@@ -153,57 +153,13 @@ type LLMRecord struct {
 	RawResponse string `json:"raw_response,omitempty"`
 }
 
-const llmRecordsSchema = `
-CREATE TABLE IF NOT EXISTS llm_records (
-    id            BIGSERIAL PRIMARY KEY,
-    ts            TIMESTAMPTZ DEFAULT now(),
-    model         TEXT,
-    profile_name  TEXT,
-    session_id    TEXT,
-    task_id       TEXT,
-    worker        TEXT,
-    latency_ms    INTEGER,
-    input_tokens  INTEGER,
-    output_tokens INTEGER,
-    cache_read    INTEGER,
-    cache_write   INTEGER,
-    status        TEXT,
-    error         TEXT,
-    request_body  TEXT,
-    response_body TEXT,
-    raw_request   TEXT,
-    raw_response  TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_llm_records_ts ON llm_records(ts);
-CREATE INDEX IF NOT EXISTS idx_llm_records_session ON llm_records(session_id);
-`
-
-// llmRecordsMigrate adds new columns to existing tables.
-const llmRecordsMigrate = `
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS task_id TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS worker TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS profile_name TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_request TEXT;
-ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_response TEXT;
-`
-
-// EnsureLLMRecordsTable creates the llm_records table if it does not exist.
+// EnsureLLMRecordsTable creates the llm_records table if it does not exist, by
+// re-running the idempotent migration 0001_llm_records.sql. Open already applies
+// it through the migration runner; this stays for callers that want the table
+// guaranteed. It takes the schema lock in its own transaction, so it must not be
+// called while Open holds that lock.
 func (d *DB) EnsureLLMRecordsTable() error {
-	tx, err := d.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback() //nolint:errcheck
-	if err := coordinateWithSchemaMigration(tx); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmRecordsSchema); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
-		return err
-	}
-	return tx.Commit()
+	return d.execEmbeddedMigration(llmRecordsMigration)
 }
 
 // InsertLLMRecord stores one LLM call record.

db/db.go

@@ -1,6 +1,15 @@
 // Package db is the PostgreSQL data source for ARTEX (replacing the old graph
 // single-file SQLite). It opens the connection, applies the schema, and seeds
 // the built-in agents and variable catalog.
+//
+// Schema changes:
+//   - schema.sql remains the idempotent base, executed on every startup.
+//   - Any new DDL goes in a numbered migration, migrations/NNNN_slug.sql.
+//     Open applies pending migrations once, in version order, and records
+//     each in schema_migrations with the SHA-256 of its file.
+//   - A published migration is never edited: Open refuses to start when an
+//     applied migration's checksum no longer matches its file. Fix forward
+//     with a new migration instead.
 package db
 
 import (
@@ -9,6 +18,7 @@ import (
 	_ "embed"
 	"errors"
 	"fmt"
+	"io/fs"
 	"net/url"
 	"os"
 	"strings"
@@ -153,8 +163,15 @@ func ensureDatabase(dsn string) error {
 	return nil
 }
 
-// Open connects, applies the schema (idempotent), and seeds builtin rows.
+// Open connects, applies the schema (idempotent) and the embedded versioned
+// migrations, and seeds builtin rows.
 func Open(dsn string) (*DB, error) {
+	return openWithMigrations(dsn, migrationsFS())
+}
+
+// openWithMigrations is Open with the migration set as a parameter, so tests
+// can pass an fstest.MapFS.
+func openWithMigrations(dsn string, migrations fs.FS) (*DB, error) {
 	if err := ensureDatabase(dsn); err != nil {
 		return nil, err
 	}
@@ -187,6 +204,11 @@ func Open(dsn string) (*DB, error) {
 		if err := applySchemaWithRetry(context.Background(), conn, time.Sleep); err != nil {
 			return fmt.Errorf("apply schema: %w", err)
 		}
+		// Runs on conn, the lock holder; it must not call EnsureLLM*Table,
+		// whose own lock request would wait on conn forever.
+		if err := applyMigrations(context.Background(), conn, migrations); err != nil {
+			return fmt.Errorf("apply migrations: %w", err)
+		}
 		if err := d.seedBuiltins(); err != nil {
 			return fmt.Errorf("seed builtins: %w", err)
 		}

db/llm_records_migrate_test.go

@@ -1,11 +1,12 @@
 package db
 
 import (
+	"io/fs"
 	"testing"
 )
 
 // The raw-body columns were added after release, so existing installs only get
-// them through llmRecordsMigrate. This runs the real migration against the dev
+// them through the ALTERs in 0001_llm_records.sql. This runs the real migration against the dev
 // PG on a table stripped back to its pre-upgrade shape, inside a transaction
 // that is always rolled back — PG does transactional DDL, so nothing persists.
 func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
@@ -17,6 +18,11 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	if err := d.EnsureLLMRecordsTable(); err != nil {
 		t.Fatalf("ensure: %v", err)
 	}
+	content, err := fs.ReadFile(migrationsFS(), llmRecordsMigration)
+	if err != nil {
+		t.Fatal(err)
+	}
+	llmRecordsMigrateSQL := string(content)
 
 	tx, err := d.Begin()
 	if err != nil {
@@ -28,7 +34,7 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	if _, err := tx.Exec(`ALTER TABLE llm_records DROP COLUMN IF EXISTS raw_request, DROP COLUMN IF EXISTS raw_response`); err != nil {
 		t.Fatalf("simulate old table: %v", err)
 	}
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
+	if _, err := tx.Exec(llmRecordsMigrateSQL); err != nil {
 		t.Fatalf("migrate: %v", err)
 	}
 
@@ -45,7 +51,7 @@ func TestLLMRecordsMigrateAddsRawColumnsToOldTable(t *testing.T) {
 	}
 
 	// Re-running must stay a no-op (the migration runs on every startup).
-	if _, err := tx.Exec(llmRecordsMigrate); err != nil {
+	if _, err := tx.Exec(llmRecordsMigrateSQL); err != nil {
 		t.Fatalf("migrate is not idempotent: %v", err)
 	}
 

db/llm_usage.go

@@ -26,41 +26,11 @@ type LLMUsage struct {
 	Status        string `json:"status"` // ok | error
 }
 
-const llmUsageSchema = `
-CREATE TABLE IF NOT EXISTS llm_usage (
-    id             BIGSERIAL PRIMARY KEY,
-    ts             TIMESTAMPTZ NOT NULL DEFAULT now(),
-    task_id        TEXT,
-    exploration_id BIGINT,
-    worker         TEXT,
-    model          TEXT,
-    profile_name   TEXT,
-    latency_ms     INTEGER,
-    input_tokens   INTEGER NOT NULL DEFAULT 0,
-    output_tokens  INTEGER NOT NULL DEFAULT 0,
-    cache_read     INTEGER NOT NULL DEFAULT 0,
-    cache_write    INTEGER NOT NULL DEFAULT 0,
-    status         TEXT
-);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_task  ON llm_usage(task_id);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_model ON llm_usage(task_id, model);
-CREATE INDEX IF NOT EXISTS idx_llm_usage_exp   ON llm_usage(exploration_id);
-`
-
-// EnsureLLMUsageTable creates the llm_usage metering table if it does not exist.
+// EnsureLLMUsageTable creates the llm_usage metering table if it does not exist,
+// by re-running the idempotent migration 0002_llm_usage.sql (see
+// EnsureLLMRecordsTable for the locking rule).
 func (d *DB) EnsureLLMUsageTable() error {
-	tx, err := d.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback() //nolint:errcheck
-	if err := coordinateWithSchemaMigration(tx); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(llmUsageSchema); err != nil {
-		return err
-	}
-	return tx.Commit()
+	return d.execEmbeddedMigration(llmUsageMigration)
 }
 
 // InsertLLMUsage appends one metering row. Best-effort: callers log and continue on

db/migrate.go

@@ -0,0 +1,217 @@
+package db
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"embed"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"path"
+	"regexp"
+	"sort"
+	"strconv"
+)
+
+// Versioned migrations, applied by Open after schema.sql (rules in the package
+// comment, db.go).
+//
+//go:embed migrations/*.sql
+var embeddedMigrationFiles embed.FS
+
+// Embedded migration files that EnsureLLMRecordsTable / EnsureLLMUsageTable
+// re-run on demand.
+const (
+	llmRecordsMigration = "0001_llm_records.sql"
+	llmUsageMigration   = "0002_llm_usage.sql"
+)
+
+// migrationsFS is the embedded migrations directory, rooted so file names are
+// the bare NNNN_slug.sql names.
+func migrationsFS() fs.FS {
+	sub, err := fs.Sub(embeddedMigrationFiles, "migrations")
+	if err != nil {
+		panic(err) // the embed pattern guarantees the directory exists
+	}
+	return sub
+}
+
+const schemaMigrationsTable = `
+CREATE TABLE IF NOT EXISTS schema_migrations (
+    version    INTEGER PRIMARY KEY,
+    name       TEXT,
+    checksum   TEXT,
+    applied_at TIMESTAMPTZ DEFAULT now()
+);
+`
+
+var migrationNameRE = regexp.MustCompile(`^([0-9]{4})_[a-z0-9_]+\.sql$`)
+
+type migration struct {
+	version  int
+	name     string // file name, e.g. 0001_llm_records.sql
+	sql      string
+	checksum string // hex SHA-256 of the file content
+}
+
+// loadMigrations reads every *.sql file at the root of fsys, sorted by version.
+// A .sql file that does not follow NNNN_slug.sql, or a duplicated version, is
+// an error so a typo cannot silently skip a migration.
+func loadMigrations(fsys fs.FS) ([]migration, error) {
+	entries, err := fs.ReadDir(fsys, ".")
+	if err != nil {
+		return nil, fmt.Errorf("read migrations: %w", err)
+	}
+	var out []migration
+	seen := map[int]string{}
+	for _, e := range entries {
+		if e.IsDir() || path.Ext(e.Name()) != ".sql" {
+			continue
+		}
+		m := migrationNameRE.FindStringSubmatch(e.Name())
+		if m == nil {
+			return nil, fmt.Errorf("migration file %s: name must be NNNN_slug.sql", e.Name())
+		}
+		version, _ := strconv.Atoi(m[1])
+		if version == 0 {
+			return nil, fmt.Errorf("migration file %s: version must be > 0", e.Name())
+		}
+		if prev, dup := seen[version]; dup {
+			return nil, fmt.Errorf("migration version %04d is used by both %s and %s", version, prev, e.Name())
+		}
+		seen[version] = e.Name()
+		content, err := fs.ReadFile(fsys, e.Name())
+		if err != nil {
+			return nil, fmt.Errorf("read migration %s: %w", e.Name(), err)
+		}
+		sum := sha256.Sum256(content)
+		out = append(out, migration{version: version, name: e.Name(), sql: string(content), checksum: hex.EncodeToString(sum[:])})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].version < out[j].version })
+	return out, nil
+}
+
+// applyMigrations brings the database up to the migrations in fsys. conn must
+// be the connection that holds the schema advisory lock (see Open); every
+// statement runs on it, so nothing here waits on that lock again.
+//
+// It fails closed before applying anything when the database records a version
+// this binary does not ship, or an applied migration whose name or checksum
+// differs from the file, or a pending migration is older than the newest
+// applied one. Each pending migration then runs in its own
+// transaction together with its schema_migrations row; a failure rolls that one
+// back and stops, leaving earlier migrations applied.
+func applyMigrations(ctx context.Context, conn *sql.Conn, fsys fs.FS) error {
+	migrations, err := loadMigrations(fsys)
+	if err != nil {
+		return err
+	}
+	if _, err := conn.ExecContext(ctx, schemaMigrationsTable); err != nil {
+		return fmt.Errorf("create schema_migrations: %w", err)
+	}
+
+	type appliedRow struct{ name, checksum string }
+	applied := map[int]appliedRow{}
+	rows, err := conn.QueryContext(ctx, `SELECT version, COALESCE(name,''), COALESCE(checksum,'') FROM schema_migrations ORDER BY version`)
+	if err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+	var appliedVersions []int
+	for rows.Next() {
+		var v int
+		var r appliedRow
+		if err := rows.Scan(&v, &r.name, &r.checksum); err != nil {
+			rows.Close()
+			return fmt.Errorf("read schema_migrations: %w", err)
+		}
+		applied[v] = r
+		appliedVersions = append(appliedVersions, v)
+	}
+	if err := rows.Close(); err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("read schema_migrations: %w", err)
+	}
+
+	byVersion := map[int]migration{}
+	for _, m := range migrations {
+		byVersion[m.version] = m
+	}
+	for _, v := range appliedVersions {
+		r := applied[v]
+		m, ok := byVersion[v]
+		if !ok {
+			return fmt.Errorf("migration %04d (%s) is applied in the database but missing from this binary", v, r.name)
+		}
+		if r.name != m.name {
+			return fmt.Errorf("migration %04d: database records file %s but this binary has %s", v, r.name, m.name)
+		}
+		if r.checksum != m.checksum {
+			return fmt.Errorf("migration %04d (%s): checksum changed since it was applied (database %s, file %s); published migrations must not be edited", v, m.name, r.checksum, m.checksum)
+		}
+	}
+
+	// A pending version below the newest applied one would run out of order on
+	// this database: refuse it before writing anything.
+	if len(appliedVersions) > 0 {
+		maxVersion := appliedVersions[len(appliedVersions)-1]
+		for _, m := range migrations {
+			if _, done := applied[m.version]; !done && m.version < maxVersion {
+				return fmt.Errorf("migration %s (%04d) is older than the applied migration %s (%04d); migrations must be appended, never inserted",
+					m.name, m.version, applied[maxVersion].name, maxVersion)
+			}
+		}
+	}
+
+	for _, m := range migrations {
+		if _, done := applied[m.version]; done {
+			continue
+		}
+		if err := applyMigration(ctx, conn, m); err != nil {
+			return fmt.Errorf("migration %04d (%s): %w", m.version, m.name, err)
+		}
+	}
+	return nil
+}
+
+func applyMigration(ctx context.Context, conn *sql.Conn, m migration) error {
+	tx, err := conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck // no-op after Commit
+	// No args: pgx sends it over the simple protocol, which allows several statements.
+	if _, err := tx.ExecContext(ctx, m.sql); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migrations(version, name, checksum) VALUES ($1, $2, $3)`, m.version, m.name, m.checksum); err != nil {
+		return fmt.Errorf("record: %w", err)
+	}
+	return tx.Commit()
+}
+
+// execEmbeddedMigration re-runs one embedded (idempotent) migration file in its
+// own transaction under the schema lock. It does not touch schema_migrations.
+// The lock is taken on a pooled connection, so this must never be called from
+// inside Open's withSchemaMigrationLock callback: it would wait on the lock
+// that callback's connection holds.
+func (d *DB) execEmbeddedMigration(name string) error {
+	content, err := fs.ReadFile(migrationsFS(), name)
+	if err != nil {
+		return err
+	}
+	tx, err := d.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck
+	if err := coordinateWithSchemaMigration(tx); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(string(content)); err != nil {
+		return err
+	}
+	return tx.Commit()
+}

db/migrate_test.go

@@ -0,0 +1,415 @@
+package db
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"io/fs"
+	"net/url"
+	"strings"
+	"testing"
+	"testing/fstest"
+	"time"
+)
+
+func TestLoadMigrationsValidation(t *testing.T) {
+	ms, err := loadMigrations(fstest.MapFS{
+		"0010_b.sql": {Data: []byte("SELECT 10;")},
+		"0002_a.sql": {Data: []byte("SELECT 2;")},
+		"README.md":  {Data: []byte("not a migration")},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(ms) != 2 || ms[0].version != 2 || ms[1].version != 10 {
+		t.Fatalf("want versions [2 10], got %+v", ms)
+	}
+	sum := sha256.Sum256([]byte("SELECT 2;"))
+	if ms[0].checksum != hex.EncodeToString(sum[:]) || ms[0].name != "0002_a.sql" {
+		t.Fatalf("migration 0002 = %+v", ms[0])
+	}
+
+	for name, fsys := range map[string]fstest.MapFS{
+		"bad name":          {"2_a.sql": {Data: []byte("SELECT 1;")}},
+		"duplicate version": {"0001_a.sql": {Data: []byte("SELECT 1;")}, "0001_b.sql": {Data: []byte("SELECT 1;")}},
+	} {
+		if _, err := loadMigrations(fsys); err == nil {
+			t.Errorf("%s: want error", name)
+		}
+	}
+}
+
+// scratchDSN returns the DSN of a new, empty database named after the test
+// database plus suffix, dropped again when the test ends.
+func scratchDSN(t *testing.T, suffix string) string {
+	t.Helper()
+	u, err := url.Parse(testDSN(t))
+	if err != nil || strings.Trim(u.Path, "/") == "" {
+		t.Skipf("DSN has no database name — skipping")
+	}
+	name := strings.TrimPrefix(u.Path, "/") + "_mig_" + suffix
+	admin := *u
+	admin.Path = "/postgres"
+	conn, err := sql.Open("pgx", admin.String())
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	if err := conn.Ping(); err != nil {
+		conn.Close()
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	drop := func() error {
+		_, err := conn.Exec(`DROP DATABASE IF EXISTS "` + name + `" WITH (FORCE)`)
+		return err
+	}
+	if err := drop(); err != nil {
+		t.Fatalf("drop leftover %s: %v", name, err)
+	}
+	t.Cleanup(func() {
+		if err := drop(); err != nil {
+			t.Errorf("drop %s: %v", name, err)
+		}
+		conn.Close()
+	})
+	u.Path = "/" + name
+	return u.String()
+}
+
+// withEmbedded returns the embedded migrations plus extra files.
+func withEmbedded(t *testing.T, extra fstest.MapFS) fstest.MapFS {
+	t.Helper()
+	out := fstest.MapFS{}
+	for _, name := range []string{llmRecordsMigration, llmUsageMigration} {
+		content, err := fs.ReadFile(migrationsFS(), name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		out[name] = &fstest.MapFile{Data: content}
+	}
+	for name, f := range extra {
+		out[name] = f
+	}
+	return out
+}
+
+type appliedMigration struct {
+	version        int
+	name, checksum string
+	appliedAt      time.Time
+}
+
+func readApplied(t *testing.T, d *DB) []appliedMigration {
+	t.Helper()
+	rows, err := d.Query(`SELECT version, name, checksum, applied_at FROM schema_migrations ORDER BY version`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var out []appliedMigration
+	for rows.Next() {
+		var m appliedMigration
+		if err := rows.Scan(&m.version, &m.name, &m.checksum, &m.appliedAt); err != nil {
+			t.Fatal(err)
+		}
+		out = append(out, m)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+
+// assertEmbeddedApplied checks that both LLM tables exist and that 0001 and
+// 0002 are recorded with the SHA-256 of their embedded files.
+func assertEmbeddedApplied(t *testing.T, d *DB) []appliedMigration {
+	t.Helper()
+	for _, table := range []string{"llm_records", "llm_usage"} {
+		var exists bool
+		if err := d.QueryRow(`SELECT to_regclass($1) IS NOT NULL`, table).Scan(&exists); err != nil {
+			t.Fatal(err)
+		}
+		if !exists {
+			t.Errorf("table %s missing after Open", table)
+		}
+	}
+	applied := readApplied(t, d)
+	want := []struct {
+		version int
+		name    string
+	}{{1, llmRecordsMigration}, {2, llmUsageMigration}}
+	if len(applied) != len(want) {
+		t.Fatalf("schema_migrations = %+v, want versions 1 and 2", applied)
+	}
+	for i, w := range want {
+		content, err := fs.ReadFile(migrationsFS(), w.name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		sum := sha256.Sum256(content)
+		got := applied[i]
+		if got.version != w.version || got.name != w.name || got.checksum != hex.EncodeToString(sum[:]) {
+			t.Errorf("schema_migrations row %d = %+v, want version %d name %s checksum %x", i, got, w.version, w.name, sum)
+		}
+	}
+	return applied
+}
+
+// assertReopenIsNoop opens dsn again and checks that no migration row was added
+// or re-applied, then that both Ensure functions stay idempotent.
+func assertReopenIsNoop(t *testing.T, dsn string, before []appliedMigration) {
+	t.Helper()
+	d, err := Open(dsn)
+	if err != nil {
+		t.Fatalf("second Open: %v", err)
+	}
+	defer d.Close()
+	after := assertEmbeddedApplied(t, d)
+	for i := range before {
+		if !after[i].appliedAt.Equal(before[i].appliedAt) {
+			t.Errorf("migration %04d applied_at changed: %v -> %v", before[i].version, before[i].appliedAt, after[i].appliedAt)
+		}
+	}
+	for range 2 {
+		if err := d.EnsureLLMRecordsTable(); err != nil {
+			t.Fatalf("EnsureLLMRecordsTable after Open: %v", err)
+		}
+		if err := d.EnsureLLMUsageTable(); err != nil {
+			t.Fatalf("EnsureLLMUsageTable after Open: %v", err)
+		}
+	}
+	if got := readApplied(t, d); len(got) != len(before) {
+		t.Errorf("Ensure functions changed schema_migrations: %+v", got)
+	}
+}
+
+// TestOpenRecordsEmbeddedMigrations runs against the package test database:
+// fresh on a first run, already migrated on a second run.
+func TestOpenRecordsEmbeddedMigrations(t *testing.T) {
+	dsn := testDSN(t)
+	d, err := Open(dsn)
+	if err != nil {
+		t.Skipf("postgres unavailable (%v) — skipping", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+func TestOpenMigratesFreshDatabase(t *testing.T) {
+	dsn := scratchDSN(t, "fresh")
+	d, err := Open(dsn)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+// An install that predates schema_migrations already has both tables (created
+// by the Ensure functions, possibly without the later llm_records columns).
+// Open must adopt them: run 0001/0002 over them and record both.
+func TestOpenAdoptsExistingTablesWithoutSchemaMigrations(t *testing.T) {
+	dsn := scratchDSN(t, "legacy")
+	d, err := openWithMigrations(dsn, fstest.MapFS{})
+	if err != nil {
+		t.Fatalf("Open without migrations: %v", err)
+	}
+	if err := d.EnsureLLMRecordsTable(); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.EnsureLLMUsageTable(); err != nil {
+		t.Fatal(err)
+	}
+	for _, q := range []string{
+		`ALTER TABLE llm_records DROP COLUMN raw_request, DROP COLUMN raw_response`,
+		`INSERT INTO llm_records(model) VALUES ('legacy')`,
+		`INSERT INTO llm_usage(model, status) VALUES ('legacy', 'ok')`,
+		`DROP TABLE schema_migrations`,
+	} {
+		if _, err := d.Exec(q); err != nil {
+			t.Fatalf("%s: %v", q, err)
+		}
+	}
+	d.Close()
+
+	d, err = Open(dsn)
+	if err != nil {
+		t.Fatalf("Open on existing tables: %v", err)
+	}
+	applied := assertEmbeddedApplied(t, d)
+	var records, usage, rawCols int
+	if err := d.QueryRow(`SELECT count(*) FROM llm_records WHERE model='legacy'`).Scan(&records); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.QueryRow(`SELECT count(*) FROM llm_usage WHERE model='legacy'`).Scan(&usage); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.QueryRow(`SELECT count(*) FROM information_schema.columns
+		WHERE table_name='llm_records' AND column_name IN ('raw_request','raw_response')`).Scan(&rawCols); err != nil {
+		t.Fatal(err)
+	}
+	if records != 1 || usage != 1 || rawCols != 2 {
+		t.Errorf("after adopt: llm_records legacy rows=%d llm_usage legacy rows=%d raw columns=%d, want 1 1 2", records, usage, rawCols)
+	}
+	d.Close()
+	assertReopenIsNoop(t, dsn, applied)
+}
+
+// probeMigration records how many migrations were already applied when it
+// ran, and fails if it ever runs twice (CREATE TABLE without IF NOT EXISTS).
+var probeMigration = &fstest.MapFile{Data: []byte(`
+CREATE TABLE migration_probe (applied_before INTEGER);
+INSERT INTO migration_probe SELECT count(*) FROM schema_migrations;
+`)}
+
+func TestMigrationsApplyNewVersionOnceInOrder(t *testing.T) {
+	dsn := scratchDSN(t, "order")
+	fsys := withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration})
+	for pass := 1; pass <= 2; pass++ {
+		d, err := openWithMigrations(dsn, fsys)
+		if err != nil {
+			t.Fatalf("pass %d: %v", pass, err)
+		}
+		applied := readApplied(t, d)
+		if len(applied) != 3 || applied[2].version != 3 || applied[2].name != "0003_probe.sql" {
+			t.Fatalf("pass %d: schema_migrations = %+v", pass, applied)
+		}
+		var runs, before int
+		if err := d.QueryRow(`SELECT count(*), max(applied_before) FROM migration_probe`).Scan(&runs, &before); err != nil {
+			t.Fatal(err)
+		}
+		if runs != 1 || before != 2 {
+			t.Fatalf("pass %d: probe ran %d times with %d migrations before it, want 1 and 2", pass, runs, before)
+		}
+		d.Close()
+	}
+}
+
+func TestMigrationsFailClosed(t *testing.T) {
+	dsn := scratchDSN(t, "closed")
+	d, err := openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	d.Close()
+
+	cases := []struct {
+		name string
+		fsys fs.FS
+		want []string
+	}{
+		{"checksum changed", withEmbedded(t, fstest.MapFS{"0003_probe.sql": {Data: append([]byte("-- edited\n"), probeMigration.Data...)}}),
+			[]string{"0003", "0003_probe.sql", "checksum"}},
+		{"version missing from binary", migrationsFS(),
+			[]string{"0003", "0003_probe.sql", "missing from this binary"}},
+		{"name changed", withEmbedded(t, fstest.MapFS{"0003_renamed.sql": probeMigration}),
+			[]string{"0003", "0003_probe.sql", "0003_renamed.sql"}},
+	}
+	for _, c := range cases {
+		d, err := openWithMigrations(dsn, c.fsys)
+		if err == nil {
+			d.Close()
+			t.Errorf("%s: Open succeeded, want error", c.name)
+			continue
+		}
+		for _, w := range c.want {
+			if !strings.Contains(err.Error(), w) {
+				t.Errorf("%s: error %q does not mention %q", c.name, err, w)
+			}
+		}
+	}
+}
+
+func TestFailingMigrationRollsBackAndBlocksOpen(t *testing.T) {
+	dsn := scratchDSN(t, "fail")
+	fsys := withEmbedded(t, fstest.MapFS{
+		"0003_probe.sql": probeMigration,
+		"0004_broken.sql": {Data: []byte(`
+CREATE TABLE migration_broken (x INTEGER);
+SELECT 1/0;
+`)},
+	})
+	d, err := openWithMigrations(dsn, fsys)
+	if err == nil {
+		d.Close()
+		t.Fatal("Open succeeded with a failing migration")
+	}
+	if !strings.Contains(err.Error(), "0004") || !strings.Contains(err.Error(), "0004_broken.sql") {
+		t.Errorf("error %q does not name the failing migration", err)
+	}
+
+	// Inspect with the migrations that succeeded: earlier ones stay applied,
+	// the failing one left nothing behind.
+	d, err = openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	applied := readApplied(t, d)
+	if len(applied) != 3 || applied[2].version != 3 {
+		t.Fatalf("schema_migrations = %+v, want versions 1..3", applied)
+	}
+	var broken bool
+	if err := d.QueryRow(`SELECT to_regclass('migration_broken') IS NOT NULL`).Scan(&broken); err != nil {
+		t.Fatal(err)
+	}
+	if broken {
+		t.Error("migration_broken exists: the failed migration was not rolled back")
+	}
+}
+
+// A database already at 0003 must refuse a binary that adds 0002 below it,
+// before writing anything.
+func TestMigrationsRefuseOutOfOrderVersion(t *testing.T) {
+	dsn := scratchDSN(t, "outoforder")
+	records, err := fs.ReadFile(migrationsFS(), llmRecordsMigration)
+	if err != nil {
+		t.Fatal(err)
+	}
+	d, err := openWithMigrations(dsn, fstest.MapFS{
+		llmRecordsMigration: {Data: records},
+		"0003_probe.sql":    probeMigration,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	before := readApplied(t, d)
+	d.Close()
+	if len(before) != 2 || before[0].version != 1 || before[1].version != 3 {
+		t.Fatalf("setup schema_migrations = %+v, want versions 1 and 3", before)
+	}
+
+	d, err = openWithMigrations(dsn, withEmbedded(t, fstest.MapFS{"0003_probe.sql": probeMigration}))
+	if err == nil {
+		d.Close()
+		t.Fatal("Open succeeded with 0002 pending below applied 0003")
+	}
+	for _, w := range []string{llmUsageMigration, "(0002)", "0003_probe.sql", "(0003)", "never inserted"} {
+		if !strings.Contains(err.Error(), w) {
+			t.Errorf("error %q does not mention %q", err, w)
+		}
+	}
+
+	raw, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer raw.Close()
+	after := readApplied(t, &DB{raw})
+	if len(after) != len(before) {
+		t.Fatalf("schema_migrations changed: before %+v, after %+v", before, after)
+	}
+	for i := range before {
+		if after[i].version != before[i].version || after[i].checksum != before[i].checksum || !after[i].appliedAt.Equal(before[i].appliedAt) {
+			t.Errorf("schema_migrations row %d changed: before %+v, after %+v", i, before[i], after[i])
+		}
+	}
+	var usage bool
+	if err := raw.QueryRow(`SELECT to_regclass('llm_usage') IS NOT NULL`).Scan(&usage); err != nil {
+		t.Fatal(err)
+	}
+	if usage {
+		t.Error("llm_usage exists: the refused 0002 migration ran")
+	}
+}

db/migrations/0001_llm_records.sql

@@ -0,0 +1,29 @@
+
+CREATE TABLE IF NOT EXISTS llm_records (
+    id            BIGSERIAL PRIMARY KEY,
+    ts            TIMESTAMPTZ DEFAULT now(),
+    model         TEXT,
+    profile_name  TEXT,
+    session_id    TEXT,
+    task_id       TEXT,
+    worker        TEXT,
+    latency_ms    INTEGER,
+    input_tokens  INTEGER,
+    output_tokens INTEGER,
+    cache_read    INTEGER,
+    cache_write   INTEGER,
+    status        TEXT,
+    error         TEXT,
+    request_body  TEXT,
+    response_body TEXT,
+    raw_request   TEXT,
+    raw_response  TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_llm_records_ts ON llm_records(ts);
+CREATE INDEX IF NOT EXISTS idx_llm_records_session ON llm_records(session_id);
+
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS task_id TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS worker TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS profile_name TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_request TEXT;
+ALTER TABLE llm_records ADD COLUMN IF NOT EXISTS raw_response TEXT;

db/migrations/0002_llm_usage.sql

@@ -0,0 +1,19 @@
+
+CREATE TABLE IF NOT EXISTS llm_usage (
+    id             BIGSERIAL PRIMARY KEY,
+    ts             TIMESTAMPTZ NOT NULL DEFAULT now(),
+    task_id        TEXT,
+    exploration_id BIGINT,
+    worker         TEXT,
+    model          TEXT,
+    profile_name   TEXT,
+    latency_ms     INTEGER,
+    input_tokens   INTEGER NOT NULL DEFAULT 0,
+    output_tokens  INTEGER NOT NULL DEFAULT 0,
+    cache_read     INTEGER NOT NULL DEFAULT 0,
+    cache_write    INTEGER NOT NULL DEFAULT 0,
+    status         TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_task  ON llm_usage(task_id);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_model ON llm_usage(task_id, model);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_exp   ON llm_usage(exploration_id);