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

feat(FN-2501): merge fusion/fn-2501

Browse Source
This commit is contained in:
gsxdsm
2026-04-25 14:24:35 -07:00
parent 633644a665
commit 1a8058f334
10 changed files with 231 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/cli/CHANGELOG.md
View File

@@ -1,5 +1,11 @@
# @runfusion/fusion
## Unreleased
### Patch Changes
- FN-2501: Agent pause/resume controls now act immediately. Pausing stops an active heartbeat run right away, and resuming to `active` triggers an immediate on-demand heartbeat instead of waiting for the next timer tick.
## 0.2.7
### Patch Changes

7
packages/dashboard/app/components/AgentsView.tsx
View File

@@ -444,13 +444,6 @@ export function AgentsView({ addToast, projectId }: AgentsViewProps) {
try {
await updateAgentState(agentId, newState, projectId);
addToast(`Agent state updated to ${newState}`, "success");
if (newState === "active") {
try {
await startAgentRun(agentId, projectId);
} catch (runErr) {
addToast(`Agent activated, but failed to start run: ${getErrorMessage(runErr)}`, "error");
}
}
void loadAgents();
} catch (err) {
addToast(`Failed to update state: ${getErrorMessage(err)}`, "error");

28
packages/dashboard/app/components/__tests__/AgentsView.test.tsx
View File

@@ -1078,9 +1078,9 @@ describe("AgentsView", () => {
await waitFor(() => {
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-001", "active", undefined);
expect(mockStartAgentRun).toHaveBeenCalledWith("agent-001", undefined);
});
expect(mockStartAgentRun).not.toHaveBeenCalled();
expect(mockAddToast).toHaveBeenCalledWith(
expect.stringContaining("active"),
"success"
@@ -1114,7 +1114,7 @@ describe("AgentsView", () => {
});
});
it("can resume paused agent", async () => {
it("can resume paused agent without manual run trigger", async () => {
render(<AgentsView addToast={mockAddToast} />);
await waitFor(() => {
@@ -1125,8 +1125,9 @@ describe("AgentsView", () => {
await waitFor(() => {
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-003", "active", undefined);
expect(mockStartAgentRun).toHaveBeenCalledWith("agent-003", undefined);
});
expect(mockStartAgentRun).not.toHaveBeenCalled();
});
it("handles state change error gracefully", async () => {
@@ -1148,27 +1149,6 @@ describe("AgentsView", () => {
});
});
it("shows error toast when startAgentRun fails but still updates state", async () => {
mockStartAgentRun.mockRejectedValue(new Error("Run failed"));
render(<AgentsView addToast={mockAddToast} />);
await waitFor(() => {
expect(screen.getByTitle("Activate")).toBeTruthy();
});
fireEvent.click(screen.getByTitle("Activate"));
await waitFor(() => {
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-001", "active", undefined);
expect(mockStartAgentRun).toHaveBeenCalledWith("agent-001", undefined);
expect(mockAddToast).toHaveBeenCalledWith(
expect.stringContaining("failed to start run"),
"error"
);
});
});
it("does not start run when pausing agent", async () => {
render(<AgentsView addToast={mockAddToast} />);

70
packages/dashboard/src/__tests__/routes-agent-runs.test.ts
View File

@@ -118,6 +118,26 @@ describe("Agent runs routes (without HeartbeatMonitor)", () => {
vi.restoreAllMocks();
});
describe("POST /api/agents/:id/state", () => {
it("pausing with no active run remains successful", async () => {
mockGetActiveHeartbeatRun.mockResolvedValue(null);
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "paused" });
const response = await request(
app,
"POST",
"/api/agents/agent-001/state",
JSON.stringify({ state: "paused" }),
{ "content-type": "application/json" },
);
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "paused" });
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(mockGetActiveHeartbeatRun).not.toHaveBeenCalled();
});
});
describe("POST /api/agents/:id/runs", () => {
it("returns 201 with run record (fallback behavior without HeartbeatMonitor)", async () => {
const mockRun = createMockRun();
@@ -403,6 +423,56 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
vi.restoreAllMocks();
});
describe("POST /api/agents/:id/state", () => {
it("pauses by stopping active run before updating state", async () => {
mockGetActiveHeartbeatRun.mockResolvedValue(createMockRun({ id: "run-pause-1" }));
mockStopRun.mockResolvedValue(undefined);
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "paused" });
const response = await request(
app,
"POST",
"/api/agents/agent-001/state",
JSON.stringify({ state: "paused" }),
{ "content-type": "application/json" },
);
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "paused" });
expect(mockStopRun).toHaveBeenCalledWith("agent-001");
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(mockStopRun.mock.invocationCallOrder[0]).toBeLessThan(mockUpdateAgentState.mock.invocationCallOrder[0]);
});
it("resuming to active triggers immediate on-demand heartbeat exactly once", async () => {
mockGetAgent.mockResolvedValue({ id: "agent-001", state: "paused" });
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "active" });
mockExecuteHeartbeat.mockResolvedValue(createMockRun({ id: "run-resume-1", status: "completed" }));
const response = await request(
app,
"POST",
"/api/agents/agent-001/state",
JSON.stringify({ state: "active" }),
{ "content-type": "application/json" },
);
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "active" });
expect(mockExecuteHeartbeat).toHaveBeenCalledTimes(1);
expect(mockExecuteHeartbeat).toHaveBeenCalledWith({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "Triggered from state resume",
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from state resume",
triggerSource: "state-resume",
},
});
});
});
describe("POST /api/agents/:id/runs", () => {
it("delegates to heartbeatMonitor.executeHeartbeat when available", async () => {
const mockRun = createMockRun({ invocationSource: "on_demand", triggerDetail: "Triggered from dashboard" });

21
packages/dashboard/src/__tests__/routes.test.ts
View File

@@ -15740,6 +15740,27 @@ describe("Agent create/update routes", () => {
expect(res.body.error).toContain("soul must be at most 10,000 characters");
});
it("POST /api/agents/:id/state pauses successfully without heartbeat monitor wiring", async () => {
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: fusionDir });
await agentStore.init();
await agentStore.updateAgentState(agentId, "active");
const res = await REQUEST(
buildAgentApp(),
"POST",
`/api/agents/${agentId}/state`,
JSON.stringify({ state: "paused" }),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(200);
expect(res.body).toMatchObject({ id: agentId, state: "paused" });
const updatedAgent = await agentStore.getAgent(agentId);
expect(updatedAgent?.state).toBe("paused");
});
it("POST /api/agents/:id/state returns 400 for invalid state transitions", async () => {
const res = await REQUEST(
buildAgentApp(),

39
packages/dashboard/src/routes.ts
View File

@@ -13419,13 +13419,48 @@ async function persistImportedSkills(
throw badRequest("state is required");
}
const nextState = state as import("@fusion/core").AgentState;
const agentId = req.params.id;
const { store: scopedStore } = await getProjectContext(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
const agent = await agentStore.updateAgentState(req.params.id, state as import("@fusion/core").AgentState);
res.json(agent);
const currentAgent = await agentStore.getAgent(agentId);
if (!currentAgent) {
throw notFound("Agent not found");
}
const projectHeartbeatMonitor = hasHeartbeatExecutor
&& heartbeatMonitor
&& isHeartbeatMonitorForProject(scopedStore)
? heartbeatMonitor
: null;
if (nextState === "paused" && projectHeartbeatMonitor) {
const activeRun = await agentStore.getActiveHeartbeatRun(agentId);
if (activeRun) {
await projectHeartbeatMonitor.stopRun(agentId);
}
}
const updatedAgent = await agentStore.updateAgentState(agentId, nextState);
if (nextState === "active" && projectHeartbeatMonitor) {
await projectHeartbeatMonitor.executeHeartbeat({
agentId,
source: "on_demand",
triggerDetail: "Triggered from state resume",
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from state resume",
triggerSource: "state-resume",
},
});
}
res.json(updatedAgent);
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;

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

@@ -459,6 +459,27 @@ describe("InProcessRuntime", () => {
expect(registeredAgents).toContain(createdAgent.id);
});
it("does not register paused agents on startup", async () => {
await runtime.start();
const store = getAgentStore(runtime);
const pausedAgent = await store.createAgent({
name: "Paused Agent",
role: "executor",
runtimeConfig: { heartbeatIntervalMs: 30000, enabled: true },
});
await store.updateAgentState(pausedAgent.id, "active");
await store.updateAgentState(pausedAgent.id, "paused");
await runtime.stop();
runtime = new InProcessRuntime(buildTestConfig(testDir), mockCentralCore);
await runtime.start();
const scheduler = runtime.getTriggerScheduler();
expect(scheduler).toBeDefined();
expect(scheduler!.getRegisteredAgents()).not.toContain(pausedAgent.id);
});
it("routes assignment triggers through executeHeartbeat", async () => {
await runtime.start();
@@ -885,6 +906,59 @@ describe("InProcessRuntime", () => {
expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
});
it("clears timers on pause and re-arms from resume without stale pre-pause firing", async () => {
const store = getAgentStore(runtime);
const monitor = runtime.getHeartbeatMonitor();
expect(monitor).toBeDefined();
const executeHeartbeatSpy = vi
.spyOn(monitor!, "executeHeartbeat")
.mockResolvedValue({ id: "run-resume-test" } as any);
const agent = await store.createAgent({
name: "resume-timer-agent",
role: "executor",
runtimeConfig: {
enabled: true,
heartbeatIntervalMs: 1000,
},
});
await store.updateAgentState(agent.id, "active");
const scheduler = runtime.getTriggerScheduler();
expect(scheduler).toBeDefined();
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
await vi.advanceTimersByTimeAsync(400);
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
await store.updateAgentState(agent.id, "paused");
expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
// Advance beyond the original tick window; stale pre-pause timer must not fire.
await vi.advanceTimersByTimeAsync(800);
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
await store.updateAgentState(agent.id, "active");
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
// Resume should start a fresh interval from now, not from pre-pause start.
await vi.advanceTimersByTimeAsync(900);
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(100);
await vi.waitFor(() => {
expect(executeHeartbeatSpy).toHaveBeenCalledTimes(1);
});
expect(executeHeartbeatSpy).toHaveBeenCalledWith(
expect.objectContaining({
agentId: agent.id,
source: "timer",
}),
);
});
it("removes event listeners when runtime is stopped", async () => {
// Create a new agent before stopping
const store = getAgentStore(runtime);

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

@@ -492,15 +492,19 @@ export class InProcessRuntime
);
this.triggerScheduler.start();
// Dynamic registration follows per-agent heartbeat enablement.
// Non-ephemeral agents are managed unless runtimeConfig.enabled is
// explicitly false. Ephemeral/task-worker agents are never armed.
// Dynamic registration follows per-agent heartbeat enablement and tickable state.
// Non-ephemeral agents are managed unless runtimeConfig.enabled is explicitly false.
// Paused/error/terminated states are never timer-armed.
const isHeartbeatEnabledAgent = (agent: import("@fusion/core").Agent) =>
!isEphemeralAgent(agent) && agent.runtimeConfig?.enabled !== false;
const isTickableHeartbeatState = (state: import("@fusion/core").AgentState) =>
state === "active" || state === "running" || state === "idle";
const isTimerManagedAgent = (agent: import("@fusion/core").Agent) =>
isHeartbeatEnabledAgent(agent) && isTickableHeartbeatState(agent.state);
this.agentCreatedListener = (agent) => {
if (!this.triggerScheduler) return;
if (!isHeartbeatEnabledAgent(agent)) return;
if (!isTimerManagedAgent(agent)) return;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
@@ -512,7 +516,7 @@ export class InProcessRuntime
this.agentUpdatedListener = (agent) => {
if (!this.triggerScheduler) return;
if (!isHeartbeatEnabledAgent(agent)) {
if (!isTimerManagedAgent(agent)) {
this.triggerScheduler.unregisterAgent(agent.id);
runtimeLog.log(`Unregistered agent ${agent.id} from heartbeat triggers`);
return;
@@ -569,12 +573,12 @@ export class InProcessRuntime
};
this.agentStore.on("agent:stateChanged", this.ephemeralTerminationListener);
// Register existing non-ephemeral agents with heartbeat enabled.
// Register existing non-ephemeral, heartbeat-enabled agents in tickable states.
try {
const agents = await this.agentStore.listAgents();
let registeredCount = 0;
for (const agent of agents) {
if (!isHeartbeatEnabledAgent(agent)) continue;
if (!isTimerManagedAgent(agent)) continue;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,