test(FN-4772): add hybrid wiring integration and lifecycle coverage

Fusion-Task-Id: FN-4772
Fusion-Task-Lineage: d296e96a-fe44-4896-928a-fa44da62b41a

packages/cli/src/commands/__tests__/daemon.test.ts

@@ -4,10 +4,16 @@ import { mkdtempSync, rmSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
-const { mockSyncStartupModels } = vi.hoisted(() => ({
+const { mockSyncStartupModels, mockShouldUseHybridExecutor, mockHybridExecutorCtor, mockHybridExecutorInitialize, mockHybridExecutorShutdown } = vi.hoisted(() => ({
   mockSyncStartupModels: vi.fn().mockResolvedValue(undefined),
+  mockShouldUseHybridExecutor: vi.fn().mockResolvedValue({ enabled: false, reason: "single-project-local-only" }),
+  mockHybridExecutorInitialize: vi.fn().mockResolvedValue(undefined),
+  mockHybridExecutorShutdown: vi.fn().mockResolvedValue(undefined),
+  mockHybridExecutorCtor: vi.fn().mockImplementation(() => ({
+    initialize: mockHybridExecutorInitialize,
+    shutdown: mockHybridExecutorShutdown,
+  })),
 }));
-
 vi.mock("../startup-model-sync.js", () => ({
   syncStartupModels: mockSyncStartupModels,
 }));
@@ -596,9 +602,10 @@ vi.mock("@fusion/engine", async (importOriginal) => {
   createAiPromptExecutor: mocks.createAiPromptExecutorMock,
   HeartbeatMonitor: mocks.heartbeatMonitorCtor,
   HeartbeatTriggerScheduler: mocks.heartbeatTriggerSchedulerCtor,
+  shouldUseHybridExecutor: mockShouldUseHybridExecutor,
+  HybridExecutor: mockHybridExecutorCtor,
   });
 });
-
 vi.mock("@mariozechner/pi-coding-agent", () => ({
   AuthStorage: {
     create: vi.fn(() => mocks.authStorage),
@@ -889,6 +896,23 @@ describe("runDaemon", () => {
     expect(listenCall.server.close).toHaveBeenCalledTimes(1);
     expect(mocks.taskStores[0].close).toHaveBeenCalledTimes(1);
   });
+
+  it("enables HybridExecutor with env override and shuts down before engine stop", async () => {
+    process.env.FUSION_HYBRID_EXECUTOR = "1";
+    mockShouldUseHybridExecutor.mockResolvedValue({ enabled: true, reason: "env-override" });
+
+    await runDaemon({});
+    expect(mockHybridExecutorCtor).toHaveBeenCalledTimes(1);
+    expect(mockHybridExecutorInitialize).toHaveBeenCalledTimes(1);
+
+    await triggerSignal("SIGTERM");
+
+    expect(mockHybridExecutorShutdown).toHaveBeenCalledTimes(1);
+    expect(mockHybridExecutorShutdown.mock.invocationCallOrder[0]).toBeLessThan(
+      mocks.projectEngineInstances[0].stop.mock.invocationCallOrder[0],
+    );
+    delete process.env.FUSION_HYBRID_EXECUTOR;
+  });
 });
 
 describe("runDaemon --token-only mode", () => {

packages/cli/src/commands/__tests__/serve.test.ts

@@ -4,10 +4,16 @@ import { mkdtempSync, rmSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 
-const { mockSyncStartupModels } = vi.hoisted(() => ({
+const { mockSyncStartupModels, mockShouldUseHybridExecutor, mockHybridExecutorCtor, mockHybridExecutorInitialize, mockHybridExecutorShutdown } = vi.hoisted(() => ({
   mockSyncStartupModels: vi.fn().mockResolvedValue(undefined),
+  mockShouldUseHybridExecutor: vi.fn().mockResolvedValue({ enabled: false, reason: "single-project-local-only" }),
+  mockHybridExecutorInitialize: vi.fn().mockResolvedValue(undefined),
+  mockHybridExecutorShutdown: vi.fn().mockResolvedValue(undefined),
+  mockHybridExecutorCtor: vi.fn().mockImplementation(() => ({
+    initialize: mockHybridExecutorInitialize,
+    shutdown: mockHybridExecutorShutdown,
+  })),
 }));
-
 vi.mock("../startup-model-sync.js", () => ({
   syncStartupModels: mockSyncStartupModels,
 }));
@@ -655,9 +661,10 @@ vi.mock("@fusion/engine", async (importOriginal) => {
   createAiPromptExecutor: mocks.createAiPromptExecutorMock,
   HeartbeatMonitor: mocks.heartbeatMonitorCtor,
   HeartbeatTriggerScheduler: mocks.heartbeatTriggerSchedulerCtor,
+  shouldUseHybridExecutor: mockShouldUseHybridExecutor,
+  HybridExecutor: mockHybridExecutorCtor,
   });
 });
-
 vi.mock("@mariozechner/pi-coding-agent", () => ({
   AuthStorage: {
     create: vi.fn(() => mocks.authStorage),
@@ -847,6 +854,25 @@ describe("runServe", () => {
     expect(mocks.taskStores[0].close).toHaveBeenCalledTimes(1);
   });
 
+  it("enables HybridExecutor when env override is set and shuts it down before engine stop", async () => {
+    process.env.FUSION_HYBRID_EXECUTOR = "1";
+    mockShouldUseHybridExecutor.mockResolvedValue({ enabled: true, reason: "env-override" });
+
+    await runServe(4040, {});
+    expect(mockHybridExecutorCtor).toHaveBeenCalledTimes(1);
+    expect(mockHybridExecutorInitialize).toHaveBeenCalledTimes(1);
+
+    await triggerSignal("SIGTERM");
+
+    expect(mockHybridExecutorShutdown).toHaveBeenCalledTimes(1);
+    const shutdownOrder = [
+      mockHybridExecutorShutdown.mock.invocationCallOrder[0],
+      mocks.projectEngineInstances[0].stop.mock.invocationCallOrder[0],
+    ];
+    expect(shutdownOrder[0]).toBeLessThan(shutdownOrder[1]);
+    delete process.env.FUSION_HYBRID_EXECUTOR;
+  });
+
   it("listens on 127.0.0.1 by default and respects a custom host", async () => {
     await runServe(3010, {});
     expect(mocks.listenCalls[0]).toMatchObject({

packages/dashboard/src/__tests__/server.test.ts

@@ -177,6 +177,15 @@ async function flushStartupCleanupTasks(): Promise<void> {
   await Promise.resolve();
 }
 
+describe("createServer options", () => {
+  it("round-trips hybridExecutor on app locals", () => {
+    const store = createMockStore();
+    const hybridExecutor = { initialize: vi.fn(), shutdown: vi.fn() } as unknown as import("@fusion/engine").HybridExecutor;
+    const app = createServer(store, { hybridExecutor });
+    expect(app.locals.hybridExecutor).toBe(hybridExecutor);
+  });
+});
+
 describe("createServer AI session startup cleanup diagnostics", () => {
   const originalNodeEnv = process.env.NODE_ENV;
 

packages/dashboard/src/server.ts

@@ -193,6 +193,8 @@ export interface ServerOptions {
   /** ProjectEngineManager for uniform multi-project engine lifecycle.
    *  When provided, the server can resolve per-project engines for route handlers. */
   engineManager?: import("@fusion/engine").ProjectEngineManager;
+  /** Optional HybridExecutor orchestration context for multi-project runtime plumbing. */
+  hybridExecutor?: import("@fusion/engine").HybridExecutor;
   /** Shared CentralCore instance used by the engine manager.
    *  Routes that mutate central runtime state should use this instance so
    *  in-process listeners (for example global concurrency changes) are notified. */
@@ -597,6 +599,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
   }
 
   const app = express();
+  app.locals.hybridExecutor = options?.hybridExecutor;
   const runtimeLogger = options?.runtimeLogger ?? createRuntimeLogger("server");
   const mutationRateLimit = rateLimit(RATE_LIMITS.mutation);
   const setupRateLimit = rateLimit(RATE_LIMITS.api);

packages/engine/src/__tests__/hybrid-executor-startup.integration.test.ts

@@ -0,0 +1,108 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { CentralCore, RegisteredProject } from "@fusion/core";
+import { HybridExecutor } from "../hybrid-executor.js";
+import { shouldUseHybridExecutor } from "../hybrid-executor-gate.js";
+
+const projectManagerState = vi.hoisted(() => ({
+  projectIds: [] as string[],
+  stopAll: vi.fn().mockResolvedValue(undefined),
+}));
+
+vi.mock("../project-manager.js", () => ({
+  ProjectManager: vi.fn().mockImplementation(() => ({
+    on: vi.fn(),
+    addProject: vi.fn().mockImplementation(async (config: { projectId: string }) => {
+      projectManagerState.projectIds.push(config.projectId);
+    }),
+    getProjectIds: vi.fn().mockImplementation(() => [...projectManagerState.projectIds]),
+    stopAll: projectManagerState.stopAll,
+  })),
+}));
+
+vi.mock("../node-health-monitor.js", () => ({
+  NodeHealthMonitor: vi.fn().mockImplementation(() => ({
+    start: vi.fn().mockResolvedValue(undefined),
+    stop: vi.fn().mockResolvedValue(undefined),
+  })),
+}));
+
+function createCentralCore(overrides?: {
+  listNodes?: () => Promise<Array<{ id: string; type: "local" | "remote" }>>;
+  listProjects?: () => Promise<RegisteredProject[]>;
+}): CentralCore {
+  const now = new Date().toISOString();
+  const baseProject = {
+    id: "proj-1",
+    name: "Project 1",
+    path: "/tmp/proj-1",
+    status: "active",
+    isolationMode: "in-process",
+    createdAt: now,
+    updatedAt: now,
+  } as RegisteredProject;
+
+  return {
+    listNodes: overrides?.listNodes ?? (async () => [{ id: "local", type: "local" }]),
+    listProjects: overrides?.listProjects ?? (async () => [baseProject]),
+    getProject: vi.fn().mockResolvedValue(baseProject),
+    resolveLocalProjectWorkingDirectory: vi.fn().mockResolvedValue("/tmp/proj-1"),
+    on: vi.fn(),
+    removeAllListeners: vi.fn(),
+  } as unknown as CentralCore;
+}
+
+describe("hybrid executor startup integration", () => {
+  const originalEnv = process.env.FUSION_HYBRID_EXECUTOR;
+
+  beforeEach(() => {
+    projectManagerState.projectIds.length = 0;
+    projectManagerState.stopAll.mockClear();
+    delete process.env.FUSION_HYBRID_EXECUTOR;
+  });
+
+  afterEach(() => {
+    if (originalEnv === undefined) delete process.env.FUSION_HYBRID_EXECUTOR;
+    else process.env.FUSION_HYBRID_EXECUTOR = originalEnv;
+  });
+
+  it("disables gate for single local project", async () => {
+    const central = createCentralCore();
+    await expect(shouldUseHybridExecutor(central)).resolves.toEqual({
+      enabled: false,
+      reason: "single-project-local-only",
+    });
+  });
+
+  it("enables and initializes via env override", async () => {
+    process.env.FUSION_HYBRID_EXECUTOR = "1";
+    const central = createCentralCore();
+    await expect(shouldUseHybridExecutor(central)).resolves.toEqual({ enabled: true, reason: "env-override" });
+
+    const executor = new HybridExecutor(central);
+    await executor.initialize();
+
+    expect(executor.getProjectIds()).toContain("proj-1");
+    expect(executor.getNodeHealthMonitor()).not.toBeNull();
+  });
+
+  it("enables gate for multi-node", async () => {
+    const central = createCentralCore({
+      listNodes: async () => [
+        { id: "local", type: "local" },
+        { id: "remote", type: "remote" },
+      ],
+    });
+
+    await expect(shouldUseHybridExecutor(central)).resolves.toEqual({ enabled: true, reason: "multi-node" });
+  });
+
+  it("shutdown clears initialized state", async () => {
+    const central = createCentralCore();
+    const executor = new HybridExecutor(central);
+    await executor.initialize();
+    expect(executor.isInitialized()).toBe(true);
+
+    await executor.shutdown();
+    expect(executor.isInitialized()).toBe(false);
+  });
+});