fix: refineTask/duplicateTask fail in backend (PostgreSQL) mode (#2253)
## 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) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## 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`. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
7
.changeset/fix-refinement-backend-mode.md
Normal file
7
.changeset/fix-refinement-backend-mode.md
Normal file
@@ -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.
|
||||
@@ -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",
|
||||
|
||||
@@ -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<PgTestHarness> {
|
||||
harness = await createTaskStoreForTest({ prefix: "fusion_refine_dup" });
|
||||
return harness;
|
||||
}
|
||||
|
||||
async function teardown(): Promise<void> {
|
||||
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;
|
||||
@@ -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<PgTestHarness> {
|
||||
harness = await createTaskStoreForTest({ prefix: "fusion_sync_residue" });
|
||||
return harness;
|
||||
}
|
||||
|
||||
async function teardown(): Promise<void> {
|
||||
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;
|
||||
@@ -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<PgTestHarness> {
|
||||
harness = await createTaskStoreForTest({ prefix: "fusion_verif_cache" });
|
||||
return harness;
|
||||
}
|
||||
|
||||
async function teardown(): Promise<void> {
|
||||
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;
|
||||
@@ -988,7 +988,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
}
|
||||
|
||||
/** Residual B (U13/U9): per-branch progress snapshots for the given tasks, */
|
||||
getBranchProgressByTask( taskIds: readonly string[], ): Map<string, Array<{ branchId: string; nodeId: string; status: string }>> {
|
||||
async getBranchProgressByTask( taskIds: readonly string[], ): Promise<Map<string, Array<{ branchId: string; nodeId: string; status: string }>>> {
|
||||
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<TaskStoreEvents> {
|
||||
}
|
||||
|
||||
/** 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<void> {
|
||||
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<Array<{
|
||||
taskId: string;
|
||||
runId: string;
|
||||
branchId: string;
|
||||
currentNodeId: string;
|
||||
status: "running" | "completed" | "failed" | "aborted";
|
||||
}> {
|
||||
}>> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<import("./types.js").WorkflowRunStepInstance[]> {
|
||||
return loadWorkflowRunStepInstancesImpl(this, taskId, runId);
|
||||
}
|
||||
clearWorkflowRunStepInstances(taskId: string, keepRunId?: string): void {
|
||||
async clearWorkflowRunStepInstances(taskId: string, keepRunId?: string): Promise<void> {
|
||||
return clearWorkflowRunStepInstancesImpl(this, taskId, keepRunId);
|
||||
}
|
||||
|
||||
@@ -1234,7 +1234,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
public parseWorkflowPromptOverrideJson(raw: string | null | undefined): Record<string, string> {
|
||||
return parseWorkflowPromptOverrideJsonImpl(this, raw);
|
||||
}
|
||||
listWorkflowPromptOverridesForProject(): Record<string, Record<string, string>> {
|
||||
async listWorkflowPromptOverridesForProject(): Promise<Record<string, Record<string, string>>> {
|
||||
return listWorkflowPromptOverridesForProjectImpl(this);
|
||||
}
|
||||
getWorkflowPromptOverrides(workflowId: string, projectId: string): Record<string, string> {
|
||||
@@ -2140,7 +2140,7 @@ Issue #2149 requires read-only type filtering to occur in the file-store before
|
||||
public async migrateMovedSettingsToWorkflowValuesOnce(): Promise<void> {
|
||||
return migrateMovedSettingsImpl(this);
|
||||
}
|
||||
public readRawProjectSettings(): Record<string, unknown> {
|
||||
public async readRawProjectSettings(): Promise<Record<string, unknown>> {
|
||||
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<void> {
|
||||
return recordVerificationCachePassImpl(this, treeSha, testCommand, buildCommand, taskId);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void> => {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
/*
|
||||
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(
|
||||
|
||||
@@ -912,7 +912,7 @@ export async function migrateMovedSettingsImpl(store: TaskStore): Promise<void>
|
||||
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<string, unknown> = {};
|
||||
try {
|
||||
rawGlobalSettings = await store.globalSettingsStore.readRaw();
|
||||
|
||||
@@ -778,21 +778,30 @@ export async function updateIssueInfoImpl(store: TaskStore, id: string, issueInf
|
||||
|
||||
export async function listWorkflowStepsImpl(store: TaskStore): Promise<import("../types.js").WorkflowStep[]> {
|
||||
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<typeof store.toStoredWorkflowStep>[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<typeof store.toStoredWorkflowStep>[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;
|
||||
|
||||
@@ -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<string, Column>();
|
||||
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<string>();
|
||||
const resultPg = new Map<string, Column>();
|
||||
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<string, Record<string, string>> {
|
||||
export async function listWorkflowPromptOverridesForProjectImpl(store: TaskStore): Promise<Record<string, Record<string, string>>> {
|
||||
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<string, Record<string, string>> = {};
|
||||
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<string, string> = {};
|
||||
for (const [nodeId, value] of Object.entries(overrides as Record<string, unknown>)) {
|
||||
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 }>;
|
||||
|
||||
@@ -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<void> {
|
||||
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<string, unknown>, 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<string, unknown> | 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;
|
||||
|
||||
@@ -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<string, Array<{ branchId: string; nodeId: string; status: string }>> {
|
||||
): Promise<Map<string, Array<{ branchId: string; nodeId: string; status: string }>>> {
|
||||
const result = new Map<string, Array<{ branchId: string; nodeId: string; status: string }>>();
|
||||
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<string, { runId: string; updatedAt: string }>();
|
||||
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<Array<{
|
||||
taskId: string;
|
||||
runId: string;
|
||||
branchId: string;
|
||||
currentNodeId: string;
|
||||
status: "running" | "completed" | "failed" | "aborted";
|
||||
}> {
|
||||
}>> {
|
||||
/*
|
||||
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<void> {
|
||||
/*
|
||||
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<import("../types.js").WorkflowRunStepInstance[]> {
|
||||
// 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<void> {
|
||||
// FNXC:PostgresOnlyDataAccess 2026-07-16-13:40: see saveWorkflowRunStepInstanceImpl.
|
||||
if (store.backendMode) {
|
||||
return clearWorkflowRunStepInstancesAsyncImpl(store, taskId, keepRunId);
|
||||
}
|
||||
try {
|
||||
if (keepRunId === undefined) {
|
||||
store.db
|
||||
|
||||
@@ -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<v
|
||||
store.db.bumpLastModified();
|
||||
}
|
||||
|
||||
export function readRawProjectSettingsImpl(store: TaskStore): Record<string, unknown> {
|
||||
export async function readRawProjectSettingsImpl(store: TaskStore): Promise<Record<string, unknown>> {
|
||||
// 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<string, unknown>)
|
||||
: {};
|
||||
} 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<void> {
|
||||
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)
|
||||
|
||||
@@ -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<void> {
|
||||
// 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(
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<void>;
|
||||
loadWorkflowRunBranches?: (taskId: string, runId: string) => WorkflowBranchRunState[] | Promise<WorkflowBranchRunState[]>;
|
||||
clearWorkflowRunBranches?: (taskId: string, keepRunId: string) => void | Promise<void>;
|
||||
};
|
||||
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<void>;
|
||||
loadWorkflowRunStepInstancesAsync?: (taskId: string, runId: string) => Promise<WorkflowStepInstanceState[]>;
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user