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

Merge pull request #1540 from Runfusion/feature/workflow-mapping

Browse Source 
fix(FN-6035): close workflow dispatch projection gaps
This commit is contained in:
gsxdsm
2026-06-09 00:06:07 -07:00
committed by GitHub
parent 9c43e3f30e 40d05c8139
commit cb2b64e6f0
15 changed files with 502 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/workflow-projection-capacity.md Normal file
View File

@@ -0,0 +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.

4
packages/core/src/__tests__/builtin-coding-workflow-ir.test.ts
View File

@@ -58,6 +58,10 @@ describe("builtin coding workflow ir", () => {
expect(traitsFor("in-review")).toEqual(["merge-blocker", "human-review", "stall-detection", "merge"]);
expect(traitsFor("done")).toEqual(["complete"]);
expect(traitsFor("archived")).toEqual(["archived"]);
// in-progress owns the legacy execution concurrency policy in workflow data:
// the limit is supplied by the project maxConcurrent setting.
const wip = byId.get("in-progress")!.traits.find((t) => t.trait === "wip");
expect(wip?.config).toEqual({ limitSetting: "maxConcurrent", countPending: true });
// todo's hold is capacity-released (legacy "pull from todo when a slot frees").
const hold = byId.get("todo")!.traits.find((t) => t.trait === "hold");
expect(hold?.config?.release).toBe("capacity");

8
packages/core/src/__tests__/builtin-traits.test.ts
View File

@@ -109,6 +109,14 @@ describe("default workflow columns validate cleanly", () => {
expect(flags.timing).toBe(true);
});
it("wip trait schema supports explicit settings-backed limits", () => {
const r = freshRegistry();
const fields = r.getTrait("wip")?.configSchema?.fields ?? [];
const limitSetting = fields.find((field) => field.key === "limitSetting");
expect(limitSetting?.type).toBe("enum");
expect(limitSetting?.enumValues).toEqual(["maxConcurrent"]);
});
it("the default workflow's in-review column resolves review and merge flags", () => {
const r = freshRegistry();
const ir = BUILTIN_CODING_WORKFLOW_IR as WorkflowIrV2;

38
packages/core/src/__tests__/store-update.test.ts
View File

@@ -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<void>((resolve) => {
releaseFirst = resolve;
});
let markFirstRead: () => void = () => {};
const firstRead = new Promise<void>((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 });
});
});
});

6
packages/core/src/builtin-coding-workflow-ir.ts
View File

@@ -37,7 +37,11 @@ const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = {
{
id: "in-progress",
name: "In progress",
traits: [{ trait: "wip" }, { trait: "abort-on-exit" }, { trait: "timing" }],
traits: [
{ trait: "wip", config: { limitSetting: "maxConcurrent", countPending: true } },
{ trait: "abort-on-exit" },
{ trait: "timing" },
],
},
{
id: "in-review",

8
packages/core/src/builtin-traits.ts
View File

@@ -77,7 +77,13 @@ export const BUILTIN_TRAIT_DEFINITIONS: TraitDefinition[] = [
flags: { countsTowardWip: true },
configSchema: {
fields: [
{ key: "limit", type: "number", required: true, description: "Max concurrent cards" },
{ key: "limit", type: "number", description: "Max concurrent cards" },
{
key: "limitSetting",
type: "enum",
enumValues: ["maxConcurrent"],
description: "Project setting that supplies the capacity limit",
},
{ key: "countPending", type: "boolean", description: "Count mid-transition cards" },
],
},

17
packages/core/src/store.ts
View File

@@ -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<TaskStore["updateTask"]>[1] | null | undefined | Promise<Parameters<TaskStore["updateTask"]>[1] | null | undefined>,
runContext?: RunMutationContext,
): Promise<Task> {
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

13
packages/core/src/workflow-capacity.ts
View File

@@ -71,10 +71,12 @@ function isDefaultWorkflowColumns(ir: WorkflowIr): boolean {
*
* Limit resolution order:
* 1. An explicit numeric `limit` in the column's `wip` trait config wins.
* 2. Otherwise, for the DEFAULT workflow's `in-progress` column, read through
* 2. A `limitSetting: "maxConcurrent"` declaration reads through to the
* project setting, making the built-in workflow's capacity policy explicit.
* 3. Otherwise, for the DEFAULT workflow's `in-progress` column, read through
* to `settings.maxConcurrent` (default 2) so the legacy knob keeps working
* and flag-ON default-workflow scheduling matches flag-OFF (legacy parity).
* 3. Otherwise the column has a capacity trait but no resolvable finite limit
* 4. Otherwise the column has a capacity trait but no resolvable finite limit
* → `Infinity` (does not gate; the trait is inert until configured).
*/
export function resolveColumnCapacity(
@@ -91,6 +93,7 @@ export function resolveColumnCapacity(
// The capacity trait config (the `wip` trait carries `limit` + `countPending`).
// Find the first trait config whose trait sets countsTowardWip.
let configLimit: number | undefined;
let limitSetting: string | undefined;
let countPending = true;
for (const ct of column.traits) {
const def = getTraitRegistry().getTrait(ct.trait);
@@ -99,6 +102,9 @@ export function resolveColumnCapacity(
if (typeof cfg.limit === "number" && Number.isFinite(cfg.limit)) {
configLimit = cfg.limit;
}
if (typeof cfg.limitSetting === "string") {
limitSetting = cfg.limitSetting;
}
if (typeof cfg.countPending === "boolean") {
countPending = cfg.countPending;
}
@@ -108,6 +114,9 @@ export function resolveColumnCapacity(
let limit: number;
if (configLimit !== undefined) {
limit = configLimit;
} else if (limitSetting === "maxConcurrent") {
const maxConcurrent = settings?.maxConcurrent;
limit = typeof maxConcurrent === "number" && Number.isFinite(maxConcurrent) ? maxConcurrent : 2;
} else if (columnId === DEFAULT_WIP_COLUMN_ID && isDefaultWorkflowColumns(ir)) {
// Read-through: legacy maxConcurrent maps onto the default workflow's
// in-progress WIP limit (U6 scheduler integration).

42
packages/engine/src/__tests__/scheduler.test.ts
View File

@@ -584,6 +584,48 @@ describe("Scheduler", () => {
expect(onMoves).toContainEqual(["FN-1", "in-progress"]);
});
it("re-reads tasks after flag-ON hold-release sweep before legacy dispatch", async () => {
vi.mocked(existsSync).mockReturnValue(true);
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
const tasks = new Map<string, Task>(
Array.from({ length: 6 }, (_, index) => {
const id = `FN-${String(index + 1).padStart(3, "0")}`;
return [id, createMockTask({ id, column: "todo", dependencies: [] })];
}),
);
const moveTask = vi.fn(async (taskId: string, column: Task["column"]) => {
const current = tasks.get(taskId);
if (!current) throw new Error(`missing task ${taskId}`);
if (column === "in-progress") {
const inProgressCount = [...tasks.values()].filter((task) => task.column === "in-progress").length;
if (inProgressCount >= 3) {
throw new Error("capacity-exhausted");
}
}
const updated = { ...current, column } as Task;
tasks.set(taskId, updated);
return updated;
});
const store = createMockStore({
listTasks: vi.fn(async () => [...tasks.values()]),
getTask: vi.fn(async (taskId: string) => tasks.get(taskId) ?? null),
getSettings: vi.fn().mockResolvedValue({
maxConcurrent: 3,
maxWorktrees: 10,
experimentalFeatures: { workflowColumns: true },
}),
moveTask,
});
const scheduler = new Scheduler(store);
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect([...tasks.values()].filter((task) => task.column === "in-progress")).toHaveLength(3);
expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress")).toHaveLength(6);
expect(vi.mocked(store.listTasks).mock.calls.length).toBeGreaterThanOrEqual(2);
});
it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => {
const off = setupTodoStore(false);
await off.scheduler.schedule();

217
packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts
View File

@@ -117,6 +117,223 @@ describe("WorkflowGraphExecutor traversal", () => {
expect(result.visitedNodeIds).not.toContain("right");
});
it("publishes workflow node task projections for dispatcher and UI", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "projection",
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 executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({
outcome: "success",
contextPatch: {
touchedFiles: ["./packages/engine/src/workflow-graph-executor.ts", "packages\\core\\src\\store.ts"],
filesChanged: 2,
summary: "workflow published task metadata",
},
}),
},
publishTaskProjection,
});
await executor.run(task, settingsOn(), ir);
expect(publishTaskProjection).toHaveBeenCalledWith(
task.id,
{
modifiedFiles: ["packages/core/src/store.ts", "packages/engine/src/workflow-graph-executor.ts"],
mergeDetails: { filesChanged: 2 },
summary: "workflow published task metadata",
},
{ nodeId: "a", nodeKind: "prompt" },
);
});
it("keeps projection writes to safe task metadata fields", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "safe-projection",
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 executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({
outcome: "success",
contextPatch: {
modifiedFiles: ["src/index.ts"],
mergeDetails: {
commitSha: "engine-owned",
mergeConfirmed: true,
filesChanged: 3,
insertions: 12.8,
deletions: 1,
},
status: "done",
error: "bypass",
review: {},
reviewState: {},
workflowStepResults: [{}],
tokenUsage: {},
},
}),
},
publishTaskProjection,
});
await executor.run(task, settingsOn(), ir);
expect(publishTaskProjection).toHaveBeenCalledWith(
task.id,
{
modifiedFiles: ["src/index.ts"],
mergeDetails: { filesChanged: 3, insertions: 12, deletions: 1 },
},
{ nodeId: "a", nodeKind: "prompt" },
);
});
it("publishes projections from loop template nodes", async () => {
const ir: WorkflowIr = {
version: "v2",
name: "loop-projection",
columns: [
{ id: "todo", name: "Todo", traits: [] },
{ id: "done", name: "Done", traits: [{ trait: "complete" }] },
],
nodes: [
{ id: "start", kind: "start", column: "todo" },
{
id: "loop",
kind: "loop",
column: "todo",
config: {
maxIterations: 1,
exitWhen: { type: "output-contains", value: "done" },
template: {
nodes: [{ id: "inner", kind: "prompt" }],
edges: [],
},
},
},
{ id: "end", kind: "end", column: "done" },
],
edges: [
{ from: "start", to: "loop" },
{ from: "loop", to: "end", condition: "success" },
],
};
const publishTaskProjection = vi.fn();
const executor = new WorkflowGraphExecutor({
handlers: {
prompt: async () => ({
outcome: "success",
value: "done",
contextPatch: { modifiedFiles: ["src/from-loop.ts"] },
}),
},
publishTaskProjection,
});
await executor.run(task, settingsOn(), ir);
expect(publishTaskProjection).toHaveBeenCalledWith(
task.id,
{ modifiedFiles: ["src/from-loop.ts"] },
{ nodeId: "inner", nodeKind: "prompt" },
);
});
it("does not retry an already-executed node when projection publishing fails", async () => {
const ir: WorkflowIr = {
version: "v1",
name: "projection-failure",
nodes: [
{ id: "start", kind: "start" },
{ id: "a", kind: "prompt" },
{ id: "end", kind: "end" },
],
edges: [
{ from: "start", to: "a" },
{ from: "a", to: "end", condition: "failure" },
],
};
const handler = vi.fn(async () => ({
outcome: "success" as const,
contextPatch: { modifiedFiles: ["src/once.ts"] },
}));
const publishTaskProjection = vi.fn(async () => {
throw new Error("store unavailable");
});
const executor = new WorkflowGraphExecutor({
handlers: { prompt: handler },
maxRetriesPerNode: 3,
publishTaskProjection,
});
const result = await executor.run(task, settingsOn(), ir);
expect(handler).toHaveBeenCalledTimes(1);
expect(publishTaskProjection).toHaveBeenCalledTimes(1);
expect(result.outcome).toBe("failure");
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",

14
packages/engine/src/executor.ts
View File

@@ -3751,6 +3751,20 @@ export class TaskExecutor {
seams: this.createAuthoritativeWorkflowSeams(settings),
runCustomNode: (node, nodeTask) =>
this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)),
publishTaskProjection: async (taskId, patch) => {
await this.store.updateTaskAtomic(taskId, (liveTask) => {
const update: Parameters<TaskStore["updateTask"]>[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
// executor writes each branch's currentNodeId/status to

11
packages/engine/src/scheduler.ts
View File

@@ -1225,10 +1225,8 @@ export class Scheduler {
this.scheduling = true;
try {
const tasks = await this.store.listTasks({ slim: true, includeArchived: false });
const settings = await this.store.getSettings();
const maxConcurrent = settings.maxConcurrent ?? this.options.maxConcurrent ?? 2;
const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4;
let tasks = await this.store.listTasks({ slim: true, includeArchived: false });
let settings = await this.store.getSettings();
this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak(
this.options.semaphore,
tasks,
@@ -1275,8 +1273,13 @@ export class Scheduler {
// workflow hold handling and the generalized capacity-release path.
if (isWorkflowColumnsEnabled(settings)) {
await this.runHoldReleaseSweepPass();
tasks = await this.store.listTasks({ slim: true, includeArchived: false });
settings = await this.store.getSettings();
}
const maxConcurrent = settings.maxConcurrent ?? this.options.maxConcurrent ?? 2;
const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4;
// Count only in-progress tasks toward the worktree limit.
// In-review tasks with worktrees are idle (waiting to merge) and
// should not block new tasks from starting.

111
packages/engine/src/workflow-graph-executor.ts
View File

@@ -38,6 +38,16 @@ export interface WorkflowNodeResult {
contextPatch?: Record<string, unknown>;
}
export interface WorkflowTaskProjection {
modifiedFiles?: string[];
mergeDetails?: {
filesChanged?: number;
insertions?: number;
deletions?: number;
};
summary?: string;
}
export interface WorkflowNodeExecutionContext {
task: TaskDetail;
settings: Pick<Settings, "experimentalFeatures"> | undefined;
@@ -131,6 +141,10 @@ export interface WorkflowGraphExecutorDeps {
resumeReconcile?: ForeachEnvironment["resumeReconcile"];
/** FIX 4 (context gap): task-level log sink for integration-conflict rework. */
logTaskEntry?: ForeachEnvironment["logTaskEntry"];
/** 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<void>;
/** @deprecated use publishTaskProjection. Kept for older callers. */
publishTouchedFiles?: (taskId: string, files: string[], source: { nodeId: string; nodeKind: WorkflowIrNode["kind"] }) => void | Promise<void>;
}
export interface WorkflowGraphExecutorResult {
@@ -147,6 +161,66 @@ const TERMINAL_FAILURE: WorkflowGraphExecutorResult = {
visitedNodeIds: [],
};
function normalizeTouchedFile(value: unknown): string | undefined {
if (typeof value === "string") {
const trimmed = value.trim().replaceAll("\\", "/").replace(/^\.\//, "");
return trimmed.length > 0 ? trimmed : undefined;
}
if (value && typeof value === "object" && "path" in value) {
return normalizeTouchedFile((value as { path?: unknown }).path);
}
return undefined;
}
function extractTouchedFiles(contextPatch: Record<string, unknown> | undefined): string[] {
if (!contextPatch) return [];
const raw = contextPatch.modifiedFiles ?? contextPatch.touchedFiles ?? contextPatch.changedFiles;
if (!Array.isArray(raw)) return [];
return [...new Set(raw.map(normalizeTouchedFile).filter((file): file is string => file !== undefined))].sort();
}
function objectRecord(value: unknown): Record<string, unknown> | undefined {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: undefined;
}
function finiteCount(value: unknown): number | undefined {
return typeof value === "number" && Number.isFinite(value) && value >= 0
? Math.floor(value)
: undefined;
}
function extractTaskProjection(contextPatch: Record<string, unknown> | undefined): WorkflowTaskProjection {
if (!contextPatch) return {};
const patch: WorkflowTaskProjection = {};
const files = extractTouchedFiles(contextPatch);
if (files.length > 0) patch.modifiedFiles = files;
const mergeDetails = objectRecord(contextPatch.mergeDetails);
const safeMergeDetails = {
filesChanged: finiteCount(contextPatch.filesChanged ?? mergeDetails?.filesChanged),
insertions: finiteCount(mergeDetails?.insertions),
deletions: finiteCount(mergeDetails?.deletions),
};
if (
safeMergeDetails.filesChanged !== undefined
|| safeMergeDetails.insertions !== undefined
|| safeMergeDetails.deletions !== undefined
) {
patch.mergeDetails = Object.fromEntries(
Object.entries(safeMergeDetails).filter(([, value]) => value !== undefined),
) as NonNullable<WorkflowTaskProjection["mergeDetails"]>;
}
if (typeof contextPatch.summary === "string") patch.summary = contextPatch.summary;
return patch;
}
function hasTaskProjection(patch: WorkflowTaskProjection): boolean {
return Object.keys(patch).length > 0;
}
export class WorkflowGraphExecutor {
private readonly maxRetriesPerNode: number;
@@ -611,11 +685,14 @@ export class WorkflowGraphExecutor {
if (signal?.aborted) return { outcome: "failure", value: "aborted" };
try {
const pluginResult = await this.executePluginNodeHandler(node, task, workflow, context, signal);
if (pluginResult) return pluginResult;
if (pluginResult) {
return await this.publishTaskProjectionFromResult(task.id, node, pluginResult);
}
if (!handler) {
throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`);
}
return await handler(node, { task, settings, context, signal });
const result = await handler(node, { task, settings, context, signal });
return await this.publishTaskProjectionFromResult(task.id, node, result);
} catch (error) {
lastError = error;
}
@@ -629,4 +706,34 @@ export class WorkflowGraphExecutor {
},
};
}
private async publishTaskProjectionFromResult(
taskId: string,
node: WorkflowIrNode,
result: WorkflowNodeResult,
): Promise<WorkflowNodeResult> {
const patch = extractTaskProjection(result.contextPatch);
if (!hasTaskProjection(patch)) return result;
const source = { nodeId: node.id, nodeKind: node.kind };
try {
await this.deps.publishTaskProjection?.(taskId, patch, source);
} catch (error) {
return {
outcome: "failure",
value: "projection-error",
contextPatch: {
...(result.contextPatch ?? {}),
[`node:${node.id}:projectionError`]: error instanceof Error ? error.message : String(error),
},
};
}
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;
}
}

8
packages/engine/src/workflow-graph-task-runner.ts
View File

@@ -1,7 +1,7 @@
import type { Settings, TaskDetail, WorkflowDefinition } from "@fusion/core";
import { getBuiltinWorkflow, isBuiltinWorkflowId, isExperimentalFeatureEnabled } from "@fusion/core";
import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js";
import { WorkflowGraphExecutor, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js";
import type {
CodeNodeRunner,
ForeachActiveContext,
@@ -85,6 +85,10 @@ export interface WorkflowGraphTaskRunnerDeps {
resumeReconcile?: ForeachEnvironment["resumeReconcile"];
/** FIX 4 (context gap): task-level log sink for integration-conflict rework. */
logTaskEntry?: ForeachEnvironment["logTaskEntry"];
/** Project node-published task metadata onto the task row for dispatcher/UI. */
publishTaskProjection?: (taskId: string, patch: WorkflowTaskProjection, source: { nodeId: string; nodeKind: string }) => void | Promise<void>;
/** @deprecated use publishTaskProjection. */
publishTouchedFiles?: (taskId: string, files: string[], source: { nodeId: string; nodeKind: string }) => void | Promise<void>;
/**
* Step-inversion (KTD-6): the production run id, threaded from the caller so it
* is the SINGLE source of truth shared with the executor-side persistence deps
@@ -229,6 +233,8 @@ export class WorkflowGraphTaskRunner {
semaphoreAvailability: this.deps.semaphoreAvailability,
resumeReconcile: this.deps.resumeReconcile,
logTaskEntry: this.deps.logTaskEntry,
publishTaskProjection: this.deps.publishTaskProjection,
publishTouchedFiles: this.deps.publishTouchedFiles,
// Single source of truth (KTD-6): prefer the caller-threaded run id so the
// executor's persistence deps probe/flip rows under the SAME id; fall back
// to the canonical derivation when unthreaded.

12
packages/engine/src/workflow-node-handlers.ts
View File

@@ -325,10 +325,20 @@ export function createPrimitivePromptLikeHandler(
};
}
const result = await primitives.runCodingSession(primitiveCtx, context.task, prepared.data);
const contextPatch = prepared.contextPatch || result.contextPatch
const sessionPatch: Record<string, unknown> = {};
if (result.data?.modifiedFiles && result.data.modifiedFiles.length > 0) {
sessionPatch.modifiedFiles = result.data.modifiedFiles;
} else if (prepared.data.modifiedFiles && prepared.data.modifiedFiles.length > 0) {
sessionPatch.modifiedFiles = prepared.data.modifiedFiles;
}
if (result.data?.summary) {
sessionPatch.summary = result.data.summary;
}
const contextPatch = prepared.contextPatch || result.contextPatch || Object.keys(sessionPatch).length > 0
? {
...(prepared.contextPatch ?? {}),
...(result.contextPatch ?? {}),
...sessionPatch,
}
: undefined;
return {