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

feat(FN-2920): improve remote tunnel setup and heartbeat scheduling

Browse Source 
- Add cloudflared install/detection support in remote settings API, UI, and route tests
- Surface Cloudflare tunnel prerequisites in Settings modal with remote access docs updates
- Harden heartbeat runtime scheduling by avoiding stale timeout state and simplifying runtime timeout handling
- Expand CLI/core/dashboard/engine coverage for task lifecycle, agent health, and runtime heartbeat behavior
- Add changesets for heartbeat scheduling fixes and PR approval setting updates

Fusion-Task-Id: FN-2920
This commit is contained in:
Fusion
2026-04-29 13:49:05 -07:00
committed by gsxdsm
parent b91533ce43
commit 17a072c924
25 changed files with 819 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
packages/engine/src/__tests__/agent-heartbeat.test.ts
View File

@@ -4217,13 +4217,9 @@ describe("HeartbeatTriggerScheduler", () => {
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
});
it("registers regardless of the legacy enabled flag (state is the source of truth)", () => {
// runtimeConfig.enabled is no longer honored by the scheduler — pause
// and resume happen through agent.state, and the agent:updated listener
// drives register/unregister. Callers that still pass `enabled: false`
// should not silently lose the timer.
it("does not register when heartbeat is explicitly disabled", () => {
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 10000, enabled: false });
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
});
it("applies default 3600-second interval when intervalMs is undefined", async () => {

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

@@ -17,7 +17,7 @@
* - onTerminated: Called when an unresponsive agent is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot, RunMutationContext, Settings } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot, RunMutationContext, Settings, AgentConfigRevision } from "@fusion/core";
import { buildExecutionMemoryInstructions, isEphemeralAgent, hasAgentIdentity } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
@@ -1750,11 +1750,14 @@ function isHeartbeatManaged(agent: Agent): boolean {
/**
* HeartbeatTriggerScheduler manages timer-based heartbeat triggers for agents.
*
* State is the source of truth: state ∈ {active, running} on a non-ephemeral
* agent arms the timer; any other state or any ephemeral agent doesn't. The
* `runtimeConfig.enabled` flag is no longer consulted here — pause/resume
* happens through `agent.state`, and the `agent:updated` listener arms or
* clears the timer on transitions.
* Timers are armed only for durable agents where all of the following hold:
* - `runtimeConfig.enabled !== false`
* - `state ∈ {active, running, idle}`
*
* Any other state, or any ephemeral/task-worker agent, clears the timer.
* State changes and heartbeat config updates are observed via AgentStore
* lifecycle events, while callers can still explicitly register existing
* agents during startup bootstrap.
*
* Other config knobs still apply:
* - `heartbeatIntervalMs`: Timer interval (default 1h)
@@ -1768,7 +1771,9 @@ export class HeartbeatTriggerScheduler {
private registrationEpochs: Map<string, number> = new Map();
private running = false;
private assignedListener: ((agent: import("@fusion/core").Agent, taskId: string) => void) | null = null;
private createdListener: ((agent: import("@fusion/core").Agent) => void) | null = null;
private updatedListener: ((agent: import("@fusion/core").Agent) => void) | null = null;
private configRevisionListener: ((agentId: string, revision: AgentConfigRevision) => void) | null = null;
private deletedListener: ((agentId: string) => void) | null = null;
constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore) {
@@ -1779,7 +1784,7 @@ export class HeartbeatTriggerScheduler {
/**
* Start the scheduler. Enables assignment watching.
* Individual agents must be registered separately via registerAgent().
* Existing agents still need one startup bootstrap pass via registerAgent().
*/
start(): void {
if (this.running) return;
@@ -1826,9 +1831,10 @@ export class HeartbeatTriggerScheduler {
* @param config - Per-agent heartbeat config
*/
registerAgent(agentId: string, config: AgentHeartbeatConfig): void {
// State drives whether an agent ticks; this method no longer honors
// `config.enabled` as a registration gate. Callers filter based on
// state + ephemeral classification before calling through.
if (config.enabled === false) {
this.unregisterAgent(agentId);
return;
}
// Apply default interval if not explicitly configured
// This ensures agents with heartbeat monitoring enabled but no explicit interval
@@ -2042,46 +2048,112 @@ export class HeartbeatTriggerScheduler {
}
}
private isTimerEligibleAgent(agent: Agent): boolean {
return isHeartbeatManaged(agent)
&& agent.runtimeConfig?.enabled !== false
&& isTickableState(agent.state);
}
private getAgentTimerConfig(agent: Agent): AgentHeartbeatConfig {
const rc = (agent.runtimeConfig ?? {}) as {
enabled?: boolean;
heartbeatIntervalMs?: number;
maxConcurrentRuns?: number;
};
return {
enabled: rc.enabled,
heartbeatIntervalMs: rc.heartbeatIntervalMs,
maxConcurrentRuns: rc.maxConcurrentRuns,
};
}
private syncTimerForAgent(agent: Agent, reason: string): void {
if (!this.isTimerEligibleAgent(agent)) {
this.unregisterAgent(agent.id);
return;
}
if (this.timers.has(agent.id)) {
// Already ticking — non-config updates should not reset the interval.
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
heartbeatLog.log(`Timer armed for ${agent.id} (${reason})`);
}
private async syncTimerForAgentFromStore(agentId: string, reason: string): Promise<void> {
const agent = await this.store.getAgent(agentId);
if (!agent) {
this.unregisterAgent(agentId);
return;
}
if (!this.isTimerEligibleAgent(agent)) {
this.unregisterAgent(agentId);
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
heartbeatLog.log(`Timer refreshed for ${agent.id} (${reason})`);
}
private didHeartbeatScheduleChange(revision: AgentConfigRevision): boolean {
const before = (revision.before.runtimeConfig ?? {}) as Record<string, unknown>;
const after = (revision.after.runtimeConfig ?? {}) as Record<string, unknown>;
const pickScheduleFields = (runtimeConfig: Record<string, unknown>) => ({
enabled: runtimeConfig.enabled,
heartbeatIntervalMs: runtimeConfig.heartbeatIntervalMs,
maxConcurrentRuns: runtimeConfig.maxConcurrentRuns,
});
return JSON.stringify(pickScheduleFields(before)) !== JSON.stringify(pickScheduleFields(after));
}
private watchAgentLifecycle(): void {
if (this.updatedListener || this.deletedListener) return;
if (this.createdListener || this.updatedListener || this.configRevisionListener || this.deletedListener) return;
this.createdListener = (agent) => {
this.syncTimerForAgent(agent, `created:${agent.state}`);
};
// State-driven registration: when an agent transitions into a tickable
// state (active/running) arm the timer; transitioning out clears it.
// state arm the timer; transitioning out clears it. Existing timers are
// left alone here so unrelated agent updates do not reset the interval.
this.updatedListener = (agent) => {
if (!isHeartbeatManaged(agent) || !isTickableState(agent.state)) {
this.unregisterAgent(agent.id);
this.syncTimerForAgent(agent, `state:${agent.state}`);
};
this.configRevisionListener = (agentId, revision) => {
if (!this.didHeartbeatScheduleChange(revision)) {
return;
}
if (this.timers.has(agent.id)) {
// Already ticking — re-registering would reset the interval mid-cycle
// on every unrelated agent update.
return;
}
const rc = (agent.runtimeConfig ?? {}) as {
heartbeatIntervalMs?: number;
maxConcurrentRuns?: number;
};
this.registerAgent(agent.id, {
heartbeatIntervalMs: rc.heartbeatIntervalMs,
maxConcurrentRuns: rc.maxConcurrentRuns,
});
heartbeatLog.log(
`State-driven registration: ${agent.id} is ${agent.state} — timer armed`,
);
void this.syncTimerForAgentFromStore(agentId, "runtime-config-updated");
};
this.deletedListener = (agentId) => {
this.unregisterAgent(agentId);
};
this.store.on("agent:created", this.createdListener);
this.store.on("agent:updated", this.updatedListener);
this.store.on("agent:configRevision", this.configRevisionListener);
this.store.on("agent:deleted", this.deletedListener);
}
private unwatchAgentLifecycle(): void {
if (this.createdListener) {
this.store.off("agent:created", this.createdListener);
this.createdListener = null;
}
if (this.updatedListener) {
this.store.off("agent:updated", this.updatedListener);
this.updatedListener = null;
}
if (this.configRevisionListener) {
this.store.off("agent:configRevision", this.configRevisionListener);
this.configRevisionListener = null;
}
if (this.deletedListener) {
this.store.off("agent:deleted", this.deletedListener);
this.deletedListener = null;

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

@@ -862,24 +862,46 @@ describe("InProcessRuntime", () => {
expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
});
it("re-registers an existing agent when agent:updated event is emitted", async () => {
// Create a new agent
it("does not reset an armed timer on unrelated agent updates", async () => {
const store = getAgentStore(runtime);
const monitor = runtime.getHeartbeatMonitor();
expect(monitor).toBeDefined();
const executeHeartbeatSpy = vi
.spyOn(monitor!, "executeHeartbeat")
.mockResolvedValue({ id: "run-update-timer-stability" } as any);
const agent = await store.createAgent({
name: "test-agent-update",
role: "executor",
runtimeConfig: {
enabled: true,
heartbeatIntervalMs: 1000,
},
});
const scheduler = runtime.getTriggerScheduler();
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
// Update the agent
await vi.advanceTimersByTimeAsync(400);
await store.updateAgent(agent.id, {
name: "test-agent-update-renamed",
});
// Verify the agent is still registered (re-registration succeeded)
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
await vi.advanceTimersByTimeAsync(599);
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
await vi.waitFor(() => {
expect(executeHeartbeatSpy).toHaveBeenCalledTimes(1);
});
expect(executeHeartbeatSpy).toHaveBeenCalledWith(
expect.objectContaining({
agentId: agent.id,
source: "timer",
}),
);
});
it("unregisters an agent when enabled is set to false in update", async () => {
@@ -907,6 +929,41 @@ describe("InProcessRuntime", () => {
expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
});
it("re-arms the timer when heartbeat interval changes", async () => {
const store = getAgentStore(runtime);
const monitor = runtime.getHeartbeatMonitor();
expect(monitor).toBeDefined();
const executeHeartbeatSpy = vi
.spyOn(monitor!, "executeHeartbeat")
.mockResolvedValue({ id: "run-interval-change" } as any);
const agent = await store.createAgent({
name: "interval-change-agent",
role: "executor",
runtimeConfig: {
enabled: true,
heartbeatIntervalMs: 1000,
},
});
await vi.advanceTimersByTimeAsync(400);
await store.updateAgent(agent.id, {
runtimeConfig: {
enabled: true,
heartbeatIntervalMs: 2000,
},
});
await vi.advanceTimersByTimeAsync(1599);
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(401);
await vi.waitFor(() => {
expect(executeHeartbeatSpy).toHaveBeenCalledTimes(1);
});
});
it("clears timers on pause and re-arms from resume without stale pre-pause firing", async () => {
const store = getAgentStore(runtime);
const monitor = runtime.getHeartbeatMonitor();

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

@@ -98,8 +98,6 @@ export class InProcessRuntime
private triageProcessor?: TriageProcessor;
private messageStore?: MessageStore;
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 */
@@ -506,9 +504,8 @@ export class InProcessRuntime
);
this.triggerScheduler.start();
// 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.
// Startup bootstrap for already-persisted agents. Ongoing lifecycle
// updates are handled inside HeartbeatTriggerScheduler itself.
const isHeartbeatEnabledAgent = (agent: import("@fusion/core").Agent) =>
!isEphemeralAgent(agent) && agent.runtimeConfig?.enabled !== false;
const isTickableHeartbeatState = (state: import("@fusion/core").AgentState) =>
@@ -516,34 +513,6 @@ export class InProcessRuntime
const isTimerManagedAgent = (agent: import("@fusion/core").Agent) =>
isHeartbeatEnabledAgent(agent) && isTickableHeartbeatState(agent.state);
this.agentCreatedListener = (agent) => {
if (!this.triggerScheduler) return;
if (!isTimerManagedAgent(agent)) return;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
});
runtimeLog.log(`Registered new agent ${agent.id} for heartbeat triggers`);
};
this.agentStore.on("agent:created", this.agentCreatedListener);
this.agentUpdatedListener = (agent) => {
if (!this.triggerScheduler) return;
if (!isTimerManagedAgent(agent)) {
this.triggerScheduler.unregisterAgent(agent.id);
runtimeLog.log(`Unregistered agent ${agent.id} from heartbeat triggers`);
return;
}
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
});
runtimeLog.log(`Re-registered agent ${agent.id} for heartbeat triggers`);
};
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.
@@ -595,6 +564,7 @@ export class InProcessRuntime
if (!isTimerManagedAgent(agent)) continue;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
});
@@ -792,16 +762,6 @@ export class InProcessRuntime
// 3. Remove agent event listeners (before stopping trigger scheduler)
// Guard on this.agentStore being defined - it may not exist if AgentStore init failed
if (this.agentCreatedListener && this.agentStore) {
this.agentStore.off("agent:created", this.agentCreatedListener);
this.agentCreatedListener = undefined;
runtimeLog.log("AgentStore agent:created listener removed");
}
if (this.agentUpdatedListener && this.agentStore) {
this.agentStore.off("agent:updated", this.agentUpdatedListener);
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;