feat(FN-4221): complete Step 3 — wire research dispatcher into engine

Fusion-Task-Id: FN-4221
Fusion-Task-Lineage: d710caa1-18b8-40d9-a70a-c5a4568484e3

packages/engine/src/__tests__/research-dispatcher.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, it, vi } from "vitest";
+import type { ResearchRun, ResearchStore } from "@fusion/core";
+import type { ResearchOrchestrator } from "../research-orchestrator.js";
+import { ResearchRunDispatcher } from "../research-dispatcher.js";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe("ResearchRunDispatcher", () => {
+  function createStore(runs: ResearchRun[]): ResearchStore {
+    return {
+      listRuns: vi.fn(() => runs),
+    } as unknown as ResearchStore;
+  }
+
+  it("dispatches queued runs", async () => {
+    const runs = [{ id: "RR-1", query: "hello", status: "queued" } as ResearchRun];
+    const store = createStore(runs);
+    const startRun = vi.fn(async () => ({ id: "RR-1" } as ResearchRun));
+    const orchestrator = { startRun } as unknown as ResearchOrchestrator;
+
+    const dispatcher = new ResearchRunDispatcher({ store, orchestrator, tickIntervalMs: 10 });
+    dispatcher.start();
+    await sleep(30);
+
+    expect(startRun).toHaveBeenCalledWith("RR-1", "hello", expect.objectContaining({ abortSignal: expect.any(AbortSignal) }));
+    await dispatcher.stop();
+  });
+
+  it("does not double-dispatch in-flight runs", async () => {
+    const runs = [{ id: "RR-1", query: "hello", status: "queued" } as ResearchRun];
+    const store = createStore(runs);
+    let resolveRun: (() => void) | undefined;
+    const startRun = vi.fn(() => new Promise<ResearchRun>((resolve) => {
+      resolveRun = () => resolve({ id: "RR-1" } as ResearchRun);
+    }));
+    const orchestrator = { startRun } as unknown as ResearchOrchestrator;
+
+    const dispatcher = new ResearchRunDispatcher({ store, orchestrator, tickIntervalMs: 10 });
+    dispatcher.start();
+    await sleep(40);
+    expect(startRun).toHaveBeenCalledTimes(1);
+
+    resolveRun?.();
+    await sleep(10);
+    await dispatcher.stop();
+  });
+
+  it("survives startRun rejection", async () => {
+    const runs = [{ id: "RR-1", query: "hello", status: "queued" } as ResearchRun];
+    const store = createStore(runs);
+    const startRun = vi.fn(async () => {
+      throw new Error("boom");
+    });
+    const orchestrator = { startRun } as unknown as ResearchOrchestrator;
+
+    const dispatcher = new ResearchRunDispatcher({ store, orchestrator, tickIntervalMs: 10 });
+    dispatcher.start();
+    await sleep(40);
+
+    expect(startRun).toHaveBeenCalled();
+    await dispatcher.stop();
+  });
+
+  it("stop cancels timer", async () => {
+    const runs = [{ id: "RR-1", query: "hello", status: "queued" } as ResearchRun];
+    const store = createStore(runs);
+    const startRun = vi.fn(async () => ({ id: "RR-1" } as ResearchRun));
+    const orchestrator = { startRun } as unknown as ResearchOrchestrator;
+
+    const dispatcher = new ResearchRunDispatcher({ store, orchestrator, tickIntervalMs: 10 });
+    dispatcher.start();
+    await sleep(30);
+    await dispatcher.stop();
+
+    const callsAfterStop = startRun.mock.calls.length;
+    await sleep(40);
+    expect(startRun).toHaveBeenCalledTimes(callsAfterStop);
+  });
+});

packages/engine/src/project-engine.ts

@@ -28,6 +28,7 @@ import { PRIORITY_MERGE } from "./concurrency.js";
 import { runtimeLog } from "./logger.js";
 import type { HeartbeatTriggerScheduler } from "./agent-heartbeat.js";
 import { ResearchOrchestrator } from "./research-orchestrator.js";
+import { ResearchRunDispatcher } from "./research-dispatcher.js";
 import { ResearchStepRunner } from "./research-step-runner.js";
 import { TunnelProcessManager } from "./remote-access/tunnel-process-manager.js";
 import type {
@@ -155,6 +156,7 @@ export class ProjectEngine {
   private cronRunner?: CronRunner;
   private automationStore?: AutomationStoreType;
   private researchOrchestrator?: ResearchOrchestrator;
+  private researchDispatcher?: ResearchRunDispatcher;
   private remoteTunnelManager?: TunnelProcessManager;
   private remoteTunnelRestoreDiagnostics: TunnelRestoreDiagnostics = {
     outcome: "skipped",
@@ -279,6 +281,11 @@ export class ProjectEngine {
         stepRunner: new ResearchStepRunner(),
         maxConcurrentRuns: settings.researchMaxConcurrentRuns ?? 3,
       });
+      this.researchDispatcher = new ResearchRunDispatcher({
+        store: store.getResearchStore(),
+        orchestrator: this.researchOrchestrator,
+      });
+      this.researchDispatcher.start();
     }
 
     this.remoteTunnelManager = new TunnelProcessManager();
@@ -544,6 +551,9 @@ export class ProjectEngine {
     this.gridlockDetector?.stop();
     this.cronRunner?.stop();
     this.setAutomationSubsystemHealth("not-initialized", "Automation subsystem stopped");
+    await this.researchDispatcher?.stop();
+    this.researchDispatcher = undefined;
+    this.researchOrchestrator = undefined;
 
     const tunnelManager = this.remoteTunnelManager;
     this.remoteTunnelManager = undefined;
@@ -648,6 +658,11 @@ export class ProjectEngine {
     return this.researchOrchestrator;
   }
 
+  /** Get the ResearchRunDispatcher (if initialized). Returns undefined before start(). */
+  getResearchDispatcher(): ResearchRunDispatcher | undefined {
+    return this.researchDispatcher;
+  }
+
   /** Get the remote tunnel manager (available after start()). */
   getRemoteTunnelManager(): TunnelProcessManager | undefined {
     return this.remoteTunnelManager;

packages/engine/src/research-dispatcher.ts

@@ -0,0 +1,86 @@
+import type { ResearchRun, ResearchStore } from "@fusion/core";
+import { createLogger, formatError } from "./logger.js";
+import type { ResearchOrchestrator } from "./research-orchestrator.js";
+
+const log = createLogger("research-dispatcher");
+
+export interface ResearchRunDispatcherOptions {
+  store: ResearchStore;
+  orchestrator: ResearchOrchestrator;
+  tickIntervalMs?: number;
+  shutdownTimeoutMs?: number;
+}
+
+export class ResearchRunDispatcher {
+  private readonly store: ResearchStore;
+  private readonly orchestrator: ResearchOrchestrator;
+  private readonly tickIntervalMs: number;
+  private readonly shutdownTimeoutMs: number;
+  private timer: NodeJS.Timeout | null = null;
+  private running = false;
+  private readonly inFlight = new Set<string>();
+  private readonly controllers = new Map<string, AbortController>();
+
+  constructor(options: ResearchRunDispatcherOptions) {
+    this.store = options.store;
+    this.orchestrator = options.orchestrator;
+    this.tickIntervalMs = Math.max(100, options.tickIntervalMs ?? 1_000);
+    this.shutdownTimeoutMs = Math.max(500, options.shutdownTimeoutMs ?? 5_000);
+  }
+
+  start(): void {
+    if (this.running) return;
+    this.running = true;
+    this.timer = setInterval(() => {
+      void this.tick();
+    }, this.tickIntervalMs);
+    void this.tick();
+  }
+
+  async stop(): Promise<void> {
+    this.running = false;
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = null;
+    }
+
+    for (const controller of this.controllers.values()) {
+      controller.abort(new Error("Research dispatcher stopped"));
+    }
+
+    const start = Date.now();
+    while (this.inFlight.size > 0 && Date.now() - start < this.shutdownTimeoutMs) {
+      await new Promise((resolve) => setTimeout(resolve, 25));
+    }
+  }
+
+  private async tick(): Promise<void> {
+    if (!this.running) return;
+
+    let queuedRuns: ResearchRun[] = [];
+    try {
+      queuedRuns = this.store.listRuns({ status: "queued" });
+    } catch (error) {
+      const { message, detail } = formatError(error);
+      log.warn(`Failed to list queued research runs: ${message}\n${detail}`);
+      return;
+    }
+
+    for (const run of queuedRuns) {
+      if (this.inFlight.has(run.id)) continue;
+      const controller = new AbortController();
+      this.inFlight.add(run.id);
+      this.controllers.set(run.id, controller);
+      void this.orchestrator
+        .startRun(run.id, run.query, { abortSignal: controller.signal })
+        .catch((error) => {
+          const { message, detail } = formatError(error);
+          log.warn(`Failed to dispatch research run ${run.id}: ${message}\n${detail}`);
+        })
+        .finally(() => {
+          this.inFlight.delete(run.id);
+          this.controllers.delete(run.id);
+        });
+    }
+  }
+}