FN-7082: add live running-agent count seam
Add a side-effect-safe core seam so global concurrency reads can report live running-agent counts. - Add core helpers to register a running-agent count source and derive active project totals. - Wire CentralCore and dashboard server setup to read live counts from already-open stores without starting runtimes. - Update the global concurrency route to preserve slot bookkeeping while sourcing currentlyActive/projectsActive from the live seam. - Cover the seam, dashboard store resolver, server wiring, and route behavior with focused tests and document the multi-project behavior. Files changed: .changeset/fn-7082-live-running-agent-count-seam.md | 7 ++ docs/multi-project.md | 2 +- packages/core/src/__tests__/central-core.test.ts | 106 +++++++++++++++++++++ packages/core/src/central-core.ts | 28 ++++++ packages/core/src/index.ts | 7 ++ packages/core/src/live-agent-count.ts | 38 ++++++++ .../dashboard/src/__tests__/project-routes.test.ts | 20 +++- .../src/__tests__/project-store-resolver.test.ts | 71 ++++++++++++++ packages/dashboard/src/__tests__/server.test.ts | 45 ++++++++- packages/dashboard/src/project-store-resolver.ts | 24 +++++ packages/dashboard/src/routes.ts | 25 +---- packages/dashboard/src/server.ts | 42 +++++++- 12 files changed, 387 insertions(+), 28 deletions(-) Fusion-Task-Id: FN-7082 Fusion-Task-Lineage: 4dbca204-8bea-4de5-b56b-7468b61575ce Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7082-live-running-agent-count-seam.md
Normal file
7
.changeset/fn-7082-live-running-agent-count-seam.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Concurrency panels now read running-agent counts from a single live source shared across the app.
|
||||
category: internal
|
||||
dev: Adds a side-effect-safe CentralCore.getLiveRunningAgentCounts() seam (DI source via setRunningAgentCountSource) that derives counts from in-progress task columns of already-open project stores without starting engines/watchers or mutating slot/health bookkeeping; GET /api/global-concurrency is rewired onto it, preserving globalMaxConcurrent/queuedCount and acquireGlobalSlot/releaseGlobalSlot semantics.
|
||||
@@ -110,7 +110,7 @@ Central health tracking keeps mutable project metrics, including:
|
||||
|
||||
## Global Concurrency Management
|
||||
|
||||
A singleton central record enforces system-wide limits so one project cannot monopolize all execution slots.
|
||||
A singleton central record enforces system-wide limits so one project cannot monopolize all execution slots. Slot acquire/release bookkeeping remains separate from read-only running-agent displays: live read surfaces derive `currentlyActive` and per-project active counts from `in-progress` tasks in already-open project stores, while the persisted `globalMaxConcurrent` cap and `queuedCount` continue to come from central concurrency state.
|
||||
|
||||
## Plugin Scope in Multi-Project Mode
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createHash } from "node:crypto";
|
||||
import { CentralCore } from "../central-core.js";
|
||||
import { setRunningAgentCountSource } from "../live-agent-count.js";
|
||||
import { NodeDiscovery } from "../node-discovery.js";
|
||||
import { NodeConnection, type ConnectionResult } from "../node-connection.js";
|
||||
import { getAppVersion } from "../app-version.js";
|
||||
@@ -32,6 +33,7 @@ describe("CentralCore", () => {
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
setRunningAgentCountSource(undefined);
|
||||
await central.close();
|
||||
vi.useRealTimers();
|
||||
vi.restoreAllMocks();
|
||||
@@ -2663,6 +2665,110 @@ describe("CentralCore", () => {
|
||||
it("should throw when releasing for non-existent project", async () => {
|
||||
await expect(central.releaseGlobalSlot("nonexistent")).rejects.toThrow("not found");
|
||||
});
|
||||
|
||||
async function registerProjectForLiveCount(name: string) {
|
||||
const projectPath = join(tempDir, name);
|
||||
mkdirSync(projectPath);
|
||||
projectPaths.push(projectPath);
|
||||
return central.registerProject({ name, path: projectPath });
|
||||
}
|
||||
|
||||
it("derives live running-agent counts from a side-effect-safe source across project data states", async () => {
|
||||
const projectA = await registerProjectForLiveCount("live-count-a");
|
||||
const projectB = await registerProjectForLiveCount("live-count-b");
|
||||
const unopenedProject = await registerProjectForLiveCount("live-count-unopened");
|
||||
await central.updateGlobalConcurrency({ globalMaxConcurrent: 2 });
|
||||
|
||||
const source = vi.fn(async (projectIds: readonly string[]) => {
|
||||
expect(projectIds).toEqual([projectA.id, projectB.id, unopenedProject.id]);
|
||||
return {
|
||||
[projectA.id]: 3,
|
||||
[projectB.id]: 2,
|
||||
[unopenedProject.id]: 0,
|
||||
};
|
||||
});
|
||||
|
||||
const counts = await central.getLiveRunningAgentCounts({ source });
|
||||
|
||||
expect(source).toHaveBeenCalledOnce();
|
||||
expect(counts).toEqual({
|
||||
currentlyActive: 5,
|
||||
projectsActive: {
|
||||
[projectA.id]: 3,
|
||||
[projectB.id]: 2,
|
||||
},
|
||||
});
|
||||
expect(counts.currentlyActive).toBeGreaterThan(2);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "zero in-progress", perProject: { a: 0 }, expected: { currentlyActive: 0, projectsActive: {} } },
|
||||
{ name: "one in-progress", perProject: { a: 1 }, expected: { currentlyActive: 1, projectsActive: { a: 1 } } },
|
||||
{ name: "multiple in one project", perProject: { a: 4 }, expected: { currentlyActive: 4, projectsActive: { a: 4 } } },
|
||||
{ name: "multiple projects", perProject: { a: 2, b: 3, c: 0 }, expected: { currentlyActive: 5, projectsActive: { a: 2, b: 3 } } },
|
||||
])("normalizes live running-agent count data state: $name", async ({ perProject, expected }) => {
|
||||
await registerProjectForLiveCount("live-count-state");
|
||||
|
||||
await expect(central.getLiveRunningAgentCounts({ source: async () => perProject })).resolves.toEqual(expected);
|
||||
});
|
||||
|
||||
it("falls back to persisted slot and health bookkeeping when no live source is registered", async () => {
|
||||
const project = await registerProjectForLiveCount("live-count-fallback");
|
||||
await central.acquireGlobalSlot(project.id);
|
||||
await central.acquireGlobalSlot(project.id);
|
||||
|
||||
const persisted = await central.getGlobalConcurrencyState();
|
||||
const counts = await central.getLiveRunningAgentCounts();
|
||||
|
||||
expect(counts).toEqual({
|
||||
currentlyActive: persisted.currentlyActive,
|
||||
projectsActive: persisted.projectsActive,
|
||||
});
|
||||
expect(counts.projectsActive).toEqual({ [project.id]: 2 });
|
||||
});
|
||||
|
||||
it("does not mutate slot or health bookkeeping during a live-count read", async () => {
|
||||
const project = await registerProjectForLiveCount("live-count-no-mutation");
|
||||
await central.acquireGlobalSlot(project.id);
|
||||
await central.updateGlobalConcurrency({ queuedCount: 4 });
|
||||
const beforeGlobal = await central.getGlobalConcurrencyState();
|
||||
const beforeHealth = await central.getProjectHealth(project.id);
|
||||
const watchSpy = vi.fn();
|
||||
const engineStartSpy = vi.fn();
|
||||
|
||||
const counts = await central.getLiveRunningAgentCounts({
|
||||
source: async () => {
|
||||
expect(watchSpy).not.toHaveBeenCalled();
|
||||
expect(engineStartSpy).not.toHaveBeenCalled();
|
||||
return { [project.id]: 7 };
|
||||
},
|
||||
});
|
||||
|
||||
expect(counts).toEqual({ currentlyActive: 7, projectsActive: { [project.id]: 7 } });
|
||||
expect(await central.getGlobalConcurrencyState()).toEqual(beforeGlobal);
|
||||
expect(await central.getProjectHealth(project.id)).toEqual(beforeHealth);
|
||||
expect(watchSpy).not.toHaveBeenCalled();
|
||||
expect(engineStartSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps acquire and release slot bookkeeping isolated from live count reads", async () => {
|
||||
const project = await registerProjectForLiveCount("live-count-limiter-isolation");
|
||||
const source = vi.fn(async () => ({ [project.id]: 5 }));
|
||||
await central.acquireGlobalSlot(project.id);
|
||||
|
||||
expect((await central.getGlobalConcurrencyState()).currentlyActive).toBe(1);
|
||||
expect(await central.getLiveRunningAgentCounts({ source })).toEqual({
|
||||
currentlyActive: 5,
|
||||
projectsActive: { [project.id]: 5 },
|
||||
});
|
||||
|
||||
await central.releaseGlobalSlot(project.id);
|
||||
expect((await central.getGlobalConcurrencyState()).currentlyActive).toBe(0);
|
||||
expect(await central.getLiveRunningAgentCounts({ source })).toEqual({
|
||||
currentlyActive: 5,
|
||||
projectsActive: { [project.id]: 5 },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("utility methods", () => {
|
||||
|
||||
@@ -97,6 +97,12 @@ import {
|
||||
ensureGitRepositoryForProjectPath,
|
||||
type GitRepositoryEnsureOutcome,
|
||||
} from "./git-repository.js";
|
||||
import {
|
||||
deriveRunningAgentCounts,
|
||||
getRunningAgentCountSource,
|
||||
type RunningAgentCountSource,
|
||||
type RunningAgentCounts,
|
||||
} from "./live-agent-count.js";
|
||||
// ── Event Types ───────────────────────────────────────────────────────────
|
||||
|
||||
export interface CentralCoreEvents {
|
||||
@@ -2649,6 +2655,28 @@ export class CentralCore extends EventEmitter<CentralCoreEvents> {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Read live running-agent counts through the side-effect-safe host seam.
|
||||
* Falls back to persisted concurrency/health bookkeeping when no host source
|
||||
* is registered so headless core callers keep their previous semantics.
|
||||
*/
|
||||
async getLiveRunningAgentCounts(options?: { source?: RunningAgentCountSource }): Promise<RunningAgentCounts> {
|
||||
this.ensureInitialized();
|
||||
|
||||
const source = options?.source ?? getRunningAgentCountSource();
|
||||
if (!source) {
|
||||
const state = await this.getGlobalConcurrencyState();
|
||||
return {
|
||||
currentlyActive: state.currentlyActive,
|
||||
projectsActive: state.projectsActive,
|
||||
};
|
||||
}
|
||||
|
||||
const projectIds = (await this.listProjects()).map((project) => project.id);
|
||||
const perProject = await source(projectIds);
|
||||
return deriveRunningAgentCounts(perProject);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update global concurrency settings.
|
||||
* Only allows updating globalMaxConcurrent, currentlyActive, and queuedCount.
|
||||
|
||||
@@ -404,6 +404,13 @@ export {
|
||||
getCreateInteractiveAiSessionFactory,
|
||||
type AgentMessage,
|
||||
} from "./ai-engine-loader.js";
|
||||
export {
|
||||
setRunningAgentCountSource,
|
||||
getRunningAgentCountSource,
|
||||
deriveRunningAgentCounts,
|
||||
type RunningAgentCountSource,
|
||||
type RunningAgentCounts,
|
||||
} from "./live-agent-count.js";
|
||||
export {
|
||||
setTaskCreatedHook,
|
||||
getTaskCreatedHook,
|
||||
|
||||
38
packages/core/src/live-agent-count.ts
Normal file
38
packages/core/src/live-agent-count.ts
Normal file
@@ -0,0 +1,38 @@
|
||||
export type RunningAgentCountSource = (projectIds: readonly string[]) => Promise<Record<string, number>> | Record<string, number>;
|
||||
|
||||
let runningAgentCountSource: RunningAgentCountSource | undefined;
|
||||
|
||||
/**
|
||||
* FNXC:GlobalConcurrencyControls 2026-06-26-17:22:
|
||||
* Live running-agent counts must come from side-effect-safe reads of `in-progress` task columns, not from stale slot or health bookkeeping. This DI seam lets dashboard, CLI, remote-node, and plugin consumers share one core path without starting project engines/runtimes, opening watchers, or mutating `globalConcurrency.currentlyActive`, `globalConcurrency.queuedCount`, or `projectHealth.inFlightAgentCount`.
|
||||
*/
|
||||
export function setRunningAgentCountSource(fn: RunningAgentCountSource | undefined): void {
|
||||
runningAgentCountSource = fn;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the registered side-effect-safe running-agent count source, if one has been wired by the host process.
|
||||
*/
|
||||
export function getRunningAgentCountSource(): RunningAgentCountSource | undefined {
|
||||
return runningAgentCountSource;
|
||||
}
|
||||
|
||||
export interface RunningAgentCounts {
|
||||
currentlyActive: number;
|
||||
projectsActive: Record<string, number>;
|
||||
}
|
||||
|
||||
export function deriveRunningAgentCounts(perProject: Record<string, number>): RunningAgentCounts {
|
||||
const projectsActive: Record<string, number> = {};
|
||||
let currentlyActive = 0;
|
||||
|
||||
for (const [projectId, rawCount] of Object.entries(perProject)) {
|
||||
const count = Number.isFinite(rawCount) ? Math.max(0, Math.trunc(rawCount)) : 0;
|
||||
currentlyActive += count;
|
||||
if (count > 0) {
|
||||
projectsActive[projectId] = count;
|
||||
}
|
||||
}
|
||||
|
||||
return { currentlyActive, projectsActive };
|
||||
}
|
||||
@@ -24,6 +24,7 @@ const {
|
||||
mockGetProjectHealth,
|
||||
mockGetRecentActivity,
|
||||
mockGetGlobalConcurrencyState,
|
||||
mockGetLiveRunningAgentCounts,
|
||||
mockUpdateGlobalConcurrency,
|
||||
mockInit,
|
||||
mockClose,
|
||||
@@ -94,6 +95,10 @@ const {
|
||||
queuedCount: 0,
|
||||
projectsActive: { proj_test123: 2 },
|
||||
}),
|
||||
mockGetLiveRunningAgentCounts: vi.fn().mockResolvedValue({
|
||||
currentlyActive: 2,
|
||||
projectsActive: { proj_test123: 2 },
|
||||
}),
|
||||
mockUpdateGlobalConcurrency: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 10,
|
||||
currentlyActive: 2,
|
||||
@@ -158,6 +163,7 @@ vi.mock("@fusion/core", async () => {
|
||||
getProjectHealth: mockGetProjectHealth,
|
||||
getRecentActivity: mockGetRecentActivity,
|
||||
getGlobalConcurrencyState: mockGetGlobalConcurrencyState,
|
||||
getLiveRunningAgentCounts: mockGetLiveRunningAgentCounts,
|
||||
updateGlobalConcurrency: mockUpdateGlobalConcurrency,
|
||||
reconcileProjectStatuses: mockReconcileProjectStatuses,
|
||||
listNodes: mockListNodes,
|
||||
@@ -176,6 +182,7 @@ vi.mock("@fusion/core", async () => {
|
||||
// Mock project-store-resolver for multi-project health tests
|
||||
vi.mock("../project-store-resolver.js", () => ({
|
||||
getOrCreateProjectStore: mockGetOrCreateProjectStore,
|
||||
countRunningAgentsInRegisteredProjectStores: vi.fn().mockResolvedValue({}),
|
||||
invalidateAllGlobalSettingsCaches: vi.fn(),
|
||||
}));
|
||||
|
||||
@@ -1241,6 +1248,7 @@ describe("GET /api/global-concurrency route handler", () => {
|
||||
it("omits projects and reports zero when no tasks are in progress", async () => {
|
||||
const storeA = storeWithColumns(["todo", "in-review", "done", "archived"]);
|
||||
mockListProjects.mockResolvedValue([project("proj_a")]);
|
||||
mockGetLiveRunningAgentCounts.mockResolvedValue({ currentlyActive: 0, projectsActive: {} });
|
||||
mockGetOrCreateProjectStore.mockResolvedValue(storeA);
|
||||
|
||||
const app = await createApp(new MockStoreForRoutes());
|
||||
@@ -1253,7 +1261,9 @@ describe("GET /api/global-concurrency route handler", () => {
|
||||
queuedCount: 7,
|
||||
projectsActive: {},
|
||||
});
|
||||
expect(storeA.listTasks).toHaveBeenCalledWith({ slim: true });
|
||||
expect(mockGetLiveRunningAgentCounts).toHaveBeenCalledWith();
|
||||
expect(mockGetOrCreateProjectStore).not.toHaveBeenCalled();
|
||||
expect(storeA.listTasks).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
@@ -1300,6 +1310,7 @@ describe("GET /api/global-concurrency route handler", () => {
|
||||
const storesByProject = new Map(
|
||||
Object.entries(stores).map(([projectId, columns]) => [projectId, storeWithColumns(columns)]),
|
||||
);
|
||||
mockGetLiveRunningAgentCounts.mockResolvedValue({ currentlyActive: expectedTotal, projectsActive: expectedProjects });
|
||||
mockGetOrCreateProjectStore.mockImplementation(async (projectId: string) => storesByProject.get(projectId) ?? storeWithColumns([]));
|
||||
|
||||
const app = await createApp(new MockStoreForRoutes());
|
||||
@@ -1314,9 +1325,10 @@ describe("GET /api/global-concurrency route handler", () => {
|
||||
});
|
||||
expect((res.body as { currentlyActive: number }).currentlyActive).toBeGreaterThanOrEqual(expectedTotal);
|
||||
expect((res.body as { projectsActive: Record<string, number> }).projectsActive).not.toHaveProperty("stale_project");
|
||||
for (const [projectId, mockStore] of storesByProject) {
|
||||
expect(mockGetOrCreateProjectStore).toHaveBeenCalledWith(projectId);
|
||||
expect(mockStore.listTasks).toHaveBeenCalledWith({ slim: true });
|
||||
expect(mockGetLiveRunningAgentCounts).toHaveBeenCalledWith();
|
||||
expect(mockGetOrCreateProjectStore).not.toHaveBeenCalled();
|
||||
for (const [, mockStore] of storesByProject) {
|
||||
expect(mockStore.listTasks).not.toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -12,6 +12,7 @@ import { EventEmitter } from "node:events";
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import {
|
||||
getOrCreateProjectStore,
|
||||
countRunningAgentsInRegisteredProjectStores,
|
||||
evictProjectStore,
|
||||
evictAllProjectStores,
|
||||
listRegisteredProjectStores,
|
||||
@@ -353,3 +354,73 @@ describe("project-store-resolver", () => {
|
||||
expect(sseMessages[2]).toContain('"to":"todo"');
|
||||
});
|
||||
});
|
||||
|
||||
describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
beforeEach(() => {
|
||||
evictAllProjectStores();
|
||||
createdStores.length = 0;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
evictAllProjectStores();
|
||||
createdStores.length = 0;
|
||||
});
|
||||
|
||||
function installTaskList(store: TaskStore, columns: string[]) {
|
||||
const tasks = columns.map((column, index) => ({ id: `FN-${index + 1}`, column }));
|
||||
const listTasks = vi.fn().mockImplementation(async (options?: { column?: string }) => (
|
||||
options?.column ? tasks.filter((task) => task.column === options.column) : tasks
|
||||
));
|
||||
(store as TaskStore & { listTasks: typeof listTasks }).listTasks = listTasks;
|
||||
return listTasks;
|
||||
}
|
||||
|
||||
it("counts in-progress tasks from already-open stores without opening unopened projects", async () => {
|
||||
const openStore = await getOrCreateProjectStore("proj_open");
|
||||
const listTasks = installTaskList(openStore, ["todo", "in-progress", "in-progress", "done"]);
|
||||
const openEntry = createdStores.find((entry) => entry.projectId === "proj_open");
|
||||
expect(openEntry).toBeDefined();
|
||||
vi.clearAllMocks();
|
||||
|
||||
const counts = await countRunningAgentsInRegisteredProjectStores(["proj_open", "proj_unopened"]);
|
||||
|
||||
expect(counts).toEqual({ proj_open: 2 });
|
||||
expect(listTasks).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
expect(createdStores).toHaveLength(1);
|
||||
expect(openEntry?.watchMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("excludes cached stores that are not requested and reports zero for open stores with no running tasks", async () => {
|
||||
const zeroStore = await getOrCreateProjectStore("proj_zero");
|
||||
const ignoredStore = await getOrCreateProjectStore("proj_ignored");
|
||||
const zeroListTasks = installTaskList(zeroStore, ["todo", "in-review", "done"]);
|
||||
const ignoredListTasks = installTaskList(ignoredStore, ["in-progress", "in-progress"]);
|
||||
const zeroEntry = createdStores.find((entry) => entry.projectId === "proj_zero");
|
||||
const ignoredEntry = createdStores.find((entry) => entry.projectId === "proj_ignored");
|
||||
vi.clearAllMocks();
|
||||
|
||||
const counts = await countRunningAgentsInRegisteredProjectStores(["proj_zero", "proj_unopened"]);
|
||||
|
||||
expect(counts).toEqual({ proj_zero: 0 });
|
||||
expect(zeroListTasks).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
expect(ignoredListTasks).not.toHaveBeenCalled();
|
||||
expect(createdStores).toHaveLength(2);
|
||||
expect(zeroEntry?.watchMock).not.toHaveBeenCalled();
|
||||
expect(ignoredEntry?.watchMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns per-project counts for multiple already-open stores", async () => {
|
||||
const storeA = await getOrCreateProjectStore("proj_a");
|
||||
const storeB = await getOrCreateProjectStore("proj_b");
|
||||
const listTasksA = installTaskList(storeA, ["in-progress", "todo"]);
|
||||
const listTasksB = installTaskList(storeB, ["todo", "in-progress", "in-progress"]);
|
||||
vi.clearAllMocks();
|
||||
|
||||
const counts = await countRunningAgentsInRegisteredProjectStores(["proj_a", "proj_b"]);
|
||||
|
||||
expect(counts).toEqual({ proj_a: 1, proj_b: 2 });
|
||||
expect(listTasksA).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
expect(listTasksB).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
expect(createdStores).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -12,7 +12,7 @@ import express from "express";
|
||||
import { createServer, setupTerminalWebSocket } from "../server.js";
|
||||
import { toSessionTag } from "../terminal-websocket-diagnostics.js";
|
||||
import { RATE_LIMITS } from "../rate-limit.js";
|
||||
import { Database, TaskStore } from "@fusion/core";
|
||||
import { Database, TaskStore, getRunningAgentCountSource, setRunningAgentCountSource, type CentralCore } from "@fusion/core";
|
||||
import { get as performGet, request as performRequest } from "../test-request.js";
|
||||
|
||||
// Mock terminal-service before any imports that use it
|
||||
@@ -190,12 +190,55 @@ async function flushStartupCleanupTasks(): Promise<void> {
|
||||
}
|
||||
|
||||
describe("createServer options", () => {
|
||||
afterEach(() => {
|
||||
setRunningAgentCountSource(undefined);
|
||||
});
|
||||
|
||||
it("round-trips hybridExecutor on app locals", () => {
|
||||
const store = createMockStore();
|
||||
const hybridExecutor = { initialize: vi.fn(), shutdown: vi.fn() } as unknown as import("@fusion/engine").HybridExecutor;
|
||||
const app = createServer(store, { hybridExecutor });
|
||||
expect(app.locals.hybridExecutor).toBe(hybridExecutor);
|
||||
});
|
||||
|
||||
it("registers live counts for the already-open default store by central project id", async () => {
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([{ id: "FN-1" }, { id: "FN-2" }]),
|
||||
});
|
||||
const centralCore = {
|
||||
getDefaultProjectId: vi.fn().mockResolvedValue("proj_default"),
|
||||
};
|
||||
|
||||
createServer(store, { centralCore: centralCore as unknown as CentralCore });
|
||||
const source = getRunningAgentCountSource();
|
||||
|
||||
expect(source).toBeDefined();
|
||||
if (!source) throw new Error("expected running-agent count source");
|
||||
await expect(source(["proj_default", "proj_unopened"])).resolves.toEqual({ proj_default: 2 });
|
||||
expect(store.listTasks).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
});
|
||||
|
||||
it("registers live counts for already-open engine-manager stores without starting engines", async () => {
|
||||
const store = createMockStore();
|
||||
const engineStore = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([{ id: "FN-3" }]),
|
||||
});
|
||||
const getEngine = vi.fn((projectId: string) => projectId === "proj_engine"
|
||||
? { getTaskStore: vi.fn(() => engineStore) }
|
||||
: undefined);
|
||||
const engineManager = { getEngine };
|
||||
|
||||
createServer(store, { engineManager: engineManager as unknown as import("@fusion/engine").ProjectEngineManager });
|
||||
const source = getRunningAgentCountSource();
|
||||
|
||||
expect(source).toBeDefined();
|
||||
if (!source) throw new Error("expected running-agent count source");
|
||||
await expect(source(["proj_engine", "proj_unopened"])).resolves.toEqual({ proj_engine: 1 });
|
||||
expect(getEngine).toHaveBeenCalledWith("proj_engine");
|
||||
expect(getEngine).toHaveBeenCalledWith("proj_unopened");
|
||||
expect(engineStore.listTasks).toHaveBeenCalledWith({ column: "in-progress", slim: true });
|
||||
expect(store.listTasks).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("createServer AI session startup cleanup diagnostics", () => {
|
||||
|
||||
@@ -158,6 +158,30 @@ export function listRegisteredProjectStores(): Array<{ projectId: string; store:
|
||||
return Array.from(storeCache.entries(), ([projectId, store]) => ({ projectId, store }));
|
||||
}
|
||||
|
||||
export async function countRunningAgentsInStore(store: TaskStore): Promise<number> {
|
||||
const tasks = await store.listTasks({ column: "in-progress", slim: true });
|
||||
return tasks.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:GlobalConcurrencyControls 2026-06-26-17:22:
|
||||
* The dashboard live-count source is restricted to already-open project stores so global concurrency reads never open a project, start a watcher, or start an engine/runtime just to answer currently-active counts.
|
||||
*/
|
||||
export async function countRunningAgentsInRegisteredProjectStores(projectIds: readonly string[]): Promise<Record<string, number>> {
|
||||
const requestedProjectIds = new Set(projectIds);
|
||||
const counts: Record<string, number> = {};
|
||||
|
||||
await Promise.all(listRegisteredProjectStores().map(async ({ projectId, store }) => {
|
||||
if (!requestedProjectIds.has(projectId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
counts[projectId] = await countRunningAgentsInStore(store);
|
||||
}));
|
||||
|
||||
return counts;
|
||||
}
|
||||
|
||||
export function onProjectStoreRegistered(listener: (projectId: string, store: TaskStore) => void): () => void {
|
||||
projectRegisteredListeners.add(listener);
|
||||
return () => {
|
||||
|
||||
@@ -57,7 +57,6 @@ import {
|
||||
} from "./api-error.js";
|
||||
import { createPluginRouter, resolvePluginManifest } from "./plugin-routes.js";
|
||||
import { fetchFromRemoteNode } from "./routes/register-settings-sync-helpers.js";
|
||||
import { getOrCreateProjectStore } from "./project-store-resolver.js";
|
||||
import { hermesRuntimeMetadata } from "@fusion-plugin-examples/hermes-runtime";
|
||||
import { openclawRuntimeMetadata } from "@fusion-plugin-examples/openclaw-runtime";
|
||||
|
||||
@@ -4638,30 +4637,16 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
||||
if (shouldClose || (typeof central.isInitialized === "function" && !central.isInitialized())) await central.init();
|
||||
|
||||
const state = await central.getGlobalConcurrencyState();
|
||||
const projects = await central.listProjects();
|
||||
const projectCounts = await Promise.all(projects.map(async (project) => {
|
||||
const projectStore = await getOrCreateProjectStore(project.id);
|
||||
const tasks = await projectStore.listTasks({ slim: true });
|
||||
return [project.id, tasks.filter((task) => task.column === "in-progress").length] as const;
|
||||
}));
|
||||
|
||||
const projectsActive: Record<string, number> = {};
|
||||
let currentlyActive = 0;
|
||||
for (const [projectId, activeCount] of projectCounts) {
|
||||
currentlyActive += activeCount;
|
||||
if (activeCount > 0) {
|
||||
projectsActive[projectId] = activeCount;
|
||||
}
|
||||
}
|
||||
const liveCounts = await central.getLiveRunningAgentCounts();
|
||||
|
||||
/*
|
||||
FNXC:GlobalConcurrencyControls 2026-06-26-12:00:
|
||||
The footer EngineControlMenu and Command Center Concurrency card need running-agent counts from live task state. Slot bookkeeping (`globalConcurrency.currentlyActive`) and polled project health are not synced in the default in-process runtime, so derive read-only currentlyActive/projectsActive from authoritative `in-progress` task columns without mutating the slot limiter or editable cap.
|
||||
FNXC:GlobalConcurrencyControls 2026-06-26-17:22:
|
||||
The published global-concurrency route reads currentlyActive/projectsActive through CentralCore's live seam while preserving globalMaxConcurrent/queuedCount from slot bookkeeping. The dashboard-registered source only inspects already-open project stores, so this read stays side-effect-safe and never opens watchers or starts project runtimes.
|
||||
*/
|
||||
const liveState = {
|
||||
...state,
|
||||
currentlyActive,
|
||||
projectsActive,
|
||||
currentlyActive: liveCounts.currentlyActive,
|
||||
projectsActive: liveCounts.projectsActive,
|
||||
};
|
||||
|
||||
if (shouldClose) await central.close();
|
||||
|
||||
@@ -16,13 +16,19 @@ import type {
|
||||
AgentLogEntry,
|
||||
TaskIdIntegrityReport,
|
||||
} from "@fusion/core";
|
||||
import { AgentStore, ChatStore } from "@fusion/core";
|
||||
import { AgentStore, ChatStore, setRunningAgentCountSource } from "@fusion/core";
|
||||
import type { AuthStorageLike, ModelRegistryLike } from "./routes.js";
|
||||
import { createApiRoutes } from "./routes.js";
|
||||
import { createSSE, disconnectSSEClient, markSSEClientAlive } from "./sse.js";
|
||||
import { rateLimit, RATE_LIMITS } from "./rate-limit.js";
|
||||
import { ApiError, sendErrorResponse } from "./api-error.js";
|
||||
import { getOrCreateProjectStore, evictAllProjectStores, setOnProjectFirstCreated } from "./project-store-resolver.js";
|
||||
import {
|
||||
countRunningAgentsInRegisteredProjectStores,
|
||||
countRunningAgentsInStore,
|
||||
getOrCreateProjectStore,
|
||||
evictAllProjectStores,
|
||||
setOnProjectFirstCreated,
|
||||
} from "./project-store-resolver.js";
|
||||
import { getOrCreateScopedChatStore } from "./chat-project-services.js";
|
||||
import { getTerminalService, STALE_SESSION_THRESHOLD_MS } from "./terminal-service.js";
|
||||
import { WebSocketServer, type WebSocket } from "ws";
|
||||
@@ -826,6 +832,37 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
|
||||
if (options?.onProjectFirstAccessed) {
|
||||
setOnProjectFirstCreated(options.onProjectFirstAccessed);
|
||||
}
|
||||
/*
|
||||
FNXC:GlobalConcurrencyControls 2026-06-26-17:22:
|
||||
Dashboard bootstrap wires CentralCore's live running-agent source to the already-open project-store cache. This avoids duplicating route-local counting while preserving the read-path rule that unopened projects are not initialized just to compute global concurrency counts.
|
||||
|
||||
FNXC:GlobalConcurrencyControls 2026-06-26-23:41:
|
||||
The default in-process TaskStore is already open but is intentionally not part of the secondary project-store cache. Include it by central default project id, and include any engine-manager stores already resident in memory, so live reads cover every already-open store without calling getOrCreateProjectStore(), watch(), or runtime startup paths.
|
||||
*/
|
||||
setRunningAgentCountSource(async (projectIds) => {
|
||||
const requestedProjectIds = new Set(projectIds);
|
||||
const counts = await countRunningAgentsInRegisteredProjectStores(projectIds);
|
||||
|
||||
if (options?.engineManager) {
|
||||
await Promise.all(projectIds.map(async (projectId) => {
|
||||
if (counts[projectId] !== undefined) {
|
||||
return;
|
||||
}
|
||||
const engine = options.engineManager?.getEngine(projectId);
|
||||
if (!engine) {
|
||||
return;
|
||||
}
|
||||
counts[projectId] = await countRunningAgentsInStore(engine.getTaskStore());
|
||||
}));
|
||||
}
|
||||
|
||||
const defaultProjectId = await options?.centralCore?.getDefaultProjectId?.();
|
||||
if (defaultProjectId && requestedProjectIds.has(defaultProjectId)) {
|
||||
counts[defaultProjectId] = await countRunningAgentsInStore(store);
|
||||
}
|
||||
|
||||
return counts;
|
||||
});
|
||||
|
||||
const app = express();
|
||||
app.locals.hybridExecutor = options?.hybridExecutor;
|
||||
@@ -2444,6 +2481,7 @@ export function setupBadgeWebSocket(
|
||||
wss.close();
|
||||
// Clean up cached project-scoped stores (stop watchers, close DB connections)
|
||||
evictAllProjectStores();
|
||||
setRunningAgentCountSource(undefined);
|
||||
dashboardApp.terminalWsServer = null;
|
||||
dashboardApp.badgeWsServer = null;
|
||||
dashboardApp.badgeWsManager = null;
|
||||
|
||||
Reference in New Issue
Block a user