FN-8462: fix legacy-adoption marker permissions

Prevent inaccessible legacy-adoption markers from triggering repeated CLI sweeps and warning spam.

- Grant fusion_runtime public schema access for the drained-marker read path
- Deduplicate permanent marker infrastructure diagnostics while preserving fail-open sweeps
- Cover runtime-role marker access and migration registration
- Add a patch changeset for the CLI fix

Files changed:
 .../fn-8462-legacy-adoption-drained-marker.md      |  7 ++
 .../core/src/__tests__/legacy-adoption.test.ts     | 37 ++++++++-
 .../legacy-adoption-runtime-marker.test.ts         | 90 ++++++++++++++++++++++
 .../src/__tests__/postgres/schema-applier.test.ts  | 16 ++++
 ...gacy_adoption_drained_marker_runtime_grants.sql |  8 ++
 packages/core/src/task-store/lifecycle-ops.ts      | 70 +++++++++++++----
 6 files changed, 210 insertions(+), 18 deletions(-)

Fusion-Task-Id: FN-8462

Fusion-Task-Lineage: cb93de43-32f3-4c13-9989-b3dd08d33af8

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

.changeset/fn-8462-legacy-adoption-drained-marker.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Stop legacy-adoption drained-marker warn spam on every CLI open under embedded Postgres.
+category: fix
+dev: Grant fusion_runtime public schema usage plus SELECT and a restricted SECURITY DEFINER marker write; permanently unavailable marker infrastructure logs once per process.

packages/core/src/__tests__/legacy-adoption.test.ts

@@ -6,7 +6,7 @@ lacks an adoption-table row — so a status added during the cutover window is c
 at build time instead of mass-parking rows `paused` at upgrade. Plus adoption-action
 + reviewLevel-backfill unit coverage (fixture rows resume owned; never both fields).
 */
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import { readFileSync, readdirSync } from "node:fs";
 import { fileURLToPath } from "node:url";
 import { dirname, join } from "node:path";
@@ -306,7 +306,7 @@ bookkeeping table → no marker, sweep always runs).
 */
 function makeFakeStore(
   rows: Array<Partial<Task> & { id: string }>,
-  opts?: { backend?: boolean; markerPresent?: boolean; markerReadThrows?: boolean },
+  opts?: { backend?: boolean; markerPresent?: boolean; markerReadThrows?: boolean; markerWriteThrows?: boolean; markerError?: Error },
 ) {
   const listCalls: Array<{ limit?: number; offset?: number }> = [];
   const markerWrites: string[] = [];
@@ -330,12 +330,13 @@ function makeFakeStore(
             // helper (SELECT public.fusion_mark_legacy_adoption_drained()); the read path is
             // SELECT version FROM … WHERE version = ….
             if (text.includes("fusion_mark_legacy_adoption_drained")) {
+              if (opts?.markerWriteThrows) throw opts.markerError ?? new Error("marker write boom");
               markerWrites.push(text);
               markerPresent = true;
               return [];
             }
             if (text.includes("SELECT") && text.includes("version")) {
-              if (opts?.markerReadThrows) throw new Error("marker read boom");
+              if (opts?.markerReadThrows) throw opts.markerError ?? new Error("marker read boom");
               return markerPresent ? [{ version: "legacy-adoption-drained" }] : [];
             }
             return [];
@@ -453,4 +454,34 @@ describe("adoptLegacyTaskRowsOnOpen — drained-marker completion short-circuit"
     expect(await adoptLegacyTaskRowsOnOpen(store)).toBe(1);
     expect(rows[0].status).toBeNull();
   });
+
+  it("reports a permanent marker privilege failure once per SQLSTATE class", async () => {
+    const cause = Object.assign(new Error("permission denied for table fusion_schema_migrations"), {code: "42501"});
+    const markerError = new Error("Failed query: SELECT version FROM public.fusion_schema_migrations", {cause});
+    const stderr = vi.spyOn(console, "error").mockImplementation(() => undefined);
+    try {
+      const {store} = makeFakeStore([{id: "task-1", status: "done"}], {
+        backend: true,
+        markerReadThrows: true,
+        markerWriteThrows: true,
+        markerError,
+      });
+
+      await adoptLegacyTaskRowsOnOpen(store);
+      await adoptLegacyTaskRowsOnOpen(store);
+
+      const diagnostics = stderr.mock.calls.filter(([message]) =>
+        String(message).includes("Legacy-adoption drained-marker infrastructure is unavailable"),
+      );
+      expect(diagnostics).toHaveLength(1);
+      expect(diagnostics[0]?.[1]).toMatchObject({
+        operation: "read",
+        sqlstate: "42501",
+        sqlstateClass: "42",
+        hint: expect.stringContaining("schema baseline 0032+"),
+      });
+    } finally {
+      stderr.mockRestore();
+    }
+  });
 });

packages/core/src/__tests__/postgres/legacy-adoption-runtime-marker.test.ts

@@ -0,0 +1,90 @@
+/*
+FNXC:LegacyAdoption 2026-07-22-10:15:
+#2387 only reproduced after startup re-bound TaskStore to fusion_runtime. Unit
+fakes cannot prove the public-schema grant or SECURITY DEFINER boundary, so this
+PG integration test exercises the same restricted role and proves a clean
+second store-open sweep short-circuits instead of recreating CLI warn spam.
+*/
+import {expect, it, vi} from "vitest";
+import {sql} from "drizzle-orm";
+import {TaskStore} from "../../store.js";
+import {createConnectionSetFromUrl} from "../../postgres/connection.js";
+import {createAsyncDataLayer} from "../../postgres/data-layer.js";
+import {
+  LEGACY_ADOPTION_DRAINED_MARKER,
+  LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION,
+  MIGRATION_BOOKKEEPING_TABLE,
+} from "../../postgres/schema-applier.js";
+import {createTaskStoreForTest, pgDescribe} from "../../__test-utils__/pg-test-harness.js";
+
+pgDescribe("legacy-adoption drained marker: fusion_runtime integration (#2387)", () => {
+  it("reads and writes the marker under fusion_runtime, then short-circuits a second clean sweep", async () => {
+    const harness = await createTaskStoreForTest({prefix: "legacy_adoption_runtime_marker", copyFromGolden: true});
+    const createRuntimeLayer = async () => {
+      const connections = await createConnectionSetFromUrl(
+        {
+          mode: "external",
+          runtimeUrl: harness.testUrl,
+          migrationUrl: harness.testUrl,
+          migrationUrlOverridden: false,
+        },
+        {poolMax: 1, connectTimeoutSeconds: 5, projectId: "legacy-marker-test", useRuntimeRole: true},
+      );
+      return createAsyncDataLayer(connections, {projectId: "legacy-marker-test"});
+    };
+    let firstStore: TaskStore | undefined;
+    let secondStore: TaskStore | undefined;
+    let runtimeLayer: ReturnType<typeof createAsyncDataLayer> | undefined;
+    let stderr: ReturnType<typeof vi.spyOn> | undefined;
+    let warnings: ReturnType<typeof vi.spyOn> | undefined;
+    try {
+      runtimeLayer = await createRuntimeLayer();
+      await harness.adminDb.execute(sql`DELETE FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} WHERE version = ${LEGACY_ADOPTION_DRAINED_MARKER}`);
+
+      // Direct role proof: no broad INSERT is granted, but SELECT and the restricted helper work.
+      await expect(runtimeLayer.db.execute(sql`SELECT current_user`)).resolves.toEqual([{current_user: "fusion_runtime"}]);
+      await expect(runtimeLayer.db.execute(
+        sql`SELECT version FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} WHERE version = ${LEGACY_ADOPTION_DRAINED_MARKER}`,
+      )).resolves.toEqual([]);
+      await expect(runtimeLayer.db.execute(sql`SELECT public.${sql.identifier(LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION)}()`)).resolves.toBeDefined();
+      await expect(runtimeLayer.db.execute(
+        sql`SELECT version FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} WHERE version = ${LEGACY_ADOPTION_DRAINED_MARKER}`,
+      )).resolves.toEqual([{version: LEGACY_ADOPTION_DRAINED_MARKER}]);
+      await runtimeLayer.close();
+      runtimeLayer = undefined;
+
+      // FNXC:LegacyAdoption 2026-07-22-16:10:
+      // #2387 must be proven through TaskStore.init(), the production store-open hook.
+      // Two independently runtime-bound stores model two CLI opens: the first scans and
+      // writes the marker; the second must read it before listTasks can start a census.
+      stderr = vi.spyOn(console, "error").mockImplementation(() => undefined);
+      warnings = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+      await harness.adminDb.execute(sql`DELETE FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} WHERE version = ${LEGACY_ADOPTION_DRAINED_MARKER}`);
+
+      firstStore = new TaskStore(harness.rootDir, undefined, {asyncLayer: await createRuntimeLayer()});
+      const firstListTasks = vi.spyOn(firstStore, "listTasks");
+      await firstStore.init();
+      expect(firstListTasks).toHaveBeenCalled();
+      await expect(harness.adminDb.execute(
+        sql`SELECT version FROM public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} WHERE version = ${LEGACY_ADOPTION_DRAINED_MARKER}`,
+      )).resolves.toEqual([{version: LEGACY_ADOPTION_DRAINED_MARKER}]);
+      await firstStore.close();
+      firstStore = undefined;
+
+      secondStore = new TaskStore(harness.rootDir, undefined, {asyncLayer: await createRuntimeLayer()});
+      const secondListTasks = vi.spyOn(secondStore, "listTasks");
+      await secondStore.init();
+      expect(secondListTasks).not.toHaveBeenCalled();
+      expect([...stderr.mock.calls, ...warnings.mock.calls].filter(([message]) =>
+        String(message).includes("Legacy-adoption drained-marker"),
+      )).toEqual([]);
+    } finally {
+      stderr?.mockRestore();
+      warnings?.mockRestore();
+      if (firstStore) await firstStore.close();
+      if (secondStore) await secondStore.close();
+      if (runtimeLayer) await runtimeLayer.close();
+      await harness.teardown();
+    }
+  });
+});

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

@@ -75,6 +75,7 @@ import {
   PLANNING_ACTIVE_TIMING_VERSION,
   SQLITE_MIGRATION_RUNTIME_READ_VERSION,
   WORKFLOW_TASK_CONTINUATIONS_VERSION,
+  LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
 } from "../../postgres/schema-applier.js";
 import { ProjectPartitionRekeyError, rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js";
 import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js";
@@ -186,6 +187,16 @@ describe("schema-applier: immutable migration identities", () => {
     expect(Number(SCHEMA_BASELINE_VERSION)).toBeGreaterThanOrEqual(Number(SYMBOL_LOCKS_SCHEMA_VERSION));
   });
 
+  /*
+  FNXC:LegacyAdoption 2026-07-22-10:45:
+  #2387 requires the runtime-role grants to run as an explicit forward migration;
+  a baseline bump alone would leave already-created embedded clusters warn-spamming.
+  */
+  it("registers runtime drained-marker grants at migration version 0032", () => {
+    expect(LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION).toBe("0032");
+    expect(Number(SCHEMA_BASELINE_VERSION)).toBeGreaterThanOrEqual(Number(LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION));
+  });
+
   /*
   FNXC:PostgresBigintCounters 2026-07-19-12:00:
   0026 widens overflow-prone counters to bigint. Keep identity fixed and at-or-before SCHEMA_BASELINE_VERSION.
@@ -1582,6 +1593,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
       PLANNING_ACTIVE_TIMING_VERSION,
       SQLITE_MIGRATION_RUNTIME_READ_VERSION,
       WORKFLOW_TASK_CONTINUATIONS_VERSION,
+      LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
     ]);
     expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false);
   });
@@ -1639,6 +1651,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
       PLANNING_ACTIVE_TIMING_VERSION,
       SQLITE_MIGRATION_RUNTIME_READ_VERSION,
       WORKFLOW_TASK_CONTINUATIONS_VERSION,
+      LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
     ]);
   });
 
@@ -1829,6 +1842,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
       PLANNING_ACTIVE_TIMING_VERSION,
       SQLITE_MIGRATION_RUNTIME_READ_VERSION,
       WORKFLOW_TASK_CONTINUATIONS_VERSION,
+      LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
     ]);
   });
 
@@ -1900,6 +1914,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
       PLANNING_ACTIVE_TIMING_VERSION,
       SQLITE_MIGRATION_RUNTIME_READ_VERSION,
       WORKFLOW_TASK_CONTINUATIONS_VERSION,
+      LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
     ]);
   });
 
@@ -1971,6 +1986,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
       PLANNING_ACTIVE_TIMING_VERSION,
       SQLITE_MIGRATION_RUNTIME_READ_VERSION,
       WORKFLOW_TASK_CONTINUATIONS_VERSION,
+      LEGACY_ADOPTION_DRAINED_MARKER_RUNTIME_GRANTS_VERSION,
     ]);
   });
 });

packages/core/src/postgres/migrations/0032_legacy_adoption_drained_marker_runtime_grants.sql

@@ -18,6 +18,14 @@ BEGIN
     RETURN;
   END IF;
 
+  /*
+  FNXC:LegacyAdoption 2026-07-22-10:15:
+  #2387 was permission denied, not a missing bookkeeping table: store open runs
+  as fusion_runtime after the migration connection creates public.fusion_schema_migrations.
+  Schema USAGE is required in addition to table SELECT on clusters that hardened
+  public's default privileges. Keep the write capability restricted to the helper.
+  */
+  GRANT USAGE ON SCHEMA public TO fusion_runtime;
   GRANT SELECT ON public.fusion_schema_migrations TO fusion_runtime;
 
   CREATE OR REPLACE FUNCTION public.fusion_mark_legacy_adoption_drained()

packages/core/src/task-store/lifecycle-ops.ts

@@ -325,22 +325,68 @@ Safety rules:
 - SQLite (non-backend) mode has no bookkeeping table → no marker, sweep always runs.
 */
 /*
-FNXC:LegacyAdoption 2026-07-21-17:30:
-Drizzle wraps Postgres errors as "Failed query: <SQL> params: …" while the real
-SQLSTATE lives on err.cause. Walk a short cause chain so drained-marker failures
-surface as permission denied / missing function instead of opaque query text.
+FNXC:LegacyAdoption 2026-07-22-10:30:
+#2387's Drizzle wrapper says only "Failed query" while PostgreSQL puts the
+actionable SQLSTATE on a nested cause. Preserve that chain in the diagnostic and
+only report permanent marker infrastructure classes once per process: a CLI
+process may open many stores, but repeated permission-denied spam hides real
+startup failures. Read failures still fail open to preserve adoption correctness.
 */
+const reportedLegacyAdoptionMarkerFailureClasses = new Set<string>();
+const PERMANENT_LEGACY_ADOPTION_MARKER_SQLSTATES = new Set(["42501", "42P01", "42883", "3F000"]);
+
+type StoreOpenDbError = {cause?: unknown; code?: unknown; message?: unknown};
+
+function getStoreOpenDbErrorSqlstate(error: unknown): string | undefined {
+  let current: unknown = error;
+  for (let depth = 0; current !== undefined && current !== null && depth < 4; depth += 1) {
+    if (typeof current !== "object") break;
+    const candidate = current as StoreOpenDbError;
+    if (typeof candidate.code === "string" && /^[0-9A-Z]{5}$/.test(candidate.code)) return candidate.code;
+    current = candidate.cause;
+  }
+  return undefined;
+}
+
 function describeStoreOpenDbError(error: unknown): string {
   const parts: string[] = [];
   let current: unknown = error;
   for (let depth = 0; current !== undefined && current !== null && depth < 4; depth += 1) {
-    const message = current instanceof Error ? current.message : String(current);
-    parts.push(message.length > 400 ? `${message.slice(0, 200)} … ${message.slice(-120)}` : message);
-    current = current instanceof Error ? current.cause : undefined;
+    const candidate = typeof current === "object" ? current as StoreOpenDbError : undefined;
+    const message = current instanceof Error
+      ? current.message
+      : typeof candidate?.message === "string" ? candidate.message : String(current);
+    const sqlstate = typeof candidate?.code === "string" && /^[0-9A-Z]{5}$/.test(candidate.code)
+      ? ` [SQLSTATE ${candidate.code}]`
+      : "";
+    parts.push(`${message.length > 400 ? `${message.slice(0, 200)} … ${message.slice(-120)}` : message}${sqlstate}`);
+    current = candidate?.cause;
   }
   return parts.join(" ⇐ ");
 }
 
+function reportLegacyAdoptionMarkerFailure(operation: "read" | "write", error: unknown): void {
+  const sqlstate = getStoreOpenDbErrorSqlstate(error);
+  if (sqlstate && PERMANENT_LEGACY_ADOPTION_MARKER_SQLSTATES.has(sqlstate)) {
+    const sqlstateClass = sqlstate.slice(0, 2);
+    if (reportedLegacyAdoptionMarkerFailureClasses.has(sqlstateClass)) return;
+    reportedLegacyAdoptionMarkerFailureClasses.add(sqlstateClass);
+    storeLog.error("Legacy-adoption drained-marker infrastructure is unavailable; sweeping will continue", {
+      phase: "init:legacy-adoption",
+      operation,
+      sqlstate,
+      sqlstateClass,
+      error: describeStoreOpenDbError(error),
+      hint: "Apply schema baseline 0032+ and verify fusion_runtime has public schema USAGE, bookkeeping SELECT, and marker-helper EXECUTE.",
+    });
+    return;
+  }
+  storeLog.warn(`Legacy-adoption drained-marker ${operation} failed${operation === "read" ? " — sweeping anyway" : ""}`, {
+    phase: "init:legacy-adoption",
+    error: describeStoreOpenDbError(error),
+  });
+}
+
 async function hasLegacyAdoptionDrainedMarker(store: TaskStore): Promise<boolean> {
   const db = store.asyncLayer?.db;
   if (!db) return false;
@@ -351,10 +397,7 @@ async function hasLegacyAdoptionDrainedMarker(store: TaskStore): Promise<boolean
     return rows.length > 0;
   } catch (error) {
     // Fail-open toward correctness: an unreadable marker means sweep.
-    storeLog.warn("Legacy-adoption drained-marker read failed — sweeping anyway", {
-      phase: "init:legacy-adoption",
-      error: describeStoreOpenDbError(error),
-    });
+    reportLegacyAdoptionMarkerFailure("read", error);
     return false;
   }
 }
@@ -372,10 +415,7 @@ async function writeLegacyAdoptionDrainedMarker(store: TaskStore): Promise<void>
     await db.execute(sql`SELECT public.${sql.identifier(LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION)}()`);
   } catch (error) {
     // Non-fatal: the next fully-clean drain writes it again.
-    storeLog.warn("Legacy-adoption drained-marker write failed", {
-      phase: "init:legacy-adoption",
-      error: describeStoreOpenDbError(error),
-    });
+    reportLegacyAdoptionMarkerFailure("write", error);
   }
 }