Merge branch 'main' into timothyjlaurent/stuck-spinners
This commit is contained in:
gsxdsm
@@ -2505,6 +2505,19 @@ describe("buildExecutionPrompt", () => {
|
||||
expect(result).toContain('git commit -m "feat(FN-001): complete Step N — description" -m "Ref: runfusion/fusion#2915"');
|
||||
});
|
||||
|
||||
it("falls back to externalIssueId for commit source issue reference when issueNumber is missing", () => {
|
||||
const task = createMockTaskDetail({
|
||||
sourceIssue: {
|
||||
provider: "github",
|
||||
repository: "runfusion/fusion",
|
||||
externalIssueId: "2915",
|
||||
},
|
||||
} as any);
|
||||
|
||||
const result = buildExecutionPrompt(task, "/home/user/project");
|
||||
expect(result).toContain('git commit -m "feat(FN-001): complete Step N — description" -m "Ref: runfusion/fusion#2915"');
|
||||
});
|
||||
|
||||
it("omits source issue reference from commit instruction when sourceIssue is missing", () => {
|
||||
const task = createMockTaskDetail();
|
||||
const result = buildExecutionPrompt(task, "/home/user/project");
|
||||
@@ -3854,7 +3867,6 @@ describe("swallowed async store failure observability", () => {
|
||||
});
|
||||
|
||||
await (executor as any).terminateChildAgent("child-007");
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
await Promise.resolve();
|
||||
|
||||
expect(warnSpy).toHaveBeenCalledWith(
|
||||
@@ -10809,71 +10821,38 @@ describe("Agent Spawning - Child Termination", () => {
|
||||
expect(internals.totalSpawnedCount).toBe(0);
|
||||
});
|
||||
|
||||
it("terminateChildAgent auto-deletes agent after 5 second delay", async () => {
|
||||
vi.useFakeTimers();
|
||||
it("terminateChildAgent auto-deletes agent immediately", async () => {
|
||||
const agentStore = createMockAgentStore() as any;
|
||||
agentStore.deleteAgent = vi.fn().mockResolvedValue(undefined);
|
||||
const store = createMockStore();
|
||||
|
||||
try {
|
||||
const agentStore = createMockAgentStore() as any;
|
||||
// Add deleteAgent mock to the agent store
|
||||
agentStore.deleteAgent = vi.fn().mockResolvedValue(undefined);
|
||||
const store = createMockStore();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { agentStore } as any);
|
||||
const internals = executor as any;
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { agentStore } as any);
|
||||
const internals = executor as any;
|
||||
const mockSession = { dispose: vi.fn() };
|
||||
const childId = "agent-auto-delete-test";
|
||||
internals.childSessions.set(childId, mockSession);
|
||||
internals.totalSpawnedCount = 1;
|
||||
|
||||
const mockSession = { dispose: vi.fn() };
|
||||
const childId = "agent-auto-delete-test";
|
||||
internals.childSessions.set(childId, mockSession);
|
||||
internals.totalSpawnedCount = 1;
|
||||
await internals.terminateChildAgent(childId);
|
||||
|
||||
// Terminate the child
|
||||
const terminatePromise = internals.terminateChildAgent(childId);
|
||||
await terminatePromise;
|
||||
|
||||
// Session should be disposed immediately
|
||||
expect(mockSession.dispose).toHaveBeenCalled();
|
||||
expect(internals.pendingEphemeralDeletions.has(childId)).toBe(true);
|
||||
|
||||
// deleteAgent should not be called yet (before 5 seconds)
|
||||
expect(agentStore.deleteAgent).not.toHaveBeenCalled();
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// Now deleteAgent should have been called
|
||||
expect(agentStore.deleteAgent).toHaveBeenCalledTimes(1);
|
||||
expect(agentStore.deleteAgent).toHaveBeenCalledWith(childId);
|
||||
expect(internals.pendingEphemeralDeletions.has(childId)).toBe(false);
|
||||
|
||||
// Should not throw even when delete fails
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(mockSession.dispose).toHaveBeenCalled();
|
||||
expect(agentStore.deleteAgent).toHaveBeenCalledTimes(1);
|
||||
expect(agentStore.deleteAgent).toHaveBeenCalledWith(childId);
|
||||
expect(internals.pendingEphemeralDeletions.has(childId)).toBe(false);
|
||||
});
|
||||
|
||||
it("disposeEphemeralTimers clears pending spawned cleanup timers", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const agentStore = createMockAgentStore() as any;
|
||||
agentStore.deleteAgent = vi.fn().mockResolvedValue(undefined);
|
||||
const store = createMockStore();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { agentStore } as any);
|
||||
const internals = executor as any;
|
||||
it("disposeEphemeralTimers clears pending deletion bookkeeping", async () => {
|
||||
const agentStore = createMockAgentStore() as any;
|
||||
agentStore.deleteAgent = vi.fn().mockResolvedValue(undefined);
|
||||
const store = createMockStore();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { agentStore } as any);
|
||||
const internals = executor as any;
|
||||
|
||||
internals.childSessions.set("agent-dispose-test", { dispose: vi.fn() });
|
||||
internals.totalSpawnedCount = 1;
|
||||
await internals.terminateChildAgent("agent-dispose-test");
|
||||
expect(internals.pendingEphemeralDeletions.has("agent-dispose-test")).toBe(true);
|
||||
internals.pendingEphemeralDeletions.add("agent-dispose-test");
|
||||
executor.disposeEphemeralTimers();
|
||||
|
||||
executor.disposeEphemeralTimers();
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
expect(agentStore.deleteAgent).not.toHaveBeenCalled();
|
||||
expect(internals.pendingEphemeralDeletions.size).toBe(0);
|
||||
expect(internals.ephemeralCleanupTimers.size).toBe(0);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
expect(internals.pendingEphemeralDeletions.size).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -5362,38 +5362,62 @@ describe("aiMergeTask — merge details collection", () => {
|
||||
expect(mergeDetailsCall?.[1].mergeDetails.mergeCommitMessage).toBe("- feat: something");
|
||||
});
|
||||
|
||||
it("stores partial mergeDetails when branch is not found", async () => {
|
||||
it("recovers owned landed commit when branch is not found", async () => {
|
||||
const store = createMockStore(
|
||||
{ id: "FN-050", worktree: "/tmp/root/.worktrees/KB-050" },
|
||||
[{ id: "FN-050", worktree: "/tmp/root/.worktrees/KB-050", column: "in-review" } as Task],
|
||||
{
|
||||
id: "FN-3469",
|
||||
worktree: "/tmp/root/.worktrees/FN-3469",
|
||||
baseCommitSha: "base3469",
|
||||
mergeDetails: { commitSha: "a47b1e5d78d626f8b480f1e90d3d64be2625ff6a" } as any,
|
||||
},
|
||||
[{ id: "FN-3469", worktree: "/tmp/root/.worktrees/FN-3469", column: "in-review" } as Task],
|
||||
);
|
||||
|
||||
mockedExecSync.mockImplementation((cmd: any) => {
|
||||
const cmdStr = String(cmd);
|
||||
// Branch verification fails → branch not found
|
||||
if (cmdStr.includes("rev-parse --verify")) throw new Error("not found");
|
||||
// But rev-parse HEAD still works → can capture commitSha (encoding: utf-8 → string)
|
||||
if (cmdStr === "git rev-parse HEAD" || cmdStr.startsWith("git rev-parse HEAD "))
|
||||
return "existingheadsha999";
|
||||
if (cmdStr.includes("merge-base --is-ancestor a47b1e5d78d626f8b480f1e90d3d64be2625ff6a HEAD")) return Buffer.from("");
|
||||
if (cmdStr.includes("log -1 --format=%H%x1f%s%x1f%b a47b1e5d78d626f8b480f1e90d3d64be2625ff6a")) {
|
||||
return "a47b1e5d78d626f8b480f1e90d3d64be2625ff6a\u001ffix(FN-3469): title\u001fFusion-Task-Id: FN-3469" as any;
|
||||
}
|
||||
if (cmdStr.includes("show --shortstat --format= a47b1e5d78d626f8b480f1e90d3d64be2625ff6a")) {
|
||||
return "2 files changed, 84 insertions(+), 2 deletions(-)" as any;
|
||||
}
|
||||
return Buffer.from("");
|
||||
});
|
||||
|
||||
const result = await aiMergeTask(store, "/tmp/root", "FN-050");
|
||||
|
||||
const result = await aiMergeTask(store, "/tmp/root", "FN-3469");
|
||||
expect(result.merged).toBe(false);
|
||||
expect(result.error).toContain("not found");
|
||||
|
||||
// Find the updateTask call that set mergeDetails
|
||||
const updateCalls = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls;
|
||||
const mergeDetailsCall = updateCalls.find(
|
||||
const mergeDetailsCall = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls.find(
|
||||
(call: any[]) => call[1]?.mergeDetails !== undefined,
|
||||
);
|
||||
expect(mergeDetailsCall).toBeDefined();
|
||||
expect(mergeDetailsCall?.[1].mergeDetails).toEqual(expect.objectContaining({
|
||||
commitSha: "a47b1e5d78d626f8b480f1e90d3d64be2625ff6a",
|
||||
mergeCommitMessage: "fix(FN-3469): title",
|
||||
mergeConfirmed: true,
|
||||
}));
|
||||
});
|
||||
|
||||
const mergeDetails = mergeDetailsCall![1].mergeDetails;
|
||||
expect(mergeDetails.commitSha).toBe("existingheadsha999");
|
||||
expect(mergeDetails.mergedAt).toBeDefined();
|
||||
expect(mergeDetails.mergeConfirmed).toBe(false);
|
||||
it("does not persist misleading mergeDetails when branch is not found and no owned commit exists", async () => {
|
||||
const store = createMockStore(
|
||||
{ id: "FN-3373", worktree: "/tmp/root/.worktrees/FN-3373" },
|
||||
[{ id: "FN-3373", worktree: "/tmp/root/.worktrees/FN-3373", column: "in-review" } as Task],
|
||||
);
|
||||
|
||||
mockedExecSync.mockImplementation((cmd: any) => {
|
||||
const cmdStr = String(cmd);
|
||||
if (cmdStr.includes("rev-parse --verify")) throw new Error("not found");
|
||||
return Buffer.from("");
|
||||
});
|
||||
|
||||
const result = await aiMergeTask(store, "/tmp/root", "FN-3373");
|
||||
expect(result.merged).toBe(false);
|
||||
|
||||
const mergeDetailsCall = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls.find(
|
||||
(call: any[]) => call[1]?.mergeDetails !== undefined,
|
||||
);
|
||||
expect(mergeDetailsCall).toBeUndefined();
|
||||
});
|
||||
|
||||
it("completes merge even when git commands fail during merge details collection", async () => {
|
||||
@@ -5582,6 +5606,15 @@ describe("buildSourceIssueRef", () => {
|
||||
})).toBe("runfusion/fusion#123");
|
||||
});
|
||||
|
||||
it("falls back to externalIssueId when issueNumber is missing", async () => {
|
||||
const { buildSourceIssueRef } = await import("../merger.js");
|
||||
expect(buildSourceIssueRef({
|
||||
provider: "github",
|
||||
repository: "runfusion/fusion",
|
||||
externalIssueId: "321",
|
||||
} as any)).toBe("runfusion/fusion#321");
|
||||
});
|
||||
|
||||
it("returns empty string for non-GitHub providers", async () => {
|
||||
const { buildSourceIssueRef } = await import("../merger.js");
|
||||
expect(buildSourceIssueRef({
|
||||
@@ -5596,6 +5629,11 @@ describe("buildSourceIssueRef", () => {
|
||||
const { buildSourceIssueRef } = await import("../merger.js");
|
||||
expect(buildSourceIssueRef(undefined)).toBe("");
|
||||
expect(buildSourceIssueRef(null)).toBe("");
|
||||
expect(buildSourceIssueRef({
|
||||
provider: "github",
|
||||
repository: "runfusion/fusion",
|
||||
externalIssueId: "not-a-number",
|
||||
} as any)).toBe("");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -594,6 +594,153 @@ describe("SelfHealingManager", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("recoverStaleHeartbeatRuns", () => {
|
||||
function createMockAgentStore(activeRuns: Array<{ id: string; agentId: string; startedAt: string; processPid?: number; status?: string }>): {
|
||||
store: AgentStore;
|
||||
ended: Array<{ runId: string; status: string }>;
|
||||
saved: Array<Partial<{ id: string; status: string; stderrExcerpt: string }>>;
|
||||
} {
|
||||
const ended: Array<{ runId: string; status: string }> = [];
|
||||
const saved: Array<Partial<{ id: string; status: string; stderrExcerpt: string }>> = [];
|
||||
const detailById = new Map<string, any>();
|
||||
for (const r of activeRuns) {
|
||||
detailById.set(r.id, { id: r.id, agentId: r.agentId, startedAt: r.startedAt, endedAt: null, status: r.status ?? "active", processPid: r.processPid });
|
||||
}
|
||||
const agentStore = {
|
||||
listActiveHeartbeatRuns: vi.fn().mockResolvedValue(
|
||||
activeRuns.map((r) => ({ id: r.id, agentId: r.agentId, startedAt: r.startedAt, endedAt: null, status: "active" as const, processPid: r.processPid })),
|
||||
),
|
||||
getRunDetail: vi.fn().mockImplementation((_agentId: string, runId: string) => Promise.resolve(detailById.get(runId) ?? null)),
|
||||
saveRun: vi.fn().mockImplementation((run: any) => {
|
||||
saved.push({ id: run.id, status: run.status, stderrExcerpt: run.stderrExcerpt });
|
||||
return Promise.resolve();
|
||||
}),
|
||||
endHeartbeatRun: vi.fn().mockImplementation((runId: string, status: string) => {
|
||||
ended.push({ runId, status });
|
||||
return Promise.resolve();
|
||||
}),
|
||||
} as unknown as AgentStore;
|
||||
return { store: agentStore, ended, saved };
|
||||
}
|
||||
|
||||
it("returns 0 when no agentStore is configured", async () => {
|
||||
const result = await manager.recoverStaleHeartbeatRuns();
|
||||
expect(result).toBe(0);
|
||||
});
|
||||
|
||||
it("terminates active runs whose processPid does not match this process", async () => {
|
||||
const { store: agentStore, ended, saved } = createMockAgentStore([
|
||||
{ id: "run-orphan", agentId: "agent-a", startedAt: new Date().toISOString(), processPid: 999_999 },
|
||||
]);
|
||||
const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
|
||||
|
||||
const result = await m.recoverStaleHeartbeatRuns();
|
||||
|
||||
expect(result).toBe(1);
|
||||
expect(ended).toEqual([{ runId: "run-orphan", status: "terminated" }]);
|
||||
expect(saved[0]?.status).toBe("terminated");
|
||||
expect(saved[0]?.stderrExcerpt).toMatch(/Auto-recovered orphaned heartbeat run/);
|
||||
m.stop();
|
||||
});
|
||||
|
||||
it("leaves young runs from the current process alone", async () => {
|
||||
const { store: agentStore, ended } = createMockAgentStore([
|
||||
{ id: "run-mine", agentId: "agent-b", startedAt: new Date().toISOString(), processPid: process.pid },
|
||||
]);
|
||||
const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
|
||||
|
||||
const result = await m.recoverStaleHeartbeatRuns();
|
||||
|
||||
expect(result).toBe(0);
|
||||
expect(ended).toEqual([]);
|
||||
m.stop();
|
||||
});
|
||||
|
||||
it("terminates legacy active runs that have no recorded processPid", async () => {
|
||||
const { store: agentStore, ended } = createMockAgentStore([
|
||||
{ id: "run-legacy", agentId: "agent-c", startedAt: new Date().toISOString() },
|
||||
]);
|
||||
const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
|
||||
|
||||
const result = await m.recoverStaleHeartbeatRuns();
|
||||
|
||||
expect(result).toBe(1);
|
||||
expect(ended[0]?.runId).toBe("run-legacy");
|
||||
m.stop();
|
||||
});
|
||||
|
||||
it("terminates current-process runs that exceed the max-age threshold", async () => {
|
||||
const tooOld = new Date(Date.now() - 7 * 60 * 60 * 1000).toISOString(); // 7h ago
|
||||
const { store: agentStore, ended } = createMockAgentStore([
|
||||
{ id: "run-stuck", agentId: "agent-d", startedAt: tooOld, processPid: process.pid },
|
||||
]);
|
||||
const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
|
||||
|
||||
const result = await m.recoverStaleHeartbeatRuns();
|
||||
|
||||
expect(result).toBe(1);
|
||||
expect(ended[0]?.runId).toBe("run-stuck");
|
||||
m.stop();
|
||||
});
|
||||
|
||||
it("runStartupRecovery includes the stale heartbeat runs step", async () => {
|
||||
vi.mocked(store.getSettings).mockResolvedValue({
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
} as unknown as Settings);
|
||||
const spy = vi.spyOn(manager, "recoverStaleHeartbeatRuns").mockResolvedValue(0);
|
||||
|
||||
await manager.runStartupRecovery();
|
||||
|
||||
expect(spy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// Documents the race between recovery and a concurrent live startRun().
|
||||
// Sequence: recovery loads the stale row, then a fresh startRun() saves a
|
||||
// brand-new run for the same agent, then recovery calls endHeartbeatRun()
|
||||
// on the stale row. The new run must remain untouched — recovery must
|
||||
// only terminate the run id it sampled, never the agent's "any active
|
||||
// run." Otherwise we'd kill the very run we just spawned.
|
||||
it("only terminates the sampled run id even if a fresh run is started concurrently", async () => {
|
||||
const oldStarted = new Date(Date.now() - 7 * 60 * 60 * 1000).toISOString();
|
||||
const ended: Array<{ runId: string; status: string }> = [];
|
||||
const saved: Array<{ id: string; status: string }> = [];
|
||||
|
||||
const agentStore = {
|
||||
listActiveHeartbeatRuns: vi.fn().mockResolvedValue([
|
||||
{ id: "run-stale", agentId: "agent-x", startedAt: oldStarted, endedAt: null, status: "active", processPid: 999_999 },
|
||||
]),
|
||||
// Simulate the live process spawning a NEW run after recovery sampled the stale one
|
||||
// but before it called endHeartbeatRun. getRunDetail still returns the stale row
|
||||
// because the new run has a different id.
|
||||
getRunDetail: vi.fn().mockImplementation((_agentId: string, runId: string) => {
|
||||
if (runId === "run-stale") {
|
||||
return Promise.resolve({ id: "run-stale", agentId: "agent-x", startedAt: oldStarted, endedAt: null, status: "active", processPid: 999_999 });
|
||||
}
|
||||
return Promise.resolve(null);
|
||||
}),
|
||||
saveRun: vi.fn().mockImplementation((run: any) => {
|
||||
saved.push({ id: run.id, status: run.status });
|
||||
return Promise.resolve();
|
||||
}),
|
||||
endHeartbeatRun: vi.fn().mockImplementation((runId: string, status: string) => {
|
||||
ended.push({ runId, status });
|
||||
return Promise.resolve();
|
||||
}),
|
||||
} as unknown as AgentStore;
|
||||
|
||||
const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
|
||||
const result = await m.recoverStaleHeartbeatRuns();
|
||||
|
||||
expect(result).toBe(1);
|
||||
expect(ended).toEqual([{ runId: "run-stale", status: "terminated" }]);
|
||||
// The hypothetical concurrent run-fresh must not have been touched.
|
||||
expect(ended.some((e) => e.runId === "run-fresh")).toBe(false);
|
||||
expect(saved.every((s) => s.id === "run-stale")).toBe(true);
|
||||
m.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("recoverNoProgressNoTaskDoneFailures", () => {
|
||||
it("requeues clean in-progress no-task_done failures with no step progress", async () => {
|
||||
const managerWithRecovery = new SelfHealingManager(store, {
|
||||
@@ -3006,6 +3153,82 @@ describe("stale triage processing eviction before recovery", () => {
|
||||
|
||||
// ── Maintenance cycle concurrency ──────────────────────────────────
|
||||
|
||||
describe("recoverDoneTaskMergeMetadata", () => {
|
||||
it("upgrades done task metadata to an owned landed commit", async () => {
|
||||
const store = createMockStore();
|
||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-3469",
|
||||
column: "done",
|
||||
paused: false,
|
||||
baseCommitSha: "base",
|
||||
mergeDetails: { commitSha: "sharedsha", mergeConfirmed: false },
|
||||
modifiedFiles: ["AGENTS.md"],
|
||||
},
|
||||
]);
|
||||
|
||||
mockedExecSync.mockImplementation((command) => {
|
||||
const cmd = String(command);
|
||||
if (cmd.includes("merge-base --is-ancestor sharedsha HEAD")) return "" as any;
|
||||
if (cmd.includes("log -1 --format=%H%x1f%s%x1f%b sharedsha")) {
|
||||
return "sharedsha\u001ffix(FN-3468): other\u001fFusion-Task-Id: FN-3468" as any;
|
||||
}
|
||||
if (cmd.includes("Fusion-Task-Id: FN-3469")) {
|
||||
return "a47b1e5\u001ffix(FN-3469): correct lazy-loaded views\n" as any;
|
||||
}
|
||||
if (cmd.includes("show --shortstat --format= a47b1e5")) {
|
||||
return "2 files changed, 84 insertions(+), 2 deletions(-)" as any;
|
||||
}
|
||||
return "" as any;
|
||||
});
|
||||
|
||||
const repaired = await manager.recoverDoneTaskMergeMetadata();
|
||||
|
||||
expect(repaired).toBe(1);
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-3469", {
|
||||
mergeDetails: expect.objectContaining({
|
||||
commitSha: "a47b1e5",
|
||||
mergeConfirmed: true,
|
||||
}),
|
||||
});
|
||||
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("clears unowned shared SHA for done task when no owned landed commit exists", async () => {
|
||||
const store = createMockStore();
|
||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-3373",
|
||||
column: "done",
|
||||
paused: false,
|
||||
mergeDetails: { commitSha: "196adbd", mergeConfirmed: false },
|
||||
modifiedFiles: ["packages/cli/src/extension.ts"],
|
||||
},
|
||||
]);
|
||||
|
||||
mockedExecSync.mockImplementation((command) => {
|
||||
const cmd = String(command);
|
||||
if (cmd.includes("merge-base --is-ancestor 196adbd HEAD")) return "" as any;
|
||||
if (cmd.includes("log -1 --format=%H%x1f%s%x1f%b 196adbd")) {
|
||||
return "196adbd\u001ffeat(FN-3372): add safety net\u001fFusion-Task-Id: FN-3372" as any;
|
||||
}
|
||||
return "" as any;
|
||||
});
|
||||
|
||||
const repaired = await manager.recoverDoneTaskMergeMetadata();
|
||||
|
||||
expect(repaired).toBe(1);
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-3373", { mergeDetails: undefined });
|
||||
|
||||
manager.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("maintenance cycle concurrency", () => {
|
||||
let store: TaskStore & EventEmitter;
|
||||
let manager: SelfHealingManager;
|
||||
@@ -3178,13 +3401,14 @@ describe("maintenance cycle concurrency", () => {
|
||||
makeSlow("recoverOrphanedPlanningTasks");
|
||||
makeSlow("recoverGhostReviewTasks");
|
||||
makeSlow("recoverOrphanedAgents");
|
||||
makeSlow("recoverStaleHeartbeatRuns");
|
||||
|
||||
await (manager as any).runMaintenance();
|
||||
|
||||
// Operations run sequentially (one at a time), not in parallel.
|
||||
expect(maxConcurrent).toBe(1);
|
||||
// All operations should have run (including last one)
|
||||
expect(executionOrder[executionOrder.length - 1]).toBe("recoverOrphanedAgents");
|
||||
expect(executionOrder[executionOrder.length - 1]).toBe("recoverStaleHeartbeatRuns");
|
||||
});
|
||||
|
||||
it("one failing batch 2 operation does not abort the batch", async () => {
|
||||
@@ -3202,6 +3426,7 @@ describe("maintenance cycle concurrency", () => {
|
||||
"recoverOrphanedPlanningTasks",
|
||||
"recoverGhostReviewTasks",
|
||||
"recoverOrphanedAgents",
|
||||
"recoverStaleHeartbeatRuns",
|
||||
] as const;
|
||||
|
||||
// Make one operation fail
|
||||
|
||||
@@ -263,6 +263,7 @@ describe("buildSpecificationPrompt", () => {
|
||||
expect(prompt).toContain(feedback);
|
||||
expect(prompt).not.toContain("Existing Specification");
|
||||
expect(prompt).toContain("without carrying forward stale assumptions");
|
||||
expect(prompt).toContain("Treat the current task title and description as required primary inputs");
|
||||
});
|
||||
|
||||
it("includes attachments when provided", () => {
|
||||
@@ -1170,6 +1171,35 @@ describe("Re-specification flow", () => {
|
||||
|
||||
expect(revisionLogEntry?.outcome).toBe("Most recent feedback");
|
||||
});
|
||||
|
||||
it("prefers latest comment-triggered re-spec feedback log over legacy revision requests", () => {
|
||||
const taskWithCommentTriggeredFeedback: Task = {
|
||||
...taskWithRevisionRequest,
|
||||
log: [
|
||||
{
|
||||
timestamp: "2026-01-01T00:00:00.000Z",
|
||||
action: "AI spec revision requested",
|
||||
outcome: "Older feedback",
|
||||
},
|
||||
{
|
||||
timestamp: "2026-01-01T00:03:00.000Z",
|
||||
action: "User comment requested re-specification of planned task",
|
||||
outcome: "Latest feedback",
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const feedbackLogEntry = [...taskWithCommentTriggeredFeedback.log]
|
||||
.reverse()
|
||||
.find((entry) =>
|
||||
entry.action === "User comment requested re-specification of planned task"
|
||||
|| entry.action === "User comment invalidated spec approval — task needs re-specification"
|
||||
|| entry.action === "AI spec revision requested"
|
||||
);
|
||||
|
||||
expect(feedbackLogEntry?.outcome).toBe("Latest feedback");
|
||||
});
|
||||
|
||||
});
|
||||
|
||||
describe("requirePlanApproval setting", () => {
|
||||
|
||||
@@ -603,10 +603,8 @@ export class TaskExecutor {
|
||||
private completedTaskWatchdogs = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
/** One-shot watchdogs for workflow reruns that should have bounced back to in-progress. */
|
||||
private workflowRerunWatchdogs = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
/** Set of ephemeral spawned agent IDs with scheduled cleanup (prevents duplicate deletion attempts). */
|
||||
/** Set of ephemeral spawned agent IDs with in-flight cleanup (prevents duplicate deletion attempts). */
|
||||
private pendingEphemeralDeletions = new Set<string>();
|
||||
/** Map of spawned agent IDs to scheduled cleanup timer handles for shutdown disposal. */
|
||||
private ephemeralCleanupTimers = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
|
||||
private async finalizeAlreadyReviewedTask(taskId: string): Promise<"merged" | "blocked" | "missing"> {
|
||||
const latestTask = await this.store.getTask(taskId);
|
||||
@@ -736,15 +734,7 @@ export class TaskExecutor {
|
||||
}
|
||||
|
||||
disposeEphemeralTimers(): void {
|
||||
const timerCount = this.ephemeralCleanupTimers.size;
|
||||
for (const timerId of this.ephemeralCleanupTimers.values()) {
|
||||
clearTimeout(timerId);
|
||||
}
|
||||
this.ephemeralCleanupTimers.clear();
|
||||
this.pendingEphemeralDeletions.clear();
|
||||
if (timerCount > 0) {
|
||||
executorLog.log(`Cleared ${timerCount} pending spawned-agent cleanup timer(s)`);
|
||||
}
|
||||
}
|
||||
|
||||
private isBenignEphemeralDeleteRaceError(agentId: string, err: unknown): boolean {
|
||||
@@ -6656,23 +6646,17 @@ and show an appropriate message to the user.\`
|
||||
executorLog.warn(`Failed to update spawned child ${childId} state to 'terminated' during cleanup: ${msg}`);
|
||||
}
|
||||
|
||||
// Auto-delete the child agent after a short delay so the UI can observe
|
||||
// the terminal state before the agent is removed.
|
||||
this.pendingEphemeralDeletions.add(childId);
|
||||
const timerId = setTimeout(async () => {
|
||||
this.ephemeralCleanupTimers.delete(childId);
|
||||
this.pendingEphemeralDeletions.delete(childId);
|
||||
try {
|
||||
await this.options.agentStore?.deleteAgent(childId);
|
||||
} catch (err: unknown) {
|
||||
if (this.isBenignEphemeralDeleteRaceError(childId, err)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await this.options.agentStore?.deleteAgent(childId);
|
||||
} catch (err: unknown) {
|
||||
if (!this.isBenignEphemeralDeleteRaceError(childId, err)) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
executorLog.warn(`Failed to delete spawned agent ${childId}: ${msg}`);
|
||||
}
|
||||
}, 5000);
|
||||
this.ephemeralCleanupTimers.set(childId, timerId);
|
||||
} finally {
|
||||
this.pendingEphemeralDeletions.delete(childId);
|
||||
}
|
||||
|
||||
this.totalSpawnedCount = Math.max(0, this.totalSpawnedCount - 1);
|
||||
}
|
||||
@@ -6906,6 +6890,21 @@ function scopePromptToWorktree(prompt: string, rootDir?: string, worktreePath?:
|
||||
.replaceAll(`${worktreePath}/.fusion/`, `${rootDir}/.fusion/`);
|
||||
}
|
||||
|
||||
function buildSourceIssueRef(sourceIssue: TaskDetail["sourceIssue"]): string {
|
||||
if (!sourceIssue || sourceIssue.provider !== "github" || !sourceIssue.repository) {
|
||||
return "";
|
||||
}
|
||||
|
||||
const issueNumber = sourceIssue.issueNumber
|
||||
?? Number.parseInt(sourceIssue.externalIssueId ?? "", 10);
|
||||
|
||||
if (!Number.isInteger(issueNumber) || issueNumber < 1) {
|
||||
return "";
|
||||
}
|
||||
|
||||
return `${sourceIssue.repository}#${issueNumber}`;
|
||||
}
|
||||
|
||||
export function buildExecutionPrompt(task: TaskDetail, rootDir?: string, settings?: Settings, worktreePath?: string): string {
|
||||
const prompt = scopePromptToWorktree(task.prompt, rootDir, worktreePath);
|
||||
const reviewMatch = prompt.match(/##\s*Review Level[:\s]*(\d)/);
|
||||
@@ -6916,9 +6915,7 @@ export function buildExecutionPrompt(task: TaskDetail, rootDir?: string, setting
|
||||
? ` --author="${settings?.commitAuthorName || "Fusion"} <${settings?.commitAuthorEmail || "noreply@runfusion.ai"}>"`
|
||||
: "";
|
||||
|
||||
const sourceIssueRef = task.sourceIssue?.provider === "github" && task.sourceIssue.repository && task.sourceIssue.issueNumber
|
||||
? `${task.sourceIssue.repository}#${task.sourceIssue.issueNumber}`
|
||||
: "";
|
||||
const sourceIssueRef = buildSourceIssueRef(task.sourceIssue);
|
||||
|
||||
// Build step progress for resume
|
||||
const hasProgress = task.steps.length > 0 && task.steps.some((s) => s.status !== "pending");
|
||||
|
||||
@@ -46,6 +46,7 @@ import {
|
||||
type AgentPromptsConfig,
|
||||
type CanonicalMergeConflictStrategy,
|
||||
type TaskSourceIssue,
|
||||
type Task,
|
||||
} from "@fusion/core";
|
||||
import { describeModel, promptWithFallback } from "./pi.js";
|
||||
import { accumulateSessionTokenUsage } from "./session-token-usage.js";
|
||||
@@ -250,6 +251,72 @@ interface InferredTestCommand {
|
||||
buildSource?: "explicit" | "inferred";
|
||||
}
|
||||
|
||||
interface OwnedLandedCommit {
|
||||
sha: string;
|
||||
subject?: string;
|
||||
filesChanged?: number;
|
||||
insertions?: number;
|
||||
deletions?: number;
|
||||
}
|
||||
|
||||
function commitOwnedByTask(taskId: string, subject: string, body: string): boolean {
|
||||
return body.includes(`${FUSION_TASK_ID_TRAILER_KEY}: ${taskId}`) || subject.includes(taskId);
|
||||
}
|
||||
|
||||
async function findOwnedLandedCommitForTask(rootDir: string, task: Task): Promise<OwnedLandedCommit | null> {
|
||||
const tryHydrate = async (sha: string): Promise<OwnedLandedCommit | null> => {
|
||||
try {
|
||||
await execFileAsync("git", ["merge-base", "--is-ancestor", sha, "HEAD"], { cwd: rootDir });
|
||||
const { stdout } = await execFileAsync("git", ["log", "-1", "--format=%H%x1f%s%x1f%b", sha], {
|
||||
cwd: rootDir,
|
||||
encoding: "utf-8",
|
||||
});
|
||||
const [resolvedSha, subject = "", body = ""] = stdout.trim().split("\x1f");
|
||||
if (!resolvedSha || !commitOwnedByTask(task.id, subject, body)) return null;
|
||||
const owned: OwnedLandedCommit = { sha: resolvedSha, subject };
|
||||
try {
|
||||
const { stdout: statsOut } = await execFileAsync("git", ["show", "--shortstat", "--format=", resolvedSha], {
|
||||
cwd: rootDir,
|
||||
encoding: "utf-8",
|
||||
});
|
||||
Object.assign(owned, parseDiffStat(statsOut));
|
||||
} catch {
|
||||
// stats optional
|
||||
}
|
||||
return owned;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
};
|
||||
|
||||
if (task.mergeDetails?.commitSha) {
|
||||
const ownedStored = await tryHydrate(task.mergeDetails.commitSha);
|
||||
if (ownedStored) return ownedStored;
|
||||
}
|
||||
|
||||
const trailer = `${FUSION_TASK_ID_TRAILER_KEY}: ${task.id}`;
|
||||
const searches: string[][] = [
|
||||
["log", "--format=%H%x1f%s", "--max-count=20", "--fixed-strings", `--grep=${trailer}`, "HEAD"],
|
||||
["log", "--format=%H%x1f%s", "--max-count=20", "--fixed-strings", `--grep=${task.id}`, "HEAD"],
|
||||
];
|
||||
|
||||
for (const args of searches) {
|
||||
try {
|
||||
const { stdout } = await execFileAsync("git", args, { cwd: rootDir, encoding: "utf-8" });
|
||||
const first = stdout.trim().split("\n").find(Boolean);
|
||||
if (!first) continue;
|
||||
const [sha] = first.split("\x1f");
|
||||
if (!sha) continue;
|
||||
const owned = await tryHydrate(sha);
|
||||
if (owned) return owned;
|
||||
} catch {
|
||||
// continue
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Infer a default test command based on project files.
|
||||
* Returns the command and whether it was explicitly configured or inferred.
|
||||
@@ -1770,9 +1837,13 @@ function getCommitAuthorArg(settings: {
|
||||
}
|
||||
|
||||
export function buildSourceIssueRef(sourceIssue?: TaskSourceIssue | null): string {
|
||||
if (!sourceIssue || sourceIssue.provider !== "github") return "";
|
||||
if (!sourceIssue.repository || !sourceIssue.issueNumber) return "";
|
||||
return `${sourceIssue.repository}#${sourceIssue.issueNumber}`;
|
||||
if (!sourceIssue || sourceIssue.provider !== "github" || !sourceIssue.repository) return "";
|
||||
|
||||
const issueNumber = sourceIssue.issueNumber
|
||||
?? Number.parseInt(sourceIssue.externalIssueId ?? "", 10);
|
||||
|
||||
if (!Number.isInteger(issueNumber) || issueNumber < 1) return "";
|
||||
return `${sourceIssue.repository}#${issueNumber}`;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -2658,25 +2729,23 @@ export async function aiMergeTask(
|
||||
});
|
||||
} catch {
|
||||
result.error = `Branch '${branch}' not found — moving to done without merge`;
|
||||
// Best-effort: try to capture current HEAD commitSha even though branch is missing
|
||||
try {
|
||||
const commitSha = execSyncText("git rev-parse HEAD", {
|
||||
cwd: rootDir,
|
||||
stdio: "pipe",
|
||||
encoding: "utf-8",
|
||||
}).trim() || undefined;
|
||||
if (commitSha) {
|
||||
await store.updateTask(taskId, {
|
||||
mergeDetails: {
|
||||
commitSha,
|
||||
mergedAt: new Date().toISOString(),
|
||||
mergeConfirmed: false,
|
||||
},
|
||||
});
|
||||
mergerLog.log(`${taskId}: branch not found but captured commitSha ${commitSha.slice(0, 8)}`);
|
||||
}
|
||||
} catch {
|
||||
// No commit SHA available — task will show summary fallback
|
||||
// Branch is gone; never infer ownership from raw HEAD. Only persist commit
|
||||
// metadata when we can prove a landed commit belongs to this task.
|
||||
const ownedCommit = await findOwnedLandedCommitForTask(rootDir, task);
|
||||
if (ownedCommit) {
|
||||
await store.updateTask(taskId, {
|
||||
mergeDetails: {
|
||||
commitSha: ownedCommit.sha,
|
||||
filesChanged: ownedCommit.filesChanged,
|
||||
insertions: ownedCommit.insertions,
|
||||
deletions: ownedCommit.deletions,
|
||||
mergeCommitMessage: ownedCommit.subject,
|
||||
mergedAt: new Date().toISOString(),
|
||||
mergeConfirmed: true,
|
||||
prNumber: task.prInfo?.number,
|
||||
},
|
||||
});
|
||||
mergerLog.log(`${taskId}: branch missing; recovered owned landed commit ${ownedCommit.sha.slice(0, 8)}`);
|
||||
}
|
||||
// Audit trail: record merge completion (FN-1404)
|
||||
await audit.database({ type: "task:move", target: taskId, metadata: { to: "done", merged: false } });
|
||||
|
||||
@@ -747,7 +747,7 @@ describe("InProcessRuntime", () => {
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
it("auto-deletes task-worker agent on task completion after 5 second delay", async () => {
|
||||
it("auto-deletes task-worker agent on task completion immediately", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
try {
|
||||
@@ -774,14 +774,9 @@ describe("InProcessRuntime", () => {
|
||||
deleteAgentSpy.mockClear();
|
||||
executorOptions.onComplete?.({ id: "FN-AUTO1" } as Task);
|
||||
|
||||
// Verify deleteAgent was not called immediately (before 5 seconds)
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// Now deleteAgent should have been called
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
@@ -820,7 +815,7 @@ describe("InProcessRuntime", () => {
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
it("auto-deletes task-worker agent on task error after 5 second delay", async () => {
|
||||
it("auto-deletes task-worker agent on task error immediately", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
try {
|
||||
@@ -849,14 +844,9 @@ describe("InProcessRuntime", () => {
|
||||
deleteAgentSpy.mockClear();
|
||||
executorOptions.onError?.({ id: "FN-AUTO2" } as Task, new Error("Task failed"));
|
||||
|
||||
// Verify deleteAgent was not called immediately (before 5 seconds)
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// Now deleteAgent should have been called
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
@@ -1275,14 +1265,9 @@ describe("InProcessRuntime", () => {
|
||||
// Wait for async handler
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Verify deleteAgent was NOT called immediately (needs 5s delay)
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// Now deleteAgent should have been called
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
|
||||
|
||||
// Note: We verified deleteAgent was called, which is the key behavior.
|
||||
@@ -1319,8 +1304,7 @@ describe("InProcessRuntime", () => {
|
||||
// Wait for async handler
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Advance timers to ensure cleanup would have run
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// deleteAgent should NOT have been called for non-ephemeral agent
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
@@ -1361,11 +1345,9 @@ describe("InProcessRuntime", () => {
|
||||
// Wait for async handlers
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// deleteAgent should have been called only once (deduplicated)
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
@@ -1398,9 +1380,8 @@ describe("InProcessRuntime", () => {
|
||||
// Emit termination event
|
||||
store.emit("agent:stateChanged", agent.id, "running", "terminated");
|
||||
|
||||
// Wait for async handler, then fire delayed cleanup
|
||||
// Wait for async handler
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// Cleanup should still be attempted
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
@@ -1443,8 +1424,7 @@ describe("InProcessRuntime", () => {
|
||||
// Wait for async handler
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Advance timers to trigger deletion
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Should have attempted deletion
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
@@ -1463,7 +1443,7 @@ describe("InProcessRuntime", () => {
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
it("clears pending timers on runtime stop", async () => {
|
||||
it("handles runtime stop racing with in-flight cleanup", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
try {
|
||||
@@ -1488,14 +1468,9 @@ describe("InProcessRuntime", () => {
|
||||
// Wait for async handler
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Stop runtime before timer fires
|
||||
await runtime.stop();
|
||||
|
||||
// Advance timers - deletion should NOT happen because timer was cleared
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
|
||||
// deleteAgent should NOT have been called (timer was cleared)
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
expect(deleteAgentSpy.mock.calls.length).toBeLessThanOrEqual(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
@@ -1528,10 +1503,11 @@ describe("InProcessRuntime", () => {
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Advance timers by 5 seconds
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// deleteAgent should have been called for spawned ephemeral agent
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
@@ -1561,14 +1537,15 @@ describe("InProcessRuntime", () => {
|
||||
executorOptions.onComplete?.({ id: "FN-DUP-COMPLETE" } as Task);
|
||||
store.emit("agent:stateChanged", worker!.id, "running", "terminated");
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
await vi.waitFor(() => {
|
||||
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
it("clears onComplete cleanup timer on stop", async () => {
|
||||
it("handles onComplete cleanup racing with runtime stop", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
await runtime.start();
|
||||
@@ -1583,8 +1560,7 @@ describe("InProcessRuntime", () => {
|
||||
executorOptions.onComplete?.({ id: "FN-STOP-COMPLETE" } as Task);
|
||||
|
||||
await runtime.stop();
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(deleteAgentSpy).not.toHaveBeenCalled();
|
||||
expect(deleteAgentSpy.mock.calls.length).toBeLessThanOrEqual(1);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
|
||||
@@ -106,10 +106,8 @@ export class InProcessRuntime
|
||||
private triageProcessor?: TriageProcessor;
|
||||
private messageStore?: MessageStore;
|
||||
private concurrencyChangedListener?: (state: { globalMaxConcurrent: number }) => void;
|
||||
/** Set of agent IDs with scheduled ephemeral cleanup (prevents duplicate deletion) */
|
||||
/** Set of agent IDs with in-flight ephemeral cleanup (prevents duplicate deletion) */
|
||||
private pendingEphemeralDeletions = new Set<string>();
|
||||
/** Map of agent IDs to their cleanup timer IDs */
|
||||
private ephemeralCleanupTimers = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
/** Listener for agent:stateChanged events to clean up terminated ephemeral agents */
|
||||
private ephemeralTerminationListener?: (agentId: string, from: import("@fusion/core").AgentState, to: import("@fusion/core").AgentState) => void;
|
||||
/**
|
||||
@@ -459,11 +457,7 @@ export class InProcessRuntime
|
||||
});
|
||||
this.taskAgentMap.delete(task.id);
|
||||
if (!ephemeral) return;
|
||||
// Auto-delete the task-worker agent after a short delay so the UI
|
||||
// can observe the terminal state before the agent is removed.
|
||||
const timerId = setTimeout(async () => {
|
||||
this.ephemeralCleanupTimers.delete(agentId);
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
void (async () => {
|
||||
try {
|
||||
await this.agentStore?.deleteAgent(agentId);
|
||||
} catch (err: unknown) {
|
||||
@@ -472,9 +466,10 @@ export class InProcessRuntime
|
||||
}
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
runtimeLog.warn(`Failed to delete agent ${agentId} after completion: ${msg}`);
|
||||
} finally {
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
}
|
||||
}, 5000);
|
||||
this.ephemeralCleanupTimers.set(agentId, timerId);
|
||||
})();
|
||||
}
|
||||
},
|
||||
onError: (task, error) => {
|
||||
@@ -514,11 +509,7 @@ export class InProcessRuntime
|
||||
});
|
||||
this.taskAgentMap.delete(task.id);
|
||||
if (!ephemeral) return;
|
||||
// Auto-delete the task-worker agent after a short delay so the UI
|
||||
// can observe the terminal state before the agent is removed.
|
||||
const timerId = setTimeout(async () => {
|
||||
this.ephemeralCleanupTimers.delete(agentId);
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
void (async () => {
|
||||
try {
|
||||
await this.agentStore?.deleteAgent(agentId);
|
||||
} catch (err: unknown) {
|
||||
@@ -527,9 +518,10 @@ export class InProcessRuntime
|
||||
}
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
runtimeLog.warn(`Failed to delete agent ${agentId} after error: ${msg}`);
|
||||
} finally {
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
}
|
||||
}, 5000);
|
||||
this.ephemeralCleanupTimers.set(agentId, timerId);
|
||||
})();
|
||||
}
|
||||
},
|
||||
};
|
||||
@@ -622,22 +614,18 @@ export class InProcessRuntime
|
||||
if (!agent) return;
|
||||
if (!isEphemeralAgent(agent)) return;
|
||||
|
||||
// Schedule deletion after delay so UI can observe terminal state
|
||||
this.pendingEphemeralDeletions.add(agentId);
|
||||
const timerId = setTimeout(async () => {
|
||||
this.ephemeralCleanupTimers.delete(agentId);
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
try {
|
||||
await this.agentStore?.deleteAgent(agentId);
|
||||
} catch (err: unknown) {
|
||||
if (this.isBenignEphemeralDeleteRaceError(agentId, err)) {
|
||||
return;
|
||||
}
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
runtimeLog.warn(`Failed to delete ephemeral agent ${agentId} after termination: ${msg}`);
|
||||
try {
|
||||
await this.agentStore?.deleteAgent(agentId);
|
||||
} catch (err: unknown) {
|
||||
if (this.isBenignEphemeralDeleteRaceError(agentId, err)) {
|
||||
return;
|
||||
}
|
||||
}, 5000);
|
||||
this.ephemeralCleanupTimers.set(agentId, timerId);
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
runtimeLog.warn(`Failed to delete ephemeral agent ${agentId} after termination: ${msg}`);
|
||||
} finally {
|
||||
this.pendingEphemeralDeletions.delete(agentId);
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
runtimeLog.warn(`Failed to process termination event for agent ${agentId}: ${msg}`);
|
||||
@@ -912,12 +900,6 @@ export class InProcessRuntime
|
||||
this.ephemeralTerminationListener = undefined;
|
||||
runtimeLog.log("AgentStore agent:stateChanged listener removed");
|
||||
}
|
||||
// Clear any pending ephemeral cleanup timers to prevent leaks during shutdown
|
||||
for (const [agentId, timerId] of this.ephemeralCleanupTimers) {
|
||||
clearTimeout(timerId);
|
||||
runtimeLog.log(`Cleared pending cleanup timer for ephemeral agent ${agentId}`);
|
||||
}
|
||||
this.ephemeralCleanupTimers.clear();
|
||||
this.pendingEphemeralDeletions.clear();
|
||||
this.executor?.disposeEphemeralTimers();
|
||||
|
||||
|
||||
@@ -117,6 +117,10 @@ interface LandedTaskCommit {
|
||||
deletions?: number;
|
||||
}
|
||||
|
||||
function commitOwnedByTask(taskId: string, subject: string, body: string): boolean {
|
||||
return body.includes(`Fusion-Task-Id: ${taskId}`) || subject.includes(taskId);
|
||||
}
|
||||
|
||||
function shellQuote(value: string): string {
|
||||
return `'${value.replace(/'/g, "'\\''")}'`;
|
||||
}
|
||||
@@ -193,12 +197,14 @@ export class SelfHealingManager {
|
||||
{ name: "stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks().then(() => undefined) },
|
||||
{ name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) },
|
||||
{ name: "interrupted-merging", fn: () => this.recoverInterruptedMergingTasks().then(() => undefined) },
|
||||
{ name: "done-merge-metadata", fn: () => this.recoverDoneTaskMergeMetadata().then(() => undefined) },
|
||||
{ name: "misclassified-failures", fn: () => this.recoverMisclassifiedFailures().then(() => undefined) },
|
||||
{ name: "partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures().then(() => undefined) },
|
||||
{ name: "orphaned-executions", fn: () => this.recoverOrphanedExecutions().then(() => undefined) },
|
||||
{ name: "approved-triage", fn: () => this.recoverApprovedTriageTasks().then(() => undefined) },
|
||||
{ name: "orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks().then(() => undefined) },
|
||||
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents().then(() => undefined) },
|
||||
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns().then(() => undefined) },
|
||||
];
|
||||
|
||||
for (const step of steps) {
|
||||
@@ -467,18 +473,16 @@ export class SelfHealingManager {
|
||||
const storedSha = task.mergeDetails?.commitSha;
|
||||
if (storedSha) {
|
||||
try {
|
||||
// Reachable from HEAD? Use --quiet --exit-code on rev-list.
|
||||
await execAsync(
|
||||
`git merge-base --is-ancestor ${shellQuote(storedSha)} HEAD`,
|
||||
{ cwd: this.options.rootDir },
|
||||
);
|
||||
// Yes — fetch its subject + stats.
|
||||
const { stdout } = await execAsync(
|
||||
`git log -1 --format=%H%x1f%s ${shellQuote(storedSha)}`,
|
||||
`git log -1 --format=%H%x1f%s%x1f%b ${shellQuote(storedSha)}`,
|
||||
{ cwd: this.options.rootDir, maxBuffer: 1024 * 1024 },
|
||||
);
|
||||
const [sha, subject] = stdout.trim().split("\x1f");
|
||||
if (sha) {
|
||||
const [sha, subject = "", body = ""] = stdout.trim().split("\x1f");
|
||||
if (sha && commitOwnedByTask(task.id, subject, body)) {
|
||||
const commit: LandedTaskCommit = { sha, subject };
|
||||
try {
|
||||
const stats = await execAsync(`git show --shortstat --format= ${shellQuote(sha)}`, {
|
||||
@@ -644,6 +648,7 @@ export class SelfHealingManager {
|
||||
{ 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-done-merge-metadata", fn: () => this.recoverDoneTaskMergeMetadata() },
|
||||
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
|
||||
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
|
||||
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
|
||||
@@ -654,6 +659,7 @@ export class SelfHealingManager {
|
||||
{ name: "recover-orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks() },
|
||||
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
|
||||
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
|
||||
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
|
||||
];
|
||||
for (const fn of batch2Fns) {
|
||||
try {
|
||||
@@ -1218,6 +1224,59 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
async recoverDoneTaskMergeMetadata(): Promise<number> {
|
||||
try {
|
||||
const tasks = await this.store.listTasks({ column: "done", slim: true });
|
||||
const candidates = tasks.filter((task) => task.column === "done" && !task.paused && Boolean(task.mergeDetails?.commitSha));
|
||||
if (candidates.length === 0) return 0;
|
||||
|
||||
let repaired = 0;
|
||||
for (const task of candidates) {
|
||||
try {
|
||||
const landed = await this.findLandedTaskCommit(task);
|
||||
if (!landed) {
|
||||
if (task.mergeDetails?.mergeConfirmed === false) {
|
||||
await this.store.updateTask(task.id, { mergeDetails: undefined });
|
||||
await this.store.logEntry(task.id, "Auto-recovered: cleared unowned done-task mergeDetails commitSha");
|
||||
repaired++;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const needsRepair =
|
||||
task.mergeDetails?.commitSha !== landed.sha ||
|
||||
task.mergeDetails?.mergeConfirmed !== true ||
|
||||
task.mergeDetails?.filesChanged === undefined;
|
||||
|
||||
if (!needsRepair) continue;
|
||||
|
||||
await this.store.updateTask(task.id, {
|
||||
mergeDetails: {
|
||||
...task.mergeDetails,
|
||||
commitSha: landed.sha,
|
||||
filesChanged: landed.filesChanged,
|
||||
insertions: landed.insertions,
|
||||
deletions: landed.deletions,
|
||||
mergeCommitMessage: landed.subject,
|
||||
mergedAt: task.mergeDetails?.mergedAt ?? new Date().toISOString(),
|
||||
mergeConfirmed: true,
|
||||
prNumber: task.prInfo?.number,
|
||||
},
|
||||
});
|
||||
await this.store.logEntry(task.id, `Auto-recovered: reconciled done-task mergeDetails to owned commit ${landed.sha.slice(0, 8)}`);
|
||||
repaired++;
|
||||
} catch (err: unknown) {
|
||||
log.error(`Failed done-task merge metadata recovery for ${task.id}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
}
|
||||
|
||||
return repaired;
|
||||
} catch (err: unknown) {
|
||||
log.error(`Done-task merge metadata recovery failed: ${err instanceof Error ? err.message : String(err)}`);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Misclassified failure recovery ───────────────────────────────
|
||||
|
||||
/**
|
||||
@@ -1462,6 +1521,104 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Default cap (in ms) on how long an active heartbeat run from the current
|
||||
* process is allowed to remain open before self-healing will terminate it.
|
||||
* Six hours is well past any legitimate heartbeat tick (default 1 h
|
||||
* interval, configurable up to a few hours) so reaching this threshold
|
||||
* means the run record was never closed — typically a process that died
|
||||
* without our watchdog catching it.
|
||||
*/
|
||||
private static readonly STALE_ACTIVE_RUN_MAX_AGE_MS = 6 * 60 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* Terminate orphaned `agentRuns` rows left in `status = 'active'` by a
|
||||
* process that crashed before calling endHeartbeatRun(). These rows
|
||||
* silently break heartbeat scheduling: HeartbeatTriggerScheduler.onTimerTick
|
||||
* skips every tick that finds an active run, so the agent never gets called
|
||||
* again until something cleans up.
|
||||
*
|
||||
* A run is considered stale when:
|
||||
* - `processPid` was recorded and does not match the current `process.pid`
|
||||
* (i.e., the writer process is gone — guaranteed orphan), or
|
||||
* - `processPid` is missing (legacy data), or
|
||||
* - the run has been active for longer than STALE_ACTIVE_RUN_MAX_AGE_MS,
|
||||
* even from the current process (defense in depth against a writer that
|
||||
* leaks the row without crashing the whole runtime).
|
||||
*
|
||||
* The matching `processPid` + young run case is left alone — that is a
|
||||
* legitimately in-flight heartbeat.
|
||||
*/
|
||||
async recoverStaleHeartbeatRuns(): Promise<number> {
|
||||
const agentStore = this.options.agentStore;
|
||||
if (!agentStore) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
let activeRuns;
|
||||
try {
|
||||
activeRuns = await agentStore.listActiveHeartbeatRuns();
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Stale heartbeat run recovery — listing failed: ${errorMessage}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (activeRuns.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const currentPid = process.pid;
|
||||
const maxAgeMs = SelfHealingManager.STALE_ACTIVE_RUN_MAX_AGE_MS;
|
||||
let recovered = 0;
|
||||
|
||||
for (const run of activeRuns) {
|
||||
const startedMs = Date.parse(run.startedAt);
|
||||
const ageMs = Number.isFinite(startedMs) ? Math.max(0, now - startedMs) : Infinity;
|
||||
const recordedPid = run.processPid;
|
||||
|
||||
const pidMismatch = typeof recordedPid === "number" && recordedPid !== currentPid;
|
||||
const pidMissing = typeof recordedPid !== "number";
|
||||
const tooOld = ageMs >= maxAgeMs;
|
||||
|
||||
if (!pidMismatch && !pidMissing && !tooOld) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const reason = pidMismatch
|
||||
? `writer pid ${recordedPid} is no longer this process (current pid ${currentPid})`
|
||||
: pidMissing
|
||||
? `no processPid recorded`
|
||||
: `active for ${Math.round(ageMs / 1000)}s (>= ${Math.round(maxAgeMs / 1000)}s threshold)`;
|
||||
|
||||
try {
|
||||
const detail = await agentStore.getRunDetail(run.agentId, run.id);
|
||||
if (detail) {
|
||||
await agentStore.saveRun({
|
||||
...detail,
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "terminated",
|
||||
stderrExcerpt: `Auto-recovered orphaned heartbeat run: ${reason}`,
|
||||
});
|
||||
}
|
||||
await agentStore.endHeartbeatRun(run.id, "terminated");
|
||||
log.log(
|
||||
`Auto-recovered: orphan heartbeat run ${run.id} for ${run.agentId} (${reason})`,
|
||||
);
|
||||
recovered++;
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Failed to recover stale heartbeat run ${run.id} for ${run.agentId}: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (recovered > 0) {
|
||||
log.log(`Recovered ${recovered} stale heartbeat run(s)`);
|
||||
}
|
||||
return recovered;
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover `in-progress` tasks that failed only because the agent exited
|
||||
* without calling task_done, and where there is no sign of work to preserve.
|
||||
|
||||
@@ -471,7 +471,10 @@ export function createSkillsOverrideFromSelection(
|
||||
if (newDiagnostics.length > 0) {
|
||||
const _purpose = sessionPurpose ? `[${sessionPurpose}]` : "skills";
|
||||
for (const diag of newDiagnostics) {
|
||||
piLog.warn(`[skills] ${diag.type}: ${diag.message}`);
|
||||
const msg = `[skills] ${diag.type}: ${diag.message}`;
|
||||
if (diag.type === "error") piLog.error(msg);
|
||||
else if (diag.type === "warning") piLog.warn(msg);
|
||||
else piLog.log(msg);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1077,11 +1077,24 @@ export class TriageProcessor {
|
||||
let feedback: string | undefined;
|
||||
|
||||
if (isReplan) {
|
||||
// Extract feedback from the most recent "AI spec revision requested" log entry
|
||||
const revisionLogEntry = [...task.log]
|
||||
// Prefer explicit re-specification feedback logged by comment-triggered
|
||||
// and approval-invalidation flows; fall back to legacy revision logs.
|
||||
const feedbackLogEntry = [...task.log]
|
||||
.reverse()
|
||||
.find((entry) => entry.action === "AI spec revision requested");
|
||||
feedback = revisionLogEntry?.outcome;
|
||||
.find((entry) =>
|
||||
entry.action === "User comment requested re-specification of planned task"
|
||||
|| entry.action === "User comment invalidated spec approval — task needs re-specification"
|
||||
|| entry.action === "AI spec revision requested"
|
||||
);
|
||||
feedback = feedbackLogEntry?.outcome;
|
||||
|
||||
// Ensure the latest user feedback is always actionable for re-plans.
|
||||
if (!feedback) {
|
||||
const latestUserComment = [...(detail.comments || [])]
|
||||
.reverse()
|
||||
.find((comment) => comment.author === "user");
|
||||
feedback = latestUserComment?.text;
|
||||
}
|
||||
|
||||
planLog.log(
|
||||
`${task.id} re-planning with feedback: ${feedback?.slice(0, 100)}...`,
|
||||
@@ -2270,7 +2283,7 @@ Please revise the specification above to address this feedback. Write the comple
|
||||
## Re-specification Instructions
|
||||
You are creating a fresh replacement specification based on user feedback.
|
||||
|
||||
**Important:** Do not reuse stale PROMPT.md content. Start from the current task description, inspect the codebase, and write a complete new specification that addresses the feedback below.
|
||||
**Important:** Do not reuse stale PROMPT.md content. Treat the current task title and description as required primary inputs, inspect the codebase, and write a complete new specification that addresses the feedback below.
|
||||
|
||||
## User Feedback
|
||||
${feedback}
|
||||
@@ -2340,7 +2353,7 @@ ${task.breakIntoSubtasks ? "- **Break into subtasks:** Yes (user requested)" : "
|
||||
${task.dependencies.length > 0 ? `- **Dependencies:** ${task.dependencies.join(", ")}` : ""}${revisionSection}${subtaskSection}
|
||||
|
||||
## Instructions
|
||||
${isRevision ? "1. Review the existing specification and user feedback carefully\n2. Revise the PROMPT.md to address the feedback while maintaining the structure\n3. Ensure the specification is detailed enough for an AI agent to execute" : isFreshRespecification ? "1. Read the project structure to understand context (package.json, source files, etc.)\n2. Write a fresh complete PROMPT.md specification to the given path following the format in your system prompt\n3. Address the user feedback without carrying forward stale assumptions from the old spec\n4. Name actual files, functions, and patterns from the codebase — be specific" : "1. Read the project structure to understand context (package.json, source files, etc.)\n2. Write a complete PROMPT.md specification to the given path following the format in your system prompt\n3. The specification must be detailed enough for an autonomous AI agent to implement without asking questions\n4. Name actual files, functions, and patterns from the codebase — be specific"}
|
||||
${isRevision ? "1. Review the existing specification and user feedback carefully\n2. Revise the PROMPT.md to address the feedback while maintaining the structure\n3. Ensure the specification is detailed enough for an AI agent to execute" : isFreshRespecification ? "1. Read the project structure to understand context (package.json, source files, etc.)\n2. Treat the current task title and description as mandatory primary inputs for a new spec\n3. Write a fresh complete PROMPT.md specification to the given path following the format in your system prompt\n4. Address the user feedback without carrying forward stale assumptions from the old spec\n5. Name actual files, functions, and patterns from the codebase — be specific" : "1. Read the project structure to understand context (package.json, source files, etc.)\n2. Write a complete PROMPT.md specification to the given path following the format in your system prompt\n3. The specification must be detailed enough for an autonomous AI agent to implement without asking questions\n4. Name actual files, functions, and patterns from the codebase — be specific"}
|
||||
|
||||
Use the write tool to write the specification file.${commandsSection}${completionDocumentationSection}${memorySection}${attachmentsSection}${userCommentsSection}`;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user