fusion/packages/core/src/postgres/schema-applier.ts

/**
 * PostgreSQL schema applier.
 *
 * FNXC:PostgresSchema 2026-06-24-03:40:
 * Applies the fresh Drizzle migration baseline to a PostgreSQL connection
 * and records it in a migration bookkeeping table. The baseline migration
 * (migrations/0000_initial.sql) is the snapshot of the final SQLite schema
 * (SCHEMA_VERSION=128) translated to PostgreSQL — applying it to an empty
 * database yields final-schema parity (VAL-SCHEMA-001).
 *
 * After the baseline lands, plugin-owned tables are materialized via the
 * schema-init hook (VAL-SCHEMA-007). The applier calls each registered plugin
 * hook so plugins evolve their own tables independently of the core migration.
 *
 * Migration tracking uses a single-row bookkeeping table in the public schema
 * so the applier is idempotent: re-running against an already-migrated database
 * is a no-op. The version-gate discipline (the institutional learning that
 * fresh-DB tests cannot catch a skipped-on-upgrade migration) is carried
 * forward via the applier's explicit baseline marker.
 */

import { readFile } from "node:fs/promises";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import type { PostgresJsDatabase } from "drizzle-orm/postgres-js";
import { sql } from "drizzle-orm";
import { runPluginSchemaInitHooks, DEFAULT_PLUGIN_SCHEMA_INIT_HOOKS, type PluginSchemaInitHook } from "./plugin-schema-hook.js";

/** The latest PostgreSQL schema version known to this applier. */
/*
FNXC:MultiProjectIsolation 2026-07-15-23:40:
Advances to 0012 after the owner_project_id domain/partition split and chat pin timestamp. Per-migration identities above stay fixed; only this latest-version marker moves.
*/
export const SCHEMA_BASELINE_VERSION = "0017";
const INITIAL_SCHEMA_VERSION = "0000";
const AUTOMATION_ISOLATION_SCHEMA_VERSION = "0001";
const ANALYTICS_ISOLATION_SCHEMA_VERSION = "0002";
/**
 * FNXC:PostgresMigrationIdentity 2026-07-14-01:41:
 * Each migration keeps an immutable bookkeeping identity even as SCHEMA_BASELINE_VERSION advances to newer migrations. Upgrade checks and inserts must use this dedicated 0003 identifier so a later latest-version marker cannot make an unrecorded monitor/approval migration look applied.
 */
export const MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION = "0003";
export const LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION = "0004";
export const MULTI_PROJECT_CUTOVER_SCHEMA_VERSION = "0005";
export const PROJECT_OWNERSHIP_SCHEMA_VERSION = "0006";
export const SQLITE_SCHEMA_PARITY_VERSION = "0007";
/**
 * FNXC:PlannerOversight 2026-07-14-18:49:
 * Version 0008 adds project.tasks.session_advisor_enabled for per-task session
 * advisor overrides. Keep this identity fixed when SCHEMA_BASELINE_VERSION advances.
 */
export const SESSION_ADVISOR_ENABLED_SCHEMA_VERSION = "0008";
export const MISSION_FIX_IDEMPOTENCY_VERSION = "0009";
/*
FNXC:GitHubImportTranslate 2026-07-15-09:30:
Import-translation cache advances to 0010. Migrations are registered here explicitly (not auto-discovered from the migrations dir), so a new .sql file that is not wired through a version constant + bookkeeping check silently never runs.
*/
export const IMPORT_TRANSLATION_CACHE_VERSION = "0010";
/**
 * FNXC:GitHubImportTranslate 2026-07-16-23:30:
 * Existing databases already recorded 0010, so the cache scope correction is
 * deliberately a new forward migration rather than a retroactive SQL edit.
 */
export const IMPORT_TRANSLATION_CACHE_SCOPE_FIX_VERSION = "0016";
/*
FNXC:MultiProjectIsolation 2026-07-15-23:40:
Version 0011 splits the domain "project" field from the RLS partition on the tables
that conflated them: `project_id` stays the trigger/GUC-owned isolation partition,
`owner_project_id` becomes the caller-supplied domain field. Writing domain values
into the partition put parent rows and child rows in different partitions and broke
the composite FKs (SQLSTATE 23503). Keep this identity fixed when
SCHEMA_BASELINE_VERSION advances.
*/
export const OWNER_PROJECT_ID_SPLIT_VERSION = "0011";
/*
FNXC:ChatPinned 2026-07-16-12:30:
Version 0012 makes the persisted pin timestamp available on databases that
already applied the baseline before Direct conversations can be pinned.
*/
export const CHAT_SESSION_PINS_VERSION = "0012";
/** FNXC:ExecutorToolFailureRetry 2026-07-16-12:00: upgrades existing PostgreSQL task rows before retry-state reads. */
export const EXECUTOR_TOOL_FAILURE_RETRY_VERSION = "0013";
/** FNXC:ExecutorEscalation 2026-07-16-21:00: Existing clusters need the durable single-shot latch before executor reads it during post-FN-7996 escalation. */
export const EXECUTOR_ESCALATION_ATTEMPT_VERSION = "0014";
/** FNXC:PostgresSchema 2026-07-16-22:00: central global routines follow main's already-landed 0014 migration. */
export const GLOBAL_ROUTINES_SCHEMA_VERSION = "0015";
/** FNXC:Settings-MergerModel 2026-07-16-12:00: per-task merger lane is an additive upgrade. */
export const TASK_MERGER_MODEL_LANE_VERSION = "0017";

/** Bookkeeping table for the fresh Drizzle migration history. */
export const MIGRATION_BOOKKEEPING_TABLE = "fusion_schema_migrations";

const __dirname = dirname(fileURLToPath(import.meta.url));
const BASELINE_MIGRATION_PATH = join(__dirname, "migrations", "0000_initial.sql");
const AUTOMATION_ISOLATION_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0001_automation_project_isolation.sql",
);
const ANALYTICS_ISOLATION_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0002_analytics_project_isolation.sql",
);
const MONITOR_APPROVAL_ISOLATION_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0003_monitor_approval_project_isolation.sql",
);
const LEGACY_CUTOVER_PRESERVATION_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0004_legacy_cutover_preservation.sql",
);
const MULTI_PROJECT_CUTOVER_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0005_multi_project_cutover.sql",
);
const PROJECT_OWNERSHIP_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0006_project_ownership.sql",
);
const SQLITE_SCHEMA_PARITY_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0007_sqlite_schema_parity.sql",
);
const SESSION_ADVISOR_ENABLED_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0008_session_advisor_enabled.sql",
);
const MISSION_FIX_IDEMPOTENCY_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0009_mission_fix_idempotency.sql",
);
const IMPORT_TRANSLATION_CACHE_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0010_import_translation_cache.sql",
);
const IMPORT_TRANSLATION_CACHE_SCOPE_FIX_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0016_import_translation_cache_scope_fix.sql",
);
const OWNER_PROJECT_ID_SPLIT_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0011_owner_project_id.sql",
);
const CHAT_SESSION_PINS_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0012_chat_session_pins.sql",
);
const EXECUTOR_TOOL_FAILURE_RETRY_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0013_executor_tool_failure_retry.sql",
);
const EXECUTOR_ESCALATION_ATTEMPT_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0014_executor_escalation_attempt.sql",
);
const GLOBAL_ROUTINES_MIGRATION_PATH = join(
  __dirname,
  "migrations",
  "0015_global_routines.sql",
);
const TASK_MERGER_MODEL_LANE_MIGRATION_PATH = join(__dirname, "migrations", "0017_task_merger_model_lane.sql");

/**
 * Ensure the migration bookkeeping table exists. Lives in the public schema so
 * it survives across the three application schemas and is queryable without
 * search_path qualification.
 */
async function ensureBookkeepingTable(db: PostgresJsDatabase<Record<string, never>>): Promise<void> {
  await db.execute(sql.raw(`
    CREATE TABLE IF NOT EXISTS public.${MIGRATION_BOOKKEEPING_TABLE} (
      version text PRIMARY KEY,
      applied_at timestamptz NOT NULL DEFAULT now()
    )
  `));
}

/** Read the baseline migration SQL from disk. Exported for tests. */
export async function readBaselineMigrationSql(): Promise<string> {
  return readFile(BASELINE_MIGRATION_PATH, "utf8");
}

/** Return the set of already-applied migration versions, or empty if none. */
export async function getAppliedMigrations(
  db: PostgresJsDatabase<Record<string, never>>,
): Promise<string[]> {
  await ensureBookkeepingTable(db);
  const rows = (await db.execute(
    sql`SELECT version FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} ORDER BY version`,
  )) as unknown as Array<{ version: string }>;
  return rows.map((row) => row.version);
}

/**
 * Apply the fresh baseline migration to the given connection.
 *
 * Idempotent: if the baseline version is already recorded, this is a no-op.
 * After the baseline lands, all registered plugin schema-init hooks run so
 * plugin-owned tables (e.g. roadmap) materialize (VAL-SCHEMA-007).
 *
 * The baseline SQL is applied as a single batch via postgres.js's file/unsafe
 * execution path. It uses CREATE TABLE IF NOT EXISTS and CREATE INDEX IF NOT
 * EXISTS throughout, so a partial prior apply is safe to resume.
 */
export async function applySchemaBaseline(
  db: PostgresJsDatabase<Record<string, never>>,
  options: { pluginHooks?: readonly PluginSchemaInitHook[] } = {},
): Promise<{ applied: boolean; pluginHooksRun: number }> {
  /*
   * FNXC:PostgresSchema 2026-07-14-00:05:
   * Schema versions are a cluster-wide invariant. Serialize version discovery,
   * DDL, and bookkeeping in one transaction so concurrent Fusion processes
   * cannot both apply a version or race its primary-key marker.
  */
  return db.transaction(async (tx) => {
    await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:schema-applier'))`);
    await ensureBookkeepingTable(tx);
    /*
    FNXC:PostgresSchema 2026-07-16-00:55:
    FN-8051 requires project, central, and archive to exist before plugin schema-init hooks run.
    Hooks run even when migration markers are already recorded and target project tables, so
    ensure the namespaces unconditionally inside the advisory-locked transaction rather than
    relying on the baseline batch that a marker-present database skips.
    */
    await tx.execute(sql.raw(`
      CREATE SCHEMA IF NOT EXISTS project;
      CREATE SCHEMA IF NOT EXISTS central;
      CREATE SCHEMA IF NOT EXISTS archive;
    `));
    const applied = await getAppliedMigrations(tx);
    const baselineAlreadyApplied = applied.includes(INITIAL_SCHEMA_VERSION);
    const automationIsolationAlreadyApplied = applied.includes(AUTOMATION_ISOLATION_SCHEMA_VERSION);
    const analyticsIsolationAlreadyApplied = applied.includes(ANALYTICS_ISOLATION_SCHEMA_VERSION);
    const monitorApprovalIsolationAlreadyApplied = applied.includes(MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION);
    const legacyCutoverPreservationAlreadyApplied = applied.includes(LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION);
    const multiProjectCutoverAlreadyApplied = applied.includes(MULTI_PROJECT_CUTOVER_SCHEMA_VERSION);
    const projectOwnershipAlreadyApplied = applied.includes(PROJECT_OWNERSHIP_SCHEMA_VERSION);
    const sqliteSchemaParityAlreadyApplied = applied.includes(SQLITE_SCHEMA_PARITY_VERSION);
    const sessionAdvisorEnabledAlreadyApplied = applied.includes(SESSION_ADVISOR_ENABLED_SCHEMA_VERSION);
    const missionFixIdempotencyAlreadyApplied = applied.includes(MISSION_FIX_IDEMPOTENCY_VERSION);
    const importTranslationCacheAlreadyApplied = applied.includes(IMPORT_TRANSLATION_CACHE_VERSION);
    const importTranslationCacheScopeFixAlreadyApplied = applied.includes(IMPORT_TRANSLATION_CACHE_SCOPE_FIX_VERSION);
    const ownerProjectIdSplitAlreadyApplied = applied.includes(OWNER_PROJECT_ID_SPLIT_VERSION);
    const chatSessionPinsAlreadyApplied = applied.includes(CHAT_SESSION_PINS_VERSION);
    const executorToolFailureRetryAlreadyApplied = applied.includes(EXECUTOR_TOOL_FAILURE_RETRY_VERSION);
    const executorEscalationAttemptAlreadyApplied = applied.includes(EXECUTOR_ESCALATION_ATTEMPT_VERSION);
    const globalRoutinesAlreadyApplied = applied.includes(GLOBAL_ROUTINES_SCHEMA_VERSION);
    const taskMergerModelLaneAlreadyApplied = applied.includes(TASK_MERGER_MODEL_LANE_VERSION);
    let schemaChanged = false;

    if (!baselineAlreadyApplied) {
      const baselineSql = await readBaselineMigrationSql();
      // The baseline contains multiple statements including CREATE SCHEMA, CREATE
      // TABLE, CREATE INDEX, and seed INSERTs. postgres.js executes a single
      // query string as one batch (simple query protocol when unparameterized).
      await tx.execute(sql.raw(baselineSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${INITIAL_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

  /*
   * FNXC:AutomationIsolation 2026-07-13-22:37:
   * A database that already recorded the initial PostgreSQL baseline must still receive project-scoped automation storage. Apply this version independently of 0000; ambiguous legacy ownership fails closed before any bound cron runner can silently omit those schedules.
   */
    if (!automationIsolationAlreadyApplied) {
      const migrationSql = await readFile(AUTOMATION_ISOLATION_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${AUTOMATION_ISOLATION_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:AnalyticsIsolation 2026-07-14-00:05:
    Existing PostgreSQL databases that already recorded 0001 must independently receive analytics project partitions before project-scoped readers and writers start. Keep 0002 versioned so a fresh baseline cannot hide a skipped upgrade path.
    */
    if (!analyticsIsolationAlreadyApplied) {
      const migrationSql = await readFile(ANALYTICS_ISOLATION_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${ANALYTICS_ISOLATION_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:CommandCenterTenantIsolation 2026-07-14-01:04:
    Version 0003 supplies durable ownership for monitor and approval analytics. It must run independently after 0002 so databases that already accepted the earlier analytics migration cannot silently skip the remaining tenant partitions.
    */
    if (!monitorApprovalIsolationAlreadyApplied) {
      const migrationSql = await readFile(MONITOR_APPROVAL_ISOLATION_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:PostgresMigrationCompleteness 2026-07-14-09:27:
    Apply retired-table preservation independently of 0000 so an older partial migration target can retry without losing board, project-auth, or task-reviewer rows. The DDL is additive and idempotent.
    */
    if (!legacyCutoverPreservationAlreadyApplied) {
      const migrationSql = await readFile(LEGACY_CUTOVER_PRESERVATION_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
    Existing targets may already contain one completed project plus a partially copied second project. Apply metadata partitioning and collision-safe revision identity before any retry builds its migration plan.
    */
    if (!multiProjectCutoverAlreadyApplied) {
      const migrationSql = await readFile(MULTI_PROJECT_CUTOVER_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MULTI_PROJECT_CUTOVER_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

  // Run plugin schema-init hooks regardless of whether the baseline was just
  // applied or already present — plugin tables must exist on every connection
  // the applier touches. The hooks are themselves idempotent (CREATE TABLE IF
  // NOT EXISTS), so re-running is safe.
    const pluginHooks = options.pluginHooks ?? DEFAULT_PLUGIN_SCHEMA_INIT_HOOKS;
    await runPluginSchemaInitHooks(tx, pluginHooks);

    /*
    FNXC:ProjectDataIsolation 2026-07-14-12:10:
    Run universal ownership once, after plugin hooks, so first application covers core and plugin tables without duplicate DDL. Later boots validate that every newly introduced plugin table declared the same ownership contract instead of rebuilding primary keys, foreign keys, and policies on every startup.

    FNXC:ProjectArchiveIsolation 2026-07-14-14:31:
    The steady-state audit includes archive.archived_tasks because archived task IDs are project-local and must retain the same forced-RLS boundary as live task rows.
    */
    if (!projectOwnershipAlreadyApplied) {
      const projectOwnershipSql = await readFile(PROJECT_OWNERSHIP_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(projectOwnershipSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${PROJECT_OWNERSHIP_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    } else {
      const ownershipGaps = (await tx.execute(sql`
        SELECT n.nspname || '.' || c.relname AS table_name
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace
        LEFT JOIN information_schema.columns col
          ON col.table_schema = n.nspname
         AND col.table_name = c.relname
         AND col.column_name = 'project_id'
        WHERE (n.nspname = 'project' OR (n.nspname = 'archive' AND c.relname = 'archived_tasks'))
          AND c.relkind = 'r'
          AND (
            col.column_name IS NULL
            OR NOT c.relrowsecurity
            OR NOT c.relforcerowsecurity
            OR NOT EXISTS (
              SELECT 1 FROM pg_policy p
              WHERE p.polrelid = c.oid AND p.polname = 'fusion_project_isolation'
            )
          )
        ORDER BY c.relname
      `)) as unknown as Array<{ table_name: string }>;
      if (ownershipGaps.length > 0) {
        throw new Error(
          `Project-owned tables are missing required isolation: ${ownershipGaps.map(({ table_name }) => table_name).join(", ")}`,
        );
      }
      const relationalGaps = (await tx.execute(sql`
        SELECT c.conrelid::regclass::text AS object_name, c.conname AS detail
        FROM pg_constraint c
        JOIN pg_class t ON t.oid = c.conrelid
        JOIN pg_namespace n ON n.oid = t.relnamespace
        WHERE (n.nspname = 'project' OR (n.nspname = 'archive' AND t.relname = 'archived_tasks'))
          AND c.contype IN ('p', 'u')
          AND NOT EXISTS (
            SELECT 1 FROM unnest(c.conkey) key_attnum
            JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = key_attnum
            WHERE a.attname = 'project_id'
          )
        UNION ALL
        SELECT t.oid::regclass::text, idx.relname
        FROM pg_index i
        JOIN pg_class t ON t.oid = i.indrelid
        JOIN pg_class idx ON idx.oid = i.indexrelid
        JOIN pg_namespace n ON n.oid = t.relnamespace
        WHERE n.nspname = 'project' AND i.indisunique
          AND NOT EXISTS (
            SELECT 1 FROM unnest(i.indkey::smallint[]) key_attnum
            JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = key_attnum
            WHERE a.attname = 'project_id'
          )
        UNION ALL
        SELECT c.conrelid::regclass::text, c.conname
        FROM pg_constraint c
        JOIN pg_class child ON child.oid = c.conrelid
        JOIN pg_namespace child_ns ON child_ns.oid = child.relnamespace
        JOIN pg_class parent ON parent.oid = c.confrelid
        JOIN pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace
        WHERE c.contype = 'f' AND child_ns.nspname = 'project' AND parent_ns.nspname = 'project'
          AND (
            NOT EXISTS (
              SELECT 1 FROM unnest(c.conkey) key_attnum
              JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = key_attnum
              WHERE a.attname = 'project_id'
            ) OR NOT EXISTS (
              SELECT 1 FROM unnest(c.confkey) key_attnum
              JOIN pg_attribute a ON a.attrelid = c.confrelid AND a.attnum = key_attnum
              WHERE a.attname = 'project_id'
            )
          )
        ORDER BY 1, 2
      `)) as unknown as Array<{ object_name: string; detail: string }>;
      if (relationalGaps.length > 0) {
        throw new Error(
          `Project-owned keys or relationships are globally scoped: ${relationalGaps.map(({ object_name, detail }) => `${object_name}.${detail}`).join(", ")}`,
        );
      }
    }

    /*
    FNXC:PostgresMigrationColumnCoverage 2026-07-14-13:17:
    Apply SQLite schema parity independently of the original baseline and ownership migration. Existing partial cutover targets must gain all late source columns before the idempotent migration retry rebuilds its table plan.
    */
    if (!sqliteSchemaParityAlreadyApplied) {
      const sqliteSchemaParitySql = await readFile(SQLITE_SCHEMA_PARITY_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(sqliteSchemaParitySql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${SQLITE_SCHEMA_PARITY_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:PlannerOversight 2026-07-14-18:49:
    Apply session_advisor_enabled independently of 0007 so databases that already
    recorded SQLite schema parity still gain the per-task session-advisor column
    before TaskStore/Drizzle SELECT paths run on boot.
    */
    if (!sessionAdvisorEnabledAlreadyApplied) {
      const sessionAdvisorEnabledSql = await readFile(SESSION_ADVISOR_ENABLED_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(sessionAdvisorEnabledSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${SESSION_ADVISOR_ENABLED_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:MissionFixIdempotency 2026-07-14-18:55:
    Existing PostgreSQL databases receive the validator-run lineage uniqueness invariant independently of earlier schema versions. Duplicate historical rows fail the migration visibly instead of being silently discarded.

    FNXC:PostgresConflictResolution 2026-07-14-20:52:
    Main assigned migration 0008 to session-advisor state before the cutover landed, so mission lineage uniqueness advances to 0009. Both migrations must run in order; sharing a bookkeeping version would silently skip one invariant.
    */
    if (!missionFixIdempotencyAlreadyApplied) {
      const migrationSql = await readFile(MISSION_FIX_IDEMPOTENCY_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MISSION_FIX_IDEMPOTENCY_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:GitHubImportTranslate 2026-07-15-09:30:
    Create the import-translation cache table independently of earlier schema versions so existing databases gain it on boot before any Import Tasks translate/import read runs against it.
    */
    if (!importTranslationCacheAlreadyApplied) {
      const importTranslationCacheSql = await readFile(IMPORT_TRANSLATION_CACHE_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(importTranslationCacheSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${IMPORT_TRANSLATION_CACHE_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:MultiProjectIsolation 2026-07-15-23:40:
    Apply the owner_project_id domain/partition split independently of earlier
    schema versions so existing databases gain the domain column (backfilled from
    the previously conflated partition value) before any store read/write path
    that now targets owner_project_id runs on boot.
    */
    if (!ownerProjectIdSplitAlreadyApplied) {
      const ownerProjectIdSplitSql = await readFile(OWNER_PROJECT_ID_SPLIT_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(ownerProjectIdSplitSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${OWNER_PROJECT_ID_SPLIT_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:ChatPinned 2026-07-16-12:30:
    Apply the pin timestamp separately from the baseline so all pre-existing
    databases can safely read and write Direct chat pins after this rollout.
    */
    if (!chatSessionPinsAlreadyApplied) {
      const chatSessionPinsSql = await readFile(CHAT_SESSION_PINS_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(chatSessionPinsSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${CHAT_SESSION_PINS_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    if (!executorToolFailureRetryAlreadyApplied) {
      const executorToolFailureRetrySql = await readFile(EXECUTOR_TOOL_FAILURE_RETRY_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(executorToolFailureRetrySql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${EXECUTOR_TOOL_FAILURE_RETRY_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    if (!executorEscalationAttemptAlreadyApplied) {
      const executorEscalationAttemptSql = await readFile(EXECUTOR_ESCALATION_ATTEMPT_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(executorEscalationAttemptSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${EXECUTOR_ESCALATION_ATTEMPT_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    if (!globalRoutinesAlreadyApplied) {
      const migrationSql = await readFile(GLOBAL_ROUTINES_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${GLOBAL_ROUTINES_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }
    if (!taskMergerModelLaneAlreadyApplied) {
      const migrationSql = await readFile(TASK_MERGER_MODEL_LANE_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${TASK_MERGER_MODEL_LANE_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    /*
    FNXC:GitHubImportTranslate 2026-07-16-23:30:
    0010's marker prevents its corrected fresh-install definition from running
    on upgrades. Apply 0016 separately before runtime cache reads so existing
    rows, RLS, and unbound compatibility stores share one partition contract.
    */
    if (!importTranslationCacheScopeFixAlreadyApplied) {
      const migrationSql = await readFile(IMPORT_TRANSLATION_CACHE_SCOPE_FIX_MIGRATION_PATH, "utf8");
      await tx.execute(sql.raw(migrationSql));
      await tx.execute(
        sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${IMPORT_TRANSLATION_CACHE_SCOPE_FIX_VERSION}) ON CONFLICT (version) DO NOTHING`,
      );
      schemaChanged = true;
    }

    return { applied: schemaChanged, pluginHooksRun: pluginHooks.length };
  });
}