diff --git a/packages/cli/skill/fusion/references/engine-tools.md b/packages/cli/skill/fusion/references/engine-tools.md index 08ee7968ea..79e3dae523 100644 --- a/packages/cli/skill/fusion/references/engine-tools.md +++ b/packages/cli/skill/fusion/references/engine-tools.md @@ -16,6 +16,7 @@ These tools are **not** part of the user-invokable extension surface. They are i | `fn_task_document_write` | triage, executor, heartbeat | Save/update a named task document revision | `key` (string), `content` (string), `author?` (string) | | `fn_task_document_read` | triage, executor, heartbeat | Read one task document or list all | `key?` (string) | | `fn_workflow_list` | executor | List the project's custom workflows (read-only built-ins plus user definitions) | none | +| `fn_workflow_get` | executor | Fetch one workflow definition by id — name, description, builtin flag, and the full IR (nodes/edges/columns/artifacts/fields) as JSON | `workflow_id` (string) | | `fn_workflow_select` | executor | Assign a custom workflow to a task (defaults to the current task) | `workflow_id` (string), `task_id?` (string) | | `fn_workflow_create` | executor | Create a custom workflow definition from a graph IR (validated server-side). v2 IR supports step-inversion constructs: `parse-steps`, `foreach` (mode/isolation/concurrency/maxReworkCycles), `step-execute`, `step-review`, `code` nodes, `rework` edges, plus `artifacts` and custom `fields` declarations | `name` (string), `description?` (string), `ir` (object), `layout?` (object) | | `fn_workflow_update` | executor | Update a custom workflow definition's name/description/ir/layout (built-ins cannot be edited; same step-inversion IR constructs as create; editing `fields` orphans rather than destroys existing task values) | `workflow_id` (string), `name?` (string), `description?` (string), `ir?` (object), `layout?` (object), `rehome_to?` (string) | diff --git a/packages/dashboard/app/api/legacy.ts b/packages/dashboard/app/api/legacy.ts index 680c2dfde8..81e06dc866 100644 --- a/packages/dashboard/app/api/legacy.ts +++ b/packages/dashboard/app/api/legacy.ts @@ -79,6 +79,10 @@ import type { TaskIdIntegrityReport, BranchGroup, BranchGroupPrState, + WorkflowFieldDefinition, + WorkflowFieldType, + WorkflowFieldOption, + WorkflowFieldRender, } from "@fusion/core"; import type { PlanningQuestion, PlanningSummary } from "@fusion/core"; import type { GithubIssueAction, ScheduledTask, ScheduledTaskCreateInput, ScheduledTaskUpdateInput, AutomationRunResult, Routine, RoutineCreateInput, RoutineUpdateInput, RoutineExecutionResult } from "@fusion/core"; @@ -552,44 +556,9 @@ export interface BoardWorkflowColumn { flags: BoardWorkflowColumnFlags; } -/** Supported custom-field value types (mirrors core `WorkflowFieldType`, KTD-13). - * Duplicated client-side (same posture as the BoardWorkflow* types above) since - * the core field-schema types are not exported through the `@fusion/core` - * barrel. */ -export type WorkflowFieldType = - | "string" - | "text" - | "number" - | "boolean" - | "enum" - | "multi-enum" - | "date" - | "url"; - -/** A single enum/multi-enum option (KTD-13). */ -export interface WorkflowFieldOption { - value: string; - label: string; - color?: string; -} - -/** Rendering instructions for a custom field (KTD-14). */ -export interface WorkflowFieldRender { - placement?: "card" | "detail" | "detail-section"; - widget?: "select" | "radio" | "chips" | "input" | "textarea" | "toggle"; - badge?: boolean; -} - -/** A workflow-defined custom task field (KTD-13). */ -export interface WorkflowFieldDefinition { - id: string; - name: string; - type: WorkflowFieldType; - required?: boolean; - default?: unknown; - options?: WorkflowFieldOption[]; - render?: WorkflowFieldRender; -} +// WorkflowFieldDefinition, WorkflowFieldType, WorkflowFieldOption, WorkflowFieldRender +// are re-exported from @fusion/core above (KTD-13/14). +export type { WorkflowFieldDefinition, WorkflowFieldType, WorkflowFieldOption, WorkflowFieldRender }; export interface BoardWorkflowDefinition { id: string; diff --git a/packages/dashboard/app/components/Board.tsx b/packages/dashboard/app/components/Board.tsx index 2e7bb45160..9b7ed08a90 100644 --- a/packages/dashboard/app/components/Board.tsx +++ b/packages/dashboard/app/components/Board.tsx @@ -379,27 +379,32 @@ export function Board({ tasks, projectId, maxConcurrent, onMoveTask, onPauseTask return result; }, [boardWorkflows, flagOn, tasks]); - // Card-placed custom field definitions per task (U13/KTD-14). Resolves each - // task's workflow from the board-workflows payload and exposes that workflow's - // card-placed field defs so TaskCard can render value badges. Empty map when - // no workflow declares card fields — cards stay byte-identical. - const taskCardFieldDefs = useMemo(() => { + // Card-placed field defs grouped by workflow id (U13/KTD-14). Only recomputes + // when the board-workflows payload changes, not on every SSE task tick. + const cardDefsByWorkflow = useMemo(() => { const map = new Map(); if (!boardWorkflows) return map; - const { workflows, taskWorkflowIds, defaultWorkflowId } = boardWorkflows; - const cardDefsByWorkflow = new Map(); - for (const wf of workflows) { + for (const wf of boardWorkflows.workflows) { const cardDefs = (wf.fields ?? []).filter((f) => f.render?.placement === "card"); - if (cardDefs.length > 0) cardDefsByWorkflow.set(wf.id, cardDefs); + if (cardDefs.length > 0) map.set(wf.id, cardDefs); } + return map; + }, [boardWorkflows]); + + // Per-task card field defs (U13/KTD-14). Recomputes on task list changes but + // reuses the stable cardDefsByWorkflow map so the inner loop is cheap. + const taskCardFieldDefs = useMemo(() => { + const map = new Map(); if (cardDefsByWorkflow.size === 0) return map; + if (!boardWorkflows) return map; + const { taskWorkflowIds, defaultWorkflowId } = boardWorkflows; for (const task of tasks) { const workflowId = taskWorkflowIds[task.id] ?? defaultWorkflowId; const defs = cardDefsByWorkflow.get(workflowId); if (defs) map.set(task.id, defs); } return map; - }, [boardWorkflows, tasks]); + }, [cardDefsByWorkflow, tasks, boardWorkflows]); // Drag pre-check (R17): adjacency + capacity from the lane's column metadata. // Cross-lane drag → workflow-mismatch. Deterministic rejections return a diff --git a/packages/dashboard/app/components/TaskCard.tsx b/packages/dashboard/app/components/TaskCard.tsx index 0b32a8fe55..373e1b4824 100644 --- a/packages/dashboard/app/components/TaskCard.tsx +++ b/packages/dashboard/app/components/TaskCard.tsx @@ -541,7 +541,9 @@ function areTaskCardPropsEqual(previous: TaskCardProps, next: TaskCardProps): bo previous.prAuthAvailable === next.prAuthAvailable && previous.autoMergeEnabled === next.autoMergeEnabled && previous.cardFieldDefs === next.cardFieldDefs && - JSON.stringify(previousTask.customFields ?? null) === JSON.stringify(nextTask.customFields ?? null) && + (previous.cardFieldDefs == null && next.cardFieldDefs == null + ? true + : JSON.stringify(previousTask.customFields ?? null) === JSON.stringify(nextTask.customFields ?? null)) && previous.onOpenDetail === next.onOpenDetail && previous.onOpenGroupModal === next.onOpenGroupModal && previous.addToast === next.addToast && diff --git a/packages/dashboard/app/components/TaskDetailModal.tsx b/packages/dashboard/app/components/TaskDetailModal.tsx index 655fbfc77c..0d32ed7567 100644 --- a/packages/dashboard/app/components/TaskDetailModal.tsx +++ b/packages/dashboard/app/components/TaskDetailModal.tsx @@ -308,6 +308,11 @@ export interface TaskDetailModalProps { initialTab?: TabId; /** Mobile-only header affordance mode. */ mobileHeaderMode?: "close" | "back"; + /** Pre-resolved workflow field defs for this task's workflow (U13/KTD-14). + * When provided (e.g. threaded from a Board that already holds the payload) + * the modal skips its own board-workflows fetch entirely. Falls back to the + * self-fetch when absent (e.g. modal opened from non-board contexts). */ + workflowFieldDefs?: WorkflowFieldDefinition[] | null; } export type TaskDetailContentProps = Omit & { @@ -483,6 +488,7 @@ export function TaskDetailContent({ mobileHeaderMode = "close", embedded = false, onRequestClose, + workflowFieldDefs: workflowFieldDefsProp, }: TaskDetailContentProps) { const { t } = useTranslation("app"); const columnLabel = useColumnLabel(); @@ -610,7 +616,11 @@ export function TaskDetailContent({ // Custom field definitions (U13/KTD-14). Resolved for this task's workflow // from the board-workflows payload; absent when the workflow declares none, // in which case the fields section renders nothing (today's UI byte-identical). - const [customFieldDefs, setCustomFieldDefs] = useState(null); + // When `workflowFieldDefsProp` is provided by the caller (e.g. the Board + // already holds the payload) we skip the self-fetch entirely. + const [customFieldDefs, setCustomFieldDefs] = useState( + workflowFieldDefsProp !== undefined ? (workflowFieldDefsProp ?? null) : null, + ); const [customFieldValues, setCustomFieldValues] = useState>(task.customFields ?? {}); const [customFieldError, setCustomFieldError] = useState(null); @@ -619,9 +629,15 @@ export function TaskDetailContent({ setCustomFieldValues(task.customFields ?? {}); }, [task.id, task.customFields]); - // Resolve this task's workflow field definitions once per task. Best-effort: + // Resolve this task's workflow field definitions once per task. Skipped when + // the caller supplies `workflowFieldDefs` directly (Board context). Best-effort: // a failed fetch (or flag-OFF empty payload) leaves defs null → no section. useEffect(() => { + if (workflowFieldDefsProp !== undefined) { + // Prop-driven path: keep in sync if the prop changes (task switch etc.). + setCustomFieldDefs(workflowFieldDefsProp ?? null); + return; + } let cancelled = false; void fetchBoardWorkflows(projectId) .then((payload) => { @@ -636,7 +652,7 @@ export function TaskDetailContent({ return () => { cancelled = true; }; - }, [task.id, projectId]); + }, [task.id, projectId, workflowFieldDefsProp]); const handleSaveCustomFields = useCallback( async (patch: Record) => { diff --git a/packages/dashboard/app/components/WorkflowNodeEditor.tsx b/packages/dashboard/app/components/WorkflowNodeEditor.tsx index 6c159b1174..1afbb8eb00 100644 --- a/packages/dashboard/app/components/WorkflowNodeEditor.tsx +++ b/packages/dashboard/app/components/WorkflowNodeEditor.tsx @@ -197,7 +197,7 @@ function InnerEditor({ setNodes(flow.nodes); setEdges(flow.edges); setColumns(columnsOf(activeWorkflow)); - setFields(fieldsOf(activeWorkflow) as WorkflowFieldDefinition[]); + setFields(fieldsOf(activeWorkflow)); setSelectedNodeId(null); setSelectedEdgeId(null); setValidationError(null); diff --git a/packages/dashboard/app/components/__tests__/TaskCard.test.tsx b/packages/dashboard/app/components/__tests__/TaskCard.test.tsx index c1f1910247..aea3b5c08b 100644 --- a/packages/dashboard/app/components/__tests__/TaskCard.test.tsx +++ b/packages/dashboard/app/components/__tests__/TaskCard.test.tsx @@ -4187,6 +4187,42 @@ describe("TaskCard memo comparator provenance behavior", () => { ), ).toBe(false); }); + + it("skips customFields JSON.stringify when both cardFieldDefs are absent", () => { + // Without cardFieldDefs present, two tasks with different customFields should + // compare equal (JSON.stringify is skipped — guard path). + const taskA = makeTask({ customFields: { x: "a" } }); + const taskB = makeTask({ customFields: { x: "b" } }); + expect( + __test_areTaskCardPropsEqual( + { task: taskA, onOpenDetail: noop, addToast: noop } as any, + { task: taskB, onOpenDetail: noop, addToast: noop } as any, + ), + ).toBe(true); + }); + + it("detects customFields change when cardFieldDefs are present on both sides", () => { + const defs = [{ id: "sev", name: "Severity", type: "enum" as const, render: { placement: "card" as const } }]; + const taskA = makeTask({ customFields: { sev: "low" } }); + const taskB = makeTask({ customFields: { sev: "high" } }); + expect( + __test_areTaskCardPropsEqual( + { task: taskA, cardFieldDefs: defs, onOpenDetail: noop, addToast: noop } as any, + { task: taskB, cardFieldDefs: defs, onOpenDetail: noop, addToast: noop } as any, + ), + ).toBe(false); + }); + + it("detects customFields change when only one side has cardFieldDefs", () => { + const defs = [{ id: "sev", name: "Severity", type: "enum" as const, render: { placement: "card" as const } }]; + const task = makeTask({ customFields: { sev: "low" } }); + expect( + __test_areTaskCardPropsEqual( + { task, cardFieldDefs: undefined, onOpenDetail: noop, addToast: noop } as any, + { task, cardFieldDefs: defs, onOpenDetail: noop, addToast: noop } as any, + ), + ).toBe(false); + }); }); describe("TaskCard mission badge", () => { diff --git a/packages/dashboard/app/components/__tests__/TaskDetailModal.custom-fields.test.tsx b/packages/dashboard/app/components/__tests__/TaskDetailModal.custom-fields.test.tsx index e25a228376..b542cde581 100644 --- a/packages/dashboard/app/components/__tests__/TaskDetailModal.custom-fields.test.tsx +++ b/packages/dashboard/app/components/__tests__/TaskDetailModal.custom-fields.test.tsx @@ -1,5 +1,6 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import { render, screen, waitFor } from "@testing-library/react"; +import type { WorkflowFieldDefinition } from "../../api"; import { makeTask, noop, @@ -67,4 +68,51 @@ describe("TaskDetailModal custom fields (U13/KTD-14)", () => { await waitFor(() => expect(screen.getByTestId("task-fields-section")).toBeTruthy()); expect((screen.getByLabelText("Owner") as HTMLInputElement).value).toBe("alice"); }); + + it("uses workflowFieldDefs prop directly and skips the board-workflows fetch", async () => { + const fetchSpy = vi.spyOn(dashboardApi, "fetchBoardWorkflows"); + const defs: WorkflowFieldDefinition[] = [ + { id: "owner", name: "Owner", type: "string", render: { placement: "detail" } }, + ]; + render( + + + , + ); + await waitFor(() => expect(screen.getByTestId("task-fields-section")).toBeTruthy()); + expect((screen.getByLabelText("Owner") as HTMLInputElement).value).toBe("bob"); + // The fetch must NOT have been triggered since the prop was provided. + expect(fetchSpy).not.toHaveBeenCalled(); + }); + + it("renders no fields section when workflowFieldDefs prop is an empty array", async () => { + const fetchSpy = vi.spyOn(dashboardApi, "fetchBoardWorkflows"); + render( + + + , + ); + // Give React a tick to settle; no section should appear. + await new Promise((r) => setTimeout(r, 50)); + expect(screen.queryByTestId("task-fields-section")).toBeNull(); + expect(fetchSpy).not.toHaveBeenCalled(); + }); }); diff --git a/packages/dashboard/app/components/workflow-flow-mapping.ts b/packages/dashboard/app/components/workflow-flow-mapping.ts index 44cb4df8c4..6260ca8083 100644 --- a/packages/dashboard/app/components/workflow-flow-mapping.ts +++ b/packages/dashboard/app/components/workflow-flow-mapping.ts @@ -6,6 +6,7 @@ import type { WorkflowIrNode, WorkflowIrEdge, WorkflowDefinition, + WorkflowFieldDefinition, } from "@fusion/core"; import type { WorkflowFlowNodeData, WorkflowEditorNodeKind } from "./nodes/WorkflowNodeTypes"; @@ -21,19 +22,10 @@ interface WorkflowForeachConfig { template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] }; } -/** Local mirror of @fusion/core's WorkflowFieldDefinition (KTD-13). The core - * barrel does not re-export it and the dashboard build aliases @fusion/core to - * a types-only entry; the editor only needs to carry the array through the - * IR<->flow round-trip without inspecting it, so this minimal shape suffices. */ -export interface WorkflowFieldDefinitionShape { - id: string; - name: string; - type: string; - required?: boolean; - default?: unknown; - options?: { value: string; label: string; color?: string }[]; - render?: { placement?: string; widget?: string; badge?: boolean }; -} +// WorkflowFieldDefinition is imported from @fusion/core above (KTD-13/14). +// Re-exported so existing importers that reference WorkflowFieldDefinitionShape +// can migrate; callers should prefer WorkflowFieldDefinition directly. +export type { WorkflowFieldDefinition as WorkflowFieldDefinitionShape }; // ── foreach template region (KTD-3, U8) ────────────────────────────────────── // @@ -286,7 +278,7 @@ export function flowToIr( nodes: FlowNode[], edges: FlowEdge[], columns?: WorkflowIrColumn[], - fields?: WorkflowFieldDefinitionShape[], + fields?: WorkflowFieldDefinition[], ): { ir: WorkflowIr; layout: Record } { const realNodes = nodes.filter((n) => !isColumnBandNode(n.id)); // Partition by parentId: foreach group children reassemble into that group's @@ -540,8 +532,8 @@ export function columnsOf(def: WorkflowDefinition): WorkflowIrColumn[] { /** Extract the editor's working custom-field list from a definition (KTD-13). * v2 with `fields` → a deep-ish copy; v1 or no fields → empty. */ -export function fieldsOf(def: WorkflowDefinition): WorkflowFieldDefinitionShape[] { - const ir = def.ir as { fields?: WorkflowFieldDefinitionShape[] }; +export function fieldsOf(def: WorkflowDefinition): WorkflowFieldDefinition[] { + const ir = def.ir as { fields?: WorkflowFieldDefinition[] }; if (!isV2(def.ir) || !Array.isArray(ir.fields)) return []; return ir.fields.map((f) => ({ ...f, diff --git a/packages/dashboard/src/routes/board-workflows.ts b/packages/dashboard/src/routes/board-workflows.ts index 9aeeae188f..0a328d0048 100644 --- a/packages/dashboard/src/routes/board-workflows.ts +++ b/packages/dashboard/src/routes/board-workflows.ts @@ -28,26 +28,13 @@ import { type TraitFlags, type WorkflowIr, type WorkflowIrV2, + type WorkflowFieldDefinition, } from "@fusion/core"; -/** A workflow-defined custom task field as the board client needs it (U13/ - * KTD-14). Structurally mirrors core's `WorkflowFieldDefinition`; declared - * locally because the core field-schema types are not exported through the - * `@fusion/core` barrel. The payload is a verbatim pass-through of the IR's - * `fields` array. */ -export interface BoardWorkflowField { - id: string; - name: string; - type: "string" | "text" | "number" | "boolean" | "enum" | "multi-enum" | "date" | "url"; - required?: boolean; - default?: unknown; - options?: Array<{ value: string; label: string; color?: string }>; - render?: { - placement?: "card" | "detail" | "detail-section"; - widget?: "select" | "radio" | "chips" | "input" | "textarea" | "toggle"; - badge?: boolean; - }; -} +/** A workflow-defined custom task field as the board client needs it (U13/KTD-14). + * Uses @fusion/core's WorkflowFieldDefinition directly now that it is exported + * through the barrel. The payload is a verbatim pass-through of the IR's `fields` array. */ +export type BoardWorkflowField = WorkflowFieldDefinition; /** Stable id the client uses for the implicit default lane (null selection). */ export const DEFAULT_WORKFLOW_LANE_ID = "builtin:coding"; @@ -102,7 +89,7 @@ function describeFields(ir: WorkflowIr): BoardWorkflowField[] | undefined { const v2 = toV2(ir); const fields = v2?.fields; if (!fields || fields.length === 0) return undefined; - return fields as BoardWorkflowField[]; + return fields; } async function describeWorkflow( diff --git a/packages/engine/src/__tests__/agent-tools.test.ts b/packages/engine/src/__tests__/agent-tools.test.ts index 3bd495269b..8f48082085 100644 --- a/packages/engine/src/__tests__/agent-tools.test.ts +++ b/packages/engine/src/__tests__/agent-tools.test.ts @@ -16,6 +16,7 @@ import { createPostRoomMessageTool, createResearchTools, createWorkflowListTool, + createWorkflowGetTool, createWorkflowSelectTool, createTaskPromoteTool, createWorkflowCreateTool, @@ -393,6 +394,55 @@ describe("createWorkflowListTool", () => { }); }); +describe("createWorkflowGetTool", () => { + it("returns the definition with builtin flag and full IR as JSON", async () => { + const ir = { + version: "v2", + name: "QA", + columns: [{ id: "intake", name: "Intake", traits: [] }], + nodes: [{ id: "n1", kind: "step-execute" }], + edges: [], + fields: [{ id: "severity", name: "Severity", type: "enum", options: [{ value: "low", label: "Low" }] }], + }; + const store = { + getWorkflowDefinition: vi.fn().mockResolvedValue({ id: "WF-003", name: "QA", description: "QA flow", ir }), + }; + const tool = createWorkflowGetTool(store as any); + const result = await tool.execute("call-1", { workflow_id: "WF-003" } as any, undefined, undefined, {} as any); + expect(store.getWorkflowDefinition).toHaveBeenCalledWith("WF-003"); + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + const parsed = JSON.parse(text); + expect(parsed).toMatchObject({ id: "WF-003", name: "QA", description: "QA flow", builtin: false }); + expect(parsed.ir.fields[0].id).toBe("severity"); + expect(result.details).toMatchObject({ workflowId: "WF-003", builtin: false }); + }); + + it("marks a builtin id as builtin", async () => { + const store = { + getWorkflowDefinition: vi.fn().mockResolvedValue({ + id: "builtin:coding", + name: "Coding", + description: "Standard", + ir: { version: "v2", name: "Coding", columns: [], nodes: [], edges: [] }, + }), + }; + const tool = createWorkflowGetTool(store as any); + const result = await tool.execute("call-1", { workflow_id: "builtin:coding" } as any, undefined, undefined, {} as any); + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + expect(JSON.parse(text).builtin).toBe(true); + expect(result.details).toMatchObject({ builtin: true }); + }); + + it("returns an error result for an unknown id", async () => { + const store = { getWorkflowDefinition: vi.fn().mockResolvedValue(undefined) }; + const tool = createWorkflowGetTool(store as any); + const result = await tool.execute("call-1", { workflow_id: "WF-404" } as any, undefined, undefined, {} as any); + expect((result as { isError?: boolean }).isError).toBe(true); + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + expect(text).toMatch(/Unknown workflow id 'WF-404'/); + }); +}); + describe("createWorkflowSelectTool", () => { it("selects for the current task by default and reports enabled step count", async () => { const store = { diff --git a/packages/engine/src/__tests__/executor-prompt.test.ts b/packages/engine/src/__tests__/executor-prompt.test.ts index cc0164f084..e1e6727942 100644 --- a/packages/engine/src/__tests__/executor-prompt.test.ts +++ b/packages/engine/src/__tests__/executor-prompt.test.ts @@ -145,6 +145,34 @@ describe("buildExecutionPrompt", () => { expect(result).not.toContain("## Attachments"); }); + it("includes Custom fields section listing id/name/type, enum options, required, and current value", () => { + const task = createMockTaskDetail({ customFields: { severity: "high" } }); + const result = buildExecutionPrompt(task, "/home/user/project", undefined, undefined, undefined, [ + { id: "severity", name: "Severity", type: "enum", required: true, options: [ + { value: "low", label: "Low" }, + { value: "high", label: "High" }, + ] }, + { id: "notes", name: "Notes", type: "text" }, + ] as any); + + expect(result).toContain("## Custom fields"); + expect(result).toContain("`severity` (Severity) — type: enum"); + expect(result).toContain("options: [low (Low), high (High)]"); + expect(result).toContain("required"); + expect(result).toContain('current: "high"'); + // The unset field reports "unset". + expect(result).toContain("`notes` (Notes) — type: text; current: unset"); + }); + + it("omits Custom fields section when no field defs are provided", () => { + const task = createMockTaskDetail(); + const result = buildExecutionPrompt(task, "/home/user/project"); + expect(result).not.toContain("## Custom fields"); + + const resultEmpty = buildExecutionPrompt(task, "/home/user/project", undefined, undefined, undefined, []); + expect(resultEmpty).not.toContain("## Custom fields"); + }); + it("includes Project Commands section with test command when settings.testCommand is set", () => { const task = createMockTaskDetail(); const result = buildExecutionPrompt(task, "/home/user/project", { @@ -2514,3 +2542,33 @@ describe("TaskExecutor global pause behavior", () => { }); }); +describe("fn_task_update bare-call guard (P1 api-contract)", () => { + // createTaskUpdateTool is a private executor method; the bare-call guard runs + // before any store access, so we reach it via the lowest-cost seam: construct + // a TaskExecutor over a mock store and invoke the private method with `as any`. + function makeTool() { + const store = createMockStore(); + const executor = new TaskExecutor(store, "/tmp/test"); + return (executor as any).createTaskUpdateTool("FN-001", new Map(), { current: null }, new Map()); + } + + it("returns isError with a self-describing message when no fields are supplied", async () => { + const tool = makeTool(); + const result = await tool.execute("call-1", {}); + expect(result.isError).toBe(true); + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + expect(text).toContain("fn_task_update requires at least one of"); + // The legacy no-op text is preserved as the detail. + expect(text).toContain("No-op: provide a step+status, dependencies, or custom_fields to update."); + }); + + it("does not trigger the guard when a dependencies-only patch is supplied", async () => { + const tool = makeTool(); + const result = await tool.execute("call-1", { dependencies: [] }); + // Reaches the dependencies path, not the bare-call guard. + expect(result.isError).not.toBe(true); + const text = result.content[0]?.type === "text" ? result.content[0].text : ""; + expect(text).not.toContain("fn_task_update requires at least one of"); + }); +}); + diff --git a/packages/engine/src/__tests__/workflow-foreach-wiring.test.ts b/packages/engine/src/__tests__/workflow-foreach-wiring.test.ts new file mode 100644 index 0000000000..bb77a7f189 --- /dev/null +++ b/packages/engine/src/__tests__/workflow-foreach-wiring.test.ts @@ -0,0 +1,237 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { TaskStore } from "@fusion/core"; +import type { TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core"; + +/** The exact param type the store's save method expects (WorkflowRunStepInstance + * is not exported via the barrel; derive it from the method signature). */ +type SaveInstanceArg = Parameters[0]; + +import { WorkflowGraphExecutor } from "../workflow-graph-executor.js"; +import type { + IntegrationGitOps, + IntegrationProjection, +} from "../step-integration.js"; +import type { WorkflowStepInstancePersistence, WorkflowStepInstanceState } from "../workflow-graph-foreach.js"; +import { type WorkflowLegacySeams } from "../workflow-node-handlers.js"; + +/** + * runId/foreachNodeId wiring regression coverage (FIX 1). These tests wire the + * REAL store (an in-memory TaskStore) through store-backed persistence + projection + * adapters that MIRROR the executor's production adapters, then assert that: + * (i) after a foreach expands + persists, a pin-protection probe under the + * PRODUCTION runId sees the rows (would be empty under the old `:run` literal); + * (ii) markInstanceIntegrated flips the SAME row the sub-walk persisted (status + * completed, integratedAt set) with NO orphan row; + * (iii) a resume load under the production runId sees the persisted rows. + */ + +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +function makeTmpDir(): string { + return mkdtempSync(join(tmpdir(), "fn-foreach-wiring-")); +} + +/** The production run id derivation: `${task.id}:${definition.id}`. */ +const DEFINITION_ID = "wf-coding"; +const runIdFor = (taskId: string) => `${taskId}:${DEFINITION_ID}`; + +/** Store-backed step-instance persistence — MIRRORS executor.buildStepInstancePersistence. */ +function storePersistence(store: TaskStore): WorkflowStepInstancePersistence { + return { + saveInstanceState: (state) => + store.saveWorkflowRunStepInstance(state as unknown as SaveInstanceArg), + loadInstanceStates: (taskId, runId) => + store.loadWorkflowRunStepInstances(taskId, runId) as unknown as WorkflowStepInstanceState[], + clearStaleInstanceStates: (taskId, keepRunId) => store.clearWorkflowRunStepInstances(taskId, keepRunId), + }; +} + +/** Store-backed projection — MIRRORS executor.buildForeachWorktreeDeps.integrationProjection. + * Critically, markInstanceIntegrated flips the EXISTING row by its REAL identity. */ +function storeProjection(store: TaskStore): IntegrationProjection { + return { + markStepDone: async (stepIndex) => { + await store.updateStep(STORED_TASK_ID, stepIndex, "done", { source: "graph" }); + }, + markInstanceIntegrated: async (stepIndex, integratedAt, identity) => { + const rows = store.loadWorkflowRunStepInstances(STORED_TASK_ID, identity.runId); + const existing = rows.find( + (r) => r.foreachNodeId === identity.foreachNodeId && r.stepIndex === stepIndex, + ); + store.saveWorkflowRunStepInstance({ + ...(existing ?? {}), + taskId: STORED_TASK_ID, + runId: identity.runId, + foreachNodeId: identity.foreachNodeId, + stepIndex, + pinnedStepCount: identity.pinnedStepCount, + currentNodeId: existing?.currentNodeId ?? "", + status: "completed", + reworkCount: existing?.reworkCount ?? 0, + branchName: identity.branchName, + integratedAt, + } as unknown as SaveInstanceArg); + }, + }; +} + +let STORED_TASK_ID = ""; + +function fakeGitOps(): IntegrationGitOps { + return { + integrate: async () => ({ kind: "integrated" as const, integratedAt: "2026-01-01T00:00:00Z" }), + discardBranch: async () => {}, + }; +} + +function singleExecuteTemplate(): { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] } { + return { + nodes: [{ id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }], + edges: [], + }; +} + +function foreachIr(config: Record): WorkflowIr { + return { + version: "v2", + name: "wiring-test", + columns: [{ id: "work", name: "Work", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "fe", kind: "foreach", config: { source: "task-steps", template: singleExecuteTemplate(), ...config } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "fe" }, + { from: "fe", to: "end", condition: "success" }, + ], + } as WorkflowIr; +} + +function baseSeams(overrides: Partial): WorkflowLegacySeams { + const ok = async () => ({ outcome: "success" as const }); + return { planning: ok, execute: ok, review: ok, merge: ok, schedule: ok, ...overrides }; +} + +describe("foreach runId/foreachNodeId wiring (FIX 1)", () => { + let rootDir: string; + let globalDir: string; + let store: TaskStore; + let taskId: string; + + beforeEach(async () => { + rootDir = makeTmpDir(); + globalDir = join(rootDir, ".fusion-global"); + store = new TaskStore(rootDir, globalDir); + await store.init(); + const task = await store.createTask({ description: "wiring task" }); + taskId = task.id; + STORED_TASK_ID = taskId; + // Two steps so the foreach expands two instances. + await store.updateTask(taskId, { + steps: [ + { name: "Step 1", status: "pending" }, + { name: "Step 2", status: "pending" }, + ], + }); + }); + + afterEach(async () => { + store?.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + function makeExecutor() { + const stepExecuteCalls: number[] = []; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx["foreach:active"] as { stepIndex: number } | undefined; + if (active) stepExecuteCalls.push(active.stepIndex); + return { outcome: "success" as const, value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ + seams, + runCustomNode: async () => ({ outcome: "success" as const }), + stepInstancePersistence: storePersistence(store), + // Worktree isolation deps (parallel). + allocateInstanceWorktree: async (stepIndex) => ({ + worktreePath: `/wt/step-${stepIndex}`, + branchName: `fusion/${taskId.toLowerCase()}-step-${stepIndex}`, + }), + resolveIntegrationBase: async () => "base", + integrationGitOps: fakeGitOps(), + integrationProjection: storeProjection(store), + semaphoreAvailability: () => 8, + // The PRODUCTION runId, threaded as the single source of truth. + runId: runIdFor(taskId), + }); + return { executor, stepExecuteCalls }; + } + + it("(i)+(ii) expands+persists under the production runId and integration flips the SAME row (no orphans)", async () => { + const detail = (await store.getTask(taskId)) as unknown as TaskDetail; + const { executor } = makeExecutor(); + const result = await executor.run(detail, settingsOn(), foreachIr({ mode: "parallel" })); + expect(result.outcome).toBe("success"); + + const prodRunId = runIdFor(taskId); + + // (i) Pin-protection probe under the PRODUCTION runId sees rows; the old buggy + // `${taskId}:run` literal would see nothing. + const rowsProd = store.loadWorkflowRunStepInstances(taskId, prodRunId); + expect(rowsProd.length).toBe(2); + expect(store.loadWorkflowRunStepInstances(taskId, `${taskId}:run`)).toEqual([]); + + // (ii) Each instance row was FLIPPED in place to completed/integratedAt — and + // there are NO orphan rows (no foreachNodeId:"" rows, exactly 2 rows total). + expect(rowsProd.every((r) => r.foreachNodeId === "fe")).toBe(true); + expect(rowsProd.every((r) => r.status === "completed")).toBe(true); + expect(rowsProd.every((r) => typeof r.integratedAt === "string" && r.integratedAt)).toBe(true); + expect(rowsProd.some((r) => r.foreachNodeId === "")).toBe(false); + // Both steps are done in the projection. + const after = await store.getTask(taskId); + expect(after.steps.map((s) => s.status)).toEqual(["done", "done"]); + }); + + it("(iii) a resume load under the production runId sees the persisted rows", async () => { + const detail = (await store.getTask(taskId)) as unknown as TaskDetail; + const { executor } = makeExecutor(); + await executor.run(detail, settingsOn(), foreachIr({ mode: "parallel" })); + + // Resume-equivalent probe: load under the production runId. + const rows = store.loadWorkflowRunStepInstances(taskId, runIdFor(taskId)); + expect(rows.length).toBe(2); + expect(rows.map((r) => r.stepIndex).sort()).toEqual([0, 1]); + expect(rows.every((r) => r.branchName?.includes("step-"))).toBe(true); + }); + + it("prunes stale-run instance rows at run start, keeping the current run", async () => { + // Seed a stale row from a prior run. + store.saveWorkflowRunStepInstance({ + taskId, + runId: `${taskId}:stale-run`, + foreachNodeId: "fe", + stepIndex: 0, + pinnedStepCount: 1, + currentNodeId: "exec", + status: "in-progress", + reworkCount: 0, + updatedAt: new Date().toISOString(), + }); + expect(store.loadWorkflowRunStepInstances(taskId, `${taskId}:stale-run`).length).toBe(1); + + const detail = (await store.getTask(taskId)) as unknown as TaskDetail; + const { executor } = makeExecutor(); + await executor.run(detail, settingsOn(), foreachIr({ mode: "parallel" })); + + // The stale run's rows were pruned at run start (keepRunId = production runId). + expect(store.loadWorkflowRunStepInstances(taskId, `${taskId}:stale-run`)).toEqual([]); + // The current run's rows survive. + expect(store.loadWorkflowRunStepInstances(taskId, runIdFor(taskId)).length).toBe(2); + }); +}); diff --git a/packages/engine/src/__tests__/workflow-graph-step-rerun.test.ts b/packages/engine/src/__tests__/workflow-graph-step-rerun.test.ts new file mode 100644 index 0000000000..68436922fe --- /dev/null +++ b/packages/engine/src/__tests__/workflow-graph-step-rerun.test.ts @@ -0,0 +1,96 @@ +// -nocheck +import { describe, it, expect, beforeEach, vi } from "vitest"; +import "./executor-test-helpers.js"; +import { TaskExecutor } from "../executor.js"; +import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js"; +import type { Task } from "@fusion/core"; + +/** + * FIX 3: runGraphTaskStep single-flight-per-attempt + rejection memo clearing. + * + * The implementation phase is memoized once per run (graphStepRunOnce) so each + * foreach instance's runStep observes the projection instead of re-running the + * agent. Two regressions are covered: + * - a REJECTED phase must clear the memo so a rework cycle RE-INVOKES the + * implementation (the prior code re-awaited the stored rejection forever); + * - the projection consult must NOT mask a step-session failure: a non-terminal + * step with no deferred review returns success:false (the prior code returned + * success on both branches). + */ +describe("runGraphTaskStep (FIX 3)", () => { + beforeEach(() => resetExecutorMocks()); + + function makeExecutor(stepStatus: string | undefined, active?: { deferDoneToReview?: boolean }) { + const store = createMockStore(); + store.getTask = vi.fn().mockResolvedValue({ + id: "FN-001", + steps: stepStatus ? [{ name: "S1", status: stepStatus }] : [{ name: "S1", status: "pending" }], + }); + const executor: any = new TaskExecutor(store, "/tmp/test", {}); + // Stamp the active foreach context the seam would normally stamp. + if (active) executor.graphStepActiveContext.set("FN-001", { stepIndex: 0, ...active }); + return { executor, store }; + } + + const task = { id: "FN-001" } as Task; + + it("re-invokes the implementation after a rejected phase (rework retries)", async () => { + const { executor } = makeExecutor("pending", { deferDoneToReview: true }); + let calls = 0; + executor.runImplementationPhase = vi.fn().mockImplementation(async () => { + calls += 1; + if (calls === 1) throw new Error("impl failed"); + return { taskDone: true, modifiedFiles: [] }; + }); + + // First attempt: implementation rejects → failure, memo cleared. + const first = await executor.runGraphTaskStep(task, 0); + expect(first.success).toBe(false); + expect(calls).toBe(1); + + // Rework re-run: the memo was cleared, so the implementation is invoked AGAIN + // (the bug left a poisoned rejected promise that was re-awaited forever). + const second = await executor.runGraphTaskStep(task, 0); + expect(calls).toBe(2); + expect(second.success).toBe(true); + }); + + it("single-flight within one attempt: concurrent callers share one phase", async () => { + const { executor } = makeExecutor("done"); + let calls = 0; + executor.runImplementationPhase = vi.fn().mockImplementation(async () => { + calls += 1; + await Promise.resolve(); + return { taskDone: true, modifiedFiles: [] }; + }); + const [a, b] = await Promise.all([ + executor.runGraphTaskStep(task, 0), + executor.runGraphTaskStep(task, 0), + ]); + expect(a.success).toBe(true); + expect(b.success).toBe(true); + expect(calls).toBe(1); // memoized — exactly one implementation pass. + }); + + it("does NOT mask a step-session failure: non-terminal step without review → failure", async () => { + const { executor } = makeExecutor("in-progress"); // never reaches done/skipped, no deferDoneToReview + executor.runImplementationPhase = vi.fn().mockResolvedValue({ taskDone: false, modifiedFiles: [] }); + const result = await executor.runGraphTaskStep(task, 0); + expect(result.success).toBe(false); + expect(result.error).toMatch(/not completed/); + }); + + it("deferDoneToReview: a non-terminal step is success (review authors done)", async () => { + const { executor } = makeExecutor("in-progress", { deferDoneToReview: true }); + executor.runImplementationPhase = vi.fn().mockResolvedValue({ taskDone: false, modifiedFiles: [] }); + const result = await executor.runGraphTaskStep(task, 0); + expect(result.success).toBe(true); + }); + + it("terminal step (done) is success regardless of review", async () => { + const { executor } = makeExecutor("done"); + executor.runImplementationPhase = vi.fn().mockResolvedValue({ taskDone: true, modifiedFiles: [] }); + const result = await executor.runGraphTaskStep(task, 0); + expect(result.success).toBe(true); + }); +}); diff --git a/packages/engine/src/__tests__/workflow-step-parallel.test.ts b/packages/engine/src/__tests__/workflow-step-parallel.test.ts index 921962d085..ab32165ebd 100644 --- a/packages/engine/src/__tests__/workflow-step-parallel.test.ts +++ b/packages/engine/src/__tests__/workflow-step-parallel.test.ts @@ -261,6 +261,8 @@ describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => { stepReview: WorkflowLegacySeams["stepReview"]; onReworkReset: (a: ForeachActiveContext) => void; template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] }; + signal: AbortSignal; + logTaskEntry: (summary: string, detail?: string) => void; }> = {}, ) { const startOrder: number[] = []; @@ -279,6 +281,8 @@ describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => { ...backend.deps, ...(overrides.semaphoreAvailability ? { semaphoreAvailability: overrides.semaphoreAvailability } : {}), ...(overrides.onReworkReset ? { onReworkReset: overrides.onReworkReset as never } : {}), + ...(overrides.signal ? { signal: overrides.signal } : {}), + ...(overrides.logTaskEntry ? { logTaskEntry: overrides.logTaskEntry } : {}), }); const ir = foreachIr(overrides.template ?? singleExecuteTemplate(), config); const result = await executor.run(task, settingsOn(), ir); @@ -392,6 +396,25 @@ describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => { expect(step1Allocs[1].base).toBe("main@1"); }); + it("FIX 4: an integration conflict writes a task-level log entry naming the conflicted files", async () => { + const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }]); + const backend = makeFakeBackend({ conflictOnceSteps: new Set([1]) }); + const logged: Array<{ summary: string; detail?: string }> = []; + const { result } = await runScenario( + task, + { mode: "parallel", isolation: "worktree", concurrency: 2 }, + backend, + { logTaskEntry: (summary, detail) => logged.push({ summary, detail }) }, + ); + expect(result.outcome).toBe("success"); + const conflictLog = logged.find((l) => l.summary.includes("integration conflict on step 1")); + expect(conflictLog).toBeDefined(); + expect(conflictLog!.summary).toContain("reworking on updated base"); + // The fake backend reports `step-1.ts` as the conflicted file. + expect(conflictLog!.summary).toContain("step-1.ts"); + expect(conflictLog!.detail).toContain("step-1.ts"); + }); + it("conflict rework exhaustion routes rework-exhausted", async () => { const task = taskWithSteps([{ dependsOn: [] }]); const backend = makeFakeBackend({ conflictSteps: new Set([0]) }); // always conflicts. @@ -408,6 +431,59 @@ describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => { expect(backend.doneSteps).toEqual([]); }); + it("FIX 2: a failed instance releases its allocated worktree exactly once", async () => { + const task = taskWithSteps([{ dependsOn: [] }]); + const backend = makeFakeBackend(); + const { result } = await runScenario( + task, + { mode: "parallel", isolation: "worktree", concurrency: 1 }, + backend, + { + // The instance allocates a worktree, then its step-execute FAILS — the + // scheduler never enqueues it for integration, so without explicit + // release its worktree+branch would leak. + stepExecute: async () => ({ outcome: "failure", value: "boom" }), + }, + ); + expect(result.outcome).toBe("failure"); + // The allocated branch was released exactly once (discard==release in the fake). + expect(backend.allocations.map((a) => a.branchName)).toEqual(["fusion/fn-par-step-0"]); + expect(backend.released).toEqual(["fusion/fn-par-step-0"]); + expect(backend.released.filter((b) => b === "fusion/fn-par-step-0").length).toBe(1); + // It never integrated. + expect(backend.doneSteps).toEqual([]); + }); + + it("FIX 2: abort mid-run releases every allocated instance worktree", async () => { + const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }]); + const backend = makeFakeBackend(); + const controller = new AbortController(); + let executed = 0; + const { result } = await runScenario( + task, + { mode: "parallel", isolation: "worktree", concurrency: 2 }, + backend, + { + // Both instances allocate worktrees and run; abort after the first batch's + // step-execute so the scheduler's top-of-loop abort check fires while their + // worktrees are still allocated. + stepExecute: async () => { + executed += 1; + if (executed >= 1) controller.abort(); + return { outcome: "success", value: "step-done" }; + }, + signal: controller.signal, + }, + ); + expect(result.outcome).toBe("failure"); + // Every allocated branch was released (no leak), each exactly once. + const allocated = backend.allocations.map((a) => a.branchName); + expect(allocated.length).toBeGreaterThan(0); + for (const b of allocated) { + expect(backend.released.filter((r) => r === b).length).toBe(1); + } + }); + it("integration order is step order even when completion order inverts", async () => { const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }, { dependsOn: [] }]); const backend = makeFakeBackend(); diff --git a/packages/engine/src/agent-tools.ts b/packages/engine/src/agent-tools.ts index fca5fde9a6..602d5ef505 100644 --- a/packages/engine/src/agent-tools.ts +++ b/packages/engine/src/agent-tools.ts @@ -12,7 +12,7 @@ import { existsSync } from "node:fs"; import { createHash } from "node:crypto"; import { join, relative, resolve } from "node:path"; import type { AgentStore, AgentState, AgentCapability, AgentUpdateInput, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, TaskCreateInput, ReflectionStore, ApprovalRequestStore, ProjectSettings, ChatStore } from "@fusion/core"; -import { listTraits } from "@fusion/core"; +import { listTraits, isBuiltinWorkflowId } from "@fusion/core"; import { promoteHeldTask } from "./hold-release.js"; import { DASHBOARD_USER_ID, canAgentTakeImplementationTaskForExplicitRouting, dailyMemoryPath, ensureOpenClawMemoryFiles, extractAgentProvisioningRequest, formatRoleMismatchReason, getMemoryBackendCapabilities, getProjectMemory, isEphemeralAgent, memoryLongTermPath, normalizeMessageParticipant, reconcileDeterministicDuplicate, resolveAgentProvisioningPolicy, resolveMemoryBackend, resolveResearchSettings, resolveTaskGithubTracking, resolveTitleSummarizerSettingsModel, runDeterministicDuplicateGuard, scheduleQmdProjectMemoryRefresh, searchProjectMemory, shouldSkipBackgroundQmdRefresh, summarizeTitle } from "@fusion/core"; import { ResearchOrchestrator } from "./research-orchestrator.js"; @@ -65,6 +65,14 @@ export const taskDocumentReadParams = Type.Object({ export const workflowListParams = Type.Object({}); +export const workflowGetParams = Type.Object({ + workflow_id: Type.String({ + description: + "The workflow definition ID to fetch (e.g. 'WF-003', or a 'builtin:*' id). " + + "Use fn_workflow_list to discover available IDs.", + }), +}); + export const workflowSelectParams = Type.Object({ workflow_id: Type.String({ description: @@ -1031,6 +1039,64 @@ export function createWorkflowListTool(store: TaskStore): ToolDefinition { }; } +/** + * Create a `fn_workflow_get` tool that returns a single workflow definition by + * id — id/name/description, whether it is a read-only built-in, and the full + * resolved IR (nodes/edges/columns/artifacts/fields) as JSON. Agent-native + * read parity with the dashboard's workflow inspector; the companion read tool + * to fn_workflow_list. Read-only; an unknown id is reported as a tool error. + */ +export function createWorkflowGetTool(store: TaskStore): ToolDefinition { + return { + name: "fn_workflow_get", + label: "Get Workflow", + description: + "Fetch a single workflow definition by its ID — its name, description, whether it is a " + + "read-only built-in, and its full IR (nodes, edges, columns, artifacts, and custom fields) " + + "as JSON. Use fn_workflow_list to discover IDs first.", + parameters: workflowGetParams, + execute: async (_id: string, params: Static) => { + const workflowId = params.workflow_id?.trim(); + if (!workflowId) { + return { + content: [{ type: "text" as const, text: "ERROR: workflow_id is required." }], + details: {}, + isError: true, + }; + } + try { + const def = await store.getWorkflowDefinition(workflowId); + if (!def) { + return { + content: [{ type: "text" as const, text: `ERROR: Unknown workflow id '${workflowId}'. Use fn_workflow_list to discover valid IDs.` }], + details: {}, + isError: true, + }; + } + const builtin = isBuiltinWorkflowId(def.id); + const payload = { + id: def.id, + name: def.name, + description: def.description, + builtin, + ir: def.ir, + }; + return { + content: [{ type: "text" as const, text: JSON.stringify(payload, null, 2) }], + details: { workflowId: def.id, builtin }, + }; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + } catch (err: any) { + return { + content: [{ type: "text" as const, text: `ERROR: Failed to get workflow: ${err?.message ?? err}` }], + details: {}, + isError: true, + }; + } + }, + }; +} + /** * Create a `fn_workflow_select` tool that assigns a workflow definition to a * task (defaulting to the current task). Mirrors the dashboard's per-task diff --git a/packages/engine/src/code-node-runner.ts b/packages/engine/src/code-node-runner.ts index 760fb7a667..ad13ceeaee 100644 --- a/packages/engine/src/code-node-runner.ts +++ b/packages/engine/src/code-node-runner.ts @@ -334,6 +334,11 @@ function defaultSpawnRunner(params: { { cwd: params.cwd, timeout: params.timeoutMs, + // Ensure the timeout actually KILLS a child that traps/ignores SIGTERM: + // execFile defaults to SIGTERM, which a long-running or signal-trapping + // script can swallow, letting it outlive the timeout. SIGKILL cannot be + // trapped, so the timeout is enforceable. + killSignal: "SIGKILL", // Minimal env: PATH + a few harmless basics; no inherited secrets beyond // what the worktree-scoped script tier already has access to (KTD-15: // same trust as existing script steps). @@ -390,13 +395,10 @@ export async function validateCodeNodeSources( error: err instanceof CodeNodeError ? err.message : String(err), }); } - // Recurse into foreach templates (code nodes are legal inside them, KTD-15). - const template = (node.config as { template?: { nodes?: WorkflowIrNode[] } } | undefined)?.template; - if (template?.nodes) { - failures.push(...(await validateCodeNodeSources({ nodes: template.nodes }))); - } + // (A `code` node has no `template` — the only template recursion is the + // foreach pass below; the prior code-node-loop recursion here was dead.) } - // Also recurse into any foreach templates at the top level. + // Recurse into any foreach templates (code nodes are legal inside them, KTD-15). for (const node of ir.nodes) { if (node.kind !== "foreach") continue; const template = (node.config as { template?: { nodes?: WorkflowIrNode[] } } | undefined)?.template; diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 9165f19e2e..628c3ad123 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -10,7 +10,7 @@ import { existsSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core"; import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask } from "@fusion/core"; -import type { TaskStep, WorkflowIr } from "@fusion/core"; +import type { TaskStep, WorkflowIr, WorkflowFieldDefinition } from "@fusion/core"; import { buildWorkflowObservationFromTask, buildWorkflowObservation, @@ -157,6 +157,7 @@ import { createTaskDocumentWriteTool as sharedCreateTaskDocumentWriteTool, createTaskLogTool as sharedCreateTaskLogTool, createWorkflowListTool as sharedCreateWorkflowListTool, + createWorkflowGetTool as sharedCreateWorkflowGetTool, createWorkflowSelectTool as sharedCreateWorkflowSelectTool, createTaskPromoteTool as sharedCreateTaskPromoteTool, createWorkflowCreateTool as sharedCreateWorkflowCreateTool, @@ -3243,6 +3244,14 @@ export class TaskExecutor { * Keyed by task id; cleared alongside the pin. */ private graphStepRunOnce = new Map>(); + /** Step-inversion (KTD-4): the foreach instance the step-execute seam is + * currently driving for a graph-owned task, so `runGraphTaskStep` can honor + * `deferDoneToReview` when deciding whether a non-terminal step is a success + * (review will author done) or a failure (implementation left it incomplete). + * Stamped by the stepExecute seam around the runTaskStep call; cleared with the + * per-run pins. */ + private graphStepActiveContext = new Map(); + /** Tasks currently being orchestrated by the graph runner. Process-wide for * the same reason as executingTaskLock (FN-4811): duplicate execute() * invocations can arrive from different TaskExecutor instances in one @@ -3288,8 +3297,25 @@ export class TaskExecutor { } if (!selection) return false; + // Resolve the production run id ONCE, here, so it is the single source of + // truth shared by the runner AND the executor-side persistence deps + // (parse-steps pin probe, foreach instance-row flips, resume reconcile). The + // runner derives `${task.id}:${definition.id}`; we mirror that derivation + // from the resolved definition and thread it everywhere. Best-effort: if the + // definition cannot be resolved (older store), the runner falls back to its + // own derivation and the deps fall back to the legacy `:run` literal — the + // prior behavior — so this never strands a task. + let resolvedRunId: string | undefined; + try { + const definition = await this.store.getWorkflowDefinition?.(selection.workflowId); + if (definition) resolvedRunId = `${task.id}:${definition.id}`; + } catch { + // Definition load failure — leave undefined; deps/runner use fallbacks. + } + const runner = new WorkflowGraphTaskRunner({ store: this.store, + runId: resolvedRunId, seams: this.createGraphSeams(settings), runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings), onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), @@ -3309,7 +3335,7 @@ export class TaskExecutor { // Step-inversion (KTD-12, U12): parse-steps node handler deps — artifact // read (through task-documents with PROMPT.md fallback), step-list write // (graph-source projection), pin-protection probe, and audit. - parseStepsDeps: this.buildParseStepsDeps(), + parseStepsDeps: this.buildParseStepsDeps(resolvedRunId), // Step-inversion (KTD-15, U14): code node runner — esbuild compile + // child-process execution with the harness contract. runCode: this.buildCodeNodeRunner(), @@ -3317,7 +3343,15 @@ export class TaskExecutor { // parallel scheduling. Per-instance worktrees branched off the task's main // branch tip; integration rebases each branch in step order; the projection // flips done-iff-integrated. Shared isolation never invokes these. - ...this.buildForeachWorktreeDeps(task), + ...this.buildForeachWorktreeDeps(task, resolvedRunId), + // FIX 4 (context gap): task-level log sink so an integration-conflict + // rework writes a visible "reworking on updated base (files: ...)" entry + // the re-running agent can read. Best-effort; logging failures swallowed. + logTaskEntry: (summary: string, detail?: string) => { + void this.store + .logEntry(task.id, summary, detail, this.getRunContextFor(task.id)) + .catch(() => {}); + }, }); let result: WorkflowGraphTaskRunResult; try { @@ -3344,6 +3378,7 @@ export class TaskExecutor { // Clear per-run step-inversion pins (KTD-8: pinned only for the run's life). this.graphStepSessionPinned.delete(task.id); this.graphStepRunOnce.delete(task.id); + this.graphStepActiveContext.delete(task.id); } } @@ -3407,6 +3442,23 @@ export class TaskExecutor { return undefined; } + /** + * Resolve the custom field definitions declared by a task's selected workflow + * (KTD-13) so the executor prompt can surface the schema and current values to + * the agent. Pure read; degrades to undefined on any resolution failure (no + * selection, missing/corrupt definition, older store) so prompt-building never + * throws and legacy tasks see no custom-fields section. + */ + private async resolveTaskCustomFieldDefs(taskId: string): Promise { + try { + const ir = await resolveWorkflowIrForTask(this.store, taskId); + const fields = ir.version === "v2" ? ir.fields : undefined; + return fields && fields.length > 0 ? fields : undefined; + } catch { + return undefined; + } + } + /** * Build the parse-steps node handler deps (KTD-12, U12): artifact read through * the task-documents machinery (PROMPT.md falls back to the task's own PROMPT @@ -3414,28 +3466,34 @@ export class TaskExecutor { * projection (`updateTask({ steps })`), pin-protection probe (persisted instance * rows exist → re-parse illegal, KTD-3), and a logEntry-backed audit sink. */ - private buildParseStepsDeps(): ParseStepsHandlerDeps { + /** + * Read a task artifact by key through the task-documents layer, falling back to + * the task's own PROMPT content for the default `PROMPT.md` step-source artifact + * (the same source the legacy step-init reads). Shared by the parse-steps and + * code-node deps (FIX 7: one source of truth for the fallback). + */ + private async readTaskArtifact(taskId: string, key: string): Promise { + // Declared artifacts ride the task-documents layer. + try { + const doc = await this.store.getTaskDocument(taskId, key); + if (doc) return doc.content; + } catch { + // Fall through to the PROMPT fallback below. + } + if (key === "PROMPT.md") { + try { + const detail = await this.store.getTask(taskId); + if (typeof detail.prompt === "string") return detail.prompt; + } catch { + // No PROMPT available. + } + } + return undefined; + } + + private buildParseStepsDeps(runId?: string): ParseStepsHandlerDeps { return { - readArtifact: async (task, key): Promise => { - // Declared artifacts ride the task-documents layer. - try { - const doc = await this.store.getTaskDocument(task.id, key); - if (doc) return doc.content; - } catch { - // Fall through to the PROMPT fallback below. - } - // Default step-source artifact (PROMPT.md): fall back to the task's PROMPT - // content (the same source the legacy step-init reads). - if (key === "PROMPT.md") { - try { - const detail = await this.store.getTask(task.id); - if (typeof detail.prompt === "string") return detail.prompt; - } catch { - // No PROMPT available. - } - } - return undefined; - }, + readArtifact: (task, key): Promise => this.readTaskArtifact(task.id, key), writeSteps: async (task, steps: TaskStep[]): Promise => { await this.store.updateTask(task.id, { steps }); }, @@ -3445,9 +3503,12 @@ export class TaskExecutor { }; if (typeof store.loadWorkflowRunStepInstances !== "function") return false; try { - // Any persisted instance row for this task (any run) means a foreach has - // expanded — re-parsing would desynchronize the pinned instance set. - const rows = store.loadWorkflowRunStepInstances(task.id, `${task.id}:run`); + // Any persisted instance row for THIS run means a foreach has expanded — + // re-parsing would desynchronize the pinned instance set (KTD-3). Probe + // under the REAL run id (threaded from maybeExecuteWorkflowGraph) so the + // pin protection actually fires; fall back to the legacy literal only when + // the run id was not threaded (older store / no definition). + const rows = store.loadWorkflowRunStepInstances(task.id, runId ?? `${task.id}:run`); return Array.isArray(rows) && rows.length > 0; } catch { return false; @@ -3484,14 +3545,11 @@ export class TaskExecutor { } catch { // No documents — pass an empty artifact map. } - // Surface PROMPT.md from the task prompt when not already a document. + // Surface PROMPT.md from the task prompt when not already a document + // (shared artifact-read fallback — FIX 7). if (out["PROMPT.md"] === undefined) { - try { - const detail = await this.store.getTask(task.id); - if (typeof detail.prompt === "string") out["PROMPT.md"] = detail.prompt; - } catch { - // No prompt available. - } + const prompt = await this.readTaskArtifact(task.id, "PROMPT.md"); + if (typeof prompt === "string") out["PROMPT.md"] = prompt; } return out; }, @@ -3537,7 +3595,7 @@ export class TaskExecutor { * Best-effort throughout: a git failure routes the foreach to a clean failure * (parked for human review) rather than crashing the run. */ - private buildForeachWorktreeDeps(task: Task): { + private buildForeachWorktreeDeps(task: Task, runId?: string): { allocateInstanceWorktree: ( stepIndex: number, base: string | undefined, @@ -3651,22 +3709,39 @@ export class TaskExecutor { // dependency order; predecessors are integrated (done) by construction. await this.store.updateStep(taskId, stepIndex, "done", { source: "graph" }); }, - markInstanceIntegrated: async (stepIndex, integratedAt): Promise => { + markInstanceIntegrated: async (stepIndex, integratedAt, identity): Promise => { const store = this.store as unknown as { saveWorkflowRunStepInstance?: (state: WorkflowStepInstanceState) => void; + loadWorkflowRunStepInstances?: (taskId: string, runId: string) => WorkflowStepInstanceState[]; }; if (typeof store.saveWorkflowRunStepInstance !== "function") return; + // The upsert is keyed by (taskId, runId, foreachNodeId, stepIndex). The + // queue passes the REAL identity (the same runId + foreachNodeId the + // foreach sub-walk persisted the row under) so this FLIPS the existing + // row to completed/integratedAt instead of writing an orphan (FIX 1). + // Load the current row to preserve its fields (currentNodeId, baseline, + // reworkCount) we don't otherwise carry on the identity. + let existing: WorkflowStepInstanceState | undefined; + try { + const rows = store.loadWorkflowRunStepInstances?.(taskId, identity.runId) ?? []; + existing = rows.find( + (r) => r.foreachNodeId === identity.foreachNodeId && r.stepIndex === stepIndex, + ); + } catch { + // Best-effort read; fall back to a minimal flip below. + } try { store.saveWorkflowRunStepInstance({ + ...(existing ?? {}), taskId, - runId: `${taskId}:run`, - foreachNodeId: "", + runId: identity.runId, + foreachNodeId: identity.foreachNodeId, stepIndex, - pinnedStepCount: 0, - currentNodeId: "", + pinnedStepCount: identity.pinnedStepCount, + currentNodeId: existing?.currentNodeId ?? "", status: "completed", - reworkCount: 0, - branchName: canonicalStepInstanceBranchName(taskId, stepIndex), + reworkCount: existing?.reworkCount ?? 0, + branchName: identity.branchName || canonicalStepInstanceBranchName(taskId, stepIndex), integratedAt, } as WorkflowStepInstanceState); } catch { @@ -3690,7 +3765,9 @@ export class TaskExecutor { if (typeof store.loadWorkflowRunStepInstances !== "function") return []; let rows: WorkflowStepInstanceState[] = []; try { - rows = store.loadWorkflowRunStepInstances(taskId, `${taskId}:run`) ?? []; + // Load under the REAL run id (threaded) so resume actually sees the rows + // the sub-walk persisted; the legacy literal is the unthreaded fallback. + rows = store.loadWorkflowRunStepInstances(taskId, runId ?? `${taskId}:run`) ?? []; } catch { return []; } @@ -3942,6 +4019,14 @@ export class TaskExecutor { // Pin step-session physics for the run before the implementation pass. this.graphStepSessionPinned.add(task.id); + // Single-flight per attempt (KTD-2/KTD-8): the implementation phase runs once + // per run, memoized by task id, so each foreach instance's `runStep` observes + // the projection rather than re-running the agent. A REJECTED phase must NOT + // poison later attempts: a rework cycle re-enters `runStep` and would otherwise + // re-await the same stored rejection forever, so the implementation is never + // retried. On rejection we therefore clear the memo entry so the NEXT call + // (the rework re-run) re-invokes the implementation phase. Concurrent + // in-flight callers within a single attempt still share the one promise. let phase = this.graphStepRunOnce.get(task.id); if (!phase) { phase = this.runImplementationPhase(task); @@ -3950,27 +4035,46 @@ export class TaskExecutor { try { await phase; } catch (err) { + // Clear the poisoned memo so a rework cycle can retry the implementation + // (only if it is still the same rejected promise — do not clobber a fresh + // attempt another caller may have already installed). + if (this.graphStepRunOnce.get(task.id) === phase) { + this.graphStepRunOnce.delete(task.id); + } return { success: false, error: err instanceof Error ? err.message : String(err) }; } // Consult the projection (the single source of truth, KTD-7) for this step's // terminal state. The step-session pass marks each step done/skipped as it - // completes; a step-review node (when present) decides done-ness instead, so - // here we treat a completed step-session pass as success for this step and let - // the review gate the projection write. + // completes; a step-review node (when present) decides done-ness instead. try { const live = await this.store.getTask(task.id); + const active = this.foreachActiveForTask(task.id); const status = live.steps[stepIndex]?.status; if (status === "done" || status === "skipped") return { success: true }; - // Step-session pass completed but this step is not yet terminal — when a - // review will mark it done (deferDoneToReview) the pass having run is the - // success signal; otherwise the implementation left it incomplete. - return { success: true }; + // Step not terminal after the pass: when a review will author done-ness + // (deferDoneToReview), the pass having RUN is the success signal — the review + // gates the projection write. Otherwise the implementation pass failed to + // complete this step, so report failure rather than masking it (FIX 3: the + // prior code returned success on both branches, hiding step-session failures). + if (active?.deferDoneToReview === true) return { success: true }; + return { + success: false, + error: `step ${stepIndex} not completed by implementation pass (status: ${status ?? "unknown"})`, + }; } catch (err) { return { success: false, error: err instanceof Error ? err.message : String(err) }; } } + /** Read the active foreach instance context for a graph-owned task (if any) so + * the step driver can honor `deferDoneToReview`. The active context is threaded + * through the foreach sub-walk; we surface it via a per-task slot the + * step-execute seam stamps. Returns undefined outside a foreach instance. */ + private foreachActiveForTask(taskId: string): ForeachActiveContext | undefined { + return this.graphStepActiveContext.get(taskId); + } + /** Seam implementations delegating to the legacy engine (KTD-1: delegate, never reimplement). */ private createGraphSeams(_settings: Settings): WorkflowLegacySeams { return { @@ -4054,6 +4158,9 @@ export class TaskExecutor { // worktree (shared isolation — unchanged). The file-scope guard the session // machinery installs applies to either worktree unchanged (not bypassed). const worktreePath = active.worktreePath || live.worktree || this.rootDir; + // Stamp the active instance so `runGraphTaskStep` can honor + // `deferDoneToReview` when judging a non-terminal step (FIX 3). + this.graphStepActiveContext.set(seamTask.id, active); const result = await runTaskStep( { store: this.store, @@ -5518,6 +5625,7 @@ export class TaskExecutor { this.createTaskDocumentWriteTool(task.id), this.createTaskDocumentReadTool(task.id), this.createWorkflowListTool(), + this.createWorkflowGetTool(), this.createWorkflowSelectTool(task.id), this.createTaskPromoteTool(task.id), this.createWorkflowCreateTool(), @@ -5776,12 +5884,14 @@ export class TaskExecutor { "Review the current state of your worktree and proceed with the next pending step.", ].join("\n")); } else { + const customFieldDefs = await this.resolveTaskCustomFieldDefs(task.id); const agentPrompt = buildExecutionPrompt( detail, this.rootDir, settings, worktreePath, this.options.pluginRunner, + customFieldDefs, ); await promptWithFallback(session, agentPrompt); } @@ -6139,6 +6249,7 @@ export class TaskExecutor { }, worktreePath); stuckDetector?.trackTask(task.id, retrySession); + const retryCustomFieldDefs = await this.resolveTaskCustomFieldDefs(task.id); let retryPrompt: string; if (pseudoPause.kind !== "none") { const shortMatch = (pseudoPause.matched ?? "").slice(0, 120); @@ -6158,7 +6269,7 @@ export class TaskExecutor { "Do NOT ask for permission. Do NOT write a summary. Just call a tool and keep working.", "", "Original task:", - buildExecutionPrompt(detail, this.rootDir, settings, worktreePath, this.options.pluginRunner), + buildExecutionPrompt(detail, this.rootDir, settings, worktreePath, this.options.pluginRunner, retryCustomFieldDefs), ].join("\n"); } else { retryPrompt = [ @@ -6168,7 +6279,7 @@ export class TaskExecutor { "2. If there is remaining work, finish it and then call fn_task_done.", "", "Original task:", - buildExecutionPrompt(detail, this.rootDir, settings, worktreePath, this.options.pluginRunner), + buildExecutionPrompt(detail, this.rootDir, settings, worktreePath, this.options.pluginRunner, retryCustomFieldDefs), ].join("\n"); } @@ -7103,6 +7214,23 @@ export class TaskExecutor { execute: async (_id: string, params: Static) => { const { step, status, dependencies, custom_fields } = params; + // Bare-call guard (P1 api-contract): a call with none of + // step/status/dependencies/custom_fields silently no-op'd, which the + // agent cannot observe. Reject it up front so the failure is visible and + // self-describing. The legacy no-op text is preserved as the detail. + if (step === undefined && status === undefined && dependencies === undefined && custom_fields === undefined) { + return { + content: [{ + type: "text" as const, + text: "ERROR: fn_task_update requires at least one of: step+status (report step progress), " + + "dependencies (array of task ids), or custom_fields (workflow-defined field patch). " + + "No-op: provide a step+status, dependencies, or custom_fields to update.", + }], + details: {}, + isError: true, + }; + } + // Custom-field patch (KTD-13): routed through the store's single write // authority, which validates each value against the task's workflow field // schema. A typed rejection surfaces the offending field id + reason as a @@ -7112,10 +7240,28 @@ export class TaskExecutor { const res = await store.updateTaskCustomFields(taskId, custom_fields); if (!res.ok) { const r = res.rejection; + // Self-correcting rejection text: append the valid field ids (and, + // for an enum violation, the valid values for the offending field) + // resolved from the task's workflow field schema so a failed write + // carries everything the agent needs to retry. Best-effort: a + // resolution failure just omits the hint (the base reason still ships). + let hint = ""; + try { + const defs = await this.resolveTaskCustomFieldDefs(taskId); + if (defs && defs.length > 0) { + if (r.code === "unknown-field" || r.code === "no-fields-defined") { + hint = ` Valid field ids: ${defs.map((f) => f.id).join(", ")}.`; + } else if (r.code === "enum-violation") { + const field = defs.find((f) => f.id === r.fieldId); + const opts = field?.options?.map((o) => o.value) ?? []; + if (opts.length > 0) hint = ` Valid values for '${r.fieldId}': ${opts.join(", ")}.`; + } + } + } catch { /* hint is best-effort */ } return { content: [{ type: "text" as const, - text: `ERROR: custom field '${r.fieldId}' rejected (${r.code}): ${r.detail}`, + text: `ERROR: custom field '${r.fieldId}' rejected (${r.code}): ${r.detail}${hint}`, }], details: { fieldId: r.fieldId, code: r.code, detail: r.detail }, isError: true, @@ -7351,6 +7497,10 @@ export class TaskExecutor { return sharedCreateWorkflowListTool(this.store); } + private createWorkflowGetTool(): ToolDefinition { + return sharedCreateWorkflowGetTool(this.store); + } + private createWorkflowSelectTool(taskId: string): ToolDefinition { return sharedCreateWorkflowSelectTool(this.store, taskId); } @@ -12420,6 +12570,7 @@ export function buildExecutionPrompt( settings?: Settings, worktreePath?: string, pluginRunner?: PluginRunner, + customFieldDefs?: WorkflowFieldDefinition[], ): string { const prompt = scopePromptToWorktree(task.prompt, rootDir, worktreePath); const reviewLevel = parseReviewLevelFromPrompt(prompt); @@ -12518,6 +12669,35 @@ git log --oneline steeringSection = lines.join("\n"); } + // Build custom fields section (KTD-13): when the task's workflow declares + // custom fields, the executor agent can write them via fn_task_update + // (custom_fields) — but without the schema it is writing blind. List each + // field's id/name/type, enum options, required flag, and current value so + // the write is informed and self-correcting. Compact: one line per field. + let customFieldsSection = ""; + if (customFieldDefs && customFieldDefs.length > 0) { + const current = task.customFields ?? {}; + const lines = [ + "", + "## Custom fields", + "", + "This task's workflow declares custom fields. Set them with `fn_task_update(custom_fields={...})` keyed by field id (pass null to clear).", + "", + ]; + for (const f of customFieldDefs) { + const parts = [`- \`${f.id}\` (${f.name}) — type: ${f.type}`]; + if ((f.type === "enum" || f.type === "multi-enum") && f.options && f.options.length > 0) { + const opts = f.options.map((o) => (o.label && o.label !== o.value ? `${o.value} (${o.label})` : o.value)).join(", "); + parts.push(`options: [${opts}]`); + } + if (f.required) parts.push("required"); + const hasValue = Object.prototype.hasOwnProperty.call(current, f.id) && current[f.id] !== null && current[f.id] !== undefined; + parts.push(`current: ${hasValue ? JSON.stringify(current[f.id]) : "unset"}`); + lines.push(parts.join("; ")); + } + customFieldsSection = lines.join("\n") + "\n"; + } + const taskPromptContributions = pluginRunner?.getPromptContributionsForSurface("executor-task") ?? []; if (taskPromptContributions.length > 0) { executorLog.log(`${task.id}: applied ${taskPromptContributions.length} plugin prompt contributions for executor-task surface`); @@ -12533,7 +12713,7 @@ ${task.dependencies.length > 0 ? `Dependencies: ${task.dependencies.join(", ")}` ## PROMPT.md ${prompt} -${attachmentsSection}${commandsSection}${memorySection}${progressSection}${steeringSection} +${attachmentsSection}${commandsSection}${memorySection}${progressSection}${steeringSection}${customFieldsSection} ## Review level: ${reviewLevel} ${reviewLevel === 0 ? "No reviews required. Implement directly." : ""} diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index fa5b8f9ab6..dec5919a47 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -8,6 +8,7 @@ export { createSendMessageTool, createReadMessagesTool, createWorkflowListTool, + createWorkflowGetTool, createWorkflowSelectTool, taskCreateParams, taskDocumentReadParams, diff --git a/packages/engine/src/step-integration.ts b/packages/engine/src/step-integration.ts index 61b1f9983a..81e05e922f 100644 --- a/packages/engine/src/step-integration.ts +++ b/packages/engine/src/step-integration.ts @@ -68,6 +68,21 @@ export interface IntegrationGitOps { discardBranch(branchName: string, stepIndex: number): Promise; } +/** + * Identity of the persisted instance row to flip on integration. The queue + * sources this from the foreach environment (the SINGLE source of truth for + * runId/foreachNodeId/pinnedStepCount — the same values the sub-walk persisted + * the row under) so `markInstanceIntegrated` updates the EXISTING row keyed by + * `(taskId, runId, foreachNodeId, stepIndex)` instead of writing an orphan. + */ +export interface IntegrationInstanceIdentity { + runId: string; + foreachNodeId: string; + pinnedStepCount: number; + /** The instance branch being integrated (carried onto the flipped row). */ + branchName: string; +} + /** Projection + persistence side-effects the queue performs on a successful * integration (KTD-7 projection-first ordering). Injected so the queue stays * engine-agnostic and unit-testable. */ @@ -80,9 +95,16 @@ export interface IntegrationProjection { markStepDone(stepIndex: number): Promise; /** * Mark the instance row `completed` with `integratedAt` AFTER the projection - * flip (projection-first ordering). Optional — a fully in-memory run needs none. + * flip (projection-first ordering). The queue passes the row's REAL identity + * (runId/foreachNodeId/pinnedStepCount/branchName) so the production impl flips + * the SAME row the sub-walk persisted — never an orphan. Optional — a fully + * in-memory run needs none. */ - markInstanceIntegrated?(stepIndex: number, integratedAt: string): Promise | void; + markInstanceIntegrated?( + stepIndex: number, + integratedAt: string, + identity: IntegrationInstanceIdentity, + ): Promise | void; } /** One enqueued, completed instance awaiting ordered integration. */ @@ -126,6 +148,11 @@ export class IntegrationQueue { private readonly gitOps: IntegrationGitOps, private readonly projection: IntegrationProjection, private readonly pinnedStepCount: number, + /** Identity context for instance-row flips on integration (KTD-6/KTD-11): + * the REAL runId + foreachNodeId the sub-walk persisted rows under, so + * `markInstanceIntegrated` updates the existing row, not an orphan. Optional + * for fully in-memory runs that pass no `markInstanceIntegrated`. */ + private readonly rowIdentity?: { runId: string; foreachNodeId: string }, ) {} /** Enqueue a completed instance awaiting integration. Idempotent per step. */ @@ -181,7 +208,12 @@ export class IntegrationQueue { // here the worktree is released after a clean integration via discardBranch // which the production op treats as "release, branch already merged"). await this.projection.markStepDone(ready.stepIndex); - await this.projection.markInstanceIntegrated?.(ready.stepIndex, result.integratedAt); + await this.projection.markInstanceIntegrated?.(ready.stepIndex, result.integratedAt, { + runId: this.rowIdentity?.runId ?? "", + foreachNodeId: this.rowIdentity?.foreachNodeId ?? "", + pinnedStepCount: this.pinnedStepCount, + branchName: ready.branchName, + }); // Release the instance worktree post-integration (pool hygiene). The branch // is already on the base; discardBranch in production only releases here. await this.safeDiscard(ready.branchName, ready.stepIndex); diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 2f4d7c27d9..1649dfa917 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -116,6 +116,8 @@ export interface WorkflowGraphExecutorDeps { semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"]; /** Step-inversion (KTD-11, U10): crash-resume reconciliation hook. */ resumeReconcile?: ForeachEnvironment["resumeReconcile"]; + /** FIX 4 (context gap): task-level log sink for integration-conflict rework. */ + logTaskEntry?: ForeachEnvironment["logTaskEntry"]; } export interface WorkflowGraphExecutorResult { @@ -192,6 +194,11 @@ export class WorkflowGraphExecutor { // load so this run's own (taskId, runId) rows survive while every stale run // is removed. Never throws into the run. await this.pruneStaleBranches(task.id, runId); + // Same posture for foreach step-instance rows (KTD-6, U4): prune every stale + // run's instance rows, keeping only this run's, so the table does not + // accumulate historical runs for a long-lived task. The resume reconcile path + // (foreach worktree scheduler) loads THIS run's rows, which survive. + await this.pruneStaleInstances(task.id, runId); // Shared branch environment: built lazily so the sequential path pays nothing. const branchEnv = (): BranchEnvironment => ({ @@ -280,6 +287,7 @@ export class WorkflowGraphExecutor { integrationProjection: this.deps.integrationProjection, semaphoreAvailability: this.deps.semaphoreAvailability, resumeReconcile: this.deps.resumeReconcile, + logTaskEntry: this.deps.logTaskEntry, }); visitedNodeIds.push(...foreachResult.visitedNodeIds); const result: WorkflowNodeResult = { @@ -334,6 +342,7 @@ export class WorkflowGraphExecutor { // Prune again on run completion (#1412): keeps only this run's rows so the // table does not accumulate historical runs for a long-lived task. await this.pruneStaleBranches(task.id, runId); + await this.pruneStaleInstances(task.id, runId); return { executed: true, outcome: terminal.outcome, @@ -363,6 +372,16 @@ export class WorkflowGraphExecutor { } } + /** Best-effort prune of stale-run foreach instance rows (KTD-6, U4); identical + * keepRunId posture as {@link pruneStaleBranches}. Never throws into the run. */ + private async pruneStaleInstances(taskId: string, keepRunId: string): Promise { + try { + await this.deps.stepInstancePersistence?.clearStaleInstanceStates?.(taskId, keepRunId); + } catch { + // Pruning is additive bookkeeping — 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-foreach.ts b/packages/engine/src/workflow-graph-foreach.ts index d391aba4f7..9c0ee826c0 100644 --- a/packages/engine/src/workflow-graph-foreach.ts +++ b/packages/engine/src/workflow-graph-foreach.ts @@ -4,6 +4,7 @@ import { WorkflowIrError } from "@fusion/core"; import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js"; import { FOREACH_ACTIVE_CONTEXT_KEY, + INTEGRATION_CONFLICT_CONTEXT_KEY, type ForeachActiveContext, } from "./workflow-node-handlers.js"; import { @@ -224,6 +225,15 @@ export interface ForeachEnvironment { ) => | Promise> | Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>; + /** + * Optional task-level log sink (FIX 4 — context gap). When an integration + * conflict routes a step instance to rework, the re-running agent has no record + * of WHY its base changed. Wiring this to the engine's task log path + * (`store.logEntry`) writes a visible entry so the rework is explicable. Best + * effort: a logging failure must never affect the run. Absent under tests / a + * foreach env without a logging dep (the conflict still logs to schedulerLog). + */ + logTaskEntry?: (summary: string, detail?: string) => void | Promise; } export interface ForeachRunResult { @@ -489,6 +499,10 @@ async function runForeachWorktree( env.integrationGitOps, env.integrationProjection, pinnedStepCount, + // Single source of truth for the instance-row identity (KTD-6/KTD-11): the + // REAL runId + foreachNodeId the sub-walk persisted rows under, so integration + // flips the SAME row instead of writing an orphan. + { runId: env.runId, foreachNodeId: foreachNode.id }, ); // Crash-resume reconciliation (KTD-11): integrated → skip; branch-exists → re-enter @@ -520,6 +534,45 @@ async function runForeachWorktree( const depsIntegrated = (i: number): boolean => resolveDependsOn(env.steps, i).every((d) => isIntegrated(d)); + // Track instances whose worktree/branch has been released so a terminal sweep + // (and the integration queue) never double-discards. The integration queue + // releases integrated/conflicted branches itself; this set covers the branches + // we release directly (failed / aborted / stuck instances that allocated a + // worktree but the queue never resolved). discardBranch is best-effort either + // way, but the guard keeps the release "exactly once" contract honest (FIX 2). + const released = new Set(); + const releaseInstanceWorktree = async (inst: WorktreeInstance): Promise => { + if (!inst.branchName) return; // never allocated. + if (released.has(inst.stepIndex)) return; + released.add(inst.stepIndex); + try { + await env.integrationGitOps!.discardBranch(inst.branchName, inst.stepIndex); + } catch (err) { + schedulerLog.warn( + `foreach ${foreachNode.id} step ${inst.stepIndex}: worktree release failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } + }; + // Terminal cleanup: release every allocated instance the integration queue did + // NOT own. The queue itself releases `integrated` (post-integration discard) and + // `awaiting-integration` (via discardAllPending on the abort/fail/stuck paths) + // and conflicted branches (marked in `released`). This sweep covers the + // remainder — `failed` / `running` / `pending` instances that allocated a + // worktree — so failure / rework-exhausted / abort / scheduler-stuck never leak + // an allocated worktree+branch. The `released` guard keeps it exactly once. + const releaseUnresolvedAllocated = async (): Promise => { + await Promise.all( + instances + .filter( + (i) => + i.branchName && + i.state !== "integrated" && + i.state !== "awaiting-integration", + ) + .map((i) => releaseInstanceWorktree(i)), + ); + }; + // Run one instance's sub-walk in an isolated context + freshly-allocated // worktree off the CURRENT integration base. Returns awaiting-integration (with // the branch) on a clean sub-walk, or a failure outcome (rework-exhausted etc.). @@ -538,6 +591,10 @@ async function runForeachWorktree( } inst.branchName = allocation.branchName; inst.worktreePath = allocation.worktreePath; + // Fresh allocation: this instance now holds a live, un-released worktree again + // (a prior conflict/rework may have flagged it released). Clear the marker so a + // later failure of THIS run releases the new worktree exactly once. + released.delete(inst.stepIndex); return runWorktreeInstanceSubWalk(foreachNode, env, plan, pinnedStepCount, visitedNodeIds, inst); }; @@ -546,6 +603,7 @@ async function runForeachWorktree( for (;;) { if (env.signal?.aborted) { await queue.discardAllPending(); + await releaseUnresolvedAllocated(); return { outcome: "failure", value: "aborted", visitedNodeIds }; } @@ -575,8 +633,11 @@ async function runForeachWorktree( } else { inst.state = "failed"; // A failed instance is skipped in the ordered queue so the cursor can - // advance past it; the foreach reports the failure value below. + // advance past it; the foreach reports the failure value below. Release + // its allocated worktree+branch immediately (FIX 2 — the queue never + // sees a failed instance, so it would otherwise leak). queue.skip(inst.stepIndex); + await releaseInstanceWorktree(inst); (inst as WorktreeInstance & { failValue?: string }).failValue = result.value; } }), @@ -593,10 +654,14 @@ async function runForeachWorktree( if (outcome.status === "integrated") { inst.state = "integrated"; } else { - // integration-conflict: route as rework on the UPDATED base (KTD-11). The - // explicit `outcome:integration-conflict` edge (if authored) is honored by - // the sub-walk; the DEFAULT here is the rework path (re-execute on the - // updated base, budget-counted). Either way we re-run the instance. + // integration-conflict: the queue's drain() already discarded the + // conflicting branch (safeDiscard) before reporting the conflict, so the + // branch is released — record it so the terminal sweep never double-discards. + released.add(inst.stepIndex); + // Route as rework on the UPDATED base (KTD-11). The explicit + // `outcome:integration-conflict` edge (if authored) is honored by the + // sub-walk; the DEFAULT here is the rework path (re-execute on the updated + // base, budget-counted). Either way we re-run the instance. if (inst.reworkBudget <= 0) { inst.state = "failed"; queue.skip(inst.stepIndex); @@ -612,9 +677,22 @@ async function runForeachWorktree( // (the edge overrides the implicit "from entry" rework). (inst as WorktreeInstance & { lastIntegrationConflict?: boolean }).lastIntegrationConflict = plan.hasExplicitIntegrationConflictEdge; + const conflictedFiles = + outcome.status === "conflict" && Array.isArray(outcome.conflictedFiles) + ? outcome.conflictedFiles + : []; schedulerLog.log( `foreach ${foreachNode.id} step ${inst.stepIndex}: integration-conflict — reworking on updated base (budget left ${inst.reworkBudget})`, ); + // Task-level audit so the re-running agent sees WHY its base moved + // (FIX 4). Best-effort; a logging failure never affects the run. + try { + const filesNote = conflictedFiles.length > 0 ? ` (files: ${conflictedFiles.join(", ")})` : ""; + await env.logTaskEntry?.( + `integration conflict on step ${inst.stepIndex}: reworking on updated base${filesNote}`, + conflictedFiles.length > 0 ? `Conflicted files:\n${conflictedFiles.map((f) => `- ${f}`).join("\n")}` : undefined, + ); + } catch { /* log is best-effort */ } } } } @@ -623,6 +701,7 @@ async function runForeachWorktree( const failed = instances.find((i) => i.state === "failed"); if (failed) { await queue.discardAllPending(); + await releaseUnresolvedAllocated(); const value = (failed as WorktreeInstance & { failValue?: string }).failValue ?? "instance-failed"; return { outcome: "failure", value, visitedNodeIds }; } @@ -636,6 +715,7 @@ async function runForeachWorktree( const anyRunnable = instances.some((i) => i.state === "pending" && depsIntegrated(i.stepIndex)); if (!progressed && !anyActive && !anyRunnable) { await queue.discardAllPending(); + await releaseUnresolvedAllocated(); return { outcome: "failure", value: "scheduler-stuck", visitedNodeIds }; } } @@ -679,8 +759,14 @@ async function runWorktreeInstanceSubWalk( instanceContext[FOREACH_ACTIVE_CONTEXT_KEY] = active; // Surface a prior integration-conflict to the author's nodes when the template // authored an explicit `outcome:integration-conflict` edge (KTD-11 override). - const lastConflict = (inst as WorktreeInstance & { lastIntegrationConflict?: boolean }).lastIntegrationConflict; - if (lastConflict) instanceContext["integration:conflict"] = true; + // Consume the one-shot signal here: clear it on the instance so a LATER clean + // rework (e.g. a fresh rethink with no conflict) does not re-surface a stale + // conflict on the seeded context. + const conflictHolder = inst as WorktreeInstance & { lastIntegrationConflict?: boolean }; + if (conflictHolder.lastIntegrationConflict) { + instanceContext[INTEGRATION_CONFLICT_CONTEXT_KEY] = true; + conflictHolder.lastIntegrationConflict = false; + } const persist = (status: WorkflowStepInstanceState["status"], currentNodeId: string): Promise => persistInstanceState(env.persistence, { diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index e68fea9cf8..5cbaf23fb3 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -77,6 +77,16 @@ export interface WorkflowGraphTaskRunnerDeps { integrationProjection?: ForeachEnvironment["integrationProjection"]; semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"]; resumeReconcile?: ForeachEnvironment["resumeReconcile"]; + /** FIX 4 (context gap): task-level log sink for integration-conflict rework. */ + logTaskEntry?: ForeachEnvironment["logTaskEntry"]; + /** + * 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 + * (`buildParseStepsDeps` / `buildForeachWorktreeDeps` probe and flip rows under + * the SAME id). When omitted the runner derives `${task.id}:${definition.id}` — + * the same formula — so a caller that does not thread it keeps prior behavior. + */ + runId?: string; } /** @@ -189,7 +199,11 @@ export class WorkflowGraphTaskRunner { integrationProjection: this.deps.integrationProjection, semaphoreAvailability: this.deps.semaphoreAvailability, resumeReconcile: this.deps.resumeReconcile, - runId: `${task.id}:${definition.id}`, + logTaskEntry: this.deps.logTaskEntry, + // 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. + runId: this.deps.runId ?? `${task.id}:${definition.id}`, onBranchProgress: (progress) => { this.branchProgress.set(progress.branchId, progress); try { diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 8c0ca3c2cd..40ab61869f 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -81,6 +81,15 @@ export const FOREACH_ACTIVE_CONTEXT_KEY = "foreach:active"; */ export const SPLIT_ACTIVE_CONTEXT_KEY = "split:active"; +/** + * Reserved context marker (KTD-11) the worktree-isolation foreach seeds into an + * instance's context for ONE re-run after an integration-conflict, when the + * template authored an explicit `outcome:integration-conflict` edge. Author nodes + * read it to branch on the conflict; the sub-walk clears it after seeding so a + * later clean rework does not re-surface a stale conflict signal. + */ +export const INTEGRATION_CONFLICT_CONTEXT_KEY = "integration:conflict"; + /** Shape of the value stored under {@link FOREACH_ACTIVE_CONTEXT_KEY}. */ export interface ForeachActiveContext { foreachNodeId: string;