root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-8419: safeguard project partition reconciliation

Browse Source 
Safely reconcile fallback and registered project partitions during dashboard startup.

- Merge duplicate partition rows with fallback data taking precedence.
- Validate unique indexes and foreign-key dependencies before rekeying.
- Bind safely after failed promotion and stop non-retryable dashboard failures.
- Add PostgreSQL reconciliation and supervisor coverage.

Files changed:
 .changeset/fn-8419-rekey-partition-merge.md        |   7 +
 packages/cli/src/bin.ts                            |   5 +-
 .../commands/__tests__/dashboard-supervise.test.ts |  16 +-
 packages/cli/src/commands/dashboard.ts             |  41 ++-
 .../src/__tests__/postgres/schema-applier.test.ts  | 178 +++++++++++-
 packages/core/src/async-secrets-store.ts           |   9 +-
 packages/core/src/index.ts                         |   8 +-
 packages/core/src/postgres-errors.ts               |   9 +
 packages/core/src/postgres/index.ts                |   5 +
 packages/core/src/postgres/migration-stamping.ts   | 318 ++++++++++++++++-----
 packages/core/src/postgres/startup-factory.ts      |  49 +++-
 packages/core/src/process-supervisor.ts            |   3 +
 packages/core/src/task-store/async-persistence.ts  |   7 +-
 13 files changed, 546 insertions(+), 109 deletions(-)

Fusion-Task-Id: FN-8419

Fusion-Task-Lineage: bfc54e40-a31e-4b61-b6ae-01eb147efde1

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-20 01:54:42 -07:00
parent 02f8bffb2e
commit 4f0d89e106
13 changed files with 546 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-8419-rekey-partition-merge.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Fix startup crash when a project has both fallback and registered partition data.
category: fix
dev: Rekey merges catalog-discovered dual partition conflicts fallback-wins with NULL-correct matching and fail-closed FK checks; startup degrades to fallback data and unique failures stop supervised retry loops.

5
packages/cli/src/bin.ts
View File

@@ -2336,7 +2336,10 @@ async function main() {
}
} catch (err) {
console.error(`Error: ${err instanceof Error ? err.message : String(err)}`);
process.exit(1);
const { isPostgresUniqueError, ProjectPartitionRekeyError, FUSION_NON_RETRYABLE_EXIT_CODE } = await import("@fusion/core");
process.exit(isPostgresUniqueError(err) || err instanceof ProjectPartitionRekeyError
? FUSION_NON_RETRYABLE_EXIT_CODE
: 1);
}
}

16
packages/cli/src/commands/__tests__/dashboard-supervise.test.ts
View File

@@ -1,5 +1,6 @@
import { afterEach, describe, expect, it } from "vitest";
import { resolveSupervisorRespawnCommand, shouldSuperviseDashboard } from "../dashboard.js";
import { classifyDashboardFatalExit, resolveSupervisorRespawnCommand, shouldSuperviseDashboard } from "../dashboard.js";
import { FUSION_NON_RETRYABLE_EXIT_CODE } from "@fusion/core";
/*
FNXC:SystemPanel 2026-07-12-14:25:
@@ -32,6 +33,19 @@ describe("shouldSuperviseDashboard", () => {
});
});
describe("classifyDashboardFatalExit", () => {
it("stops unique-constraint failures without consuming restart attempts", () => {
expect(classifyDashboardFatalExit({ cause: { code: "23505" } })).toEqual({
exitCode: FUSION_NON_RETRYABLE_EXIT_CODE,
nonRetryable: true,
});
});
it("leaves ordinary startup failures retryable", () => {
expect(classifyDashboardFatalExit(new Error("port unavailable"))).toEqual({ exitCode: 1, nonRetryable: false });
});
});
describe("resolveSupervisorRespawnCommand", () => {
const originalBun = (globalThis as { Bun?: unknown }).Bun;

41
packages/cli/src/commands/dashboard.ts
View File

@@ -32,6 +32,9 @@ import {
type TraitFlags,
createTaskStoreForBackend,
FUSION_RESTART_EXIT_CODE,
FUSION_NON_RETRYABLE_EXIT_CODE,
isPostgresUniqueError,
ProjectPartitionRekeyError,
} from "@fusion/core";
import {
createServer,
@@ -914,14 +917,24 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
host: selectedHost,
log: (message) => logSink.log(message, "dashboard"),
});
const dashboardBackendBoot = await phaseTime(
"backend.factory",
() => createTaskStoreForBackend({
rootDir: cwd,
onMigrationProgress: (event) => migrationHoldingServer?.setMigrationProgress(event),
}),
logPhase,
);
let dashboardBackendBoot: Awaited<ReturnType<typeof createTaskStoreForBackend>>;
try {
dashboardBackendBoot = await phaseTime(
"backend.factory",
() => createTaskStoreForBackend({
rootDir: cwd,
onMigrationProgress: (event) => migrationHoldingServer?.setMigrationProgress(event),
}),
logPhase,
);
} catch (error) {
const fatal = classifyDashboardFatalExit(error);
if (fatal.nonRetryable) {
console.error(`[dashboard] startup stopped: unique constraint or project partition reconciliation failure: ${error instanceof Error ? error.message : String(error)}`);
process.exit(fatal.exitCode);
}
throw error;
}
// FNXC:PostgresFinalCutover 2026-07-14-17:20: Dashboard runtime storage is
// PostgreSQL-only; factory failure is surfaced instead of creating a dead store.
store = dashboardBackendBoot.taskStore;
@@ -3537,6 +3550,13 @@ export function shouldSuperviseDashboard(
* This does NOT use shell detachment wrappers, shell kill loops, or unbounded retries.
* Port 4040 processes are never killed — the child binds its own port.
*/
export function classifyDashboardFatalExit(error: unknown): { exitCode: number; nonRetryable: boolean } {
if (isPostgresUniqueError(error) || error instanceof ProjectPartitionRekeyError) {
return { exitCode: FUSION_NON_RETRYABLE_EXIT_CODE, nonRetryable: true };
}
return { exitCode: 1, nonRetryable: false };
}
export async function runDashboardSupervised(
port: number,
_opts: Parameters<typeof runDashboard>[1] = {},
@@ -3629,6 +3649,11 @@ export async function runDashboardSupervised(
and reset the crash budget — an intentional restart must never consume
SUPERVISE_MAX_RESTARTS or incur crash backoff.
*/
if (exitCode === FUSION_NON_RETRYABLE_EXIT_CODE) {
console.error("[dashboard:supervisor] dashboard stopped after a non-retryable unique constraint or project partition identity reconciliation failure; inspect the preceding startup error.");
process.exit(exitCode);
}
if (exitCode === FUSION_RESTART_EXIT_CODE) {
console.log("[dashboard:supervisor] restart requested — restarting now");
restartCount = 0;

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

@@ -73,7 +73,7 @@ import {
TASK_VERIFICATION_REQUEST_VERSION,
TASK_DECLARED_SYMBOLS_VERSION,
} from "../../postgres/schema-applier.js";
import { rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js";
import { ProjectPartitionRekeyError, rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js";
import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js";
const PG_ADMIN_URL =
@@ -1107,6 +1107,182 @@ pgDescribe("schema-applier: VAL-SCHEMA-001 final-schema parity (table counts)",
`)).resolves.toHaveLength(1);
});
it("merges dual partitions fallback-wins with NULL-correct catalog unique rules", async () => {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql`
CREATE TABLE project.fn8419_null_unique_probe (project_id text NOT NULL, tag text, payload text NOT NULL);
CREATE UNIQUE INDEX fn8419_null_unique_probe_uq ON project.fn8419_null_unique_probe (project_id, tag);
INSERT INTO project.config(project_id, updated_at, settings)
VALUES ('local-fallback', 'fallback-time', '{"winner":"fallback"}'), ('registered-project', 'scaffold-time', '{"winner":"scaffold"}');
INSERT INTO project.fn8419_null_unique_probe(project_id, tag, payload)
VALUES ('local-fallback', 'same', 'fallback-wins'), ('registered-project', 'same', 'scaffold'), ('registered-project', NULL, 'must-survive');
`);
await expect(rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project")).resolves.toBe(true);
await expect(ctx.db.execute(sql`
SELECT project_id, tag, payload FROM project.fn8419_null_unique_probe ORDER BY tag NULLS FIRST
`)).resolves.toEqual([
{ project_id: "registered-project", tag: null, payload: "must-survive" },
{ project_id: "registered-project", tag: "same", payload: "fallback-wins" },
]);
await expect(ctx.db.execute(sql`
SELECT settings FROM project.config WHERE project_id = 'registered-project'
`)).resolves.toEqual([{ settings: { winner: "fallback" } }]);
await expect(ctx.db.execute(sql`
SELECT count(*)::int AS count FROM project.config WHERE project_id = 'local-fallback'
`)).resolves.toEqual([{ count: 0 }]);
});
it("retains an inbound non-project dependent for each separate FK constraint", async () => {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql`
CREATE TABLE public.fn8419_external_dependents (
first_parent text,
second_parent text,
CONSTRAINT fn8419_first_parent_fk FOREIGN KEY (first_parent)
REFERENCES project.config(project_id) ON DELETE CASCADE ON UPDATE CASCADE,
CONSTRAINT fn8419_second_parent_fk FOREIGN KEY (second_parent)
REFERENCES project.config(project_id) ON DELETE CASCADE ON UPDATE CASCADE
);
INSERT INTO project.config(project_id, updated_at, settings)
VALUES ('local-fallback', 'fallback-time', '{"winner":"fallback"}'), ('registered-project', 'scaffold-time', '{"winner":"scaffold"}');
INSERT INTO public.fn8419_external_dependents(first_parent, second_parent)
VALUES ('registered-project', NULL);
`);
await expect(rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project"))
.rejects.toMatchObject({
name: "ProjectPartitionRekeyError",
reason: "unreplaced-fk-dependent",
} satisfies Partial<ProjectPartitionRekeyError>);
await expect(ctx.db.execute(sql`
SELECT first_parent, second_parent FROM public.fn8419_external_dependents
`)).resolves.toEqual([{ first_parent: "registered-project", second_parent: null }]);
await expect(ctx.db.execute(sql`
SELECT project_id FROM project.config ORDER BY project_id
`)).resolves.toEqual([{ project_id: "local-fallback" }, { project_id: "registered-project" }]);
});
it("refuses deferred UPDATE SET NULL and SET DEFAULT partition mutations", async () => {
for (const [name, action] of [["set_null", "SET NULL"], ["set_default", "SET DEFAULT"]] as const) {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql.raw(`
CREATE TABLE public.fn8419_${name}_dependent (
parent_id text DEFAULT 'local-fallback' REFERENCES project.config(project_id)
ON UPDATE ${action} DEFERRABLE INITIALLY DEFERRED
);
INSERT INTO project.config(project_id, updated_at, settings)
VALUES ('local-fallback', 'fallback-time', '{"winner":"fallback"}');
INSERT INTO public.fn8419_${name}_dependent(parent_id) VALUES ('local-fallback');
`));
await expect(rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project"))
.rejects.toMatchObject({
name: "ProjectPartitionRekeyError",
reason: "unsafe-fk-update-graph",
} satisfies Partial<ProjectPartitionRekeyError>);
await expect(ctx.db.execute(sql.raw(`SELECT parent_id FROM public.fn8419_${name}_dependent`)))
.resolves.toEqual([{ parent_id: "local-fallback" }]);
await teardownDb(ctx);
ctx = null;
}
});
it("allows a conflict-deletable registered child to be replaced before its parent", async () => {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql.raw(`
CREATE TABLE project.fn8419_replace_parent (project_id text PRIMARY KEY, payload text NOT NULL);
CREATE TABLE project.fn8419_replace_child (
project_id text PRIMARY KEY,
parent_id text NOT NULL REFERENCES project.fn8419_replace_parent(project_id)
ON DELETE RESTRICT ON UPDATE CASCADE,
payload text NOT NULL
);
INSERT INTO project.fn8419_replace_parent VALUES
('local-fallback', 'fallback-parent'), ('registered-project', 'scaffold-parent');
INSERT INTO project.fn8419_replace_child VALUES
('local-fallback', 'local-fallback', 'fallback-child'),
('registered-project', 'registered-project', 'scaffold-child');
`));
await expect(rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project")).resolves.toBe(true);
await expect(ctx.db.execute(sql`SELECT project_id, parent_id, payload FROM project.fn8419_replace_child`))
.resolves.toEqual([{ project_id: "registered-project", parent_id: "registered-project", payload: "fallback-child" }]);
});
it("retains unreplaced registered dependents for every delete action", async () => {
for (const [name, deleteAction, childDefinition] of [
["restrict", "RESTRICT", "parent_id text NOT NULL"],
["cascade", "CASCADE", "parent_id text NOT NULL"],
["set_null", "SET NULL", "parent_id text"],
["set_default", "SET DEFAULT", "parent_id text NOT NULL DEFAULT 'registered-project'"],
] as const) {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql.raw(`
CREATE TABLE project.fn8419_delete_${name}_parent (project_id text PRIMARY KEY, payload text NOT NULL);
CREATE TABLE project.fn8419_delete_${name}_child (
project_id text PRIMARY KEY,
${childDefinition} REFERENCES project.fn8419_delete_${name}_parent(project_id)
ON DELETE ${deleteAction} ON UPDATE CASCADE,
payload text NOT NULL
);
INSERT INTO project.fn8419_delete_${name}_parent VALUES
('local-fallback', 'fallback'), ('registered-project', 'scaffold');
INSERT INTO project.fn8419_delete_${name}_child(project_id, parent_id, payload)
VALUES ('registered-project', 'registered-project', 'must-survive');
`));
await expect(rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project"))
.rejects.toMatchObject({ name: "ProjectPartitionRekeyError", reason: "unreplaced-fk-dependent" } satisfies Partial<ProjectPartitionRekeyError>);
await expect(ctx.db.execute(sql.raw(`SELECT project_id, parent_id, payload FROM project.fn8419_delete_${name}_child`)))
.resolves.toEqual([{ project_id: "registered-project", parent_id: "registered-project", payload: "must-survive" }]);
await expect(ctx.db.execute(sql.raw(`SELECT project_id FROM project.fn8419_delete_${name}_parent ORDER BY project_id`)))
.resolves.toEqual([{ project_id: "local-fallback" }, { project_id: "registered-project" }]);
await teardownDb(ctx);
ctx = null;
}
});
it("rejects unsafe fallback update FK graphs but permits deferred and cascade controls", async () => {
for (const [name, updateClause, expected] of [
["unsafe", "ON UPDATE NO ACTION", "unsafe-fk-update-graph"],
["deferred", "ON UPDATE NO ACTION DEFERRABLE INITIALLY DEFERRED", undefined],
["cascade", "ON UPDATE CASCADE", undefined],
] as const) {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);
await ctx.db.execute(sql.raw(`
CREATE TABLE project.fn8419_update_${name}_parent (
project_id text NOT NULL, id text NOT NULL, PRIMARY KEY(project_id, id)
);
CREATE TABLE project.fn8419_update_${name}_child (
project_id text NOT NULL, parent_id text NOT NULL,
PRIMARY KEY(project_id, parent_id),
FOREIGN KEY(project_id, parent_id) REFERENCES project.fn8419_update_${name}_parent(project_id, id) ${updateClause}
);
INSERT INTO project.fn8419_update_${name}_parent VALUES ('local-fallback', 'parent');
INSERT INTO project.fn8419_update_${name}_child VALUES ('local-fallback', 'parent');
`));
const rekey = rekeyFallbackProjectPartition(ctx.db, "local-fallback", "registered-project");
if (expected) {
await expect(rekey).rejects.toMatchObject({ name: "ProjectPartitionRekeyError", reason: expected } satisfies Partial<ProjectPartitionRekeyError>);
await expect(ctx.db.execute(sql.raw(`SELECT project_id FROM project.fn8419_update_${name}_child`)))
.resolves.toEqual([{ project_id: "local-fallback" }]);
} else {
await expect(rekey).resolves.toBe(true);
await expect(ctx.db.execute(sql.raw(`SELECT project_id FROM project.fn8419_update_${name}_child`)))
.resolves.toEqual([{ project_id: "registered-project" }]);
}
await teardownDb(ctx);
ctx = null;
}
});
it("quarantines ownerless rows when complete and failed migrations name different projects", async () => {
ctx = await setupFreshDb();
await applySchemaBaseline(ctx.db);

9
packages/core/src/async-secrets-store.ts
View File

@@ -41,6 +41,7 @@
*/
import { randomUUID } from "node:crypto";
import { asc, eq, sql } from "drizzle-orm";
import { isPostgresUniqueError } from "./postgres-errors.js";
import * as schema from "./postgres/schema/index.js";
import type { AsyncDataLayer, DbTransaction } from "./postgres/data-layer.js";
import {
@@ -139,14 +140,6 @@ function tableForScope(scope: SecretScope): ProjectSecretsTable {
: (schema.central.secretsGlobal as unknown as ProjectSecretsTable);
}
function isPostgresUniqueError(error: unknown): boolean {
// PostgreSQL unique_violation (23505). The code may be on the error directly
// (raw postgres.js) or on the `cause` (Drizzle wraps postgres errors).
const directCode = (error as { code?: string } | null)?.code;
const causeCode = (error as { cause?: { code?: string } } | null)?.cause?.code;
return directCode === "23505" || causeCode === "23505";
}
function isAccessPolicy(value: string): value is SecretAccessPolicy {
return value === "auto" || value === "prompt" || value === "deny";
}

8
packages/core/src/index.ts
View File

@@ -1029,7 +1029,8 @@ export {
readProjectIdentityAsync,
writeProjectIdentityAsync,
} from "./project-identity.js";
export { ProcessSupervisor, superviseSpawn, FUSION_RESTART_EXIT_CODE } from "./process-supervisor.js";
export { ProcessSupervisor, superviseSpawn, FUSION_RESTART_EXIT_CODE, FUSION_NON_RETRYABLE_EXIT_CODE } from "./process-supervisor.js";
export { isPostgresUniqueError } from "./postgres-errors.js";
export type {
SuperviseSpawnOptions,
SupervisedChild,
@@ -2444,6 +2445,9 @@ export {
// to the central-registry project id on BOTH cutover paths.
stampMigratedProjectRows,
lookupRegisteredProjectIdByPath,
rekeyFallbackProjectPartition,
ProjectPartitionRekeyError,
selectDegradedBindTarget,
applySchemaBaseline,
getAppliedMigrations,
SCHEMA_BASELINE_VERSION,
@@ -2494,6 +2498,8 @@ export type {
TableMigrationResult,
StampMigratedProjectRowsInput,
StampMigratedProjectRowsResult,
ProjectPartitionOwnership,
ProjectPartitionRekeyReason,
BackendBootResult,
CreateTaskStoreForBackendOptions,
LoadedPluginSchemaContract,

9
packages/core/src/postgres-errors.ts Normal file
View File

@@ -0,0 +1,9 @@
/** PostgreSQL unique_violation (23505), including Drizzle's wrapped cause shape. */
export function isPostgresUniqueError(error: unknown): boolean {
let current: unknown = error;
for (let depth = 0; current && depth < 5; depth += 1) {
if ((current as { code?: string }).code === "23505") return true;
current = (current as { cause?: unknown }).cause;
}
return false;
}

5
packages/core/src/postgres/index.ts
View File

@@ -201,8 +201,13 @@ export {
export {
stampMigratedProjectRows,
lookupRegisteredProjectIdByPath,
rekeyFallbackProjectPartition,
ProjectPartitionRekeyError,
selectDegradedBindTarget,
type StampMigratedProjectRowsInput,
type StampMigratedProjectRowsResult,
type ProjectPartitionOwnership,
type ProjectPartitionRekeyReason,
} from "./migration-stamping.js";
/**

318
packages/core/src/postgres/migration-stamping.ts
View File

@@ -49,12 +49,63 @@ export interface StampMigratedProjectRowsResult {
readonly stamped: boolean;
}
export interface ProjectPartitionOwnership {
readonly fallbackProjectId: string;
readonly registeredProjectId: string;
readonly fallbackOwnedRows: boolean;
readonly registeredOwnedRows: boolean;
readonly ownershipByRelation: Record<string, { fallback: boolean; registered: boolean }>;
}
export type ProjectPartitionRekeyReason =
| "unreplaced-fk-dependent"
| "unsupported-unique-metadata"
| "unsafe-fk-update-graph"
| "unique-violation"
| "unknown";
/** A fail-closed promotion result that also preserves the pre-write bind decision. */
export class ProjectPartitionRekeyError extends Error {
constructor(
readonly reason: ProjectPartitionRekeyReason,
readonly ownership: ProjectPartitionOwnership,
readonly details: string,
options?: ErrorOptions,
) {
super(`Project partition reconciliation refused (${reason}): ${details}`, options);
this.name = "ProjectPartitionRekeyError";
}
}
export function selectDegradedBindTarget(
ownership: ProjectPartitionOwnership | undefined,
): "fallback" | "registered" | "refuse" {
if (!ownership) return "refuse";
if (ownership.fallbackOwnedRows) return "fallback";
return ownership.registeredOwnedRows ? "registered" : "refuse";
}
type RekeyTarget = { schema: string; table: string };
type UniqueRule = { schema: string; table: string; name: string; columns: string[]; nullsNotDistinct: boolean; partial: boolean; expression: boolean };
type ForeignKey = { constraintId: string; constraintName: string; childSchema: string; childTable: string; parentSchema: string; parentTable: string; childColumns: string[]; parentColumns: string[]; deferrable: boolean; updateAction: string; deleteAction: string };
function quoteIdentifier(identifier: string): string {
return `"${identifier.replaceAll('"', '""')}"`;
}
function relationName(target: RekeyTarget): string { return `${quoteIdentifier(target.schema)}.${quoteIdentifier(target.table)}`; }
function relationKey(target: RekeyTarget): string { return `${target.schema}.${target.table}`; }
// Project identifiers originate from central storage, but quote them here because catalog-driven SQL cannot bind identifiers and must remain injection-safe.
function quoteLiteral(value: string): string { return `'${value.replaceAll("'", "''")}'`; }
/**
* FNXC:ProjectIdentityPromotion 2026-07-14-14:08:
* An unregistered project first runs under a stable path-derived partition. If
* central registration later assigns its canonical ID, atomically promote all
* project and archive rows plus migration bookkeeping so the next boot cannot
* strand the project in the fallback partition or repeat its SQLite cutover.
* FNXC:ProjectPartitionMerge 2026-07-20-12:00:
* A registered project can legitimately have older path-fallback rows. Promotion is a single catalog-driven merge, not a NOT EXISTS skip: fallback rows win conflicts so registration scaffolding cannot discard operator settings. Normal UNIQUE keys require non-NULL equality while NULLS NOT DISTINCT keys use IS NOT DISTINCT FROM.
*
* FNXC:ProjectPartitionMerge 2026-07-20-12:00:
* Before writes, reject metadata we cannot model and unsafe update FK edges. Registered rows are deleted only for a proven unique conflict; any inbound registered dependent makes that delete fail closed regardless of CASCADE/SET NULL/SET DEFAULT, rather than using an FK action as data cleanup. The typed error retains pre-transaction ownership so startup can safely bind fallback after rollback.
*
* FNXC:ProjectPartitionMerge 2026-07-20-12:30:
* Evaluate each catalog FK constraint independently: aggregating separate edges can miss a dependent and permit a collateral cascade. DEFERRABLE only permits deferred checks; ON UPDATE SET NULL/SET DEFAULT remain unsafe because they can mutate retained children during promotion.
*/
export async function rekeyFallbackProjectPartition(
db: MigrationDb,
@@ -64,83 +115,204 @@ export async function rekeyFallbackProjectPartition(
if (!fallbackProjectId || fallbackProjectId === registeredProjectId) return false;
return db.transaction(async (tx) => {
// This remains the first statement: migration-stamping-lock-order.test.ts enforces it.
await acquireSqliteMigrationStateLock(tx);
const tables = (await tx.execute(sql`
SELECT table_name
const targets = (await tx.execute(sql`
SELECT table_schema AS schema, table_name AS table
FROM information_schema.columns
WHERE table_schema = 'project' AND column_name = 'project_id'
ORDER BY table_name
`)) as unknown as Array<{ table_name: string }>;
/*
FNXC:ProjectIdentityPromotion 2026-07-14-18:58:
Fallback ownership can exist only in satellite tables such as agents, reports, or mission state. Inspect every project-owned table before deciding promotion is a no-op; task/archive-only detection stranded those partitions after central registration.
*/
let ownsProjectRows = false;
for (const { table_name: tableName } of tables) {
const rows = (await tx.execute(sql`
SELECT EXISTS (
SELECT 1 FROM ${sql.identifier("project")}.${sql.identifier(tableName)}
WHERE project_id = ${fallbackProjectId}
) AS found
`)) as unknown as Array<{ found: boolean }>;
if (rows[0]?.found) {
ownsProjectRows = true;
break;
UNION ALL SELECT 'archive', 'archived_tasks'
WHERE to_regclass('archive.archived_tasks') IS NOT NULL
ORDER BY 1, 2
`)) as unknown as RekeyTarget[];
const ownershipByRelation: Record<string, { fallback: boolean; registered: boolean }> = {};
for (const target of targets) {
const rows = (await tx.execute(sql.raw(`SELECT EXISTS (SELECT 1 FROM ${relationName(target)} WHERE project_id = ${quoteLiteral(fallbackProjectId)}) AS fallback, EXISTS (SELECT 1 FROM ${relationName(target)} WHERE project_id = ${quoteLiteral(registeredProjectId)}) AS registered`))) as unknown as Array<{ fallback: boolean; registered: boolean }>;
ownershipByRelation[relationKey(target)] = rows[0] ?? { fallback: false, registered: false };
}
const stateExists = (await tx.execute(sql`SELECT to_regclass('public.fusion_sqlite_migrations') IS NOT NULL AS exists`)) as unknown as Array<{ exists: boolean }>;
let fallbackMigration = false;
let registeredMigration = false;
if (stateExists[0]?.exists) {
const rows = (await tx.execute(sql`SELECT EXISTS (SELECT 1 FROM public.fusion_sqlite_migrations WHERE project_id = ${fallbackProjectId}) AS fallback, EXISTS (SELECT 1 FROM public.fusion_sqlite_migrations WHERE project_id = ${registeredProjectId}) AS registered`)) as unknown as Array<{ fallback: boolean; registered: boolean }>;
fallbackMigration = rows[0]?.fallback === true;
registeredMigration = rows[0]?.registered === true;
ownershipByRelation["public.fusion_sqlite_migrations"] = { fallback: fallbackMigration, registered: registeredMigration };
}
const ownership: ProjectPartitionOwnership = {
fallbackProjectId, registeredProjectId,
fallbackOwnedRows: fallbackMigration || Object.values(ownershipByRelation).some((row) => row.fallback),
registeredOwnedRows: registeredMigration || Object.values(ownershipByRelation).some((row) => row.registered),
ownershipByRelation,
};
if (!ownership.fallbackOwnedRows) return false;
const uniqueRules = (await tx.execute(sql`
SELECT ns.nspname AS schema, cls.relname AS table, idx.relname AS name,
array_agg(att.attname ORDER BY ord.n) AS columns,
COALESCE(ind.indnullsnotdistinct, false) AS "nullsNotDistinct",
ind.indpred IS NOT NULL AS partial, ind.indexprs IS NOT NULL AS expression
FROM pg_index ind
JOIN pg_class cls ON cls.oid = ind.indrelid
JOIN pg_namespace ns ON ns.oid = cls.relnamespace
JOIN pg_class idx ON idx.oid = ind.indexrelid
-- FNXC:ProjectPartitionMerge 2026-07-20-12:45: indkey also contains non-key INCLUDE attributes; only indnkeyatts participate in a unique conflict after project_id is rewritten.
JOIN LATERAL unnest(ind.indkey::smallint[]) WITH ORDINALITY ord(attnum, n)
ON ord.n <= ind.indnkeyatts
LEFT JOIN pg_attribute att ON att.attrelid = cls.oid AND att.attnum = ord.attnum
WHERE ind.indisunique
AND (ns.nspname = 'project' OR (ns.nspname = 'archive' AND cls.relname = 'archived_tasks'))
GROUP BY ns.nspname, cls.relname, idx.relname, ind.indnullsnotdistinct, ind.indpred, ind.indexprs
`)) as unknown as UniqueRule[];
const targetKeys = new Set(targets.map(relationKey));
const relevantRules = uniqueRules.filter((rule) => {
const rows = ownershipByRelation[`${rule.schema}.${rule.table}`];
/* FNXC:ProjectPartitionMerge 2026-07-20-12:45: A partial/expression rule can depend on project_id even when its key attrs do not. We cannot evaluate that predicate safely, so include it in the fail-closed metadata pass whenever fallback rows will be moved. */
return targetKeys.has(`${rule.schema}.${rule.table}`) && rows?.fallback
&& (rule.columns.includes("project_id") || rule.expression || rule.partial)
// Without a registered row this rewrite cannot introduce a second
// member of the index, so opaque predicates have no conflict to model.
&& rows.registered;
});
for (const rule of relevantRules) {
if (rule.expression || rule.partial || rule.columns.some((column) => !column)) {
throw new ProjectPartitionRekeyError("unsupported-unique-metadata", ownership, `${rule.schema}.${rule.table}.${rule.name}`);
}
}
if (!ownsProjectRows) {
const rows = (await tx.execute(sql`
SELECT EXISTS (
SELECT 1 FROM archive.archived_tasks WHERE project_id = ${fallbackProjectId}
) AS found
`)) as unknown as Array<{ found: boolean }>;
ownsProjectRows = rows[0]?.found === true;
const fks = (await tx.execute(sql`
SELECT con.oid::text AS "constraintId", con.conname AS "constraintName",
cns.nspname AS "childSchema", child.relname AS "childTable",
pns.nspname AS "parentSchema", parent.relname AS "parentTable",
array_agg(ca.attname ORDER BY keys.n) FILTER (WHERE ca.attname IS NOT NULL) AS "childColumns",
array_agg(pa.attname ORDER BY keys.n) FILTER (WHERE pa.attname IS NOT NULL) AS "parentColumns",
con.condeferrable AS deferrable, con.confupdtype AS "updateAction", con.confdeltype AS "deleteAction"
FROM pg_constraint con
JOIN pg_class child ON child.oid = con.conrelid JOIN pg_namespace cns ON cns.oid = child.relnamespace
JOIN pg_class parent ON parent.oid = con.confrelid JOIN pg_namespace pns ON pns.oid = parent.relnamespace
JOIN LATERAL unnest(con.conkey) WITH ORDINALITY keys(attnum, n) ON true
LEFT JOIN pg_attribute ca ON ca.attrelid = con.conrelid AND ca.attnum = keys.attnum
LEFT JOIN pg_attribute pa ON pa.attrelid = con.confrelid AND pa.attnum = con.confkey[keys.n]
WHERE con.contype = 'f'
GROUP BY con.oid, con.conname, cns.nspname, child.relname, pns.nspname, parent.relname, con.condeferrable, con.confupdtype, con.confdeltype
`)) as unknown as ForeignKey[];
const updateUnsafe = fks.find((fk) => {
const parentRows = ownershipByRelation[`${fk.parentSchema}.${fk.parentTable}`];
const childRows = ownershipByRelation[`${fk.childSchema}.${fk.childTable}`];
const rewritesFkIdentity = fk.parentColumns.includes("project_id")
|| (targetKeys.has(`${fk.childSchema}.${fk.childTable}`) && fk.childColumns.includes("project_id"));
// SET NULL/DEFAULT mutates retained children even when the check is deferred.
const mutatesChild = fk.updateAction === "n" || fk.updateAction === "d";
const childIsRekeyed = targetKeys.has(`${fk.childSchema}.${fk.childTable}`);
/* FNXC:ProjectPartitionMerge 2026-07-20-12:45: A deferred check only makes a two-sided rekey safe: an external child is never rewritten by this loop and would still reference the fallback key at commit. CASCADE is the only catalog action that can safely update it. */
const deferredExternalChild = fk.deferrable && !childIsRekeyed;
return targetKeys.has(`${fk.parentSchema}.${fk.parentTable}`)
&& (parentRows?.fallback || childRows?.fallback)
&& rewritesFkIdentity
&& (mutatesChild || deferredExternalChild || (!fk.deferrable && fk.updateAction !== "c"));
});
if (updateUnsafe) {
throw new ProjectPartitionRekeyError("unsafe-fk-update-graph", ownership, `${updateUnsafe.childSchema}.${updateUnsafe.childTable} -> ${updateUnsafe.parentSchema}.${updateUnsafe.parentTable}`);
}
const migrationState = (await tx.execute(sql`
SELECT to_regclass('public.fusion_sqlite_migrations') IS NOT NULL AS exists
`)) as unknown as Array<{ exists: boolean }>;
let ownsMigrationState = false;
if (migrationState[0]?.exists) {
const markerRows = (await tx.execute(sql`
SELECT EXISTS (
SELECT 1 FROM public.fusion_sqlite_migrations
WHERE project_id = ${fallbackProjectId}
) AS found
`)) as unknown as Array<{ found: boolean }>;
ownsMigrationState = markerRows[0]?.found === true;
}
if (!ownsProjectRows && !ownsMigrationState) return false;
await tx.execute(sql`SET CONSTRAINTS ALL DEFERRED`);
for (const { table_name: tableName } of tables) {
await tx.execute(sql`
UPDATE ${sql.identifier("project")}.${sql.identifier(tableName)}
SET project_id = ${registeredProjectId}
WHERE project_id = ${fallbackProjectId}
`);
await tx.execute(sql.raw(`CREATE TEMP TABLE fusion_rekey_conflict_candidates (
schema_name text NOT NULL, table_name text NOT NULL, row_id tid NOT NULL,
PRIMARY KEY (schema_name, table_name, row_id)
) ON COMMIT DROP`));
// Compute the complete delete plan before deleting a parent. A dependent is
// safe only when it is itself a registered conflict candidate; this is the
// proof that its fallback counterpart replaces it after promotion.
for (const rule of relevantRules) {
const target: RekeyTarget = { schema: rule.schema, table: rule.table };
const nonProjectColumns = rule.columns.filter((column) => column !== "project_id");
const predicate = nonProjectColumns.length === 0
? "TRUE"
: rule.nullsNotDistinct
? nonProjectColumns.map((column) => `reg.${quoteIdentifier(column)} IS NOT DISTINCT FROM fb.${quoteIdentifier(column)}`).join(" AND ")
: nonProjectColumns.map((column) => `reg.${quoteIdentifier(column)} IS NOT NULL AND fb.${quoteIdentifier(column)} IS NOT NULL AND reg.${quoteIdentifier(column)} = fb.${quoteIdentifier(column)}`).join(" AND ");
await tx.execute(sql.raw(`INSERT INTO fusion_rekey_conflict_candidates(schema_name, table_name, row_id)
SELECT ${quoteLiteral(rule.schema)}, ${quoteLiteral(rule.table)}, reg.ctid
FROM ${relationName(target)} reg
WHERE reg.project_id = ${quoteLiteral(registeredProjectId)}
AND EXISTS (SELECT 1 FROM ${relationName(target)} fb
WHERE fb.project_id = ${quoteLiteral(fallbackProjectId)} AND ${predicate})
ON CONFLICT DO NOTHING`));
}
await tx.execute(sql`
UPDATE archive.archived_tasks
SET project_id = ${registeredProjectId}
WHERE project_id = ${fallbackProjectId}
`);
if (migrationState[0]?.exists) {
await tx.execute(sql`
UPDATE public.fusion_sqlite_migrations
SET project_id = ${registeredProjectId}, updated_at = now()
WHERE project_id = ${fallbackProjectId}
`);
await tx.execute(sql`
INSERT INTO public.fusion_sqlite_migrations(migration_key, project_id, status, last_error, updated_at)
SELECT ${`project:${registeredProjectId}`}, ${registeredProjectId}, status, last_error, now()
FROM public.fusion_sqlite_migrations
WHERE migration_key = ${`project:${fallbackProjectId}`}
ON CONFLICT (migration_key) DO UPDATE
SET project_id = EXCLUDED.project_id,
status = EXCLUDED.status,
last_error = EXCLUDED.last_error,
updated_at = now()
`);
for (const fk of fks) {
const parentKey = `${fk.parentSchema}.${fk.parentTable}`;
if (!targetKeys.has(parentKey)) continue;
const childTarget = targetKeys.has(`${fk.childSchema}.${fk.childTable}`);
const join = fk.childColumns.map((column, index) => `child.${quoteIdentifier(column)} IS NOT DISTINCT FROM parent.${quoteIdentifier(fk.parentColumns[index]!)}`).join(" AND ");
const registeredChildScope = childTarget
? ` AND child.${quoteIdentifier("project_id")} = ${quoteLiteral(registeredProjectId)}`
: "";
const unreplaced = (await tx.execute(sql.raw(`SELECT EXISTS (
SELECT 1 FROM ${relationName({ schema: fk.childSchema, table: fk.childTable })} child
JOIN ${relationName({ schema: fk.parentSchema, table: fk.parentTable })} parent ON ${join}
JOIN fusion_rekey_conflict_candidates parent_candidate
ON parent_candidate.schema_name = ${quoteLiteral(fk.parentSchema)}
AND parent_candidate.table_name = ${quoteLiteral(fk.parentTable)}
AND parent_candidate.row_id = parent.ctid
WHERE 1 = 1${registeredChildScope}
AND NOT EXISTS (SELECT 1 FROM fusion_rekey_conflict_candidates child_candidate
WHERE child_candidate.schema_name = ${quoteLiteral(fk.childSchema)}
AND child_candidate.table_name = ${quoteLiteral(fk.childTable)}
AND child_candidate.row_id = child.ctid)
) AS found`))) as unknown as Array<{ found: boolean }>;
if (unreplaced[0]?.found) {
throw new ProjectPartitionRekeyError("unreplaced-fk-dependent", ownership, `${fk.constraintName || fk.constraintId}: ${fk.childSchema}.${fk.childTable} (${fk.deleteAction}) depends on ${fk.parentSchema}.${fk.parentTable}`);
}
}
/*
FNXC:ProjectPartitionMerge 2026-07-20-13:15:
Delete only the precomputed registered conflict set, children before parents.
This permits a fallback-wins parent/child merge when every removed child has
its own fallback replacement, while an unreplaced child aborts before any
ON DELETE CASCADE, SET NULL, or SET DEFAULT side effect can run.
*/
const pending = new Set(targets.map(relationKey));
while (pending.size > 0) {
const next = [...pending].find((key) => !fks.some((fk) =>
`${fk.parentSchema}.${fk.parentTable}` === key
&& pending.has(`${fk.childSchema}.${fk.childTable}`)
&& `${fk.childSchema}.${fk.childTable}` !== key,
));
// A non-deferrable delete cycle has no statement-safe child-first order.
const deferredCycle = !next && [...pending].every((key) => fks
.filter((fk) => `${fk.parentSchema}.${fk.parentTable}` === key && pending.has(`${fk.childSchema}.${fk.childTable}`))
.every((fk) => fk.deferrable));
if (!next && !deferredCycle) {
throw new ProjectPartitionRekeyError("unknown", ownership, "non-deferrable conflict-delete FK cycle");
}
const deleteKey = next ?? [...pending][0]!;
const [schema, table] = deleteKey.split(".");
await tx.execute(sql.raw(`DELETE FROM ${relationName({ schema: schema!, table: table! })} reg
USING fusion_rekey_conflict_candidates candidate
WHERE candidate.schema_name = ${quoteLiteral(schema!)}
AND candidate.table_name = ${quoteLiteral(table!)}
AND candidate.row_id = reg.ctid`));
pending.delete(deleteKey);
}
// ON UPDATE CASCADE changes child keys with its parent statement, so its
// parent must run first; deferred edges have no statement-order constraint.
const updatePending = new Map(targets.map((target) => [relationKey(target), target]));
while (updatePending.size > 0) {
const target = [...updatePending.values()].find((candidate) => !fks.some((fk) =>
`${fk.childSchema}.${fk.childTable}` === relationKey(candidate)
&& updatePending.has(`${fk.parentSchema}.${fk.parentTable}`)
&& fk.updateAction === "c",
));
const next = target ?? updatePending.values().next().value as RekeyTarget;
await tx.execute(sql.raw(`UPDATE ${relationName(next)} SET project_id = ${quoteLiteral(registeredProjectId)} WHERE project_id = ${quoteLiteral(fallbackProjectId)}`));
updatePending.delete(relationKey(next));
}
if (stateExists[0]?.exists) {
await tx.execute(sql`UPDATE public.fusion_sqlite_migrations SET project_id = ${registeredProjectId}, updated_at = now() WHERE project_id = ${fallbackProjectId}`);
await tx.execute(sql`INSERT INTO public.fusion_sqlite_migrations(migration_key, project_id, status, last_error, updated_at) SELECT ${`project:${registeredProjectId}`}, ${registeredProjectId}, status, last_error, now() FROM public.fusion_sqlite_migrations WHERE migration_key = ${`project:${fallbackProjectId}`} ON CONFLICT (migration_key) DO UPDATE SET project_id = EXCLUDED.project_id, status = EXCLUDED.status, last_error = EXCLUDED.last_error, updated_at = now()`);
}
return true;
});

49
packages/core/src/postgres/startup-factory.ts
View File

@@ -66,7 +66,9 @@ import {
import { runLoadedPluginSchemaInitHooks, type LoadedPluginSchemaContract } from "./plugin-schema-hook.js";
import {
lookupRegisteredProjectIdByPath,
ProjectPartitionRekeyError,
rekeyFallbackProjectPartition,
selectDegradedBindTarget,
stampMigratedProjectRows,
} from "./migration-stamping.js";
@@ -851,11 +853,22 @@ export async function createTaskStoreForBackend(
const fallbackProjectId = fallbackProjectIdForRoot(rootDir);
migrationProjectId ??= fallbackProjectId;
if (migrationProjectId !== fallbackProjectId) {
await rekeyFallbackProjectPartition(
connections.migration,
fallbackProjectId,
migrationProjectId,
);
try {
await rekeyFallbackProjectPartition(connections.migration, fallbackProjectId, migrationProjectId);
} catch (error) {
const target = error instanceof ProjectPartitionRekeyError
? selectDegradedBindTarget(error.ownership)
: "refuse";
if (target === "fallback") {
/* FNXC:ProjectPartitionMerge 2026-07-20-12:00: Failed promotion must carry the pre-transaction fallback binding through every first-boot migration consumer; logging while continuing on the registered scaffold hides user data. */
log.warn(`startup-factory: partition promotion failed for ${rootDir} (${fallbackProjectId} -> ${migrationProjectId}); binding fallback data for this session (${error instanceof ProjectPartitionRekeyError ? error.reason : "unknown"})`);
migrationProjectId = fallbackProjectId;
} else if (target === "registered") {
log.warn(`startup-factory: partition promotion failed for ${rootDir} (${fallbackProjectId} -> ${migrationProjectId}); fallback was empty, retaining registered binding`);
} else {
throw new Error(`startup-factory: refusing project ${rootDir}; project partition reconciliation failed for ${fallbackProjectId} -> ${migrationProjectId}`, { cause: error });
}
}
}
/*
FNXC:MultiProjectIsolation 2026-07-11:
@@ -1036,18 +1049,32 @@ export async function createTaskStoreForBackend(
Unregistered paths resolve to undefined and boot unbound, preserving legacy
single-project behavior.
*/
const resolvedProjectId = options.projectId
let resolvedProjectId = options.projectId
?? (rootDir
? (await lookupRegisteredProjectIdByPath(connections.migration, rootDir))
?? fallbackProjectIdForRoot(rootDir)
: undefined);
if (rootDir && resolvedProjectId) {
await rekeyFallbackProjectPartition(
connections.migration,
fallbackProjectIdForRoot(rootDir),
resolvedProjectId,
);
const fallbackProjectId = fallbackProjectIdForRoot(rootDir);
if (fallbackProjectId !== resolvedProjectId) {
try {
await rekeyFallbackProjectPartition(connections.migration, fallbackProjectId, resolvedProjectId);
} catch (error) {
const target = error instanceof ProjectPartitionRekeyError
? selectDegradedBindTarget(error.ownership)
: "refuse";
if (target === "fallback") {
/* FNXC:ProjectPartitionMerge 2026-07-20-12:00: A rolled-back merge has a reliable ownership snapshot on its typed error. Bind fallback rather than deepening a split by writing the registered scaffold. */
log.warn(`startup-factory: partition promotion failed for ${rootDir} (${fallbackProjectId} -> ${resolvedProjectId}); binding fallback data for this session (${error instanceof ProjectPartitionRekeyError ? error.reason : "unknown"})`);
resolvedProjectId = fallbackProjectId;
} else if (target === "registered") {
log.warn(`startup-factory: partition promotion failed for ${rootDir} (${fallbackProjectId} -> ${resolvedProjectId}); fallback was empty, retaining registered binding`);
} else {
throw new Error(`startup-factory: refusing project ${rootDir}; project partition reconciliation failed for ${fallbackProjectId} -> ${resolvedProjectId}`, { cause: error });
}
}
}
}
/*

3
packages/core/src/process-supervisor.ts
View File

@@ -16,6 +16,9 @@ scripts/dev-with-memory.mjs.
*/
export const FUSION_RESTART_EXIT_CODE = 86;
/* FNXC:ProjectPartitionMerge 2026-07-20-12:00: A classified unique-constraint startup failure is deterministic, so supervised dashboard boot must stop once rather than consume the crash-restart budget. */
export const FUSION_NON_RETRYABLE_EXIT_CODE = 87;
const DEFAULT_KILL_GRACE_MS = 2_000;
const DEFAULT_MAX_LIFETIME_MS = 600_000;
const MAX_KILL_WAIT_MS = 1_000;

7
packages/core/src/task-store/async-persistence.ts
View File

@@ -33,6 +33,7 @@ import { and, Column, eq, is, isNull, sql, type SQL } from "drizzle-orm";
import type { PgColumn } from "drizzle-orm/pg-core";
import * as schema from "../postgres/schema/index.js";
import type { AsyncDataLayer, DbTransaction } from "../postgres/data-layer.js";
import { isPostgresUniqueError } from "../postgres-errors.js";
import { taskProjectScope } from "../postgres/data-layer.js";
import {
TASK_COLUMN_DESCRIPTORS,
@@ -491,9 +492,5 @@ export function isTaskIdConflictError(error: unknown): boolean {
if (/SQLITE_CONSTRAINT|UNIQUE constraint failed: tasks\.(id|proposalClaimId)|PRIMARY KEY constraint failed: tasks\.id/i.test(message)) {
return true;
}
// PostgreSQL unique_violation (23505). The code may be on the error directly
// (raw postgres.js) or on the `cause` (Drizzle wraps postgres errors).
const directCode = (error as { code?: string } | null)?.code;
const causeCode = (error as { cause?: { code?: string } } | null)?.cause?.code;
return directCode === "23505" || causeCode === "23505";
return isPostgresUniqueError(error);
}