FN-8051: ensure PostgreSQL schemas initialize before plugin hooks

Ensure required PostgreSQL namespaces exist before plugin initialization on every boot.

- Create project, central, and archive schemas under the schema advisory lock before hooks run
- Cover marker-present databases with a plugin-hook schema availability regression test
- Add a patch changeset for the reliability fix

Files changed:
 .changeset/fn-8051-schema-init.md                  |  7 ++++
 .../src/__tests__/postgres/schema-applier.test.ts  | 43 ++++++++++++++++++++++
 packages/core/src/postgres/schema-applier.ts       | 12 ++++++
 3 files changed, 62 insertions(+)

Fusion-Task-Id: FN-8051

Fusion-Task-Lineage: a3b20683-a742-4a8c-9cfc-fbf316c5649b

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>

.changeset/fn-8051-schema-init.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Ensure required database schemas always initialize before plugin tables on boot.
+category: fix
+dev: applySchemaBaseline now runs CREATE SCHEMA IF NOT EXISTS project/central/archive unconditionally before plugin schema-init hooks (FN-8051).

packages/core/src/__tests__/postgres/schema-applier.test.ts

@@ -47,6 +47,7 @@ import {
   SQLITE_SCHEMA_PARITY_VERSION,
 } from "../../postgres/schema-applier.js";
 import { rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js";
+import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js";
 
 const PG_ADMIN_URL =
   process.env.FUSION_PG_TEST_ADMIN_URL ?? "postgresql://localhost:5432/postgres";
@@ -372,6 +373,48 @@ pgDescribe("schema-applier: VAL-SCHEMA-008 three-database topology", () => {
     `)) as unknown as Array<{ schema_name: string }>;
     expect(rows.map((r) => r.schema_name)).toEqual(["archive", "central", "project"]);
   });
+
+  it("ensures schemas before hooks when all migration markers are already recorded", async () => {
+    ctx = await setupFreshDb();
+    await ctx.db.execute(sql.raw(`
+      CREATE TABLE public.fusion_schema_migrations (
+        version text PRIMARY KEY,
+        applied_at timestamptz NOT NULL DEFAULT now()
+      );
+      INSERT INTO public.fusion_schema_migrations (version)
+      SELECT lpad(n::text, 4, '0')
+      FROM generate_series(0, ${Number(SCHEMA_BASELINE_VERSION)}) AS migration(n);
+    `));
+
+    const observedSchemas: string[] = [];
+    const assertSchemasHook: PluginSchemaInitHook = {
+      pluginId: "assert-required-schemas",
+      async init(db) {
+        const rows = (await db.execute(sql`
+          SELECT schema_name FROM information_schema.schemata
+          WHERE schema_name IN ('project', 'central', 'archive')
+          ORDER BY schema_name
+        `)) as unknown as Array<{ schema_name: string }>;
+        observedSchemas.push(...rows.map(({ schema_name }) => schema_name));
+        if (rows.length !== 3) {
+          throw new Error(`Required schemas missing at plugin hook time: ${rows.map(({ schema_name }) => schema_name).join(", ")}`);
+        }
+      },
+    };
+
+    await expect(applySchemaBaseline(ctx.db, { pluginHooks: [assertSchemasHook] })).resolves.toEqual({
+      applied: false,
+      pluginHooksRun: 1,
+    });
+    expect(observedSchemas).toEqual(["archive", "central", "project"]);
+
+    const schemas = (await ctx.db.execute(sql`
+      SELECT schema_name FROM information_schema.schemata
+      WHERE schema_name IN ('project', 'central', 'archive')
+      ORDER BY schema_name
+    `)) as unknown as Array<{ schema_name: string }>;
+    expect(schemas.map(({ schema_name }) => schema_name)).toEqual(["archive", "central", "project"]);
+  });
 });
 
 pgDescribe("schema-applier: VAL-SCHEMA-001 final-schema parity (table counts)", () => {

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

@@ -182,6 +182,18 @@ export async function applySchemaBaseline(
   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);