diff --git a/.changeset/fn-7082-live-running-agent-count-seam.md b/.changeset/fn-7082-live-running-agent-count-seam.md new file mode 100644 index 0000000000..b3ff4df522 --- /dev/null +++ b/.changeset/fn-7082-live-running-agent-count-seam.md @@ -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. diff --git a/docs/multi-project.md b/docs/multi-project.md index 9f31b5a5c7..29be655d24 100644 --- a/docs/multi-project.md +++ b/docs/multi-project.md @@ -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 diff --git a/packages/core/src/__tests__/central-core.test.ts b/packages/core/src/__tests__/central-core.test.ts index 32e18aa812..df549523df 100644 --- a/packages/core/src/__tests__/central-core.test.ts +++ b/packages/core/src/__tests__/central-core.test.ts @@ -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", () => { diff --git a/packages/core/src/central-core.ts b/packages/core/src/central-core.ts index 98ee6a678d..309b20a426 100644 --- a/packages/core/src/central-core.ts +++ b/packages/core/src/central-core.ts @@ -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 { }; } + /** + * 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 { + 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. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 01f3a7ebac..5ba4a5cc44 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -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, diff --git a/packages/core/src/live-agent-count.ts b/packages/core/src/live-agent-count.ts new file mode 100644 index 0000000000..4ea062f7d5 --- /dev/null +++ b/packages/core/src/live-agent-count.ts @@ -0,0 +1,38 @@ +export type RunningAgentCountSource = (projectIds: readonly string[]) => Promise> | Record; + +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; +} + +export function deriveRunningAgentCounts(perProject: Record): RunningAgentCounts { + const projectsActive: Record = {}; + 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 }; +} diff --git a/packages/dashboard/src/__tests__/project-routes.test.ts b/packages/dashboard/src/__tests__/project-routes.test.ts index 0ff3fc016f..939dfcc3ce 100644 --- a/packages/dashboard/src/__tests__/project-routes.test.ts +++ b/packages/dashboard/src/__tests__/project-routes.test.ts @@ -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 }).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(); } }); }); diff --git a/packages/dashboard/src/__tests__/project-store-resolver.test.ts b/packages/dashboard/src/__tests__/project-store-resolver.test.ts index 087b5b0df9..af86aca4c9 100644 --- a/packages/dashboard/src/__tests__/project-store-resolver.test.ts +++ b/packages/dashboard/src/__tests__/project-store-resolver.test.ts @@ -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); + }); +}); diff --git a/packages/dashboard/src/__tests__/server.test.ts b/packages/dashboard/src/__tests__/server.test.ts index 52312f0ca6..8101b7c0a5 100644 --- a/packages/dashboard/src/__tests__/server.test.ts +++ b/packages/dashboard/src/__tests__/server.test.ts @@ -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 { } 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", () => { diff --git a/packages/dashboard/src/project-store-resolver.ts b/packages/dashboard/src/project-store-resolver.ts index e0a2490c28..cdc0f83ac0 100644 --- a/packages/dashboard/src/project-store-resolver.ts +++ b/packages/dashboard/src/project-store-resolver.ts @@ -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 { + 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> { + const requestedProjectIds = new Set(projectIds); + const counts: Record = {}; + + 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 () => { diff --git a/packages/dashboard/src/routes.ts b/packages/dashboard/src/routes.ts index ffa8bfab05..6e50a6eb66 100644 --- a/packages/dashboard/src/routes.ts +++ b/packages/dashboard/src/routes.ts @@ -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 = {}; - 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(); diff --git a/packages/dashboard/src/server.ts b/packages/dashboard/src/server.ts index 5a931558cf..cd9b8f242a 100644 --- a/packages/dashboard/src/server.ts +++ b/packages/dashboard/src/server.ts @@ -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;