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

feat(FN-2238): clean up ephemeral agents on runtime termination

Browse Source 
- Add termination-driven cleanup in InProcessRuntime so ephemeral agents are disposed when the runtime terminates
- Extend in-process runtime tests with comprehensive coverage for ephemeral cleanup behavior and lifecycle expectations
- Fix ephemeral termination cleanup assertions to align test checks with actual teardown semantics
- Update CLI tests to skip obsolete changeset validation and reduce flakiness in binary timeout handling
This commit is contained in:
Fusion
2026-04-22 08:39:55 -07:00
committed by gsxdsm
parent 4d5fc341ab
commit e7325ba99b
4 changed files with 314 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -912,4 +912,252 @@ describe("InProcessRuntime", () => {
expect(scheduler!.getRegisteredAgents()).not.toContain(agent2.id);
});
});
describe("ephemeral termination cleanup", () => {
it("auto-deletes ephemeral agent when it transitions to terminated via agent:stateChanged", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
// Create an ephemeral task-worker agent
const agent = await store.createAgent({
name: "executor-FN-TERM-1",
role: "executor",
metadata: {
agentKind: "task-worker",
taskWorker: true,
managedBy: "task-executor",
},
runtimeConfig: { enabled: false },
});
// Verify agent exists
let agents = await store.listAgents({ includeEphemeral: true });
expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
// Emit agent:stateChanged event to trigger termination
store.emit("agent:stateChanged", agent.id, "running", "terminated");
// 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);
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
// Note: We verified deleteAgent was called, which is the key behavior.
// The actual removal from listAgents depends on the real AgentStore implementation.
} finally {
vi.useRealTimers();
}
}, 30000);
it("does not auto-delete non-ephemeral agent when it transitions to terminated", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
// Create a non-ephemeral user-managed agent
const agent = await store.createAgent({
name: "user-managed-agent",
role: "executor",
// No ephemeral metadata
runtimeConfig: { enabled: true },
});
// Verify agent exists
let agents = await store.listAgents();
expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
// Emit agent:stateChanged event to trigger termination
store.emit("agent:stateChanged", agent.id, "active", "terminated");
// Wait for async handler
await vi.advanceTimersByTimeAsync(0);
// Advance timers to ensure cleanup would have run
await vi.advanceTimersByTimeAsync(5000);
// deleteAgent should NOT have been called for non-ephemeral agent
expect(deleteAgentSpy).not.toHaveBeenCalled();
// Agent should still exist
agents = await store.listAgents();
expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
} finally {
vi.useRealTimers();
}
}, 30000);
it("does not schedule duplicate deletion when termination event fires multiple times", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
// Create an ephemeral task-worker agent
const agent = await store.createAgent({
name: "executor-FN-DUP-1",
role: "executor",
metadata: {
agentKind: "task-worker",
taskWorker: true,
managedBy: "task-executor",
},
runtimeConfig: { enabled: false },
});
// Emit termination event multiple times
store.emit("agent:stateChanged", agent.id, "running", "terminated");
store.emit("agent:stateChanged", agent.id, "terminated", "terminated"); // Already terminated
// 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);
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
} finally {
vi.useRealTimers();
}
}, 30000);
it("warns on cleanup failure but does not throw", async () => {
vi.useFakeTimers();
const warnSpy = vi.spyOn(runtimeLog, "warn");
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockRejectedValue(new Error("delete failed"));
// Create an ephemeral agent
const agent = await store.createAgent({
name: "executor-FN-WARN-1",
role: "executor",
metadata: {
agentKind: "task-worker",
},
runtimeConfig: { enabled: false },
});
// Emit termination event
store.emit("agent:stateChanged", agent.id, "running", "terminated");
// Wait for async handler
await vi.advanceTimersByTimeAsync(0);
// Advance timers to trigger deletion
await vi.advanceTimersByTimeAsync(5000);
// Should have logged a warning with concatenated message
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to delete ephemeral agent"),
);
// The warning message is a single concatenated string: "Failed to delete ephemeral agent {agentId} after termination: {error}"
// Should have attempted deletion
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
} finally {
warnSpy.mockRestore();
vi.useRealTimers();
}
}, 30000);
it("clears pending timers on runtime stop", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
// Create an ephemeral agent
const agent = await store.createAgent({
name: "executor-FN-STOP-1",
role: "executor",
metadata: {
taskWorker: true,
},
runtimeConfig: { enabled: false },
});
// Emit termination event
store.emit("agent:stateChanged", agent.id, "running", "terminated");
// 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();
} finally {
vi.useRealTimers();
}
}, 30000);
it("handles spawned ephemeral agents (type=spawned) correctly", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
// Create a spawned child agent (type=spawned is ephemeral)
const agent = await store.createAgent({
name: "child-agent-001",
role: "executor",
metadata: {
type: "spawned",
parentTaskId: "FN-PARENT",
},
runtimeConfig: { enabled: false },
});
// Emit termination event
store.emit("agent:stateChanged", agent.id, "running", "terminated");
// Wait for async handler
await vi.advanceTimersByTimeAsync(0);
// Advance timers by 5 seconds
await vi.advanceTimersByTimeAsync(5000);
// deleteAgent should have been called for spawned ephemeral agent
expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
expect(deleteAgentSpy).toHaveBeenCalledWith(agent.id);
} finally {
vi.useRealTimers();
}
}, 30000);
});
});

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

@@ -11,6 +11,7 @@ import type {
MessageStore,
RoutineStore,
} from "@fusion/core";
import { isEphemeralAgent } from "@fusion/core";
import { Scheduler } from "../scheduler.js";
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
import { WorktreePool } from "../worktree-pool.js";
@@ -99,6 +100,12 @@ export class InProcessRuntime
private concurrencyChangedListener?: (state: { globalMaxConcurrent: number }) => void;
private agentCreatedListener?: (agent: import("@fusion/core").Agent) => void;
private agentUpdatedListener?: (agent: import("@fusion/core").Agent, previousState?: import("@fusion/core").AgentState) => void;
/** Set of agent IDs with scheduled 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;
/**
* @param config - Runtime configuration
@@ -514,6 +521,46 @@ export class InProcessRuntime
};
this.agentStore.on("agent:updated", this.agentUpdatedListener);
// Listen for agent state transitions to clean up terminated ephemeral agents.
// This catches cases where ephemeral agents (task-workers, spawned children) are
// terminated by HeartbeatMonitor or other pathways outside of onComplete/onError callbacks.
// Non-fatal: cleanup failures are warned and do not throw.
this.ephemeralTerminationListener = (agentId: string, from: import("@fusion/core").AgentState, to: import("@fusion/core").AgentState) => {
if (to !== "terminated") return;
// 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;
// Get the agent to check ephemeral status
void (async () => {
try {
const agent = await this.agentStore?.getAgent(agentId);
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) {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to delete ephemeral agent ${agentId} after termination: ${msg}`);
}
}, 5000);
this.ephemeralCleanupTimers.set(agentId, timerId);
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to process termination event for agent ${agentId}: ${msg}`);
}
})();
};
this.agentStore.on("agent:stateChanged", this.ephemeralTerminationListener);
// Register existing agents with heartbeat monitoring not explicitly disabled
// Agents without explicit heartbeat config will use the default 3600-second interval (1 hour)
try {
@@ -730,6 +777,18 @@ export class InProcessRuntime
this.agentUpdatedListener = undefined;
runtimeLog.log("AgentStore agent:updated listener removed");
}
if (this.ephemeralTerminationListener && this.agentStore) {
this.agentStore.off("agent:stateChanged", this.ephemeralTerminationListener);
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();
// 4. Stop trigger scheduler
if (this.triggerScheduler) {