feat(FN-1225): add headless fn serve command with health checks

- Add a new serve command implementation that runs the dashboard server in headless mode and exposes a health endpoint
- Register the serve command in CLI routing and extend bin command coverage for dispatch behavior
- Add comprehensive serve command tests for startup flow, options handling, and health-check responses
- Fix mission event ordering in MissionStore to keep health snapshots deterministic under concurrent updates
- Include a changeset for @gsxdsm/fusion documenting the new fn serve capability
This commit is contained in:
gsxdsm
2026-04-08 09:26:56 -07:00
parent f0878ce7b6
commit cc4c4dc7e1
8 changed files with 1235 additions and 7 deletions

View File

@@ -0,0 +1,5 @@
---
"@gsxdsm/fusion": minor
---
Add `fn serve` command to run Fusion as a headless node for remote mesh connections

View File

@@ -3,6 +3,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
const runTaskCreate = vi.fn(); const runTaskCreate = vi.fn();
const runTaskList = vi.fn(); const runTaskList = vi.fn();
const runDesktop = vi.fn(); const runDesktop = vi.fn();
const runServe = vi.fn();
const runTaskPlan = vi.fn(); const runTaskPlan = vi.fn();
const runTaskImportFromGitHub = vi.fn(); const runTaskImportFromGitHub = vi.fn();
const runSettingsShow = vi.fn(); const runSettingsShow = vi.fn();
@@ -41,6 +42,10 @@ vi.mock("../commands/desktop.js", () => ({
runDesktop, runDesktop,
})); }));
vi.mock("../commands/serve.js", () => ({
runServe,
}));
vi.mock("../commands/task.js", () => ({ vi.mock("../commands/task.js", () => ({
runTaskCreate, runTaskCreate,
runTaskList, runTaskList,
@@ -277,6 +282,16 @@ describe("bin", () => {
expect(errorSpy).toHaveBeenCalledWith("Unknown subcommand: node wat"); expect(errorSpy).toHaveBeenCalledWith("Unknown subcommand: node wat");
}); });
it("routes serve command with port, host, paused, and interactive flags", async () => {
await runBin(["serve", "--port", "5050", "--host", "127.0.0.1", "--paused", "--interactive"]);
expect(runServe).toHaveBeenCalledWith(5050, {
paused: true,
interactive: true,
host: "127.0.0.1",
});
});
it("routes desktop command flags", async () => { it("routes desktop command flags", async () => {
await runBin(["desktop", "--dev", "--paused", "--interactive"]); await runBin(["desktop", "--dev", "--paused", "--interactive"]);
@@ -323,6 +338,7 @@ describe("bin", () => {
const help = logSpy.mock.calls.map((call) => String(call[0])).join("\n"); const help = logSpy.mock.calls.map((call) => String(call[0])).join("\n");
expect(help).toContain("fn project list | ls"); expect(help).toContain("fn project list | ls");
expect(help).toContain("fn node list | ls"); expect(help).toContain("fn node list | ls");
expect(help).toContain("fn serve [--port <port>] [--host <host>] [--paused]");
expect(help).toContain("fn task comments <id>"); expect(help).toContain("fn task comments <id>");
expect(help).toContain("--project, -P <name>"); expect(help).toContain("--project, -P <name>");
}); });

View File

@@ -39,6 +39,7 @@ if (isBunBinary) {
// Dynamic imports so the pi-coding-agent config module sees PI_PACKAGE_DIR // Dynamic imports so the pi-coding-agent config module sees PI_PACKAGE_DIR
const { runDashboard } = await import("./commands/dashboard.js"); const { runDashboard } = await import("./commands/dashboard.js");
const { runServe } = await import("./commands/serve.js");
const { runDesktop } = await import("./commands/desktop.js"); const { runDesktop } = await import("./commands/desktop.js");
const { runTaskCreate, runTaskList, runTaskMove, runTaskMerge, runTaskUpdate, runTaskLog, runTaskLogs, runTaskShow, runTaskAttach, runTaskPause, runTaskUnpause, runTaskImportFromGitHub, runTaskDuplicate, runTaskArchive, runTaskUnarchive, runTaskRefine, runTaskPlan, runTaskDelete, runTaskRetry, runTaskComment, runTaskComments, runTaskSteer, runTaskPrCreate } = await import("./commands/task.js"); const { runTaskCreate, runTaskList, runTaskMove, runTaskMerge, runTaskUpdate, runTaskLog, runTaskLogs, runTaskShow, runTaskAttach, runTaskPause, runTaskUnpause, runTaskImportFromGitHub, runTaskDuplicate, runTaskArchive, runTaskUnarchive, runTaskRefine, runTaskPlan, runTaskDelete, runTaskRetry, runTaskComment, runTaskComments, runTaskSteer, runTaskPrCreate } = await import("./commands/task.js");
const { runSettingsShow, runSettingsSet } = await import("./commands/settings.js"); const { runSettingsShow, runSettingsSet } = await import("./commands/settings.js");
@@ -63,6 +64,8 @@ Usage:
fn dashboard --paused Start with automation paused fn dashboard --paused Start with automation paused
fn dashboard --dev Start web UI only (no AI engine) fn dashboard --dev Start web UI only (no AI engine)
fn dashboard --interactive Start with interactive port selection fn dashboard --interactive Start with interactive port selection
fn serve [--port <port>] [--host <host>] [--paused]
Start Fusion as a headless node (API + engine, no UI)
fn desktop Launch the Fusion desktop app (Electron) fn desktop Launch the Fusion desktop app (Electron)
fn desktop --dev Launch with hot-reload (connects to Vite dev server) fn desktop --dev Launch with hot-reload (connects to Vite dev server)
fn desktop --paused Launch with automation paused fn desktop --paused Launch with automation paused
@@ -138,7 +141,8 @@ Usage:
Options: Options:
--project, -P <name> Target a specific project (bypasses CWD detection) --project, -P <name> Target a specific project (bypasses CWD detection)
--port, -p <port> Dashboard port (default: 4040) --port, -p <port> Dashboard/serve port (default: 4040)
--host <host> Serve host (default: 0.0.0.0)
--interactive Interactive mode (port selection for dashboard, issue selection for import) --interactive Interactive mode (port selection for dashboard, issue selection for import)
--paused Start with engine paused (automation disabled) --paused Start with engine paused (automation disabled)
--dev Start dashboard only (no AI engine) --dev Start dashboard only (no AI engine)
@@ -299,6 +303,19 @@ async function main() {
break; break;
} }
case "serve": {
const portIdx = args.indexOf("--port");
const portIdxShort = args.indexOf("-p");
const pi = portIdx !== -1 ? portIdx : portIdxShort;
const port = pi !== -1 ? parseInt(args[pi + 1], 10) : 4040;
const paused = args.includes("--paused");
const interactive = args.includes("--interactive");
const hostIdx = args.indexOf("--host");
const host = hostIdx !== -1 && hostIdx + 1 < args.length ? args[hostIdx + 1] : undefined;
await runServe(port, { paused, interactive, host });
break;
}
case "desktop": { case "desktop": {
const paused = args.includes("--paused"); const paused = args.includes("--paused");
const dev = args.includes("--dev"); const dev = args.includes("--dev");

View File

@@ -0,0 +1,459 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { EventEmitter } from "node:events";
const mocks = vi.hoisted(() => {
type ListenCall = {
port: number;
host?: string;
server: {
close: ReturnType<typeof vi.fn>;
address: ReturnType<typeof vi.fn>;
once: (event: string, cb: (...args: unknown[]) => void) => void;
on: (event: string, cb: (...args: unknown[]) => void) => void;
emit: (event: string, ...args: unknown[]) => boolean;
};
};
const taskStores: any[] = [];
const automationStores: any[] = [];
const agentStores: any[] = [];
const centralInstances: any[] = [];
const triageInstances: any[] = [];
const executorInstances: any[] = [];
const schedulerInstances: any[] = [];
const stuckDetectorInstances: any[] = [];
const selfHealingInstances: any[] = [];
const cronRunnerInstances: any[] = [];
const missionAutopilotInstances: any[] = [];
const notifierInstances: any[] = [];
const listenCalls: ListenCall[] = [];
function createTaskStoreMock() {
const emitter = new EventEmitter();
const missionStore = {
listMissions: vi.fn().mockResolvedValue([]),
};
return {
init: vi.fn().mockResolvedValue(undefined),
watch: vi.fn().mockResolvedValue(undefined),
close: vi.fn(),
getFusionDir: vi.fn().mockReturnValue("/repo/.fusion"),
getMissionStore: vi.fn().mockReturnValue(missionStore),
getSettings: vi.fn().mockResolvedValue({
maxConcurrent: 2,
recycleWorktrees: false,
autoMerge: false,
pollIntervalMs: 60_000,
openrouterModelSync: false,
}),
updateSettings: vi.fn().mockResolvedValue(undefined),
listTasks: vi.fn().mockResolvedValue([]),
getTask: vi.fn(),
updateTask: vi.fn().mockResolvedValue(undefined),
moveTask: vi.fn().mockResolvedValue(undefined),
updatePrInfo: vi.fn().mockResolvedValue(undefined),
logEntry: vi.fn().mockResolvedValue(undefined),
on: vi.fn((event: string, handler: (...args: unknown[]) => void) => {
emitter.on(event, handler);
}),
off: vi.fn((event: string, handler: (...args: unknown[]) => void) => {
emitter.off(event, handler);
}),
emit: emitter.emit.bind(emitter),
};
}
function createMockServer(port: number) {
const emitter = new EventEmitter();
return Object.assign(emitter, {
close: vi.fn((cb?: () => void) => cb?.()),
address: vi.fn(() => ({ port, family: "IPv4", address: "0.0.0.0" })),
once: emitter.once.bind(emitter),
on: emitter.on.bind(emitter),
});
}
const taskStoreCtor = vi.fn().mockImplementation(() => {
const store = createTaskStoreMock();
taskStores.push(store);
return store;
});
const automationStoreCtor = vi.fn().mockImplementation(() => {
const automationStore = {
init: vi.fn().mockResolvedValue(undefined),
};
automationStores.push(automationStore);
return automationStore;
});
const agentStoreCtor = vi.fn().mockImplementation(() => {
const agentStore = {
init: vi.fn().mockResolvedValue(undefined),
};
agentStores.push(agentStore);
return agentStore;
});
const centralCoreCtor = vi.fn().mockImplementation(() => {
const instance = {
init: vi.fn().mockResolvedValue(undefined),
close: vi.fn().mockResolvedValue(undefined),
getProjectByPath: vi.fn().mockResolvedValue({ id: "project-1" }),
listNodes: vi.fn().mockResolvedValue([
{ id: "node-local", name: "local", type: "local", status: "offline" },
]),
updateNode: vi.fn().mockResolvedValue(undefined),
};
centralInstances.push(instance);
return instance;
});
const createServerMock = vi.fn().mockImplementation(() => ({
listen: vi.fn((port: number, host?: string) => {
const actualPort = port === 0 ? 5050 : port;
const server = createMockServer(actualPort);
listenCalls.push({ port, host, server });
queueMicrotask(() => {
server.emit("listening");
});
return server;
}),
}));
const triageCtor = vi.fn().mockImplementation(() => {
const triage = {
start: vi.fn(),
stop: vi.fn(),
markStuckAborted: vi.fn(),
};
triageInstances.push(triage);
return triage;
});
const executorCtor = vi.fn().mockImplementation(() => {
const executor = {
resumeOrphaned: vi.fn().mockResolvedValue(undefined),
markStuckAborted: vi.fn(),
handleLoopDetected: vi.fn().mockResolvedValue(false),
recoverCompletedTask: vi.fn().mockResolvedValue(false),
getExecutingTaskIds: vi.fn().mockReturnValue(new Set()),
};
executorInstances.push(executor);
return executor;
});
const schedulerCtor = vi.fn().mockImplementation(() => {
const scheduler = {
start: vi.fn(),
stop: vi.fn(),
};
schedulerInstances.push(scheduler);
return scheduler;
});
const stuckDetectorCtor = vi.fn().mockImplementation(() => {
const detector = {
start: vi.fn(),
stop: vi.fn(),
checkNow: vi.fn().mockResolvedValue(undefined),
};
stuckDetectorInstances.push(detector);
return detector;
});
const selfHealingCtor = vi.fn().mockImplementation(() => {
const manager = {
start: vi.fn(),
stop: vi.fn(),
checkStuckBudget: vi.fn().mockResolvedValue(true),
};
selfHealingInstances.push(manager);
return manager;
});
const cronRunnerCtor = vi.fn().mockImplementation(() => {
const cron = {
start: vi.fn(),
stop: vi.fn(),
};
cronRunnerInstances.push(cron);
return cron;
});
const missionAutopilotCtor = vi.fn().mockImplementation(() => {
const autopilot = {
start: vi.fn(),
stop: vi.fn(),
setScheduler: vi.fn(),
};
missionAutopilotInstances.push(autopilot);
return autopilot;
});
const notifierCtor = vi.fn().mockImplementation(() => {
const notifier = {
start: vi.fn(),
stop: vi.fn(),
};
notifierInstances.push(notifier);
return notifier;
});
const authStorage = {
getApiKey: vi.fn().mockResolvedValue(undefined),
};
const modelRegistry = {
registerProvider: vi.fn(),
refresh: vi.fn(),
};
return {
taskStores,
automationStores,
agentStores,
centralInstances,
triageInstances,
executorInstances,
schedulerInstances,
stuckDetectorInstances,
selfHealingInstances,
cronRunnerInstances,
missionAutopilotInstances,
notifierInstances,
listenCalls,
taskStoreCtor,
automationStoreCtor,
agentStoreCtor,
centralCoreCtor,
createServerMock,
triageCtor,
executorCtor,
schedulerCtor,
stuckDetectorCtor,
selfHealingCtor,
cronRunnerCtor,
missionAutopilotCtor,
notifierCtor,
authStorage,
modelRegistry,
reset() {
taskStores.length = 0;
automationStores.length = 0;
agentStores.length = 0;
centralInstances.length = 0;
triageInstances.length = 0;
executorInstances.length = 0;
schedulerInstances.length = 0;
stuckDetectorInstances.length = 0;
selfHealingInstances.length = 0;
cronRunnerInstances.length = 0;
missionAutopilotInstances.length = 0;
notifierInstances.length = 0;
listenCalls.length = 0;
},
};
});
vi.mock("@fusion/core", () => ({
TaskStore: mocks.taskStoreCtor,
AutomationStore: mocks.automationStoreCtor,
AgentStore: mocks.agentStoreCtor,
CentralCore: mocks.centralCoreCtor,
getTaskMergeBlocker: vi.fn().mockReturnValue(null),
}));
vi.mock("@fusion/dashboard", () => ({
createServer: mocks.createServerMock,
GitHubClient: vi.fn().mockImplementation(() => ({})),
}));
vi.mock("@fusion/engine", () => ({
TriageProcessor: mocks.triageCtor,
TaskExecutor: mocks.executorCtor,
Scheduler: mocks.schedulerCtor,
AgentSemaphore: vi.fn().mockImplementation(() => ({
run: (fn: () => Promise<unknown>) => fn(),
})),
WorktreePool: vi.fn().mockImplementation(() => ({
rehydrate: vi.fn(),
})),
aiMergeTask: vi.fn().mockResolvedValue({ merged: true }),
UsageLimitPauser: vi.fn().mockImplementation(() => ({})),
PRIORITY_MERGE: 100,
scanIdleWorktrees: vi.fn().mockResolvedValue([]),
cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0),
NtfyNotifier: mocks.notifierCtor,
PrMonitor: vi.fn().mockImplementation(() => ({
onNewComments: vi.fn(),
})),
PrCommentHandler: vi.fn().mockImplementation(() => ({
handleNewComments: vi.fn(),
createFollowUpTask: vi.fn().mockResolvedValue(undefined),
})),
CronRunner: mocks.cronRunnerCtor,
StuckTaskDetector: mocks.stuckDetectorCtor,
SelfHealingManager: mocks.selfHealingCtor,
MissionAutopilot: mocks.missionAutopilotCtor,
createAiPromptExecutor: vi.fn().mockResolvedValue(vi.fn().mockResolvedValue("ok")),
}));
vi.mock("@mariozechner/pi-coding-agent", () => ({
AuthStorage: {
create: vi.fn(() => mocks.authStorage),
},
DefaultPackageManager: vi.fn().mockImplementation(() => ({
resolve: vi.fn().mockResolvedValue({ extensions: [] }),
})),
ModelRegistry: vi.fn().mockImplementation(() => mocks.modelRegistry),
SettingsManager: {
create: vi.fn(() => ({})),
},
discoverAndLoadExtensions: vi.fn().mockResolvedValue({
runtime: { pendingProviderRegistrations: [] },
errors: [],
}),
getAgentDir: vi.fn(() => "/mock-agent-dir"),
createExtensionRuntime: vi.fn(),
}));
vi.mock("../dashboard.js", () => ({
promptForPort: vi.fn(async (port: number) => port),
getMergeStrategy: vi.fn((settings: { mergeStrategy?: "direct" | "pull-request" }) => settings.mergeStrategy ?? "direct"),
processPullRequestMergeTask: vi.fn().mockResolvedValue("waiting"),
}));
const { runServe } = await import("../serve.js");
describe("runServe", () => {
const originalCwd = process.cwd;
const originalOn = process.on;
const originalExit = process.exit;
let signalHandlers: Record<"SIGINT" | "SIGTERM", Array<() => void>>;
let logSpy: ReturnType<typeof vi.spyOn>;
let warnSpy: ReturnType<typeof vi.spyOn>;
let errorSpy: 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: [] };
logSpy = vi.spyOn(console, "log").mockImplementation(() => {});
warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
errorSpy = vi.spyOn(console, "error").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(() => {
logSpy.mockRestore();
warnSpy.mockRestore();
errorSpy.mockRestore();
cwdSpy.mockRestore();
processOnSpy.mockRestore();
process.cwd = originalCwd;
process.on = originalOn;
process.exit = originalExit;
});
it("initializes stores, starts engine services, and creates a headless server", async () => {
await runServe(4040, {});
expect(mocks.taskStoreCtor).toHaveBeenCalledWith("/repo");
expect(mocks.taskStores[0].init).toHaveBeenCalledTimes(1);
expect(mocks.taskStores[0].watch).toHaveBeenCalledTimes(1);
expect(mocks.automationStoreCtor).toHaveBeenCalledWith("/repo");
expect(mocks.automationStores[0].init).toHaveBeenCalledTimes(1);
expect(mocks.agentStores[0].init).toHaveBeenCalledTimes(1);
expect(mocks.createServerMock).toHaveBeenCalledTimes(1);
expect(mocks.createServerMock.mock.calls[0][1]).toMatchObject({
headless: true,
});
expect(mocks.triageInstances[0].start).toHaveBeenCalledTimes(1);
expect(mocks.schedulerInstances[0].start).toHaveBeenCalledTimes(1);
expect(mocks.missionAutopilotInstances[0].start).toHaveBeenCalledTimes(1);
expect(mocks.stuckDetectorInstances[0].start).toHaveBeenCalledTimes(1);
expect(mocks.selfHealingInstances[0].start).toHaveBeenCalledTimes(1);
expect(mocks.executorInstances[0].resumeOrphaned).toHaveBeenCalledTimes(1);
await triggerSignal("SIGINT");
});
it("sets enginePaused when started with paused=true", async () => {
await runServe(0, { paused: true });
expect(mocks.taskStores[0].updateSettings).toHaveBeenCalledWith({ enginePaused: true });
await triggerSignal("SIGTERM");
});
it("updates the local node status online on startup and offline on shutdown", async () => {
await runServe(4040, {});
const nodeCentral = mocks.centralInstances.find((instance) => instance.listNodes.mock.calls.length > 0);
expect(nodeCentral).toBeDefined();
expect(nodeCentral.updateNode).toHaveBeenCalledWith("node-local", { status: "online" });
await triggerSignal("SIGINT");
expect(nodeCentral.updateNode).toHaveBeenCalledWith("node-local", { status: "offline" });
});
it("stops engine services during shutdown", async () => {
await runServe(4040, {});
const listenCall = mocks.listenCalls[0];
expect(listenCall).toBeDefined();
await triggerSignal("SIGTERM");
expect(mocks.selfHealingInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.stuckDetectorInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.missionAutopilotInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.triageInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.schedulerInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.cronRunnerInstances[0].stop).toHaveBeenCalledTimes(1);
expect(mocks.notifierInstances[0].stop).toHaveBeenCalledTimes(1);
expect(listenCall.server.close).toHaveBeenCalledTimes(1);
expect(mocks.taskStores[0].close).toHaveBeenCalledTimes(1);
});
it("listens on 0.0.0.0 by default and respects a custom host", async () => {
await runServe(3010, {});
expect(mocks.listenCalls[0]).toMatchObject({
port: 3010,
host: "0.0.0.0",
});
await triggerSignal("SIGINT");
await runServe(3020, { host: "127.0.0.1" });
expect(mocks.listenCalls[1]).toMatchObject({
port: 3020,
host: "127.0.0.1",
});
await triggerSignal("SIGINT");
});
});

View File

@@ -0,0 +1,688 @@
import type { AddressInfo } from "node:net";
import {
TaskStore,
AutomationStore,
CentralCore,
AgentStore,
getTaskMergeBlocker,
} from "@fusion/core";
import { createServer, GitHubClient } from "@fusion/dashboard";
import {
TriageProcessor,
TaskExecutor,
Scheduler,
AgentSemaphore,
WorktreePool,
aiMergeTask,
UsageLimitPauser,
PRIORITY_MERGE,
scanIdleWorktrees,
cleanupOrphanedWorktrees,
NtfyNotifier,
PrMonitor,
PrCommentHandler,
CronRunner,
StuckTaskDetector,
SelfHealingManager,
MissionAutopilot,
createAiPromptExecutor,
} from "@fusion/engine";
import {
AuthStorage,
DefaultPackageManager,
ModelRegistry,
SettingsManager,
discoverAndLoadExtensions,
getAgentDir,
createExtensionRuntime,
} from "@mariozechner/pi-coding-agent";
import {
promptForPort,
getMergeStrategy,
processPullRequestMergeTask,
} from "./dashboard.js";
export async function runServe(
port: number,
opts: { interactive?: boolean; paused?: boolean; host?: string } = {},
) {
let selectedPort = port;
if (opts.interactive) {
try {
selectedPort = await promptForPort(port);
} catch (err: any) {
if (err.message === "Interactive prompt cancelled") {
console.log("Cancelled — exiting");
process.exit(0);
}
throw err;
}
}
const selectedHost = opts.host ?? "0.0.0.0";
const cwd = process.cwd();
const store = new TaskStore(cwd);
await store.init();
await store.watch();
const automationStore = new AutomationStore(cwd);
await automationStore.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
await agentStore.init();
let ntfyProjectId: string | undefined;
try {
const central = new CentralCore();
await central.init();
const registered = await central.getProjectByPath(cwd);
await central.close();
if (registered) {
ntfyProjectId = registered.id;
}
} catch {
// Central DB unavailable or project not registered — backward compatible
}
const notifier = new NtfyNotifier(store, { projectId: ntfyProjectId });
notifier.start();
if (opts.paused) {
await store.updateSettings({ enginePaused: true });
console.log("[engine] Starting in paused mode — automation disabled");
}
const initialSettings = await store.getSettings();
let cachedMaxConcurrent = initialSettings.maxConcurrent;
const semaphore = new AgentSemaphore(() => cachedMaxConcurrent);
const pool = new WorktreePool();
if (initialSettings.recycleWorktrees) {
const idlePaths = await scanIdleWorktrees(cwd, store);
if (idlePaths.length > 0) {
pool.rehydrate(idlePaths);
console.log(`[engine] Rehydrated pool with ${idlePaths.length} idle worktree(s)`);
}
} else {
const cleaned = await cleanupOrphanedWorktrees(cwd, store);
if (cleaned > 0) {
console.log(`[engine] Cleaned up ${cleaned} orphaned worktree(s)`);
}
}
const usageLimitPauser = new UsageLimitPauser(store);
const githubClient = new GitHubClient(process.env.GITHUB_TOKEN);
let activeMergeSession: { dispose: () => void } | null = null;
const rawMerge = (taskId: string) =>
aiMergeTask(store, cwd, taskId, {
pool,
usageLimitPauser,
agentStore,
onAgentText: (delta) => process.stdout.write(delta),
onSession: (session) => {
activeMergeSession = session;
},
});
const onMerge = (taskId: string) =>
semaphore.run(() => rawMerge(taskId), PRIORITY_MERGE);
store.on("settings:updated", ({ settings, previous }) => {
if (settings.globalPause && !previous.globalPause) {
if (activeMergeSession) {
console.log("[auto-merge] Global pause — terminating active merge session");
activeMergeSession.dispose();
activeMergeSession = null;
}
}
});
const mergeQueue: string[] = [];
const mergeActive = new Set<string>();
let mergeRunning = false;
function enqueueMerge(taskId: string): void {
if (mergeActive.has(taskId)) return;
mergeActive.add(taskId);
mergeQueue.push(taskId);
void drainMergeQueue();
}
async function drainMergeQueue(): Promise<void> {
if (mergeRunning) return;
mergeRunning = true;
try {
while (mergeQueue.length > 0) {
const taskId = mergeQueue.shift()!;
try {
const settings = await store.getSettings();
if (settings.globalPause || settings.enginePaused) {
console.log(
`[auto-merge] Skipping ${taskId}${settings.globalPause ? "global pause" : "engine paused"} active`,
);
continue;
}
if (!settings.autoMerge) {
console.log(`[auto-merge] Skipping ${taskId} — autoMerge disabled`);
continue;
}
const task = await store.getTask(taskId);
if (getTaskMergeBlocker(task)) {
continue;
}
const mergeStrategy = getMergeStrategy(settings);
if (mergeStrategy === "pull-request") {
console.log(`[auto-merge] Processing PR flow for ${taskId}...`);
const result = await processPullRequestMergeTask(store, cwd, taskId, githubClient);
if (result === "merged") {
console.log(`[auto-merge] ✓ ${taskId} merged via pull request`);
} else if (result === "waiting") {
console.log(`[auto-merge] … ${taskId} waiting on PR checks or reviews`);
}
} else {
console.log(`[auto-merge] Merging ${taskId}...`);
await onMerge(taskId);
console.log(`[auto-merge] ✓ ${taskId} merged`);
if (task.mergeRetries && task.mergeRetries > 0) {
await store.updateTask(taskId, { mergeRetries: 0 });
}
}
} catch (err: any) {
const errorMsg = err.message ?? String(err);
console.log(`[auto-merge] ✗ ${taskId}: ${errorMsg}`);
const settings = await store
.getSettings()
.catch(() => ({ autoResolveConflicts: true, mergeStrategy: "direct" as const }));
const task = await store.getTask(taskId).catch(() => null);
const mergeStrategy = getMergeStrategy(settings);
if (mergeStrategy === "direct") {
const isConflictError =
errorMsg.includes("conflict") || errorMsg.includes("Conflict");
if (task && isConflictError) {
const currentRetries = task.mergeRetries ?? 0;
const maxRetries = 3;
if (settings.autoResolveConflicts !== false && currentRetries < maxRetries) {
const newRetryCount = currentRetries + 1;
await store.updateTask(taskId, {
mergeRetries: newRetryCount,
status: null,
});
const delayMs = 5000 * Math.pow(2, currentRetries);
console.log(
`[auto-merge] ↻ ${taskId}: retry ${newRetryCount}/${maxRetries} in ${delayMs / 1000}s`,
);
setTimeout(() => {
enqueueMerge(taskId);
}, delayMs);
} else {
if (currentRetries >= maxRetries) {
console.log(
`[auto-merge] ⊘ ${taskId}: max retries (${maxRetries}) exceeded — manual resolution required`,
);
} else {
console.log(
`[auto-merge] ⊘ ${taskId}: autoResolveConflicts disabled — manual resolution required`,
);
}
try {
await store.updateTask(taskId, { status: null });
} catch {
// best-effort
}
}
} else {
try {
await store.updateTask(taskId, { status: null });
} catch {
// best-effort
}
}
} else {
try {
await store.updateTask(taskId, { status: null });
} catch {
// best-effort
}
}
} finally {
mergeActive.delete(taskId);
}
}
} finally {
mergeRunning = false;
}
}
store.on("task:moved", async ({ task, to }) => {
if (to !== "in-review") return;
if (getTaskMergeBlocker(task)) return;
try {
const settings = await store.getSettings();
if (settings.globalPause || settings.enginePaused) return;
if (!settings.autoMerge) return;
enqueueMerge(task.id);
} catch {
// ignore settings read errors
}
});
const authStorage = AuthStorage.create();
const modelRegistry = new ModelRegistry(authStorage);
try {
const agentDir = getAgentDir();
const piSettingsManager = SettingsManager.create(cwd, agentDir);
const packageManager = new DefaultPackageManager({
cwd,
agentDir,
settingsManager: piSettingsManager,
});
const resolvedPaths = await packageManager.resolve();
const packageExtensionPaths = resolvedPaths.extensions
.filter((r) => r.enabled)
.map((r) => r.path);
const extensionsResult = await discoverAndLoadExtensions(
packageExtensionPaths,
cwd,
undefined,
);
for (const { path, error } of extensionsResult.errors) {
console.log(`[extensions] Failed to load ${path}: ${error}`);
}
for (const {
name,
config,
extensionPath,
} of extensionsResult.runtime.pendingProviderRegistrations) {
try {
modelRegistry.registerProvider(name, config);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
console.log(
`[extensions] Failed to register provider from ${extensionPath}: ${message}`,
);
}
}
extensionsResult.runtime.pendingProviderRegistrations = [];
modelRegistry.refresh();
(async () => {
try {
const settings = await store.getSettings();
if (settings.openrouterModelSync === false) return;
const hasOrAuth = await authStorage.getApiKey("openrouter");
const headers: Record<string, string> = {};
if (hasOrAuth) headers["Authorization"] = `Bearer ${hasOrAuth}`;
const res = await fetch("https://openrouter.ai/api/v1/models", {
headers,
});
if (!res.ok) return;
const json = (await res.json()) as {
data?: Array<{
id: string;
name: string;
context_length?: number;
top_provider?: { max_completion_tokens?: number };
pricing?: Record<string, string>;
architecture?: {
modality?: string;
input_modalities?: string[];
};
}>;
};
const orModels = (json.data || []).map((m: any) => {
const id = (m.id || "").toLowerCase();
const name = (m.name || "").toLowerCase();
const reasoning =
id.includes(":thinking") ||
id.includes("-r1") ||
id.includes("/r1") ||
id.includes("o1-") ||
id.includes("o3-") ||
id.includes("o4-") ||
id.includes("reasoner") ||
name.includes("thinking") ||
name.includes("reasoner");
const hasVision =
m.architecture?.input_modalities?.includes("image") ??
m.architecture?.modality?.includes("multimodal") ??
false;
function parseCost(v?: string) {
const n = parseFloat(v || "0");
return isNaN(n) ? 0 : n * 1_000_000;
}
return {
id: m.id,
name: m.name || m.id,
reasoning,
input: (hasVision ? ["text", "image"] : ["text"]) as (
| "text"
| "image"
)[],
cost: {
input: parseCost(m.pricing?.prompt),
output: parseCost(m.pricing?.completion),
cacheRead: parseCost(m.pricing?.input_cache_read),
cacheWrite: parseCost(m.pricing?.input_cache_write),
},
contextWindow: m.context_length || 128000,
maxTokens: m.top_provider?.max_completion_tokens || 16384,
};
});
modelRegistry.registerProvider("openrouter", {
baseUrl: "https://openrouter.ai/api/v1",
apiKey: "OPENROUTER_API_KEY",
api: "openai-completions",
models: orModels,
});
console.log(
`[openrouter] Synced ${orModels.length} models from OpenRouter API`,
);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.log(`[openrouter] Failed to sync models: ${message}`);
}
})();
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
console.log(`[extensions] Failed to discover extensions: ${message}`);
createExtensionRuntime();
modelRegistry.refresh();
}
const missionAutopilot = new MissionAutopilot(store, store.getMissionStore());
const app = createServer(store, {
onMerge,
authStorage,
modelRegistry,
automationStore,
missionAutopilot,
headless: true,
});
const executorRef: { current: TaskExecutor | null } = { current: null };
const triageRef: { current: TriageProcessor | null } = { current: null };
const selfHealing = new SelfHealingManager(store, {
rootDir: cwd,
recoverCompletedTask: (task) =>
executorRef.current?.recoverCompletedTask(task) ?? Promise.resolve(false),
getExecutingTaskIds: () => executorRef.current?.getExecutingTaskIds() ?? new Set(),
});
const stuckTaskDetector = new StuckTaskDetector(store, {
beforeRequeue: (taskId) => selfHealing.checkStuckBudget(taskId),
onLoopDetected: (event) =>
executorRef.current?.handleLoopDetected(event) ?? Promise.resolve(false),
onStuck: (event) => {
triageRef.current?.markStuckAborted(event.taskId);
executorRef.current?.markStuckAborted(event.taskId, event.shouldRequeue);
console.log(
`[engine] ⚠ ${event.taskId} stuck (${event.reason}) — ` +
`no progress for ${Math.round(event.noProgressMs / 60_000)}min, ` +
`${event.activitySinceProgress} events since last progress — ` +
`terminated, ${event.shouldRequeue ? "will retry" : "budget exhausted"}`,
);
},
});
const triage = new TriageProcessor(store, cwd, {
semaphore,
usageLimitPauser,
stuckTaskDetector,
agentStore,
onSpecifyStart: (t) => console.log(`[engine] Specifying ${t.id}...`),
onSpecifyComplete: (t) => console.log(`[engine] ✓ ${t.id} → todo`),
onSpecifyError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
});
triageRef.current = triage;
const executor = new TaskExecutor(store, cwd, {
semaphore,
pool,
usageLimitPauser,
stuckTaskDetector,
agentStore,
onStart: (t, p) => console.log(`[engine] Executing ${t.id} in ${p}`),
onComplete: (t) => console.log(`[engine] ✓ ${t.id} → in-review`),
onError: (t, e) => console.log(`[engine] ✗ ${t.id}: ${e.message}`),
});
executorRef.current = executor;
const settings = await store.getSettings();
const prMonitor = new PrMonitor();
const prCommentHandler = new PrCommentHandler(store);
prMonitor.onNewComments((taskId, prInfo, comments) =>
prCommentHandler.handleNewComments(taskId, prInfo, comments),
);
const scheduler = new Scheduler(store, {
semaphore,
prMonitor,
missionStore: store.getMissionStore(),
missionAutopilot,
onSchedule: (t) => console.log(`[engine] Scheduled ${t.id}`),
onBlocked: (t, deps) =>
console.log(`[engine] ${t.id} blocked by ${deps.join(", ")}`),
onClosedPrFeedback: async (taskId, prInfo, comments) => {
await prCommentHandler.createFollowUpTask(taskId, prInfo, comments);
},
});
missionAutopilot.setScheduler(scheduler);
const aiPromptExecutor = await createAiPromptExecutor(cwd);
const cronRunner = new CronRunner(store, automationStore, { aiPromptExecutor });
cronRunner.start();
triage.start();
scheduler.start();
missionAutopilot.start();
stuckTaskDetector.start();
selfHealing.start();
executor.resumeOrphaned().catch((err) =>
console.error("[engine] Failed to resume orphaned tasks:", err),
);
if (settings.autoMerge) {
const existing = await store.listTasks();
const inReview = existing.filter((t) => !getTaskMergeBlocker(t));
if (inReview.length > 0) {
console.log(
`[auto-merge] Startup sweep: enqueueing ${inReview.length} in-review task(s)`,
);
for (const t of inReview) {
enqueueMerge(t.id);
}
}
}
store.on("settings:updated", async ({ settings: s, previous: prev }) => {
if (prev.globalPause && !s.globalPause) {
console.log("[engine] Global unpause — resuming agentic activity");
cachedMaxConcurrent = s.maxConcurrent ?? cachedMaxConcurrent;
executor.resumeOrphaned().catch((err) =>
console.error("[engine] Failed to resume orphaned tasks on unpause:", err),
);
if (s.autoMerge) {
try {
const tasks = await store.listTasks();
for (const t of tasks) {
if (!getTaskMergeBlocker(t)) {
enqueueMerge(t.id);
}
}
} catch {
// ignore errors in unpause sweep
}
}
}
});
store.on("settings:updated", async ({ settings: s, previous: prev }) => {
if (prev.enginePaused && !s.enginePaused) {
console.log("[engine] Engine unpaused — resuming agentic activity");
cachedMaxConcurrent = s.maxConcurrent ?? cachedMaxConcurrent;
executor.resumeOrphaned().catch((err) =>
console.error(
"[engine] Failed to resume orphaned tasks on engine unpause:",
err,
),
);
if (s.autoMerge) {
try {
const tasks = await store.listTasks();
for (const t of tasks) {
if (!getTaskMergeBlocker(t)) {
enqueueMerge(t.id);
}
}
} catch {
// ignore errors in unpause sweep
}
}
}
});
store.on("settings:updated", async ({ settings: s, previous: prev }) => {
if (s.taskStuckTimeoutMs !== prev.taskStuckTimeoutMs) {
console.log(
`[stuck-detector] Timeout changed to ${s.taskStuckTimeoutMs}ms — running immediate check`,
);
await stuckTaskDetector.checkNow();
}
});
let mergeRetryTimer: ReturnType<typeof setTimeout> | null = null;
async function scheduleMergeRetry(): Promise<void> {
const currentSettings = await store.getSettings().catch(() => settings);
const interval = currentSettings.pollIntervalMs ?? 15_000;
mergeRetryTimer = setTimeout(async () => {
try {
const s = await store.getSettings();
cachedMaxConcurrent = s.maxConcurrent;
if (!s.globalPause && !s.enginePaused && s.autoMerge) {
const tasks = await store.listTasks();
for (const t of tasks) {
if (!getTaskMergeBlocker(t)) {
enqueueMerge(t.id);
}
}
}
} catch {
// ignore errors in periodic sweep
}
void scheduleMergeRetry();
}, interval);
}
void scheduleMergeRetry();
const server = app.listen(selectedPort, selectedHost);
await new Promise<void>((resolve, reject) => {
server.once("listening", resolve);
server.once("error", reject);
});
const actualPort = (server.address() as AddressInfo).port;
let centralCore: CentralCore | null = null;
let localNodeId: string | undefined;
try {
centralCore = new CentralCore();
await centralCore.init();
const nodes = await centralCore.listNodes();
const localNode = nodes.find((node) => node.type === "local");
if (localNode) {
localNodeId = localNode.id;
await centralCore.updateNode(localNode.id, { status: "online" });
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to set local node online: ${message}`);
}
console.log();
console.log(` Fusion Node`);
console.log(` ────────────────────────`);
console.log(` → http://${selectedHost}:${actualPort}`);
console.log();
console.log(` Health: GET /api/health`);
console.log(` API: /api/*`);
console.log(` AI engine: ✓ active`);
console.log(` Press Ctrl+C to stop`);
console.log();
let shuttingDown = false;
const shutdown = async () => {
if (shuttingDown) return;
shuttingDown = true;
selfHealing.stop();
stuckTaskDetector.stop();
missionAutopilot.stop();
triage.stop();
scheduler.stop();
cronRunner.stop();
notifier.stop();
if (mergeRetryTimer) {
clearTimeout(mergeRetryTimer);
mergeRetryTimer = null;
}
if (centralCore && localNodeId) {
try {
await centralCore.updateNode(localNodeId, { status: "offline" });
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
console.warn(`[serve] Failed to set local node offline: ${message}`);
}
}
if (centralCore) {
await centralCore.close().catch(() => {
// best-effort
});
centralCore = null;
}
try {
server.close();
} catch {
// best-effort
}
store.close();
process.exit(0);
};
process.on("SIGINT", () => {
void shutdown();
});
process.on("SIGTERM", () => {
void shutdown();
});
}

View File

@@ -405,7 +405,7 @@ export class MissionStore extends EventEmitter<MissionStoreEvents> {
SELECT * SELECT *
FROM mission_events FROM mission_events
WHERE ${whereSql} WHERE ${whereSql}
ORDER BY timestamp DESC ORDER BY timestamp DESC, rowid DESC
LIMIT ? OFFSET ? LIMIT ? OFFSET ?
`).all(...params, limit, offset) as any[]; `).all(...params, limit, offset) as any[];

View File

@@ -91,6 +91,33 @@ async function REQUEST(
return performRequest(app, method, path, body, headers); return performRequest(app, method, path, body, headers);
} }
describe("createServer health and headless mode", () => {
it("returns liveness payload from /api/health", async () => {
const store = createMockStore();
const app = createServer(store);
const res = await GET(app, "/api/health");
expect(res.status).toBe(200);
expect(res.body).toEqual({
status: "ok",
version: expect.any(String),
uptime: expect.any(Number),
});
});
it("serves API routes but no frontend when headless=true", async () => {
const store = createMockStore();
const app = createServer(store, { headless: true });
const tasksRes = await GET(app, "/api/tasks");
expect(tasksRes.status).toBe(200);
const rootRes = await GET(app, "/");
expect(rootRes.status).toBe(404);
});
});
describe("API Error Handling Middleware", () => { describe("API Error Handling Middleware", () => {
let store: TaskStore; let store: TaskStore;

View File

@@ -43,6 +43,8 @@ const MAX_AI_SESSION_CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
export interface ServerOptions { export interface ServerOptions {
/** Custom merge handler — when provided, used instead of store.mergeTask */ /** Custom merge handler — when provided, used instead of store.mergeTask */
onMerge?: (taskId: string) => Promise<MergeResult>; onMerge?: (taskId: string) => Promise<MergeResult>;
/** When true, run API/websocket server only (skip frontend static assets + SPA fallback) */
headless?: boolean;
/** Maximum concurrent worktrees / execution slots (default 2) */ /** Maximum concurrent worktrees / execution slots (default 2) */
maxConcurrent?: number; maxConcurrent?: number;
/** Optional GitHub token for PR operations — falls back to GITHUB_TOKEN env var */ /** Optional GitHub token for PR operations — falls back to GITHUB_TOKEN env var */
@@ -142,6 +144,8 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
// Initialize terminal service with project root // Initialize terminal service with project root
getTerminalService(store.getRootDir()); getTerminalService(store.getRootDir());
const isHeadless = options?.headless === true;
// Serve built React app // Serve built React app
// Resolution order: // Resolution order:
// 1. FUSION_CLIENT_DIR env override (explicit) // 1. FUSION_CLIENT_DIR env override (explicit)
@@ -157,7 +161,9 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
? join(__dirname, "..", "dist", "client") ? join(__dirname, "..", "dist", "client")
: join(__dirname, "..", "client"); : join(__dirname, "..", "client");
app.use(express.static(clientDir)); if (!isHeadless) {
app.use(express.static(clientDir));
}
// Rate limiting — stricter limit on SSE connections // Rate limiting — stricter limit on SSE connections
app.get("/api/events", rateLimit(RATE_LIMITS.sse), async (req, res) => { app.get("/api/events", rateLimit(RATE_LIMITS.sse), async (req, res) => {
@@ -375,6 +381,14 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
); );
} }
app.get("/api/health", (_req, res) => {
res.json({
status: "ok",
version: process.env.npm_package_version ?? "0.4.0",
uptime: Math.floor(process.uptime()),
});
});
// REST API // REST API
app.use("/api", createApiRoutes(store, { ...options, aiSessionStore })); app.use("/api", createApiRoutes(store, { ...options, aiSessionStore }));
@@ -397,10 +411,12 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
res.status(500).json({ error: "Internal server error" }); res.status(500).json({ error: "Internal server error" });
}); });
// SPA fallback if (!isHeadless) {
app.get("/{*splat}", (_req, res) => { // SPA fallback
res.sendFile(join(clientDir, "index.html")); app.get("/{*splat}", (_req, res) => {
}); res.sendFile(join(clientDir, "index.html"));
});
}
const dashboardApp = app as DashboardExpressApp; const dashboardApp = app as DashboardExpressApp;
dashboardApp.terminalWsServer = null; dashboardApp.terminalWsServer = null;