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

test(core): migrate 12 hand-rolled PG test files onto the shared harness

Browse Source 
Slow-test trim, OPTIMIZE tier (mechanical sweep). These files each
created and baselined a fresh database per test (~3-4.6s/test); they now
use the shared golden-template harness (TRUNCATE per test) or, where a
private database is genuinely required (postgres-health drift/VACUUM
subjects, allocator-cross-project RLS connection sets), a per-test
template clone instead of raw DDL. Zero assertion lines changed
(verified), identical test counts, 110/110 green. Deliberately left
alone: startup-factory-integration and embedded-lifecycle (own-database
boot IS the subject) and schema-applier (real DDL is the subject).
Measured: ~195s serial → ~30s for these files.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-08-15 20:55:11 -07:00
parent f1a8b271bd
commit ea90eeed1c
12 changed files with 360 additions and 930 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
packages/core/src/__tests__/postgres/allocator-cross-project.test.ts
View File

@@ -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<void>;
connectionsA: PostgresConnections;
connectionsB: PostgresConnections;
layerA: AsyncDataLayer;
@@ -58,27 +51,16 @@ interface TestCtx {
}
async function setupCtx(): Promise<TestCtx> {
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<TestCtx> {
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<void> {
@@ -107,7 +89,7 @@ async function teardownCtx(ctx: TestCtx | null): Promise<void> {
// best-effort
}
try {
adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`);
await ctx.baseTeardown();
} catch {
// best-effort
}

79
packages/core/src/__tests__/postgres/approval-request-lifecycle.pg.test.ts
View File

@@ -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<StoreTestCtx> {
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<void> {
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 });

77
packages/core/src/__tests__/postgres/archive-pagination.pg.test.ts
View File

@@ -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<Ctx> {
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<void> {
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()));

88
packages/core/src/__tests__/postgres/central-archive-secrets.test.ts
View File

@@ -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<TestCtx> {
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<void> {
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<Buffer> {
return async () => Buffer.from(key);
@@ -107,17 +74,22 @@ function sampleArchiveEntry(overrides: Partial<ArchivedTaskEntry> = {}): 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<Record<string, unknown>> = [];
const store = new AsyncSecretsStore(ctx.layer, fixedMasterKeyProvider(), {

143
packages/core/src/__tests__/postgres/central-core-backend.test.ts
View File

@@ -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<TestCtx> {
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<void> {
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<TestCtx> {
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);

87
packages/core/src/__tests__/postgres/chat-store-content-search-edit.pg.test.ts
View File

@@ -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<Ctx> {
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<void> {
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 =

131
packages/core/src/__tests__/postgres/fts-replacement.test.ts
View File

@@ -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<typeof postgres>;
}
async function setupCtx(): Promise<TestCtx> {
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<void> {
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, unknown>[]): 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" });

91
packages/core/src/__tests__/postgres/postgres-health.test.ts
View File

@@ -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<typeof postgres>;
teardown: () => Promise<void>;
}
async function setupCtx(): Promise<TestCtx> {
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<void> {
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
}

117
packages/core/src/__tests__/postgres/secrets-roundtrip.test.ts
View File

@@ -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<typeof postgres>;
db: ReturnType<typeof drizzle>;
}
async function setupCtx(): Promise<SecretTestCtx> {
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<void> {
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<Bu
}
pgDescribe("PostgreSQL secrets round-trip (VAL-CROSS-011)", () => {
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");

143
packages/core/src/__tests__/postgres/taskstore-persistence.test.ts
View File

@@ -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<typeof postgres>;
adminDb: ReturnType<typeof drizzle>;
}
async function setupCtx(): Promise<TestCtx> {
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<void> {
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<string, unknown> {
}
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<void> {
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

129
packages/core/src/__tests__/postgres/u15-engine-dashboard-consumers.test.ts
View File

@@ -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<typeof postgres>;
adminDb: ReturnType<typeof drizzle>;
}
async function setupCtx(): Promise<TestCtx> {
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<void> {
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);

137
packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts
View File

@@ -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<typeof postgres>;
}
async function setupCtx(): Promise<TestCtx> {
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<void> {
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<void> {
@@ -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();