From cb1691f473f554929b665d96d7b2671d5fa3cbdf Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 25 Jun 2026 22:33:08 -0700 Subject: [PATCH] feat(FN-7039): record graph workflow-step results into task.workflowStepResults Enabled optional-group nodes now upsert their outcome into the existing task.workflowStepResults field keyed by node id, and emit [pre-merge] logs at parity with the legacy runWorkflowSteps path. Disabled groups stay byte-inert. Reuses the existing WorkflowStepResult type + store.updateTask path (no new table/type/store method). Unblocks the unified progress bar for graph-run steps. Plan U2. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...iltin-coding-workflow-step-results.test.ts | 174 ++++++++++++++++++ packages/engine/src/executor.ts | 27 +++ .../engine/src/workflow-graph-executor.ts | 97 ++++++++++ packages/engine/src/workflow-graph-loop.ts | 14 +- .../engine/src/workflow-graph-task-runner.ts | 7 +- 5 files changed, 316 insertions(+), 3 deletions(-) create mode 100644 packages/engine/src/__tests__/builtin-coding-workflow-step-results.test.ts diff --git a/packages/engine/src/__tests__/builtin-coding-workflow-step-results.test.ts b/packages/engine/src/__tests__/builtin-coding-workflow-step-results.test.ts new file mode 100644 index 0000000000..15ab6eefdd --- /dev/null +++ b/packages/engine/src/__tests__/builtin-coding-workflow-step-results.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, it } from "vitest"; +import type { TaskDetail, WorkflowIr, WorkflowStepResult } from "@fusion/core"; + +import { WorkflowGraphExecutor, type WorkflowNodeHandler, type WorkflowNodeResult } from "../workflow-graph-executor.js"; + +/* +FNXC:WorkflowStepResults 2026-06-25-12:00: +Plan U2 coverage: an ENABLED optional-group node must upsert a WorkflowStepResult +into `task.workflowStepResults` keyed by the GROUP node id, with status mapped from +the inner exit node's outcome/verdict (APPROVE → passed; advisory REVISE → +advisory_failure; gate REVISE / group failure → failed). A DISABLED group records +NOTHING (byte-inert). Uses the executor `handlers` override pattern to inject inner +prompt/gate node results (real executor runs, not traversal-only). +*/ + +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +/** A graph with one `optional-group` ("Code review") between `before` and `after`. + * The group's template runs a single `reviewstep` prompt when enabled. */ +function codeReviewGroupIr(): WorkflowIr { + return { + version: "v2", + name: "code-review-results-test", + columns: [{ id: "work", name: "Work", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "before", kind: "prompt", config: { prompt: "before" } }, + { + id: "code-review", + kind: "optional-group", + config: { + name: "Code review", + defaultOn: false, + template: { + nodes: [{ id: "reviewstep", kind: "prompt", config: { prompt: "review" } }], + edges: [], + }, + }, + }, + { id: "after", kind: "prompt", config: { prompt: "after" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "before" }, + { from: "before", to: "code-review" }, + { from: "code-review", to: "after", condition: "success" }, + // Route a group failure away from the success edge so the run terminates + // cleanly when the inner gate REVISEs / fails. + { from: "code-review", to: "after", condition: "failure" }, + { from: "after", to: "end" }, + ], + }; +} + +function taskWith(enabled: string[] | undefined): TaskDetail { + return { id: "FN-CR", enabledWorkflowSteps: enabled } as TaskDetail; +} + +/** A capturing sink that mirrors the production executor adapter's upsert-by-id + * semantics so the final recorded state can be asserted. */ +function makeRecorder() { + const results: WorkflowStepResult[] = []; + const calls: WorkflowStepResult[] = []; + const record = async (_taskId: string, result: WorkflowStepResult) => { + calls.push(result); + const idx = results.findIndex((r) => r.workflowStepId === result.workflowStepId); + if (idx >= 0) results[idx] = result; + else results.push(result); + }; + return { results, calls, record }; +} + +/** Inject a fixed inner-node result for `reviewstep`; everything else succeeds. */ +function innerHandler(reviewResult: WorkflowNodeResult): WorkflowNodeHandler { + return async (node) => (node.id === "reviewstep" ? reviewResult : { outcome: "success" }); +} + +describe("WorkflowGraphExecutor optional-group → task.workflowStepResults (plan U2)", () => { + it("(a) enabled group + APPROVE verdict → one entry keyed by the group node id with status 'passed'", async () => { + const recorder = makeRecorder(); + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: innerHandler({ outcome: "success", value: "APPROVE" }) }, + recordWorkflowStepResult: recorder.record, + }); + + const result = await executor.run(taskWith(["code-review"]), settingsOn(), codeReviewGroupIr()); + expect(result.outcome).toBe("success"); + + // Exactly one entry, keyed by the GROUP node id (not the inner template id). + expect(recorder.results).toHaveLength(1); + const entry = recorder.results[0]; + expect(entry.workflowStepId).toBe("code-review"); + expect(entry.workflowStepName).toBe("Code review"); + expect(entry.phase).toBe("pre-merge"); + expect(entry.status).toBe("passed"); + expect(entry.verdict).toBe("APPROVE"); + // Upsert: a pending entry was written first, then replaced by the terminal one + // (same startedAt preserved; completedAt added). + expect(recorder.calls).toHaveLength(2); + expect(recorder.calls[0].status).toBe("pending"); + expect(recorder.calls[0].startedAt).toBeDefined(); + expect(entry.startedAt).toBe(recorder.calls[0].startedAt); + expect(entry.completedAt).toBeDefined(); + }); + + it("(b) advisory REVISE (success outcome, REVISE verdict) → status 'advisory_failure'", async () => { + const recorder = makeRecorder(); + const executor = new WorkflowGraphExecutor({ + // Advisory: the inner node returns success (non-blocking) but a REVISE verdict. + handlers: { prompt: innerHandler({ outcome: "success", value: "REVISE" }) }, + recordWorkflowStepResult: recorder.record, + }); + + await executor.run(taskWith(["code-review"]), settingsOn(), codeReviewGroupIr()); + + expect(recorder.results).toHaveLength(1); + expect(recorder.results[0].workflowStepId).toBe("code-review"); + expect(recorder.results[0].status).toBe("advisory_failure"); + expect(recorder.results[0].verdict).toBe("REVISE"); + }); + + it("(c) gate REVISE / group failure → status 'failed'", async () => { + // Gate REVISE: a blocking gate surfaces the REVISE as a failure OUTCOME. + const gateRecorder = makeRecorder(); + const gateExecutor = new WorkflowGraphExecutor({ + handlers: { prompt: innerHandler({ outcome: "failure", value: "REVISE" }) }, + recordWorkflowStepResult: gateRecorder.record, + }); + await gateExecutor.run(taskWith(["code-review"]), settingsOn(), codeReviewGroupIr()); + expect(gateRecorder.results).toHaveLength(1); + expect(gateRecorder.results[0].status).toBe("failed"); + + // Hard group failure (no verdict) → also 'failed'. + const failRecorder = makeRecorder(); + const failExecutor = new WorkflowGraphExecutor({ + handlers: { prompt: innerHandler({ outcome: "failure", value: "failed" }) }, + recordWorkflowStepResult: failRecorder.record, + }); + await failExecutor.run(taskWith(["code-review"]), settingsOn(), codeReviewGroupIr()); + expect(failRecorder.results).toHaveLength(1); + expect(failRecorder.results[0].status).toBe("failed"); + }); + + it("(d) DISABLED group → no entry recorded (byte-inert)", async () => { + const recorder = makeRecorder(); + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: innerHandler({ outcome: "success", value: "APPROVE" }) }, + recordWorkflowStepResult: recorder.record, + }); + + const result = await executor.run(taskWith([]), settingsOn(), codeReviewGroupIr()); + expect(result.outcome).toBe("success"); + + expect(recorder.calls).toHaveLength(0); + expect(recorder.results).toHaveLength(0); + }); + + it("emits parity [pre-merge] logs for the enabled group via logTaskEntry", async () => { + const logs: string[] = []; + const recorder = makeRecorder(); + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: innerHandler({ outcome: "success", value: "APPROVE" }) }, + recordWorkflowStepResult: recorder.record, + logTaskEntry: (summary: string) => { + logs.push(summary); + }, + }); + + await executor.run(taskWith(["code-review"]), settingsOn(), codeReviewGroupIr()); + + expect(logs).toContain("[pre-merge] Starting workflow step: Code review"); + expect(logs).toContain("[pre-merge] Workflow step completed: Code review"); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index df189829e6..bd177bdde8 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -4352,6 +4352,33 @@ export class TaskExecutor { .logEntry(task.id, summary, detail, this.getRunContextFor(task.id)) .catch(() => {}); }, + /* + FNXC:WorkflowStepResults 2026-06-25-12:00: + Plan U2 (KTD-1/KTD-2): persistence adapter for an ENABLED optional-group + node's outcome. The graph records each enabled group's WorkflowStepResult + into the EXISTING `task.workflowStepResults` field keyed by `node.id` so the + unified progress bar (getUnifiedTaskProgress) reflects graph-run steps — + NO new table/type/store method. Upsert by `workflowStepId === node.id` + (replace-if-present else append) through the existing + `store.updateTask({workflowStepResults})` path. Fail-soft: degrade to a + no-op when the store lacks updateTask, and swallow read/write errors (the + executor wrapper also swallows) so result recording never affects the run. + */ + recordWorkflowStepResult: async (taskId: string, result: import("@fusion/core").WorkflowStepResult) => { + if (typeof this.store.updateTask !== "function") return; + try { + const live = await this.store.getTask(taskId); + const existing = Array.isArray(live?.workflowStepResults) + ? [...live.workflowStepResults] + : []; + const idx = existing.findIndex((r) => r.workflowStepId === result.workflowStepId); + if (idx >= 0) existing[idx] = result; + else existing.push(result); + await this.store.updateTask(taskId, { workflowStepResults: existing }, this.getRunContextFor(taskId)); + } catch { + // Result recording is additive visibility — never affect the run. + } + }, }); let result: WorkflowGraphTaskRunResult; try { diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 3f591d50c3..ef1998e97e 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -7,6 +7,7 @@ import type { WorkflowIrNode, WorkflowIrNodeKind, WorkflowNodeExtensionResult, + WorkflowStepResult, } from "@fusion/core"; import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, getWorkflowExtensionRegistry, resolveMaxReworkCycles } from "@fusion/core"; @@ -153,6 +154,18 @@ export interface WorkflowGraphExecutorDeps { resumeReconcile?: ForeachEnvironment["resumeReconcile"]; /** FIX 4 (context gap): task-level log sink for integration-conflict rework. */ logTaskEntry?: ForeachEnvironment["logTaskEntry"]; + /* + * FNXC:WorkflowStepResults 2026-06-25-12:00: + * Fail-soft persistence sink for an ENABLED optional-group node's outcome + * (plan U2, KTD-1/KTD-2). The graph upserts each enabled group's result into the + * EXISTING `task.workflowStepResults` field keyed by `node.id` so the unified + * progress bar (`getUnifiedTaskProgress`) reflects graph-run steps. Optional: when + * absent the executor records NOTHING (keeps in-memory tests byte-inert), so a + * disabled group and an unwired store both record nothing. The upsert-by-id + + * `store.updateTask({workflowStepResults})` wiring lives in the executor adapter; + * this seam only forwards the terminal/pending entry. + */ + recordWorkflowStepResult?: (taskId: string, result: WorkflowStepResult) => void | Promise; /** Project node-published task metadata onto the task row for dispatcher/UI. */ publishTaskProjection?: (taskId: string, patch: WorkflowTaskProjection, source: { nodeId: string; nodeKind: WorkflowIrNode["kind"] }) => void | Promise; /** @deprecated use publishTaskProjection. Kept for older callers. */ @@ -502,6 +515,32 @@ export class WorkflowGraphExecutor { // invariant. (Code review: CodeRabbit.) return await traverseChildren(node, { outcome: "success" }); } + /* + * FNXC:WorkflowStepResults 2026-06-25-12:00: + * Record an enabled optional-group's outcome into the EXISTING + * `task.workflowStepResults` field keyed by `node.id` (plan U2, + * KTD-1/KTD-2/KTD-3) + emit `[pre-merge]` logs at parity with the legacy + * `runWorkflowSteps`. A `pending` entry (with `startedAt`) is written when + * the enabled group STARTS so the dashboard can show live status; after + * `runOptionalGroup` returns, the entry is UPSERT-replaced by the terminal + * record (same `startedAt`, plus `completedAt`). Disabled groups take the + * bypass branch above and record NOTHING (byte-inert). Recording is + * fail-soft via the optional `recordWorkflowStepResult` dep — absent → no + * record (in-memory tests unchanged). + */ + const groupName = typeof node.config?.name === "string" && node.config.name.trim() + ? node.config.name.trim() + : node.id; + const stepStartedAt = new Date().toISOString(); + await this.recordOptionalGroupStepResult(task.id, { + workflowStepId: node.id, + workflowStepName: groupName, + phase: "pre-merge", + status: "pending", + startedAt: stepStartedAt, + }); + this.deps.logTaskEntry?.(`[pre-merge] Starting workflow step: ${groupName}`); + const groupResult = await runOptionalGroup(node, { context, runTemplateNode: (tNode, sig, contextOverride) => @@ -509,6 +548,49 @@ export class WorkflowGraphExecutor { shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src), signal: this.deps.signal, }); + // Map the group outcome → a WorkflowStepResult status (mirrors + // `mapWorkflowStatus` in taskProgress.ts): a `failure` outcome (gate REVISE + // or hard failure) → "failed"; an advisory REVISE (success outcome, REVISE + // verdict) → "advisory_failure" (non-blocking); otherwise → "passed". + const exitResult = groupResult.exitStepRecord; + const verdictRaw = typeof (exitResult?.value ?? groupResult.value) === "string" + ? (exitResult?.value ?? groupResult.value) as string + : undefined; + const verdict = + verdictRaw === "APPROVE" || verdictRaw === "APPROVE_WITH_NOTES" || verdictRaw === "REVISE" + ? verdictRaw + : undefined; + let stepStatus: WorkflowStepResult["status"]; + if (groupResult.outcome === "failure") stepStatus = "failed"; + else if (verdict === "REVISE") stepStatus = "advisory_failure"; + else stepStatus = "passed"; + const exitContextPatch = exitResult?.contextPatch; + const stepOutput = typeof exitContextPatch?.output === "string" ? exitContextPatch.output : undefined; + const stepNotes = typeof exitContextPatch?.notes === "string" ? exitContextPatch.notes : undefined; + await this.recordOptionalGroupStepResult(task.id, { + workflowStepId: node.id, + workflowStepName: groupName, + phase: "pre-merge", + status: stepStatus, + ...(verdict ? { verdict } : {}), + ...(stepOutput !== undefined ? { output: stepOutput } : {}), + ...(stepNotes !== undefined ? { notes: stepNotes } : {}), + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + }); + // `[pre-merge]` terminal logs at parity with the legacy path + // (executor.ts runWorkflowSteps: "completed" / "requested revision" / + // "failed" + the advisory variant). + if (stepStatus === "passed") { + this.deps.logTaskEntry?.(`[pre-merge] Workflow step completed: ${groupName}`); + } else if (stepStatus === "advisory_failure") { + this.deps.logTaskEntry?.(`[pre-merge] Workflow step requested revision: ${groupName}`, stepOutput); + this.deps.logTaskEntry?.(`[pre-merge] Advisory workflow step failed: ${groupName}`); + } else if (verdict === "REVISE") { + this.deps.logTaskEntry?.(`[pre-merge] Workflow step requested revision: ${groupName}`, stepOutput); + } else { + this.deps.logTaskEntry?.(`[pre-merge] Workflow step failed: ${groupName}`, stepOutput); + } visitedNodeIds.push(...groupResult.visitedNodeIds); const result: WorkflowNodeResult = { outcome: groupResult.outcome, @@ -689,6 +771,21 @@ export class WorkflowGraphExecutor { } } + /* + * FNXC:WorkflowStepResults 2026-06-25-12:00: + * Fail-soft forward to the `recordWorkflowStepResult` persistence sink (plan U2). + * Recording is additive visibility bookkeeping — a sink failure (or absent sink) + * must NEVER affect graph execution, so swallow errors and no-op when unwired. + */ + private async recordOptionalGroupStepResult(taskId: string, result: WorkflowStepResult): Promise { + if (!this.deps.recordWorkflowStepResult) return; + try { + await this.deps.recordWorkflowStepResult(taskId, result); + } catch { + // Result recording is additive — a failure must not affect the run. + } + } + private shouldTraverseEdge(edge: WorkflowIrEdge, sourceResult: WorkflowNodeResult): boolean { if (!edge.condition) return sourceResult.outcome === "success"; if (edge.condition === "success") return sourceResult.outcome === "success"; diff --git a/packages/engine/src/workflow-graph-loop.ts b/packages/engine/src/workflow-graph-loop.ts index f3f5c9de6b..309d0b0270 100644 --- a/packages/engine/src/workflow-graph-loop.ts +++ b/packages/engine/src/workflow-graph-loop.ts @@ -229,6 +229,16 @@ export interface OptionalGroupRunResult { outcome: WorkflowNodeOutcome; value?: string; visitedNodeIds: string[]; + /* + * FNXC:WorkflowStepResults 2026-06-25-12:00: + * The exit (last-run) inner template node's result, surfaced so the executor can + * record the group's outcome into `task.workflowStepResults` (KTD-1/KTD-2, plan + * U2). For a prompt/gate inner node `value` carries the structured verdict + * (APPROVE / APPROVE_WITH_NOTES / REVISE); `outcome` distinguishes a gate REVISE + * (failure) from an advisory REVISE (success). Notes/output are not surfaced by + * the WorkflowNodeResult contract, so they are recorded only when present. + */ + exitStepRecord?: WorkflowNodeResult; } function resolveOptionalGroupTemplate( @@ -278,7 +288,7 @@ export async function runOptionalGroup( // Publish accumulated template context, then surface the failure as the // group's outcome so its failure/outcome: edges route. Object.assign(env.context, groupContext); - return { outcome: "failure", value: lastResult.value, visitedNodeIds }; + return { outcome: "failure", value: lastResult.value, visitedNodeIds, exitStepRecord: lastResult }; } const edges: WorkflowIrEdge[] = outgoing.get(current.id) ?? []; @@ -290,5 +300,5 @@ export async function runOptionalGroup( // Single pass complete: publish the template's context onto the shared context. Object.assign(env.context, groupContext); - return { outcome: lastResult.outcome, value: lastResult.value, visitedNodeIds }; + return { outcome: lastResult.outcome, value: lastResult.value, visitedNodeIds, exitStepRecord: lastResult }; } diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index dc14195e5b..ee11889f8f 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -1,4 +1,4 @@ -import type { Settings, TaskDetail, WorkflowDefinition } from "@fusion/core"; +import type { Settings, TaskDetail, WorkflowDefinition, WorkflowStepResult } from "@fusion/core"; import { getBuiltinWorkflow, isBuiltinWorkflowId } from "@fusion/core"; import { WorkflowGraphExecutor, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js"; @@ -88,6 +88,10 @@ export interface WorkflowGraphTaskRunnerDeps { resumeReconcile?: ForeachEnvironment["resumeReconcile"]; /** FIX 4 (context gap): task-level log sink for integration-conflict rework. */ logTaskEntry?: ForeachEnvironment["logTaskEntry"]; + /** Plan U2 (KTD-1/KTD-2): fail-soft sink that upserts an enabled optional-group + * node's outcome into `task.workflowStepResults` keyed by node id. Additive; + * absent → graph records nothing (disabled groups + unwired stores byte-inert). */ + recordWorkflowStepResult?: (taskId: string, result: WorkflowStepResult) => void | Promise; /** Project node-published task metadata onto the task row for dispatcher/UI. */ publishTaskProjection?: (taskId: string, patch: WorkflowTaskProjection, source: { nodeId: string; nodeKind: string }) => void | Promise; /** @deprecated use publishTaskProjection. */ @@ -233,6 +237,7 @@ export class WorkflowGraphTaskRunner { semaphoreAvailability: this.deps.semaphoreAvailability, resumeReconcile: this.deps.resumeReconcile, logTaskEntry: this.deps.logTaskEntry, + recordWorkflowStepResult: this.deps.recordWorkflowStepResult, publishTaskProjection: this.deps.publishTaskProjection, publishTouchedFiles: this.deps.publishTouchedFiles, // Single source of truth (KTD-6): prefer the caller-threaded run id so the