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

fix(FN-6035): project workflow dispatch metadata

Browse Source
This commit is contained in:
gsxdsm
2026-06-08 23:34:35 -07:00
parent 9c43e3f30e
commit 83565a535a
13 changed files with 309 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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;

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" },
],
},

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();

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

@@ -117,6 +117,99 @@ 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("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("caps retries and converts exceptions to failure", async () => {
const ir: WorkflowIr = {
version: "v1",

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

@@ -3751,6 +3751,27 @@ export class TaskExecutor {
seams: this.createAuthoritativeWorkflowSeams(settings),
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<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 } as Task["mergeDetails"];
}
if (patch.summary !== undefined) update.summary = patch.summary;
if (patch.review !== undefined) update.review = patch.review as unknown as Task["review"];
if (patch.reviewState !== undefined) update.reviewState = patch.reviewState as unknown as Task["reviewState"];
if (patch.workflowStepResults !== undefined) update.workflowStepResults = patch.workflowStepResults as Task["workflowStepResults"];
if (patch.tokenUsage !== undefined) update.tokenUsage = patch.tokenUsage as unknown as Task["tokenUsage"];
if (patch.error !== undefined) update.error = patch.error;
if (patch.status !== undefined) update.status = patch.status;
if (Object.keys(update).length > 0) {
await this.store.updateTask(taskId, 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.

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

@@ -38,6 +38,18 @@ export interface WorkflowNodeResult {
contextPatch?: Record<string, unknown>;
}
export interface WorkflowTaskProjection {
modifiedFiles?: string[];
mergeDetails?: Record<string, unknown>;
summary?: string;
review?: Record<string, unknown>;
reviewState?: Record<string, unknown>;
workflowStepResults?: unknown[];
tokenUsage?: Record<string, unknown>;
error?: string | null;
status?: string | null;
}
export interface WorkflowNodeExecutionContext {
task: TaskDetail;
settings: Pick<Settings, "experimentalFeatures"> | undefined;
@@ -131,6 +143,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 +163,59 @@ 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 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);
if (mergeDetails) patch.mergeDetails = mergeDetails;
if (typeof contextPatch.filesChanged === "number" && Number.isFinite(contextPatch.filesChanged)) {
patch.mergeDetails = { ...(patch.mergeDetails ?? {}), filesChanged: contextPatch.filesChanged };
}
if (typeof contextPatch.summary === "string") patch.summary = contextPatch.summary;
const review = objectRecord(contextPatch.review);
if (review) patch.review = review;
const reviewState = objectRecord(contextPatch.reviewState);
if (reviewState) patch.reviewState = reviewState;
if (Array.isArray(contextPatch.workflowStepResults)) patch.workflowStepResults = contextPatch.workflowStepResults;
const tokenUsage = objectRecord(contextPatch.tokenUsage);
if (tokenUsage) patch.tokenUsage = tokenUsage;
if (typeof contextPatch.error === "string" || contextPatch.error === null) patch.error = contextPatch.error;
if (typeof contextPatch.status === "string" || contextPatch.status === null) patch.status = contextPatch.status;
return patch;
}
function hasTaskProjection(patch: WorkflowTaskProjection): boolean {
return Object.keys(patch).length > 0;
}
export class WorkflowGraphExecutor {
private readonly maxRetriesPerNode: number;
@@ -611,11 +680,16 @@ 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) {
await this.publishTaskProjectionFromResult(task.id, node, pluginResult);
return 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 });
await this.publishTaskProjectionFromResult(task.id, node, result);
return result;
} catch (error) {
lastError = error;
}
@@ -629,4 +703,18 @@ export class WorkflowGraphExecutor {
},
};
}
private async publishTaskProjectionFromResult(
taskId: string,
node: WorkflowIrNode,
result: WorkflowNodeResult,
): Promise<void> {
const patch = extractTaskProjection(result.contextPatch);
if (!hasTaskProjection(patch)) return;
const source = { nodeId: node.id, nodeKind: node.kind };
await this.deps.publishTaskProjection?.(taskId, patch, source);
if (patch.modifiedFiles && patch.modifiedFiles.length > 0) {
await this.deps.publishTouchedFiles?.(taskId, patch.modifiedFiles, source);
}
}
}

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 {