fix(FN-2040): prevent false unresponsive agent states

- Apply a default 30s heartbeat interval when scheduler config is missing or invalid - Dynamically register and update heartbeat triggers from AgentStore create/update events - Register all non-disabled agents at startup and guard listener cleanup when stopping runtime - Treat non-periodic agents as healthy in dashboard health checks to avoid stale-time false alarms - Expand engine and dashboard tests and add a patch changeset for @gsxdsm/fusion
2026-04-17 20:40:06 -07:00
parent 0a89068ef0
commit e4f43a8b43
7 changed files with 396 additions and 37 deletions
--- a/packages/engine/src/agent-heartbeat.test.ts
+++ b/packages/engine/src/agent-heartbeat.test.ts
@@ -3340,14 +3340,66 @@ describe("HeartbeatTriggerScheduler", () => {
      expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
    });

-    it("skips registration when intervalMs is undefined", () => {
+    it("applies default 30-second interval when intervalMs is undefined", async () => {
+      vi.useFakeTimers();
      scheduler.registerAgent("agent-001", {});
-      expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
+      expect(scheduler.getRegisteredAgents()).toContain("agent-001");
+
+      // Verify the default 30-second interval fires
+      expect(callback).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(30_000);
+      expect(callback).toHaveBeenCalledOnce();
+      vi.useRealTimers();
    });

-    it("skips registration when intervalMs is 0", () => {
+    it("applies default 30-second interval when intervalMs is 0", async () => {
+      vi.useFakeTimers();
      scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 0 });
-      expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
+      expect(scheduler.getRegisteredAgents()).toContain("agent-001");
+
+      // Verify the default 30-second interval fires
+      expect(callback).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(30_000);
+      expect(callback).toHaveBeenCalledOnce();
+      vi.useRealTimers();
+    });
+
+    it("applies default 30-second interval when heartbeatIntervalMs is not set", async () => {
+      vi.useFakeTimers();
+      scheduler.registerAgent("agent-001", { enabled: true });
+      expect(scheduler.getRegisteredAgents()).toContain("agent-001");
+
+      // Should fire at exactly 30 seconds (default interval)
+      await vi.advanceTimersByTimeAsync(29_999);
+      expect(callback).not.toHaveBeenCalled();
+
+      await vi.advanceTimersByTimeAsync(1); // Now at exactly 30 seconds
+      expect(callback).toHaveBeenCalledOnce();
+      expect(callback).toHaveBeenCalledWith("agent-001", "timer", {
+        wakeReason: "timer",
+        triggerDetail: "scheduled",
+        intervalMs: 30_000,
+      });
+      vi.useRealTimers();
+    });
+
+    it("uses explicit interval over default when both are provided", async () => {
+      vi.useFakeTimers();
+      scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 15_000, enabled: true });
+      expect(scheduler.getRegisteredAgents()).toContain("agent-001");
+
+      // Should fire at 15 seconds (explicit), not 30
+      await vi.advanceTimersByTimeAsync(14_999);
+      expect(callback).not.toHaveBeenCalled();
+
+      await vi.advanceTimersByTimeAsync(1); // Now at exactly 15 seconds
+      expect(callback).toHaveBeenCalledOnce();
+      expect(callback).toHaveBeenCalledWith("agent-001", "timer", {
+        wakeReason: "timer",
+        triggerDetail: "scheduled",
+        intervalMs: 15_000,
+      });
+      vi.useRealTimers();
    });

    it("clears previous timer when re-registering", () => {
--- a/packages/engine/src/agent-heartbeat.ts
+++ b/packages/engine/src/agent-heartbeat.ts
@@ -1459,6 +1459,9 @@ export class HeartbeatTriggerScheduler {
    return this.running;
  }

+  /** Default heartbeat interval when not explicitly configured (30 seconds) */
+  private static readonly DEFAULT_HEARTBEAT_INTERVAL_MS = 30_000;
+
  /**
   * Register an agent for timer-based heartbeat triggers.
   * @param agentId - The agent ID
@@ -1471,11 +1474,14 @@ export class HeartbeatTriggerScheduler {
      return;
    }

-    // Skip if no interval configured
-    const rawIntervalMs = config.heartbeatIntervalMs;
+    // Apply default interval if not explicitly configured
+    // This ensures agents with heartbeat monitoring enabled but no explicit interval
+    // still get periodic timer triggers (matching HeartbeatMonitor constructor default)
+    let rawIntervalMs = config.heartbeatIntervalMs;
+    let usingDefaultInterval = false;
    if (!rawIntervalMs || typeof rawIntervalMs !== "number" || !Number.isFinite(rawIntervalMs) || rawIntervalMs <= 0) {
-      heartbeatLog.log(`Skipping timer registration for ${agentId} (no interval)`);
-      return;
+      rawIntervalMs = HeartbeatTriggerScheduler.DEFAULT_HEARTBEAT_INTERVAL_MS;
+      usingDefaultInterval = true;
    }

    const intervalMs = Math.max(1000, Math.round(rawIntervalMs));
@@ -1488,7 +1494,11 @@ export class HeartbeatTriggerScheduler {
    }, intervalMs);

    this.timers.set(agentId, { intervalMs, handle });
-    heartbeatLog.log(`Registered timer for ${agentId} (every ${intervalMs}ms)`);
+    heartbeatLog.log(
+      usingDefaultInterval
+        ? `Registered timer for ${agentId} (every ${intervalMs}ms, default interval)`
+        : `Registered timer for ${agentId} (every ${intervalMs}ms)`,
+    );
  }

  /**
--- a/packages/engine/src/runtimes/in-process-runtime.test.ts
+++ b/packages/engine/src/runtimes/in-process-runtime.test.ts
@@ -684,4 +684,153 @@ describe("InProcessRuntime", () => {
      expect(MessageStore).toHaveBeenCalled();
    });
  });
+
+  describe("dynamic agent registration with HeartbeatTriggerScheduler", () => {
+    beforeEach(async () => {
+      vi.useFakeTimers();
+      await runtime.start();
+    });
+
+    afterEach(async () => {
+      await runtime.stop();
+      vi.useRealTimers();
+    });
+
+    it("registers a new agent when agent:created event is emitted", async () => {
+      // Create a new agent via the AgentStore
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-dynamic",
+        role: "executor",
+      });
+
+      // Verify the agent was registered with the trigger scheduler
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler).toBeDefined();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+    });
+
+    it("registers agent without explicit heartbeatIntervalMs using default 30s interval", async () => {
+      // Create a new agent with only enabled: true (no heartbeatIntervalMs)
+      // This tests that the default 30-second interval is applied
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-default-interval",
+        role: "executor",
+        runtimeConfig: { enabled: true }, // No heartbeatIntervalMs - should use default 30s
+      });
+
+      // Verify the agent was registered with the trigger scheduler
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler).toBeDefined();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+    });
+
+    it("registers a new agent with explicit heartbeatIntervalMs", async () => {
+      // Create a new agent with explicit heartbeat config
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-explicit",
+        role: "executor",
+        runtimeConfig: {
+          heartbeatIntervalMs: 15000,
+          enabled: true,
+        },
+      });
+
+      // Verify the agent was registered with the trigger scheduler
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler).toBeDefined();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+    });
+
+    it("does not register a new agent when enabled is false", async () => {
+      // Create a new agent with heartbeat disabled
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-disabled",
+        role: "executor",
+        runtimeConfig: {
+          enabled: false,
+        },
+      });
+
+      // Verify the agent was NOT registered with the trigger scheduler
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler).toBeDefined();
+      expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
+    });
+
+    it("re-registers an existing agent when agent:updated event is emitted", async () => {
+      // Create a new agent
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-update",
+        role: "executor",
+      });
+
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+
+      // Update the agent
+      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);
+    });
+
+    it("unregisters an agent when enabled is set to false in update", async () => {
+      // Create a new agent with heartbeat enabled
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-toggle",
+        role: "executor",
+        runtimeConfig: {
+          enabled: true,
+        },
+      });
+
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+
+      // Update the agent to disable heartbeat
+      await store.updateAgent(agent.id, {
+        runtimeConfig: {
+          enabled: false,
+        },
+      });
+
+      // Verify the agent was unregistered
+      expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
+    });
+
+    it("removes event listeners when runtime is stopped", async () => {
+      // Create a new agent before stopping
+      const store = getAgentStore(runtime);
+      const agent = await store.createAgent({
+        name: "test-agent-cleanup",
+        role: "executor",
+      });
+
+      const scheduler = runtime.getTriggerScheduler();
+      expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
+
+      // Stop the runtime
+      await runtime.stop();
+
+      // The agent should still be registered (unregister is internal to scheduler)
+      // But the listeners should be removed - verify by checking they don't fire
+      // Create another agent - it won't be registered since runtime is stopped
+      const agent2 = await store.createAgent({
+        name: "test-agent-after-stop",
+        role: "executor",
+      });
+
+      // Since runtime is stopped, trigger scheduler is stopped
+      // The agent won't be in registered list
+      expect(scheduler!.getRegisteredAgents()).not.toContain(agent2.id);
+    });
+  });
 });
--- a/packages/engine/src/runtimes/in-process-runtime.ts
+++ b/packages/engine/src/runtimes/in-process-runtime.ts
@@ -97,6 +97,8 @@ 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;

  /**
   * @param config - Runtime configuration
@@ -409,21 +411,55 @@ export class InProcessRuntime
        );
        this.triggerScheduler.start();

-        // Register existing agents that have heartbeat config
+        // Set up dynamic registration for agents created or updated after startup
+        this.agentCreatedListener = (agent) => {
+          if (!this.triggerScheduler) return;
+          const rc = agent.runtimeConfig;
+          if (rc?.enabled === false) return;
+          this.triggerScheduler.registerAgent(agent.id, {
+            heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
+            enabled: rc?.enabled as boolean | 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;
+          const rc = agent.runtimeConfig;
+          if (rc?.enabled === false) {
+            this.triggerScheduler.unregisterAgent(agent.id);
+            runtimeLog.log(`Unregistered agent ${agent.id} from heartbeat triggers (disabled)`);
+          } else {
+            this.triggerScheduler.registerAgent(agent.id, {
+              heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
+              enabled: rc?.enabled as boolean | undefined,
+              maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
+            });
+            runtimeLog.log(`Re-registered agent ${agent.id} for heartbeat triggers`);
+          }
+        };
+        this.agentStore.on("agent:updated", this.agentUpdatedListener);
+
+        // Register existing agents with heartbeat monitoring not explicitly disabled
+        // Agents without explicit heartbeat config will use the default 30-second interval
        try {
          const agents = await this.agentStore.listAgents();
+          let registeredCount = 0;
          for (const agent of agents) {
            const rc = agent.runtimeConfig;
-            if (rc && (rc.heartbeatIntervalMs || rc.enabled !== undefined || rc.maxConcurrentRuns)) {
+            if (rc?.enabled !== false) {
              this.triggerScheduler.registerAgent(agent.id, {
-                heartbeatIntervalMs: rc.heartbeatIntervalMs as number | undefined,
-                enabled: rc.enabled as boolean | undefined,
-                maxConcurrentRuns: rc.maxConcurrentRuns as number | undefined,
+                heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
+                enabled: rc?.enabled as boolean | undefined,
+                maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
              });
+              registeredCount++;
            }
          }
          if (agents.length > 0) {
-            runtimeLog.log(`Registered ${this.triggerScheduler.getRegisteredAgents().length} agents for heartbeat triggers`);
+            runtimeLog.log(`Registered ${registeredCount} of ${agents.length} agents for heartbeat triggers`);
          }
        } catch (regErr) {
          runtimeLog.warn(`Failed to register agents for heartbeat triggers:`, regErr);
@@ -612,7 +648,20 @@ export class InProcessRuntime
        runtimeLog.log("RoutineScheduler stopped");
      }

-      // 3. Stop trigger scheduler
+      // 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");
+      }
+
+      // 4. Stop trigger scheduler
      if (this.triggerScheduler) {
        this.triggerScheduler.stop();
        runtimeLog.log("TriggerScheduler stopped");