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

feat(FN-3580): restore canonical agent lifecycle and remove terminated agen

Browse Source 
This merge restores the canonical agent lifecycle with termination scoped at the run level (FN-3580, 4 steps), adds sender-side wake recipient override for messages, and introduces test isolation CI enforcement with a stuck-requeue race fix. UI changes remove terminated-agent indicators from AgentDe

Fusion-Task-Id: FN-3580
This commit is contained in:
Fusion
2026-05-06 10:18:27 -07:00
committed by gsxdsm
parent 7f90308485
commit a31c4323a8
24 changed files with 216 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
packages/engine/src/__tests__/heartbeat-executor.test.ts
View File

@@ -297,19 +297,34 @@ describe("executeHeartbeat", () => {
expect(store.updateAgentState).not.toHaveBeenCalledWith("agent-001", "active");
});
it("completes with invalid_state when agent state is terminated", async () => {
const store = createStoreWithAgentForExec({ state: "terminated" });
it("completes with invalid_state when agent state is error", async () => {
const store = createStoreWithAgentForExec({ state: "error" });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(result).toBeDefined();
expect(result.status).toBe("completed");
expect(result.resultJson).toEqual({ reason: "invalid_state", state: "terminated" });
expect(result.resultJson).toEqual({ reason: "invalid_state", state: "error" });
expect(mockedCreateFnAgent).not.toHaveBeenCalled();
expect(store.updateAgentState).not.toHaveBeenCalledWith("agent-001", "active");
});
it("keeps terminated as a run status while pausing the agent", async () => {
const store = createStoreWithAgentForExec({ state: "running" });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const run = await monitor.startRun("agent-001", { source: "on_demand" });
await monitor.completeRun("agent-001", run.id, {
status: "terminated",
stderrExcerpt: "Run stopped by user",
});
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "running");
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(store.endHeartbeatRun).toHaveBeenCalledWith(run.id, "terminated");
});
it("completes as failed when agent not found in store", async () => {
const store = createStoreWithAgentForExec();
(store.getAgent as ReturnType<typeof vi.fn>).mockResolvedValue(null);

10
packages/engine/src/__tests__/heartbeat-scheduler.test.ts
View File

@@ -725,23 +725,23 @@ describe("HeartbeatTriggerScheduler", () => {
expect(callback).toHaveBeenCalledOnce();
});
it("timer is unregistered when agent becomes terminated (should clear timer)", async () => {
it("timer is unregistered when agent becomes paused (should clear timer)", async () => {
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 5000 });
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
// Update to terminated state
// Update to paused state
(eventStore.getAgent as ReturnType<typeof vi.fn>).mockImplementation((agentId: string) => ({
id: agentId,
name: `Agent ${agentId}`,
role: "executor" as const,
state: "terminated" as const,
state: "paused" as const,
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
metadata: {},
}));
eventStore.emit("agent:updated", { id: "agent-001", state: "terminated", metadata: {} } as import("@fusion/core").Agent);
eventStore.emit("agent:updated", { id: "agent-001", state: "paused", metadata: {} } as import("@fusion/core").Agent);
// Timer should be cleared for terminated agents
// Timer should be cleared for paused agents
expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
await vi.advanceTimersByTimeAsync(10000);

4
packages/engine/src/__tests__/heartbeat-session-prompt.test.ts
View File

@@ -560,7 +560,7 @@ describe("Budget Governance", () => {
expect(store.updateAgent).not.toHaveBeenCalledWith("agent-001", { pauseReason: "budget-exhausted" });
});
it("does not pause agent when run is terminated", async () => {
it("keeps terminated as a run status while pausing the agent", async () => {
const store = createCompleteRunBudgetStore({
budgetStatus: createBudgetStatus({ isOverBudget: true, isOverThreshold: true }),
});
@@ -572,7 +572,7 @@ describe("Budget Governance", () => {
});
expect(store.getBudgetStatus).not.toHaveBeenCalled();
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "terminated");
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(store.updateAgent).not.toHaveBeenCalledWith("agent-001", { pauseReason: "budget-exhausted" });
});

7
packages/engine/src/agent-heartbeat.ts
View File

@@ -14,7 +14,7 @@
* Callback pattern (not EventEmitter):
* - onMissed: Called when an agent misses its heartbeat
* - onRecovered: Called when an agent recovers after a missed heartbeat
* - onTerminated: Called when an unresponsive agent is terminated
* - onTerminated: Called when a heartbeat run is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, RunMutationContext, Settings, AgentConfigRevision, ReflectionStore } from "@fusion/core";
@@ -69,7 +69,7 @@ export interface HeartbeatMonitorOptions {
onMissed?: (agentId: string, reason: string) => void;
/** Callback when an agent recovers after a missed heartbeat */
onRecovered?: (agentId: string) => void;
/** Callback when an unresponsive agent is terminated */
/** Callback when a heartbeat run is terminated (run status only; agent state is handled separately). */
onTerminated?: (agentId: string, reason: string) => void;
/** Callback when a run starts */
onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
@@ -902,7 +902,7 @@ export class HeartbeatMonitor {
await this.store.updateAgentState(agentId, "error");
await this.store.updateAgent(agentId, { lastError: completionResult.stderrExcerpt ?? "Run failed" });
} else if (completionResult.status === "terminated") {
await this.store.updateAgentState(agentId, "terminated");
await this.store.updateAgentState(agentId, "paused");
} else {
// Completed successfully - back to active
await this.store.updateAgentState(agentId, "active");
@@ -2543,7 +2543,6 @@ const OVERDUE_FIRE_JITTER_MS = 5_000;
* States where timers should be cleared:
* - "paused" — Agent is paused by budget exhaustion or manual action
* - "error" — Agent encountered an error
* - "terminated" — Agent was explicitly stopped/terminated
*/
function isTickableState(state: Agent["state"]): boolean {
return state === "active" || state === "running" || state === "idle";

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

@@ -1258,8 +1258,8 @@ describe("InProcessRuntime", () => {
});
});
describe("ephemeral termination cleanup", () => {
it("auto-deletes ephemeral agent when it transitions to terminated via agent:stateChanged", async () => {
describe("ephemeral paused-state cleanup", () => {
it("auto-deletes ephemeral agent when it transitions to paused via agent:stateChanged", async () => {
vi.useFakeTimers();
try {
@@ -1284,7 +1284,7 @@ describe("InProcessRuntime", () => {
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
// Emit agent:stateChanged event to trigger cleanup
store.emit("agent:stateChanged", agent.id, "running", "paused");
// Wait for async handler
@@ -1302,7 +1302,7 @@ describe("InProcessRuntime", () => {
}
}, 30000);
it("does not auto-delete non-ephemeral agent when it transitions to terminated", async () => {
it("does not auto-delete non-ephemeral agent when it transitions to paused", async () => {
vi.useFakeTimers();
try {
@@ -1323,7 +1323,7 @@ describe("InProcessRuntime", () => {
let agents = await store.listAgents();
expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
// Emit agent:stateChanged event to trigger termination
// Emit agent:stateChanged event to trigger cleanup
store.emit("agent:stateChanged", agent.id, "active", "paused");
// Wait for async handler
@@ -1593,12 +1593,12 @@ describe("InProcessRuntime", () => {
});
describe("startup ephemeral sweep", () => {
it("cleans terminated ephemeral agents on startup", async () => {
it("cleans paused 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",
name: "orphan-paused",
role: "executor",
metadata: { agentKind: "task-worker" },
runtimeConfig: { enabled: false },