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

fix(FN-XXXX): quiet pi-claude-cli MCP config refresh log

Browse Source 
Demote the refresh message from console.error to debugMcp so it no
longer appears as an error in normal output.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-05-01 22:58:00 -07:00
parent 24b3ded03f
commit 31195374d4
13 changed files with 852 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
packages/engine/src/__tests__/executor.test.ts
View File

@@ -3099,6 +3099,24 @@ describe("TaskExecutor pause behavior", () => {
expect(store.logEntry).not.toHaveBeenCalledWith("FN-001", expect.anything());
});
it("skips resumeOrphaned entirely while enginePaused is active", async () => {
const store = createMockStore();
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
enginePaused: true,
globalPause: false,
});
const executor = new TaskExecutor(store, "/tmp/test");
await executor.resumeOrphaned();
expect(store.listTasks).not.toHaveBeenCalled();
expect(store.logEntry).not.toHaveBeenCalled();
});
it("resumes unpaused in-progress task with no active session", async () => {
const store = createMockStore();
const disposeFn = vi.fn();
@@ -3137,6 +3155,47 @@ describe("TaskExecutor pause behavior", () => {
expect(store.logEntry).toHaveBeenCalledWith("FN-001", "Resuming execution after unpause", undefined, undefined);
});
it("does not resume unpaused in-progress task while global pause is active", async () => {
const store = createMockStore();
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
globalPause: true,
enginePaused: false,
});
mockedCreateFnAgent.mockImplementation(async () => ({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
}) as any);
const executor = new TaskExecutor(store, "/tmp/test");
store._trigger("task:updated", {
id: "FN-001",
paused: undefined,
column: "in-progress",
description: "Test task",
title: "Paused runtime task",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
await new Promise((r) => setTimeout(r, 20));
expect(executor).toBeTruthy();
expect(mockedCreateFnAgent).not.toHaveBeenCalled();
expect(store.logEntry).not.toHaveBeenCalledWith("FN-001", "Resuming execution after unpause", undefined, undefined);
});
it("does not recursively resume when resume logging emits task updated", async () => {
const store = createMockStore();
const task = {
@@ -4134,8 +4193,17 @@ describe("TaskExecutor global pause behavior", () => {
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "failed" });
});
it("finalizes to in-review when global pause hits after fn_task_done", async () => {
it("defers completion handoff when global pause hits after fn_task_done", async () => {
const store = createMockStore();
let globalPause = false;
store.getSettings.mockImplementation(async () => ({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
globalPause,
enginePaused: false,
}));
mockedCreateFnAgent.mockImplementation(async (opts: any) => {
const customTools = opts.customTools || [];
@@ -4146,6 +4214,7 @@ describe("TaskExecutor global pause behavior", () => {
if (taskDoneTool) {
await taskDoneTool.execute("tool-1", {});
}
globalPause = true;
store._trigger("settings:updated", {
settings: { globalPause: true },
previous: { globalPause: false },
@@ -4168,11 +4237,17 @@ describe("TaskExecutor global pause behavior", () => {
createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(),
});
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review");
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "todo");
expect(
store.logEntry.mock.calls.some(
([id, action]: [string, string]) =>
id === "FN-001" && action.includes("Completion handoff deferred — global pause active"),
),
).toBe(true);
});
it("promotes todo tasks to in-progress when fn_task_done is called while paused", async () => {
it("parks todo tasks in in-progress when fn_task_done is called during global pause", async () => {
const store = createMockStore();
let capturedCustomTools: any[] = [];
@@ -4192,6 +4267,14 @@ describe("TaskExecutor global pause behavior", () => {
};
store.getTask.mockResolvedValue(todoTask);
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
globalPause: true,
enginePaused: false,
});
store.moveTask.mockImplementation(async (_id: string, to: string) => ({ ...todoTask, column: to, paused: undefined }));
mockedCreateFnAgent.mockImplementation((async (opts: any) => {
@@ -4212,13 +4295,20 @@ describe("TaskExecutor global pause behavior", () => {
const executor = new TaskExecutor(store, "/tmp/test");
await executor.execute(todoTask as any);
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { paused: false, status: null });
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { paused: false, status: null });
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: null });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review");
expect(store.logEntry).toHaveBeenCalledWith(
"FN-001",
expect.stringContaining("fn_task_done called while task was in todo"),
expect.stringContaining("fn_task_done called while task was in todo during pause"),
);
expect(
store.logEntry.mock.calls.some(
([id, action]: [string, string]) =>
id === "FN-001" && action.includes("Completion handoff deferred — global pause active"),
),
).toBe(true);
});
it("takes no action when globalPause remains false", async () => {
@@ -4339,6 +4429,62 @@ describe("TaskExecutor enginePaused soft pause (no agent termination)", () => {
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "failed" });
});
it("keeps fn_task_done on the normal completion path when enginePaused becomes true", async () => {
const store = createMockStore();
const mutableSettings = {
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
globalPause: false,
enginePaused: false,
};
let capturedCustomTools: any[] = [];
let taskDoneResult: any;
store.getSettings.mockImplementation(async () => ({ ...mutableSettings }));
mockedCreateFnAgent.mockImplementation((async (opts: any) => {
capturedCustomTools = opts.customTools || [];
return {
session: {
prompt: vi.fn().mockImplementation(async () => {
mutableSettings.enginePaused = true;
store._trigger("settings:updated", {
settings: { enginePaused: true },
previous: { enginePaused: false },
});
const taskDoneTool = capturedCustomTools.find((tool: any) => tool.name === "fn_task_done");
if (taskDoneTool) {
taskDoneResult = await taskDoneTool.execute("call-1", { summary: "done" });
}
}),
dispose: vi.fn(),
subscribe: vi.fn(),
on: vi.fn(),
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
state: {},
},
};
}) as any);
const executor = new TaskExecutor(store, "/tmp/test");
const watchdogSpy = vi.spyOn(executor as any, "scheduleCompletedTaskWatchdog");
await executor.execute({
id: "FN-001", title: "Test", description: "T", column: "in-progress",
dependencies: [], steps: [], currentStep: 0, log: [],
createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(),
});
expect(taskDoneResult.content[0].text).toBe(
"Task marked complete with summary. All steps done. Moving to in-review.",
);
expect(watchdogSpy).toHaveBeenCalledWith("FN-001", "fn_task_done");
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { paused: false, status: null });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");
});
it("does NOT move tasks to todo when enginePaused transitions false→true", async () => {
const store = createMockStore();
let capturedCustomTools: any[] = [];
@@ -8120,7 +8266,7 @@ describe("Workflow Steps Execution", () => {
dependencies: [] as string[],
steps: [{ name: "Preflight", status: "pending" as const }],
currentStep: 0,
log: [] as string[],
log: [] as any[],
enabledWorkflowSteps: ["WS-001"],
workflowStepRetries: 3, // Exhaust retries so task fails immediately
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
@@ -8264,7 +8410,7 @@ describe("Workflow Steps Execution", () => {
dependencies: [] as string[],
steps: [{ name: "Preflight", status: "pending" as const }],
currentStep: 0,
log: [] as string[],
log: [] as any[],
enabledWorkflowSteps: ["WS-001"],
workflowStepRetries: 3, // Exhaust retries so task fails immediately
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
@@ -9155,6 +9301,100 @@ describe("Workflow Steps Execution", () => {
expect(onComplete).toHaveBeenCalled();
expect(onError).not.toHaveBeenCalled();
});
it("parks a task pause during a prompt-mode workflow step instead of routing through failure recovery", async () => {
const store = createMockStore();
const mutableTask = {
id: "FN-001",
title: "Test",
description: "Test task",
column: "in-progress" as const,
paused: false,
dependencies: [] as string[],
steps: [{ name: "Preflight", status: "pending" as const }],
currentStep: 0,
log: [] as any[],
enabledWorkflowSteps: ["WS-001"],
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
store.getTask.mockImplementation(async () => mutableTask as any);
store.updateTask.mockImplementation(async (_taskId: string, patch: Record<string, unknown>) => {
Object.assign(mutableTask, patch);
return { ...mutableTask };
});
store.moveTask.mockImplementation(async (_taskId: string, column: string) => {
mutableTask.column = column as typeof mutableTask.column;
return { ...mutableTask };
});
store.getWorkflowStep.mockResolvedValue({
id: "WS-001",
name: "QA Check",
description: "Run tests",
mode: "prompt",
prompt: "Run the test suite.",
enabled: true,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
const workflowAbort = vi.fn().mockResolvedValue(undefined);
const workflowDispose = vi.fn();
let callIdx = 0;
mockedCreateFnAgent.mockImplementation((async (opts: any) => {
callIdx++;
if (callIdx === 1) {
const customTools = opts.customTools || [];
return {
session: {
prompt: vi.fn().mockImplementation(async () => {
const taskDoneTool = customTools.find((tool: any) => tool.name === "fn_task_done");
if (taskDoneTool) {
await taskDoneTool.execute("tool-1", {});
}
}),
dispose: vi.fn(),
subscribe: vi.fn(),
on: vi.fn(),
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
state: {},
},
};
}
return {
session: {
prompt: vi.fn().mockImplementation(async () => {
mutableTask.paused = true;
store._trigger("task:updated", { ...mutableTask });
throw new Error("workflow step aborted by pause");
}),
abort: workflowAbort,
dispose: workflowDispose,
subscribe: vi.fn(),
on: vi.fn(),
state: {},
},
};
}) as any);
const executor = new TaskExecutor(store, "/tmp/test");
await executor.execute({ ...mutableTask });
expect(workflowDispose).toHaveBeenCalled();
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true });
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review");
expect(store.addTaskComment).not.toHaveBeenCalled();
expect(store.logEntry).toHaveBeenCalledWith(
"FN-001",
"Execution paused during pre-merge workflow step — moved to todo",
undefined,
expect.objectContaining({ agentId: "executor" }),
);
});
});
describe("Real-time steering injection", () => {
@@ -11276,6 +11516,25 @@ describe("TaskExecutor watchdogs", () => {
);
});
it("defers workflow rerun bounce while global pause is active", async () => {
const store = createMockStore();
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: false,
globalPause: true,
enginePaused: false,
});
const executor = new TaskExecutor(store, "/tmp/test");
const outcome = await (executor as any).performWorkflowRerunBounce("FN-WD-PAUSE", "/tmp/fn-wd-pause");
expect(outcome).toBe("deferred-paused");
expect(store.getTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
});
it("preserves the original executionStartedAt during a workflow rerun bounce", async () => {
const store = createMockStore();
const originalExecutionStartedAt = "2026-04-30T05:06:43.781Z";

37
packages/engine/src/__tests__/project-engine.test.ts
View File

@@ -15,6 +15,7 @@ const mocks = vi.hoisted(() => ({
cronRunnerStop: vi.fn(),
runtimeStart: vi.fn(async () => undefined),
runtimeStop: vi.fn(async () => undefined),
runtimeResumeAfterUnpause: vi.fn(async () => undefined),
aiMergeTask: vi.fn(),
execFile: vi.fn(),
currentStore: null as Record<string, unknown> | null,
@@ -92,6 +93,7 @@ vi.mock("../runtimes/in-process-runtime.js", () => ({
InProcessRuntime: vi.fn().mockImplementation(() => ({
start: mocks.runtimeStart,
stop: mocks.runtimeStop,
resumeAfterUnpause: mocks.runtimeResumeAfterUnpause,
getTaskStore: () => mocks.currentStore,
getAgentStore: vi.fn(),
getMessageStore: vi.fn(),
@@ -230,6 +232,7 @@ function createEngine(options?: ConstructorParameters<typeof ProjectEngine>[2])
}
beforeEach(() => {
mocks.runtimeResumeAfterUnpause.mockClear();
mocks.notifierStart.mockClear();
mocks.notifierStop.mockClear();
mocks.notifierNotifyGridlock.mockClear();
@@ -1500,6 +1503,24 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => {
await engine.stop();
});
it("resumes deferred startup recovery on engine unpause", async () => {
const mockStore = createMockStore(baseSettings);
mocks.currentStore = mockStore.store;
const engine = createEngine();
await engine.start();
mocks.runtimeResumeAfterUnpause.mockClear();
await mockStore.emitSettingsUpdated(
{ ...baseSettings, enginePaused: false },
{ ...baseSettings, enginePaused: true },
);
expect(mocks.runtimeResumeAfterUnpause).toHaveBeenCalledTimes(1);
await engine.stop();
});
});
describe("ProjectEngine swallowed error hardening", () => {
@@ -1602,7 +1623,7 @@ describe("ProjectEngine swallowed error hardening", () => {
await engine.stop();
});
it("warns when resumeOrphaned dispatch fails during global unpause", async () => {
it("warns when resumeAfterUnpause dispatch fails during global unpause", async () => {
const mockStore = createMockStore(baseSettings);
mocks.currentStore = mockStore.store;
const engine = createEngine();
@@ -1610,9 +1631,9 @@ describe("ProjectEngine swallowed error hardening", () => {
warnSpy.mockClear();
const runtime = engine.getRuntime() as unknown as object;
Object.defineProperty(runtime, "executor", {
Object.defineProperty(runtime, "resumeAfterUnpause", {
get() {
throw new Error("executor broken");
throw new Error("resume hook broken");
},
configurable: true,
});
@@ -1623,7 +1644,7 @@ describe("ProjectEngine swallowed error hardening", () => {
);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Global unpause: failed to dispatch resumeOrphaned"),
expect.stringContaining("Global unpause: failed to dispatch resumeAfterUnpause"),
);
await engine.stop();
@@ -1650,7 +1671,7 @@ describe("ProjectEngine swallowed error hardening", () => {
await engine.stop();
});
it("warns when resumeOrphaned dispatch fails during engine unpause", async () => {
it("warns when resumeAfterUnpause dispatch fails during engine unpause", async () => {
const mockStore = createMockStore(baseSettings);
mocks.currentStore = mockStore.store;
const engine = createEngine();
@@ -1658,9 +1679,9 @@ describe("ProjectEngine swallowed error hardening", () => {
warnSpy.mockClear();
const runtime = engine.getRuntime() as unknown as object;
Object.defineProperty(runtime, "executor", {
Object.defineProperty(runtime, "resumeAfterUnpause", {
get() {
throw new Error("executor broken");
throw new Error("resume hook broken");
},
configurable: true,
});
@@ -1671,7 +1692,7 @@ describe("ProjectEngine swallowed error hardening", () => {
);
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Engine unpause: failed to dispatch resumeOrphaned"),
expect.stringContaining("Engine unpause: failed to dispatch resumeAfterUnpause"),
);
await engine.stop();

32
packages/engine/src/__tests__/scheduler.test.ts
View File

@@ -917,6 +917,38 @@ describe("Scheduler", () => {
expect(store.moveTask).not.toHaveBeenCalled();
});
it("aborts dispatch when globalPause becomes active mid-pass", async () => {
const todoTask = createMockTask({ id: "FN-002", column: "todo" });
vi.mocked(existsSync).mockReturnValue(true);
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
const getSettings = vi.fn()
.mockResolvedValueOnce({
maxConcurrent: 2,
maxWorktrees: 4,
globalPause: false,
enginePaused: false,
})
.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
globalPause: true,
enginePaused: false,
});
const store = createMockStore({
listTasks: vi.fn().mockResolvedValue([todoTask]),
getTask: vi.fn().mockResolvedValue(todoTask),
getSettings,
});
const scheduler = new Scheduler(store);
(scheduler as any).running = true;
await scheduler.schedule();
expect(store.getTask).toHaveBeenCalledWith("FN-002");
expect(store.updateTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
});
});
describe("engine pause", () => {

16
packages/engine/src/__tests__/self-healing.test.ts
View File

@@ -428,6 +428,10 @@ describe("SelfHealingManager", () => {
});
it("runStartupRecovery invokes the startup recovery subset", async () => {
vi.mocked(store.getSettings).mockResolvedValue({
globalPause: false,
enginePaused: false,
} as unknown as Settings);
const recoverNoProgressNoTaskDoneFailures = vi.spyOn(manager, "recoverNoProgressNoTaskDoneFailures").mockResolvedValue(1);
const recoverCompletedTasks = vi.spyOn(manager, "recoverCompletedTasks").mockResolvedValue(1);
const recoverMisclassifiedFailures = vi.spyOn(manager, "recoverMisclassifiedFailures").mockResolvedValue(1);
@@ -444,6 +448,18 @@ describe("SelfHealingManager", () => {
expect(recoverOrphanedExecutions).toHaveBeenCalledTimes(1);
expect(recoverApprovedTriageTasks).toHaveBeenCalledTimes(1);
});
it("runStartupRecovery skips while enginePaused is active", async () => {
vi.mocked(store.getSettings).mockResolvedValue({
globalPause: false,
enginePaused: true,
} as unknown as Settings);
const recoverCompletedTasks = vi.spyOn(manager, "recoverCompletedTasks").mockResolvedValue(1);
await manager.runStartupRecovery();
expect(recoverCompletedTasks).not.toHaveBeenCalled();
});
});
describe("recoverNoProgressNoTaskDoneFailures", () => {

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

@@ -563,6 +563,8 @@ export class TaskExecutor {
}>();
/** Active step-session executors per task (mutually exclusive with activeSessions). */
private activeStepExecutors = new Map<string, StepSessionExecutor>();
/** Active pre-merge workflow step sessions per task. */
private activeWorkflowStepSessions = new Map<string, AgentSession>();
/**
* Reviewer subagent sessions per task. Reviewers (`reviewer.ts`) create their
* own AgentSessions that aren't part of `activeSessions`/`activeStepExecutors`,
@@ -612,6 +614,84 @@ export class TaskExecutor {
await this.store.mergeTask(taskId);
return "merged";
}
private async getExecutionPauseLabel(): Promise<"global pause" | "engine pause" | null> {
const settings = await this.store.getSettings();
if (settings.globalPause) return "global pause";
if (settings.enginePaused) return "engine pause";
return null;
}
private async shouldDeferCompletionForGlobalPause(
taskId: string,
context: string,
): Promise<boolean> {
const settings = await this.store.getSettings();
if (!settings.globalPause) {
return false;
}
this.clearCompletedTaskWatchdog(taskId);
executorLog.log(`${taskId}: completion handoff deferred — global pause active (${context})`);
await this.store.logEntry(
taskId,
`Completion handoff deferred — global pause active (${context})`,
undefined,
this.currentRunContext,
).catch(() => undefined);
return true;
}
private async shouldDeferWorkflowStepCompletion(
taskId: string,
context: string,
): Promise<boolean> {
let latestTask: Task | null = null;
try {
latestTask = await this.store.getTask(taskId);
} catch {
latestTask = null;
}
if (latestTask?.paused || this.pausedAborted.has(taskId)) {
this.clearCompletedTaskWatchdog(taskId);
executorLog.log(`${taskId}: completion handoff deferred — task paused (${context})`);
await this.store.logEntry(
taskId,
`Completion handoff deferred — task paused (${context})`,
undefined,
this.currentRunContext,
).catch(() => undefined);
return true;
}
return this.shouldDeferCompletionForGlobalPause(taskId, context);
}
private async parkTaskAfterWorkflowStepPause(taskId: string): Promise<boolean> {
let latestTask: Task | null = null;
try {
latestTask = await this.store.getTask(taskId);
} catch {
latestTask = null;
}
if (!latestTask?.paused) {
return false;
}
executorLog.log(`${taskId}: workflow step interrupted by task pause — moving to todo`);
await this.store.logEntry(
taskId,
"Execution paused during pre-merge workflow step — moved to todo",
undefined,
this.currentRunContext,
).catch(() => undefined);
if (latestTask.column === "in-progress") {
await this.store.moveTask(taskId, "todo", { preserveResumeState: true });
}
return true;
}
/** Child agent sessions keyed by agent ID. Used for termination. */
private childSessions = new Map<string, AgentSession>();
/** Total count of currently spawned agents (across all parents). */
@@ -768,6 +848,20 @@ export class TaskExecutor {
);
this.activeStepExecutors.delete(task.id);
}
if (this.activeWorkflowStepSessions.has(task.id)) {
executorLog.log(`${task.id} moved from in-progress to ${to} — terminating workflow step session`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const workflowSession = this.activeWorkflowStepSessions.get(task.id)!;
const sessionWithAbort = workflowSession as AgentSession & { abort?: () => Promise<void> };
if (typeof sessionWithAbort.abort === "function") {
void sessionWithAbort.abort().catch((err) => {
executorLog.warn(`Failed to abort workflow step session for ${task.id}: ${err}`);
});
}
workflowSession.dispose();
this.activeWorkflowStepSessions.delete(task.id);
}
// Reviewer subagents run in their own sessions outside `activeSessions`
// and `activeStepExecutors`, so the loops above don't reach them.
// Without this, a reviewer keeps running (and emitting verdicts that
@@ -823,6 +917,25 @@ export class TaskExecutor {
this.disposeSubagentsForTask(task.id, "task paused");
return;
}
if (task.paused && this.activeWorkflowStepSessions.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating workflow step session`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const workflowSession = this.activeWorkflowStepSessions.get(task.id)!;
const sessionWithAbort = workflowSession as AgentSession & { abort?: () => Promise<void> };
if (typeof sessionWithAbort.abort === "function") {
await sessionWithAbort.abort().catch((err) =>
executorLog.warn(`Failed to abort workflow step session for pause ${task.id}: ${err}`),
);
}
workflowSession.dispose();
this.activeWorkflowStepSessions.delete(task.id);
this.loopRecoveryState.delete(task.id);
this.spawnedAgents.delete(task.id);
this.stuckAborted.delete(task.id);
this.disposeSubagentsForTask(task.id, "task paused");
return;
}
// Handle unpause of an in-progress task with no active session.
// This covers orphaned states (e.g., engine restarted while task was
@@ -833,12 +946,32 @@ export class TaskExecutor {
&& task.column === "in-progress"
&& !this.activeSessions.has(task.id)
&& !this.activeStepExecutors.has(task.id)
&& !this.activeWorkflowStepSessions.has(task.id)
) {
if (
!this.executing.has(task.id)
&& !this.resumingUnpaused.has(task.id)
&& !this.recoveringCompleted.has(task.id)
) {
const pauseLabel = await this.getExecutionPauseLabel();
if (pauseLabel) {
executorLog.log(`Skipping unpause resume for ${task.id} — ${pauseLabel} active`);
return;
}
if (this.isTaskWorkComplete(task) && !task.mergeDetails) {
this.recoveringCompleted.add(task.id);
executorLog.log(`${task.id} unpaused with completed work and no session — recovering directly to in-review`);
void this.recoverCompletedTask(task)
.catch((err) =>
executorLog.error(`Failed to recover completed unpaused task ${task.id}:`, err),
)
.finally(() => {
this.recoveringCompleted.delete(task.id);
});
return;
}
this.resumingUnpaused.add(task.id);
executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`);
try {
@@ -994,6 +1127,22 @@ export class TaskExecutor {
this.spawnedAgents.delete(taskId);
this.stuckAborted.delete(taskId);
}
for (const [taskId, workflowSession] of this.activeWorkflowStepSessions) {
executorLog.log(`Global pause — terminating workflow step session for ${taskId}`);
this.pausedAborted.add(taskId);
this.options.stuckTaskDetector?.untrackTask(taskId);
const sessionWithAbort = workflowSession as AgentSession & { abort?: () => Promise<void> };
if (typeof sessionWithAbort.abort === "function") {
void sessionWithAbort.abort().catch((err) => {
executorLog.warn(`Failed to abort workflow step session for ${taskId}: ${err}`);
});
}
workflowSession.dispose();
this.activeWorkflowStepSessions.delete(taskId);
this.loopRecoveryState.delete(taskId);
this.spawnedAgents.delete(taskId);
this.stuckAborted.delete(taskId);
}
}
});
@@ -1131,6 +1280,7 @@ export class TaskExecutor {
|| this.executing.has(taskId)
|| this.activeSessions.has(taskId)
|| this.activeStepExecutors.has(taskId)
|| this.activeWorkflowStepSessions.has(taskId)
|| this.resumingUnpaused.has(taskId)
) {
return;
@@ -1138,6 +1288,11 @@ export class TaskExecutor {
this.recoveringCompleted.add(taskId);
try {
const pauseLabel = await this.getExecutionPauseLabel();
if (pauseLabel) {
return;
}
let currentTask: Task | null = null;
try {
currentTask = await this.store.getTask(taskId);
@@ -1192,7 +1347,13 @@ export class TaskExecutor {
taskId: string,
worktreePath: string,
preserveResumeState: boolean = true,
): Promise<"bounced" | "skipped-pending"> {
): Promise<"bounced" | "skipped-pending" | "deferred-paused"> {
const pauseLabel = await this.getExecutionPauseLabel();
if (pauseLabel) {
executorLog.log(`${taskId}: workflow rerun deferred — ${pauseLabel} active`);
return "deferred-paused";
}
// Re-entry guard: if a previous bounce for the same task is still
// mid-flight (e.g., the watchdog fired before the original sequence
// completed), skip rather than racing two concurrent moveTask sequences.
@@ -1209,6 +1370,10 @@ export class TaskExecutor {
if (!latestTask) {
throw new Error("task missing during workflow rerun bounce");
}
if (latestTask.paused) {
executorLog.log(`${taskId}: workflow rerun deferred — task is paused`);
return "deferred-paused";
}
if (latestTask.column === "in-progress") {
const originalExecutionStartedAt = latestTask.executionStartedAt;
@@ -1225,12 +1390,22 @@ export class TaskExecutor {
worktree: worktreePath,
executionStartedAt: originalExecutionStartedAt ?? null,
});
const pauseLabelAfterTodo = await this.getExecutionPauseLabel();
if (pauseLabelAfterTodo) {
executorLog.log(`${taskId}: workflow rerun parked in todo — ${pauseLabelAfterTodo} became active during bounce`);
return "deferred-paused";
}
await this.store.moveTask(taskId, "in-progress");
return "bounced";
}
if (latestTask.column === "todo") {
await this.store.updateTask(taskId, { worktree: worktreePath });
const pauseLabelBeforeResume = await this.getExecutionPauseLabel();
if (pauseLabelBeforeResume) {
executorLog.log(`${taskId}: workflow rerun parked in todo — ${pauseLabelBeforeResume} became active before resume`);
return "deferred-paused";
}
await this.store.moveTask(taskId, "in-progress");
return "bounced";
}
@@ -1254,8 +1429,10 @@ export class TaskExecutor {
const outcome = await this.performWorkflowRerunBounce(taskId, worktreePath, preserveResumeState);
if (outcome === "bounced") {
executorLog.log(successMessage);
} else {
} else if (outcome === "skipped-pending") {
executorLog.warn(`${taskId}: rerun bounce skipped — another bounce already in flight`);
} else {
executorLog.log(`${taskId}: rerun bounce deferred while pause is active`);
}
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
@@ -1266,6 +1443,12 @@ export class TaskExecutor {
const watchdog = setTimeout(async () => {
this.workflowRerunWatchdogs.delete(taskId);
const pauseLabel = await this.getExecutionPauseLabel();
if (pauseLabel) {
executorLog.log(`${taskId}: workflow rerun watchdog skipped — ${pauseLabel} active`);
return;
}
let currentTask: Task | null = null;
try {
currentTask = await this.store.getTask(taskId);
@@ -1293,7 +1476,7 @@ export class TaskExecutor {
const outcome = await this.performWorkflowRerunBounce(taskId, worktreePath, preserveResumeState);
if (outcome === "bounced") {
executorLog.warn(`${taskId}: workflow rerun watchdog retry succeeded`);
} else {
} else if (outcome === "skipped-pending") {
// The original bounce is still mid-flight, which means *it* is the
// one that's hung — not us. Log honestly so operators don't see a
// false "succeeded" message while the task is actually stranded.
@@ -1304,6 +1487,8 @@ export class TaskExecutor {
taskId,
`Workflow rerun watchdog retry skipped — original bounce still in flight after ${WORKFLOW_RERUN_WATCHDOG_MS / 1000}s; task may be stuck`,
).catch(() => undefined);
} else {
executorLog.log(`${taskId}: workflow rerun watchdog retry deferred while pause is active`);
}
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
@@ -1499,12 +1684,21 @@ export class TaskExecutor {
this.executing.has(task.id)
|| this.activeSessions.has(task.id)
|| this.activeStepExecutors.has(task.id)
|| this.activeWorkflowStepSessions.has(task.id)
|| this.resumingUnpaused.has(task.id)
) {
executorLog.log(`${task.id}: skipping recoverCompletedTask — task has active execution in flight`);
return false;
}
const settings = await this.store.getSettings();
if (settings.globalPause || settings.enginePaused) {
executorLog.log(
`${task.id}: skipping recoverCompletedTask — ${
settings.globalPause ? "global pause" : "engine pause"
} active`,
);
return false;
}
// Capture modified files if the worktree still exists
if (task.worktree && existsSync(task.worktree)) {
@@ -1516,7 +1710,16 @@ export class TaskExecutor {
// Run workflow steps before transitioning — skip in fast mode
if (task.executionMode !== "fast") {
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps during completed-task recovery")) {
return false;
}
const workflowResult = await this.runWorkflowSteps(task, task.worktree, settings);
if (workflowResult === "deferred-paused") {
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
}
return false;
}
if (!workflowResult.allPassed) {
// For recovery path, treat any failure (including revision) as hard failure
// Send back to in-progress so executor can attempt to fix the issues
@@ -1528,6 +1731,9 @@ export class TaskExecutor {
}
}
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before in-review transition during completed-task recovery")) {
return false;
}
await this.persistTokenUsage(task.id);
await this.store.moveTask(task.id, "in-review");
this.clearCompletedTaskWatchdog(task.id);
@@ -1599,6 +1805,16 @@ export class TaskExecutor {
* directly to in-review without spawning a new agent session.
*/
async resumeOrphaned(): Promise<void> {
const settings = await this.store.getSettings();
if (settings.globalPause || settings.enginePaused) {
executorLog.log(
`resumeOrphaned skipped — ${
settings.globalPause ? "global pause" : "engine pause"
} is active`,
);
return;
}
const tasks = await this.store.listTasks({ slim: true, column: "in-progress" });
const inProgress = tasks.filter(
(t) => t.column === "in-progress" && !this.executing.has(t.id) && !t.paused,
@@ -2180,10 +2396,23 @@ export class TaskExecutor {
}
this.scheduleCompletedTaskWatchdog(task.id, "step-session completion");
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after step-session completion")) {
return;
}
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
return;
}
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
}
return;
}
if (!workflowResult.allPassed) {
// Check if revision was requested
if (workflowResult.revisionRequested) {
@@ -2206,6 +2435,9 @@ export class TaskExecutor {
// Reset retry counters on success
await this.store.updateTask(task.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before in-review transition after step-session completion")) {
return;
}
await this.store.moveTask(task.id, "in-review");
this.clearCompletedTaskWatchdog(task.id);
@@ -2649,6 +2881,9 @@ export class TaskExecutor {
this.pausedAborted.delete(task.id);
wasPaused = true;
if (await this.shouldFinalizeCompletedTask(task.id, taskDone)) {
if (await this.shouldDeferCompletionForGlobalPause(task.id, "paused after completion")) {
return;
}
executorLog.log(`${task.id} paused after completion (graceful session exit) — finalizing to in-review`);
await this.store.logEntry(task.id, "Execution paused after completion — finalizing to in-review");
await this.persistTokenUsage(task.id);
@@ -2698,10 +2933,25 @@ export class TaskExecutor {
}
this.scheduleCompletedTaskWatchdog(task.id, "task completion");
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after task completion")) {
return;
}
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
wasPaused = true;
return;
}
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
wasPaused = true;
}
return;
}
if (!workflowResult.allPassed) {
// Check if revision was requested
if (workflowResult.revisionRequested) {
@@ -2724,6 +2974,9 @@ export class TaskExecutor {
// Reset retry counters on success
await this.store.updateTask(task.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before in-review transition after task completion")) {
return;
}
await this.persistTokenUsage(task.id);
await this.store.moveTask(task.id, "in-review");
@@ -2863,10 +3116,25 @@ export class TaskExecutor {
}
this.scheduleCompletedTaskWatchdog(task.id, "task completion retry");
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after task completion retry")) {
return;
}
// Run workflow steps before moving to in-review — skip in fast mode
if (executionMode !== "fast") {
const workflowResult = await this.runWorkflowSteps(task, worktreePath, settings);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
wasPaused = true;
return;
}
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
wasPaused = true;
}
return;
}
if (!workflowResult.allPassed) {
if (workflowResult.revisionRequested) {
await this.handleWorkflowRevisionRequest(task, worktreePath, workflowResult.feedback, workflowResult.stepName);
@@ -2881,6 +3149,9 @@ export class TaskExecutor {
}
await this.store.updateTask(task.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before in-review transition after task completion retry")) {
return;
}
await this.persistTokenUsage(task.id);
await this.store.moveTask(task.id, "in-review");
@@ -2989,6 +3260,9 @@ export class TaskExecutor {
// Task was paused mid-execution — clean up worktree and move to todo
this.pausedAborted.delete(task.id);
if (await this.shouldFinalizeCompletedTask(task.id, taskDone)) {
if (await this.shouldDeferCompletionForGlobalPause(task.id, "paused after completion")) {
return;
}
executorLog.log(`${task.id} paused after completion — finalizing to in-review`);
await this.store.logEntry(task.id, "Execution paused after completion — finalizing to in-review", undefined, this.currentRunContext);
await this.persistTokenUsage(task.id);
@@ -3494,7 +3768,13 @@ export class TaskExecutor {
if (params.summary) {
await store.updateTask(taskId, { summary: params.summary });
}
await store.updateTask(taskId, { paused: false, status: null });
const settings = await store.getSettings();
const hardPauseActive = Boolean(task.paused || settings.globalPause);
if (hardPauseActive) {
await store.updateTask(taskId, { status: null });
} else {
await store.updateTask(taskId, { paused: false, status: null });
}
await store.logEntry(taskId, "Task marked done by agent");
const latestTask = await store.getTask(taskId);
@@ -3502,17 +3782,21 @@ export class TaskExecutor {
if (latestColumn === "todo") {
await store.logEntry(
taskId,
"fn_task_done called while task was in todo — promoting to in-progress before completion handoff",
hardPauseActive
? "fn_task_done called while task was in todo during pause — promoting to in-progress for deferred completion handoff"
: "fn_task_done called while task was in todo — promoting to in-progress before completion handoff",
);
await store.moveTask(taskId, "in-progress");
latestColumn = "in-progress";
}
if (latestColumn === "in-progress") {
if (latestColumn === "in-progress" && !hardPauseActive) {
this.scheduleCompletedTaskWatchdog(taskId, "fn_task_done");
}
const successMessage = params.summary
const successMessage = hardPauseActive
? "Task marked complete. Completion handoff deferred until pause is cleared."
: params.summary
? "Task marked complete with summary. All steps done. Moving to in-review."
: "Task marked complete. All steps done. Moving to in-review.";
return {
@@ -4183,7 +4467,7 @@ ${failureFeedback}
task: Task,
worktreePath: string,
settings: Settings,
): Promise<WorkflowStepResult> {
): Promise<WorkflowStepResult | "deferred-paused"> {
// Check if task has enabled workflow steps
const currentTask = await this.store.getTask(task.id);
if (!currentTask.enabledWorkflowSteps?.length) return { allPassed: true };
@@ -4243,6 +4527,10 @@ ${failureFeedback}
continue;
}
if (await this.shouldDeferWorkflowStepCompletion(task.id, `before workflow step '${ws.name}'`)) {
return "deferred-paused";
}
await this.store.logEntry(task.id, `[pre-merge] Starting workflow step: ${ws.name} (${stepMode} mode)`);
executorLog.log(`${task.id} — [pre-merge] running workflow step: ${ws.name} (${stepMode} mode)`);
@@ -4263,6 +4551,9 @@ ${failureFeedback}
const result: WorkflowStepOutcome = stepMode === "script"
? await this.executeScriptWorkflowStep(task, ws, worktreePath, settings)
: await this.executeWorkflowStep(task, ws, worktreePath, settings);
if (await this.shouldDeferWorkflowStepCompletion(task.id, `workflow step '${ws.name}'`)) {
return "deferred-paused";
}
const completedAt = new Date().toISOString();
if (result.success) {
@@ -4334,6 +4625,9 @@ ${failureFeedback}
};
}
} catch (err: unknown) {
if (await this.shouldDeferWorkflowStepCompletion(task.id, `workflow step '${ws.name}'`)) {
return "deferred-paused";
}
const { message: errorMessage, detail: errorDetail, stack: errorStack } = formatError(err);
const completedAt = new Date().toISOString();
await this.store.logEntry(
@@ -4539,6 +4833,7 @@ and show an appropriate message to the user.\`
task.id,
`Workflow step '${workflowStep.name}' using model: ${describeModel(session)}${useOverride && attemptLabel === "primary" ? " (workflow step override)" : ""}${attemptLabel === "fallback" ? " (fallback after timeout)" : ""}`,
);
this.activeWorkflowStepSessions.set(task.id, session);
let output = "";
session.subscribe((event) => {
@@ -4612,6 +4907,10 @@ and show an appropriate message to the user.\`
return { success: false, error: errorMessage };
} finally {
if (timeoutHandle) clearTimeout(timeoutHandle);
const activeWorkflowStepSession = this.activeWorkflowStepSessions.get(task.id);
if (activeWorkflowStepSession === session) {
this.activeWorkflowStepSessions.delete(task.id);
}
// Suppress unused-variable warning; `timedOut` documents intent.
void timedOut;
}

18
packages/engine/src/project-engine.ts
View File

@@ -1824,14 +1824,13 @@ export class ProjectEngine {
runtimeLog.log("Global unpause — resuming agentic activity");
try {
const executor = (this.runtime as any).executor;
executor?.resumeOrphaned?.().catch((err: Error) =>
runtimeLog.error("Failed to resume orphaned tasks on unpause:", err),
const runtime = this.runtime as any;
runtime.resumeAfterUnpause?.().catch((err: Error) =>
runtimeLog.error("Failed to resume agentic activity on unpause:", err),
);
} catch (err: unknown) {
runtimeLog.warn(
`Global unpause: failed to dispatch resumeOrphaned: ${err instanceof Error ? err.message : String(err)}`,
`Global unpause: failed to dispatch resumeAfterUnpause: ${err instanceof Error ? err.message : String(err)}`,
);
}
@@ -1862,14 +1861,13 @@ export class ProjectEngine {
runtimeLog.log("Engine unpaused — resuming agentic activity");
try {
const executor = (this.runtime as any).executor;
executor?.resumeOrphaned?.().catch((err: Error) =>
runtimeLog.error("Failed to resume orphaned tasks on engine unpause:", err),
const runtime = this.runtime as any;
runtime.resumeAfterUnpause?.().catch((err: Error) =>
runtimeLog.error("Failed to resume agentic activity on engine unpause:", err),
);
} catch (err: unknown) {
runtimeLog.warn(
`Engine unpause: failed to dispatch resumeOrphaned: ${err instanceof Error ? err.message : String(err)}`,
`Engine unpause: failed to dispatch resumeAfterUnpause: ${err instanceof Error ? err.message : String(err)}`,
);
}

60
packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts
View File

@@ -15,6 +15,8 @@ const {
mockRecoverNoProgressNoTaskDoneFailures,
mockRunStartupRecovery,
mockExecutorCtor,
mockResumeOrphaned,
mockTaskStoreSettings,
mockMessageStoreSetHook,
} = vi.hoisted(() => ({
mockSelfHealingStart: vi.fn(),
@@ -23,6 +25,8 @@ const {
mockRecoverNoProgressNoTaskDoneFailures: vi.fn().mockResolvedValue(0),
mockRunStartupRecovery: vi.fn().mockResolvedValue(undefined),
mockExecutorCtor: vi.fn(),
mockResumeOrphaned: vi.fn().mockResolvedValue(undefined),
mockTaskStoreSettings: {} as Record<string, unknown>,
mockMessageStoreSetHook: vi.fn(),
}));
@@ -46,7 +50,7 @@ vi.mock("@fusion/core", async () => {
self.getDatabase = vi.fn().mockReturnValue(mockDatabase);
self.init = vi.fn().mockResolvedValue(undefined);
self.listTasks = vi.fn().mockResolvedValue([]);
self.getSettings = vi.fn().mockResolvedValue({});
self.getSettings = vi.fn().mockImplementation(async () => structuredClone(mockTaskStoreSettings));
self.getMissionStore = vi.fn().mockReturnValue({
getMissionWithHierarchy: vi.fn().mockReturnValue(null),
findNextPendingSlice: vi.fn().mockReturnValue(null),
@@ -141,7 +145,7 @@ vi.mock("../../executor.js", async () => {
TaskExecutor: vi.fn().mockImplementation((_store, _rootDir, options) => {
mockExecutorCtor(options);
const self = {} as Record<string, unknown>;
self.resumeOrphaned = vi.fn().mockResolvedValue(undefined);
self.resumeOrphaned = mockResumeOrphaned;
self.recoverCompletedTask = vi.fn().mockResolvedValue(true);
self.getExecutingTaskIds = vi.fn().mockReturnValue(new Set());
self.handleLoopDetected = vi.fn().mockResolvedValue(false);
@@ -184,6 +188,9 @@ describe("InProcessRuntime", () => {
}
beforeEach(() => {
for (const key of Object.keys(mockTaskStoreSettings)) {
delete mockTaskStoreSettings[key];
}
// Create a unique temp directory for this test run
testDir = mkdtempSync(join(tmpdir(), `fn-test-${randomUUID().slice(0, 8)}-`));
@@ -243,6 +250,55 @@ describe("InProcessRuntime", () => {
await runtime.start();
expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
}, 30000);
it("defers startup recovery while enginePaused is active", async () => {
mockTaskStoreSettings.enginePaused = true;
await runtime.start();
expect(mockRecoverNoProgressNoTaskDoneFailures).not.toHaveBeenCalled();
expect(mockResumeOrphaned).not.toHaveBeenCalled();
expect(mockRunStartupRecovery).not.toHaveBeenCalled();
}, 30000);
it("resumes deferred startup recovery after engine pause is cleared in startup order", async () => {
mockTaskStoreSettings.enginePaused = true;
await runtime.start();
mockRecoverNoProgressNoTaskDoneFailures.mockClear();
mockResumeOrphaned.mockClear();
mockRunStartupRecovery.mockClear();
mockTaskStoreSettings.enginePaused = false;
await runtime.resumeAfterUnpause();
expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
expect(mockRecoverNoProgressNoTaskDoneFailures.mock.invocationCallOrder[0]).toBeLessThan(
mockResumeOrphaned.mock.invocationCallOrder[0],
);
expect(mockResumeOrphaned.mock.invocationCallOrder[0]).toBeLessThan(
mockRunStartupRecovery.mock.invocationCallOrder[0],
);
}, 30000);
it("coalesces concurrent unpause recovery dispatches", async () => {
mockTaskStoreSettings.enginePaused = true;
await runtime.start();
mockRecoverNoProgressNoTaskDoneFailures.mockClear();
mockResumeOrphaned.mockClear();
mockRunStartupRecovery.mockClear();
mockTaskStoreSettings.enginePaused = false;
await Promise.all([runtime.resumeAfterUnpause(), runtime.resumeAfterUnpause()]);
expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
}, 30000);

85
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -116,6 +116,10 @@ export class InProcessRuntime
* before `start()` via `setMergeEnqueuer`.
*/
private mergeEnqueuer?: (taskId: string) => void;
/** Tracks whether startup recovery was intentionally deferred due to pause state. */
private startupRecoveryDeferred = false;
/** Prevent duplicate unpause recovery dispatches from racing each other. */
private resumeAfterUnpauseRunning = false;
/**
* @param config - Runtime configuration
@@ -669,20 +673,19 @@ export class InProcessRuntime
// 8. Set up event forwarding from TaskStore
this.setupEventForwarding();
// 9. Requeue no-progress no-task_done failures before resumeOrphaned
// can restart them.
await this.selfHealingManager.recoverNoProgressNoTaskDoneFailures();
const startupSettings = await this.taskStore.getSettings();
if (startupSettings.globalPause || startupSettings.enginePaused) {
this.startupRecoveryDeferred = true;
runtimeLog.log(
`Startup recovery deferred — ${
startupSettings.globalPause ? "global pause" : "engine pause"
} is active`,
);
} else {
this.startupRecoveryDeferred = false;
// 10. Resume orphaned in-progress tasks
await this.executor.resumeOrphaned();
// Some "stuck" tasks are already orphaned by the time the runtime boots:
// they no longer have a tracked session/worktree, so the stuck detector
// cannot recover them. Delegate the startup recovery pass to
// SelfHealingManager so the policy lives in one place.
void this.selfHealingManager.runStartupRecovery().catch((err) => {
runtimeLog.error("Self-healing startup recovery failed:", err);
});
await this.resumeStartupRecoverySequence();
}
// 11. Start scheduler and triage processor
this.scheduler.start();
@@ -907,6 +910,62 @@ export class InProcessRuntime
this.mergeEnqueuer = enqueueMerge;
}
/**
* Resume executor/self-healing activity after an unpause transition.
*
* When startup recovery had been deferred, this replays the original startup
* ordering so orphan resume and self-healing cannot race each other.
*/
async resumeAfterUnpause(): Promise<void> {
if (!this.taskStore || !this.executor || !this.selfHealingManager) {
return;
}
if (this.resumeAfterUnpauseRunning) {
return;
}
this.resumeAfterUnpauseRunning = true;
try {
const settings = await this.taskStore.getSettings();
if (settings.globalPause || settings.enginePaused) {
runtimeLog.log(
`Unpause recovery still blocked — ${
settings.globalPause ? "global pause" : "engine pause"
} remains active`,
);
return;
}
if (this.startupRecoveryDeferred) {
await this.resumeStartupRecoverySequence();
this.startupRecoveryDeferred = false;
return;
}
await this.executor.resumeOrphaned();
} finally {
this.resumeAfterUnpauseRunning = false;
}
}
private async resumeStartupRecoverySequence(): Promise<void> {
// Requeue no-progress no-task_done failures before resumeOrphaned can
// restart other orphaned executions.
await this.selfHealingManager!.recoverNoProgressNoTaskDoneFailures();
// Resume orphaned in-progress tasks before the broader self-healing scan
// so the executor can claim or fast-path eligible tasks first.
await this.executor!.resumeOrphaned();
// Some "stuck" tasks are already orphaned by the time the runtime boots:
// they no longer have a tracked session/worktree, so the stuck detector
// cannot recover them. Delegate the startup recovery pass to
// SelfHealingManager so the policy lives in one place.
void this.selfHealingManager!.runStartupRecovery().catch((err) => {
runtimeLog.error("Self-healing startup recovery failed:", err);
});
}
/**
* Get the project's TaskStore instance.
* @throws Error if runtime has not been started

10
packages/engine/src/scheduler.ts
View File

@@ -757,6 +757,16 @@ export class Scheduler {
continue;
}
const latestSettings = await this.store.getSettings();
if (latestSettings.globalPause) {
schedulerLog.log(`Task ${task.id} dispatch aborted — globalPause became active mid-pass`);
continue;
}
if (latestSettings.enginePaused) {
schedulerLog.log(`Task ${task.id} dispatch aborted — enginePaused became active mid-pass`);
continue;
}
// Resolve effective node for routing
let effectiveNode = resolveEffectiveNode(freshTask, settings);
schedulerLog.log(`Task ${task.id} routed to node=${effectiveNode.nodeId ?? "local"} (source=${effectiveNode.source})`);

63
packages/engine/src/self-healing.ts
View File

@@ -173,6 +173,16 @@ export class SelfHealingManager {
* stale in-progress/planning tasks that no longer have a live worker.
*/
async runStartupRecovery(): Promise<void> {
const settings = await this.store.getSettings();
if (settings.globalPause || settings.enginePaused) {
log.log(
`Startup recovery skipped — ${
settings.globalPause ? "global pause" : "engine pause"
} is active`,
);
return;
}
// Each recovery step is isolated — one failure doesn't prevent subsequent steps.
const steps: Array<{ name: string; fn: () => Promise<unknown> }> = [
{ name: "no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures().then(() => undefined) },
@@ -616,28 +626,37 @@ export class SelfHealingManager {
}
}
// Batch 2 — Task recovery (operations are independent of each other)
const batch2Fns: Array<{ name: string; fn: () => Promise<unknown> }> = [
{ name: "recover-completed-tasks", fn: () => this.recoverCompletedTasks() },
{ name: "recover-stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks() },
{ name: "recover-failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps() },
{ name: "recover-interrupted-merging", fn: () => this.recoverInterruptedMergingTasks() },
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
{ name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() },
{ name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() },
{ name: "recover-orphaned-executions", fn: () => this.recoverOrphanedExecutions() },
{ name: "recover-approved-triage", fn: () => this.recoverApprovedTriageTasks() },
{ name: "recover-orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks() },
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
];
for (const fn of batch2Fns) {
try {
await fn.fn();
log.log(`Maintenance batch 2 step "${fn.name}" succeeded`);
} catch (stepErr) {
log.error(`Maintenance batch 2 step "${fn.name}" failed: ${stepErr instanceof Error ? stepErr.message : String(stepErr)}`);
const recoverySettings = await this.store.getSettings();
if (recoverySettings.globalPause || recoverySettings.enginePaused) {
log.log(
`Maintenance batch 2 skipped — ${
recoverySettings.globalPause ? "global pause" : "engine pause"
} is active`,
);
} else {
// Batch 2 — Task recovery (operations are independent of each other)
const batch2Fns: Array<{ name: string; fn: () => Promise<unknown> }> = [
{ name: "recover-completed-tasks", fn: () => this.recoverCompletedTasks() },
{ name: "recover-stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks() },
{ name: "recover-failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps() },
{ name: "recover-interrupted-merging", fn: () => this.recoverInterruptedMergingTasks() },
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
{ name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() },
{ name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() },
{ name: "recover-orphaned-executions", fn: () => this.recoverOrphanedExecutions() },
{ name: "recover-approved-triage", fn: () => this.recoverApprovedTriageTasks() },
{ name: "recover-orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks() },
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
];
for (const fn of batch2Fns) {
try {
await fn.fn();
log.log(`Maintenance batch 2 step "${fn.name}" succeeded`);
} catch (stepErr) {
log.error(`Maintenance batch 2 step "${fn.name}" failed: ${stepErr instanceof Error ? stepErr.message : String(stepErr)}`);
}
}
}