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
@@ -3,6 +3,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
const runTaskCreate = vi.fn();
|
||||
const runTaskList = vi.fn();
|
||||
const runDesktop = vi.fn();
|
||||
const runServe = vi.fn();
|
||||
const runTaskPlan = vi.fn();
|
||||
const runTaskImportFromGitHub = vi.fn();
|
||||
const runSettingsShow = vi.fn();
|
||||
@@ -41,6 +42,10 @@ vi.mock("../commands/desktop.js", () => ({
|
||||
runDesktop,
|
||||
}));
|
||||
|
||||
vi.mock("../commands/serve.js", () => ({
|
||||
runServe,
|
||||
}));
|
||||
|
||||
vi.mock("../commands/task.js", () => ({
|
||||
runTaskCreate,
|
||||
runTaskList,
|
||||
@@ -277,6 +282,16 @@ describe("bin", () => {
|
||||
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 () => {
|
||||
await runBin(["desktop", "--dev", "--paused", "--interactive"]);
|
||||
|
||||
@@ -323,6 +338,7 @@ describe("bin", () => {
|
||||
const help = logSpy.mock.calls.map((call) => String(call[0])).join("\n");
|
||||
expect(help).toContain("fn project 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("--project, -P <name>");
|
||||
});
|
||||
|
||||
@@ -39,6 +39,7 @@ if (isBunBinary) {
|
||||
|
||||
// Dynamic imports so the pi-coding-agent config module sees PI_PACKAGE_DIR
|
||||
const { runDashboard } = await import("./commands/dashboard.js");
|
||||
const { runServe } = await import("./commands/serve.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 { runSettingsShow, runSettingsSet } = await import("./commands/settings.js");
|
||||
@@ -63,6 +64,8 @@ Usage:
|
||||
fn dashboard --paused Start with automation paused
|
||||
fn dashboard --dev Start web UI only (no AI engine)
|
||||
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 --dev Launch with hot-reload (connects to Vite dev server)
|
||||
fn desktop --paused Launch with automation paused
|
||||
@@ -138,7 +141,8 @@ Usage:
|
||||
|
||||
Options:
|
||||
--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)
|
||||
--paused Start with engine paused (automation disabled)
|
||||
--dev Start dashboard only (no AI engine)
|
||||
@@ -299,6 +303,19 @@ async function main() {
|
||||
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": {
|
||||
const paused = args.includes("--paused");
|
||||
const dev = args.includes("--dev");
|
||||
|
||||
459
packages/cli/src/commands/__tests__/serve.test.ts
Normal file
459
packages/cli/src/commands/__tests__/serve.test.ts
Normal 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");
|
||||
});
|
||||
});
|
||||
688
packages/cli/src/commands/serve.ts
Normal file
688
packages/cli/src/commands/serve.ts
Normal 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();
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user