fix(review): apply code-review fixes — runId/foreachNodeId wiring, worktree-leak cleanup, rework re-execution, instance pruning, SIGKILL fallback, type dedup, board memo split, agent-native field-schema context + fn_workflow_get
17 findings from 12-reviewer code review applied: - P1 runId trio (pin-probe/resume/markIntegrated used placeholder runId; 4-reviewer corroboration) + production-wiring tests - P1 worktree/branch release on instance failure/exhaustion/abort - P2 runGraphTaskStep no longer masks step-session failures; rejected memo cleared so rework re-executes - P2 clearStaleInstanceStates wired at run start/end (mirrors branch pruning) - P2 code-node timeout killSignal SIGKILL; dead template-recursion removed - P1/P2 field-type re-declarations replaced with @fusion/core imports (stale comments removed) - P2 Board memo split + TaskCard comparator stringify guard + modal prop-driven field defs - HIGH agent-native: executor prompt injects custom-field schema/values; self-correcting rejection text; fn_workflow_get; fn_task_update bare-call guard; integration-conflict task log Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -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) |
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, import("../api").WorkflowFieldDefinition[]>();
|
||||
if (!boardWorkflows) return map;
|
||||
const { workflows, taskWorkflowIds, defaultWorkflowId } = boardWorkflows;
|
||||
const cardDefsByWorkflow = new Map<string, import("../api").WorkflowFieldDefinition[]>();
|
||||
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<string, import("../api").WorkflowFieldDefinition[]>();
|
||||
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
|
||||
|
||||
@@ -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 &&
|
||||
|
||||
@@ -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<TaskDetailModalProps, "onClose"> & {
|
||||
@@ -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<WorkflowFieldDefinition[] | null>(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<WorkflowFieldDefinition[] | null>(
|
||||
workflowFieldDefsProp !== undefined ? (workflowFieldDefsProp ?? null) : null,
|
||||
);
|
||||
const [customFieldValues, setCustomFieldValues] = useState<Record<string, unknown>>(task.customFields ?? {});
|
||||
const [customFieldError, setCustomFieldError] = useState<CustomFieldRejection | null>(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<string, unknown>) => {
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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", () => {
|
||||
|
||||
@@ -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(
|
||||
<FileBrowserProvider openFile={vi.fn()}>
|
||||
<TaskDetailModal
|
||||
task={makeTask({ id: "FN-002", column: "done", customFields: { owner: "bob" } })}
|
||||
workflowFieldDefs={defs}
|
||||
onClose={noop}
|
||||
onMoveTask={noopMove}
|
||||
onDeleteTask={noopDelete}
|
||||
onMergeTask={noopMerge}
|
||||
onOpenDetail={noopOpenDetail}
|
||||
addToast={noop}
|
||||
/>
|
||||
</FileBrowserProvider>,
|
||||
);
|
||||
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(
|
||||
<FileBrowserProvider openFile={vi.fn()}>
|
||||
<TaskDetailModal
|
||||
task={makeTask({ id: "FN-003", column: "done" })}
|
||||
workflowFieldDefs={[]}
|
||||
onClose={noop}
|
||||
onMoveTask={noopMove}
|
||||
onDeleteTask={noopDelete}
|
||||
onMergeTask={noopMerge}
|
||||
onOpenDetail={noopOpenDetail}
|
||||
addToast={noop}
|
||||
/>
|
||||
</FileBrowserProvider>,
|
||||
);
|
||||
// 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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<WorkflowFlowNodeData>[],
|
||||
edges: FlowEdge[],
|
||||
columns?: WorkflowIrColumn[],
|
||||
fields?: WorkflowFieldDefinitionShape[],
|
||||
fields?: WorkflowFieldDefinition[],
|
||||
): { ir: WorkflowIr; layout: Record<string, { x: number; y: number }> } {
|
||||
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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
237
packages/engine/src/__tests__/workflow-foreach-wiring.test.ts
Normal file
237
packages/engine/src/__tests__/workflow-foreach-wiring.test.ts
Normal file
@@ -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<TaskStore["saveWorkflowRunStepInstance"]>[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<string, unknown>): 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>): 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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
|
||||
@@ -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<typeof workflowGetParams>) => {
|
||||
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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, Promise<{ taskDone: boolean; modifiedFiles: string[] }>>();
|
||||
|
||||
/** 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<string, ForeachActiveContext>();
|
||||
|
||||
/** 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<WorkflowFieldDefinition[] | undefined> {
|
||||
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<string | undefined> {
|
||||
// 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<string | undefined> => {
|
||||
// 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<string | undefined> => this.readTaskArtifact(task.id, key),
|
||||
writeSteps: async (task, steps: TaskStep[]): Promise<void> => {
|
||||
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<void> => {
|
||||
markInstanceIntegrated: async (stepIndex, integratedAt, identity): Promise<void> => {
|
||||
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<typeof taskUpdateParams>) => {
|
||||
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." : ""}
|
||||
|
||||
@@ -8,6 +8,7 @@ export {
|
||||
createSendMessageTool,
|
||||
createReadMessagesTool,
|
||||
createWorkflowListTool,
|
||||
createWorkflowGetTool,
|
||||
createWorkflowSelectTool,
|
||||
taskCreateParams,
|
||||
taskDocumentReadParams,
|
||||
|
||||
@@ -68,6 +68,21 @@ export interface IntegrationGitOps {
|
||||
discardBranch(branchName: string, stepIndex: number): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void>;
|
||||
/**
|
||||
* 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> | void;
|
||||
markInstanceIntegrated?(
|
||||
stepIndex: number,
|
||||
integratedAt: string,
|
||||
identity: IntegrationInstanceIdentity,
|
||||
): Promise<void> | 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);
|
||||
|
||||
@@ -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<void> {
|
||||
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";
|
||||
|
||||
@@ -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 }>>
|
||||
| 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<void>;
|
||||
}
|
||||
|
||||
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<number>();
|
||||
const releaseInstanceWorktree = async (inst: WorktreeInstance): Promise<void> => {
|
||||
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<void> => {
|
||||
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<void> =>
|
||||
persistInstanceState(env.persistence, {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user