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

feat(FN-1805): integrate PeerExchangeService into dashboard and serve runtimes

Browse Source 
- Wire PeerExchangeService into dashboard runtime with proper lifecycle management
- Wire PeerExchangeService into serve runtime for headless node mode
- Add mDNS discovery startup/shutdown coordination during runtime lifecycle
- Add comprehensive dashboard tests for peer exchange and discovery lifecycle
- Add serve tests covering node lifecycle and peer discovery operations
This commit is contained in:
Fusion
2026-04-15 01:23:57 -07:00
committed by gsxdsm
parent cfc2e16148
commit afd8608683
4 changed files with 431 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
packages/cli/src/commands/__tests__/dashboard.test.ts
View File

@@ -1,6 +1,12 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { EventEmitter } from "node:events";
// Use vi.hoisted to define mocks that need to be referenced in vi.mock
const { centralInstances } = vi.hoisted(() => {
const centralInstances: any[] = [];
return { centralInstances };
});
// ── Capture arguments ───────────────────────────────────────────────
// Minimal mock store backed by EventEmitter so `store.on` works
@@ -56,17 +62,27 @@ const mockProcessAndAudit = vi.fn().mockResolvedValue({
vi.mock("@fusion/core", () => ({
TaskStore: vi.fn().mockImplementation(() => makeMockStore()),
CentralCore: vi.fn().mockImplementation(() => ({
init: vi.fn().mockResolvedValue(undefined),
close: vi.fn().mockResolvedValue(undefined),
getProjectByPath: vi.fn().mockResolvedValue({ id: "project-1" }),
getProject: vi.fn().mockImplementation((id: string) =>
Promise.resolve({ id, name: `Project ${id}`, path: process.cwd(), status: "active", isolationMode: "in-process", createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() }),
),
listProjects: vi.fn().mockResolvedValue([
{ id: "project-1", name: "Test Project", path: process.cwd(), status: "active", isolationMode: "in-process", createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
]),
})),
CentralCore: vi.fn().mockImplementation(() => {
const instance = {
init: vi.fn().mockResolvedValue(undefined),
close: vi.fn().mockResolvedValue(undefined),
getProjectByPath: vi.fn().mockResolvedValue({ id: "project-1" }),
getProject: vi.fn().mockImplementation((id: string) =>
Promise.resolve({ id, name: `Project ${id}`, path: process.cwd(), status: "active", isolationMode: "in-process", createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() }),
),
listProjects: vi.fn().mockResolvedValue([
{ id: "project-1", name: "Test Project", path: process.cwd(), status: "active", isolationMode: "in-process", createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
]),
listNodes: vi.fn().mockResolvedValue([
{ id: "node-local", name: "local", type: "local", status: "offline" },
]),
updateNode: vi.fn().mockResolvedValue(undefined),
startDiscovery: vi.fn().mockResolvedValue({}),
stopDiscovery: vi.fn(),
};
centralInstances.push(instance);
return instance;
}),
AutomationStore: vi.fn().mockImplementation(() => ({
init: vi.fn().mockResolvedValue(undefined),
listSchedules: vi.fn().mockResolvedValue([]),
@@ -244,6 +260,10 @@ vi.mock("@fusion/engine", async (importOriginal) => {
addProject: vi.fn().mockResolvedValue({}),
stopAll: vi.fn().mockResolvedValue(undefined),
})),
PeerExchangeService: vi.fn().mockImplementation(() => ({
start: vi.fn(),
stop: vi.fn().mockResolvedValue(undefined),
})),
};
});
@@ -606,3 +626,59 @@ describe("runDashboard — per-project engine manager (multi-project)", () => {
});
});
describe("runDashboard — Peer exchange and discovery", () => {
beforeEach(async () => {
vi.clearAllMocks();
centralInstances.length = 0;
mockDiscoverAndLoadExtensions.mockResolvedValue({
runtime: { pendingProviderRegistrations: [] },
errors: [],
});
const { TaskStore } = await import("@fusion/core");
(TaskStore as unknown as ReturnType<typeof vi.fn>).mockImplementation(() => makeMockStore());
});
it("creates PeerExchangeService with CentralCore and calls start() in non-dev mode", async () => {
const { PeerExchangeService } = await import("@fusion/engine");
await runDashboard(0, {});
expect(PeerExchangeService).toHaveBeenCalledTimes(1);
const peerExchangeInstance = PeerExchangeService.mock.results[0]?.value;
expect(peerExchangeInstance.start).toHaveBeenCalledTimes(1);
});
it("creates CentralCore with startDiscovery and updateNode methods in non-dev mode", async () => {
await runDashboard(0, {});
// Verify CentralCore was created with the required methods
expect(centralInstances.length).toBeGreaterThanOrEqual(1);
const meshCentral = centralInstances[0];
expect(meshCentral).toBeDefined();
expect(typeof meshCentral.startDiscovery).toBe("function");
expect(typeof meshCentral.updateNode).toBe("function");
});
it("creates CentralCore and PeerExchangeService in dev mode", async () => {
const { PeerExchangeService: PeerExchangeServiceEngine } = await import("@fusion/engine");
await runDashboard(0, { dev: true });
// In dev mode, we create a separate CentralCore for mesh
expect(centralInstances.length).toBeGreaterThanOrEqual(1);
expect(PeerExchangeServiceEngine).toHaveBeenCalledTimes(1);
const peerExchangeInstance = PeerExchangeServiceEngine.mock.results[0]?.value;
expect(peerExchangeInstance.start).toHaveBeenCalledTimes(1);
});
it("creates CentralCore with startDiscovery and updateNode methods in dev mode", async () => {
await runDashboard(0, { dev: true });
expect(centralInstances.length).toBeGreaterThanOrEqual(1);
const meshCentral = centralInstances[0];
expect(meshCentral).toBeDefined();
expect(typeof meshCentral.startDiscovery).toBe("function");
expect(typeof meshCentral.updateNode).toBe("function");
});
});

147
packages/cli/src/commands/__tests__/serve.test.ts
View File

@@ -116,6 +116,8 @@ const mocks = vi.hoisted(() => {
{ id: "node-local", name: "local", type: "local", status: "offline" },
]),
updateNode: vi.fn().mockResolvedValue(undefined),
startDiscovery: vi.fn().mockResolvedValue({}),
stopDiscovery: vi.fn(),
};
centralInstances.push(instance);
return instance;
@@ -496,6 +498,10 @@ vi.mock("@fusion/engine", () => ({
startReconciliation: vi.fn(),
};
}),
PeerExchangeService: vi.fn().mockImplementation(() => ({
start: vi.fn(),
stop: vi.fn().mockResolvedValue(undefined),
})),
TriageProcessor: mocks.triageCtor,
TaskExecutor: mocks.executorCtor,
Scheduler: mocks.schedulerCtor,
@@ -1091,3 +1097,144 @@ describe("runServe — Semaphore boundary (task lanes only)", () => {
await triggerSignal("SIGINT");
});
});
describe("runServe — Peer exchange and discovery", () => {
const originalCwd = process.cwd;
const originalOn = process.on;
const originalExit = process.exit;
let signalHandlers: Record<"SIGINT" | "SIGTERM", Array<() => void>>;
let warnSpy: ReturnType<typeof vi.spyOn>;
let cwdSpy: ReturnType<typeof vi.spyOn>;
let processOnSpy: ReturnType<typeof vi.spyOn>;
async function triggerSignal(signal: "SIGINT" | "SIGTERM") {
const handlers = signalHandlers[signal];
expect(handlers.length).toBeGreaterThan(0);
handlers[handlers.length - 1]();
await new Promise((resolve) => setTimeout(resolve, 0));
}
beforeEach(() => {
vi.clearAllMocks();
mocks.reset();
signalHandlers = { SIGINT: [], SIGTERM: [] };
warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
cwdSpy = vi.spyOn(process, "cwd").mockReturnValue("/repo");
processOnSpy = vi.spyOn(process, "on").mockImplementation(((event: string, listener: () => void) => {
if (event === "SIGINT" || event === "SIGTERM") {
signalHandlers[event].push(listener);
}
return process;
}) as typeof process.on);
process.exit = vi.fn() as never;
});
afterEach(() => {
warnSpy.mockRestore();
cwdSpy.mockRestore();
processOnSpy.mockRestore();
process.cwd = originalCwd;
process.on = originalOn;
process.exit = originalExit;
});
it("creates PeerExchangeService with CentralCore and calls start()", async () => {
const { PeerExchangeService } = await import("@fusion/engine");
await runServe(4040, {});
expect(PeerExchangeService).toHaveBeenCalledTimes(1);
const peerExchangeInstance = PeerExchangeService.mock.results[0]?.value;
expect(peerExchangeInstance.start).toHaveBeenCalledTimes(1);
await triggerSignal("SIGINT");
});
it("calls centralCore.startDiscovery() with correct config after server starts", async () => {
await runServe(4040, {});
// Find the central core instance that was used
const nodeCentral = mocks.centralInstances.find((instance) => instance.listNodes.mock.calls.length > 0);
expect(nodeCentral).toBeDefined();
// startDiscovery should have been called with broadcast, listen, and correct port
expect(nodeCentral.startDiscovery).toHaveBeenCalledWith(
expect.objectContaining({
broadcast: true,
listen: true,
serviceType: "_fusion._tcp",
port: 4040,
staleTimeoutMs: 300_000,
}),
);
await triggerSignal("SIGINT");
});
it("starts discovery with port 5050 when port 0 is requested", async () => {
await runServe(0, {});
const nodeCentral = mocks.centralInstances.find((instance) => instance.listNodes.mock.calls.length > 0);
expect(nodeCentral).toBeDefined();
// Port 0 maps to 5050 in the mock
expect(nodeCentral.startDiscovery).toHaveBeenCalledWith(
expect.objectContaining({
port: 5050,
}),
);
await triggerSignal("SIGINT");
});
it("calls peerExchangeService.stop() on shutdown before engineManager.stopAll()", async () => {
const { PeerExchangeService } = await import("@fusion/engine");
await runServe(4040, {});
// Get the peer exchange instance
const peerExchangeInstance = PeerExchangeService.mock.results[0]?.value;
expect(peerExchangeInstance).toBeDefined();
// Reset mocks to isolate shutdown behavior
peerExchangeInstance.stop.mockClear();
await triggerSignal("SIGTERM");
// stop() should have been called
expect(peerExchangeInstance.stop).toHaveBeenCalledTimes(1);
});
it("calls centralCore.stopDiscovery() on shutdown before closing", async () => {
await runServe(4040, {});
const nodeCentral = mocks.centralInstances.find((instance) => instance.listNodes.mock.calls.length > 0);
expect(nodeCentral).toBeDefined();
// Reset to isolate shutdown behavior
nodeCentral.stopDiscovery.mockClear();
await triggerSignal("SIGTERM");
// stopDiscovery should have been called
expect(nodeCentral.stopDiscovery).toHaveBeenCalledTimes(1);
});
it("sets local node to offline on shutdown", async () => {
await runServe(4040, {});
const nodeCentral = mocks.centralInstances.find((instance) => instance.listNodes.mock.calls.length > 0);
expect(nodeCentral).toBeDefined();
// Reset to isolate shutdown behavior
nodeCentral.updateNode.mockClear();
await triggerSignal("SIGTERM");
// Should have been called twice: once to set online, once to set offline
expect(nodeCentral.updateNode).toHaveBeenCalledWith("node-local", { status: "offline" });
});
});

142
packages/cli/src/commands/dashboard.ts
View File

@@ -1,7 +1,7 @@
import type { AddressInfo } from "node:net";
import { TaskStore, AutomationStore, CentralCore, AgentStore, PluginStore, PluginLoader, getTaskMergeBlocker } from "@fusion/core";
import { createServer, GitHubClient, createSkillsAdapter, getProjectSettingsPath } from "@fusion/dashboard";
import { aiMergeTask, MissionAutopilot, MissionExecutionLoop, HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext, ProjectEngineManager } from "@fusion/engine";
import { aiMergeTask, MissionAutopilot, MissionExecutionLoop, HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext, ProjectEngineManager, PeerExchangeService } from "@fusion/engine";
import { AuthStorage, DefaultPackageManager, ModelRegistry, discoverAndLoadExtensions, getAgentDir, createExtensionRuntime } from "@mariozechner/pi-coding-agent";
import {
getMergeStrategy,
@@ -486,6 +486,16 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
//
let app: ReturnType<typeof createServer>;
// ── Mesh networking: peer exchange + mDNS discovery ──────────────────
//
// peerExchangeService: periodically syncs peer info with connected nodes
// centralCoreForMesh: CentralCore for discovery/node lifecycle (may differ from centralCoreForEngine)
// localNodeIdForMesh: tracks the local node ID for cleanup on shutdown
//
let peerExchangeService: PeerExchangeService | null = null;
let centralCoreForMesh: CentralCore | null = null;
let localNodeIdForMesh: string | undefined;
// Start the AI engine (unless in dev mode)
if (!opts.dev) {
// ── ProjectEngineManager: uniform engine lifecycle for all projects ──
@@ -521,6 +531,21 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
// required for correctness — reconciliation handles all cases.
engineManager.startReconciliation();
// ── PeerExchangeService: gossip protocol for mesh peer discovery ──────
//
// Reuse centralCoreForEngine for peer exchange since it handles all mesh ops.
//
peerExchangeService = new PeerExchangeService(centralCoreForEngine);
try {
peerExchangeService.start();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to start peer exchange service: ${message}`);
}
// Use the same CentralCore instance for mesh operations
centralCoreForMesh = centralCoreForEngine;
// Resolve the cwd project's engine for the dashboard's HTTP layer defaults.
// The engine for the cwd project provides onMerge, automationStore, etc.
// for requests that arrive without ?projectId=. This is transitional —
@@ -590,6 +615,33 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
// Stop all project engines uniformly
await engineManager.stopAll();
// Stop peer exchange service
if (peerExchangeService) {
try {
await peerExchangeService.stop();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to stop peer exchange service: ${message}`);
}
}
// Stop mDNS discovery and set local node offline
if (centralCoreForMesh && localNodeIdForMesh) {
try {
centralCoreForMesh.stopDiscovery();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to stop mDNS discovery: ${message}`);
}
try {
await centralCoreForMesh.updateNode(localNodeIdForMesh, { status: "offline" });
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to set local node offline: ${message}`);
}
}
await centralCoreForEngine.close().catch(() => {});
store.close();
@@ -607,6 +659,23 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
});
} else {
// Dev mode: create HeartbeatMonitor + TriggerScheduler inline (engine not started)
// ── Mesh networking for dev mode ─────────────────────────────────────
//
// In dev mode we don't use the engine's CentralCore, so create a separate
// instance for peer exchange and mDNS discovery.
//
try {
centralCoreForMesh = new CentralCore();
await centralCoreForMesh.init();
peerExchangeService = new PeerExchangeService(centralCoreForMesh);
peerExchangeService.start();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to initialize mesh networking: ${message}`);
}
try {
heartbeatMonitorImpl = new HeartbeatMonitor({
store: agentStore,
@@ -727,6 +796,37 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
stopDiagnosticInterval();
if (triggerScheduler) triggerScheduler.stop();
if (heartbeatMonitorImpl) heartbeatMonitorImpl.stop();
// Stop peer exchange service
if (peerExchangeService) {
try {
await peerExchangeService.stop();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to stop peer exchange service: ${message}`);
}
}
// Stop mDNS discovery and set local node offline
if (centralCoreForMesh && localNodeIdForMesh) {
try {
centralCoreForMesh.stopDiscovery();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to stop mDNS discovery: ${message}`);
}
try {
await centralCoreForMesh.updateNode(localNodeIdForMesh, { status: "offline" });
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to set local node offline: ${message}`);
}
}
if (centralCoreForMesh) {
await centralCoreForMesh.close().catch(() => {});
}
store.close();
process.exit(0);
};
@@ -750,13 +850,51 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
}
});
server.on("listening", () => {
server.on("listening", async () => {
const actualPort = (server.address() as AddressInfo).port;
if (actualPort !== selectedPort) {
console.log(`⚠ Port ${selectedPort} in use, using ${actualPort} instead`);
}
// ── mDNS discovery: broadcast presence and listen for other nodes ───────
//
// Advertises this node on the local network and discovers other Fusion nodes
// without requiring manual configuration.
//
if (centralCoreForMesh) {
try {
await centralCoreForMesh.startDiscovery({
broadcast: true,
listen: true,
serviceType: "_fusion._tcp",
port: actualPort,
staleTimeoutMs: 300_000,
});
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to start mDNS discovery: ${message}`);
}
}
// ── CentralCore: set local node online ─────────────────────────────────
//
// Find the local node and mark it as online now that we know the port.
//
if (centralCoreForMesh) {
try {
const nodes = await centralCoreForMesh.listNodes();
const localNode = nodes.find((node) => node.type === "local");
if (localNode) {
localNodeIdForMesh = localNode.id;
await centralCoreForMesh.updateNode(localNode.id, { status: "online" });
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[dashboard] Failed to set local node online: ${message}`);
}
}
console.log();
console.log(` fn board`);
console.log(` ────────────────────────`);

58
packages/cli/src/commands/serve.ts
View File

@@ -20,7 +20,7 @@ import {
} from "@fusion/core";
import type { AutomationRunResult, ScheduledTask } from "@fusion/core";
import { createServer, GitHubClient, createSkillsAdapter, getProjectSettingsPath } from "@fusion/dashboard";
import { ProjectEngineManager } from "@fusion/engine";
import { ProjectEngineManager, PeerExchangeService } from "@fusion/engine";
import {
AuthStorage,
DefaultPackageManager,
@@ -310,6 +310,23 @@ export async function runServe(
// required for correctness — reconciliation handles all cases.
engineManager.startReconciliation();
// ── PeerExchangeService: gossip protocol for mesh peer discovery ──────
//
// Periodically exchanges peer information with connected remote nodes
// to keep the mesh state up-to-date across all nodes.
// Uses sharedCentralCore since it's the CentralCore instance available at this point.
//
let peerExchangeService: PeerExchangeService | null = null;
if (sharedCentralCore) {
peerExchangeService = new PeerExchangeService(sharedCentralCore);
try {
peerExchangeService.start();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to start peer exchange service: ${message}`);
}
}
// Get the cwd project's engine and store for the HTTP layer.
// serve.ts needs a store for plugin setup, diagnostics, and the server.
const cwdEngine = ntfyProjectId ? engineManager.getEngine(ntfyProjectId) : undefined;
@@ -541,6 +558,27 @@ export async function runServe(
const actualPort = (server.address() as AddressInfo).port;
// ── mDNS discovery: broadcast presence and listen for other nodes ───────
//
// Advertises this node on the local network and discovers other Fusion nodes
// without requiring manual configuration.
// Uses sharedCentralCore since it's the CentralCore instance available at this point.
//
if (sharedCentralCore) {
try {
await sharedCentralCore.startDiscovery({
broadcast: true,
listen: true,
serviceType: "_fusion._tcp",
port: actualPort,
staleTimeoutMs: 300_000,
});
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to start mDNS discovery: ${message}`);
}
}
// ── CentralCore: node registration ────────────────────────────────────
//
// Reuse the shared CentralCore instance created earlier (for ntfyProjectId).
@@ -610,6 +648,16 @@ export async function runServe(
// Stop all project engines uniformly
await engineManager.stopAll();
// Stop peer exchange service
if (peerExchangeService) {
try {
await peerExchangeService.stop();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to stop peer exchange service: ${message}`);
}
}
if (centralCore && localNodeId) {
try {
await centralCore.updateNode(localNodeId, { status: "offline" });
@@ -620,6 +668,14 @@ export async function runServe(
}
if (centralCore) {
// Stop mDNS discovery
try {
centralCore.stopDiscovery();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to stop mDNS discovery: ${message}`);
}
await centralCore.close().catch(() => {
// best-effort
});