feat(engine): U10 — parallel step execution: dependency scheduler, per-instance worktrees, ordered integration, conflict→rework (KTD-11)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -386,7 +386,10 @@ describe("WorkflowGraphExecutor foreach (U3)", () => {
|
||||
expect(result.outcome).toBe("failure");
|
||||
});
|
||||
|
||||
it("parallel mode is guarded with a clear not-yet-wired failure (U10 replaces it)", async () => {
|
||||
it("parallel mode (now worktree isolation, U10) fails cleanly without isolation wiring", async () => {
|
||||
// U10: parallel mode defaults to worktree isolation. Without the worktree /
|
||||
// integration deps wired, the foreach fails with a routable value rather than
|
||||
// running shared-mode physics (which would be an unguardable concurrent-write race).
|
||||
const seams = baseSeams({
|
||||
stepExecute: async () => ({ outcome: "success", value: "step-done" }),
|
||||
});
|
||||
@@ -397,7 +400,7 @@ describe("WorkflowGraphExecutor foreach (U3)", () => {
|
||||
foreachIr(singleExecuteTemplate(), { config: { mode: "parallel", concurrency: 2 } }),
|
||||
);
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(result.context["node:fe:value"]).toBe("parallel-not-wired");
|
||||
expect(result.context["node:fe:value"]).toBe("worktree-isolation-unwired");
|
||||
});
|
||||
|
||||
it("getTaskSteps dep is used to read a fresh count when injected", async () => {
|
||||
|
||||
561
packages/engine/src/__tests__/workflow-step-parallel.test.ts
Normal file
561
packages/engine/src/__tests__/workflow-step-parallel.test.ts
Normal file
@@ -0,0 +1,561 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { TaskDetail, TaskStep, WorkflowIr, WorkflowIrNode } from "@fusion/core";
|
||||
|
||||
import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
|
||||
import {
|
||||
FOREACH_ACTIVE_CONTEXT_KEY,
|
||||
type ForeachActiveContext,
|
||||
type WorkflowLegacySeams,
|
||||
} from "../workflow-node-handlers.js";
|
||||
import {
|
||||
IntegrationQueue,
|
||||
type IntegrationGitOps,
|
||||
type IntegrationProjection,
|
||||
type IntegrationAttemptResult,
|
||||
} from "../step-integration.js";
|
||||
import type { WorkflowStepInstanceState } from "../workflow-graph-foreach.js";
|
||||
|
||||
const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } });
|
||||
|
||||
// ── shared test scaffolding ─────────────────────────────────────────────────
|
||||
|
||||
/** Build a TaskDetail with a step list; dependsOn (0-indexed) per step optional. */
|
||||
function taskWithSteps(specs: Array<{ dependsOn?: number[] }> | number): TaskDetail {
|
||||
const list: Array<{ dependsOn?: number[] }> =
|
||||
typeof specs === "number" ? Array.from({ length: specs }, () => ({})) : specs;
|
||||
const steps: TaskStep[] = list.map((s, i) => ({
|
||||
name: `Step ${i + 1}`,
|
||||
status: "pending" as const,
|
||||
...(s.dependsOn ? { dependsOn: s.dependsOn } : {}),
|
||||
}));
|
||||
return { id: "FN-PAR", steps } as unknown as TaskDetail;
|
||||
}
|
||||
|
||||
/** Base no-op seams with an optional override. */
|
||||
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 };
|
||||
}
|
||||
|
||||
/** A single step-execute template. */
|
||||
function singleExecuteTemplate() {
|
||||
return {
|
||||
nodes: [{ id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }],
|
||||
edges: [],
|
||||
};
|
||||
}
|
||||
|
||||
/** exec → step-review template (review routes approve/revise/rethink). */
|
||||
function reviewTemplate() {
|
||||
return {
|
||||
nodes: [
|
||||
{ id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } },
|
||||
{ id: "review", kind: "step-review" as const, config: { type: "code" } },
|
||||
],
|
||||
edges: [
|
||||
{ from: "exec", to: "review", condition: "success" },
|
||||
{ from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const },
|
||||
{ from: "review", to: "exec", condition: "outcome:rethink", kind: "rework" as const },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
function foreachIr(
|
||||
template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] },
|
||||
config: Record<string, unknown> = {},
|
||||
): WorkflowIr {
|
||||
return {
|
||||
version: "v2",
|
||||
name: "parallel-test",
|
||||
columns: [{ id: "work", name: "Work", traits: [] }],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "fe", kind: "foreach", config: { source: "task-steps", template, ...config } },
|
||||
{ id: "end", kind: "end" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "fe" },
|
||||
{ from: "fe", to: "end", condition: "success" },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* A fake worktree+git+integration backend the executor deps are wired to. Models
|
||||
* a per-instance branch and an ordered integration base purely in-memory; tests
|
||||
* script which (stepIndex) integrations conflict.
|
||||
*/
|
||||
function makeFakeBackend(opts: {
|
||||
conflictSteps?: Set<number>;
|
||||
/** Steps whose integration conflicts only on the FIRST attempt (then succeed). */
|
||||
conflictOnceSteps?: Set<number>;
|
||||
} = {}) {
|
||||
const conflictSteps = opts.conflictSteps ?? new Set<number>();
|
||||
const conflictOnceSteps = opts.conflictOnceSteps ?? new Set<number>();
|
||||
const integrateAttempts = new Map<number, number>();
|
||||
|
||||
const allocations: Array<{ stepIndex: number; branchName: string; base: string | undefined }> = [];
|
||||
const integrationOrder: number[] = [];
|
||||
const discarded: string[] = [];
|
||||
const released: string[] = [];
|
||||
const doneSteps: number[] = [];
|
||||
const instanceIntegrated: Array<{ stepIndex: number; at: string }> = [];
|
||||
let integrationBase = "main@0";
|
||||
let integratedCount = 0;
|
||||
const resetBranches: string[] = [];
|
||||
|
||||
const gitOps: IntegrationGitOps = {
|
||||
integrate: async (branchName, stepIndex): Promise<IntegrationAttemptResult> => {
|
||||
const attempt = (integrateAttempts.get(stepIndex) ?? 0) + 1;
|
||||
integrateAttempts.set(stepIndex, attempt);
|
||||
const conflictNow =
|
||||
conflictSteps.has(stepIndex) || (conflictOnceSteps.has(stepIndex) && attempt === 1);
|
||||
if (conflictNow) {
|
||||
return { kind: "conflict", conflictedFiles: [`step-${stepIndex}.ts`] };
|
||||
}
|
||||
integrationOrder.push(stepIndex);
|
||||
integratedCount += 1;
|
||||
integrationBase = `main@${integratedCount}`; // base advances on each integration
|
||||
return { kind: "integrated", integratedAt: `t${stepIndex}` };
|
||||
},
|
||||
discardBranch: async (branchName) => {
|
||||
discarded.push(branchName);
|
||||
released.push(branchName);
|
||||
},
|
||||
};
|
||||
|
||||
const projection: IntegrationProjection = {
|
||||
markStepDone: async (stepIndex) => {
|
||||
doneSteps.push(stepIndex);
|
||||
},
|
||||
markInstanceIntegrated: async (stepIndex, at) => {
|
||||
instanceIntegrated.push({ stepIndex, at });
|
||||
},
|
||||
};
|
||||
|
||||
return {
|
||||
gitOps,
|
||||
projection,
|
||||
allocations,
|
||||
integrationOrder,
|
||||
discarded,
|
||||
released,
|
||||
doneSteps,
|
||||
instanceIntegrated,
|
||||
resetBranches,
|
||||
getBase: () => integrationBase,
|
||||
deps: {
|
||||
allocateInstanceWorktree: async (stepIndex: number, base: string | undefined) => {
|
||||
const branchName = `fusion/fn-par-step-${stepIndex}`;
|
||||
allocations.push({ stepIndex, branchName, base });
|
||||
return { worktreePath: `/wt/step-${stepIndex}`, branchName };
|
||||
},
|
||||
resolveIntegrationBase: async () => integrationBase,
|
||||
integrationGitOps: gitOps,
|
||||
integrationProjection: projection,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ── IntegrationQueue state machine (TEST-FIRST) ─────────────────────────────
|
||||
|
||||
describe("IntegrationQueue (ordered integration state machine)", () => {
|
||||
function queueHarness(pinned: number, integrate: (b: string, i: number) => IntegrationAttemptResult) {
|
||||
const order: number[] = [];
|
||||
const done: number[] = [];
|
||||
const integrated: number[] = [];
|
||||
const discarded: string[] = [];
|
||||
const git: IntegrationGitOps = {
|
||||
integrate: async (b, i) => {
|
||||
const r = integrate(b, i);
|
||||
if (r.kind === "integrated") order.push(i);
|
||||
return r;
|
||||
},
|
||||
discardBranch: async (b) => {
|
||||
discarded.push(b);
|
||||
},
|
||||
};
|
||||
const proj: IntegrationProjection = {
|
||||
markStepDone: async (i) => {
|
||||
done.push(i);
|
||||
},
|
||||
markInstanceIntegrated: async (i) => {
|
||||
integrated.push(i);
|
||||
},
|
||||
};
|
||||
const q = new IntegrationQueue(git, proj, pinned);
|
||||
return { q, order, done, integrated, discarded };
|
||||
}
|
||||
|
||||
it("integrates strictly in step order even when completion order inverts", async () => {
|
||||
const h = queueHarness(3, () => ({ kind: "integrated", integratedAt: "t" }));
|
||||
// Enqueue out of order: 2 first, then 0, then 1.
|
||||
h.q.enqueue(2, "b2");
|
||||
let outcomes = await h.q.drain();
|
||||
expect(outcomes).toEqual([]); // 0 not ready → nothing integrates.
|
||||
h.q.enqueue(0, "b0");
|
||||
outcomes = await h.q.drain();
|
||||
expect(h.order).toEqual([0]); // only 0 (1 is the next gap).
|
||||
h.q.enqueue(1, "b1");
|
||||
await h.q.drain();
|
||||
expect(h.order).toEqual([0, 1, 2]); // 1 then 2 cascade.
|
||||
expect(h.q.isDrained()).toBe(true);
|
||||
});
|
||||
|
||||
it("projection-first: markStepDone precedes markInstanceIntegrated per step", async () => {
|
||||
const events: string[] = [];
|
||||
const git: IntegrationGitOps = {
|
||||
integrate: async () => ({ kind: "integrated", integratedAt: "t" }),
|
||||
discardBranch: async () => {},
|
||||
};
|
||||
const proj: IntegrationProjection = {
|
||||
markStepDone: async (i) => {
|
||||
events.push(`done:${i}`);
|
||||
},
|
||||
markInstanceIntegrated: async (i) => {
|
||||
events.push(`row:${i}`);
|
||||
},
|
||||
};
|
||||
const q = new IntegrationQueue(git, proj, 1);
|
||||
q.enqueue(0, "b0");
|
||||
await q.drain();
|
||||
expect(events).toEqual(["done:0", "row:0"]);
|
||||
});
|
||||
|
||||
it("conflict stops the drain, discards the branch, and does not mark done", async () => {
|
||||
const h = queueHarness(2, (_b, i) =>
|
||||
i === 0 ? { kind: "conflict", conflictedFiles: ["x"] } : { kind: "integrated", integratedAt: "t" },
|
||||
);
|
||||
h.q.enqueue(0, "b0");
|
||||
h.q.enqueue(1, "b1");
|
||||
const outcomes = await h.q.drain();
|
||||
expect(outcomes).toEqual([{ stepIndex: 0, status: "conflict", conflictedFiles: ["x"] }]);
|
||||
expect(h.done).toEqual([]);
|
||||
expect(h.discarded).toContain("b0");
|
||||
// Step 1 must NOT integrate ahead of the unresolved step 0.
|
||||
expect(h.order).toEqual([]);
|
||||
});
|
||||
|
||||
it("skip advances the cursor past a failed step", async () => {
|
||||
const h = queueHarness(3, () => ({ kind: "integrated", integratedAt: "t" }));
|
||||
h.q.skip(0);
|
||||
h.q.enqueue(1, "b1");
|
||||
h.q.enqueue(2, "b2");
|
||||
await h.q.drain();
|
||||
expect(h.order).toEqual([1, 2]);
|
||||
expect(h.q.isDrained()).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
// ── full U10 scenarios ──────────────────────────────────────────────────────
|
||||
|
||||
describe("WorkflowGraphExecutor parallel/worktree foreach (U10)", () => {
|
||||
/** Run a foreach IR with a fake backend; record the order steps START. */
|
||||
async function runScenario(
|
||||
task: TaskDetail,
|
||||
config: Record<string, unknown>,
|
||||
backend: ReturnType<typeof makeFakeBackend>,
|
||||
overrides: Partial<{
|
||||
semaphoreAvailability: () => number;
|
||||
stepExecute: WorkflowLegacySeams["stepExecute"];
|
||||
stepReview: WorkflowLegacySeams["stepReview"];
|
||||
onReworkReset: (a: ForeachActiveContext) => void;
|
||||
template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] };
|
||||
}> = {},
|
||||
) {
|
||||
const startOrder: number[] = [];
|
||||
const seams = baseSeams({
|
||||
stepExecute:
|
||||
overrides.stepExecute ??
|
||||
(async (_t, ctx) => {
|
||||
const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
|
||||
startOrder.push(active.stepIndex);
|
||||
return { outcome: "success", value: "step-done" };
|
||||
}),
|
||||
...(overrides.stepReview ? { stepReview: overrides.stepReview } : {}),
|
||||
});
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
seams,
|
||||
...backend.deps,
|
||||
...(overrides.semaphoreAvailability ? { semaphoreAvailability: overrides.semaphoreAvailability } : {}),
|
||||
...(overrides.onReworkReset ? { onReworkReset: overrides.onReworkReset as never } : {}),
|
||||
});
|
||||
const ir = foreachIr(overrides.template ?? singleExecuteTemplate(), config);
|
||||
const result = await executor.run(task, settingsOn(), ir);
|
||||
return { result, startOrder };
|
||||
}
|
||||
|
||||
it("diamond dep graph (0 ← 1,2 ← 3) runs 1∥2 then 3", async () => {
|
||||
// Step 0 root; 1 and 2 depend on 0; 3 depends on 1 and 2.
|
||||
const task = taskWithSteps([
|
||||
{ dependsOn: [] },
|
||||
{ dependsOn: [0] },
|
||||
{ dependsOn: [0] },
|
||||
{ dependsOn: [1, 2] },
|
||||
]);
|
||||
const backend = makeFakeBackend();
|
||||
const { result, startOrder } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 4 },
|
||||
backend,
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
// 0 first; 1 and 2 after 0 integrated; 3 last.
|
||||
expect(startOrder[0]).toBe(0);
|
||||
expect(new Set([startOrder[1], startOrder[2]])).toEqual(new Set([1, 2]));
|
||||
expect(startOrder[3]).toBe(3);
|
||||
// Ordered integration is step order.
|
||||
expect(backend.integrationOrder).toEqual([0, 1, 2, 3]);
|
||||
expect(backend.doneSteps).toEqual([0, 1, 2, 3]);
|
||||
});
|
||||
|
||||
it("sequential + worktree runs one at a time with per-step branches + ordered integration", async () => {
|
||||
const task = taskWithSteps(3);
|
||||
const backend = makeFakeBackend();
|
||||
const concurrentPeak = { value: 0 };
|
||||
let active = 0;
|
||||
const { result } = await runScenario(task, { mode: "sequential", isolation: "worktree" }, backend, {
|
||||
stepExecute: async () => {
|
||||
active += 1;
|
||||
concurrentPeak.value = Math.max(concurrentPeak.value, active);
|
||||
await Promise.resolve();
|
||||
active -= 1;
|
||||
return { outcome: "success", value: "step-done" };
|
||||
},
|
||||
});
|
||||
expect(result.outcome).toBe("success");
|
||||
expect(concurrentPeak.value).toBe(1); // never more than one at a time.
|
||||
expect(backend.allocations.map((a) => a.branchName)).toEqual([
|
||||
"fusion/fn-par-step-0",
|
||||
"fusion/fn-par-step-1",
|
||||
"fusion/fn-par-step-2",
|
||||
]);
|
||||
expect(backend.integrationOrder).toEqual([0, 1, 2]);
|
||||
});
|
||||
|
||||
it("unannotated plan stays fully sequential at concurrency 4", async () => {
|
||||
const task = taskWithSteps(4); // no dependsOn → each implicitly depends on prev.
|
||||
const backend = makeFakeBackend();
|
||||
const concurrentPeak = { value: 0 };
|
||||
const order: number[] = [];
|
||||
let active = 0;
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 4 },
|
||||
backend,
|
||||
{
|
||||
stepExecute: async (_t, ctx) => {
|
||||
const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
|
||||
order.push(a.stepIndex);
|
||||
active += 1;
|
||||
concurrentPeak.value = Math.max(concurrentPeak.value, active);
|
||||
await Promise.resolve();
|
||||
active -= 1;
|
||||
return { outcome: "success", value: "step-done" };
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
expect(concurrentPeak.value).toBe(1);
|
||||
expect(order).toEqual([0, 1, 2, 3]);
|
||||
expect(backend.integrationOrder).toEqual([0, 1, 2, 3]);
|
||||
});
|
||||
|
||||
it("conflict between parallel steps → loser reworks on updated base and succeeds", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }]);
|
||||
// Step 1 conflicts the first integration attempt, then succeeds.
|
||||
const backend = makeFakeBackend({ conflictOnceSteps: new Set([1]) });
|
||||
const execStarts: number[] = [];
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 2 },
|
||||
backend,
|
||||
{
|
||||
stepExecute: async (_t, ctx) => {
|
||||
const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
|
||||
execStarts.push(a.stepIndex);
|
||||
return { outcome: "success", value: "step-done" };
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
// Step 1 executed twice (initial + rework after conflict).
|
||||
expect(execStarts.filter((s) => s === 1).length).toBe(2);
|
||||
// Both eventually integrated, in step order.
|
||||
expect(backend.integrationOrder).toEqual([0, 1]);
|
||||
expect(backend.doneSteps).toEqual([0, 1]);
|
||||
// The conflicting branch was discarded before re-running.
|
||||
expect(backend.discarded).toContain("fusion/fn-par-step-1");
|
||||
// The rework re-allocated off the UPDATED base (after step 0 integrated).
|
||||
const step1Allocs = backend.allocations.filter((a) => a.stepIndex === 1);
|
||||
expect(step1Allocs.length).toBe(2);
|
||||
expect(step1Allocs[1].base).toBe("main@1");
|
||||
});
|
||||
|
||||
it("conflict rework exhaustion routes rework-exhausted", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }]);
|
||||
const backend = makeFakeBackend({ conflictSteps: new Set([0]) }); // always conflicts.
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 1, maxReworkCycles: 2 },
|
||||
backend,
|
||||
);
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(result.context).toBeDefined();
|
||||
// The foreach node's value surfaces rework-exhausted.
|
||||
expect(result.visitedNodeIds).toContain("fe");
|
||||
// Never marked done.
|
||||
expect(backend.doneSteps).toEqual([]);
|
||||
});
|
||||
|
||||
it("integration order is step order even when completion order inverts", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }, { dependsOn: [] }]);
|
||||
const backend = makeFakeBackend();
|
||||
// Make later steps complete FIRST by delaying step 0's execution.
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 3 },
|
||||
backend,
|
||||
{
|
||||
stepExecute: async (_t, ctx) => {
|
||||
const a = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext;
|
||||
// step 0 yields the most → completes last.
|
||||
const delays = 2 - a.stepIndex;
|
||||
for (let i = 0; i < delays; i++) await Promise.resolve();
|
||||
return { outcome: "success", value: "step-done" };
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
expect(backend.integrationOrder).toEqual([0, 1, 2]);
|
||||
expect(backend.doneSteps).toEqual([0, 1, 2]);
|
||||
});
|
||||
|
||||
it("semaphore starvation degrades to sequential without deadlock", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [] }, { dependsOn: [] }]);
|
||||
const backend = makeFakeBackend();
|
||||
const concurrentPeak = { value: 0 };
|
||||
let active = 0;
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 4 },
|
||||
backend,
|
||||
{
|
||||
semaphoreAvailability: () => 0, // fully starved.
|
||||
stepExecute: async () => {
|
||||
active += 1;
|
||||
concurrentPeak.value = Math.max(concurrentPeak.value, active);
|
||||
await Promise.resolve();
|
||||
active -= 1;
|
||||
return { outcome: "success", value: "step-done" };
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
expect(concurrentPeak.value).toBe(1); // forced to 1 under starvation, no deadlock.
|
||||
expect(backend.integrationOrder).toEqual([0, 1, 2]);
|
||||
});
|
||||
|
||||
it("dependency cycle at expansion fails audited", async () => {
|
||||
// Step 1 depends on step 2 (a forward reference → cycle signature).
|
||||
const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [2] }, { dependsOn: [] }]);
|
||||
const backend = makeFakeBackend();
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 4 },
|
||||
backend,
|
||||
);
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(backend.allocations).toEqual([]); // never expanded any instance.
|
||||
});
|
||||
|
||||
it("RETHINK resets only the instance branch (branch-scoped)", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }]);
|
||||
const backend = makeFakeBackend();
|
||||
const resetBranches: string[] = [];
|
||||
let reviewCalls = 0;
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "parallel", isolation: "worktree", concurrency: 1 },
|
||||
backend,
|
||||
{
|
||||
template: reviewTemplate(),
|
||||
stepReview: async () => {
|
||||
reviewCalls += 1;
|
||||
return reviewCalls === 1 ? { verdict: "RETHINK" as const } : { verdict: "APPROVE" as const };
|
||||
},
|
||||
onReworkReset: (active: ForeachActiveContext) => {
|
||||
// Branch-scoped reset: the active context carries THIS instance's branch.
|
||||
resetBranches.push(active.branchName ?? "<none>");
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
expect(resetBranches).toEqual(["fusion/fn-par-step-0"]);
|
||||
expect(backend.integrationOrder).toEqual([0]);
|
||||
});
|
||||
|
||||
it("merge-blocker stays blocked until last integration (projection rule)", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }, { dependsOn: [0] }]);
|
||||
const backend = makeFakeBackend();
|
||||
// Capture doneSteps progression: step 1 must not be done until it integrates.
|
||||
const doneAfterStep0Integrated: number[] = [];
|
||||
const origMarkDone = backend.projection.markStepDone;
|
||||
backend.projection.markStepDone = async (i) => {
|
||||
await origMarkDone(i);
|
||||
doneAfterStep0Integrated.push(i);
|
||||
};
|
||||
const { result } = await runScenario(
|
||||
task,
|
||||
{ mode: "sequential", isolation: "worktree" },
|
||||
backend,
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
// done flips strictly in integration order — step 1 done ONLY after step 0.
|
||||
expect(doneAfterStep0Integrated).toEqual([0, 1]);
|
||||
});
|
||||
|
||||
it("worktree isolation without wiring fails cleanly (routable)", async () => {
|
||||
const task = taskWithSteps(2);
|
||||
const executor = new WorkflowGraphExecutor({ seams: baseSeams({}) });
|
||||
const result = await executor.run(
|
||||
task,
|
||||
settingsOn(),
|
||||
foreachIr(singleExecuteTemplate(), { mode: "parallel", isolation: "worktree" }),
|
||||
);
|
||||
expect(result.outcome).toBe("failure");
|
||||
});
|
||||
});
|
||||
|
||||
// ── crash-resume reconciliation ─────────────────────────────────────────────
|
||||
|
||||
describe("worktree-isolation crash-resume reconciliation (U10)", () => {
|
||||
it("persists branchName + awaiting-integration through the persistence hook", async () => {
|
||||
const task = taskWithSteps([{ dependsOn: [] }]);
|
||||
const backend = makeFakeBackend();
|
||||
const saved: WorkflowStepInstanceState[] = [];
|
||||
const persistence = {
|
||||
saveInstanceState: (s: WorkflowStepInstanceState) => {
|
||||
saved.push({ ...s });
|
||||
},
|
||||
};
|
||||
const seams = baseSeams({
|
||||
stepExecute: async () => ({ outcome: "success", value: "step-done" }),
|
||||
});
|
||||
const executor = new WorkflowGraphExecutor({
|
||||
seams,
|
||||
...backend.deps,
|
||||
stepInstancePersistence: persistence,
|
||||
});
|
||||
const result = await executor.run(
|
||||
task,
|
||||
settingsOn(),
|
||||
foreachIr(singleExecuteTemplate(), { mode: "sequential", isolation: "worktree" }),
|
||||
);
|
||||
expect(result.outcome).toBe("success");
|
||||
// The instance row carried branchName and reached awaiting-integration.
|
||||
const awaiting = saved.find((s) => s.status === "awaiting-integration");
|
||||
expect(awaiting).toBeDefined();
|
||||
expect(awaiting?.branchName).toBe("fusion/fn-par-step-0");
|
||||
});
|
||||
});
|
||||
@@ -45,14 +45,14 @@ import {
|
||||
resolveAgentMemoryInclusionMode,
|
||||
type RunCommandResult,
|
||||
} from "@fusion/core";
|
||||
import { findWorktreeUser } from "./merger.js";
|
||||
import { findWorktreeUser, getConflictedFiles } from "./merger.js";
|
||||
import {
|
||||
runVerificationCommand,
|
||||
summarizeVerificationOutput,
|
||||
VERIFICATION_LOG_MAX_CHARS,
|
||||
type VerificationResult,
|
||||
} from "./verification-utils.js";
|
||||
import { generateWorktreeName, resolveTaskWorkingBranch } from "./worktree-names.js";
|
||||
import { canonicalStepInstanceBranchName, generateWorktreeName, resolveTaskWorkingBranch } from "./worktree-names.js";
|
||||
import { resolveTaskWorktreePath, resolveWorktreesDir } from "./worktree-paths.js";
|
||||
import { Type, type Static } from "@earendil-works/pi-ai";
|
||||
import { describeModel, promptWithFallback, compactSessionContext } from "./pi.js";
|
||||
@@ -3306,6 +3306,11 @@ export class TaskExecutor {
|
||||
// Step-inversion (KTD-15, U14): code node runner — esbuild compile +
|
||||
// child-process execution with the harness contract.
|
||||
runCode: this.buildCodeNodeRunner(),
|
||||
// Step-inversion (KTD-11, U10): worktree isolation + ordered integration +
|
||||
// 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),
|
||||
});
|
||||
let result: WorkflowGraphTaskRunResult;
|
||||
try {
|
||||
@@ -3499,6 +3504,216 @@ export class TaskExecutor {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the worktree-isolation + ordered-integration + parallel-scheduling deps
|
||||
* for a graph-owned foreach (KTD-11, U10). Returns the additive set the
|
||||
* WorkflowGraphTaskRunner forwards to the foreach sub-walk:
|
||||
*
|
||||
* - `allocateInstanceWorktree(i, base)` — a per-instance worktree on a
|
||||
* canonical `fusion/<task>-step-<i>` branch off `base` (the main tip),
|
||||
* created via the existing `createWorktree` path (the file-scope guard the
|
||||
* session machinery installs applies unchanged to anything the instance
|
||||
* session commits in this worktree — we do NOT bypass it);
|
||||
* - `resolveIntegrationBase()` — the task's main branch tip, re-read before each
|
||||
* (re)allocation so a rework lands on the UPDATED base;
|
||||
* - `integrationGitOps` — rebase the instance branch onto the main branch in
|
||||
* the task's MAIN worktree, fast-forward main on success; on conflict reuse
|
||||
* merger.ts `getConflictedFiles` (NOT reimplemented) and abort the rebase so
|
||||
* the next instance can integrate; `discardBranch` deletes the branch + frees
|
||||
* the instance worktree (pool hygiene);
|
||||
* - `integrationProjection` — projection-first ordering (KTD-7): `markStepDone`
|
||||
* flips the step `done` via `updateStep(source:"graph")` (the dependency-order
|
||||
* guard admits it), THEN `markInstanceIntegrated` flips the persisted row;
|
||||
* - `semaphoreAvailability` — the live free-slot count so parallel scheduling
|
||||
* clamps without hold-and-wait.
|
||||
*
|
||||
* 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): {
|
||||
allocateInstanceWorktree: (
|
||||
stepIndex: number,
|
||||
base: string | undefined,
|
||||
) => Promise<{ worktreePath: string; branchName: string }>;
|
||||
resolveIntegrationBase: () => Promise<string | undefined>;
|
||||
integrationGitOps: import("./step-integration.js").IntegrationGitOps;
|
||||
integrationProjection: import("./step-integration.js").IntegrationProjection;
|
||||
semaphoreAvailability: () => number;
|
||||
resumeReconcile: (
|
||||
pinned: number,
|
||||
) => Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>>;
|
||||
} {
|
||||
const taskId = task.id;
|
||||
// Per-instance worktree paths, so discard can free them.
|
||||
const instancePaths = new Map<number, string>();
|
||||
|
||||
const mainWorktree = async (): Promise<string> => {
|
||||
try {
|
||||
return (await this.store.getTask(taskId)).worktree || this.rootDir;
|
||||
} catch {
|
||||
return this.rootDir;
|
||||
}
|
||||
};
|
||||
const mainBranch = async (): Promise<string> => {
|
||||
try {
|
||||
const detail = await this.store.getTask(taskId);
|
||||
return resolveTaskWorkingBranch(detail);
|
||||
} catch {
|
||||
return resolveTaskWorkingBranch(task);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
resolveIntegrationBase: async (): Promise<string | undefined> => {
|
||||
// The main branch tip (HEAD of the task's working branch in its worktree).
|
||||
try {
|
||||
const { stdout } = await execAsync("git rev-parse HEAD", { cwd: await mainWorktree() });
|
||||
const sha = stdout.trim();
|
||||
return sha.length > 0 ? sha : await mainBranch();
|
||||
} catch {
|
||||
return await mainBranch();
|
||||
}
|
||||
},
|
||||
allocateInstanceWorktree: async (stepIndex, base): Promise<{ worktreePath: string; branchName: string }> => {
|
||||
const branchName = canonicalStepInstanceBranchName(taskId, stepIndex);
|
||||
const worktreePath = resolveTaskWorktreePath(
|
||||
this.rootDir,
|
||||
undefined,
|
||||
`${taskId.toLowerCase()}-step-${stepIndex}`,
|
||||
);
|
||||
// createWorktree installs the file-scope guard (session machinery,
|
||||
// unchanged) and branches off `base` (the integration base / updated tip).
|
||||
const created = await this.createWorktree(branchName, worktreePath, taskId, base);
|
||||
instancePaths.set(stepIndex, created.path);
|
||||
return { worktreePath: created.path, branchName: created.branch };
|
||||
},
|
||||
integrationGitOps: {
|
||||
integrate: async (branchName, stepIndex): Promise<import("./step-integration.js").IntegrationAttemptResult> => {
|
||||
const cwd = await mainWorktree();
|
||||
const target = await mainBranch();
|
||||
try {
|
||||
// Rebase the instance branch onto the current main tip, then ff main.
|
||||
await execAsync(`git rebase ${target} ${branchName}`, { cwd });
|
||||
await execAsync(`git checkout ${target}`, { cwd });
|
||||
await execAsync(`git merge --ff-only ${branchName}`, { cwd });
|
||||
return { kind: "integrated", integratedAt: new Date().toISOString() };
|
||||
} catch (err) {
|
||||
// Conflict (or other rebase failure): classify via merger helper, abort.
|
||||
const conflictedFiles = await getConflictedFiles(cwd);
|
||||
try {
|
||||
await execAsync("git rebase --abort", { cwd });
|
||||
} catch {
|
||||
// best-effort; leave the worktree recoverable.
|
||||
}
|
||||
// Restore main checkout so the next instance integrates cleanly.
|
||||
try {
|
||||
await execAsync(`git checkout ${target}`, { cwd });
|
||||
} catch {
|
||||
// best-effort.
|
||||
}
|
||||
executorLog.warn(
|
||||
`[step-integration] ${taskId} step ${stepIndex} branch ${branchName} conflict: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
return { kind: "conflict", conflictedFiles };
|
||||
}
|
||||
},
|
||||
discardBranch: async (branchName, stepIndex): Promise<void> => {
|
||||
const cwd = await mainWorktree();
|
||||
const path = instancePaths.get(stepIndex);
|
||||
if (path) {
|
||||
// Remove the instance worktree (pool hygiene). Best-effort; force so a
|
||||
// dirty/conflicting tree is still cleaned up.
|
||||
try {
|
||||
await execAsync(`git worktree remove --force "${path}"`, { cwd: this.rootDir });
|
||||
} catch {
|
||||
// best-effort cleanup.
|
||||
}
|
||||
instancePaths.delete(stepIndex);
|
||||
}
|
||||
// Delete the (now-merged or conflicting) branch.
|
||||
try {
|
||||
await execAsync(`git branch -D ${branchName}`, { cwd });
|
||||
} catch {
|
||||
// best-effort — the branch may already be gone.
|
||||
}
|
||||
},
|
||||
},
|
||||
integrationProjection: {
|
||||
markStepDone: async (stepIndex): Promise<void> => {
|
||||
// Projection-first (KTD-7): graph-source write relaxes the guard to
|
||||
// dependency order; predecessors are integrated (done) by construction.
|
||||
await this.store.updateStep(taskId, stepIndex, "done", { source: "graph" });
|
||||
},
|
||||
markInstanceIntegrated: async (stepIndex, integratedAt): Promise<void> => {
|
||||
const store = this.store as unknown as {
|
||||
saveWorkflowRunStepInstance?: (state: WorkflowStepInstanceState) => void;
|
||||
};
|
||||
if (typeof store.saveWorkflowRunStepInstance !== "function") return;
|
||||
try {
|
||||
store.saveWorkflowRunStepInstance({
|
||||
taskId,
|
||||
runId: `${taskId}:run`,
|
||||
foreachNodeId: "",
|
||||
stepIndex,
|
||||
pinnedStepCount: 0,
|
||||
currentNodeId: "",
|
||||
status: "completed",
|
||||
reworkCount: 0,
|
||||
branchName: canonicalStepInstanceBranchName(taskId, stepIndex),
|
||||
integratedAt,
|
||||
} as WorkflowStepInstanceState);
|
||||
} catch {
|
||||
// Persistence is additive bookkeeping — never fail the integration.
|
||||
}
|
||||
},
|
||||
},
|
||||
semaphoreAvailability: (): number => this.options.semaphore?.availableCount ?? 1,
|
||||
resumeReconcile: async (
|
||||
pinned,
|
||||
): Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>> => {
|
||||
// Crash-resume reconciliation (KTD-11): reconcile each persisted instance
|
||||
// row against branch existence. integrated → done; branch exists not
|
||||
// integrated → re-enter the integration queue; branch missing → re-run.
|
||||
// NOTE (handoff): this is the per-run resume seeding only; the full
|
||||
// self-healing sweep across stale runs (recoverStaleTransitionPending
|
||||
// analogue) is out of scope for U10.
|
||||
const store = this.store as unknown as {
|
||||
loadWorkflowRunStepInstances?: (taskId: string, runId: string) => WorkflowStepInstanceState[];
|
||||
};
|
||||
if (typeof store.loadWorkflowRunStepInstances !== "function") return [];
|
||||
let rows: WorkflowStepInstanceState[] = [];
|
||||
try {
|
||||
rows = store.loadWorkflowRunStepInstances(taskId, `${taskId}:run`) ?? [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
const cwd = await mainWorktree();
|
||||
const out: Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }> = [];
|
||||
for (const row of rows) {
|
||||
if (row.stepIndex < 0 || row.stepIndex >= pinned) continue;
|
||||
if (row.status === "completed" || row.integratedAt) {
|
||||
out.push({ stepIndex: row.stepIndex, disposition: "integrated" });
|
||||
continue;
|
||||
}
|
||||
const branchName = row.branchName || canonicalStepInstanceBranchName(taskId, row.stepIndex);
|
||||
let branchExists = false;
|
||||
try {
|
||||
await execAsync(`git rev-parse --verify --quiet ${branchName}`, { cwd });
|
||||
branchExists = true;
|
||||
} catch {
|
||||
branchExists = false;
|
||||
}
|
||||
if (branchExists && row.status === "awaiting-integration") {
|
||||
out.push({ stepIndex: row.stepIndex, disposition: "reintegrate", branchName });
|
||||
} else {
|
||||
out.push({ stepIndex: row.stepIndex, disposition: "rerun" });
|
||||
}
|
||||
}
|
||||
return out;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* RETHINK reset-on-rework (KTD-4, U5): reset the active foreach instance's step
|
||||
* to its per-step baseline before the rework edge re-enters step-execute. Drives
|
||||
@@ -3509,11 +3724,18 @@ export class TaskExecutor {
|
||||
* the documented KTD-2 semantics; the git reset + step→pending are authoritative.
|
||||
*/
|
||||
private async applyGraphRethinkReset(taskId: string, active: ForeachActiveContext): Promise<void> {
|
||||
let worktreePath = this.rootDir;
|
||||
try {
|
||||
worktreePath = (await this.store.getTask(taskId)).worktree || this.rootDir;
|
||||
} catch {
|
||||
// Best-effort worktree resolution; fall back to rootDir.
|
||||
// Worktree isolation (KTD-11): reset the instance's OWN branch/worktree only —
|
||||
// sibling instances and the integration base are untouched, so the blast-radius
|
||||
// guard is STRUCTURAL (skipped) in this mode. Shared isolation resets the task's
|
||||
// main worktree and keeps the KTD-2 ancestry guard as written.
|
||||
const branchScoped = typeof active.worktreePath === "string" && active.worktreePath.length > 0;
|
||||
let worktreePath = active.worktreePath ?? this.rootDir;
|
||||
if (!branchScoped) {
|
||||
try {
|
||||
worktreePath = (await this.store.getTask(taskId)).worktree || this.rootDir;
|
||||
} catch {
|
||||
// Best-effort worktree resolution; fall back to rootDir.
|
||||
}
|
||||
}
|
||||
const liveSteps = await this.store.getTask(taskId).then((t) => t.steps).catch(() => []);
|
||||
await resetStepToBaseline(
|
||||
@@ -3524,11 +3746,16 @@ export class TaskExecutor {
|
||||
// when checkpointId resolves but no session is current (KTD-2 partial path).
|
||||
sessionRef: { current: null },
|
||||
reviewType: "code",
|
||||
blastRadiusGuard: makeAncestryBlastRadiusGuard({
|
||||
worktreePath,
|
||||
task: { id: taskId, steps: liveSteps },
|
||||
stepIndex: active.stepIndex,
|
||||
}),
|
||||
// Branch-scoped RETHINK under worktree isolation makes the guard structural
|
||||
// (the reset can only touch the instance's own branch); shared isolation
|
||||
// keeps the defensive ancestry guard (KTD-2/KTD-11).
|
||||
blastRadiusGuard: branchScoped
|
||||
? undefined
|
||||
: makeAncestryBlastRadiusGuard({
|
||||
worktreePath,
|
||||
task: { id: taskId, steps: liveSteps },
|
||||
stepIndex: active.stepIndex,
|
||||
}),
|
||||
},
|
||||
{ id: taskId, steps: liveSteps },
|
||||
active.stepIndex,
|
||||
@@ -3815,7 +4042,11 @@ export class TaskExecutor {
|
||||
return { outcome: "failure", value: "no-active-step-instance" };
|
||||
}
|
||||
const live = await this.store.getTask(seamTask.id);
|
||||
const worktreePath = live.worktree || this.rootDir;
|
||||
// Worktree isolation (KTD-11, U10): run the instance's session in ITS OWN
|
||||
// worktree when the foreach allocated one; otherwise the task's main
|
||||
// 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;
|
||||
const result = await runTaskStep(
|
||||
{
|
||||
store: this.store,
|
||||
@@ -3864,7 +4095,8 @@ export class TaskExecutor {
|
||||
}
|
||||
const stepIndex = active.stepIndex;
|
||||
const detail = await this.store.getTask(seamTask.id);
|
||||
const worktreePath = detail.worktree || this.rootDir;
|
||||
// Worktree isolation (KTD-11): review the instance's OWN worktree when set.
|
||||
const worktreePath = active.worktreePath || detail.worktree || this.rootDir;
|
||||
const stepName = detail.steps[stepIndex]?.name ?? `Step ${stepIndex + 1}`;
|
||||
const promptContent = detail.prompt ?? "";
|
||||
const settings = await this.store.getSettings();
|
||||
|
||||
241
packages/engine/src/step-integration.ts
Normal file
241
packages/engine/src/step-integration.ts
Normal file
@@ -0,0 +1,241 @@
|
||||
/**
|
||||
* step-integration — the ordered integration stage for worktree-isolated foreach
|
||||
* instances (step-inversion KTD-11, U10).
|
||||
*
|
||||
* Under `isolation: "worktree"` each foreach step instance runs in its OWN
|
||||
* worktree/branch off a common integration base. Completing the instance's
|
||||
* sub-walk does NOT mark the step done — instead the instance enqueues here as
|
||||
* `awaiting-integration`. The {@link IntegrationQueue} then lands completed
|
||||
* branches onto the task's main branch (the integration base) **strictly in step
|
||||
* order**: instance `i` integrates only after instances `0..i-1` are integrated
|
||||
* or skipped. This is the single place where a worktree-isolated instance's work
|
||||
* becomes visible on main history, so:
|
||||
*
|
||||
* - **Success**: flip the projection FIRST (`updateStep(..., "done", graph)` —
|
||||
* the dependency-order guard admits it because predecessors are done), THEN
|
||||
* mark the instance row `completed`/`integratedAt` (projection-first ordering,
|
||||
* KTD-7: closes the merge-blocker race), THEN release the instance worktree
|
||||
* (pool hygiene).
|
||||
* - **Conflict**: discard the instance branch (release worktree), and emit
|
||||
* `outcome:integration-conflict` for that instance. The foreach sub-walk
|
||||
* routes that like a rework — re-execute the step on the UPDATED integration
|
||||
* base in a fresh worktree, counting against the instance's `maxReworkCycles`
|
||||
* budget (exhaustion → rework-exhausted as usual).
|
||||
*
|
||||
* All git mechanics are behind the injectable {@link IntegrationGitOps} so this
|
||||
* module is hermetically testable with fakes. Production wires it (executor.ts)
|
||||
* to a rebase/cherry-pick onto the task's main branch that reuses merger.ts's
|
||||
* conflict-classification helpers (`getConflictedFiles`) for conflict detection —
|
||||
* NOT reimplemented here.
|
||||
*
|
||||
* Reconcile alignment (KTD-11): `complete step N` commits live on instance
|
||||
* branches before integration and on main history after it; the projection rule
|
||||
* ("done iff integrated") therefore agrees with `reconcileStepsFromGitHistory`
|
||||
* (which reads main-worktree history) by construction.
|
||||
*/
|
||||
|
||||
import { schedulerLog } from "./logger.js";
|
||||
|
||||
/** The outcome of attempting to integrate one instance branch onto the base. */
|
||||
export type IntegrationAttemptResult =
|
||||
| { kind: "integrated"; integratedAt: string }
|
||||
| { kind: "conflict"; conflictedFiles: string[] };
|
||||
|
||||
/**
|
||||
* Injectable git mechanics for the ordered integration stage (KTD-11). Production
|
||||
* (executor.ts) implements these over real git — `integrate` does a rebase /
|
||||
* cherry-pick of the instance branch onto the task's main branch and uses
|
||||
* merger.ts's `getConflictedFiles` to detect conflicts; `discardBranch` deletes
|
||||
* the conflicting branch. Tests inject fakes for fast, deterministic runs.
|
||||
*/
|
||||
export interface IntegrationGitOps {
|
||||
/**
|
||||
* Land `branchName` onto the integration base (the task's main branch) for step
|
||||
* `stepIndex`. Returns `integrated` on a clean rebase/cherry-pick (the base now
|
||||
* contains the step's commits), or `conflict` with the conflicting file list.
|
||||
* MUST NOT mutate the projection or instance rows — that is the queue's job
|
||||
* (projection-first ordering). On `conflict` the implementation MUST leave the
|
||||
* base clean (abort the rebase) so the next instance can integrate.
|
||||
*/
|
||||
integrate(
|
||||
branchName: string,
|
||||
stepIndex: number,
|
||||
): Promise<IntegrationAttemptResult>;
|
||||
/**
|
||||
* Discard a conflicting (or abandoned) instance branch and release its worktree
|
||||
* (pool hygiene). Best-effort — never throws into the queue.
|
||||
*/
|
||||
discardBranch(branchName: string, stepIndex: number): Promise<void>;
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
export interface IntegrationProjection {
|
||||
/**
|
||||
* Flip the projection FIRST (KTD-7): `updateStep(taskId, stepIndex, "done")`
|
||||
* with graph source so the dependency-order guard admits it. Awaited before the
|
||||
* instance row flips to `completed`, closing the merge-blocker race.
|
||||
*/
|
||||
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.
|
||||
*/
|
||||
markInstanceIntegrated?(stepIndex: number, integratedAt: string): Promise<void> | void;
|
||||
}
|
||||
|
||||
/** One enqueued, completed instance awaiting ordered integration. */
|
||||
export interface PendingIntegration {
|
||||
stepIndex: number;
|
||||
branchName: string;
|
||||
}
|
||||
|
||||
/** The disposition of one instance after the queue drained as far as it could. */
|
||||
export type InstanceIntegrationOutcome =
|
||||
| { stepIndex: number; status: "integrated"; integratedAt: string }
|
||||
| { stepIndex: number; status: "conflict"; conflictedFiles: string[] };
|
||||
|
||||
/**
|
||||
* Per-(task, run, foreach) ordered integration queue (KTD-11).
|
||||
*
|
||||
* Completed worktree-isolated instances enqueue via {@link enqueue}; the queue
|
||||
* lands them onto the integration base **strictly in step order**. The scheduler
|
||||
* calls {@link drain} whenever a new instance becomes available (or on each
|
||||
* scheduler tick); `drain` integrates every contiguous run of ready instances
|
||||
* starting at the lowest not-yet-resolved step index, stopping at the first gap
|
||||
* (a step not yet completed) or the first conflict. Conflicts are reported back
|
||||
* so the scheduler can route `outcome:integration-conflict` for that instance.
|
||||
*
|
||||
* The queue NEVER skips ahead past a gap: instance `i` integrates only after
|
||||
* `0..i-1` are integrated or skipped, so completion-order inversion (a later step
|
||||
* finishing first) cannot reorder integration — integration order is step order.
|
||||
*/
|
||||
export class IntegrationQueue {
|
||||
/** Instances that have completed and are waiting to integrate, by step index. */
|
||||
private readonly pending = new Map<number, PendingIntegration>();
|
||||
/** Step indices whose integration is resolved (integrated OR routed conflict). */
|
||||
private readonly resolved = new Set<number>();
|
||||
/** The next step index eligible to integrate (advances as the queue drains). */
|
||||
private cursor = 0;
|
||||
/** Steps the scheduler told us to SKIP (e.g. dependency-failed) — treated as
|
||||
* resolved so the cursor advances past them without blocking later steps. */
|
||||
private readonly skipped = new Set<number>();
|
||||
|
||||
constructor(
|
||||
private readonly gitOps: IntegrationGitOps,
|
||||
private readonly projection: IntegrationProjection,
|
||||
private readonly pinnedStepCount: number,
|
||||
) {}
|
||||
|
||||
/** Enqueue a completed instance awaiting integration. Idempotent per step. */
|
||||
enqueue(stepIndex: number, branchName: string): void {
|
||||
if (this.resolved.has(stepIndex)) return;
|
||||
this.pending.set(stepIndex, { stepIndex, branchName });
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a step index as skipped (resolved without integration), so the ordered
|
||||
* cursor can advance past it. Used when an instance failed before producing a
|
||||
* branch (the projection stays non-done; the foreach reports the failure).
|
||||
*/
|
||||
skip(stepIndex: number): void {
|
||||
if (this.resolved.has(stepIndex)) return;
|
||||
this.skipped.add(stepIndex);
|
||||
this.resolved.add(stepIndex);
|
||||
this.advanceCursor();
|
||||
}
|
||||
|
||||
/** Whether step `i` is still awaiting integration in the queue. */
|
||||
isPending(stepIndex: number): boolean {
|
||||
return this.pending.has(stepIndex);
|
||||
}
|
||||
|
||||
/** Whether step `i` has been integrated or routed to conflict/skip. */
|
||||
isResolved(stepIndex: number): boolean {
|
||||
return this.resolved.has(stepIndex);
|
||||
}
|
||||
|
||||
/**
|
||||
* Integrate every contiguous ready instance starting at the cursor, in step
|
||||
* order. Stops at the first gap (the cursor's step hasn't completed yet) or the
|
||||
* first conflict (the conflicting step is reported and NOT marked resolved here
|
||||
* — the scheduler routes it to rework, then re-enqueues or re-skips). Returns
|
||||
* the per-instance outcomes produced THIS drain (callers act on conflicts).
|
||||
*/
|
||||
async drain(): Promise<InstanceIntegrationOutcome[]> {
|
||||
const outcomes: InstanceIntegrationOutcome[] = [];
|
||||
for (;;) {
|
||||
// Advance past any already-resolved/skipped steps so the cursor points at
|
||||
// the lowest unresolved step.
|
||||
this.advanceCursor();
|
||||
if (this.cursor >= this.pinnedStepCount) break;
|
||||
|
||||
const ready = this.pending.get(this.cursor);
|
||||
if (!ready) break; // Gap: the lowest unresolved step hasn't completed yet.
|
||||
|
||||
const result = await this.gitOps.integrate(ready.branchName, ready.stepIndex);
|
||||
if (result.kind === "integrated") {
|
||||
// Projection-first ordering (KTD-7): flip the step done BEFORE the instance
|
||||
// row, then release the worktree (the discard path releases on conflict;
|
||||
// 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);
|
||||
// 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);
|
||||
this.pending.delete(this.cursor);
|
||||
this.resolved.add(this.cursor);
|
||||
outcomes.push({
|
||||
stepIndex: ready.stepIndex,
|
||||
status: "integrated",
|
||||
integratedAt: result.integratedAt,
|
||||
});
|
||||
this.advanceCursor();
|
||||
continue;
|
||||
}
|
||||
|
||||
// Conflict: discard the branch + release worktree, report the conflict, and
|
||||
// STOP draining (the conflicting step is not resolved — the scheduler routes
|
||||
// it to rework on the updated base, then re-enqueues a fresh branch or skips).
|
||||
await this.safeDiscard(ready.branchName, ready.stepIndex);
|
||||
this.pending.delete(this.cursor);
|
||||
outcomes.push({
|
||||
stepIndex: ready.stepIndex,
|
||||
status: "conflict",
|
||||
conflictedFiles: result.conflictedFiles,
|
||||
});
|
||||
break;
|
||||
}
|
||||
return outcomes;
|
||||
}
|
||||
|
||||
/** True once every step index is resolved (integrated or skipped). */
|
||||
isDrained(): boolean {
|
||||
return this.resolved.size >= this.pinnedStepCount;
|
||||
}
|
||||
|
||||
/** Drain-and-release any remaining pending branches (abort/cleanup path). */
|
||||
async discardAllPending(): Promise<void> {
|
||||
for (const { branchName, stepIndex } of this.pending.values()) {
|
||||
await this.safeDiscard(branchName, stepIndex);
|
||||
}
|
||||
this.pending.clear();
|
||||
}
|
||||
|
||||
private advanceCursor(): void {
|
||||
while (this.cursor < this.pinnedStepCount && this.resolved.has(this.cursor)) {
|
||||
this.cursor += 1;
|
||||
}
|
||||
}
|
||||
|
||||
private async safeDiscard(branchName: string, stepIndex: number): Promise<void> {
|
||||
try {
|
||||
await this.gitOps.discardBranch(branchName, stepIndex);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
schedulerLog.warn(`integration discardBranch failed for ${branchName} (step ${stepIndex}): ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
} from "./workflow-graph-branches.js";
|
||||
import {
|
||||
runForeach,
|
||||
type ForeachEnvironment,
|
||||
type WorkflowStepInstancePersistence,
|
||||
} from "./workflow-graph-foreach.js";
|
||||
|
||||
@@ -97,6 +98,24 @@ export interface WorkflowGraphExecutorDeps {
|
||||
* runs (zero behavior change for non-foreach graphs).
|
||||
*/
|
||||
signal?: AbortSignal;
|
||||
/** Step-inversion (KTD-11, U10): per-instance worktree/branch allocation off the
|
||||
* integration base, for `isolation: "worktree"`. Absent → worktree isolation
|
||||
* fails cleanly (shared isolation is unaffected). */
|
||||
allocateInstanceWorktree?: ForeachEnvironment["allocateInstanceWorktree"];
|
||||
/** Step-inversion (KTD-11, U10): resolve the current integration base (main tip)
|
||||
* so reworks land on the updated base. */
|
||||
resolveIntegrationBase?: ForeachEnvironment["resolveIntegrationBase"];
|
||||
/** Step-inversion (KTD-11, U10): ordered-integration git mechanics (rebase /
|
||||
* cherry-pick + conflict detection via merger helpers). */
|
||||
integrationGitOps?: ForeachEnvironment["integrationGitOps"];
|
||||
/** Step-inversion (KTD-11, U10): projection-first integration writes
|
||||
* (updateStep done, then instance row). */
|
||||
integrationProjection?: ForeachEnvironment["integrationProjection"];
|
||||
/** Step-inversion (KTD-11, U10): non-blocking free-semaphore-slot accessor for
|
||||
* parallel scheduling (clamps concurrency without hold-and-wait). */
|
||||
semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"];
|
||||
/** Step-inversion (KTD-11, U10): crash-resume reconciliation hook. */
|
||||
resumeReconcile?: ForeachEnvironment["resumeReconcile"];
|
||||
}
|
||||
|
||||
export interface WorkflowGraphExecutorResult {
|
||||
@@ -248,12 +267,19 @@ export class WorkflowGraphExecutor {
|
||||
runId,
|
||||
steps,
|
||||
context,
|
||||
runTemplateNode: (tNode, sig) =>
|
||||
this.executeNodeWithRetries(tNode, task, settings, context, sig),
|
||||
runTemplateNode: (tNode, sig, contextOverride) =>
|
||||
this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, sig),
|
||||
shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src),
|
||||
persistence: this.deps.stepInstancePersistence,
|
||||
onReworkReset: this.deps.onReworkReset,
|
||||
signal: this.deps.signal,
|
||||
// Worktree isolation + parallel scheduling (KTD-11, U10).
|
||||
allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
|
||||
resolveIntegrationBase: this.deps.resolveIntegrationBase,
|
||||
integrationGitOps: this.deps.integrationGitOps,
|
||||
integrationProjection: this.deps.integrationProjection,
|
||||
semaphoreAvailability: this.deps.semaphoreAvailability,
|
||||
resumeReconcile: this.deps.resumeReconcile,
|
||||
});
|
||||
visitedNodeIds.push(...foreachResult.visitedNodeIds);
|
||||
const result: WorkflowNodeResult = {
|
||||
|
||||
@@ -6,6 +6,11 @@ import {
|
||||
FOREACH_ACTIVE_CONTEXT_KEY,
|
||||
type ForeachActiveContext,
|
||||
} from "./workflow-node-handlers.js";
|
||||
import {
|
||||
IntegrationQueue,
|
||||
type IntegrationGitOps,
|
||||
type IntegrationProjection,
|
||||
} from "./step-integration.js";
|
||||
import { schedulerLog } from "./logger.js";
|
||||
|
||||
/**
|
||||
@@ -41,6 +46,10 @@ import { schedulerLog } from "./logger.js";
|
||||
const DEFAULT_MAX_REWORK_CYCLES = 3;
|
||||
/** Defensive cap mirroring core's validation clamp (KTD-5). */
|
||||
const MAX_REWORK_CYCLES_CAP = 10;
|
||||
/** Default parallel concurrency (KTD-3). */
|
||||
const DEFAULT_CONCURRENCY = 2;
|
||||
/** Hard cap on parallel concurrency (KTD-3). */
|
||||
const CONCURRENCY_CAP = 8;
|
||||
|
||||
/** The foreach node's config shape this module reads (subset of WorkflowForeachConfig). */
|
||||
interface ForeachConfig {
|
||||
@@ -72,12 +81,18 @@ export interface WorkflowStepInstanceState {
|
||||
pinnedStepCount: number;
|
||||
/** Template node id (NOT the materialized instance id) the instance is at. */
|
||||
currentNodeId: string;
|
||||
status: "in-progress" | "completed" | "failed";
|
||||
status: "in-progress" | "awaiting-integration" | "completed" | "failed";
|
||||
baselineSha?: string;
|
||||
checkpointId?: string;
|
||||
reworkCount: number;
|
||||
/** Latest authoritative step-review verdict (KTD-4/KTD-6, U5). */
|
||||
verdict?: "APPROVE" | "REVISE" | "RETHINK" | "UNAVAILABLE";
|
||||
/** Worktree-isolation (KTD-11, U10): the instance's own branch name; null/absent
|
||||
* under shared isolation. */
|
||||
branchName?: string;
|
||||
/** Worktree-isolation (KTD-11, U10): ISO timestamp the branch integrated; absent
|
||||
* until the ordered integration stage lands it. */
|
||||
integratedAt?: string;
|
||||
}
|
||||
|
||||
export interface WorkflowStepInstancePersistence {
|
||||
@@ -121,12 +136,19 @@ export interface ForeachEnvironment {
|
||||
/**
|
||||
* Runs one template node through the executor's executeNodeWithRetries (so
|
||||
* per-node maxRetries still applies inside the sub-walk). The node passed is
|
||||
* the ORIGINAL template node; the executor reads/writes the shared context,
|
||||
* which already carries `foreach:active` for the current instance.
|
||||
* the ORIGINAL template node; the executor reads/writes the supplied context,
|
||||
* which carries `foreach:active` for the current instance.
|
||||
*
|
||||
* `contextOverride` lets a worktree-isolated instance run on its OWN context
|
||||
* object (KTD-11): under parallel scheduling concurrent instances must NOT share
|
||||
* the single `foreach:active` slot, so each gets an isolated context clone. When
|
||||
* omitted the shared `env.context` is used (shared-isolation sequential path —
|
||||
* unchanged behavior).
|
||||
*/
|
||||
runTemplateNode: (
|
||||
node: WorkflowIrNode,
|
||||
signal?: AbortSignal,
|
||||
contextOverride?: Record<string, unknown>,
|
||||
) => Promise<WorkflowNodeResult>;
|
||||
shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean;
|
||||
persistence?: WorkflowStepInstancePersistence;
|
||||
@@ -145,6 +167,63 @@ export interface ForeachEnvironment {
|
||||
) => void | Promise<void>;
|
||||
/** Honored between nodes (existing posture). */
|
||||
signal?: AbortSignal;
|
||||
|
||||
// ── Worktree isolation + parallel scheduling (KTD-11, U10) ────────────────
|
||||
/**
|
||||
* Allocate the instance's OWN worktree + branch off the current integration
|
||||
* base (the task's main branch tip at instance start). Production wires this to
|
||||
* the worktree pool / `createWorktree` with a canonical
|
||||
* `fusion/<task>-step-<i>` branch name; tests inject a fake. Required when
|
||||
* `isolation: "worktree"`; absent under shared isolation. The integration base
|
||||
* is the SAME for sibling instances scheduled together (they branch from the
|
||||
* common tip), and is the UPDATED main tip when an instance re-runs after an
|
||||
* integration-conflict rework.
|
||||
*/
|
||||
allocateInstanceWorktree?: (
|
||||
stepIndex: number,
|
||||
integrationBase: string | undefined,
|
||||
) => Promise<{ worktreePath: string; branchName: string }>;
|
||||
/** Resolve the current integration base (main branch tip). Re-read before each
|
||||
* (re)allocation so a rework lands on the UPDATED base (KTD-11). Optional —
|
||||
* defaults to undefined (the allocator's own default base). */
|
||||
resolveIntegrationBase?: () => Promise<string | undefined>;
|
||||
/** Ordered-integration git mechanics (KTD-11). Required when `isolation:
|
||||
* "worktree"`; the queue uses it to land branches in step order. */
|
||||
integrationGitOps?: IntegrationGitOps;
|
||||
/**
|
||||
* Projection + instance-row writes the integration stage performs on a clean
|
||||
* integration (KTD-7 projection-first ordering). `markStepDone` flips the step
|
||||
* `done` via `updateStep(source:"graph")`; `markInstanceIntegrated` flips the
|
||||
* row to `completed`/`integratedAt`. Required when `isolation: "worktree"`.
|
||||
*/
|
||||
integrationProjection?: IntegrationProjection;
|
||||
/**
|
||||
* Non-blocking semaphore availability accessor for parallel scheduling (KTD-11):
|
||||
* how many slots are free for IMMEDIATE acquisition right now. The scheduler
|
||||
* runs up to `min(concurrency, availability)` instances concurrently and
|
||||
* degrades to fewer/sequential under contention — it NEVER hold-and-waits on
|
||||
* slots while blocking integration (each instance acquires its own lease inside
|
||||
* the step-execute seam like a normal session). Defaults to "unbounded" (the
|
||||
* concurrency cap governs) when absent. A returned value ≤ 0 degrades to 1
|
||||
* (always make forward progress; never deadlock).
|
||||
*/
|
||||
semaphoreAvailability?: () => number;
|
||||
/**
|
||||
* Crash-resume reconciliation hook (KTD-11, U10). Before scheduling, the
|
||||
* worktree scheduler reconciles each step instance against git/persistence truth:
|
||||
* - integrated (row `completed`/`integratedAt`) → mark INTEGRATED (skip);
|
||||
* - branch exists but not integrated (`awaiting-integration`) → RE-ENTER the
|
||||
* integration queue (the branch is already built);
|
||||
* - branch missing → RE-RUN the instance.
|
||||
* Returns a per-step disposition; absent → all instances run fresh (cold start).
|
||||
* The full self-healing sweep across runs is out of scope (handoff). `pinned` is
|
||||
* the count this expansion pinned so the hook can reject a `pin-mismatch`.
|
||||
*/
|
||||
resumeReconcile?: (
|
||||
pinned: number,
|
||||
) =>
|
||||
| Promise<Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>>
|
||||
| Array<{ stepIndex: number; disposition: "integrated" | "reintegrate" | "rerun"; branchName?: string }>;
|
||||
}
|
||||
|
||||
export interface ForeachRunResult {
|
||||
@@ -168,6 +247,8 @@ function resolveForeachConfig(node: WorkflowIrNode): {
|
||||
template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] };
|
||||
maxReworkCycles: number;
|
||||
mode: "sequential" | "parallel";
|
||||
isolation: "shared" | "worktree";
|
||||
concurrency: number;
|
||||
} {
|
||||
const cfg = (node.config ?? {}) as ForeachConfig;
|
||||
const template = cfg.template;
|
||||
@@ -177,7 +258,57 @@ function resolveForeachConfig(node: WorkflowIrNode): {
|
||||
const raw = typeof cfg.maxReworkCycles === "number" ? cfg.maxReworkCycles : DEFAULT_MAX_REWORK_CYCLES;
|
||||
const maxReworkCycles = Math.max(1, Math.min(MAX_REWORK_CYCLES_CAP, Math.floor(raw)));
|
||||
const mode = cfg.mode === "parallel" ? "parallel" : "sequential";
|
||||
return { template, maxReworkCycles, mode };
|
||||
// Default isolation: worktree for parallel mode, shared for sequential (KTD-3).
|
||||
// (Core validation rejects parallel+shared; this default mirrors that intent.)
|
||||
const isolation: "shared" | "worktree" =
|
||||
cfg.isolation === "worktree"
|
||||
? "worktree"
|
||||
: cfg.isolation === "shared"
|
||||
? "shared"
|
||||
: mode === "parallel"
|
||||
? "worktree"
|
||||
: "shared";
|
||||
const rawConc =
|
||||
typeof cfg.concurrency === "number" && Number.isFinite(cfg.concurrency)
|
||||
? Math.floor(cfg.concurrency)
|
||||
: DEFAULT_CONCURRENCY;
|
||||
// Concurrency only meaningful in parallel mode; sequential pins to 1 (KTD-3).
|
||||
const concurrency = mode === "parallel" ? Math.max(1, Math.min(CONCURRENCY_CAP, rawConc)) : 1;
|
||||
return { template, maxReworkCycles, mode, isolation, concurrency };
|
||||
}
|
||||
|
||||
/** The 0-indexed predecessor step indices instance `stepIndex` depends on. A step
|
||||
* with no annotation implicitly depends on the previous step (KTD-3), so an
|
||||
* unannotated plan is fully sequential regardless of mode. */
|
||||
function resolveDependsOn(steps: TaskStep[], stepIndex: number): number[] {
|
||||
const deps = steps[stepIndex]?.dependsOn;
|
||||
if (Array.isArray(deps) && deps.length > 0) return deps;
|
||||
return stepIndex > 0 ? [stepIndex - 1] : [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate the dependency DAG at expansion (KTD-3): every dependsOn index must be
|
||||
* in range and strictly earlier (a step may only depend on lower indices), and
|
||||
* the graph must be acyclic. Returns a refusal reason on violation, else null.
|
||||
* Because dependsOn entries reference lower indices only, a forward-reference or
|
||||
* self-reference is the cycle signature we reject.
|
||||
*/
|
||||
function validateDependencyDag(steps: TaskStep[]): string | null {
|
||||
for (let i = 0; i < steps.length; i++) {
|
||||
const deps = steps[i]?.dependsOn;
|
||||
if (!Array.isArray(deps)) continue;
|
||||
for (const d of deps) {
|
||||
if (!Number.isInteger(d) || d < 0 || d >= steps.length) {
|
||||
return `step ${i} depends on out-of-range step ${d}`;
|
||||
}
|
||||
if (d >= i) {
|
||||
// A dependency on an equal/later index is the only way to form a cycle
|
||||
// when edges always point to lower indices; reject it as an audited cycle.
|
||||
return `dependency cycle: step ${i} depends on step ${d} (>= itself)`;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Find the single template entry node (no non-rework incoming edge). */
|
||||
@@ -200,33 +331,21 @@ function findTemplateEntry(
|
||||
return entries[0];
|
||||
}
|
||||
|
||||
/**
|
||||
* Expand a foreach node and run its instances sequentially in step order.
|
||||
* Returns the foreach node's aggregate outcome (KTD-3).
|
||||
*/
|
||||
export async function runForeach(
|
||||
/** Compiled template state shared across instances (built once per expansion). */
|
||||
interface TemplatePlan {
|
||||
templateById: Map<string, WorkflowIrNode>;
|
||||
templateOutgoing: Map<string, WorkflowIrEdge[]>;
|
||||
entry: WorkflowIrNode;
|
||||
/** Whether the template routes an explicit `outcome:integration-conflict` edge
|
||||
* from any node (overrides the default rework routing — KTD-11). */
|
||||
hasExplicitIntegrationConflictEdge: boolean;
|
||||
templateHasStepReview: boolean;
|
||||
}
|
||||
|
||||
function compileTemplate(
|
||||
foreachNode: WorkflowIrNode,
|
||||
env: ForeachEnvironment,
|
||||
): Promise<ForeachRunResult> {
|
||||
const { template, maxReworkCycles, mode } = resolveForeachConfig(foreachNode);
|
||||
|
||||
// U3 scope guard: parallel mode is U10. Fail cleanly with a routable outcome
|
||||
// rather than silently running it as sequential.
|
||||
if (mode === "parallel") {
|
||||
return {
|
||||
outcome: "failure",
|
||||
value: "parallel-not-wired",
|
||||
visitedNodeIds: [],
|
||||
};
|
||||
}
|
||||
|
||||
// Pin the count at expansion (KTD-3). Zero steps → success edge (no instances).
|
||||
const pinnedStepCount = env.steps.length;
|
||||
const visitedNodeIds: string[] = [];
|
||||
if (pinnedStepCount === 0) {
|
||||
return { outcome: "success", visitedNodeIds };
|
||||
}
|
||||
|
||||
template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] },
|
||||
): TemplatePlan {
|
||||
const templateById = new Map(template.nodes.map((n) => [n.id, n]));
|
||||
const templateOutgoing = new Map<string, WorkflowIrEdge[]>();
|
||||
for (const edge of template.edges) {
|
||||
@@ -235,14 +354,52 @@ export async function runForeach(
|
||||
templateOutgoing.set(edge.from, list);
|
||||
}
|
||||
const entry = findTemplateEntry(template.nodes, template.edges, foreachNode.id);
|
||||
|
||||
// Single-authority done-marking (U6/KTD-4): when the template contains a
|
||||
// step-review node, step-execute SUCCESS must leave the step in-progress and the
|
||||
// review's APPROVE marks it done. Computed once and threaded into each instance.
|
||||
const hasExplicitIntegrationConflictEdge = template.edges.some(
|
||||
(e) => e.condition === "outcome:integration-conflict",
|
||||
);
|
||||
const templateHasStepReview = template.nodes.some((n) => n.kind === "step-review");
|
||||
return { templateById, templateOutgoing, entry, hasExplicitIntegrationConflictEdge, templateHasStepReview };
|
||||
}
|
||||
|
||||
// Sequential + shared: a runnable-set loop with concurrency 1 (U10 extends this
|
||||
// to parallel/worktree). Instances run strictly in step order.
|
||||
/**
|
||||
* Expand a foreach node and run its instances. Returns the foreach node's
|
||||
* aggregate outcome (KTD-3). Dispatches on the two orthogonal axes (KTD-11):
|
||||
* - shared isolation (default sequential): the unchanged step-order loop —
|
||||
* work lands in the task's main worktree, done at step completion;
|
||||
* - worktree isolation (sequential OR parallel): per-instance branches off the
|
||||
* integration base, dependency-aware scheduling up to `min(concurrency, free
|
||||
* semaphore slots)`, ordered integration in step order, integration-conflict
|
||||
* routed as rework on the updated base.
|
||||
*/
|
||||
export async function runForeach(
|
||||
foreachNode: WorkflowIrNode,
|
||||
env: ForeachEnvironment,
|
||||
): Promise<ForeachRunResult> {
|
||||
const config = resolveForeachConfig(foreachNode);
|
||||
|
||||
// Pin the count at expansion (KTD-3). Zero steps → success edge (no instances).
|
||||
const pinnedStepCount = env.steps.length;
|
||||
const visitedNodeIds: string[] = [];
|
||||
if (pinnedStepCount === 0) {
|
||||
return { outcome: "success", visitedNodeIds };
|
||||
}
|
||||
|
||||
// Dependency-cycle / out-of-range rejection at expansion (KTD-3, audited).
|
||||
const dagViolation = validateDependencyDag(env.steps);
|
||||
if (dagViolation) {
|
||||
schedulerLog.warn(
|
||||
`foreach ${foreachNode.id} for task ${env.task.id}: ${dagViolation} — failing expansion (dependency-cycle)`,
|
||||
);
|
||||
return { outcome: "failure", value: "dependency-cycle", visitedNodeIds };
|
||||
}
|
||||
|
||||
const plan = compileTemplate(foreachNode, config.template);
|
||||
|
||||
if (config.isolation === "worktree") {
|
||||
return runForeachWorktree(foreachNode, env, config, plan, pinnedStepCount, visitedNodeIds);
|
||||
}
|
||||
|
||||
// ── shared isolation (default sequential) — UNCHANGED U3 behavior ──────────
|
||||
for (let stepIndex = 0; stepIndex < pinnedStepCount; stepIndex++) {
|
||||
if (env.signal?.aborted) {
|
||||
return { outcome: "failure", value: "aborted", visitedNodeIds };
|
||||
@@ -252,13 +409,13 @@ export async function runForeach(
|
||||
foreachNode,
|
||||
stepIndex,
|
||||
pinnedStepCount,
|
||||
entry,
|
||||
templateById,
|
||||
templateOutgoing,
|
||||
maxReworkCycles,
|
||||
plan.entry,
|
||||
plan.templateById,
|
||||
plan.templateOutgoing,
|
||||
config.maxReworkCycles,
|
||||
env,
|
||||
visitedNodeIds,
|
||||
templateHasStepReview,
|
||||
plan.templateHasStepReview,
|
||||
);
|
||||
|
||||
if (instanceResult.outcome === "failure") {
|
||||
@@ -275,6 +432,335 @@ export async function runForeach(
|
||||
return { outcome: "success", visitedNodeIds };
|
||||
}
|
||||
|
||||
// ── Worktree isolation + dependency scheduler + ordered integration (KTD-11) ──
|
||||
|
||||
type InstanceState = "pending" | "running" | "awaiting-integration" | "integrated" | "failed";
|
||||
|
||||
interface WorktreeInstance {
|
||||
stepIndex: number;
|
||||
state: InstanceState;
|
||||
/** Per-instance rework budget — survives integration-conflict re-executions. */
|
||||
reworkBudget: number;
|
||||
reworkCount: number;
|
||||
/** The instance's allocated branch (set on each run). */
|
||||
branchName?: string;
|
||||
worktreePath?: string;
|
||||
baselineSha?: string;
|
||||
checkpointId?: string;
|
||||
verdict?: ForeachActiveContext["verdict"];
|
||||
}
|
||||
|
||||
/**
|
||||
* Worktree-isolation foreach (KTD-11, U10): per-instance branches off the
|
||||
* integration base, dependency-aware scheduling up to `min(concurrency, free
|
||||
* semaphore slots)`, ordered integration in step order, integration-conflict
|
||||
* routed as rework on the updated base.
|
||||
*
|
||||
* Scheduling: an instance is runnable when all of its `dependsOn` steps are
|
||||
* INTEGRATED (not merely completed). The runnable set runs concurrently up to the
|
||||
* clamped slot count; each instance acquires its own semaphore lease inside the
|
||||
* step-execute seam (we only READ availability non-blockingly to decide how many
|
||||
* to launch — never hold-and-wait, so a starved semaphore degrades to sequential
|
||||
* without deadlock). Completion enqueues the branch as `awaiting-integration`;
|
||||
* after each batch the integration queue drains in step order.
|
||||
*/
|
||||
async function runForeachWorktree(
|
||||
foreachNode: WorkflowIrNode,
|
||||
env: ForeachEnvironment,
|
||||
config: ReturnType<typeof resolveForeachConfig>,
|
||||
plan: TemplatePlan,
|
||||
pinnedStepCount: number,
|
||||
visitedNodeIds: string[],
|
||||
): Promise<ForeachRunResult> {
|
||||
if (!env.allocateInstanceWorktree || !env.integrationGitOps || !env.integrationProjection) {
|
||||
// Worktree isolation requires the full wiring; fail cleanly (routable) rather
|
||||
// than silently running shared-mode physics.
|
||||
return { outcome: "failure", value: "worktree-isolation-unwired", visitedNodeIds };
|
||||
}
|
||||
|
||||
const instances: WorktreeInstance[] = Array.from({ length: pinnedStepCount }, (_, i) => ({
|
||||
stepIndex: i,
|
||||
state: "pending",
|
||||
reworkBudget: config.maxReworkCycles,
|
||||
reworkCount: 0,
|
||||
}));
|
||||
|
||||
const queue = new IntegrationQueue(
|
||||
env.integrationGitOps,
|
||||
env.integrationProjection,
|
||||
pinnedStepCount,
|
||||
);
|
||||
|
||||
// Crash-resume reconciliation (KTD-11): integrated → skip; branch-exists → re-enter
|
||||
// the integration queue; branch-missing → re-run. Cold start (no hook) runs fresh.
|
||||
if (env.resumeReconcile) {
|
||||
try {
|
||||
const dispositions = await env.resumeReconcile(pinnedStepCount);
|
||||
for (const d of dispositions) {
|
||||
const inst = instances[d.stepIndex];
|
||||
if (!inst) continue;
|
||||
if (d.disposition === "integrated") {
|
||||
inst.state = "integrated";
|
||||
} else if (d.disposition === "reintegrate" && d.branchName) {
|
||||
// Branch already built — enqueue for ordered integration without re-running.
|
||||
inst.state = "awaiting-integration";
|
||||
inst.branchName = d.branchName;
|
||||
queue.enqueue(d.stepIndex, d.branchName);
|
||||
}
|
||||
// "rerun" leaves the instance pending (default).
|
||||
}
|
||||
} catch (err) {
|
||||
schedulerLog.warn(
|
||||
`foreach ${foreachNode.id} for task ${env.task.id}: resume reconcile failed, running cold: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const isIntegrated = (i: number): boolean => instances[i].state === "integrated";
|
||||
const depsIntegrated = (i: number): boolean =>
|
||||
resolveDependsOn(env.steps, i).every((d) => isIntegrated(d));
|
||||
|
||||
// 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.).
|
||||
const runOneInstance = async (inst: WorktreeInstance): Promise<{ outcome: WorkflowNodeOutcome; value?: string }> => {
|
||||
const integrationBase = env.resolveIntegrationBase
|
||||
? await env.resolveIntegrationBase().catch(() => undefined)
|
||||
: undefined;
|
||||
let allocation: { worktreePath: string; branchName: string };
|
||||
try {
|
||||
allocation = await env.allocateInstanceWorktree!(inst.stepIndex, integrationBase);
|
||||
} catch (err) {
|
||||
schedulerLog.warn(
|
||||
`foreach ${foreachNode.id} step ${inst.stepIndex}: worktree allocation failed: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
return { outcome: "failure", value: "worktree-alloc-failed" };
|
||||
}
|
||||
inst.branchName = allocation.branchName;
|
||||
inst.worktreePath = allocation.worktreePath;
|
||||
|
||||
return runWorktreeInstanceSubWalk(foreachNode, env, plan, pinnedStepCount, visitedNodeIds, inst);
|
||||
};
|
||||
|
||||
// Schedule + integrate loop.
|
||||
for (;;) {
|
||||
if (env.signal?.aborted) {
|
||||
await queue.discardAllPending();
|
||||
return { outcome: "failure", value: "aborted", visitedNodeIds };
|
||||
}
|
||||
|
||||
// Runnable = pending instances whose deps are all integrated.
|
||||
const runnable = instances.filter((i) => i.state === "pending" && depsIntegrated(i.stepIndex));
|
||||
const running = instances.filter((i) => i.state === "running").length;
|
||||
|
||||
if (runnable.length > 0) {
|
||||
// Clamp concurrency by free semaphore slots (non-blocking, KTD-11). Degrade
|
||||
// to at least 1 so we never deadlock waiting for slots we don't hold: a
|
||||
// starved semaphore (availability ≤ 0) still launches one at a time.
|
||||
const rawAvail = env.semaphoreAvailability ? env.semaphoreAvailability() : config.concurrency;
|
||||
const avail = Math.max(1, rawAvail);
|
||||
let slots = Math.min(config.concurrency, avail) - running;
|
||||
// Force forward progress when nothing is running (never block on slots held).
|
||||
if (slots <= 0 && running === 0) slots = 1;
|
||||
const toLaunch = slots > 0 ? runnable.slice(0, slots) : [];
|
||||
|
||||
if (toLaunch.length > 0) {
|
||||
for (const inst of toLaunch) inst.state = "running";
|
||||
await Promise.all(
|
||||
toLaunch.map(async (inst) => {
|
||||
const result = await runOneInstance(inst);
|
||||
if (result.outcome === "success") {
|
||||
inst.state = "awaiting-integration";
|
||||
queue.enqueue(inst.stepIndex, inst.branchName!);
|
||||
} 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.
|
||||
queue.skip(inst.stepIndex);
|
||||
(inst as WorktreeInstance & { failValue?: string }).failValue = result.value;
|
||||
}
|
||||
}),
|
||||
);
|
||||
continue; // re-evaluate runnable set + drain after the batch.
|
||||
}
|
||||
}
|
||||
|
||||
// Drain the integration queue in step order; route conflicts to rework.
|
||||
const outcomes = await queue.drain();
|
||||
const progressed = outcomes.length > 0;
|
||||
for (const outcome of outcomes) {
|
||||
const inst = instances[outcome.stepIndex];
|
||||
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.
|
||||
if (inst.reworkBudget <= 0) {
|
||||
inst.state = "failed";
|
||||
queue.skip(inst.stepIndex);
|
||||
(inst as WorktreeInstance & { failValue?: string }).failValue = "rework-exhausted";
|
||||
} else {
|
||||
inst.reworkBudget -= 1;
|
||||
inst.reworkCount += 1;
|
||||
inst.state = "pending"; // re-enter scheduling; re-allocates off updated base.
|
||||
// Default routing IS the rework path (re-execute the step on the updated
|
||||
// base, budget-counted — KTD-11). When the template authors an explicit
|
||||
// `outcome:integration-conflict` edge, the re-run surfaces the conflict
|
||||
// signal in the instance context so the author's node can branch on it
|
||||
// (the edge overrides the implicit "from entry" rework).
|
||||
(inst as WorktreeInstance & { lastIntegrationConflict?: boolean }).lastIntegrationConflict =
|
||||
plan.hasExplicitIntegrationConflictEdge;
|
||||
schedulerLog.log(
|
||||
`foreach ${foreachNode.id} step ${inst.stepIndex}: integration-conflict — reworking on updated base (budget left ${inst.reworkBudget})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Terminal conditions.
|
||||
const failed = instances.find((i) => i.state === "failed");
|
||||
if (failed) {
|
||||
await queue.discardAllPending();
|
||||
const value = (failed as WorktreeInstance & { failValue?: string }).failValue ?? "instance-failed";
|
||||
return { outcome: "failure", value, visitedNodeIds };
|
||||
}
|
||||
if (instances.every((i) => i.state === "integrated")) {
|
||||
return { outcome: "success", visitedNodeIds };
|
||||
}
|
||||
|
||||
// No progress and nothing runnable/running → stuck (shouldn't happen with a
|
||||
// valid DAG, but guard against a livelock).
|
||||
const anyActive = instances.some((i) => i.state === "running" || i.state === "awaiting-integration");
|
||||
const anyRunnable = instances.some((i) => i.state === "pending" && depsIntegrated(i.stepIndex));
|
||||
if (!progressed && !anyActive && !anyRunnable) {
|
||||
await queue.discardAllPending();
|
||||
return { outcome: "failure", value: "scheduler-stuck", visitedNodeIds };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one worktree-isolated instance's iterative sub-walk in an ISOLATED context
|
||||
* (its own `foreach:active` slot — required for concurrency), against the
|
||||
* instance's own worktree/branch. Returns success (sub-walk reached the exit;
|
||||
* the caller enqueues the branch for ordered integration — does NOT mark done
|
||||
* here, KTD-11) or failure (rework-exhausted / aborted / node failure).
|
||||
*
|
||||
* RETHINK under worktree isolation is branch-scoped (KTD-11): `onReworkReset`
|
||||
* resets the instance's OWN branch (the executor wires it to resetStepToBaseline
|
||||
* against this instance's worktree), so the blast-radius guard is structural.
|
||||
*/
|
||||
async function runWorktreeInstanceSubWalk(
|
||||
foreachNode: WorkflowIrNode,
|
||||
env: ForeachEnvironment,
|
||||
plan: TemplatePlan,
|
||||
pinnedStepCount: number,
|
||||
visitedNodeIds: string[],
|
||||
inst: WorktreeInstance,
|
||||
): Promise<{ outcome: WorkflowNodeOutcome; value?: string }> {
|
||||
const stepIndex = inst.stepIndex;
|
||||
// Isolated per-instance context (NOT the shared env.context) so concurrent
|
||||
// instances never collide on the `foreach:active` slot. Seeded from the shared
|
||||
// context so handlers still see prior walk context (read-only-ish).
|
||||
const instanceContext: Record<string, unknown> = { ...env.context };
|
||||
const active: ForeachActiveContext = {
|
||||
foreachNodeId: foreachNode.id,
|
||||
stepIndex,
|
||||
instanceId: `${foreachNode.id}#${stepIndex}`,
|
||||
deferDoneToReview: plan.templateHasStepReview,
|
||||
worktreePath: inst.worktreePath,
|
||||
branchName: inst.branchName,
|
||||
baselineSha: inst.baselineSha,
|
||||
checkpointId: inst.checkpointId,
|
||||
verdict: inst.verdict,
|
||||
};
|
||||
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;
|
||||
|
||||
const persist = (status: WorkflowStepInstanceState["status"], currentNodeId: string): Promise<void> =>
|
||||
persistInstanceState(env.persistence, {
|
||||
taskId: env.task.id,
|
||||
runId: env.runId,
|
||||
foreachNodeId: foreachNode.id,
|
||||
stepIndex,
|
||||
pinnedStepCount,
|
||||
currentNodeId,
|
||||
status,
|
||||
baselineSha: active.baselineSha,
|
||||
checkpointId: active.checkpointId,
|
||||
reworkCount: inst.reworkCount,
|
||||
verdict: active.verdict,
|
||||
branchName: inst.branchName,
|
||||
});
|
||||
|
||||
await persist("in-progress", plan.entry.id);
|
||||
|
||||
let currentId = plan.entry.id;
|
||||
let lastResult: WorkflowNodeResult = { outcome: "success" };
|
||||
|
||||
for (;;) {
|
||||
if (env.signal?.aborted) {
|
||||
await persist("failed", currentId);
|
||||
return { outcome: "failure", value: "aborted" };
|
||||
}
|
||||
|
||||
const node = plan.templateById.get(currentId);
|
||||
if (!node) throw new WorkflowIrError(`Unknown foreach template node: ${currentId}`);
|
||||
|
||||
visitedNodeIds.push(instanceNodeId(foreachNode.id, stepIndex, currentId));
|
||||
|
||||
lastResult = await env.runTemplateNode(node, env.signal, instanceContext);
|
||||
syncActiveFromContext(instanceContext, active);
|
||||
// Persist captured baseline/checkpoint back onto the instance (survives rework).
|
||||
inst.baselineSha = active.baselineSha;
|
||||
inst.checkpointId = active.checkpointId;
|
||||
inst.verdict = active.verdict;
|
||||
|
||||
if (lastResult.outcome === "failure") {
|
||||
await persist("failed", currentId);
|
||||
return { outcome: "failure", value: lastResult.value };
|
||||
}
|
||||
|
||||
const next = chooseNextEdge(currentId, plan.templateOutgoing, lastResult, env.shouldTraverseEdge);
|
||||
if (!next) {
|
||||
// Sub-walk exit — work complete on the branch; AWAIT INTEGRATION (not done).
|
||||
await persist("awaiting-integration", currentId);
|
||||
return { outcome: "success" };
|
||||
}
|
||||
|
||||
if (next.kind === "rework") {
|
||||
if (inst.reworkBudget <= 0) {
|
||||
await persist("failed", currentId);
|
||||
return { outcome: "failure", value: "rework-exhausted" };
|
||||
}
|
||||
inst.reworkBudget -= 1;
|
||||
inst.reworkCount += 1;
|
||||
|
||||
if (lastResult.value === "rethink" && env.onReworkReset) {
|
||||
try {
|
||||
// Branch-scoped RETHINK: resets THIS instance's branch only (KTD-11).
|
||||
await env.onReworkReset(active, "rethink");
|
||||
syncActiveFromContext(instanceContext, active);
|
||||
inst.baselineSha = active.baselineSha;
|
||||
inst.checkpointId = active.checkpointId;
|
||||
} catch (err) {
|
||||
schedulerLog.warn(
|
||||
`onReworkReset failed for task ${env.task.id} foreach ${foreachNode.id} step ${stepIndex}: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
await persist("in-progress", next.to);
|
||||
}
|
||||
|
||||
currentId = next.to;
|
||||
}
|
||||
}
|
||||
|
||||
interface InstanceResult {
|
||||
outcome: WorkflowNodeOutcome;
|
||||
value?: string;
|
||||
|
||||
@@ -14,7 +14,7 @@ import type {
|
||||
WorkflowBranchProgress,
|
||||
WorkflowBranchSemaphore,
|
||||
} from "./workflow-graph-branches.js";
|
||||
import type { WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js";
|
||||
import type { ForeachEnvironment, WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js";
|
||||
// (Both types are also used as values in the side-effect tracking wrappers below.)
|
||||
|
||||
/**
|
||||
@@ -69,6 +69,14 @@ export interface WorkflowGraphTaskRunnerDeps {
|
||||
/** Step-inversion (U14, KTD-15): `code` node runner. Additive; a workflow with
|
||||
* no code node never invokes it. */
|
||||
runCode?: CodeNodeRunner;
|
||||
/** Step-inversion (KTD-11, U10): worktree-isolation + parallel-scheduling deps.
|
||||
* Additive; a shared-isolation foreach never invokes them. */
|
||||
allocateInstanceWorktree?: ForeachEnvironment["allocateInstanceWorktree"];
|
||||
resolveIntegrationBase?: ForeachEnvironment["resolveIntegrationBase"];
|
||||
integrationGitOps?: ForeachEnvironment["integrationGitOps"];
|
||||
integrationProjection?: ForeachEnvironment["integrationProjection"];
|
||||
semaphoreAvailability?: ForeachEnvironment["semaphoreAvailability"];
|
||||
resumeReconcile?: ForeachEnvironment["resumeReconcile"];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -174,6 +182,13 @@ export class WorkflowGraphTaskRunner {
|
||||
onReworkReset: this.deps.onReworkReset,
|
||||
parseStepsDeps: this.deps.parseStepsDeps,
|
||||
runCode: this.deps.runCode,
|
||||
// Step-inversion (KTD-11, U10): worktree isolation + parallel scheduling.
|
||||
allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
|
||||
resolveIntegrationBase: this.deps.resolveIntegrationBase,
|
||||
integrationGitOps: this.deps.integrationGitOps,
|
||||
integrationProjection: this.deps.integrationProjection,
|
||||
semaphoreAvailability: this.deps.semaphoreAvailability,
|
||||
resumeReconcile: this.deps.resumeReconcile,
|
||||
runId: `${task.id}:${definition.id}`,
|
||||
onBranchProgress: (progress) => {
|
||||
this.branchProgress.set(progress.branchId, progress);
|
||||
|
||||
@@ -99,6 +99,20 @@ export interface ForeachActiveContext {
|
||||
* foreach sub-walk sets this at instance entry; the step-execute seam reads it.
|
||||
*/
|
||||
deferDoneToReview?: boolean;
|
||||
/**
|
||||
* Worktree-isolation (KTD-11, U10): the instance's OWN worktree path, branched
|
||||
* off the integration base. Set by the foreach sub-walk at instance entry under
|
||||
* `isolation: "worktree"`; the step-execute / step-review / RETHINK seams run
|
||||
* against THIS path instead of the task's main worktree. Absent under
|
||||
* `isolation: "shared"` (work lands directly in the main worktree). The file-scope
|
||||
* guard still fires for anything the instance session commits in this worktree
|
||||
* (the session machinery is unchanged — see executor stepExecute seam).
|
||||
*/
|
||||
worktreePath?: string;
|
||||
/** Worktree-isolation (KTD-11, U10): the instance's OWN branch name (e.g.
|
||||
* `fusion/<task>-step-<i>`). Set with {@link worktreePath}; the ordered
|
||||
* integration stage lands this branch onto the task's main branch. */
|
||||
branchName?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -33,6 +33,17 @@ export function canonicalFusionBranchName(taskId: string): string {
|
||||
return `fusion/${taskId.toLowerCase()}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Canonical per-instance branch name for a worktree-isolated foreach step
|
||||
* (step-inversion KTD-11, U10): `fusion/<task>-step-<i>`. Deterministic from the
|
||||
* task id + 0-based step index so crash-resume can reconstruct the branch name
|
||||
* (and probe its existence) without persisting it separately — though the
|
||||
* instance row also carries `branchName` for the integration/reconcile path.
|
||||
*/
|
||||
export function canonicalStepInstanceBranchName(taskId: string, stepIndex: number): string {
|
||||
return `${canonicalFusionBranchName(taskId)}-step-${stepIndex}`;
|
||||
}
|
||||
|
||||
export function resolveTaskWorkingBranch(task: Pick<Task, "id" | "branch" | "branchContext">): string {
|
||||
if (task.branchContext?.assignmentMode === "shared") {
|
||||
return canonicalFusionBranchName(task.id);
|
||||
|
||||
Reference in New Issue
Block a user