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

perf(executor): recover approved steps on engine restart

Browse Source 
When the engine restarts mid-step, an in-progress step may have already
passed plan + code review but not yet been flipped to done by the agent's
next task_update call. Previously, the next executor pass re-entered the
step and replayed both reviews — measured at 5-20 min of pure waste per
restart (observed in FN-2215 Step 1 and FN-2207 Step 6).

recoverApprovedStepsOnResume scans the task log for any in-progress step
whose most recent "code review Step N: APPROVE" entry is newer than its
most recent "Step N → pending" transition, and marks those steps done
before execute() runs. Safely skips steps that were reset after approval
(e.g. by a workflow revision) or only received REVISE verdicts.

Called from both the engine-restart path (resumeOrphaned) and the
unpause path, matching the two places the task log shows as vulnerable
to this race.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-21 09:02:42 -07:00
parent 322bda2da1
commit 32c551c5f4
2 changed files with 193 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -608,6 +608,7 @@ export class TaskExecutor {
try {
await this.clearResumeFailureState(task);
await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.currentRunContext);
await this.recoverApprovedStepsOnResume(task.id);
} catch (clearErr) {
executorLog.warn(`${task.id} clearResumeFailureState failed during unpause: ${clearErr instanceof Error ? clearErr.message : String(clearErr)}`);
}
@@ -999,6 +1000,7 @@ export class TaskExecutor {
try {
await this.clearResumeFailureState(task);
await this.store.logEntry(task.id, "Resumed after engine restart");
await this.recoverApprovedStepsOnResume(task.id);
} catch (err) {
executorLog.error(`Failed to write resume log for ${task.id}:`, err);
}
@@ -4204,6 +4206,74 @@ and show an appropriate message to the user.\`
}
}
/**
* When the engine restarts mid-step, an `in-progress` step may have already
* passed its code review (log: `code review Step N: APPROVE`) but not yet
* been flipped to `done` by the agent's next `task_update` call. Without
* intervention, the next executor pass re-enters the step and replays plan
* + code review, which we've measured at 5–20 min of pure waste per restart.
*
* This reconciler scans the task log for any in-progress step whose most
* recent approved code review is newer than its most recent `→ pending`
* transition, and marks those steps `done`. Subsequent resume logic then
* advances to the next actually-pending step.
*/
private async recoverApprovedStepsOnResume(taskId: string): Promise<void> {
let detail: TaskDetail;
try {
detail = await this.store.getTask(taskId);
} catch (err) {
executorLog.warn(`${taskId}: recoverApprovedStepsOnResume getTask failed: ${err instanceof Error ? err.message : String(err)}`);
return;
}
const log = detail.log ?? [];
if (log.length === 0) return;
let recovered = 0;
for (let i = 0; i < detail.steps.length; i++) {
if (detail.steps[i].status !== "in-progress") continue;
let lastPendingAt = -1;
let lastApproveAt = -1;
const stepName = detail.steps[i].name;
// Matches "Step 3 (My Step) → pending"; name is user-controlled, so match
// on prefix rather than a regex built from the name.
const transitionPrefix = `Step ${i} (${stepName}) → `;
const approvePrefix = `code review Step ${i}:`;
for (let j = 0; j < log.length; j++) {
const action = log[j].action || "";
if (action.startsWith(transitionPrefix)) {
const status = action.slice(transitionPrefix.length).trim();
if (status === "pending") lastPendingAt = j;
} else if (action.startsWith(approvePrefix) && action.includes("APPROVE")) {
lastApproveAt = j;
}
}
if (lastApproveAt > lastPendingAt) {
executorLog.log(
`${taskId}: step ${i} ("${stepName}") already has an approved code review — marking done on resume (skipping review replay)`,
);
try {
await this.store.logEntry(
taskId,
`Step ${i} (${stepName}) recovered as done on resume — code review had already approved before the engine stopped`,
);
await this.store.updateStep(taskId, i, "done");
recovered++;
} catch (err) {
executorLog.warn(
`${taskId}: failed to recover step ${i} on resume: ${err instanceof Error ? err.message : String(err)}`,
);
}
}
}
if (recovered > 0) {
executorLog.log(`${taskId}: recovered ${recovered} approved step(s) on resume`);
}
}
/**
* Check whether the task's branch has any unique commits compared to main.
* If the branch has no unique commits and the task has steps marked done,

123
packages/engine/src/restart.integration.test.ts
View File

@@ -375,6 +375,129 @@ describe("In-progress task resume after restart", () => {
expect(initCalls).toHaveLength(0);
});
it("recovers a step whose code review approved before the engine stopped (no review replay)", async () => {
const store = createMockStore();
const steps: StepStatus[] = ["done", "in-progress", "pending"];
const task = makeTask("FN-1701", "in-progress", {
steps: makeSteps(...steps),
currentStep: 1,
});
// Log order mirrors what we saw in FN-2215: plan review APPROVE → impl →
// code review requested → code review APPROVE → (engine stops before
// step status flips to done).
const detail = makeTaskDetail("FN-1701", "in-progress", {
steps: makeSteps(...steps),
currentStep: 1,
log: [
{ timestamp: "2026-04-21T00:00:00.000Z", action: "Step 1 (Step 1) → in-progress" },
{ timestamp: "2026-04-21T00:00:01.000Z", action: "plan review Step 1: APPROVE" },
{ timestamp: "2026-04-21T00:00:02.000Z", action: "code review requested for Step 1 (Step 1)" },
{ timestamp: "2026-04-21T00:00:03.000Z", action: "code review Step 1: APPROVE" },
],
});
store.listTasks.mockResolvedValue([task]);
store.getTask.mockResolvedValue(detail);
mockAgentSuccess();
const executor = new TaskExecutor(store, "/tmp/test");
await executor.resumeOrphaned();
await new Promise((r) => setTimeout(r, 50));
// The step should have been flipped to done *before* execute ran.
expect(store.updateStep).toHaveBeenCalledWith("FN-1701", 1, "done");
// A recovery log entry should have been written explaining the flip.
expect(store.logEntry).toHaveBeenCalledWith(
"FN-1701",
expect.stringContaining("recovered as done on resume"),
);
});
it("does NOT recover a step that was reset to pending after its code review approved", async () => {
const store = createMockStore();
const steps: StepStatus[] = ["done", "in-progress", "pending"];
const detail = makeTaskDetail("FN-1702", "in-progress", {
steps: makeSteps(...steps),
currentStep: 1,
log: [
{ timestamp: "2026-04-21T00:00:00.000Z", action: "Step 1 (Step 1) → in-progress" },
{ timestamp: "2026-04-21T00:00:01.000Z", action: "code review Step 1: APPROVE" },
// Workflow revision came in after the approval and reset this step.
{ timestamp: "2026-04-21T00:00:02.000Z", action: "Step 1 (Step 1) → pending" },
{ timestamp: "2026-04-21T00:00:03.000Z", action: "Step 1 (Step 1) → in-progress" },
],
});
store.listTasks.mockResolvedValue([makeTask("FN-1702", "in-progress", { steps: makeSteps(...steps), currentStep: 1 })]);
store.getTask.mockResolvedValue(detail);
mockAgentSuccess();
const executor = new TaskExecutor(store, "/tmp/test");
await executor.resumeOrphaned();
await new Promise((r) => setTimeout(r, 50));
// Must NOT mark the step done — the reset invalidated the prior approval.
const updateStepDoneCalls = store.updateStep.mock.calls.filter(
(c: any[]) => c[0] === "FN-1702" && c[2] === "done",
);
expect(updateStepDoneCalls).toHaveLength(0);
});
it("does NOT recover a step whose code review only revised (no APPROVE)", async () => {
const store = createMockStore();
const steps: StepStatus[] = ["done", "in-progress", "pending"];
const detail = makeTaskDetail("FN-1703", "in-progress", {
steps: makeSteps(...steps),
currentStep: 1,
log: [
{ timestamp: "2026-04-21T00:00:00.000Z", action: "Step 1 (Step 1) → in-progress" },
{ timestamp: "2026-04-21T00:00:01.000Z", action: "code review Step 1: REVISE" },
],
});
store.listTasks.mockResolvedValue([makeTask("FN-1703", "in-progress", { steps: makeSteps(...steps), currentStep: 1 })]);
store.getTask.mockResolvedValue(detail);
mockAgentSuccess();
const executor = new TaskExecutor(store, "/tmp/test");
await executor.resumeOrphaned();
await new Promise((r) => setTimeout(r, 50));
const updateStepDoneCalls = store.updateStep.mock.calls.filter(
(c: any[]) => c[0] === "FN-1703" && c[2] === "done",
);
expect(updateStepDoneCalls).toHaveLength(0);
});
it("recovers multiple in-progress steps that each have approved code reviews", async () => {
const store = createMockStore();
// Two consecutive steps each stuck in-progress with approved code reviews.
// (Rare but possible if the agent was processing step N+1 when the engine
// stopped after approving step N.)
const steps: StepStatus[] = ["in-progress", "in-progress", "pending"];
const detail = makeTaskDetail("FN-1704", "in-progress", {
steps: makeSteps(...steps),
currentStep: 0,
log: [
{ timestamp: "2026-04-21T00:00:00.000Z", action: "Step 0 (Step 0) → in-progress" },
{ timestamp: "2026-04-21T00:00:01.000Z", action: "code review Step 0: APPROVE" },
{ timestamp: "2026-04-21T00:00:02.000Z", action: "Step 1 (Step 1) → in-progress" },
{ timestamp: "2026-04-21T00:00:03.000Z", action: "code review Step 1: APPROVE" },
],
});
store.listTasks.mockResolvedValue([makeTask("FN-1704", "in-progress", { steps: makeSteps(...steps), currentStep: 0 })]);
store.getTask.mockResolvedValue(detail);
mockAgentSuccess();
const executor = new TaskExecutor(store, "/tmp/test");
await executor.resumeOrphaned();
await new Promise((r) => setTimeout(r, 50));
expect(store.updateStep).toHaveBeenCalledWith("FN-1704", 0, "done");
expect(store.updateStep).toHaveBeenCalledWith("FN-1704", 1, "done");
});
it("resumeOrphaned() logs 'Resumed after engine restart' for each orphaned task", async () => {
const store = createMockStore();
const task1 = makeTask("FN-040", "in-progress");