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

feat(FN-3352): improve in-process runtime cleanup reliability

Browse Source 
Adds comprehensive tests for cleanup reliability in the in-process runtime (131-line test suite) plus related coverage in the executor tests, with a small fix to the runtime implementation itself to address the reliability issue.

Fusion-Task-Id: FN-3352
This commit is contained in:
Fusion
2026-05-04 18:03:07 -07:00
committed by gsxdsm
parent 69c75feaea
commit cced0cec9a
4 changed files with 284 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,7 @@ const {
mockExecutorCtor,
mockResumeOrphaned,
mockTaskStoreSettings,
mockTaskStoreGetTask,
mockMessageStoreSetHook,
mockSchedulerConfigurePrMonitoring,
} = vi.hoisted(() => ({
@@ -28,6 +29,7 @@ const {
mockExecutorCtor: vi.fn(),
mockResumeOrphaned: vi.fn().mockResolvedValue(undefined),
mockTaskStoreSettings: {} as Record<string, unknown>,
mockTaskStoreGetTask: vi.fn().mockResolvedValue(null),
mockMessageStoreSetHook: vi.fn(),
mockSchedulerConfigurePrMonitoring: vi.fn(),
}));
@@ -52,6 +54,7 @@ vi.mock("@fusion/core", async () => {
self.getDatabase = vi.fn().mockReturnValue(mockDatabase);
self.init = vi.fn().mockResolvedValue(undefined);
self.listTasks = vi.fn().mockResolvedValue([]);
self.getTask = mockTaskStoreGetTask;
self.getSettings = vi.fn().mockImplementation(async () => structuredClone(mockTaskStoreSettings));
self.getMissionStore = vi.fn().mockReturnValue({
getMissionWithHierarchy: vi.fn().mockReturnValue(null),
@@ -153,6 +156,9 @@ vi.mock("../../executor.js", async () => {
self.getExecutingTaskIds = vi.fn().mockReturnValue(new Set());
self.handleLoopDetected = vi.fn().mockResolvedValue(false);
self.markStuckAborted = vi.fn();
self.abortAllSessionBash = vi.fn().mockResolvedValue(undefined);
self.isEphemeralDeletionPending = vi.fn().mockReturnValue(false);
self.disposeEphemeralTimers = vi.fn();
self.activeWorktrees = new Map();
return self;
}),
@@ -194,6 +200,8 @@ describe("InProcessRuntime", () => {
for (const key of Object.keys(mockTaskStoreSettings)) {
delete mockTaskStoreSettings[key];
}
mockTaskStoreGetTask.mockReset();
mockTaskStoreGetTask.mockResolvedValue(null);
// Create a unique temp directory for this test run
testDir = mkdtempSync(join(tmpdir(), `fn-test-${randomUUID().slice(0, 8)}-`));
@@ -1429,5 +1437,128 @@ describe("InProcessRuntime", () => {
vi.useRealTimers();
}
}, 30000);
it("does not double-delete when onComplete already scheduled cleanup", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onComplete?: (task: Task) => void;
};
executorOptions.onStart?.({ id: "FN-DUP-COMPLETE" } as Task, join(testDir, "worktree-FN-DUP-COMPLETE"));
let worker: Agent | undefined;
await vi.waitFor(async () => {
worker = (await store.listAgents({ includeEphemeral: true }))
.find((a: Agent) => a.name === "executor-FN-DUP-COMPLETE");
expect(worker).toBeDefined();
});
executorOptions.onComplete?.({ id: "FN-DUP-COMPLETE" } as Task);
store.emit("agent:stateChanged", worker!.id, "running", "terminated");
await vi.advanceTimersByTimeAsync(5000);
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
} finally {
vi.useRealTimers();
}
}, 30000);
it("clears onComplete cleanup timer on stop", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onComplete?: (task: Task) => void;
};
executorOptions.onStart?.({ id: "FN-STOP-COMPLETE" } as Task, join(testDir, "worktree-FN-STOP-COMPLETE"));
executorOptions.onComplete?.({ id: "FN-STOP-COMPLETE" } as Task);
await runtime.stop();
await vi.advanceTimersByTimeAsync(5000);
expect(deleteAgentSpy).not.toHaveBeenCalled();
} finally {
vi.useRealTimers();
}
}, 30000);
});
describe("startup ephemeral sweep", () => {
it("cleans terminated ephemeral agents on startup", async () => {
const { AgentStore } = await import("@fusion/core");
const preStore = new AgentStore({ rootDir: join(testDir, ".fusion") });
await preStore.init();
const orphan = await preStore.createAgent({
name: "orphan-terminated",
role: "executor",
metadata: { agentKind: "task-worker" },
runtimeConfig: { enabled: false },
});
await preStore.updateAgentState(orphan.id, "active");
await preStore.updateAgentState(orphan.id, "terminated");
await runtime.start();
const store = getAgentStore(runtime);
expect(await store.getAgent(orphan.id)).toBeNull();
}, 30000);
it("cleans ephemeral agents assigned to non-in-progress tasks", async () => {
mockTaskStoreGetTask.mockResolvedValue({ id: "FN-DONE", column: "done" });
const { AgentStore } = await import("@fusion/core");
const preStore = new AgentStore({ rootDir: join(testDir, ".fusion") });
await preStore.init();
const orphan = await preStore.createAgent({
name: "orphan-stale-task",
role: "executor",
metadata: { agentKind: "task-worker" },
runtimeConfig: { enabled: false },
});
await preStore.assignTask(orphan.id, "FN-DONE");
await runtime.start();
const store = getAgentStore(runtime);
expect(await store.getAgent(orphan.id)).toBeNull();
}, 30000);
it("continues startup sweep when one delete fails", async () => {
const warnSpy = vi.spyOn(runtimeLog, "warn");
const { AgentStore } = await import("@fusion/core");
const originalDeleteAgent = AgentStore.prototype.deleteAgent;
const deleteProtoSpy = vi
.spyOn(AgentStore.prototype, "deleteAgent")
.mockRejectedValueOnce(new Error("delete failed"))
.mockImplementation(async function(this: AgentStore, agentId: string) {
return originalDeleteAgent.call(this, agentId);
});
try {
const preStore = new AgentStore({ rootDir: join(testDir, ".fusion") });
await preStore.init();
const a1 = await preStore.createAgent({ name: "orphan-a1", role: "executor", metadata: { agentKind: "task-worker" }, runtimeConfig: { enabled: false } });
const a2 = await preStore.createAgent({ name: "orphan-a2", role: "executor", metadata: { agentKind: "task-worker" }, runtimeConfig: { enabled: false } });
await preStore.updateAgentState(a1.id, "active");
await preStore.updateAgentState(a1.id, "terminated");
await preStore.updateAgentState(a2.id, "active");
await preStore.updateAgentState(a2.id, "terminated");
await runtime.start();
const store = getAgentStore(runtime);
expect(runtime.getStatus()).toBe("active");
const remaining = await store.listAgents({ includeEphemeral: true });
expect(remaining.filter((a: Agent) => a.id === a1.id || a.id === a2.id)).toHaveLength(1);
expect(warnSpy).toHaveBeenCalled();
} finally {
warnSpy.mockRestore();
deleteProtoSpy.mockRestore();
}
}, 30000);
});
});

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

@@ -412,6 +412,9 @@ export class InProcessRuntime
// Update agent state to terminated (completed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
// Register pending deletion before flipping to terminated so
// agent:stateChanged listener doesn't schedule duplicate cleanup.
this.pendingEphemeralDeletions.add(agentId);
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (completion): ${msg}`);
@@ -419,12 +422,20 @@ export class InProcessRuntime
this.taskAgentMap.delete(task.id);
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
void setTimeout(() => {
this.agentStore?.deleteAgent(agentId).catch((err: unknown) => {
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 agent ${agentId} after completion: ${msg}`);
});
}
}, 5000);
this.ephemeralCleanupTimers.set(agentId, timerId);
}
},
onError: (task, error) => {
@@ -450,6 +461,9 @@ export class InProcessRuntime
// Update agent state to terminated (failed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
// Register pending deletion before flipping to terminated so
// agent:stateChanged listener doesn't schedule duplicate cleanup.
this.pendingEphemeralDeletions.add(agentId);
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (error): ${msg}`);
@@ -457,12 +471,20 @@ export class InProcessRuntime
this.taskAgentMap.delete(task.id);
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
void setTimeout(() => {
this.agentStore?.deleteAgent(agentId).catch((err: unknown) => {
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 agent ${agentId} after error: ${msg}`);
});
}
}, 5000);
this.ephemeralCleanupTimers.set(agentId, timerId);
}
},
};
@@ -541,8 +563,9 @@ export class InProcessRuntime
// Skip if already terminated (avoid re-scheduling)
if (from === "terminated") return;
// Check if already scheduled for deletion (e.g., by onComplete/onError callback)
if (this.pendingEphemeralDeletions.has(agentId)) return;
// Check if already scheduled for deletion (e.g., by onComplete/onError callback
// or TaskExecutor spawned-child cleanup).
if (this.pendingEphemeralDeletions.has(agentId) || this.executor?.isEphemeralDeletionPending(agentId)) return;
// Get the agent to check ephemeral status
void (async () => {
@@ -575,6 +598,57 @@ export class InProcessRuntime
};
this.agentStore.on("agent:stateChanged", this.ephemeralTerminationListener);
// Startup sweep for orphaned ephemeral agents from prior crashed/unclean runs.
// Non-fatal: best-effort cleanup that must not block runtime startup.
try {
const allAgents = await this.agentStore.listAgents({ includeEphemeral: true });
let cleanedCount = 0;
for (const agent of allAgents) {
if (!isEphemeralAgent(agent)) continue;
let shouldDelete = agent.state === "terminated" || agent.state === "error";
if (!shouldDelete && agent.taskId) {
try {
const task = await this.taskStore.getTask(agent.taskId);
if (!task || task.column !== "in-progress") {
shouldDelete = true;
}
} catch {
shouldDelete = true;
}
}
if (!shouldDelete) continue;
try {
if (agent.state !== "terminated") {
await this.agentStore.updateAgentState(agent.id, "terminated");
}
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Startup sweep failed to set ephemeral agent ${agent.id} terminated: ${msg}`);
}
try {
await this.agentStore.deleteAgent(agent.id);
cleanedCount += 1;
} catch (err: unknown) {
if (this.isBenignEphemeralDeleteRaceError(agent.id, err)) {
cleanedCount += 1;
continue;
}
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Startup sweep failed to delete ephemeral agent ${agent.id}: ${msg}`);
}
}
if (cleanedCount > 0) {
runtimeLog.log(`Startup ephemeral sweep cleaned ${cleanedCount} orphaned agent(s)`);
}
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Startup ephemeral sweep failed (continuing): ${msg}`);
}
// Register existing non-ephemeral, heartbeat-enabled agents in tickable states.
try {
const agents = await this.agentStore.listAgents();
@@ -797,6 +871,7 @@ export class InProcessRuntime
}
this.ephemeralCleanupTimers.clear();
this.pendingEphemeralDeletions.clear();
this.executor?.disposeEphemeralTimers();
// 4. Stop trigger scheduler
if (this.triggerScheduler) {