fix: recover completed tasks stuck in progress
This commit is contained in:
5
.changeset/recover-stuck-completed-tasks.md
Normal file
5
.changeset/recover-stuck-completed-tasks.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@gsxdsm/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Recover completed tasks that were left stuck in `in-progress` before they could transition to `in-review`.
|
||||||
@@ -4,12 +4,16 @@ import { EventEmitter } from "node:events";
|
|||||||
// ── Capture instances & arguments ───────────────────────────────────
|
// ── Capture instances & arguments ───────────────────────────────────
|
||||||
|
|
||||||
let capturedExecutorOpts: Record<string, unknown> | undefined;
|
let capturedExecutorOpts: Record<string, unknown> | undefined;
|
||||||
|
let capturedSelfHealingOpts: Record<string, unknown> | undefined;
|
||||||
|
|
||||||
const {
|
const {
|
||||||
mockAuthStorage,
|
mockAuthStorage,
|
||||||
mockModelRegistry,
|
mockModelRegistry,
|
||||||
mockDiscoverAndLoadExtensions,
|
mockDiscoverAndLoadExtensions,
|
||||||
mockCreateExtensionRuntime,
|
mockCreateExtensionRuntime,
|
||||||
|
mockSelfHealingStart,
|
||||||
|
mockSelfHealingStop,
|
||||||
|
mockCheckStuckBudget,
|
||||||
} = vi.hoisted(() => ({
|
} = vi.hoisted(() => ({
|
||||||
mockAuthStorage: { getAuth: vi.fn(), setAuth: vi.fn() },
|
mockAuthStorage: { getAuth: vi.fn(), setAuth: vi.fn() },
|
||||||
mockModelRegistry: {
|
mockModelRegistry: {
|
||||||
@@ -21,6 +25,9 @@ const {
|
|||||||
errors: [],
|
errors: [],
|
||||||
}),
|
}),
|
||||||
mockCreateExtensionRuntime: vi.fn(),
|
mockCreateExtensionRuntime: vi.fn(),
|
||||||
|
mockSelfHealingStart: vi.fn(),
|
||||||
|
mockSelfHealingStop: vi.fn(),
|
||||||
|
mockCheckStuckBudget: vi.fn().mockResolvedValue(true),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
// Minimal mock store backed by EventEmitter so `store.on` works
|
// Minimal mock store backed by EventEmitter so `store.on` works
|
||||||
@@ -195,6 +202,14 @@ vi.mock("@fusion/engine", async (importOriginal) => {
|
|||||||
start: vi.fn(),
|
start: vi.fn(),
|
||||||
stop: vi.fn(),
|
stop: vi.fn(),
|
||||||
})),
|
})),
|
||||||
|
SelfHealingManager: vi.fn().mockImplementation((_store: unknown, opts: unknown) => {
|
||||||
|
capturedSelfHealingOpts = opts as Record<string, unknown>;
|
||||||
|
return {
|
||||||
|
start: mockSelfHealingStart,
|
||||||
|
stop: mockSelfHealingStop,
|
||||||
|
checkStuckBudget: mockCheckStuckBudget,
|
||||||
|
};
|
||||||
|
}),
|
||||||
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
|
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
|
||||||
cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0),
|
cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0),
|
||||||
};
|
};
|
||||||
@@ -574,6 +589,7 @@ describe("runDashboard — auto-merge pause exclusion", () => {
|
|||||||
|
|
||||||
beforeEach(async () => {
|
beforeEach(async () => {
|
||||||
capturedExecutorOpts = undefined;
|
capturedExecutorOpts = undefined;
|
||||||
|
capturedSelfHealingOpts = undefined;
|
||||||
vi.clearAllMocks();
|
vi.clearAllMocks();
|
||||||
resetGitHubMocks();
|
resetGitHubMocks();
|
||||||
mockStore = makeMockStore();
|
mockStore = makeMockStore();
|
||||||
@@ -759,6 +775,17 @@ describe("runDashboard — immediate resume on unpause", () => {
|
|||||||
expect(resumeOrphaned).toHaveBeenCalled();
|
expect(resumeOrphaned).toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("passes executor recovery callbacks into SelfHealingManager", async () => {
|
||||||
|
await runDashboard(0, { open: false });
|
||||||
|
|
||||||
|
expect(capturedSelfHealingOpts).toMatchObject({
|
||||||
|
rootDir: process.cwd(),
|
||||||
|
recoverCompletedTask: expect.any(Function),
|
||||||
|
getExecutingTaskIds: expect.any(Function),
|
||||||
|
});
|
||||||
|
expect(mockSelfHealingStart).toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
it("sweeps merge queue on unpause when autoMerge is enabled", async () => {
|
it("sweeps merge queue on unpause when autoMerge is enabled", async () => {
|
||||||
// Set up settings to return autoMerge: true for the drain queue check
|
// Set up settings to return autoMerge: true for the drain queue check
|
||||||
mockStore.getSettings.mockResolvedValue({
|
mockStore.getSettings.mockResolvedValue({
|
||||||
|
|||||||
@@ -568,7 +568,11 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
|
|||||||
});
|
});
|
||||||
|
|
||||||
// ── Self-healing: auto-unpause, stuck kill budgets, maintenance ─────
|
// ── Self-healing: auto-unpause, stuck kill budgets, maintenance ─────
|
||||||
const selfHealing = new SelfHealingManager(store, { rootDir: cwd });
|
const selfHealing = new SelfHealingManager(store, {
|
||||||
|
rootDir: cwd,
|
||||||
|
recoverCompletedTask: (task) => executorRef.current?.recoverCompletedTask(task) ?? Promise.resolve(false),
|
||||||
|
getExecutingTaskIds: () => executorRef.current?.getExecutingTaskIds() ?? new Set(),
|
||||||
|
});
|
||||||
|
|
||||||
// ── Stuck task detector: monitors agent sessions for stagnation ────
|
// ── Stuck task detector: monitors agent sessions for stagnation ────
|
||||||
// Created before the executor so it can be passed in options.
|
// Created before the executor so it can be passed in options.
|
||||||
|
|||||||
@@ -7527,12 +7527,10 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
const rootDir = scopedStore.getRootDir();
|
const rootDir = scopedStore.getRootDir();
|
||||||
const sha = task.mergeDetails.commitSha;
|
const sha = task.mergeDetails.commitSha;
|
||||||
|
|
||||||
// Resolve the diff base using the same priority as resolveDiffBase():
|
// Resolve the diff base:
|
||||||
// 1. task.baseCommitSha (exact starting commit of the worktree)
|
// 1. task.baseCommitSha (exact starting commit of the worktree)
|
||||||
// 2. merge-base between the merge commit and the base branch
|
// 2. First parent of the merge commit (safe for squash merges)
|
||||||
// 3. First parent of the merge commit as fallback
|
|
||||||
let mergeBase: string | undefined;
|
let mergeBase: string | undefined;
|
||||||
const baseBranch = task.baseBranch ?? "main";
|
|
||||||
|
|
||||||
// Priority 1: Use task.baseCommitSha if it's a valid ancestor of the merge commit
|
// Priority 1: Use task.baseCommitSha if it's a valid ancestor of the merge commit
|
||||||
if (task.baseCommitSha) {
|
if (task.baseCommitSha) {
|
||||||
@@ -7547,19 +7545,9 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Priority 2: Compute merge-base between merge commit and base branch
|
// Priority 2: Fall back to first parent of the merge commit (safe for squash merges)
|
||||||
if (!mergeBase) {
|
// This is more reliable than merge-base with baseBranch, which can return incorrect
|
||||||
try {
|
// results when baseBranch is another merged feature branch (for dependent tasks).
|
||||||
mergeBase = nodeChildProcess.execSync(
|
|
||||||
`git merge-base ${sha} origin/${baseBranch} 2>/dev/null || git merge-base ${sha} ${baseBranch}`,
|
|
||||||
{ cwd: rootDir, encoding: "utf-8", timeout: 5000 },
|
|
||||||
).trim();
|
|
||||||
} catch {
|
|
||||||
// merge-base with branch failed — fall through
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Priority 3: Fall back to first parent of the merge commit
|
|
||||||
if (!mergeBase) {
|
if (!mergeBase) {
|
||||||
try {
|
try {
|
||||||
mergeBase = nodeChildProcess.execSync(
|
mergeBase = nodeChildProcess.execSync(
|
||||||
@@ -7724,12 +7712,10 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
const rootDir = scopedStore.getRootDir();
|
const rootDir = scopedStore.getRootDir();
|
||||||
const sha = task.mergeDetails.commitSha;
|
const sha = task.mergeDetails.commitSha;
|
||||||
|
|
||||||
// Resolve the diff base using the same priority as resolveDiffBase():
|
// Resolve the diff base:
|
||||||
// 1. task.baseCommitSha (exact starting commit of the worktree)
|
// 1. task.baseCommitSha (exact starting commit of the worktree)
|
||||||
// 2. merge-base between the merge commit and the base branch
|
// 2. First parent of the merge commit (safe for squash merges)
|
||||||
// 3. First parent of the merge commit as fallback
|
|
||||||
let mergeBase: string | undefined;
|
let mergeBase: string | undefined;
|
||||||
const baseBranch = task.baseBranch ?? "main";
|
|
||||||
|
|
||||||
// Priority 1: Use task.baseCommitSha if it's a valid ancestor of the merge commit
|
// Priority 1: Use task.baseCommitSha if it's a valid ancestor of the merge commit
|
||||||
if (task.baseCommitSha) {
|
if (task.baseCommitSha) {
|
||||||
@@ -7744,19 +7730,9 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Priority 2: Compute merge-base between merge commit and base branch
|
// Priority 2: Fall back to first parent of the merge commit (safe for squash merges)
|
||||||
if (!mergeBase) {
|
// This is more reliable than merge-base with baseBranch, which can return incorrect
|
||||||
try {
|
// results when baseBranch is another merged feature branch (for dependent tasks).
|
||||||
mergeBase = nodeChildProcess.execSync(
|
|
||||||
`git merge-base ${sha} origin/${baseBranch} 2>/dev/null || git merge-base ${sha} ${baseBranch}`,
|
|
||||||
{ cwd: rootDir, encoding: "utf-8", timeout: 5000 },
|
|
||||||
).trim();
|
|
||||||
} catch {
|
|
||||||
// merge-base with branch failed — fall through
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Priority 3: Fall back to first parent of the merge commit
|
|
||||||
if (!mergeBase) {
|
if (!mergeBase) {
|
||||||
try {
|
try {
|
||||||
mergeBase = nodeChildProcess.execSync(
|
mergeBase = nodeChildProcess.execSync(
|
||||||
|
|||||||
@@ -251,6 +251,11 @@ export class TaskExecutor {
|
|||||||
/** Total count of currently spawned agents (across all parents). */
|
/** Total count of currently spawned agents (across all parents). */
|
||||||
private totalSpawnedCount = 0;
|
private totalSpawnedCount = 0;
|
||||||
|
|
||||||
|
/** Returns the set of task IDs currently being executed. */
|
||||||
|
getExecutingTaskIds(): Set<string> {
|
||||||
|
return new Set(this.executing);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param store — Task store instance (also used to listen for events)
|
* @param store — Task store instance (also used to listen for events)
|
||||||
* @param rootDir — Project root directory
|
* @param rootDir — Project root directory
|
||||||
@@ -374,9 +379,61 @@ export class TaskExecutor {
|
|||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Check whether a task's work is complete — all steps are done or skipped.
|
||||||
|
* Used to detect tasks that called task_done() but never transitioned to in-review
|
||||||
|
* (e.g., killed by stuck detector after task_done but before moveTask).
|
||||||
|
*/
|
||||||
|
private isTaskWorkComplete(task: Task): boolean {
|
||||||
|
if (task.steps.length === 0) return false;
|
||||||
|
return task.steps.every((s) => s.status === "done" || s.status === "skipped");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Fast-path a completed task directly to in-review without spawning a new agent.
|
||||||
|
* Captures modified files, runs workflow steps, and transitions the task.
|
||||||
|
*
|
||||||
|
* @returns true if the task was successfully transitioned, false otherwise.
|
||||||
|
*/
|
||||||
|
async recoverCompletedTask(task: Task): Promise<boolean> {
|
||||||
|
try {
|
||||||
|
const settings = await this.store.getSettings();
|
||||||
|
|
||||||
|
// Capture modified files if the worktree still exists
|
||||||
|
if (task.worktree && existsSync(task.worktree)) {
|
||||||
|
const modifiedFiles = this.captureModifiedFiles(task.worktree, task.baseCommitSha);
|
||||||
|
if (modifiedFiles.length > 0) {
|
||||||
|
await this.store.updateTask(task.id, { modifiedFiles });
|
||||||
|
executorLog.log(`${task.id}: recovered ${modifiedFiles.length} modified files`);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Run workflow steps before transitioning
|
||||||
|
const workflowSuccess = await this.runWorkflowSteps(task, task.worktree, settings);
|
||||||
|
if (!workflowSuccess) {
|
||||||
|
await this.store.updateTask(task.id, { status: "failed", error: "Workflow step failed during recovery" });
|
||||||
|
await this.store.moveTask(task.id, "in-review");
|
||||||
|
executorLog.log(`✗ ${task.id} workflow step failed during recovery → in-review`);
|
||||||
|
return true; // Still transitioned out of in-progress
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.store.moveTask(task.id, "in-review");
|
||||||
|
await this.store.logEntry(task.id, "Auto-recovered: task work was complete but stuck in in-progress — moved to in-review");
|
||||||
|
executorLog.log(`✓ ${task.id} auto-recovered completed task → in-review`);
|
||||||
|
this.options.onComplete?.(task);
|
||||||
|
return true;
|
||||||
|
} catch (err: any) {
|
||||||
|
executorLog.error(`Failed to recover completed task ${task.id}: ${err.message}`);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Resume orphaned in-progress tasks (e.g., after crash/restart).
|
* Resume orphaned in-progress tasks (e.g., after crash/restart).
|
||||||
* Call once after engine startup.
|
* Call once after engine startup.
|
||||||
|
*
|
||||||
|
* Tasks that are already complete (all steps done/skipped) are fast-pathed
|
||||||
|
* directly to in-review without spawning a new agent session.
|
||||||
*/
|
*/
|
||||||
async resumeOrphaned(): Promise<void> {
|
async resumeOrphaned(): Promise<void> {
|
||||||
const tasks = await this.store.listTasks();
|
const tasks = await this.store.listTasks();
|
||||||
@@ -388,6 +445,14 @@ export class TaskExecutor {
|
|||||||
|
|
||||||
executorLog.log(`Found ${inProgress.length} orphaned in-progress task(s)`);
|
executorLog.log(`Found ${inProgress.length} orphaned in-progress task(s)`);
|
||||||
for (const task of inProgress) {
|
for (const task of inProgress) {
|
||||||
|
// Fast-path: if the task already completed its work (all steps done),
|
||||||
|
// move it directly to in-review instead of re-executing from scratch.
|
||||||
|
if (this.isTaskWorkComplete(task)) {
|
||||||
|
executorLog.log(`${task.id} is already complete — fast-pathing to in-review`);
|
||||||
|
await this.recoverCompletedTask(task);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
executorLog.log(`Resuming ${task.id}: ${task.title || task.description.slice(0, 60)}`);
|
executorLog.log(`Resuming ${task.id}: ${task.title || task.description.slice(0, 60)}`);
|
||||||
try {
|
try {
|
||||||
await this.store.logEntry(task.id, "Resumed after engine restart");
|
await this.store.logEntry(task.id, "Resumed after engine restart");
|
||||||
|
|||||||
@@ -330,6 +330,83 @@ describe("In-progress task resume after restart", () => {
|
|||||||
expect(store.logEntry).toHaveBeenCalledWith("FN-040", "Resumed after engine restart");
|
expect(store.logEntry).toHaveBeenCalledWith("FN-040", "Resumed after engine restart");
|
||||||
expect(store.logEntry).toHaveBeenCalledWith("FN-041", "Resumed after engine restart");
|
expect(store.logEntry).toHaveBeenCalledWith("FN-041", "Resumed after engine restart");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("resumeOrphaned() fast-paths already-complete tasks to in-review", async () => {
|
||||||
|
const store = createMockStore();
|
||||||
|
const completedTask = makeTask("FN-963", "in-progress", {
|
||||||
|
worktree: "/tmp/wt/FN-963",
|
||||||
|
baseCommitSha: "base123",
|
||||||
|
steps: makeSteps("done", "done", "skipped"),
|
||||||
|
});
|
||||||
|
store.listTasks.mockResolvedValue([completedTask]);
|
||||||
|
store.getTask.mockResolvedValue(makeTaskDetail("FN-963", "in-progress", {
|
||||||
|
worktree: "/tmp/wt/FN-963",
|
||||||
|
baseCommitSha: "base123",
|
||||||
|
steps: makeSteps("done", "done", "skipped"),
|
||||||
|
}));
|
||||||
|
|
||||||
|
const executor = new TaskExecutor(store, "/tmp/test");
|
||||||
|
const executeSpy = vi.spyOn(executor, "execute");
|
||||||
|
|
||||||
|
mockedExecSync.mockImplementation((command) => {
|
||||||
|
const cmd = String(command);
|
||||||
|
if (cmd === 'git diff --name-only base123..HEAD') {
|
||||||
|
return "packages/dashboard/app/components/SettingsModal.tsx\n" as any;
|
||||||
|
}
|
||||||
|
return "" as any;
|
||||||
|
});
|
||||||
|
|
||||||
|
await executor.resumeOrphaned();
|
||||||
|
|
||||||
|
expect(executeSpy).not.toHaveBeenCalled();
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("FN-963", {
|
||||||
|
modifiedFiles: ["packages/dashboard/app/components/SettingsModal.tsx"],
|
||||||
|
});
|
||||||
|
expect(store.moveTask).toHaveBeenCalledWith("FN-963", "in-review");
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith(
|
||||||
|
"FN-963",
|
||||||
|
"Auto-recovered: task work was complete but stuck in in-progress — moved to in-review",
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("recoverCompletedTask() marks task failed then moves to in-review when workflow fails", async () => {
|
||||||
|
const store = createMockStore({
|
||||||
|
getTask: vi.fn().mockResolvedValue(makeTaskDetail("FN-963", "in-progress", {
|
||||||
|
worktree: "/tmp/wt/FN-963",
|
||||||
|
steps: makeSteps("done"),
|
||||||
|
enabledWorkflowSteps: ["wf-1"],
|
||||||
|
})),
|
||||||
|
getWorkflowStep: vi.fn().mockResolvedValue({
|
||||||
|
id: "wf-1",
|
||||||
|
name: "Build",
|
||||||
|
mode: "script",
|
||||||
|
scriptName: "pnpm test",
|
||||||
|
phase: "pre-merge",
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
const task = makeTask("FN-963", "in-progress", {
|
||||||
|
worktree: "/tmp/wt/FN-963",
|
||||||
|
steps: makeSteps("done"),
|
||||||
|
});
|
||||||
|
|
||||||
|
mockedExecSync.mockImplementation((command) => {
|
||||||
|
const cmd = String(command);
|
||||||
|
if (cmd === "pnpm test") {
|
||||||
|
throw new Error("tests failed");
|
||||||
|
}
|
||||||
|
return "" as any;
|
||||||
|
});
|
||||||
|
|
||||||
|
const executor = new TaskExecutor(store, "/tmp/test");
|
||||||
|
const recovered = await executor.recoverCompletedTask(task);
|
||||||
|
|
||||||
|
expect(recovered).toBe(true);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("FN-963", {
|
||||||
|
status: "failed",
|
||||||
|
error: "Workflow step failed during recovery",
|
||||||
|
});
|
||||||
|
expect(store.moveTask).toHaveBeenCalledWith("FN-963", "in-review");
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
// ── Step 3: In-review merge re-queue tests ────────────────────────────────
|
// ── Step 3: In-review merge re-queue tests ────────────────────────────────
|
||||||
|
|||||||
@@ -4,6 +4,16 @@ import type { Task, TaskStore, CentralCore } from "@fusion/core";
|
|||||||
import { InProcessRuntime } from "./in-process-runtime.js";
|
import { InProcessRuntime } from "./in-process-runtime.js";
|
||||||
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
||||||
|
|
||||||
|
const {
|
||||||
|
mockSelfHealingStart,
|
||||||
|
mockSelfHealingStop,
|
||||||
|
mockSelfHealingCtor,
|
||||||
|
} = vi.hoisted(() => ({
|
||||||
|
mockSelfHealingStart: vi.fn(),
|
||||||
|
mockSelfHealingStop: vi.fn(),
|
||||||
|
mockSelfHealingCtor: vi.fn(),
|
||||||
|
}));
|
||||||
|
|
||||||
// Mock the TaskStore class
|
// Mock the TaskStore class
|
||||||
vi.mock("@fusion/core", async () => {
|
vi.mock("@fusion/core", async () => {
|
||||||
const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
|
const actual = await vi.importActual<typeof import("@fusion/core")>("@fusion/core");
|
||||||
@@ -54,6 +64,18 @@ vi.mock("../scheduler.js", async () => {
|
|||||||
};
|
};
|
||||||
});
|
});
|
||||||
|
|
||||||
|
vi.mock("../self-healing.js", async () => {
|
||||||
|
return {
|
||||||
|
SelfHealingManager: vi.fn().mockImplementation((_store, opts) => {
|
||||||
|
mockSelfHealingCtor(opts);
|
||||||
|
return {
|
||||||
|
start: mockSelfHealingStart,
|
||||||
|
stop: mockSelfHealingStop,
|
||||||
|
};
|
||||||
|
}),
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
// Mock the executor
|
// Mock the executor
|
||||||
vi.mock("../executor.js", async () => {
|
vi.mock("../executor.js", async () => {
|
||||||
return {
|
return {
|
||||||
@@ -111,6 +133,19 @@ describe("InProcessRuntime", () => {
|
|||||||
expect(runtime.getStatus()).toBe("active");
|
expect(runtime.getStatus()).toBe("active");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("passes executor recovery callbacks into SelfHealingManager", async () => {
|
||||||
|
await runtime.start();
|
||||||
|
|
||||||
|
expect(mockSelfHealingCtor).toHaveBeenCalledWith(
|
||||||
|
expect.objectContaining({
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: expect.any(Function),
|
||||||
|
getExecutingTaskIds: expect.any(Function),
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
expect(mockSelfHealingStart).toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
it("should transition to 'stopped' after stop", async () => {
|
it("should transition to 'stopped' after stop", async () => {
|
||||||
await runtime.start();
|
await runtime.start();
|
||||||
await runtime.stop();
|
await runtime.stop();
|
||||||
|
|||||||
@@ -230,6 +230,8 @@ export class InProcessRuntime
|
|||||||
// 7. Initialize SelfHealingManager
|
// 7. Initialize SelfHealingManager
|
||||||
this.selfHealingManager = new SelfHealingManager(this.taskStore, {
|
this.selfHealingManager = new SelfHealingManager(this.taskStore, {
|
||||||
rootDir: this.config.workingDirectory,
|
rootDir: this.config.workingDirectory,
|
||||||
|
recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task),
|
||||||
|
getExecutingTaskIds: () => this.executor.getExecutingTaskIds(),
|
||||||
});
|
});
|
||||||
this.selfHealingManager.start();
|
this.selfHealingManager.start();
|
||||||
|
|
||||||
|
|||||||
@@ -386,4 +386,212 @@ describe("SelfHealingManager", () => {
|
|||||||
expect(result).toBe(0);
|
expect(result).toBe(0);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ── Completed task recovery ─────────────────────────────────────────
|
||||||
|
|
||||||
|
describe("recoverCompletedTasks", () => {
|
||||||
|
it("recovers tasks with all steps done that are stuck in in-progress", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-001",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [
|
||||||
|
{ status: "done" },
|
||||||
|
{ status: "done" },
|
||||||
|
{ status: "skipped" },
|
||||||
|
],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(1);
|
||||||
|
expect(recoverFn).toHaveBeenCalledWith(
|
||||||
|
expect.objectContaining({ id: "FN-001" }),
|
||||||
|
);
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips tasks that are actively executing", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set(["FN-001"]));
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-001",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [{ status: "done" }, { status: "done" }],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(0);
|
||||||
|
expect(recoverFn).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips tasks with incomplete steps", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-002",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [
|
||||||
|
{ status: "done" },
|
||||||
|
{ status: "in-progress" },
|
||||||
|
{ status: "pending" },
|
||||||
|
],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(0);
|
||||||
|
expect(recoverFn).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips paused tasks", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-003",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: true,
|
||||||
|
steps: [{ status: "done" }, { status: "done" }],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(0);
|
||||||
|
expect(recoverFn).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips tasks with no steps", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-004",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(0);
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns 0 when no recoverCompletedTask callback is provided", async () => {
|
||||||
|
// Default manager has no recovery callback
|
||||||
|
const result = await manager.recoverCompletedTasks();
|
||||||
|
expect(result).toBe(0);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("counts only successfully recovered tasks", async () => {
|
||||||
|
const recoverFn = vi.fn()
|
||||||
|
.mockResolvedValueOnce(true)
|
||||||
|
.mockResolvedValueOnce(false);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||||
|
{
|
||||||
|
id: "FN-005",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [{ status: "done" }],
|
||||||
|
},
|
||||||
|
{
|
||||||
|
id: "FN-006",
|
||||||
|
column: "in-progress",
|
||||||
|
paused: false,
|
||||||
|
steps: [{ status: "done" }],
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(1);
|
||||||
|
expect(recoverFn).toHaveBeenCalledTimes(2);
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns 0 when listTasks throws", async () => {
|
||||||
|
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||||
|
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||||
|
|
||||||
|
const managerWithRecovery = new SelfHealingManager(store, {
|
||||||
|
rootDir: "/tmp/test-project",
|
||||||
|
recoverCompletedTask: recoverFn,
|
||||||
|
getExecutingTaskIds: getExecuting,
|
||||||
|
});
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockRejectedValueOnce(new Error("DB error"));
|
||||||
|
|
||||||
|
const result = await managerWithRecovery.recoverCompletedTasks();
|
||||||
|
|
||||||
|
expect(result).toBe(0);
|
||||||
|
|
||||||
|
managerWithRecovery.stop();
|
||||||
|
});
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -16,7 +16,7 @@
|
|||||||
import { execSync } from "node:child_process";
|
import { execSync } from "node:child_process";
|
||||||
import { existsSync, readdirSync, statSync } from "node:fs";
|
import { existsSync, readdirSync, statSync } from "node:fs";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
import type { TaskStore, Settings } from "@fusion/core";
|
import type { TaskStore, Settings, Task } from "@fusion/core";
|
||||||
import { createLogger } from "./logger.js";
|
import { createLogger } from "./logger.js";
|
||||||
import { scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js";
|
import { scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js";
|
||||||
|
|
||||||
@@ -25,6 +25,21 @@ const log = createLogger("self-healing");
|
|||||||
export interface SelfHealingOptions {
|
export interface SelfHealingOptions {
|
||||||
/** Project root directory (parent of .worktrees/) */
|
/** Project root directory (parent of .worktrees/) */
|
||||||
rootDir: string;
|
rootDir: string;
|
||||||
|
/**
|
||||||
|
* Callback to recover a completed task that is stuck in in-progress.
|
||||||
|
* Called by the periodic maintenance cycle when it detects a task whose
|
||||||
|
* work is done but was never transitioned to in-review (e.g., killed by
|
||||||
|
* stuck detector after task_done but before moveTask).
|
||||||
|
*
|
||||||
|
* Should return true if the task was successfully transitioned out of
|
||||||
|
* in-progress, false if recovery failed.
|
||||||
|
*/
|
||||||
|
recoverCompletedTask?: (task: Task) => Promise<boolean>;
|
||||||
|
/**
|
||||||
|
* Returns the set of task IDs currently being executed by the executor.
|
||||||
|
* Used to avoid recovering tasks that are actively being worked on.
|
||||||
|
*/
|
||||||
|
getExecutingTaskIds?: () => Set<string>;
|
||||||
}
|
}
|
||||||
|
|
||||||
export class SelfHealingManager {
|
export class SelfHealingManager {
|
||||||
@@ -235,6 +250,7 @@ export class SelfHealingManager {
|
|||||||
await this.cleanupOrphanedBranches();
|
await this.cleanupOrphanedBranches();
|
||||||
this.checkpointWal();
|
this.checkpointWal();
|
||||||
await this.enforceWorktreeCap();
|
await this.enforceWorktreeCap();
|
||||||
|
await this.recoverCompletedTasks();
|
||||||
|
|
||||||
const elapsedMs = Date.now() - startMs;
|
const elapsedMs = Date.now() - startMs;
|
||||||
log.log(`Maintenance cycle completed in ${elapsedMs}ms`);
|
log.log(`Maintenance cycle completed in ${elapsedMs}ms`);
|
||||||
@@ -243,6 +259,55 @@ export class SelfHealingManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Completed task recovery ──────────────────────────────────────
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Recover tasks stuck in in-progress whose work is actually complete.
|
||||||
|
*
|
||||||
|
* This catches tasks where the agent called task_done() (all steps marked
|
||||||
|
* done, summary written) but the session was killed before the executor
|
||||||
|
* could call moveTask("in-review"). Without this, such tasks sit
|
||||||
|
* indefinitely in in-progress with no active session.
|
||||||
|
*
|
||||||
|
* @returns Number of tasks recovered
|
||||||
|
*/
|
||||||
|
async recoverCompletedTasks(): Promise<number> {
|
||||||
|
const recoverFn = this.options.recoverCompletedTask;
|
||||||
|
if (!recoverFn) return 0;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const tasks = await this.store.listTasks();
|
||||||
|
const executingIds = this.options.getExecutingTaskIds?.() ?? new Set<string>();
|
||||||
|
|
||||||
|
const stuckCompleted = tasks.filter((t) =>
|
||||||
|
t.column === "in-progress" &&
|
||||||
|
!t.paused &&
|
||||||
|
!executingIds.has(t.id) &&
|
||||||
|
t.steps.length > 0 &&
|
||||||
|
t.steps.every((s) => s.status === "done" || s.status === "skipped"),
|
||||||
|
);
|
||||||
|
|
||||||
|
if (stuckCompleted.length === 0) return 0;
|
||||||
|
|
||||||
|
log.warn(`Found ${stuckCompleted.length} completed task(s) stuck in in-progress`);
|
||||||
|
|
||||||
|
let recovered = 0;
|
||||||
|
for (const task of stuckCompleted) {
|
||||||
|
log.log(`Recovering completed task ${task.id}: ${task.title || task.description.slice(0, 60)}`);
|
||||||
|
const success = await recoverFn(task);
|
||||||
|
if (success) recovered++;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (recovered > 0) {
|
||||||
|
log.log(`Recovered ${recovered} completed task(s) → in-review`);
|
||||||
|
}
|
||||||
|
return recovered;
|
||||||
|
} catch (err: any) {
|
||||||
|
log.error(`Completed task recovery failed: ${err.message}`);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/** Run `git worktree prune` to clean stale metadata. */
|
/** Run `git worktree prune` to clean stale metadata. */
|
||||||
private async pruneWorktrees(): Promise<void> {
|
private async pruneWorktrees(): Promise<void> {
|
||||||
try {
|
try {
|
||||||
|
|||||||
Reference in New Issue
Block a user