From daa34fbc38062a117fbae78c6a015e27144b7209 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 16 Jul 2026 22:35:01 -0700 Subject: [PATCH] fix: refineTask/duplicateTask fail in backend (PostgreSQL) mode (#2253) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Eliminates the remaining backend/PostgreSQL-mode sync-SQLite (`store.db`) call sites — both the crashing ones and the try/catch-masked ones that silently degraded features. Found via a full audit of `store.db`/`archiveDb` residue after the PG cutover's per-site routing missed them. **Crashes fixed:** 1. **refineTask / duplicateTask** threw `TaskStore.db: SQLite Database is not available in backend mode`. Both create rows through `createTaskWithId` callbacks calling `store.atomicCreateTaskJson()` directly, bypassing `_createTaskInternal`'s backend routing. The shared helper now routes itself (soft-delete conflict check + non-destructive insert in one AsyncDataLayer transaction). 2. **Merger verification cache**: `getVerificationCacheHit` ran sync SQLite unguarded *outside* any try/catch in `runDeterministicVerification`; `recordVerificationCachePass` was swallowed so the cache never warmed. Both are now async with a PG branch. **Silent degradations fixed (features that were dead on PG):** - Workflow run-branch + foreach step-instance persistence (`saveWorkflowRunBranch`, `loadWorkflowRunBranches`, `clearWorkflowRunBranches`, `saveWorkflowRunStepInstance`, `loadWorkflowRunStepInstances`, `clearWorkflowRunStepInstances`) — executor crash-resume checkpoints were silently never persisted. - `getBranchProgressByTask` — returned an empty map, dropping `branchProgress` from task payloads. - `runPluginColumnTransitionHooks` — plugin `onEnter`/`onExit` column-transition hooks never fired (marker bookkeeping + non-locking task read now async). - `getTaskColumns` — dashboard treated all agent-linked tasks as non-terminal. - `getWorkflowStep` / `listWorkflowSteps` — stored workflow-step rows now read from `project.workflow_steps` (listing previously returned plugin steps only); `getLegacyWorkflowStepSnapshot` returns `undefined` on PG (legacy snapshot exists only in pre-migration SQLite). - `readRawProjectSettings` / `listWorkflowPromptOverridesForProject` — now read via the async layer. These store methods became **async**; engine/dashboard callers await them (the workflow persistence interfaces already accepted `Promise`-returning impls). **PG gotcha encoded in the fixes:** migration `0006_project_ownership` rebuilds every project-schema PK to lead with `project_id`, so column-list `ON CONFLICT` inference fails (42P10) — upserts target the PK by constraint name. ## Surface Enumeration - Creators through `atomicCreateTaskJson`: `refineTaskImpl`, `duplicateTaskImpl` (fixed); `_createTaskInternalImpl` unaffected (already routed). - Verification-cache callers (all merger, all 3 sites now awaited). - Run-branch/step-instance callers: executor persistence adapters, parse-steps foreach probe, integration-queue flip, crash-resume reconcile, graph-reset cleanup; triage replan cleanup; dashboard spec-rebuild pin clears; agent-reflection rework summing — all awaited. - Audit classified everything else as guarded or sync-mode-only (dead in production — every entry point constructs stores via `createTaskStoreForBackend`). ## Symptom Verification - **Original symptoms:** refinement/duplicate creation threw; merge verification threw; workflow checkpoints/branch progress/plugin hooks/task-column lookups silently no-oped on PostgreSQL. - **Exact reproduction:** `refine-duplicate-task.pg.test.ts`, `verification-cache.pg.test.ts`, and `sync-db-residue-backend.pg.test.ts` exercise each surface against embedded-PostgreSQL backend-mode TaskStores. - **Assertion it is gone:** all suites pass (14 + 5 tests), plus `transition-pending-and-status-clear.pg.test.ts`, `create-task-reserved-id.pg.test.ts`, dashboard `routes-github.test.ts` (123), engine `triage.test.ts` (221) and `agent-reflection.test.ts` (31). Core/engine/dashboard typecheck fully clean: the 13 errors from the FN-8142 pi SDK migration are fixed by bumping @earendil-works/pi-ai/pi-coding-agent to ^0.80.10 (FN-8142 used APIs absent from the previously locked 0.80.6). Locally green: `pnpm verify:fast` (scoped typecheck + build + CLI build + boot smoke), `pnpm test:gate`, and `pnpm lint`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) ## Summary by CodeRabbit * **Bug Fixes** * Fixed refinement/duplication task creation in PostgreSQL-backed backend mode. * Improved backend-mode persistence for workflow checkpoints, foreach-step instances, branch progress, and cleanup flows (including retries/resets/transitions), so stored data reliably round-trips. * Hardened backend-mode reads for workflow steps, task columns, project settings, and prompt overrides. * Made verification-cache reads/writes complete reliably, including command-specific cache behavior. * **Tests** * Added PostgreSQL integration/regression coverage for refinement/duplication, sync residue, and verification caching. * **Chores** * Bumped `@earendil-works/pi-ai` and `@earendil-works/pi-coding-agent` to `^0.80.10`. --------- Co-authored-by: Claude Fable 5 --- .changeset/fix-refinement-backend-mode.md | 7 + packages/cli/package.json | 2 +- .../postgres/refine-duplicate-task.pg.test.ts | 116 +++++++++++++ .../sync-db-residue-backend.pg.test.ts | 155 ++++++++++++++++++ .../postgres/verification-cache.pg.test.ts | 59 +++++++ packages/core/src/store.ts | 28 ++-- packages/core/src/task-store/audit-ops.ts | 52 ++++-- .../core/src/task-store/branch-group-ops.ts | 21 ++- packages/core/src/task-store/lifecycle-ops.ts | 2 +- .../core/src/task-store/remaining-ops-1.ts | 63 +++++-- .../core/src/task-store/remaining-ops-4.ts | 57 ++++++- .../core/src/task-store/remaining-ops-5.ts | 43 ++++- .../core/src/task-store/remaining-ops-6.ts | 100 +++++++++-- .../core/src/task-store/remaining-ops-8.ts | 63 ++++++- .../src/task-store/workflow-workitems-ops.ts | 14 +- packages/dashboard/package.json | 8 +- .../routes/register-task-workflow-routes.ts | 2 +- packages/engine/package.json | 4 +- packages/engine/src/executor.ts | 13 +- packages/engine/src/merger.ts | 6 +- 20 files changed, 736 insertions(+), 79 deletions(-) create mode 100644 .changeset/fix-refinement-backend-mode.md create mode 100644 packages/core/src/__tests__/postgres/refine-duplicate-task.pg.test.ts create mode 100644 packages/core/src/__tests__/postgres/sync-db-residue-backend.pg.test.ts create mode 100644 packages/core/src/__tests__/postgres/verification-cache.pg.test.ts diff --git a/.changeset/fix-refinement-backend-mode.md b/.changeset/fix-refinement-backend-mode.md new file mode 100644 index 0000000000..4117427cd3 --- /dev/null +++ b/.changeset/fix-refinement-backend-mode.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix task refinement/duplication, merge verification, and workflow checkpoint persistence on PostgreSQL. +category: fix +dev: atomicCreateTaskJson now routes to the AsyncDataLayer in backend mode (fixing the refineTask/duplicateTask createTaskWithId paths that bypassed _createTaskInternal's backend routing), and the merger verification-cache ops (getVerificationCacheHit/recordVerificationCachePass) are now async with a PostgreSQL branch; the upsert targets verification_cache_pkey by constraint name since migration 0006 rebuilds project-schema PKs to lead with project_id. The full sync-SQLite residue sweep also ports workflow run-branch/step-instance persistence, branch progress, plugin column-transition hooks, getTaskColumns, getWorkflowStep/listWorkflowSteps stored-row reads, readRawProjectSettings, and listWorkflowPromptOverridesForProject to the async layer (several store methods became async: saveWorkflowRunBranch, loadWorkflowRunBranches, clearWorkflowRunBranches, saveWorkflowRunStepInstance, loadWorkflowRunStepInstances, clearWorkflowRunStepInstances, getBranchProgressByTask, readRawProjectSettings, listWorkflowPromptOverridesForProject). Also bumps @earendil-works/pi-ai and pi-coding-agent to ^0.80.10 — the FN-8142 pi SDK migration targeted APIs (ModelRuntime et al.) absent from the previously pinned 0.80.6, which broke the engine build. diff --git a/packages/cli/package.json b/packages/cli/package.json index 070ee674ae..49c7320caf 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -63,9 +63,9 @@ "@earendil-works/pi-ai": "^0.80.10", "@earendil-works/pi-coding-agent": "^0.80.10", "dockerode": "^4.0.12", + "electron": "^33.4.11", "embedded-postgres": "15.18.0-beta.17", "express": "^5.1.0", - "electron": "^33.4.11", "i18next": "^26.3.1", "ink": "^7.0.5", "ink-spinner": "^5.0.0", diff --git a/packages/core/src/__tests__/postgres/refine-duplicate-task.pg.test.ts b/packages/core/src/__tests__/postgres/refine-duplicate-task.pg.test.ts new file mode 100644 index 0000000000..5e5cf32a00 --- /dev/null +++ b/packages/core/src/__tests__/postgres/refine-duplicate-task.pg.test.ts @@ -0,0 +1,116 @@ +/** + * FNXC:PostgresOnlyDataAccess 2026-07-16-11:10: + * Regression: refineTask and duplicateTask create rows through the shared + * atomicCreateTaskJson helper via createTaskWithId callbacks, bypassing + * _createTaskInternal's backend routing. Before the fix, creating a refinement + * (or duplicate) in backend mode threw "TaskStore.db: SQLite Database is not + * available in backend mode". atomicCreateTaskJson now routes itself to the + * async layer, so both surfaces must persist against PostgreSQL. + */ +import { describe, it, expect } from "vitest"; +import { + pgDescribe, + createTaskStoreForTest, + type PgTestHarness, +} from "../../__test-utils__/pg-test-harness.js"; + +pgDescribe("refineTask / duplicateTask backend mode (PostgreSQL)", () => { + let harness: PgTestHarness | null = null; + + async function makeHarness(): Promise { + harness = await createTaskStoreForTest({ prefix: "fusion_refine_dup" }); + return harness; + } + + async function teardown(): Promise { + if (harness) { + await harness.teardown(); + harness = null; + } + } + + it("refineTask creates a refinement of a done task in backend mode", async () => { + const h = await makeHarness(); + try { + const source = await h.store.createTask({ + title: "Source feature", + description: "Original completed work", + column: "done", + }); + + const refined = await h.store.refineTask(source.id, "Please tighten the empty-state copy"); + + expect(refined.id).not.toBe(source.id); + expect(refined.sourceType).toBe("task_refine"); + expect(refined.sourceParentTaskId).toBe(source.id); + expect(refined.column).toBe("triage"); + expect(refined.dependencies).toEqual([source.id]); + expect(refined.description).toContain("Please tighten the empty-state copy"); + + // Round-trip through the async layer. + const fetched = await h.store.getTask(refined.id); + expect(fetched.id).toBe(refined.id); + expect(fetched.sourceType).toBe("task_refine"); + } finally { + await teardown(); + } + }); + + it("refineTask works for an in-review source task in backend mode", async () => { + const h = await makeHarness(); + try { + const source = await h.store.createTask({ + title: "In-review feature", + description: "Work awaiting review", + column: "in-review", + }); + + const refined = await h.store.refineTask(source.id, "Follow-up polish request"); + const fetched = await h.store.getTask(refined.id); + expect(fetched.sourceParentTaskId).toBe(source.id); + } finally { + await teardown(); + } + }); + + it("refineTask rejects a source task that is not done or in-review", async () => { + const h = await makeHarness(); + try { + const source = await h.store.createTask({ + title: "Live task", + description: "Still in progress", + column: "in-progress", + }); + await expect(h.store.refineTask(source.id, "too early")).rejects.toThrow(/must be in 'done' or 'in-review'/); + } finally { + await teardown(); + } + }); + + it("duplicateTask duplicates a task in backend mode", async () => { + const h = await makeHarness(); + try { + const source = await h.store.createTask({ + title: "Duplicable task", + description: "Task to duplicate", + }); + + const dup = await h.store.duplicateTask(source.id); + + expect(dup.id).not.toBe(source.id); + expect(dup.sourceType).toBe("task_duplicate"); + expect(dup.sourceParentTaskId).toBe(source.id); + expect(dup.description).toContain(`(Duplicated from ${source.id})`); + + const fetched = await h.store.getTask(dup.id); + expect(fetched.id).toBe(dup.id); + expect(fetched.sourceType).toBe("task_duplicate"); + } finally { + await teardown(); + } + }); +}); + +// Keep `describe` referenced so the import is not flagged as unused if the +// pgDescribe.skip path is taken in CI (no PG available). +void describe; diff --git a/packages/core/src/__tests__/postgres/sync-db-residue-backend.pg.test.ts b/packages/core/src/__tests__/postgres/sync-db-residue-backend.pg.test.ts new file mode 100644 index 0000000000..b9a23903d7 --- /dev/null +++ b/packages/core/src/__tests__/postgres/sync-db-residue-backend.pg.test.ts @@ -0,0 +1,155 @@ +/** + * FNXC:PostgresOnlyDataAccess 2026-07-16-12:50: + * Regression coverage for the sync-SQLite residue sweep: these TaskStore + * surfaces previously either threw "SQLite Database is not available in + * backend mode" or silently degraded to empty/no-op behind try/catch in + * backend mode. Each now routes to the AsyncDataLayer and must round-trip + * against PostgreSQL: + * - workflow run-branch persistence (save/load/clear + branch progress) + * - foreach step-instance persistence (save/load/upsert/clear) + * - getTaskColumns (active + archived + missing) + * - getWorkflowStep (stored row by id/templateId + built-in template) + * - listWorkflowSteps (stored rows, not just plugin steps) + * - readRawProjectSettings / listWorkflowPromptOverridesForProject + */ +import { describe, it, expect } from "vitest"; +import { + pgDescribe, + createTaskStoreForTest, + type PgTestHarness, +} from "../../__test-utils__/pg-test-harness.js"; + +pgDescribe("sync-db residue surfaces in backend mode (PostgreSQL)", () => { + let harness: PgTestHarness | null = null; + + async function makeHarness(): Promise { + harness = await createTaskStoreForTest({ prefix: "fusion_sync_residue" }); + return harness; + } + + async function teardown(): Promise { + if (harness) { + await harness.teardown(); + harness = null; + } + } + + it("persists, loads, and prunes workflow run branches; branch progress uses the latest run", async () => { + const h = await makeHarness(); + try { + const task = await h.store.createTask({ description: "branch persistence" }); + + await h.store.saveWorkflowRunBranch({ taskId: task.id, runId: "run-1", branchId: "b1", currentNodeId: "n1", status: "completed" }); + await h.store.saveWorkflowRunBranch({ taskId: task.id, runId: "run-2", branchId: "b1", currentNodeId: "n2", status: "running" }); + // Upsert flips the same key in place. + await h.store.saveWorkflowRunBranch({ taskId: task.id, runId: "run-2", branchId: "b1", currentNodeId: "n3", status: "completed" }); + + const run2 = await h.store.loadWorkflowRunBranches(task.id, "run-2"); + expect(run2).toHaveLength(1); + expect(run2[0]).toMatchObject({ branchId: "b1", currentNodeId: "n3", status: "completed" }); + + const progress = await h.store.getBranchProgressByTask([task.id]); + expect(progress.get(task.id)).toEqual([{ branchId: "b1", nodeId: "n3", status: "completed" }]); + + await h.store.clearWorkflowRunBranches(task.id, "run-2"); + expect(await h.store.loadWorkflowRunBranches(task.id, "run-1")).toHaveLength(0); + expect(await h.store.loadWorkflowRunBranches(task.id, "run-2")).toHaveLength(1); + } finally { + await teardown(); + } + }); + + it("persists, upserts, loads, and prunes foreach step instances", async () => { + const h = await makeHarness(); + try { + const task = await h.store.createTask({ description: "step-instance persistence" }); + const base = { taskId: task.id, runId: "run-1", foreachNodeId: "fe1", pinnedStepCount: 2 }; + + await h.store.saveWorkflowRunStepInstance({ ...base, stepIndex: 0, currentNodeId: "n1", status: "running", reworkCount: 0 } as never); + await h.store.saveWorkflowRunStepInstance({ ...base, stepIndex: 1, currentNodeId: "n1", status: "running", reworkCount: 1 } as never); + // Upsert the first row to completed. + await h.store.saveWorkflowRunStepInstance({ ...base, stepIndex: 0, currentNodeId: "n2", status: "completed", reworkCount: 2, integratedAt: "2026-07-16T00:00:00.000Z" } as never); + + const rows = await h.store.loadWorkflowRunStepInstances(task.id, "run-1"); + expect(rows).toHaveLength(2); + expect(rows[0]).toMatchObject({ stepIndex: 0, status: "completed", reworkCount: 2, integratedAt: "2026-07-16T00:00:00.000Z" }); + expect(rows[1]).toMatchObject({ stepIndex: 1, status: "running", reworkCount: 1 }); + + // keepRunId semantics: pruning keeps only the given run. + await h.store.saveWorkflowRunStepInstance({ ...base, runId: "run-2", stepIndex: 0, currentNodeId: "n1", status: "running", reworkCount: 0 } as never); + await h.store.clearWorkflowRunStepInstances(task.id, "run-2"); + expect(await h.store.loadWorkflowRunStepInstances(task.id, "run-1")).toHaveLength(0); + expect(await h.store.loadWorkflowRunStepInstances(task.id, "run-2")).toHaveLength(1); + + // No keepRunId clears everything. + await h.store.clearWorkflowRunStepInstances(task.id); + expect(await h.store.loadWorkflowRunStepInstances(task.id, "run-2")).toHaveLength(0); + } finally { + await teardown(); + } + }); + + it("getTaskColumns resolves active, archived, and missing ids", async () => { + const h = await makeHarness(); + try { + const active = await h.store.createTask({ description: "active", column: "in-progress" }); + const toArchive = await h.store.createTask({ description: "to archive", column: "done" }); + await h.store.archiveTask(toArchive.id, { cleanup: false }); + + const map = await h.store.getTaskColumns([active.id, toArchive.id, "FN-NOPE-1"]); + expect(map.get(active.id)).toBe("in-progress"); + expect(map.get(toArchive.id)).toBe("archived"); + expect(map.has("FN-NOPE-1")).toBe(false); + } finally { + await teardown(); + } + }); + + it("getWorkflowStep resolves stored rows by id and templateId, and built-in templates", async () => { + const h = await makeHarness(); + try { + const created = await h.store.createWorkflowStep({ + templateId: undefined, + name: "Residue Step", + description: "backend-mode stored step", + mode: "prompt", + phase: "pre-merge", + gateMode: "advisory", + prompt: "check things", + toolMode: "readonly", + enabled: true, + }); + + const byId = await h.store.getWorkflowStep(created.id); + expect(byId?.name).toBe("Residue Step"); + + // Stored rows appear in the listing (previously dropped in backend mode). + const listed = await h.store.listWorkflowSteps(); + expect(listed.map((s) => s.id)).toContain(created.id); + + // Unknown ids fall through to built-in templates or undefined — no throw. + const missing = await h.store.getWorkflowStep("WS-does-not-exist"); + expect(missing === undefined || typeof missing.id === "string").toBe(true); + } finally { + await teardown(); + } + }); + + it("readRawProjectSettings and listWorkflowPromptOverridesForProject read via the async layer", async () => { + const h = await makeHarness(); + try { + await h.store.updateSettings({ taskPrefix: "ZZ" }); + const raw = await h.store.readRawProjectSettings(); + expect(raw.taskPrefix).toBe("ZZ"); + + const overrides = await h.store.listWorkflowPromptOverridesForProject(); + expect(overrides).toEqual({}); + } finally { + await teardown(); + } + }); +}); + +// Keep `describe` referenced so the import is not flagged as unused if the +// pgDescribe.skip path is taken in CI (no PG available). +void describe; diff --git a/packages/core/src/__tests__/postgres/verification-cache.pg.test.ts b/packages/core/src/__tests__/postgres/verification-cache.pg.test.ts new file mode 100644 index 0000000000..ecf14ff299 --- /dev/null +++ b/packages/core/src/__tests__/postgres/verification-cache.pg.test.ts @@ -0,0 +1,59 @@ +/** + * FNXC:PostgresOnlyDataAccess 2026-07-16-11:45: + * Regression: the merger's deterministic-verification cache ops used the sync + * SQLite `store.db` unguarded, so the cache read threw "SQLite Database is not + * available in backend mode" on every merge with a resolvable tree sha (and + * cache writes were silently swallowed by the merger's try/catch). Both ops now + * route to the project-schema verification_cache table in backend mode. + */ +import { describe, it, expect } from "vitest"; +import { + pgDescribe, + createTaskStoreForTest, + type PgTestHarness, +} from "../../__test-utils__/pg-test-harness.js"; + +pgDescribe("verification cache backend mode (PostgreSQL)", () => { + let harness: PgTestHarness | null = null; + + async function makeHarness(): Promise { + harness = await createTaskStoreForTest({ prefix: "fusion_verif_cache" }); + return harness; + } + + async function teardown(): Promise { + if (harness) { + await harness.teardown(); + harness = null; + } + } + + it("records and reads back a verification pass in backend mode", async () => { + const h = await makeHarness(); + try { + const treeSha = "a".repeat(40); + + expect(await h.store.getVerificationCacheHit(treeSha, "pnpm test", "pnpm build")).toBeNull(); + + await h.store.recordVerificationCachePass(treeSha, "pnpm test", "pnpm build", "FN-1"); + const hit = await h.store.getVerificationCacheHit(treeSha, "pnpm test", "pnpm build"); + expect(hit).not.toBeNull(); + expect(hit!.taskId).toBe("FN-1"); + expect(typeof hit!.recordedAt).toBe("string"); + + // Different command set is a distinct cache key. + expect(await h.store.getVerificationCacheHit(treeSha, "pnpm test", "")).toBeNull(); + + // Re-recording the same key updates in place (upsert, no conflict throw). + await h.store.recordVerificationCachePass(treeSha, "pnpm test", "pnpm build", "FN-2"); + const updated = await h.store.getVerificationCacheHit(treeSha, "pnpm test", "pnpm build"); + expect(updated!.taskId).toBe("FN-2"); + } finally { + await teardown(); + } + }); +}); + +// Keep `describe` referenced so the import is not flagged as unused if the +// pgDescribe.skip path is taken in CI (no PG available). +void describe; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index ec8628d46d..aab00b833b 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -988,7 +988,7 @@ export class TaskStore extends EventEmitter { } /** Residual B (U13/U9): per-branch progress snapshots for the given tasks, */ - getBranchProgressByTask( taskIds: readonly string[], ): Map> { + async getBranchProgressByTask( taskIds: readonly string[], ): Promise>> { return getBranchProgressByTaskImpl(this, taskIds); } // FNXC:PostgresCutover 2026-07-04-00:00: facade delegates to async PG query in backend mode. @@ -1025,36 +1025,36 @@ export class TaskStore extends EventEmitter { } /** Persist (idempotent upsert) one branch's progress for a fan-out run (#1407). */ - saveWorkflowRunBranch(state: { taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; }): void { - saveWorkflowRunBranchImpl(this, state); + async saveWorkflowRunBranch(state: { taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; }): Promise { + return saveWorkflowRunBranchImpl(this, state); } /** Load persisted branch states for a run (crash-resume; #1407). */ - loadWorkflowRunBranches( taskId: string, runId: string, ): Array<{ + async loadWorkflowRunBranches( taskId: string, runId: string, ): Promise { + }>> { return loadWorkflowRunBranchesImpl(this, taskId, runId); } /** Prune stale branch rows for a task (#1412). */ - clearWorkflowRunBranches(taskId: string, keepRunId: string): void { - clearWorkflowRunBranchesImpl(this, taskId, keepRunId); + async clearWorkflowRunBranches(taskId: string, keepRunId: string): Promise { + return clearWorkflowRunBranchesImpl(this, taskId, keepRunId); } /** Persist (idempotent upsert) one step instance's run-state inside a foreach */ - saveWorkflowRunStepInstance( state: import("./types.js").WorkflowRunStepInstance, ): void { + async saveWorkflowRunStepInstance( state: import("./types.js").WorkflowRunStepInstance, ): Promise { return saveWorkflowRunStepInstanceImpl(this, state); } /** Load persisted step-instance run-state for a run (crash-resume; KTD-6). */ - loadWorkflowRunStepInstances( taskId: string, runId: string, ): import("./types.js").WorkflowRunStepInstance[] { + async loadWorkflowRunStepInstances( taskId: string, runId: string, ): Promise { return loadWorkflowRunStepInstancesImpl(this, taskId, runId); } - clearWorkflowRunStepInstances(taskId: string, keepRunId?: string): void { + async clearWorkflowRunStepInstances(taskId: string, keepRunId?: string): Promise { return clearWorkflowRunStepInstancesImpl(this, taskId, keepRunId); } @@ -1234,7 +1234,7 @@ export class TaskStore extends EventEmitter { public parseWorkflowPromptOverrideJson(raw: string | null | undefined): Record { return parseWorkflowPromptOverrideJsonImpl(this, raw); } - listWorkflowPromptOverridesForProject(): Record> { + async listWorkflowPromptOverridesForProject(): Promise>> { return listWorkflowPromptOverridesForProjectImpl(this); } getWorkflowPromptOverrides(workflowId: string, projectId: string): Record { @@ -2140,7 +2140,7 @@ Issue #2149 requires read-only type filtering to occur in the file-store before public async migrateMovedSettingsToWorkflowValuesOnce(): Promise { return migrateMovedSettingsImpl(this); } - public readRawProjectSettings(): Record { + public async readRawProjectSettings(): Promise> { return readRawProjectSettingsImpl(this); } public invalidateConfigCacheAfterMigration(): void { @@ -2644,12 +2644,12 @@ Issue #2149 requires read-only type filtering to occur in the file-store before // ── Verification Cache ──────────────────────────────────────────────────── /** Look up a previously recorded verification cache pass for a given tree sha */ - getVerificationCacheHit( treeSha: string, testCommand: string, buildCommand: string, ): { recordedAt: string; taskId: string | null } | null { + async getVerificationCacheHit( treeSha: string, testCommand: string, buildCommand: string, ): Promise<{ recordedAt: string; taskId: string | null } | null> { return getVerificationCacheHitImpl(this, treeSha, testCommand, buildCommand); } /** Record a successful verification pass for the given tree sha and commands. */ - recordVerificationCachePass( treeSha: string, testCommand: string, buildCommand: string, taskId: string, ): void { + async recordVerificationCachePass( treeSha: string, testCommand: string, buildCommand: string, taskId: string, ): Promise { return recordVerificationCachePassImpl(this, treeSha, testCommand, buildCommand, taskId); } diff --git a/packages/core/src/task-store/audit-ops.ts b/packages/core/src/task-store/audit-ops.ts index 7d5a5f9838..52669e43e5 100644 --- a/packages/core/src/task-store/audit-ops.ts +++ b/packages/core/src/task-store/audit-ops.ts @@ -12,6 +12,7 @@ import {findWorkflowColumn} from "../plugin-gate-verdict.js"; import {getTraitRegistry} from "../trait-registry.js"; import {makeTransitionPending} from "../transition-types.js"; import {writeTransitionPending} from "../transition-pending.js"; +import {writeTransitionPendingAsync} from "./async-transition-pending.js"; import type {WorkflowIr} from "../workflow-ir-types.js"; import "../builtin-traits.js"; import {toJson, fromJson} from "../db.js"; @@ -43,21 +44,41 @@ export async function runPluginColumnTransitionHooksImpl(store: TaskStore, taskI // mid-hook is recoverable. const hookIds = pending.map((p) => `${p.traitId}:${p.hookKind}`); const startedAt = Date.now(); - try { - writeTransitionPending( - store.db, - taskId, - makeTransitionPending(toColumn, ["default-workflow:postCommit", ...hookIds], startedAt), - ); - } catch { - // Marker bookkeeping is best-effort; proceed to run the hooks regardless. - } + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:20: + Backend mode previously threw on the sync store.db marker write / + readTaskFromDb here; callers (moves.ts, lifecycle-ops.ts recovery) swallow + the throw, so plugin onEnter/onExit column-transition hooks silently never + fired on PostgreSQL. Route both the marker bookkeeping and the non-locking + task read through the async layer. + */ + const writeMarker = async (remainingHookIds: string[]): Promise => { + try { + const marker = makeTransitionPending(toColumn, remainingHookIds, startedAt); + if (store.backendMode) { + await writeTransitionPendingAsync(store.asyncLayer!.db, taskId, marker); + } else { + writeTransitionPending(store.db, taskId, marker); + } + } catch { + // Marker bookkeeping is best-effort; proceed to run the hooks regardless. + } + }; + await writeMarker(["default-workflow:postCommit", ...hookIds]); // Read the task once for hook context. MUST be a non-locking read — this // runs inside `withTaskLock`, so `getTask` (which re-acquires the lock) - // would deadlock. `readTaskFromDb` is the in-lock-safe read. - const taskRow = store.readTaskFromDb(taskId, { includeDeleted: false }); - const taskDetail = taskRow as unknown as TaskDetail | undefined; + // would deadlock. `readTaskFromDb` is the in-lock-safe read (backend mode: + // raw readTaskRow + row conversion, same non-locking property). + let taskDetail: TaskDetail | undefined; + if (store.backendMode) { + const pgRow = await readTaskRow(store.asyncLayer!, taskId, { includeDeleted: false }); + taskDetail = pgRow + ? (store.rowToTask(store.pgRowToTaskRow(pgRow)) as unknown as TaskDetail) + : undefined; + } else { + taskDetail = store.readTaskFromDb(taskId, { includeDeleted: false }) as unknown as TaskDetail | undefined; + } const remaining = ["default-workflow:postCommit", ...hookIds]; for (const { traitId, hookKind } of pending) { @@ -97,11 +118,8 @@ export async function runPluginColumnTransitionHooksImpl(store: TaskStore, taskI // Mark this hook complete in the marker (whether it ran, degraded, or threw). const idx = remaining.indexOf(`${traitId}:${hookKind}`); if (idx >= 0) remaining.splice(idx, 1); - try { - writeTransitionPending(store.db, taskId, makeTransitionPending(toColumn, remaining, startedAt)); - } catch { - // Best-effort progress bookkeeping; the final clear is the backstop. - } + // Best-effort progress bookkeeping; the final clear is the backstop. + await writeMarker(remaining); } } diff --git a/packages/core/src/task-store/branch-group-ops.ts b/packages/core/src/task-store/branch-group-ops.ts index 7052da61cc..2d2f7e03d4 100644 --- a/packages/core/src/task-store/branch-group-ops.ts +++ b/packages/core/src/task-store/branch-group-ops.ts @@ -19,7 +19,26 @@ import {listArtifacts as listArtifactsAsync} from "./async-comments-attachments. import { and, eq, isNull, ne, sql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; -export function saveWorkflowRunBranchImpl(store: TaskStore, state: { taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; }): void { +export async function saveWorkflowRunBranchImpl(store: TaskStore, state: { taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; }): Promise { + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:15: + Backend mode previously swallowed the sync throw, so parallel-branch + checkpoints were never persisted on PostgreSQL. ON CONFLICT targets the PK + by constraint name because project-schema PKs lead with project_id (which + itself comes from the column's current_setting default under RLS). + */ + if (store.backendMode) { + await store.asyncLayer!.db.execute(sql` + INSERT INTO project.workflow_run_branches + (task_id, run_id, branch_id, current_node_id, status, updated_at) + VALUES (${state.taskId}, ${state.runId}, ${state.branchId}, ${state.currentNodeId}, ${state.status}, ${new Date().toISOString()}) + ON CONFLICT ON CONSTRAINT workflow_run_branches_pkey DO UPDATE SET + current_node_id = EXCLUDED.current_node_id, + status = EXCLUDED.status, + updated_at = EXCLUDED.updated_at + `); + return; + } try { store.db .prepare( diff --git a/packages/core/src/task-store/lifecycle-ops.ts b/packages/core/src/task-store/lifecycle-ops.ts index 8337acb025..037243d998 100644 --- a/packages/core/src/task-store/lifecycle-ops.ts +++ b/packages/core/src/task-store/lifecycle-ops.ts @@ -912,7 +912,7 @@ export async function migrateMovedSettingsImpl(store: TaskStore): Promise const projectId = store.getWorkflowSettingsProjectId(); // (1) Snapshot CUSTOMIZED moved keys from RAW persisted project + global stores. - const rawProjectSettings = store.readRawProjectSettings(); + const rawProjectSettings = await store.readRawProjectSettings(); let rawGlobalSettings: Record = {}; try { rawGlobalSettings = await store.globalSettingsStore.readRaw(); diff --git a/packages/core/src/task-store/remaining-ops-1.ts b/packages/core/src/task-store/remaining-ops-1.ts index 5debd4e035..c47d43cd78 100644 --- a/packages/core/src/task-store/remaining-ops-1.ts +++ b/packages/core/src/task-store/remaining-ops-1.ts @@ -778,21 +778,30 @@ export async function updateIssueInfoImpl(store: TaskStore, id: string, issueInf export async function listWorkflowStepsImpl(store: TaskStore): Promise { if (store.workflowStepsCache) return store.workflowStepsCache; - /* - * FNXC:SqliteFinalRemoval 2026-06-24-15:40: - * In backend mode (PostgreSQL), the workflow_steps table read path has not - * been converted to the async Drizzle helper yet. Return only the plugin- - * contributed steps (which are in-memory, not DB-backed) so task creation - * does not throw when auto-defaulting workflow steps. The stored steps are - * empty until the async workflow-step helper is implemented. This matches - * the existing fail-soft behavior (the catch block logged a warning and - * continued with no default steps). - */ if (store.backendMode) { + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:30: + Backend mode reads stored steps from project.workflow_steps via the async + layer, replacing the SqliteFinalRemoval-era interim fail-soft that + returned plugin-contributed steps only (stored steps were dropped until + the async helper existed). Listing parity with the sync branch below: + compiled-step rows stay filtered out, plugin steps are appended. + */ + const table = schema.project.workflowSteps; + const pgRows = await store.asyncLayer!.db + .select() + .from(table) + .orderBy(table.createdAt); + const storedPgSteps = pgRows + .map((row) => store.applyLegacyWorkflowStepOverrides(store.toStoredWorkflowStep({ + ...row, + migrated_fragment_id: row.migratedFragmentId, + } as unknown as Parameters[0]))) + .filter((step) => !step.templateId?.startsWith(WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX)); const pluginSteps = store._pluginWorkflowStepTemplates .map(({ template }) => store.resolvePluginWorkflowStep(template.id)) .filter((step): step is import("../types.js").WorkflowStep => Boolean(step)); - store.workflowStepsCache = pluginSteps; + store.workflowStepsCache = [...storedPgSteps, ...pluginSteps]; return store.workflowStepsCache; } const rows = store.db.prepare("SELECT * FROM workflow_steps ORDER BY createdAt ASC").all() as Array<{ @@ -834,6 +843,38 @@ export async function getWorkflowStepImpl(store: TaskStore, id: string): Promise } } + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:30: + Backend mode previously fell through to the sync store.db read and threw + "SQLite Database is not available" (reachable via ensureWorkflowStepForTemplate + and the executor's workflow-step gate). Async lookup: by id, then by + templateId (earliest created), then built-in template — same resolution + order as the sync branch. + */ + if (store.backendMode) { + const table = schema.project.workflowSteps; + const mapRow = (row: typeof table.$inferSelect) => + store.applyLegacyWorkflowStepOverrides(store.toStoredWorkflowStep({ + ...row, + migrated_fragment_id: row.migratedFragmentId, + } as unknown as Parameters[0])); + const byIdRows = await store.asyncLayer!.db + .select() + .from(table) + .where(eq(table.id, id)) + .limit(1); + if (byIdRows[0]) return mapRow(byIdRows[0]); + const byTemplateRows = await store.asyncLayer!.db + .select() + .from(table) + .where(eq(table.templateId, id)) + .orderBy(table.createdAt) + .limit(1); + if (byTemplateRows[0]) return mapRow(byTemplateRows[0]); + const pgTemplate = store.getBuiltInWorkflowTemplate(id); + return pgTemplate ? store.toBuiltInWorkflowStep(pgTemplate) : undefined; + } + const byId = store.db.prepare("SELECT * FROM workflow_steps WHERE id = ?").get(id) as | { id: string; diff --git a/packages/core/src/task-store/remaining-ops-4.ts b/packages/core/src/task-store/remaining-ops-4.ts index fe84798942..001e2d098a 100644 --- a/packages/core/src/task-store/remaining-ops-4.ts +++ b/packages/core/src/task-store/remaining-ops-4.ts @@ -13,7 +13,8 @@ import * as schema from "../postgres/schema/index.js"; import type {MoveTaskOptions, MoveTaskInternalOptions} from "../store.js"; import {TASK_BRANCH_CONTEXT_METADATA_KEY} from "../store.js"; import {randomUUID} from "node:crypto"; -import {and, eq, inArray} from "drizzle-orm"; +import {and, eq, inArray, isNull} from "drizzle-orm"; +import {filterArchived as filterArchivedAsync} from "../async-archive-db.js"; import type {Task, TaskCreateInput, Column, ColumnId, TaskDocumentWithTask, RunMutationContext, TaskCommitAssociation, GoalCitation, GoalCitationInput, TaskBranchAssignmentMode, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind} from "../types.js"; import {COLUMNS} from "../types.js"; import {parseWorkflowIr, serializeWorkflowIr} from "../workflow-ir.js"; @@ -334,6 +335,37 @@ export async function getTaskColumnsImpl(store: TaskStore, ids: string[]): Promi } const uniqueIds = [...new Set(ids)]; + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:25: + Backend mode previously threw here (dashboard's caller swallowed it, so + every agent-linked task read as non-terminal on PostgreSQL). Async reads: + live columns from project.tasks, then archive membership for the misses. + */ + if (store.backendMode) { + const layer = store.asyncLayer!; + const rows = await layer.db + .select({ id: schema.project.tasks.id, column: schema.project.tasks.column }) + .from(schema.project.tasks) + .where(and(inArray(schema.project.tasks.id, uniqueIds), isNull(schema.project.tasks.deletedAt))); + const activeByIdPg = new Map(); + for (const row of rows) { + activeByIdPg.set(row.id, row.column as Column); + } + const missingPg = uniqueIds.filter((id) => !activeByIdPg.has(id)); + const archivedPg = missingPg.length > 0 + ? await filterArchivedAsync(layer.db, missingPg, layer.projectId) + : new Set(); + const resultPg = new Map(); + for (const id of uniqueIds) { + const activeColumn = activeByIdPg.get(id); + if (activeColumn !== undefined) { + resultPg.set(id, activeColumn); + } else if (archivedPg.has(id)) { + resultPg.set(id, "archived"); + } + } + return resultPg; + } const placeholders = uniqueIds.map(() => "?").join(","); const rows = store.db .prepare(`SELECT id, "column" FROM tasks WHERE id IN (${placeholders}) AND ${TaskStore.ACTIVE_TASKS_WHERE}`) @@ -419,8 +451,29 @@ export async function updateTaskCustomFieldsImpl(store: TaskStore, taskId: strin }); } -export function listWorkflowPromptOverridesForProjectImpl(store: TaskStore): Record> { +export async function listWorkflowPromptOverridesForProjectImpl(store: TaskStore): Promise>> { const projectId = store.getWorkflowSettingsProjectId(); + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:25: backend branch added so + // this public method cannot throw the sync-SQLite error on PostgreSQL. + if (store.backendMode) { + const table = schema.project.workflowPromptOverrides; + const pgRows = await store.asyncLayer!.db + .select({ workflowId: table.workflowId, overrides: table.overrides }) + .from(table) + .where(eq(table.projectId, projectId)); + const outPg: Record> = {}; + for (const row of pgRows) { + // jsonb column: drizzle returns the parsed object (getWorkflowPromptOverridesAsyncImpl parity). + const overrides = row.overrides; + if (!overrides || typeof overrides !== "object" || Array.isArray(overrides)) continue; + const entry: Record = {}; + for (const [nodeId, value] of Object.entries(overrides as Record)) { + if (typeof value === "string" && value.trim()) entry[nodeId] = value; + } + outPg[row.workflowId] = entry; + } + return outPg; + } const rows = store.db .prepare("SELECT workflowId, overrides FROM workflow_prompt_overrides WHERE projectId = ?") .all(projectId) as Array<{ workflowId: string; overrides: string }>; diff --git a/packages/core/src/task-store/remaining-ops-5.ts b/packages/core/src/task-store/remaining-ops-5.ts index 9b6bdf0236..e94ab5e146 100644 --- a/packages/core/src/task-store/remaining-ops-5.ts +++ b/packages/core/src/task-store/remaining-ops-5.ts @@ -23,7 +23,7 @@ import { type TaskIdIntegrityReport, detectTaskIdIntegrityAnomalies } from "../t import { createBranchGroup as createBranchGroupAsync } from "./async-branch-groups.js"; import { findLiveLineageChildren as findLiveLineageChildrenAsync } from "./async-lifecycle.js"; import { recordRunAuditEvent as recordRunAuditEventAsync } from "./async-audit.js"; -import { readTaskRow } from "./async-persistence.js"; +import { insertTaskRowInTransaction, isTaskIdConflictError, readTaskRow, readTaskRowInTransaction } from "./async-persistence.js"; import { TASK_PERSIST_SQL_COLUMNS, TASK_UPSERT_SQL_ASSIGNMENTS, type TaskRow } from "./persistence.js"; import { purgeTaskWorkflowSelectionRowsAsyncImpl } from "./remaining-ops-8.js"; import { ConfigRow } from "./row-types.js"; @@ -674,6 +674,41 @@ export function getMalformedTaskMetadataReasonImpl(store: TaskStore, task: Parti export async function atomicCreateTaskJsonImpl(store: TaskStore, dir: string, task: Task, operation: string): Promise { const id = store.getTaskIdFromDir(dir); + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-11:05: + refineTask and duplicateTask create rows through this shared helper via their + createTaskWithId callbacks, bypassing _createTaskInternal's backend routing, so + creating a refinement in backend mode threw "SQLite Database is not available". + This helper must route itself: soft-delete conflict check + non-destructive + insert in one async transaction (parity with the sync transactionImmediate + block below), with unique_violation normalized to "Task ID already exists". + */ + if (store.backendMode) { + const layer = store.asyncLayer!; + const context = store.createTaskPersistSerializationContext(task); + let backendDeletedAt: string | undefined; + try { + await layer.transactionImmediate(async (tx) => { + const pgRow = await readTaskRowInTransaction(tx, id, { includeDeleted: true }, layer.projectId); + if (pgRow) { + backendDeletedAt = store.getSoftDeletedWriteConflict(id, task, store.pgRowToTaskRow(pgRow)); + if (backendDeletedAt) return; + } + await insertTaskRowInTransaction(tx, task as unknown as Record, context, layer.projectId); + }); + } catch (error) { + if (isTaskIdConflictError(error)) { + store.logTaskCreateConflict(task, operation, error); + throw new Error(`Task ID already exists: ${task.id}`); + } + throw error; + } + if (backendDeletedAt) { + store.throwSoftDeletedWriteBlocked(id, backendDeletedAt, operation); + } + await store.writeTaskJsonFile(dir, task); + return; + } let deletedAt: string | undefined; store.db.transactionImmediate(() => { deletedAt = store.getSoftDeletedWriteConflict(id, task); @@ -761,6 +796,12 @@ export function toBuiltInWorkflowStepImpl(store: TaskStore, template: import(".. } export function getLegacyWorkflowStepSnapshotImpl(store: TaskStore, id: string, templateId?: string): Record | undefined { + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:55: the legacy snapshot lives + // only in the pre-migration SQLite config.workflowSteps JSON blob; a + // PostgreSQL deployment has no legacy snapshot, so overrides never apply. + if (store.backendMode) { + return undefined; + } const row = store.db .prepare("SELECT workflowSteps FROM config WHERE id = 1") .get() as { workflowSteps?: string | null } | undefined; diff --git a/packages/core/src/task-store/remaining-ops-6.ts b/packages/core/src/task-store/remaining-ops-6.ts index 7e12e694b5..2988aecfcb 100644 --- a/packages/core/src/task-store/remaining-ops-6.ts +++ b/packages/core/src/task-store/remaining-ops-6.ts @@ -260,11 +260,48 @@ export async function recordPrThreadOutcomeImpl(store: TaskStore, store.db.bumpLastModified(); } -export function getBranchProgressByTaskImpl(store: TaskStore, +export async function getBranchProgressByTaskImpl(store: TaskStore, taskIds: readonly string[], - ): Map> { + ): Promise>> { const result = new Map>(); if (taskIds.length === 0) return result; + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:10: + Backend mode previously fell into the sync catch below and silently + returned an empty map, dropping branchProgress from every task payload on + PostgreSQL. Read the rows async and resolve the winning (latest updatedAt, + runId tie-break) run per task in JS — per-task row counts are small. + */ + if (store.backendMode) { + const table = schema.project.workflowRunBranches; + const rows = await store.asyncLayer!.db + .select({ + taskId: table.taskId, + runId: table.runId, + branchId: table.branchId, + nodeId: table.currentNodeId, + status: table.status, + updatedAt: table.updatedAt, + }) + .from(table) + .where(inArray(table.taskId, taskIds as string[])); + const latestRunByTask = new Map(); + for (const row of rows) { + const current = latestRunByTask.get(row.taskId); + if (!current + || row.updatedAt > current.updatedAt + || (row.updatedAt === current.updatedAt && row.runId > current.runId)) { + latestRunByTask.set(row.taskId, { runId: row.runId, updatedAt: row.updatedAt }); + } + } + for (const row of rows) { + if (latestRunByTask.get(row.taskId)?.runId !== row.runId) continue; + const list = result.get(row.taskId) ?? []; + list.push({ branchId: row.branchId, nodeId: row.nodeId, status: row.status }); + result.set(row.taskId, list); + } + return result; + } try { // Skip entirely when the table has no rows (cheap existence probe). const any = store.db @@ -327,16 +364,41 @@ export function getBranchProgressByTaskImpl(store: TaskStore, return result; } -export function loadWorkflowRunBranchesImpl(store: TaskStore, +export async function loadWorkflowRunBranchesImpl(store: TaskStore, taskId: string, runId: string, - ): Array<{ + ): Promise { + }>> { + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-12:10: + Backend mode previously returned [] from the sync catch, so parallel-branch + workflow runs lost their crash-recovery checkpoints on PostgreSQL. + */ + if (store.backendMode) { + const table = schema.project.workflowRunBranches; + const rows = await store.asyncLayer!.db + .select({ + taskId: table.taskId, + runId: table.runId, + branchId: table.branchId, + currentNodeId: table.currentNodeId, + status: table.status, + }) + .from(table) + .where(and(eq(table.taskId, taskId), eq(table.runId, runId))); + return rows as Array<{ + taskId: string; + runId: string; + branchId: string; + currentNodeId: string; + status: "running" | "completed" | "failed" | "aborted"; + }>; + } try { const rows = store.db .prepare( @@ -357,9 +419,19 @@ export function loadWorkflowRunBranchesImpl(store: TaskStore, } } -export function saveWorkflowRunStepInstanceImpl(store: TaskStore, +export async function saveWorkflowRunStepInstanceImpl(store: TaskStore, state: import("../types.js").WorkflowRunStepInstance, - ): void { + ): Promise { + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-13:40: + Backend mode previously swallowed the sync throw, so foreach step-instance + checkpoints were never persisted on PostgreSQL. Delegate to the FN-8157 + async sibling (single PG code path); its !backendMode branch routes back + here, guarded so there is no recursion. + */ + if (store.backendMode) { + return saveWorkflowRunStepInstanceAsyncImpl(store, state); + } try { store.db .prepare( @@ -397,10 +469,14 @@ export function saveWorkflowRunStepInstanceImpl(store: TaskStore, } } -export function loadWorkflowRunStepInstancesImpl(store: TaskStore, +export async function loadWorkflowRunStepInstancesImpl(store: TaskStore, taskId: string, runId: string, - ): import("../types.js").WorkflowRunStepInstance[] { + ): Promise { + // FNXC:PostgresOnlyDataAccess 2026-07-16-13:40: see saveWorkflowRunStepInstanceImpl. + if (store.backendMode) { + return loadWorkflowRunStepInstancesAsyncImpl(store, taskId, runId); + } try { const rows = store.db .prepare( @@ -416,7 +492,11 @@ export function loadWorkflowRunStepInstancesImpl(store: TaskStore, } } -export function clearWorkflowRunStepInstancesImpl(store: TaskStore, taskId: string, keepRunId?: string): void { +export async function clearWorkflowRunStepInstancesImpl(store: TaskStore, taskId: string, keepRunId?: string): Promise { + // FNXC:PostgresOnlyDataAccess 2026-07-16-13:40: see saveWorkflowRunStepInstanceImpl. + if (store.backendMode) { + return clearWorkflowRunStepInstancesAsyncImpl(store, taskId, keepRunId); + } try { if (keepRunId === undefined) { store.db diff --git a/packages/core/src/task-store/remaining-ops-8.ts b/packages/core/src/task-store/remaining-ops-8.ts index 585ecb11d8..515e1bd2b0 100644 --- a/packages/core/src/task-store/remaining-ops-8.ts +++ b/packages/core/src/task-store/remaining-ops-8.ts @@ -31,7 +31,7 @@ import { compactTaskActivityLog } from "./comments.js"; import { type TaskRow } from "./persistence.js"; import { ActivityLogRow } from "./row-types.js"; import { ActivityEventType, ActivityLogEntry, AgentLogEntry, ArchivedTaskEntry, DEFAULT_SETTINGS, Settings } from "../types.js"; -import { and, eq, inArray, isNull } from "drizzle-orm"; +import { and, eq, inArray, isNull, sql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; import { normalizeWorkflowIcon, type StoredWorkflowRow, type WorkflowDefinition, type WorkflowDefinitionInput, type WorkflowNodeLayout } from "../workflow-definition-types.js"; import { WorkflowIr } from "../workflow-ir-types.js"; @@ -76,7 +76,21 @@ export async function importLegacyAgentLogsOnceImpl(store: TaskStore): Promise { +export async function readRawProjectSettingsImpl(store: TaskStore): Promise> { + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:35: backend mode previously + // returned {} from the catch below, hiding raw persisted project settings + // on PostgreSQL. Read the config row via the async layer instead. + if (store.backendMode) { + try { + const config = await readProjectConfig(store.asyncLayer!); + const settings = config.settings; + return settings && typeof settings === "object" && !Array.isArray(settings) + ? (settings as Record) + : {}; + } catch { + return {}; + } + } try { const row = store.db.prepare("SELECT settings FROM config WHERE id = 1").get() as | { settings: string } @@ -984,13 +998,33 @@ export function getExperimentSessionStoreImpl(store: TaskStore): ExperimentSessi return store.experimentSessionStore; } -export function getVerificationCacheHitImpl(store: TaskStore, +/* +FNXC:PostgresOnlyDataAccess 2026-07-16-11:40: +The merger's deterministic-verification cache read at merger.ts ran the sync +SQLite branch unguarded, so every backend-mode merge with a resolvable tree sha +threw "SQLite Database is not available". Both cache ops are now async and route +to the project-schema verification_cache table in backend mode. +*/ +export async function getVerificationCacheHitImpl(store: TaskStore, treeSha: string, testCommand: string, buildCommand: string, - ): { recordedAt: string; taskId: string | null } | null { + ): Promise<{ recordedAt: string; taskId: string | null } | null> { const normalizedTest = testCommand ?? ""; const normalizedBuild = buildCommand ?? ""; + if (store.backendMode) { + const table = schema.project.verificationCache; + const rows = await store.asyncLayer!.db + .select({ recordedAt: table.recordedAt, taskId: table.taskId }) + .from(table) + .where(and( + eq(table.treeSha, treeSha), + eq(table.testCommand, normalizedTest), + eq(table.buildCommand, normalizedBuild), + )) + .limit(1); + return rows[0] ?? null; + } const row = store.db .prepare( `SELECT recordedAt, taskId FROM verification_cache @@ -1002,15 +1036,32 @@ export function getVerificationCacheHitImpl(store: TaskStore, return row ?? null; } -export function recordVerificationCachePassImpl(store: TaskStore, +export async function recordVerificationCachePassImpl(store: TaskStore, treeSha: string, testCommand: string, buildCommand: string, taskId: string, - ): void { + ): Promise { const normalizedTest = testCommand ?? ""; const normalizedBuild = buildCommand ?? ""; const recordedAt = new Date().toISOString(); + if (store.backendMode) { + /* + FNXC:PostgresOnlyDataAccess 2026-07-16-11:55: + Target the PK by constraint name: migration 0006_project_ownership + rebuilds every project-schema PK to lead with project_id, so a + column-list ON CONFLICT on the three logical key columns cannot infer + the arbiter index (42P10). project_id itself comes from the column's + current_setting('fusion.project_id') default under RLS. + */ + await store.asyncLayer!.db.execute(sql` + INSERT INTO project.verification_cache (tree_sha, test_command, build_command, recorded_at, task_id) + VALUES (${treeSha}, ${normalizedTest}, ${normalizedBuild}, ${recordedAt}, ${taskId}) + ON CONFLICT ON CONSTRAINT verification_cache_pkey + DO UPDATE SET recorded_at = EXCLUDED.recorded_at, task_id = EXCLUDED.task_id + `); + return; + } store.db .prepare( `INSERT OR REPLACE INTO verification_cache (treeSha, testCommand, buildCommand, recordedAt, taskId) diff --git a/packages/core/src/task-store/workflow-workitems-ops.ts b/packages/core/src/task-store/workflow-workitems-ops.ts index 46f1ea15da..9bc40edb82 100644 --- a/packages/core/src/task-store/workflow-workitems-ops.ts +++ b/packages/core/src/task-store/workflow-workitems-ops.ts @@ -12,8 +12,20 @@ import type {Task, MergeRequestWorkflowProjectionOptions, WorkflowWorkItem, Work import "../builtin-traits.js"; import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import {recordRunAuditEvent as recordRunAuditEventAsync} from "../postgres/data-layer.js"; +import {and, eq, ne} from "drizzle-orm"; +import * as schema from "../postgres/schema/index.js"; -export function clearWorkflowRunBranchesImpl(store: TaskStore, taskId: string, keepRunId: string): void { +export async function clearWorkflowRunBranchesImpl(store: TaskStore, taskId: string, keepRunId: string): Promise { + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:15: backend mode previously + // swallowed the sync throw, so stale-run branch rows were never pruned on + // PostgreSQL. + if (store.backendMode) { + const table = schema.project.workflowRunBranches; + await store.asyncLayer!.db + .delete(table) + .where(and(eq(table.taskId, taskId), ne(table.runId, keepRunId))); + return; + } try { store.db .prepare( diff --git a/packages/dashboard/package.json b/packages/dashboard/package.json index c2c6be9830..05928ed791 100644 --- a/packages/dashboard/package.json +++ b/packages/dashboard/package.json @@ -110,13 +110,14 @@ "@fusion-plugin-examples/cli-printing-press": "workspace:*", "@fusion-plugin-examples/compound-engineering": "workspace:*", "@fusion-plugin-examples/cursor-runtime": "workspace:*", - "@fusion-plugin-examples/grok-runtime": "workspace:*", - "@fusion-plugin-examples/omp-runtime": "workspace:*", "@fusion-plugin-examples/dependency-graph": "workspace:*", "@fusion-plugin-examples/droid-runtime": "workspace:*", + "@fusion-plugin-examples/grok-runtime": "workspace:*", "@fusion-plugin-examples/hermes-runtime": "workspace:*", + "@fusion-plugin-examples/omp-runtime": "workspace:*", "@fusion-plugin-examples/openclaw-runtime": "workspace:*", "@fusion-plugin-examples/paperclip-runtime": "workspace:*", + "@fusion-plugin-examples/quality": "workspace:*", "@fusion-plugin-examples/roadmap": "workspace:*", "@fusion/core": "workspace:*", "@fusion/engine": "workspace:*", @@ -150,8 +151,7 @@ "remark-gfm": "^4.0.1", "unified": "^11.0.5", "ws": "^8.18.0", - "zod": "^3.25.76", - "@fusion-plugin-examples/quality": "workspace:*" + "zod": "^3.25.76" }, "devDependencies": { "@testing-library/jest-dom": "^6.9.1", diff --git a/packages/dashboard/src/routes/register-task-workflow-routes.ts b/packages/dashboard/src/routes/register-task-workflow-routes.ts index 01d4ff2e03..52fc484e47 100644 --- a/packages/dashboard/src/routes/register-task-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-task-workflow-routes.ts @@ -953,7 +953,7 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork try { const settings = await scopedStore.getSettingsFast(); if (isWorkflowColumnsEnabled(settings) && tasks.length > 0) { - const byTask = scopedStore.getBranchProgressByTask(tasks.map((t) => t.id)); + const byTask = await scopedStore.getBranchProgressByTask(tasks.map((t) => t.id)); if (byTask.size > 0) { tasks = tasks.map((task) => { const branchProgress = byTask.get(task.id); diff --git a/packages/engine/package.json b/packages/engine/package.json index 7100323dcb..5f0786fc4f 100644 --- a/packages/engine/package.json +++ b/packages/engine/package.json @@ -38,10 +38,10 @@ "test:watch": "vitest src/__tests__/executor-*.test.ts --watch" }, "dependencies": { - "@fusion/core": "workspace:*", - "@fusion/pi-claude-cli": "workspace:*", "@earendil-works/pi-ai": "^0.80.10", "@earendil-works/pi-coding-agent": "^0.80.10", + "@fusion/core": "workspace:*", + "@fusion/pi-claude-cli": "workspace:*", "@modelcontextprotocol/sdk": "^1.0.0", "cron-parser": "^5.5.0", "esbuild": "^0.25.12", diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 2e896502b4..e95e3759fa 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -5596,15 +5596,18 @@ export class TaskExecutor { * mixed/partial store never throws into the run. */ private buildBranchPersistence(): WorkflowBranchPersistence | undefined { + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:40: the store methods are now + // async (PostgreSQL routing); the persistence interfaces already accept + // Promise-returning impls and await them. const store = this.store as unknown as { - saveWorkflowRunBranch?: (state: WorkflowBranchRunState) => void; - loadWorkflowRunBranches?: (taskId: string, runId: string) => WorkflowBranchRunState[]; - clearWorkflowRunBranches?: (taskId: string, keepRunId: string) => void; + saveWorkflowRunBranch?: (state: WorkflowBranchRunState) => void | Promise; + loadWorkflowRunBranches?: (taskId: string, runId: string) => WorkflowBranchRunState[] | Promise; + clearWorkflowRunBranches?: (taskId: string, keepRunId: string) => void | Promise; }; if (typeof store.saveWorkflowRunBranch !== "function") return undefined; return { saveBranchState: (state) => store.saveWorkflowRunBranch?.(state), - loadBranchStates: (taskId, runId) => store.loadWorkflowRunBranches?.(taskId, runId) ?? [], + loadBranchStates: async (taskId, runId) => (await store.loadWorkflowRunBranches?.(taskId, runId)) ?? [], clearStaleBranchStates: (taskId, keepRunId) => store.clearWorkflowRunBranches?.(taskId, keepRunId), }; } @@ -5616,6 +5619,8 @@ export class TaskExecutor { * fully in-memory — purely additive, same posture as buildBranchPersistence. */ private buildStepInstancePersistence(): WorkflowStepInstancePersistence | undefined { + // FNXC:PostgresOnlyDataAccess 2026-07-16-12:40: async store methods; the + // persistence interface awaits Promise-returning impls. const store = this.store as unknown as { saveWorkflowRunStepInstanceAsync?: (state: WorkflowStepInstanceState) => Promise; loadWorkflowRunStepInstancesAsync?: (taskId: string, runId: string) => Promise; diff --git a/packages/engine/src/merger.ts b/packages/engine/src/merger.ts index c469b98904..89b529a58b 100644 --- a/packages/engine/src/merger.ts +++ b/packages/engine/src/merger.ts @@ -770,7 +770,7 @@ async function runDeterministicVerification( } if (treeSha) { - const cacheHit = store.getVerificationCacheHit(treeSha, effectiveTestCommand, effectiveBuildCommand); + const cacheHit = await store.getVerificationCacheHit(treeSha, effectiveTestCommand, effectiveBuildCommand); if (cacheHit) { const sha7 = treeSha.slice(0, 7); const msg = `Skipping deterministic verification — cached pass for tree ${sha7} (recorded at ${cacheHit.recordedAt}, by ${cacheHit.taskId ?? "unknown"})`; @@ -1000,7 +1000,7 @@ async function runDeterministicVerification( // ── Record cache pass ────────────────────────────────────────────────── if (treeSha) { try { - store.recordVerificationCachePass(treeSha, effectiveTestCommand, effectiveBuildCommand, taskId); + await store.recordVerificationCachePass(treeSha, effectiveTestCommand, effectiveBuildCommand, taskId); mergerLog.log(`${taskId}: Recorded verification pass for tree ${treeSha.slice(0, 7)}`); await store.logEntry(taskId, `Recorded verification pass for tree ${treeSha.slice(0, 7)}`); } catch (err) { @@ -9125,7 +9125,7 @@ export async function aiMergeTask( }); const treeSha = treeOut.trim(); if (!treeSha) continue; - const cacheHit = store.getVerificationCacheHit(treeSha, effectiveTestCommand ?? "", effectiveBuildCommand ?? ""); + const cacheHit = await store.getVerificationCacheHit(treeSha, effectiveTestCommand ?? "", effectiveBuildCommand ?? ""); if (cacheHit) { verificationPassed = true; break;