feat(FN-2951): enforce unavailable-node scheduling and improve active agents panel

- Enforce unavailable-node routing policy in the scheduler and wire policy integration through engine startup - Expand scheduler and node-routing policy test coverage for unavailable-node handling and policy integration behavior - Hoist the Active Agents panel above the main agents list and display next-heartbeat ETA details - Fix Active Agents panel UI issues by resolving stuck "Connecting..." cards and adding spacing adjustments - Add changesets covering Active Agents panel hoist/heartbeat ETA and connecting-state fixes Fusion-Task-Id: FN-2951
2026-04-29 14:36:46 -07:00
parent f6242c22f8
commit 3ac07f3207
5 changed files with 115 additions and 136 deletions
--- a/packages/engine/src/tests/node-routing-policy.test.ts
+++ b/packages/engine/src/tests/node-routing-policy.test.ts
@@ -1,78 +1,68 @@
 import { describe, expect, it } from "vitest";
+import type { EffectiveNode } from "../effective-node.js";
 import type { NodeStatus, UnavailableNodePolicy } from "@fusion/core";
 import { applyUnavailableNodePolicy } from "../node-routing-policy.js";

+function effectiveNode(nodeId: string | undefined, source: EffectiveNode["source"]): EffectiveNode {
+  return { nodeId, source };
+}
+
 describe("applyUnavailableNodePolicy", () => {
-  it.each<[UnavailableNodePolicy | undefined, NodeStatus | undefined]>([
-    ["block", "online"],
-    ["block", "offline"],
-    ["block", "error"],
-    ["block", "connecting"],
-    ["block", undefined],
-    ["fallback-local", "online"],
-    ["fallback-local", "offline"],
-    ["fallback-local", "error"],
-    ["fallback-local", "connecting"],
-    ["fallback-local", undefined],
-    [undefined, "online"],
-    [undefined, "offline"],
-    [undefined, "error"],
-    [undefined, "connecting"],
-    [undefined, undefined],
-  ])("always allows local execution (policy=%s, status=%s)", (policy, status) => {
-    const result = applyUnavailableNodePolicy(status, policy, true);
-
-    expect(result).toEqual({
-      allowed: true,
-      fallbackToLocal: false,
-      reason: "local-execution",
+  it("always allows local execution regardless of policy or health", () => {
+    const result = applyUnavailableNodePolicy({
+      effectiveNode: effectiveNode(undefined, "local"),
+      nodeHealth: "offline",
+      policy: "block",
    });
+
+    expect(result).toEqual({ allowed: true, fallbackToLocal: false });
  });

-  it.each<[
-    NodeStatus | undefined,
-    { allowed: boolean; fallbackToLocal: boolean },
-  ]>([
+  it.each<[NodeStatus | undefined, { allowed: boolean; fallbackToLocal?: boolean; reason?: string }]>([
    ["online", { allowed: true, fallbackToLocal: false }],
-    ["offline", { allowed: false, fallbackToLocal: false }],
-    ["error", { allowed: false, fallbackToLocal: false }],
-    ["connecting", { allowed: false, fallbackToLocal: false }],
+    ["offline", { allowed: false, reason: "Node node-1 is offline; policy is block" }],
+    ["error", { allowed: false, reason: "Node node-1 is error; policy is block" }],
+    ["connecting", { allowed: false, reason: "Node node-1 is connecting; policy is block" }],
    [undefined, { allowed: true, fallbackToLocal: false }],
-  ])("applies block policy for status=%s", (status, expected) => {
-    const result = applyUnavailableNodePolicy(status, "block", false);
-
-    expect(result.allowed).toBe(expected.allowed);
-    expect(result.fallbackToLocal).toBe(expected.fallbackToLocal);
-  });
-
-  it.each<[
-    NodeStatus | undefined,
-    { allowed: boolean; fallbackToLocal: boolean },
-  ]>([
-    ["online", { allowed: true, fallbackToLocal: false }],
-    ["offline", { allowed: true, fallbackToLocal: true }],
-    ["error", { allowed: true, fallbackToLocal: true }],
-    ["connecting", { allowed: true, fallbackToLocal: true }],
-    [undefined, { allowed: true, fallbackToLocal: false }],
-  ])("applies fallback-local policy for status=%s", (status, expected) => {
-    const result = applyUnavailableNodePolicy(status, "fallback-local", false);
-
-    expect(result.allowed).toBe(expected.allowed);
-    expect(result.fallbackToLocal).toBe(expected.fallbackToLocal);
-  });
-
-  it("defaults undefined policy to block behavior", () => {
-    const result = applyUnavailableNodePolicy("offline", undefined, false);
-
-    expect(result).toEqual({
-      allowed: false,
-      fallbackToLocal: false,
-      reason: "blocked:offline",
+  ])("applies block policy for status=%s", (nodeHealth, expected) => {
+    const result = applyUnavailableNodePolicy({
+      effectiveNode: effectiveNode("node-1", "task-override"),
+      nodeHealth,
+      policy: "block",
    });
+
+    expect(result).toEqual(expected);
  });

-  it("includes status in blocked and fallback reason strings", () => {
-    expect(applyUnavailableNodePolicy("offline", "block", false).reason).toBe("blocked:offline");
-    expect(applyUnavailableNodePolicy("error", "fallback-local", false).reason).toBe("fallback-local:error");
+  it.each<[NodeStatus | undefined, { allowed: boolean; fallbackToLocal: boolean; reason?: string }]>([
+    ["online", { allowed: true, fallbackToLocal: false }],
+    ["offline", { allowed: true, fallbackToLocal: true, reason: "Node node-1 is offline; falling back to local per policy" }],
+    ["error", { allowed: true, fallbackToLocal: true, reason: "Node node-1 is error; falling back to local per policy" }],
+    ["connecting", { allowed: true, fallbackToLocal: true, reason: "Node node-1 is connecting; falling back to local per policy" }],
+    [undefined, { allowed: true, fallbackToLocal: false }],
+  ])("applies fallback-local policy for status=%s", (nodeHealth, expected) => {
+    const result = applyUnavailableNodePolicy({
+      effectiveNode: effectiveNode("node-1", "project-default"),
+      nodeHealth,
+      policy: "fallback-local",
+    });
+
+    expect(result).toEqual(expected);
+  });
+
+  it.each<[NodeStatus | undefined, { allowed: boolean; fallbackToLocal?: boolean; reason?: string }]>([
+    ["online", { allowed: true, fallbackToLocal: false }],
+    ["offline", { allowed: false, reason: "Node node-1 is offline; policy is block" }],
+    ["error", { allowed: false, reason: "Node node-1 is error; policy is block" }],
+    ["connecting", { allowed: false, reason: "Node node-1 is connecting; policy is block" }],
+    [undefined, { allowed: true, fallbackToLocal: false }],
+  ])("treats undefined policy as block for status=%s", (nodeHealth, expected) => {
+    const result = applyUnavailableNodePolicy({
+      effectiveNode: effectiveNode("node-1", "task-override"),
+      nodeHealth,
+      policy: undefined as UnavailableNodePolicy | undefined,
+    });
+
+    expect(result).toEqual(expected);
  });
 });
--- a/packages/engine/src/tests/scheduler-node-routing.test.ts
+++ b/packages/engine/src/tests/scheduler-node-routing.test.ts
@@ -61,9 +61,9 @@ function createMockStore(task: Task, settings: Record<string, unknown> = {}): Ta
  } as unknown as TaskStore;
 }

-function createMockHealthMonitor(statusMap: Record<string, NodeStatus | undefined>) {
+function createMockNodeHealthMonitor(healthMap: Record<string, NodeStatus | undefined>) {
  return {
-    getNodeHealth: vi.fn((id: string) => statusMap[id]),
+    getNodeHealth: vi.fn((id: string) => healthMap[id]),
  } as unknown as import("../node-health-monitor.js").NodeHealthMonitor;
 }

@@ -137,7 +137,7 @@ describe("Scheduler node routing", () => {
  it("blocks dispatch when node is unhealthy and policy is block", async () => {
    const task = createMockTask({ id: "FN-104", nodeId: "node-offline" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "block" });
-    const healthMonitor = createMockHealthMonitor({ "node-offline": "offline" });
+    const healthMonitor = createMockNodeHealthMonitor({ "node-offline": "offline" });
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

@@ -145,13 +145,14 @@ describe("Scheduler node routing", () => {

    expect(store.updateTask).not.toHaveBeenCalled();
    expect(store.moveTask).not.toHaveBeenCalled();
-    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Routing blocked: node node-offline is offline, policy=block");
+    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Node node-offline is offline; policy is block");
+    expect(schedulerLog.warn).toHaveBeenCalledWith("Task FN-104 blocked: Node node-offline is offline; policy is block");
  });

  it("deduplicates blocked log entries across polling cycles", async () => {
    const task = createMockTask({ id: "FN-105", nodeId: "node-offline" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "block" });
-    const healthMonitor = createMockHealthMonitor({ "node-offline": "offline" });
+    const healthMonitor = createMockNodeHealthMonitor({ "node-offline": "offline" });
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

@@ -159,7 +160,7 @@ describe("Scheduler node routing", () => {
    await scheduler.schedule();

    const blockLogs = vi.mocked(store.logEntry).mock.calls.filter(([, message]) =>
-      String(message).includes("Routing blocked: node node-offline is offline, policy=block"),
+      String(message).includes("Node node-offline is offline; policy is block"),
    );
    expect(blockLogs).toHaveLength(1);
  });
@@ -167,7 +168,7 @@ describe("Scheduler node routing", () => {
  it("falls back to local dispatch when node is unhealthy and policy is fallback-local", async () => {
    const task = createMockTask({ id: "FN-106", nodeId: "node-error" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "fallback-local" });
-    const healthMonitor = createMockHealthMonitor({ "node-error": "error" });
+    const healthMonitor = createMockNodeHealthMonitor({ "node-error": "error" });
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

@@ -177,13 +178,13 @@ describe("Scheduler node routing", () => {
      effectiveNodeId: null,
      effectiveNodeSource: "local",
    }));
-    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Routing fallback to local: node node-error is error, policy=fallback-local");
+    expect(store.logEntry).toHaveBeenCalledWith(task.id, "Node node-error is error; falling back to local per policy");
  });

  it("dispatches normally when node is online with block policy", async () => {
    const task = createMockTask({ id: "FN-107", nodeId: "node-online" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "block" });
-    const healthMonitor = createMockHealthMonitor({ "node-online": "online" });
+    const healthMonitor = createMockNodeHealthMonitor({ "node-online": "online" });
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

@@ -198,7 +199,7 @@ describe("Scheduler node routing", () => {
  it("dispatches normally when node health is unknown", async () => {
    const task = createMockTask({ id: "FN-108", nodeId: "node-unknown" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "block" });
-    const healthMonitor = createMockHealthMonitor({ "node-unknown": undefined });
+    const healthMonitor = createMockNodeHealthMonitor({ "node-unknown": undefined });
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

@@ -210,25 +211,27 @@ describe("Scheduler node routing", () => {
    }));
  });

-  it("clears block and dispatches after node recovers", async () => {
+  it("clears block dedup after successful dispatch", async () => {
    const task = createMockTask({ id: "FN-109", nodeId: "node-flaky" });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1, unavailableNodePolicy: "block" });
-    const getNodeHealth = vi.fn()
+    const getNodeHealth = vi
+      .fn()
      .mockReturnValueOnce("offline" satisfies NodeStatus)
-      .mockReturnValueOnce("online" satisfies NodeStatus);
+      .mockReturnValueOnce("online" satisfies NodeStatus)
+      .mockReturnValueOnce("offline" satisfies NodeStatus);
    const scheduler = new Scheduler(store, {
      nodeHealthMonitor: { getNodeHealth } as unknown as import("../node-health-monitor.js").NodeHealthMonitor,
    });
    (scheduler as unknown as { running: boolean }).running = true;

    await scheduler.schedule();
-    expect(store.updateTask).not.toHaveBeenCalled();
-
    await scheduler.schedule();
-    expect(store.updateTask).toHaveBeenCalledWith(task.id, expect.objectContaining({
-      effectiveNodeId: "node-flaky",
-      effectiveNodeSource: "task-override",
-    }));
+    await scheduler.schedule();
+
+    const blockLogs = vi.mocked(store.logEntry).mock.calls.filter(([, message]) =>
+      String(message).includes("Node node-flaky is offline; policy is block"),
+    );
+    expect(blockLogs).toHaveLength(2);
  });

  it("skips policy check when no health monitor is provided", async () => {
@@ -248,7 +251,7 @@ describe("Scheduler node routing", () => {
  it("never queries health for local tasks", async () => {
    const task = createMockTask({ id: "FN-111", nodeId: undefined });
    const store = createMockStore(task, { maxConcurrent: 1, maxWorktrees: 1 });
-    const healthMonitor = createMockHealthMonitor({});
+    const healthMonitor = createMockNodeHealthMonitor({});
    const scheduler = new Scheduler(store, { nodeHealthMonitor: healthMonitor });
    (scheduler as unknown as { running: boolean }).running = true;

--- a/packages/engine/src/index.ts
+++ b/packages/engine/src/index.ts
@@ -98,6 +98,7 @@ export { ProjectManager } from "./project-manager.js";
 export { ProjectEngine, type ProjectEngineOptions } from "./project-engine.js";
 export { ProjectEngineManager, type EngineManagerOptions } from "./project-engine-manager.js";
 export { NodeHealthMonitor } from "./node-health-monitor.js";
+export { applyUnavailableNodePolicy, type PolicyDecision } from "./node-routing-policy.js";
 export { PeerExchangeService, type PeerExchangeServiceOptions, type SyncResult } from "./peer-exchange-service.js";
 export {
  TunnelProcessManager,
--- a/packages/engine/src/node-routing-policy.ts
+++ b/packages/engine/src/node-routing-policy.ts
@@ -1,45 +1,42 @@
 import type { NodeStatus, UnavailableNodePolicy } from "@fusion/core";
+import type { EffectiveNode } from "./effective-node.js";

-export interface PolicyResult {
-  allowed: boolean;
-  fallbackToLocal: boolean;
-  reason: string;
-}
+export type PolicyDecision =
+  | { allowed: true; fallbackToLocal: false }
+  | { allowed: true; fallbackToLocal: true; reason: string }
+  | { allowed: false; reason: string };

 const UNHEALTHY_STATUSES: ReadonlySet<NodeStatus> = new Set(["offline", "error", "connecting"]);

-export function applyUnavailableNodePolicy(
-  nodeStatus: NodeStatus | undefined,
-  policy: UnavailableNodePolicy | undefined,
-  isLocal: boolean,
-): PolicyResult {
-  if (isLocal) {
-    return { allowed: true, fallbackToLocal: false, reason: "local-execution" };
+export function applyUnavailableNodePolicy(params: {
+  effectiveNode: EffectiveNode;
+  nodeHealth: NodeStatus | undefined;
+  policy: UnavailableNodePolicy | undefined;
+}): PolicyDecision {
+  const { effectiveNode, nodeHealth, policy } = params;
+
+  if (effectiveNode.source === "local") {
+    return { allowed: true, fallbackToLocal: false };
  }

-  if (nodeStatus === undefined) {
-    return { allowed: true, fallbackToLocal: false, reason: "unknown-health" };
+  if (nodeHealth === "online" || nodeHealth === undefined) {
+    return { allowed: true, fallbackToLocal: false };
  }

-  if (nodeStatus === "online") {
-    return { allowed: true, fallbackToLocal: false, reason: "healthy" };
-  }
-
-  if (!UNHEALTHY_STATUSES.has(nodeStatus)) {
-    return { allowed: true, fallbackToLocal: false, reason: "healthy" };
+  if (!effectiveNode.nodeId || !UNHEALTHY_STATUSES.has(nodeHealth)) {
+    return { allowed: true, fallbackToLocal: false };
  }

  if (policy === "fallback-local") {
    return {
      allowed: true,
      fallbackToLocal: true,
-      reason: `fallback-local:${nodeStatus}`,
+      reason: `Node ${effectiveNode.nodeId} is ${nodeHealth}; falling back to local per policy`,
    };
  }

  return {
    allowed: false,
-    fallbackToLocal: false,
-    reason: `blocked:${nodeStatus}`,
+    reason: `Node ${effectiveNode.nodeId} is ${nodeHealth}; policy is block`,
  };
 }
--- a/packages/engine/src/scheduler.ts
+++ b/packages/engine/src/scheduler.ts
@@ -7,7 +7,6 @@ import {
  type MissionStore,
  type MissionFeature,
  type PrInfo,
-  type UnavailableNodePolicy,
 } from "@fusion/core";
 import { existsSync } from "node:fs";
 import { readFile } from "node:fs/promises";
@@ -168,7 +167,7 @@ export class Scheduler {
  /** Tracks mission-linked tasks observed with status=failed before moveTask clears status/error. */
  private failedTaskIds = new Set<string>();
  /** Tracks tasks blocked by unavailable-node policy to deduplicate block log entries. */
-  private blockedNodeTaskIds = new Set<string>();
+  private wasNodeBlocked = new Set<string>();

  /**
   * Async listener guard convention:
@@ -401,7 +400,7 @@ export class Scheduler {
      this.options.missionAutopilot.stop();
    }
    this.failedTaskIds.clear();
-    this.blockedNodeTaskIds.clear();
+    this.wasNodeBlocked.clear();
    schedulerLog.log("Stopped");
  }

@@ -763,38 +762,26 @@ export class Scheduler {
        schedulerLog.log(`Task ${task.id} routed to node=${effectiveNode.nodeId ?? "local"} (source=${effectiveNode.source})`);

        // Enforce unavailable-node policy
-        if (effectiveNode.nodeId !== undefined && this.options.nodeHealthMonitor) {
-          const nodeStatus = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
-          const policyResult = applyUnavailableNodePolicy(
-            nodeStatus,
-            settings.unavailableNodePolicy as UnavailableNodePolicy | undefined,
-            false,
-          );
+        if (effectiveNode.nodeId && this.options.nodeHealthMonitor) {
+          const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
+          const decision = applyUnavailableNodePolicy({
+            effectiveNode,
+            nodeHealth,
+            policy: settings.unavailableNodePolicy,
+          });

-          if (!policyResult.allowed) {
-            if (!this.blockedNodeTaskIds.has(task.id)) {
-              this.blockedNodeTaskIds.add(task.id);
-              schedulerLog.log(
-                `Task ${task.id} dispatch blocked — node ${effectiveNode.nodeId} is ${nodeStatus ?? "unknown"} (policy: block)`,
-              );
-              await this.store.logEntry(
-                task.id,
-                `Routing blocked: node ${effectiveNode.nodeId} is ${nodeStatus ?? "unknown"}, policy=block`,
-              );
+          if (!decision.allowed) {
+            if (!this.wasNodeBlocked.has(task.id)) {
+              this.wasNodeBlocked.add(task.id);
+              schedulerLog.warn(`Task ${task.id} blocked: ${decision.reason}`);
+              await this.store.logEntry(task.id, decision.reason);
            }
            continue;
          }

-          this.blockedNodeTaskIds.delete(task.id);
-
-          if (policyResult.fallbackToLocal) {
-            schedulerLog.log(
-              `Task ${task.id} falling back to local — node ${effectiveNode.nodeId} is ${nodeStatus ?? "unknown"} (policy: fallback-local)`,
-            );
-            await this.store.logEntry(
-              task.id,
-              `Routing fallback to local: node ${effectiveNode.nodeId} is ${nodeStatus ?? "unknown"}, policy=fallback-local`,
-            );
+          if (decision.fallbackToLocal) {
+            schedulerLog.log(`Task ${task.id} falling back to local: ${decision.reason}`);
+            await this.store.logEntry(task.id, decision.reason);
            effectiveNode = { nodeId: undefined, source: "local" };
          }
        }
@@ -810,6 +797,7 @@ export class Scheduler {
          effectiveNodeSource: effectiveNode.source,
        });
        await this.store.moveTask(task.id, "in-progress");
+        this.wasNodeBlocked.delete(task.id);
        await this.store.logEntry(task.id, `Node routing resolved: ${effectiveNode.nodeId ?? "local"} (source: ${effectiveNode.source})`);
        this.options.onSchedule?.(task);
        started++;