From 969fce7aa10832dc345aaa044c59ae508acd12ce Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 16 Jul 2026 05:33:55 -0700 Subject: [PATCH] FN-8047: migrate AgentStore multi-node tests to PostgreSQL Migrate multi-node AgentStore coverage to shared PostgreSQL-backed fixtures. - Make concurrent central claim insertion resolve unique-key races as checkout conflicts. - Rework claim and owning-node handoff tests to use shared async PostgreSQL layers. - Restore PostgreSQL-compatible tests from the quarantine ledger. Files changed: packages/core/src/async-central-db.ts | 9 ++- .../cross-node-claim-mutex.integration.test.ts | 72 ++++++++++--------- .../distributed-claim-mutex.integration.test.ts | 27 +++---- .../owning-node-handoff.integration.test.ts | 41 +++++------ .../__tests__/reliability-interactions/_helpers.ts | 83 ++++++++++++++++++++-- .../multi-node-claim-mutex-interactions.test.ts | 28 +++----- .../owning-node-unavailable-interactions.test.ts | 36 +++++----- packages/engine/vitest.config.ts | 8 +-- scripts/lib/test-quarantine.json | 25 ------- 9 files changed, 180 insertions(+), 149 deletions(-) Fusion-Task-Id: FN-8047 Fusion-Task-Lineage: 3b7ee21e-0190-4364-a0cb-88aac5e2e1a3 Co-authored-by: Fusion (runfusion.ai) --- packages/core/src/async-central-db.ts | 9 +- ...cross-node-claim-mutex.integration.test.ts | 66 ++++++++------- ...istributed-claim-mutex.integration.test.ts | 27 +++--- .../owning-node-handoff.integration.test.ts | 41 ++++----- .../reliability-interactions/_helpers.ts | 83 +++++++++++++++++-- ...ulti-node-claim-mutex-interactions.test.ts | 28 +++---- ...ning-node-unavailable-interactions.test.ts | 36 ++++---- packages/engine/vitest.config.ts | 8 +- scripts/lib/test-quarantine.json | 25 ------ 9 files changed, 177 insertions(+), 146 deletions(-) diff --git a/packages/core/src/async-central-db.ts b/packages/core/src/async-central-db.ts index 0212100181..f3775737c5 100644 --- a/packages/core/src/async-central-db.ts +++ b/packages/core/src/async-central-db.ts @@ -172,6 +172,10 @@ export async function tryClaimTask( const now = input.renewedAt; if (!existing) { + /* + FNXC:AsyncCentralClaims 2026-07-16-10:55: + FN-8047 requires concurrent first claims from separate nodes to produce one winner and a normal conflict for the loser. PostgreSQL transactions can both observe an absent row, so make the unique-key collision non-throwing and classify the persisted winner below instead of leaking a database constraint error through AgentStore checkout. + */ await tx.insert(schema.central.taskClaims).values({ projectId: input.projectId, taskId: input.taskId, @@ -182,11 +186,14 @@ export async function tryClaimTask( leaseRenewedAt: now, createdAt: now, updatedAt: now, - }); + }).onConflictDoNothing(); const claim = await getTaskClaim(tx, input.projectId, input.taskId); if (!claim) { throw new Error("Task claim insert succeeded but row could not be read back"); } + if (claim.ownerNodeId !== input.nodeId || claim.ownerAgentId !== input.agentId) { + return { ok: false, reason: "conflict", current: claim }; + } return { ok: true, claim }; } diff --git a/packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts b/packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts index 8cbafb96a1..7445136abb 100644 --- a/packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts +++ b/packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts @@ -1,35 +1,43 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { mkdtempSync } from "node:fs"; -import { rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; -import { AgentStore, CheckoutConflictError, TaskStore, createCentralDatabase, type CentralDatabase } from "@fusion/core"; +import { AgentStore, AsyncCentralClaimStore, CheckoutConflictError, TaskStore } from "@fusion/core"; +import { pgDescribe } from "../../../core/src/__test-utils__/pg-test-harness.js"; +import { createPgLayer, hasPg, makePgAgentStore, makePgTaskStore } from "./reliability-interactions/_helpers.js"; -function makeTmpDir(): string { - return mkdtempSync(join(tmpdir(), "fn-cross-node-claim-test-")); -} +const pgIt = hasPg ? pgDescribe : describe.skip; -describe("cross-node claim mutex integration", () => { - let rootDir: string; - let globalDir: string; +pgIt("cross-node claim mutex integration", () => { let taskStore: TaskStore; - let centralDb: CentralDatabase; + let centralClaimStore: AsyncCentralClaimStore; let storeA: AgentStore; let storeB: AgentStore; + let cleanupTaskStore: (() => Promise) | undefined; + let cleanupCentralLayer: (() => Promise) | undefined; let agentA: string; let agentB: string; let taskId: string; beforeEach(async () => { - rootDir = makeTmpDir(); - globalDir = join(rootDir, ".fusion-global"); - taskStore = new TaskStore(rootDir, globalDir); - await taskStore.init(); - centralDb = createCentralDatabase(globalDir); - centralDb.init(); + const taskFixture = await makePgTaskStore(); + const centralFixture = await createPgLayer(); + taskStore = taskFixture.store; + cleanupTaskStore = taskFixture.cleanup; + cleanupCentralLayer = centralFixture.cleanup; + centralClaimStore = new AsyncCentralClaimStore(centralFixture.layer); - storeA = new AgentStore({ rootDir, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-a" }); - storeB = new AgentStore({ rootDir, taskStore, claimStore: centralDb, projectId: "P-1", nodeId: "node-b" }); + storeA = makePgAgentStore({ + taskStore, + layer: taskFixture.layer, + claimStore: centralClaimStore, + projectId: "P-1", + nodeId: "node-a", + }); + storeB = makePgAgentStore({ + taskStore, + layer: taskFixture.layer, + claimStore: centralClaimStore, + projectId: "P-1", + nodeId: "node-b", + }); await storeA.init(); await storeB.init(); @@ -41,13 +49,12 @@ describe("cross-node claim mutex integration", () => { afterEach(async () => { storeA?.close(); storeB?.close(); - taskStore?.close(); - centralDb?.close(); - await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + await cleanupTaskStore?.(); + await cleanupCentralLayer?.(); }); it("allows one winner per race and bumps epoch once per successful ownership acquisition", async () => { - const originalTryClaim = centralDb.tryClaimTask.bind(centralDb); + const originalTryClaim = centralClaimStore.tryClaimTask.bind(centralClaimStore); const installBarrier = () => { let waiters = 0; let releaseBarrier: (() => void) | undefined; @@ -55,13 +62,14 @@ describe("cross-node claim mutex integration", () => { releaseBarrier = resolve; }); - centralDb.tryClaimTask = (((input: Parameters[0]) => { + centralClaimStore.tryClaimTask = async (input: Parameters[0]) => { waiters += 1; if (waiters === 2) { releaseBarrier?.(); } - return barrier.then(() => originalTryClaim(input)); - }) as unknown) as CentralDatabase["tryClaimTask"]; + await barrier; + return originalTryClaim(input); + }; }; installBarrier(); @@ -79,7 +87,7 @@ describe("cross-node claim mutex integration", () => { const winner = fulfilled[0].value; expect(rejected[0].reason.currentHolderId).toBe(winner.checkedOutBy); - expect(centralDb.getTaskClaim("P-1", taskId)?.leaseEpoch).toBe(1); + expect((await centralClaimStore.getTaskClaim("P-1", taskId))?.leaseEpoch).toBe(1); expect(winner.checkoutLeaseEpoch).toBe(1); expect(["node-a", "node-b"]).toContain(winner.checkoutNodeId); @@ -97,6 +105,6 @@ describe("cross-node claim mutex integration", () => { expect(fulfilled2).toHaveLength(1); expect(rejected2).toHaveLength(1); expect(rejected2[0].reason).toBeInstanceOf(CheckoutConflictError); - expect(centralDb.getTaskClaim("P-1", taskId)?.leaseEpoch).toBe(1); + expect((await centralClaimStore.getTaskClaim("P-1", taskId))?.leaseEpoch).toBe(1); }); }); diff --git a/packages/engine/src/__tests__/distributed-claim-mutex.integration.test.ts b/packages/engine/src/__tests__/distributed-claim-mutex.integration.test.ts index 61d913fdea..448552e8eb 100644 --- a/packages/engine/src/__tests__/distributed-claim-mutex.integration.test.ts +++ b/packages/engine/src/__tests__/distributed-claim-mutex.integration.test.ts @@ -1,29 +1,23 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { mkdtempSync } from "node:fs"; -import { rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; import { AgentStore, CheckoutConflictError, TaskStore } from "@fusion/core"; +import { pgDescribe } from "../../../core/src/__test-utils__/pg-test-harness.js"; +import { hasPg, makePgAgentStore, makePgTaskStore } from "./reliability-interactions/_helpers.js"; -function makeTmpDir(): string { - return mkdtempSync(join(tmpdir(), "fn-distributed-claim-test-")); -} +const pgIt = hasPg ? pgDescribe : describe.skip; -describe("distributed claim mutex integration", () => { - let rootDir: string; +pgIt("distributed claim mutex integration", () => { let taskStore: TaskStore; let agentStore: AgentStore; + let cleanup: (() => Promise) | undefined; let winnerAgentId = ""; - let globalDir = ""; let loserAgentId = ""; let taskId = ""; beforeEach(async () => { - rootDir = makeTmpDir(); - globalDir = join(rootDir, ".fusion-global"); - taskStore = new TaskStore(rootDir, globalDir); - await taskStore.init(); - agentStore = new AgentStore({ rootDir, taskStore }); + const fixture = await makePgTaskStore(); + taskStore = fixture.store; + cleanup = fixture.cleanup; + agentStore = makePgAgentStore({ taskStore, layer: fixture.layer }); await agentStore.init(); winnerAgentId = (await agentStore.createAgent({ name: "winner", role: "executor" })).id; @@ -33,8 +27,7 @@ describe("distributed claim mutex integration", () => { afterEach(async () => { agentStore?.close(); - taskStore?.close(); - await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + await cleanup?.(); }); it("allows exactly one concurrent claimant and supports retry after release", async () => { diff --git a/packages/engine/src/__tests__/owning-node-handoff.integration.test.ts b/packages/engine/src/__tests__/owning-node-handoff.integration.test.ts index 78306bc97d..0768bf06ae 100644 --- a/packages/engine/src/__tests__/owning-node-handoff.integration.test.ts +++ b/packages/engine/src/__tests__/owning-node-handoff.integration.test.ts @@ -1,33 +1,26 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { mkdtempSync } from "node:fs"; -import { rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; import { TaskStore, type OwningNodeHandoffPolicy } from "@fusion/core"; +import { pgDescribe } from "../../../core/src/__test-utils__/pg-test-harness.js"; +import { hasPg, makePgTaskStore } from "./reliability-interactions/_helpers.js"; import { MeshLeaseManager } from "../mesh-lease-manager.js"; import type { NodeHealthMonitor } from "../node-health-monitor.js"; -function makeTmpDir(): string { - return mkdtempSync(join(tmpdir(), "fn-owning-handoff-test-")); -} +const pgIt = hasPg ? pgDescribe : describe.skip; -describe("MeshLeaseManager owning-node handoff integration", () => { - let rootDir: string; - let globalDir: string; +pgIt("MeshLeaseManager owning-node handoff integration", () => { let taskStore: TaskStore; + let cleanup: (() => Promise) | undefined; let taskId: string; beforeEach(async () => { - rootDir = makeTmpDir(); - globalDir = join(rootDir, ".fusion-global"); - taskStore = new TaskStore(rootDir, globalDir); - await taskStore.init(); + const fixture = await makePgTaskStore(); + taskStore = fixture.store; + cleanup = fixture.cleanup; taskId = (await taskStore.createTask({ description: "handoff" })).id; }); afterEach(async () => { - taskStore?.close(); - await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + await cleanup?.(); }); async function seedLease(ownerNodeId: string): Promise { @@ -54,20 +47,20 @@ describe("MeshLeaseManager owning-node handoff integration", () => { it("applies handoff policy matrix for peer-owned leases", async () => { await seedLease("node-peer"); - let baselineEventIds = new Set(taskStore.getRunAuditEvents({ taskId, limit: 200 }).map((event) => event.id)); + let baselineEventIds = new Set((await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).map((event) => event.id)); expect(await runCase("block", "node-peer")).toBe(false); let task = await taskStore.getTask(taskId); expect(task?.checkedOutBy).toBe("agent-1"); - let newEvents = taskStore.getRunAuditEvents({ taskId, limit: 200 }).filter((event) => !baselineEventIds.has(event.id)); + let newEvents = (await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).filter((event) => !baselineEventIds.has(event.id)); expect(newEvents.some((event) => event.mutationType === "node:handoff:parked" && event.metadata?.source === "mesh-lease.recover" && event.metadata?.decisionReason === "handoff_blocked_by_policy")).toBe(true); expect(newEvents.some((event) => event.mutationType === "node:lease:recovered")).toBe(false); await seedLease("node-peer"); - baselineEventIds = new Set(taskStore.getRunAuditEvents({ taskId, limit: 200 }).map((event) => event.id)); + baselineEventIds = new Set((await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).map((event) => event.id)); expect(await runCase("reassign-to-local", "node-peer")).toBe(true); task = await taskStore.getTask(taskId); expect(task?.checkedOutBy ?? null).toBeNull(); - newEvents = taskStore.getRunAuditEvents({ taskId, limit: 200 }).filter((event) => !baselineEventIds.has(event.id)); + newEvents = (await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).filter((event) => !baselineEventIds.has(event.id)); const localRecoveryEvent = newEvents.find((event) => event.mutationType === "node:lease:recovered"); expect(localRecoveryEvent).toBeTruthy(); expect(localRecoveryEvent?.metadata?.source).toBe("mesh-lease.recover"); @@ -75,11 +68,11 @@ describe("MeshLeaseManager owning-node handoff integration", () => { expect(String(localRecoveryEvent?.metadata?.recoveryReason ?? "")).toContain("test-owner-unavailable"); await seedLease("node-peer"); - baselineEventIds = new Set(taskStore.getRunAuditEvents({ taskId, limit: 200 }).map((event) => event.id)); + baselineEventIds = new Set((await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).map((event) => event.id)); expect(await runCase("reassign-any-healthy", "node-peer")).toBe(true); task = await taskStore.getTask(taskId); expect(task?.checkedOutBy ?? null).toBeNull(); - newEvents = taskStore.getRunAuditEvents({ taskId, limit: 200 }).filter((event) => !baselineEventIds.has(event.id)); + newEvents = (await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).filter((event) => !baselineEventIds.has(event.id)); const anyRecoveryEvent = newEvents.find((event) => event.mutationType === "node:lease:recovered"); expect(anyRecoveryEvent).toBeTruthy(); expect(anyRecoveryEvent?.metadata?.source).toBe("mesh-lease.recover"); @@ -90,12 +83,12 @@ describe("MeshLeaseManager owning-node handoff integration", () => { it("recovers self-owned leases regardless of policy", async () => { for (const policy of ["block", "reassign-to-local", "reassign-any-healthy"] as const) { await seedLease("node-local"); - const baselineEventIds = new Set(taskStore.getRunAuditEvents({ taskId, limit: 200 }).map((event) => event.id)); + const baselineEventIds = new Set((await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).map((event) => event.id)); const recovered = await runCase(policy, "node-local"); expect(recovered).toBe(true); const task = await taskStore.getTask(taskId); expect(task?.checkedOutBy ?? null).toBeNull(); - const newEvents = taskStore.getRunAuditEvents({ taskId, limit: 200 }).filter((event) => !baselineEventIds.has(event.id)); + const newEvents = (await taskStore.getRunAuditEventsAsync({ taskId, limit: 200 })).filter((event) => !baselineEventIds.has(event.id)); const recoveryEvents = newEvents.filter((event) => event.mutationType === "node:lease:recovered"); expect(recoveryEvents).toHaveLength(1); expect(recoveryEvents[0]?.metadata?.source).toBe("mesh-lease.recover"); diff --git a/packages/engine/src/__tests__/reliability-interactions/_helpers.ts b/packages/engine/src/__tests__/reliability-interactions/_helpers.ts index 21ce90e9b7..a61362a18a 100644 --- a/packages/engine/src/__tests__/reliability-interactions/_helpers.ts +++ b/packages/engine/src/__tests__/reliability-interactions/_helpers.ts @@ -4,8 +4,8 @@ import { join } from "node:path"; import { execSync, spawnSync, exec } from "node:child_process"; import { Worker } from "node:worker_threads"; import { - DEFAULT_SETTINGS, TaskStore, type Settings, type Task, - type AsyncDataLayer, type ResolvedBackend, + AgentStore, DEFAULT_SETTINGS, TaskStore, type Settings, type Task, + type AsyncDataLayer, type CentralClaimStore, type ResolvedBackend, createConnectionSetFromUrl, applySchemaBaseline, createAsyncDataLayer, } from "@fusion/core"; import { aiMergeTask } from "../../merger.js"; @@ -131,7 +131,18 @@ function adminExecAsync(statement: string, timeoutMs = 15_000): Promise { let relDbCounter = 0; -async function createPgLayer(): Promise<{ layer: AsyncDataLayer; dbName: string }> { +export type PgLayerFixture = { + layer: AsyncDataLayer; + dbName: string; + cleanup: () => Promise; +}; + +/** + * Create one isolated PostgreSQL schema layer for a reliability test. + * Callers must use {@link hasPg} before invoking this helper because DDL uses + * the `psql` binary as well as a TCP-reachable PostgreSQL server. + */ +export async function createPgLayer(): Promise { relDbCounter += 1; const dbName = `fusion_rel_${process.pid}_${relDbCounter}_${Math.random().toString(36).slice(2, 8)}`; try { @@ -152,7 +163,65 @@ async function createPgLayer(): Promise<{ layer: AsyncDataLayer; dbName: string await schemaConn.close(); const connections = await createConnectionSetFromUrl(backend, { poolMax: 5, connectTimeoutSeconds: 5 }); const layer = createAsyncDataLayer(connections); - return { layer, dbName }; + return { + layer, + dbName, + cleanup: async () => { + try { await layer.close(); } catch { /* best-effort */ } + try { await adminExecAsync(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* best-effort */ } + }, + }; +} + +/* +FNXC:PgMigrationQuarantine 2026-07-16-10:30: +VAL-REMOVAL-005 removed AgentStore's SQLite runtime path, so multi-node claim and handoff tests must construct TaskStore and every sibling AgentStore with one shared AsyncDataLayer. Reliability callers gate with hasGit && hasPg; integration callers compose hasPg ? pgDescribe : describe.skip because DDL requires both reachable PostgreSQL and psql, but integration tests do not require Git. +*/ +export async function makePgTaskStore(): Promise<{ + rootDir: string; + store: TaskStore; + layer: AsyncDataLayer; + cleanup: () => Promise; +}> { + const rootDir = await mkdtemp(join(reliabilityTestTempParent(), "fusion-pg-store-")); + const pg = await createPgLayer(); + const store = new TaskStore(rootDir, undefined, { asyncLayer: pg.layer }); + await store.init(); + return { + rootDir, + store, + layer: pg.layer, + cleanup: async () => { + await store.close(); + await pg.cleanup(); + await rm(rootDir, { recursive: true, force: true }); + }, + }; +} + +/** + * Construct one backend-mode AgentStore for each AgentStore instance a test + * already needs. Two-store tests call this twice with the same taskStore/layer. + */ +export function makePgAgentStore(input: { + taskStore: TaskStore; + layer: AsyncDataLayer; + rootDir?: string; + claimStore?: CentralClaimStore; + projectId?: string; + nodeId?: string; +}): AgentStore { + if (input.taskStore.getAsyncLayer() !== input.layer) { + throw new Error("makePgAgentStore requires the TaskStore's shared asyncLayer"); + } + return new AgentStore({ + rootDir: input.rootDir ?? input.taskStore.getRootDir(), + taskStore: input.taskStore, + asyncLayer: input.layer, + claimStore: input.claimStore, + projectId: input.projectId, + nodeId: input.nodeId, + }); } export type ReliabilityFixture = { @@ -194,7 +263,8 @@ export async function makeReliabilityFixture(input: { git(rootDir, 'git commit -m "chore: init"'); await mkdir(join(rootDir, ".fusion"), { recursive: true }); - const { layer, dbName } = await createPgLayer(); + const pg = await createPgLayer(); + const { layer } = pg; const store = new TaskStore(rootDir, undefined, { asyncLayer: layer }); await store.init(); const settings: Settings = { @@ -232,8 +302,7 @@ export async function makeReliabilityFixture(input: { cleanup: async () => { manager.stop(); await store.close(); - try { await layer.close(); } catch { /* best-effort */ } - try { await adminExecAsync(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* best-effort */ } + await pg.cleanup(); await rm(rootDir, { recursive: true, force: true }); await rm(worktreeRoot, { recursive: true, force: true }); }, diff --git a/packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts b/packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts index 1d1386d50e..a9dd36987f 100644 --- a/packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts @@ -1,18 +1,12 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { mkdtempSync } from "node:fs"; -import { rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; import { AgentStore, CheckoutConflictError, TaskStore } from "@fusion/core"; +import { hasGit, hasPg, makePgAgentStore, makePgTaskStore } from "./_helpers.js"; -function makeTmpDir(): string { - return mkdtempSync(join(tmpdir(), "fn-reliability-claim-mutex-")); -} +const describeIfGit = hasGit && hasPg ? describe : describe.skip; -describe("reliability interactions: multi-node claim mutex", () => { - let rootDir: string; - let globalDir: string; +describeIfGit("reliability interactions: multi-node claim mutex", () => { let taskStore: TaskStore; + let cleanup: (() => Promise) | undefined; let agentStoreA: AgentStore; let agentStoreB: AgentStore; let taskId: string; @@ -20,13 +14,12 @@ describe("reliability interactions: multi-node claim mutex", () => { let agentB: string; beforeEach(async () => { - rootDir = makeTmpDir(); - globalDir = join(rootDir, ".fusion-global"); - taskStore = new TaskStore(rootDir, globalDir); - await taskStore.init(); + const fixture = await makePgTaskStore(); + taskStore = fixture.store; + cleanup = fixture.cleanup; - agentStoreA = new AgentStore({ rootDir, taskStore }); - agentStoreB = new AgentStore({ rootDir, taskStore }); + agentStoreA = makePgAgentStore({ taskStore, layer: fixture.layer }); + agentStoreB = makePgAgentStore({ taskStore, layer: fixture.layer }); await agentStoreA.init(); await agentStoreB.init(); @@ -38,8 +31,7 @@ describe("reliability interactions: multi-node claim mutex", () => { afterEach(async () => { agentStoreA?.close(); agentStoreB?.close(); - taskStore?.close(); - await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + await cleanup?.(); }); it("prevents split-brain, preserves renewal semantics, and keeps legacy conflict shape", async () => { diff --git a/packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts b/packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts index 50a594b781..84a071364d 100644 --- a/packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts +++ b/packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts @@ -1,13 +1,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { existsSync, mkdtempSync } from "node:fs"; -import { readFile } from "node:fs/promises"; -import { rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; +import { existsSync } from "node:fs"; import type { NodeStatus, OwningNodeHandoffPolicy, Task, TaskStore } from "@fusion/core"; import { TaskStore as CoreTaskStore } from "@fusion/core"; import { MeshLeaseManager } from "../../mesh-lease-manager.js"; import { Scheduler } from "../../scheduler.js"; +import { hasGit, hasPg, makePgTaskStore } from "./_helpers.js"; + +const describeIfGit = hasGit && hasPg ? describe : describe.skip; +const readFileState = vi.hoisted(() => ({ mockPromptRead: false })); vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); @@ -21,14 +21,12 @@ vi.mock("node:fs/promises", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, - readFile: vi.fn(), + readFile: async (...args: Parameters) => ( + readFileState.mockPromptRead ? "# Task\nFN-4813" : actual.readFile(...args) + ), }; }); -function makeTmpDir(): string { - return mkdtempSync(join(tmpdir(), "fn-owning-node-handoff-")); -} - function createMockTask(overrides: Partial = {}): Task { return { id: "FN-200", @@ -72,23 +70,21 @@ function createMockHealthMonitor(statusMap: Record { - let rootDir = ""; - let globalDir = ""; +describeIfGit("reliability interactions: owning-node unavailable handoff", () => { let taskStore: CoreTaskStore; + let cleanup: (() => Promise) | undefined; beforeEach(async () => { vi.mocked(existsSync).mockReturnValue(true); - vi.mocked(readFile).mockResolvedValue("# Task\nFN-4813"); - rootDir = makeTmpDir(); - globalDir = join(rootDir, ".fusion-global"); - taskStore = new CoreTaskStore(rootDir, globalDir); - await taskStore.init(); + const fixture = await makePgTaskStore(); + taskStore = fixture.store; + cleanup = fixture.cleanup; + readFileState.mockPromptRead = true; }); afterEach(async () => { - taskStore?.close(); - await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + readFileState.mockPromptRead = false; + await cleanup?.(); }); async function seedCheckedOutTask(overrides: Partial = {}): Promise { diff --git a/packages/engine/vitest.config.ts b/packages/engine/vitest.config.ts index aca155bb5c..fe57384b08 100644 --- a/packages/engine/vitest.config.ts +++ b/packages/engine/vitest.config.ts @@ -301,11 +301,8 @@ export default defineConfig({ // getDatabase, walCheckpoint) that throw/return-empty in backend mode, or have mock // drift from the async-satellite cutover. Quarantined on sight per AGENTS.md. "src/__tests__/backlog-pressure-reporter.test.ts", - "src/__tests__/cross-node-claim-mutex.integration.test.ts", - "src/__tests__/distributed-claim-mutex.integration.test.ts", "src/__tests__/mission-autopilot.test.ts", "src/__tests__/mission-factory-parity.integration.test.ts", - "src/__tests__/owning-node-handoff.integration.test.ts", "src/__tests__/planner-overseer-intervention-wiring.test.ts", "src/__tests__/project-engine.test.ts", "src/__tests__/self-healing.test.ts", @@ -379,8 +376,9 @@ export default defineConfig({ // (now PG-backed) but fail on sync SQLite APIs (getRunAuditEvents, getDatabase) that // return [] / throw in backend mode, or on mock drift from the async-satellite cutover. // Quarantined on sight per AGENTS.md; mirrored in scripts/lib/test-quarantine.json. - "src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts", - "src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts", + // FNXC:PgMigrationQuarantine 2026-07-16-10:45: + // FN-8047 restored multi-node claim and owning-node handoff coverage with shared PG-backed + // AgentStores and AsyncCentralClaimStore; their paired ledger entries are intentionally live. "src/__tests__/reliability-interactions/integration-worktree-state.test.ts", "src/__tests__/reliability-interactions/explicit-duplicate-marker-sweep.test.ts", "src/__tests__/reliability-interactions/self-defeating-dep-reconcile.test.ts", diff --git a/scripts/lib/test-quarantine.json b/scripts/lib/test-quarantine.json index 41324e627a..310ef10edf 100644 --- a/scripts/lib/test-quarantine.json +++ b/scripts/lib/test-quarantine.json @@ -11,16 +11,6 @@ "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", "quarantinedAt": "2026-07-14" }, - { - "file": "packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, - { - "file": "packages/engine/src/__tests__/distributed-claim-mutex.integration.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, { "file": "packages/engine/src/__tests__/mission-autopilot.test.ts", "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", @@ -31,11 +21,6 @@ "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", "quarantinedAt": "2026-07-14" }, - { - "file": "packages/engine/src/__tests__/owning-node-handoff.integration.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, { "file": "packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts", "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", @@ -126,16 +111,6 @@ "reason": "VAL-REMOVAL-005 PG migration: fails due to SQLite removal or sync API incompatibility in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", "quarantinedAt": "2026-07-14" }, - { - "file": "packages/engine/src/__tests__/reliability-interactions/multi-node-claim-mutex-interactions.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: reliability fixture uses PG-backed store but sync APIs (getRunAuditEvents, getDatabase) fail in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, - { - "file": "packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts", - "reason": "VAL-REMOVAL-005 PG migration: reliability fixture uses PG-backed store but sync APIs (getRunAuditEvents, getDatabase) fail in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.", - "quarantinedAt": "2026-07-14" - }, { "file": "packages/engine/src/__tests__/reliability-interactions/integration-worktree-state.test.ts", "reason": "VAL-REMOVAL-005 PG migration: reliability fixture uses PG-backed store but sync APIs (getRunAuditEvents, getDatabase) fail in backend mode. Failing run: https://github.com/Runfusion/Fusion/actions/runs/29344576232. Mirrored in packages/engine/vitest.config.ts.",