diff --git a/packages/core/src/__tests__/postgres/allocator-cross-project.test.ts b/packages/core/src/__tests__/postgres/allocator-cross-project.test.ts index a9a42730bb..cbba098308 100644 --- a/packages/core/src/__tests__/postgres/allocator-cross-project.test.ts +++ b/packages/core/src/__tests__/postgres/allocator-cross-project.test.ts @@ -10,13 +10,15 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; +import { it, expect, afterEach } from "vitest"; import { eq } from "drizzle-orm"; -import { execSync } from "node:child_process"; import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; import { createConnectionSetFromUrl, type PostgresConnections } from "../../postgres/connection.js"; import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import { + pgDescribe, + createTaskStoreForTest, +} from "../../__test-utils__/pg-test-harness.js"; import * as schema from "../../postgres/schema/index.js"; import { insertTaskRow } from "../../task-store/async/async-persistence.js"; import { @@ -25,30 +27,21 @@ import { } from "../../task-store/async/async-allocator.js"; import type { DistributedTaskIdAllocator } from "../../tasks/distributed-task-id.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - const SHARED_PREFIX = "KB"; -function uniqueDbName(): string { - return `fusion_allocxp_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - -// FNXC:PgTestAuthFix 2026-07-14-07:30: -// The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - +/* +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated the per-test database creation off hand-rolled CREATE DATABASE + +applySchemaBaseline (~3-4s of DDL per test) onto the harness's template-cloned +`createTaskStoreForTest`. This file KEEPS a private database per test because it opens +its own pair of project-bound, RLS-enforced runtime-role connection sets against that +database — the cross-project isolation those connections provide is the subject, and +the harness database is only the substrate. The harness TaskStore's unbound init rows +live outside the proj_a/proj_b partitions, so the RLS-scoped reads never see them. +Every assertion is unchanged. +*/ interface TestCtx { - dbName: string; + baseTeardown: () => Promise; connectionsA: PostgresConnections; connectionsB: PostgresConnections; layerA: AsyncDataLayer; @@ -58,27 +51,16 @@ interface TestCtx { } async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - + const base = await createTaskStoreForTest({ + prefix: "fusion_allocxp_test", + copyFromGolden: true, + }); const backend: ResolvedBackend = { mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, + runtimeUrl: base.testUrl, + migrationUrl: base.testUrl, migrationUrlOverridden: false, }; - const schemaConnections = await createConnectionSetFromUrl(backend, { - poolMax: 1, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); const connectionsA = await createConnectionSetFromUrl(backend, { poolMax: 5, @@ -96,7 +78,7 @@ async function setupCtx(): Promise { const layerB = createAsyncDataLayer(connectionsB, { projectId: "proj_b" }); const allocatorA = createAsyncDistributedTaskIdAllocator(layerA); const allocatorB = createAsyncDistributedTaskIdAllocator(layerB); - return { dbName, connectionsA, connectionsB, layerA, layerB, allocatorA, allocatorB }; + return { baseTeardown: base.teardown, connectionsA, connectionsB, layerA, layerB, allocatorA, allocatorB }; } async function teardownCtx(ctx: TestCtx | null): Promise { @@ -107,7 +89,7 @@ async function teardownCtx(ctx: TestCtx | null): Promise { // best-effort } try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); + await ctx.baseTeardown(); } catch { // best-effort } diff --git a/packages/core/src/__tests__/postgres/approval-request-lifecycle.pg.test.ts b/packages/core/src/__tests__/postgres/approval-request-lifecycle.pg.test.ts index 11c60ed346..fbe93b7732 100644 --- a/packages/core/src/__tests__/postgres/approval-request-lifecycle.pg.test.ts +++ b/packages/core/src/__tests__/postgres/approval-request-lifecycle.pg.test.ts @@ -1,55 +1,23 @@ -import { describe, it, expect, afterEach } from "vitest"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; - -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_sat_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (the approval-request transition guards are), and every assertion is +unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface StoreTestCtx { - dbName: string; layer: AsyncDataLayer; } -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* may not exist */ } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ databaseUrl: testUrl, databaseMigrationUrl: testUrl }); - const connections = await createConnectionSetFromUrl(backend, { poolMax: 3, connectTimeoutSeconds: 5 }); - await applySchemaBaseline(connections.migration); - const layer = createAsyncDataLayer(connections); - return { dbName, layer }; -} - -async function teardownCtx(ctx: StoreTestCtx | null): Promise { - if (!ctx) return; - try { await ctx.layer.close(); } catch { /* best-effort */ } - try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* best-effort */ } -} - /* FNXC:ApprovalLifecycleSecurity 2026-07-30-13:10 (ported from the deleted sync branch): These assertions arrived with the approval-hardening work as `approval-request-store-lifecycle.test.ts`, @@ -71,11 +39,18 @@ two concurrent transactions these tests do not create. The guard stays because t simply not what is verified here. Do not read a green run as proof of it. */ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { - let ctx: StoreTestCtx | null = null; - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_sat_test", }); + let ctx: StoreTestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); const REQUESTER = { actorId: "agent-1", actorType: "agent" as const, actorName: "Bot" }; const DECIDER = { actorId: "user-1", actorType: "user" as const, actorName: "Admin" }; @@ -91,7 +66,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { } it("a same-verdict replay is rejected as an invalid transition", async () => { - ctx = await setupCtx(); const store = await seed("apr-replay"); await store.decideApprovalRequest(ctx.layer, "apr-replay", "approved", { actor: DECIDER }); @@ -101,7 +75,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { }); it("a replay does not re-stamp decidedAt or append a duplicate audit event", async () => { - ctx = await setupCtx(); const store = await seed("apr-nodup"); const first = await store.decideApprovalRequest(ctx.layer, "apr-nodup", "approved", { actor: DECIDER }); const auditBefore = await store.getApprovalAuditHistory(ctx.layer.db, "apr-nodup"); @@ -116,7 +89,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { }); it("approve then deny is rejected — the first decision stands", async () => { - ctx = await setupCtx(); const store = await seed("apr-flip"); await store.decideApprovalRequest(ctx.layer, "apr-flip", "approved", { actor: DECIDER }); @@ -127,7 +99,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { }); it("deciding a request that does not exist reports not-found", async () => { - ctx = await setupCtx(); const store = await import("../../async-stores/async-approval-request-store.js"); await expect( @@ -136,7 +107,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { }); it("markCompleted on a still-pending request is rejected", async () => { - ctx = await setupCtx(); const store = await seed("apr-pending"); await expect( @@ -149,7 +119,6 @@ pgDescribe("approval request lifecycle security (PostgreSQL)", () => { The ownership check is the containment that matters: without it any caller who learned a request id could redeem someone else's approved grant. */ - ctx = await setupCtx(); const store = await seed("apr-owner"); await store.decideApprovalRequest(ctx.layer, "apr-owner", "approved", { actor: DECIDER }); diff --git a/packages/core/src/__tests__/postgres/archive-pagination.pg.test.ts b/packages/core/src/__tests__/postgres/archive-pagination.pg.test.ts index fe52791224..f7138fb197 100644 --- a/packages/core/src/__tests__/postgres/archive-pagination.pg.test.ts +++ b/packages/core/src/__tests__/postgres/archive-pagination.pg.test.ts @@ -9,9 +9,13 @@ * unreachable (FUSION_PG_TEST_SKIP=1) so the merge gate stays green. */ -import { describe, it, expect, afterEach } from "vitest"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import { getArchivedRowCount, listArchivedTaskEntriesPage, @@ -19,54 +23,17 @@ import { } from "../../async-stores/async-archive-db.js"; import type { ArchivedTaskEntry } from "../../types.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_archive_page_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (FN-7659 pagination ordering is), and every assertion is unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface Ctx { - dbName: string; layer: AsyncDataLayer; } -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* may not exist */ } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ databaseUrl: testUrl, databaseMigrationUrl: testUrl }); - const connections = await createConnectionSetFromUrl(backend, { poolMax: 3, connectTimeoutSeconds: 5 }); - await applySchemaBaseline(connections.migration); - const layer = createAsyncDataLayer(connections); - return { dbName, layer }; -} - -async function teardownCtx(ctx: Ctx | null): Promise { - if (!ctx) return; - try { await ctx.layer.close(); } catch { /* best-effort */ } - try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* best-effort */ } -} - function makeEntry(id: string, archivedAt: string): ArchivedTaskEntry { return { id, @@ -81,21 +48,25 @@ function makeEntry(id: string, archivedAt: string): ArchivedTaskEntry { } pgDescribe("archive pagination (PostgreSQL, FN-7659)", () => { - let ctx: Ctx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_archive_page_test", }); + let ctx: Ctx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("returns [] and total 0 for an empty archive", async () => { - ctx = await setupCtx(); expect(await listArchivedTaskEntriesPage(ctx.layer.db, 100, 0)).toEqual([]); expect(await getArchivedRowCount(ctx.layer.db)).toBe(0); }); it("orders results by archivedAt DESC (newest first)", async () => { - ctx = await setupCtx(); const base = Date.parse("2026-01-01T00:00:00.000Z"); for (let i = 0; i < 10; i++) { await upsertArchivedTask(ctx.layer.db, makeEntry(`FN-${i}`, new Date(base + i * 60_000).toISOString())); @@ -107,7 +78,6 @@ pgDescribe("archive pagination (PostgreSQL, FN-7659)", () => { }); it("windows correctly with LIMIT/OFFSET across page boundaries", async () => { - ctx = await setupCtx(); const base = Date.parse("2026-01-01T00:00:00.000Z"); const total = 250; for (let i = 0; i < total; i++) { @@ -137,7 +107,6 @@ pgDescribe("archive pagination (PostgreSQL, FN-7659)", () => { }); it("handles the exact page-boundary cases (total === 100 and 101)", async () => { - ctx = await setupCtx(); const base = Date.parse("2026-01-01T00:00:00.000Z"); for (let i = 0; i < 101; i++) { await upsertArchivedTask(ctx.layer.db, makeEntry(`FN-${i}`, new Date(base + i * 60_000).toISOString())); diff --git a/packages/core/src/__tests__/postgres/central-archive-secrets.test.ts b/packages/core/src/__tests__/postgres/central-archive-secrets.test.ts index b7eb5cce30..e072c6c326 100644 --- a/packages/core/src/__tests__/postgres/central-archive-secrets.test.ts +++ b/packages/core/src/__tests__/postgres/central-archive-secrets.test.ts @@ -27,60 +27,27 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach, vi } from "vitest"; -import { execSync } from "node:child_process"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll, vi } from "vitest"; import { randomBytes } from "node:crypto"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import type { ArchivedTaskEntry } from "../../types.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_cas_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test, and every assertion is unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; layer: AsyncDataLayer; } -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* may not exist */ } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ databaseUrl: testUrl, databaseMigrationUrl: testUrl }); - const connections = await createConnectionSetFromUrl(backend, { poolMax: 3, connectTimeoutSeconds: 5 }); - await applySchemaBaseline(connections.migration); - const layer = createAsyncDataLayer(connections); - return { dbName, layer }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { await ctx.layer.close(); } catch { /* best-effort */ } - try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* best-effort */ } -} - /** A fixed 32-byte master key provider for deterministic test crypto. */ function fixedMasterKeyProvider(key: Buffer = randomBytes(32)): () => Promise { return async () => Buffer.from(key); @@ -107,17 +74,22 @@ function sampleArchiveEntry(overrides: Partial = {}): Archive } pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-central-archive-db)", () => { - let ctx: TestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_cas_test", }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); // ── Central DB: task claims ── it("CentralDatabase: tryClaimTask creates a fresh claim, then getTaskClaim reads it back", async () => { - ctx = await setupCtx(); const { tryClaimTask, getTaskClaim } = await import("../../async-stores/async-central-db.js"); const now = new Date().toISOString(); @@ -143,7 +115,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("CentralDatabase: same-owner renewal requires matching expectedEpoch", async () => { - ctx = await setupCtx(); const { tryClaimTask } = await import("../../async-stores/async-central-db.js"); const now = () => new Date().toISOString(); @@ -171,7 +142,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("CentralDatabase: different-owner takeover requires matching expectedEpoch, else conflict", async () => { - ctx = await setupCtx(); const { tryClaimTask } = await import("../../async-stores/async-central-db.js"); const now = () => new Date().toISOString(); @@ -201,7 +171,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("CentralDatabase: renewTaskClaim and releaseTaskClaim honor ownership", async () => { - ctx = await setupCtx(); const { tryClaimTask, renewTaskClaim, releaseTaskClaim, getTaskClaim } = await import("../../async-stores/async-central-db.js"); const now = () => new Date().toISOString(); @@ -242,7 +211,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("CentralDatabase: renewTaskClaim returns not_found for an absent claim", async () => { - ctx = await setupCtx(); const { renewTaskClaim } = await import("../../async-stores/async-central-db.js"); const result = await renewTaskClaim(ctx.layer, { projectId: "proj-1", taskId: "FN-MISSING", nodeId: "node-a", agentId: "agent-1", @@ -255,7 +223,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen // ── Archive DB ── it("ArchiveDatabase: upsert → get → list → filterArchived → delete", async () => { - ctx = await setupCtx(); const { upsertArchivedTask, getArchivedTask, listArchivedTasks, filterArchived, deleteArchivedTask, getArchivedRowCount } = await import("../../async-stores/async-archive-db.js"); const entry = sampleArchiveEntry({ id: "FN-ARCH-1", title: "First archived", comments: [{ id: "c1", text: "note", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }] }); @@ -282,7 +249,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("ArchiveDatabase: upsert replaces an existing entry on conflict", async () => { - ctx = await setupCtx(); const { upsertArchivedTask, getArchivedTask } = await import("../../async-stores/async-archive-db.js"); const entry = sampleArchiveEntry({ id: "FN-ARCH-2", title: "v1" }); await upsertArchivedTask(ctx.layer.db, entry); @@ -296,7 +262,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("ArchiveDatabase: search matches tokens across title/description/comments", async () => { - ctx = await setupCtx(); const { upsertArchivedTask, searchArchivedTasks } = await import("../../async-stores/async-archive-db.js"); await upsertArchivedTask(ctx.layer.db, sampleArchiveEntry({ id: "FN-S1", title: "Postgres migration", description: "convert sqlite", comments: [] })); await upsertArchivedTask(ctx.layer.db, sampleArchiveEntry({ id: "FN-S2", title: "unrelated", description: "nothing here", comments: [{ id: "c", text: "mention postgres", author: "agent", createdAt: "2026-01-01T00:00:00.000Z" }] })); @@ -312,7 +277,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen // ── SecretsStore ── it("SecretsStore: create → get → list → update → reveal → delete for project scope", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); @@ -346,7 +310,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("SecretsStore: global scope routes to central.secrets_global", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); @@ -363,7 +326,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("SecretsStore: duplicate key throws duplicate-key (unique constraint)", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore, SecretsStoreError } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); @@ -375,7 +337,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("SecretsStore: re-encrypting a value on update round-trips", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); @@ -386,7 +347,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("SecretsStore: listEnvExportable returns project-overrides-global on key collision", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); @@ -399,7 +359,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen }); it("SecretsStore: deleting an absent secret throws not-found", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider()); await expect(store.deleteSecret("nope", "project")).rejects.toMatchObject({ code: "not-found" }); @@ -410,7 +369,6 @@ pgDescribe("PostgreSQL central-db / archive-db / secrets-store (U6 satellite-cen * Secret audit notifications must cover create, update, read, and delete without ever including plaintext, and a failing observer must remain non-blocking so audit infrastructure cannot break credential operations. */ it("SecretsStore: audit events omit plaintext and emitter failures stay non-blocking", async () => { - ctx = await setupCtx(); const { AsyncSecretsStore } = await import("../../async-stores/async-secrets-store.js"); const events: Array> = []; const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider(), { diff --git a/packages/core/src/__tests__/postgres/central-core-backend.test.ts b/packages/core/src/__tests__/postgres/central-core-backend.test.ts index e14d19d3a1..75003cbc60 100644 --- a/packages/core/src/__tests__/postgres/central-core-backend.test.ts +++ b/packages/core/src/__tests__/postgres/central-core-backend.test.ts @@ -19,99 +19,36 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; -import { execSync } from "node:child_process"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { CentralCore } from "../../central/central-core.js"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; - -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_cc_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (CentralCore's backend-mode delegation is). Each test still constructs +and inits its own CentralCore against the harness layer AFTER the per-test truncate, so +init-time bootstrap (default local node) is observed per test exactly as before. +CentralCore.close() never closes an injected layer, so the shared pool survives across +tests. Every assertion is unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; layer: AsyncDataLayer; central: CentralCore; globalDir: string; projectDirs: string[]; } -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - /* may not exist */ - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ - databaseUrl: testUrl, - databaseMigrationUrl: testUrl, - }); - const connections = await createConnectionSetFromUrl(backend, { - poolMax: 3, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(connections.migration); - const layer = createAsyncDataLayer(connections); - // Pass an explicit temp global dir so resolveGlobalDir() does not throw under VITEST. - const globalDir = mkdtempSync(join(tmpdir(), "kb-cc-pg-global-")); - const central = new CentralCore(globalDir, { asyncLayer: layer }); - await central.init(); - return { dbName, layer, central, globalDir, projectDirs: [] }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { - await ctx.central.close(); - } catch { - /* best-effort */ - } - try { - await ctx.layer.close(); - } catch { - /* best-effort */ - } - for (const dir of [...ctx.projectDirs, ctx.globalDir]) { - try { - rmSync(dir, { recursive: true, force: true }); - } catch { - /* best-effort */ - } - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); - } catch { - /* best-effort */ - } -} - function makeProjectDir(ctx: TestCtx, name: string): string { const dir = mkdtempSync(join(tmpdir(), `kb-cc-pg-${name}-`)); ctx.projectDirs.push(dir); @@ -119,22 +56,49 @@ function makeProjectDir(ctx: TestCtx, name: string): string { } pgDescribe("CentralCore backend mode (PostgreSQL)", () => { - let ctx: TestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_cc_test", }); + let ctx: TestCtx; + + async function setupCtx(): Promise { + const layer = h.layer(); + // Pass an explicit temp global dir so resolveGlobalDir() does not throw under VITEST. + const globalDir = mkdtempSync(join(tmpdir(), "kb-cc-pg-global-")); + const central = new CentralCore(globalDir, { asyncLayer: layer }); + await central.init(); + return { layer, central, globalDir, projectDirs: [] }; + } + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = await setupCtx(); + }); + afterEach(async () => { + try { + await ctx.central.close(); + } catch { + /* best-effort */ + } + for (const dir of [...ctx.projectDirs, ctx.globalDir]) { + try { + rmSync(dir, { recursive: true, force: true }); + } catch { + /* best-effort */ + } + } + await h.afterEach(); + }); + afterAll(h.afterAll); it("reports backendMode=true and does not construct SQLite CentralDatabase", async () => { - ctx = await setupCtx(); expect(ctx.central.backendMode).toBe(true); // getDatabasePath returns the logical global dir in backend mode (no SQLite file). expect(ctx.central.getDatabasePath()).not.toMatch(/fusion-central\.db$/); }); it("bootstraps a default local node on init", async () => { - ctx = await setupCtx(); const nodes = await ctx.central.listNodes(); const localNodes = nodes.filter((n) => n.type === "local"); expect(localNodes.length).toBe(1); @@ -142,7 +106,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("registers, reads, and lists a project through PostgreSQL", async () => { - ctx = await setupCtx(); const projectPath = makeProjectDir(ctx, "alpha"); const created = await ctx.central.registerProject({ name: "Alpha", @@ -168,7 +131,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("updates a project and reconciles stale statuses", async () => { - ctx = await setupCtx(); const projectPath = makeProjectDir(ctx, "beta"); const created = await ctx.central.registerProject({ name: "Beta", @@ -188,7 +150,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("registers and updates a node through PostgreSQL", async () => { - ctx = await setupCtx(); const node = await ctx.central.registerNode({ name: "remote-1", type: "remote", @@ -210,7 +171,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("logs and reads activity through PostgreSQL", async () => { - ctx = await setupCtx(); const projectPath = makeProjectDir(ctx, "gamma"); const project = await ctx.central.registerProject({ name: "Gamma", @@ -236,7 +196,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { it("records project-node path mappings through PostgreSQL", async () => { - ctx = await setupCtx(); const projectPath = makeProjectDir(ctx, "epsilon"); const project = await ctx.central.registerProject({ name: "Epsilon", @@ -255,7 +214,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("records and reads a mesh snapshot through PostgreSQL", async () => { - ctx = await setupCtx(); const nodes = await ctx.central.listNodes(); const localNode = nodes.find((n) => n.type === "local")!; // project_id is part of the composite PRIMARY KEY and therefore NOT NULL @@ -280,7 +238,6 @@ pgDescribe("CentralCore backend mode (PostgreSQL)", () => { }); it("attachBackendLayer transitions a legacy CentralCore into backend mode", async () => { - ctx = await setupCtx(); // Create a fresh legacy CentralCore (no asyncLayer) then attach the layer. const legacy = new CentralCore(ctx.globalDir); expect(legacy.backendMode).toBe(false); diff --git a/packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts b/packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts index 3bff381d2c..b8a134e6be 100644 --- a/packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts +++ b/packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts @@ -14,9 +14,13 @@ * green without a running server. Mirrors the satellite-db-injected-stores harness. */ -import { describe, it, expect, afterEach } from "vitest"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import { addChatMessage, addChatMessageAttachment, @@ -30,54 +34,18 @@ import { } from "../../async-stores/async-chat-store.js"; import type { ChatInFlightGenerationState, ChatMessage, ChatSession } from "../../chat/chat-types.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_chat_search_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (the async chat-store search/edit primitives are), and every assertion +is unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface Ctx { - dbName: string; layer: AsyncDataLayer; } -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); } catch { /* may not exist */ } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ databaseUrl: testUrl, databaseMigrationUrl: testUrl }); - const connections = await createConnectionSetFromUrl(backend, { poolMax: 3, connectTimeoutSeconds: 5 }); - await applySchemaBaseline(connections.migration); - const layer = createAsyncDataLayer(connections); - return { dbName, layer }; -} - -async function teardownCtx(ctx: Ctx | null): Promise { - if (!ctx) return; - try { await ctx.layer.close(); } catch { /* best-effort */ } - try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* best-effort */ } -} - let sessionCounter = 0; let messageCounter = 0; @@ -120,16 +88,20 @@ async function addMessage( } pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () => { - let ctx: Ctx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_chat_search_test", }); + let ctx: Ctx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("matches by message content, dedups to the most recent match, and respects scope", async () => { - ctx = await setupCtx(); - const session = await makeSession(ctx, "Weekend plans"); await addMessage(ctx, session.id, "user", "Let's talk about the quarterly roadmap"); const single = await searchChatSessionsByMessageContent(ctx.layer.db, "roadmap", [session.id]); @@ -165,8 +137,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () }); it("treats literal % and _ as literal characters, not LIKE wildcards", async () => { - ctx = await setupCtx(); - const literalSession = await makeSession(ctx); await addMessage(ctx, literalSession.id, "user", "Discount is 50% off, use code A_B"); const otherSession = await makeSession(ctx); @@ -186,8 +156,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () }); it("deleteChatMessagesFrom truncates from the target (inclusive) and preserves retained order", async () => { - ctx = await setupCtx(); - const session = await makeSession(ctx); const m1 = await addMessage(ctx, session.id, "user", "first turn"); const m2 = await addMessage(ctx, session.id, "assistant", "first reply"); @@ -203,8 +171,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () }); it("deleteChatMessagesFrom is a no-op for a wrong-session or unknown target", async () => { - ctx = await setupCtx(); - const sessionA = await makeSession(ctx); const sessionB = await makeSession(ctx); const a1 = await addMessage(ctx, sessionA.id, "user", "keep me"); @@ -226,7 +192,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () invariant so a later tool-derived mutation cannot poison a chat row. */ it("sanitizes attachment appends and metadata merge updates at their jsonb boundaries", async () => { - ctx = await setupCtx(); const session = await makeSession(ctx); const message = await addMessage(ctx, session.id, "user", "hello", { clean: true }); @@ -254,8 +219,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () }); it("updateChatMessageMetadata merges by default, replaces on merge:false, and throws for missing messages", async () => { - ctx = await setupCtx(); - const session = await makeSession(ctx); const message = await addMessage(ctx, session.id, "user", "hello", { mentions: ["@a"] }); @@ -280,7 +243,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () by the live chat callback untouched. */ it("persists the NUL-marked in-flight tool snapshot and reads back its sanitized shape", async () => { - ctx = await setupCtx(); const session = await makeSession(ctx); const snapshot: ChatInFlightGenerationState = { status: "generating", @@ -336,7 +298,6 @@ pgDescribe("async chat store content search + edit primitives (PostgreSQL)", () containing a raw NUL byte round-trips instead of throwing. */ it("addChatMessage strips a raw U+0000 byte instead of throwing (regression for the CEO-agent chat crash)", async () => { - ctx = await setupCtx(); const session = await makeSession(ctx); const diagnosticDump = diff --git a/packages/core/src/__tests__/postgres/fts-replacement.test.ts b/packages/core/src/__tests__/postgres/fts-replacement.test.ts index 39fd047b4a..d6146aa138 100644 --- a/packages/core/src/__tests__/postgres/fts-replacement.test.ts +++ b/packages/core/src/__tests__/postgres/fts-replacement.test.ts @@ -23,15 +23,14 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; -import postgres from "postgres"; -import { drizzle } from "drizzle-orm/postgres-js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; import { sql, eq } from "drizzle-orm"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; -import { createConnectionSetFromUrl } from "../../postgres/connection.js"; -import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import * as schema from "../../postgres/schema/index.js"; import { insertTaskRow } from "../../task-store/async/async-persistence.js"; import { @@ -45,89 +44,17 @@ import { import { upsertArchivedTaskEntry } from "../../task-store/async/async-archive-lineage.js"; import type { ArchivedTaskEntry } from "../../types.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_fts_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (the tsvector/GIN search path is), and every assertion is unchanged. +The DROP INDEX / REINDEX cases re-create the same index state they mutate, so they stay +safe on a per-file shared database. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; - testUrl: string; layer: AsyncDataLayer; - adminSql: ReturnType; -} - -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - - const schemaBackend: ResolvedBackend = { - mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, - migrationUrlOverridden: false, - }; - const schemaConnections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 1, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); - - const connections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 5, - connectTimeoutSeconds: 5, - }); - const layer = createAsyncDataLayer(connections); - - const adminSql = postgres(testUrl, { max: 2, prepare: false, onnotice: () => {} }); - // Keep a reference so TS doesn't flag unused; adminSql is used for teardown - // via end() and direct diagnostic queries. - void drizzle(adminSql); - - return { dbName, testUrl, layer, adminSql }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { - await ctx.layer.close(); - } catch { - // best-effort - } - try { - await ctx.adminSql.end({ timeout: 5 }); - } catch { - // best-effort - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); - } catch { - // best-effort - } } /** A minimal task record with the NOT NULL columns filled. */ @@ -162,17 +89,22 @@ function resultIds(rows: Record[]): string[] { } pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => { - let ctx: TestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_fts_test", }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); // ── VAL-SEARCH-001: Search parity with FTS5 baseline (row membership) ── it("returns the same row membership as the FTS5 baseline for representative queries (VAL-SEARCH-001)", async () => { - ctx = await setupCtx(); // Seed tasks with distinct searchable text. await insertTask(ctx.layer, "FTS-001", { title: "database migration guide" }); await insertTask(ctx.layer, "FTS-002", { title: "frontend redesign" }); @@ -199,7 +131,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => }); it("matches terms across id, title, description, and comments columns (VAL-SEARCH-001)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "SEARCH-ID-1", { title: "alpha" }); await insertTask(ctx.layer, "PLAIN-002", { title: "beta", comments: [{ text: "gamma delta notes" }] }); @@ -216,7 +147,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // Prefix matching regression test: "frob" must find "frobnicator" (FTS5 * parity). // to_tsquery with :* suffix reproduces FTS5's `${token}*` prefix token. it("prefix matching: partial token finds longer indexed term (VAL-SEARCH-001)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "PREFIX-001", { title: "frobnicator setup" }); await insertTask(ctx.layer, "PREFIX-002", { title: "database tuning" }); @@ -233,7 +163,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-002: tsvector sync-on-write (insert) ── it("newly inserted task is immediately searchable without explicit reindex (VAL-SEARCH-002)", async () => { - ctx = await setupCtx(); // No tasks exist yet. const before = await searchTasksTsvector(ctx.layer.db, "freshly"); expect(before).toEqual([]); @@ -247,7 +176,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-003: tsvector sync-on-write (update) ── it("updated task text fields are reflected in search immediately (VAL-SEARCH-003)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "UPD-001", { title: "original title" }); // "renamed" not present initially. @@ -274,7 +202,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-004: tsvector sync-on-write (delete) ── it("soft-deleted task no longer appears in live search (VAL-SEARCH-004)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "DEL-001", { title: "to be deleted searchable" }); await insertTask(ctx.layer, "DEL-002", { title: "to be deleted keeper" }); @@ -294,7 +221,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => }); it("hard-deleted task row is gone from search (VAL-SEARCH-004)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "HARD-001", { title: "hard delete target" }); const before = await searchTasksTsvector(ctx.layer.db, "target"); @@ -311,7 +237,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-005: Archive search parity ── it("archived-task search returns matching rows via tsvector (VAL-SEARCH-005)", async () => { - ctx = await setupCtx(); const baseEntry = (id: string, title: string, description: string) => ({ id, @@ -343,7 +268,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-006: Non-text mutation does not regenerate tsvector ── it("a mutation touching only non-text columns leaves search_vector unchanged (VAL-SEARCH-006)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "VEC-001", { title: "stable title text" }); // Read the initial search_vector value. @@ -365,7 +289,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => }); it("a mutation touching a text column DOES regenerate the tsvector (VAL-SEARCH-006 inverse)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "VEC-002", { title: "before change" }); const svBefore = await readTaskSearchVector(ctx.layer.db, "VEC-002"); @@ -385,7 +308,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── VAL-SEARCH-007: Index rebuild restores search ── it("REINDEX on the GIN index restores correct search without data loss (VAL-SEARCH-007)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "RIDX-001", { title: "reindex probe alpha" }); await insertTask(ctx.layer, "RIDX-002", { title: "reindex probe beta" }); @@ -412,7 +334,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => }); it("DROP + re-CREATE the GIN index restores search (VAL-SEARCH-007 alternate)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "DROP-001", { title: "drop recreate search" }); // Drop the index (simulating corruption/missing index). @@ -430,7 +351,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => // ── Helpers / edge cases ── it("empty and whitespace queries return no results (no crash)", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "EDGE-001", { title: "something" }); expect(await searchTasksTsvector(ctx.layer.db, "")).toEqual([]); @@ -449,7 +369,6 @@ pgDescribe("fts-replacement: tsvector/GIN full-text search (PostgreSQL)", () => }); it("includeArchived=false excludes archived tasks from search", async () => { - ctx = await setupCtx(); await insertTask(ctx.layer, "ARCH-001", { title: "archived filter target", column: "archived" }); await insertTask(ctx.layer, "LIVE-001", { title: "archived filter target", column: "todo" }); diff --git a/packages/core/src/__tests__/postgres/postgres-health.test.ts b/packages/core/src/__tests__/postgres/postgres-health.test.ts index 139e754b24..7607e4fc4a 100644 --- a/packages/core/src/__tests__/postgres/postgres-health.test.ts +++ b/packages/core/src/__tests__/postgres/postgres-health.test.ts @@ -17,15 +17,15 @@ * Skipped when PostgreSQL is unreachable (FUSION_PG_TEST_SKIP=1). */ -import { describe, it, expect, afterEach } from "vitest"; -import { execSync } from "node:child_process"; -import postgres from "postgres"; -import { drizzle } from "drizzle-orm/postgres-js"; +import { it, expect, afterEach } from "vitest"; import { sql } from "drizzle-orm"; import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; import { createConnectionSetFromUrl } from "../../postgres/connection.js"; import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import { + pgDescribe, + createTaskStoreForTest, +} from "../../__test-utils__/pg-test-harness.js"; import { checkPostgresHealth, detectSchemaDrift, @@ -37,82 +37,37 @@ import { import { detectTaskIdIntegrityAnomaliesAsync } from "../../postgres/async-task-id-integrity.js"; import { PROJECT_SCHEMA } from "../../postgres/schema/_shared.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_u8_health_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated the per-test database creation off hand-rolled CREATE DATABASE + +applySchemaBaseline (~3-4s of DDL per test) onto the harness's template-cloned +`createTaskStoreForTest`. This file KEEPS a private database per test on purpose: the +schema-drift cases DROP real columns and the VACUUM cases assert whole-table row/dead-tuple +counts, both of which would poison a file-shared database. Only the scaffolding changed — +the harness's TaskStore init seeds config/allocator rows, so setup wipes application data +to restore the pristine tables the original fresh databases provided. Every assertion is +unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; - testUrl: string; layer: AsyncDataLayer; - adminSql: ReturnType; + teardown: () => Promise; } async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - - const schemaBackend: ResolvedBackend = { - mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, - migrationUrlOverridden: false, - }; - const schemaConnections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 1, - connectTimeoutSeconds: 5, + const harness = await createTaskStoreForTest({ + prefix: "fusion_u8_health", + copyFromGolden: true, }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); - - const connections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 5, - connectTimeoutSeconds: 5, - }); - const layer = createAsyncDataLayer(connections); - - const adminSql = postgres(testUrl, { max: 2, prepare: false, onnotice: () => {} }); - return { dbName, testUrl, layer, adminSql }; + await harness.adminDb.execute(sql`DELETE FROM project.tasks`); + await harness.adminDb.execute(sql`DELETE FROM project.archived_tasks`); + await harness.adminDb.execute(sql`DELETE FROM project.distributed_task_id_state`); + return { layer: harness.layer, teardown: harness.teardown }; } async function teardownCtx(ctx: TestCtx | null): Promise { if (!ctx) return; try { - await ctx.layer.close(); - } catch { - // best-effort - } - try { - await ctx.adminSql.end({ timeout: 3 }); - } catch { - // best-effort - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); + await ctx.teardown(); } catch { // best-effort } diff --git a/packages/core/src/__tests__/postgres/secrets-roundtrip.test.ts b/packages/core/src/__tests__/postgres/secrets-roundtrip.test.ts index f90da56cd2..828a5091fa 100644 --- a/packages/core/src/__tests__/postgres/secrets-roundtrip.test.ts +++ b/packages/core/src/__tests__/postgres/secrets-roundtrip.test.ts @@ -26,96 +26,30 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; -import postgres from "postgres"; -import { drizzle } from "drizzle-orm/postgres-js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { eq } from "drizzle-orm"; import { sql } from "drizzle-orm"; import { randomBytes } from "node:crypto"; -import { execSync } from "node:child_process"; import { createSecretCipher } from "../../secrets/secrets-crypto.js"; import * as schema from "../../postgres/schema/index.js"; - -const PG_ADMIN_URL = - process.env.FUSION_PG_TEST_ADMIN_URL ?? "postgresql://localhost:5432/postgres"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -/** - * FNXC:SecretsStore 2026-06-24-12:00: - * Create a uniquely-named fresh database for each test so tests are hermetic - * and never touch existing data. Mirrors the data-layer / schema-applier test - * harness (CREATE/DROP DATABASE cannot run inside a transaction, so psql via - * execSync is the acceptable short-DDL use per AGENTS.md). - */ -function uniqueDbName(): string { - return `fusion_secret_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (the bytea encrypt → INSERT → SELECT → decrypt cycle is), and every +assertion is unchanged. `db` is the harness's raw admin Drizzle connection, matching the +direct-connection semantics the original per-test databases used. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface SecretTestCtx { - dbName: string; - testUrl: string; - adminSql: ReturnType; - db: ReturnType; -} - -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - - // Apply the baseline schema so secrets + secrets_global exist. - const { createConnectionSetFromUrl } = await import("../../postgres/connection.js"); - const { applySchemaBaseline } = await import("../../postgres/schema-applier.js"); - const { resolveBackendWithOptions } = await import("../../postgres/backend-resolver.js"); - const backend = resolveBackendWithOptions({ - databaseUrl: testUrl, - databaseMigrationUrl: testUrl, - }); - const connections = await createConnectionSetFromUrl(backend, { - poolMax: 1, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(connections.migration); - await connections.close(); - - const adminSql = postgres(testUrl, { max: 3, prepare: false, onnotice: () => {} }); - const db = drizzle(adminSql); - return { dbName, testUrl, adminSql, db }; -} - -async function teardownCtx(ctx: SecretTestCtx | null): Promise { - if (!ctx) return; - try { - await ctx.adminSql.end({ timeout: 5 }); - } catch { - // best-effort - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); - } catch { - // best-effort - } + db: PostgresJsDatabase; } /** A fixed 32-byte master key provider for deterministic test crypto. */ @@ -124,15 +58,20 @@ function fixedMasterKeyProvider(key: Buffer = randomBytes(32)): () => Promise { - let ctx: SecretTestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_secret_test", }); + let ctx: SecretTestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { db: h.adminDb() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("round-trips a project-scoped secret through project.secrets bytea columns", async () => { - ctx = await setupCtx(); const cipher = createSecretCipher(fixedMasterKeyProvider()); const plaintext = "super-secret-api-key-12345"; const encrypted = await cipher.encrypt(plaintext); @@ -180,7 +119,6 @@ pgDescribe("PostgreSQL secrets round-trip (VAL-CROSS-011)", () => { }); it("round-trips a global-scoped secret through central.secrets_global bytea columns", async () => { - ctx = await setupCtx(); const cipher = createSecretCipher(fixedMasterKeyProvider()); const plaintext = "global-secret-token-XYZ"; const encrypted = await cipher.encrypt(plaintext); @@ -222,7 +160,6 @@ pgDescribe("PostgreSQL secrets round-trip (VAL-CROSS-011)", () => { }); it("preserves ciphertext integrity across a re-read (tamper detection via GCM auth tag)", async () => { - ctx = await setupCtx(); const cipher = createSecretCipher(fixedMasterKeyProvider()); const plaintext = "integrity-check-value"; const encrypted = await cipher.encrypt(plaintext); @@ -266,7 +203,6 @@ pgDescribe("PostgreSQL secrets round-trip (VAL-CROSS-011)", () => { }); it("enforces the access_policy CHECK constraint on project.secrets", async () => { - ctx = await setupCtx(); const cipher = createSecretCipher(fixedMasterKeyProvider()); const encrypted = await cipher.encrypt("v"); @@ -306,7 +242,6 @@ pgDescribe("PostgreSQL secrets round-trip (VAL-CROSS-011)", () => { }); it("enforces key uniqueness on project.secrets", async () => { - ctx = await setupCtx(); const cipher = createSecretCipher(fixedMasterKeyProvider()); const encrypted = await cipher.encrypt("v"); diff --git a/packages/core/src/__tests__/postgres/taskstore-persistence.test.ts b/packages/core/src/__tests__/postgres/taskstore-persistence.test.ts index f2a084f6ca..c73d214757 100644 --- a/packages/core/src/__tests__/postgres/taskstore-persistence.test.ts +++ b/packages/core/src/__tests__/postgres/taskstore-persistence.test.ts @@ -20,15 +20,15 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; -import postgres from "postgres"; -import { drizzle } from "drizzle-orm/postgres-js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { sql, eq } from "drizzle-orm"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; -import { createConnectionSetFromUrl } from "../../postgres/connection.js"; -import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import * as schema from "../../postgres/schema/index.js"; import { insertTaskRow, @@ -52,87 +52,19 @@ import { } from "../../task-store/async/async-settings.js"; import type { WorkflowTransitionNotificationMarker } from "../../types.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_u12_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test, and every assertion is unchanged. Tests that seed +project.distributed_task_id_state or assume an absent project.config row first DELETE the +rows the harness's TaskStore init/config re-seed may have created, restoring the pristine +table state the original per-test database gave them. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; - testUrl: string; layer: AsyncDataLayer; - adminSql: ReturnType; - adminDb: ReturnType; -} - -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - - const schemaBackend: ResolvedBackend = { - mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, - migrationUrlOverridden: false, - }; - const schemaConnections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 1, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); - - const connections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 5, - connectTimeoutSeconds: 5, - }); - const layer = createAsyncDataLayer(connections); - - const adminSql = postgres(testUrl, { max: 2, prepare: false, onnotice: () => {} }); - const adminDb = drizzle(adminSql); - return { dbName, testUrl, layer, adminSql, adminDb }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { - await ctx.layer.close(); - } catch { - // best-effort - } - try { - await ctx.adminSql.end({ timeout: 5 }); - } catch { - // best-effort - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); - } catch { - // best-effort - } + adminDb: PostgresJsDatabase; } /** A minimal task record with the NOT NULL columns filled. */ @@ -149,17 +81,27 @@ function makeMinimalTask(id: string, column = "todo"): Record { } pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { - let ctx: TestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_u12_test", }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer(), adminDb: h.adminDb() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); + + /** Restore the pristine allocator-state table the per-test databases used to provide. */ + async function clearAllocatorState(): Promise { + await ctx.adminDb.execute(sql`DELETE FROM project.distributed_task_id_state`); + } // ── VAL-DATA-009 / VAL-SCHEMA-004: create + JSON round-trip ─────────── it("inserts a task and reads it back via async Drizzle (VAL-DATA-009)", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("KB-001"), { lineageId: null }); const row = await readTaskRow(ctx.layer, "KB-001"); @@ -170,7 +112,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("round-trips JSON columns as JSONB with identical shape (VAL-SCHEMA-004)", async () => { - ctx = await setupCtx(); // The column descriptors read nested fields (e.g. task.tokenUsage.perModel), // so the task record carries the canonical Task shape for JSON-backed columns. const workflowTransitionNotification: WorkflowTransitionNotificationMarker = { @@ -230,7 +171,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("create-class insert is non-destructive: duplicate id raises, existing row intact (VAL-DATA-009)", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("KB-010"), { lineageId: null }); // A second insert with the same id must fail (primary-key violation), not @@ -256,7 +196,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { // ── VAL-DATA-005 / VAL-DATA-006: soft-delete visibility ─────────────── it("soft-deleted tasks are hidden from live readers (VAL-DATA-005)", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("KB-100", "todo"), { lineageId: null }); await insertTaskRow(ctx.layer, makeMinimalTask("KB-101", "todo"), { lineageId: null }); @@ -280,7 +219,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("forensic reads surface soft-deleted rows (VAL-DATA-006)", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("KB-200", "todo"), { lineageId: null }); const deletedAt = new Date().toISOString(); await softDeleteTaskRow(ctx.layer, "KB-200", deletedAt); @@ -299,7 +237,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { // listTasks → readLiveTaskRows. Without the wiring, soft-deleted tasks were // absent from the list response even when includeDeleted=true was passed. it("readLiveTaskRows surfaces soft-deleted rows when includeDeleted is set (VAL-CROSS-003)", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("KB-300", "todo"), { lineageId: null }); await insertTaskRow(ctx.layer, makeMinimalTask("KB-301", "todo"), { lineageId: null }); const deletedAt = new Date().toISOString(); @@ -323,7 +260,7 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { // ── VAL-DATA-007 / VAL-DATA-008: allocator reconciliation ───────────── it("allocator reconciliation bumps sequences to max suffix on store open (VAL-DATA-007)", async () => { - ctx = await setupCtx(); + await clearAllocatorState(); // Seed a task with a high suffix, but leave the sequence at a low value. await insertTaskRow(ctx.layer, makeMinimalTask("KB-050"), { lineageId: null }); // Manually set the sequence to a low value (below the seeded suffix). @@ -352,7 +289,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("soft-deleted IDs stay reserved (VAL-DATA-008)", async () => { - ctx = await setupCtx(); // Seed a soft-deleted task with a high suffix. await insertTaskRow(ctx.layer, makeMinimalTask("KB-099", "todo"), { lineageId: null }); await softDeleteTaskRow(ctx.layer, "KB-099", new Date().toISOString()); @@ -375,7 +311,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("reconciliation accounts for archived-task IDs (VAL-DATA-008)", async () => { - ctx = await setupCtx(); // Seed an archived task row with a high suffix. await ctx.adminDb.execute(sql` INSERT INTO project.archived_tasks (id, data, archived_at) @@ -387,7 +322,7 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("reconciliation accounts for reservation IDs", async () => { - ctx = await setupCtx(); + await clearAllocatorState(); // Seed a reservation with a high sequence. const nowIso = new Date().toISOString(); await ctx.adminDb.execute(sql` @@ -406,7 +341,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("getKnownPrefixes discovers prefixes from tasks and archived tasks", async () => { - ctx = await setupCtx(); await insertTaskRow(ctx.layer, makeMinimalTask("ABC-001"), { lineageId: null }); await ctx.adminDb.execute(sql` INSERT INTO project.archived_tasks (id, data, archived_at) @@ -423,7 +357,9 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { // ── Settings round-trip ─────────────────────────────────────────────── it("settings read/update project round-trip (VAL-SCHEMA-004 jsonb)", async () => { - ctx = await setupCtx(); + // The shared harness re-seeds a default config row per test; remove it so the + // "initially absent" contract this test pins is observed exactly as before. + await ctx.adminDb.execute(sql`DELETE FROM project.config`); // Initially absent → default. let config = await readProjectConfig(ctx.layer); @@ -449,7 +385,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("settings patch deep-merges into the existing row", async () => { - ctx = await setupCtx(); await writeProjectConfig(ctx.layer, { taskPrefix: "KB", maxConcurrent: 4 }); await patchProjectSettings(ctx.layer, { autoMerge: true }); @@ -459,7 +394,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("settings preserve nextWorkflowStepId across updates", async () => { - ctx = await setupCtx(); await writeProjectConfig(ctx.layer, { taskPrefix: "KB" }, { nextWorkflowStepId: 7 }); // A subsequent write without the option preserves the prior value. @@ -470,7 +404,6 @@ pgDescribe("U12 taskstore-persistence (PostgreSQL)", () => { }); it("config row enforces per-project singleton via project_id PK", async () => { - ctx = await setupCtx(); // FNXC:MultiProjectIsolation 2026-07-11: config is now keyed per-project on // project_id (the PK). The old singleton CHECK (id = 1) was removed so multiple // projects can each have their own config row. A duplicate project_id must diff --git a/packages/core/src/__tests__/postgres/u15-engine-dashboard-consumers.test.ts b/packages/core/src/__tests__/postgres/u15-engine-dashboard-consumers.test.ts index 840dd55bc2..ff318e66bd 100644 --- a/packages/core/src/__tests__/postgres/u15-engine-dashboard-consumers.test.ts +++ b/packages/core/src/__tests__/postgres/u15-engine-dashboard-consumers.test.ts @@ -21,15 +21,15 @@ * gate stays green without a running server. */ -import { describe, it, expect, afterEach } from "vitest"; -import postgres from "postgres"; -import { drizzle } from "drizzle-orm/postgres-js"; +import { describe, it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { and, eq } from "drizzle-orm"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; -import { createConnectionSetFromUrl } from "../../postgres/connection.js"; -import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import * as schema from "../../postgres/schema/index.js"; import { recordDeploymentAsync, @@ -51,87 +51,17 @@ import { reconcileSoftDeletedColumnDriftAsync, } from "../../task-store/async/async-self-healing.js"; -const PG_TEST_URL_BASE = - process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = - process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); - -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - -function uniqueDbName(): string { - return `fusion_u15_test_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - /* -FNXC:PgTestAuthFix 2026-07-14-00:00: -The inline adminExec used process.env.USER for the psql -U flag, which is 'runner' on GitHub Actions (not 'postgres'). Use the PG_TEST_URL_BASE connection string instead so credentials are always correct. +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per file +with TRUNCATE-based reset per test. The database setup here was scaffolding, not the +subject under test (the async monitor-store and self-healing helpers are), and every +assertion is unchanged. */ -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - interface TestCtx { - dbName: string; - testUrl: string; layer: AsyncDataLayer; - adminSql: ReturnType; - adminDb: ReturnType; -} - -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - - const schemaBackend: ResolvedBackend = { - mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, - migrationUrlOverridden: false, - }; - const schemaConnections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 1, - connectTimeoutSeconds: 5, - }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); - - const connections = await createConnectionSetFromUrl(schemaBackend, { - poolMax: 5, - connectTimeoutSeconds: 5, - }); - const layer = createAsyncDataLayer(connections); - - const adminSql = postgres(testUrl, { max: 2, prepare: false, onnotice: () => {} }); - const adminDb = drizzle(adminSql); - return { dbName, testUrl, layer, adminSql, adminDb }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { - await ctx.layer.close(); - } catch { - // best-effort - } - try { - await ctx.adminSql.end({ timeout: 5 }); - } catch { - // best-effort - } - try { - adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); - } catch { - // best-effort - } + adminDb: PostgresJsDatabase; } /** @@ -160,17 +90,22 @@ async function seedTask( } pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { - let ctx: TestCtx | null = null; - - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_u15_test", }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer(), adminDb: h.adminDb() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); // ── Monitor store: deployments ──────────────────────────────────────────── describe("monitor deployments", () => { it("records a deployment and reads it back via async Drizzle", async () => { - ctx = await setupCtx(); const deployment = await recordDeploymentAsync(ctx.layer.db, { service: "api", environment: "prod", @@ -187,7 +122,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("is idempotent by deploymentId (upsert, not duplicate)", async () => { - ctx = await setupCtx(); const first = await recordDeploymentAsync(ctx.layer.db, { deploymentId: "dep-1", status: "deployed", @@ -208,7 +142,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { // ── Monitor store: incidents + storm guard ──────────────────────────────── describe("monitor incidents + storm guard", () => { it("opens an incident then resolves it", async () => { - ctx = await setupCtx(); const { incident, created } = await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "g1", title: "API 500s", @@ -234,7 +167,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("absorbs a burst sharing one groupingKey into ONE open incident", async () => { - ctx = await setupCtx(); for (let i = 0; i < 100; i += 1) { await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "g-burst", @@ -247,13 +179,11 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("resolveIncident returns null when nothing is open", async () => { - ctx = await setupCtx(); const result = await resolveIncidentAsync(ctx.layer.db, "nope"); expect(result).toBeNull(); }); it("the atomic claim step prevents a second claim once an incident is claimed", async () => { - ctx = await setupCtx(); const { incident } = await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "g-claim", title: "Claim me", @@ -273,7 +203,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("releases a stranded sentinel claim back to NULL but never clobbers a real id", async () => { - ctx = await setupCtx(); const { incident } = await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "g-rel", title: "t", @@ -294,7 +223,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("countRecentAutoFixTasks ignores sentinel placeholders but counts real links", async () => { - ctx = await setupCtx(); const { incident: a } = await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "ga", title: "a" }); const { incident: b } = await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "gb", title: "b" }); // a is only claimed (sentinel) → must NOT count. @@ -306,7 +234,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("decideStormGuard preserves threshold, sustained, absorb, and circuit-breaker gates", async () => { - ctx = await setupCtx(); const incident = (await ingestIncidentSignalAsync(ctx.layer.db, { groupingKey: "g", title: "t" })).incident; const now = Date.parse("2026-06-24T10:00:00.000Z"); @@ -351,7 +278,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { // ── Self-healing: reconcileSoftDeletedColumnDrift ───────────────────────── describe("self-healing reconcileSoftDeletedColumnDrift", () => { it("reconciles soft-deleted non-archived tasks to archived and records an audit per row", async () => { - ctx = await setupCtx(); const deletedAt = new Date().toISOString(); // Soft-deleted tasks that drifted off archived. await seedTask(ctx, "FN-drift-1", { column: "in-review", deletedAt }); @@ -392,7 +318,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("lists only soft-deleted non-archived candidates", async () => { - ctx = await setupCtx(); const deletedAt = new Date().toISOString(); await seedTask(ctx, "FN-d1", { column: "in-review", deletedAt }); await seedTask(ctx, "FN-live", { column: "todo", deletedAt: null }); @@ -404,7 +329,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("scopes drift reconciliation to the data layer project partition", async () => { - ctx = await setupCtx(); const deletedAt = new Date().toISOString(); await seedTask(ctx, "FN-shared", { column: "todo", deletedAt, projectId: "project-a" }); await seedTask(ctx, "FN-shared", { column: "in-review", deletedAt, projectId: "project-b" }); @@ -430,7 +354,6 @@ pgDescribe("U15 engine + dashboard consumers (PostgreSQL)", () => { }); it("returns zero reconciled when no candidates exist", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-live", { column: "todo", deletedAt: null }); const result = await reconcileSoftDeletedColumnDriftAsync(ctx.layer, async () => {}); expect(result.reconciled).toBe(0); diff --git a/packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts b/packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts index 59f9a04bd3..cee51ca479 100644 --- a/packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts +++ b/packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts @@ -30,13 +30,13 @@ Skipped when PostgreSQL is unreachable (FUSION_PG_TEST_SKIP=1) so the merge gate stays green without a running server — the same posture as the sibling PG suites. */ -import { describe, it, expect, afterEach } from "vitest"; -import postgres from "postgres"; -import { execSync } from "node:child_process"; -import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js"; -import { createConnectionSetFromUrl } from "../../postgres/connection.js"; -import type { ResolvedBackend } from "../../postgres/backend-resolver.js"; -import { applySchemaBaseline } from "../../postgres/schema-applier.js"; +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; import { insertTaskRow, readTaskRow } from "../../task-store/async/async-persistence.js"; import * as schema from "../../postgres/schema/index.js"; import { and, eq } from "drizzle-orm"; @@ -50,70 +50,20 @@ import { import { createWorkflowEventBus } from "../../workflow-events.js"; import type { WorkflowLifecycleEvent } from "../../types/workflow-events.js"; -const PG_TEST_URL_BASE = process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432"; -const PG_AVAILABLE = process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE); -const pgDescribe = PG_AVAILABLE ? describe : describe.skip; - const TEST_PROJECT_ID = "proj_test_u3_outbox"; -function uniqueDbName(): string { - return `fusion_u3_outbox_${process.pid}_${Math.random().toString(36).slice(2, 8)}`; -} - -function adminExec(statement: string): void { - execSync( - `psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`, - { stdio: "pipe", env: process.env }, - ); -} - +/* +FNXC:PgTestHarnessAdoption 2026-08-16-03:45: +Migrated off the hand-rolled per-test CREATE DATABASE + applySchemaBaseline scaffolding +(~3-4s of DDL per test) onto the shared PG harness: one template-cloned database per +describe block with TRUNCATE-based reset per test. The database setup here was +scaffolding — the real-PG lease predicate is the subject and still runs against real +PostgreSQL. The harness keeps the project-BOUND layer this suite requires (an unbound +harness runs with RLS bypassed and writes rows the bound reader cannot see), and every +assertion is unchanged. +*/ interface TestCtx { - dbName: string; layer: AsyncDataLayer; - adminSql: ReturnType; -} - -async function setupCtx(): Promise { - const dbName = uniqueDbName(); - try { - adminExec(`DROP DATABASE IF EXISTS "${dbName}"`); - } catch { - // may not exist - } - adminExec(`CREATE DATABASE "${dbName}"`); - const testUrl = `${PG_TEST_URL_BASE}/${dbName}`; - const backend: ResolvedBackend = { - mode: "external", - runtimeUrl: testUrl, - migrationUrl: testUrl, - migrationUrlOverridden: false, - }; - const schemaConnections = await createConnectionSetFromUrl(backend, { poolMax: 1, connectTimeoutSeconds: 5 }); - await applySchemaBaseline(schemaConnections.migration); - await schemaConnections.close(); - - // Bind the layer to a project, as production does — an unbound harness runs - // with RLS bypassed and writes rows the bound reader cannot see. - const connections = await createConnectionSetFromUrl(backend, { - poolMax: 5, - connectTimeoutSeconds: 5, - projectId: TEST_PROJECT_ID, - }); - const layer = createAsyncDataLayer(connections, { projectId: TEST_PROJECT_ID }); - const adminSql = postgres(testUrl, { - max: 2, - prepare: false, - onnotice: () => {}, - connection: { "fusion.project_id": TEST_PROJECT_ID }, - }); - return { dbName, layer, adminSql }; -} - -async function teardownCtx(ctx: TestCtx | null): Promise { - if (!ctx) return; - try { await ctx.layer.close(); } catch { /* closing best-effort */ } - try { await ctx.adminSql.end({ timeout: 5 }); } catch { /* closing best-effort */ } - try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* dropped best-effort */ } } async function seedTask(ctx: TestCtx, id: string, column = "in-progress"): Promise { @@ -144,14 +94,21 @@ async function moveColumnInTransaction(tx: DbTransaction, taskId: string, column } pgDescribe("transactional outbox — durable follow-on work (U3 / R5)", () => { - let ctx: TestCtx | null = null; - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_u3_outbox", + projectId: TEST_PROJECT_ID, }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("a work item written INSIDE the transition transaction survives a crash before the event is emitted", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-OUT-1"); // The transition transaction: the COLUMN CHANGE and the durable follow-on @@ -179,7 +136,6 @@ pgDescribe("transactional outbox — durable follow-on work (U3 / R5)", () => { }); it("a ROLLED-BACK transition leaves no column change, no orphan work, and emits NOTHING", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-OUT-2"); // The emit point sits AFTER the transaction block in moves.ts, so a throw @@ -215,14 +171,21 @@ pgDescribe("transactional outbox — durable follow-on work (U3 / R5)", () => { }); pgDescribe("transactional outbox — delivery is AT-LEAST-ONCE (U3 / R5)", () => { - let ctx: TestCtx | null = null; - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_u3_outbox_alo", + projectId: TEST_PROJECT_ID, }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("an item whose lease EXPIRED mid-flight is re-claimable — the store redelivers", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-OUT-3"); const created = await upsertWorkflowWorkItem(ctx.layer, { @@ -250,7 +213,6 @@ pgDescribe("transactional outbox — delivery is AT-LEAST-ONCE (U3 / R5)", () => }); it("an IDEMPOTENT handler run twice over one redelivered item produces exactly one effect", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-OUT-4"); const created = await upsertWorkflowWorkItem(ctx.layer, { @@ -278,14 +240,21 @@ pgDescribe("transactional outbox — delivery is AT-LEAST-ONCE (U3 / R5)", () => }); pgDescribe("post-commit event vs. outbox — the division of labour (U3 / R5, KTD-3)", () => { - let ctx: TestCtx | null = null; - afterEach(async () => { - await teardownCtx(ctx); - ctx = null; + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_u3_outbox_evt", + projectId: TEST_PROJECT_ID, }); + let ctx: TestCtx; + + beforeAll(h.beforeAll); + beforeEach(async () => { + await h.beforeEach(); + ctx = { layer: h.layer() }; + }); + afterEach(h.afterEach); + afterAll(h.afterAll); it("dropping every subscriber loses the REACTION but not the durable work", async () => { - ctx = await setupCtx(); await seedTask(ctx, "FN-OUT-5"); const bus = createWorkflowEventBus();