From 40d05c81391b22eb6f2a0a30f087444cb00b957e Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 00:01:00 -0700 Subject: [PATCH] fix(FN-6035): address workflow projection review feedback --- .changeset/workflow-projection-capacity.md | 2 + .../core/src/__tests__/store-update.test.ts | 38 +++++++++++++++++++ packages/core/src/store.ts | 17 +++++++++ .../workflow-graph-executor-handlers.test.ts | 37 ++++++++++++++++++ packages/engine/src/executor.ts | 25 ++++++------ .../engine/src/workflow-graph-executor.ts | 10 +++-- 6 files changed, 113 insertions(+), 16 deletions(-) diff --git a/.changeset/workflow-projection-capacity.md b/.changeset/workflow-projection-capacity.md index 888c6b2175..c3dc751d9d 100644 --- a/.changeset/workflow-projection-capacity.md +++ b/.changeset/workflow-projection-capacity.md @@ -1,3 +1,5 @@ +--- "@runfusion/fusion": patch +--- Fix workflow-native dispatch capacity accounting and publish workflow node task metadata to the existing task fields used by scheduler and dashboard surfaces. diff --git a/packages/core/src/__tests__/store-update.test.ts b/packages/core/src/__tests__/store-update.test.ts index e1e5c1febe..f7749ea711 100644 --- a/packages/core/src/__tests__/store-update.test.ts +++ b/packages/core/src/__tests__/store-update.test.ts @@ -1198,5 +1198,43 @@ Task with acceptance criteria }); }); + describe("updateTaskAtomic", () => { + it("serializes read-merge-write patches against the freshest task snapshot", async () => { + const task = await store.createTask({ description: "atomic task projection" }); + let releaseFirst: () => void = () => {}; + const firstCanFinish = new Promise((resolve) => { + releaseFirst = resolve; + }); + let markFirstRead: () => void = () => {}; + const firstRead = new Promise((resolve) => { + markFirstRead = resolve; + }); + + const first = store.updateTaskAtomic(task.id, async (current) => { + markFirstRead(); + expect(current.modifiedFiles).toBeUndefined(); + await firstCanFinish; + return { + modifiedFiles: [...new Set([...(current.modifiedFiles ?? []), "src/a.ts"])].sort(), + mergeDetails: { ...(current.mergeDetails ?? {}), filesChanged: 1 }, + }; + }); + + await firstRead; + + const second = store.updateTaskAtomic(task.id, (current) => ({ + modifiedFiles: [...new Set([...(current.modifiedFiles ?? []), "src/b.ts"])].sort(), + mergeDetails: { ...(current.mergeDetails ?? {}), insertions: 2 }, + })); + + releaseFirst(); + await Promise.all([first, second]); + + const updated = await store.getTask(task.id); + expect(updated.modifiedFiles).toEqual(["src/a.ts", "src/b.ts"]); + expect(updated.mergeDetails).toEqual({ filesChanged: 1, insertions: 2 }); + }); + }); + }); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index a5166dcc71..03b98a1abc 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -7445,6 +7445,23 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} return this.withTaskLock(id, () => this.updateTaskUnlocked(id, updates, runContext)); } + async updateTaskAtomic( + id: string, + updater: ( + current: Task, + ) => Parameters[1] | null | undefined | Promise[1] | null | undefined>, + runContext?: RunMutationContext, + ): Promise { + return this.withTaskLock(id, async () => { + const current = await this.readTaskJson(this.taskDir(id)); + const updates = await updater(current); + if (!updates || Object.values(updates).every((value) => value === undefined)) { + return current; + } + return this.updateTaskUnlocked(id, updates, runContext); + }); + } + /** * Merge a validated/normalized custom-field patch into the existing values. * `null` in the patch deletes that field's value (the delete sentinel from diff --git a/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts b/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts index 637655a72f..169e515163 100644 --- a/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts @@ -297,6 +297,43 @@ describe("WorkflowGraphExecutor traversal", () => { expect(result.context["node:a:projectionError"]).toBe("store unavailable"); }); + it("does not fail the node when the deprecated touched-files hook fails", async () => { + const ir: WorkflowIr = { + version: "v1", + name: "legacy-touched-files-failure", + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a" }, + { from: "a", to: "end", condition: "success" }, + ], + }; + const publishTaskProjection = vi.fn(); + const publishTouchedFiles = vi.fn(async () => { + throw new Error("legacy sink unavailable"); + }); + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async () => ({ + outcome: "success", + contextPatch: { modifiedFiles: ["src/projected.ts"] }, + }), + }, + publishTaskProjection, + publishTouchedFiles, + }); + + const result = await executor.run(task, settingsOn(), ir); + + expect(publishTaskProjection).toHaveBeenCalledTimes(1); + expect(publishTouchedFiles).toHaveBeenCalledTimes(1); + expect(result.outcome).toBe("success"); + expect(result.context["node:a:projectionError"]).toBeUndefined(); + }); + it("caps retries and converts exceptions to failure", async () => { const ir: WorkflowIr = { version: "v1", diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 17592439b5..05f86305cb 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -3752,19 +3752,18 @@ export class TaskExecutor { runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)), publishTaskProjection: async (taskId, patch) => { - const liveTask = await this.store.getTask(taskId); - const update: Parameters[1] = {}; - if (patch.modifiedFiles) { - const merged = [...new Set([...(liveTask?.modifiedFiles ?? []), ...patch.modifiedFiles])].sort(); - if (merged.length > 0) update.modifiedFiles = merged; - } - if (patch.mergeDetails) { - update.mergeDetails = { ...(liveTask?.mergeDetails ?? {}), ...patch.mergeDetails }; - } - if (patch.summary !== undefined) update.summary = patch.summary; - if (Object.keys(update).length > 0) { - await this.store.updateTask(taskId, update); - } + await this.store.updateTaskAtomic(taskId, (liveTask) => { + const update: Parameters[1] = {}; + if (patch.modifiedFiles) { + const merged = [...new Set([...(liveTask.modifiedFiles ?? []), ...patch.modifiedFiles])].sort(); + if (merged.length > 0) update.modifiedFiles = merged; + } + if (patch.mergeDetails) { + update.mergeDetails = { ...(liveTask.mergeDetails ?? {}), ...patch.mergeDetails }; + } + if (patch.summary !== undefined) update.summary = patch.summary; + return update; + }); }, onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), // Wire SQLite-backed per-branch persistence in production (#1407): the diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 9d2dd03360..b7f642b996 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -717,9 +717,6 @@ export class WorkflowGraphExecutor { const source = { nodeId: node.id, nodeKind: node.kind }; try { await this.deps.publishTaskProjection?.(taskId, patch, source); - if (patch.modifiedFiles && patch.modifiedFiles.length > 0) { - await this.deps.publishTouchedFiles?.(taskId, patch.modifiedFiles, source); - } } catch (error) { return { outcome: "failure", @@ -730,6 +727,13 @@ export class WorkflowGraphExecutor { }, }; } + if (patch.modifiedFiles && patch.modifiedFiles.length > 0) { + try { + await this.deps.publishTouchedFiles?.(taskId, patch.modifiedFiles, source); + } catch { + // Deprecated compatibility hook; primary projection persistence owns node outcome. + } + } return result; } }