FN-7205: prefer live concurrency counts

Prefer live runtime stores for concurrency counters so dashboard totals match active agent enforcement.

- Prefer engine-manager task stores before default fallback stores when sourcing live running-agent counts.
- Count running agents from slim task lists across active lifecycle states instead of trusting stale in-progress queries.
- Add regressions for count normalization, default-project store precedence, and scoped semaphore limit changes.
- Add a patch changeset for the published Fusion CLI package.

Files changed:
 .changeset/fn-7205-concurrency-counter-enforcement.md     |  7 ++++
 packages/core/src/__tests__/live-agent-count.test.ts      | 18 ++++++++-
 packages/dashboard/src/__tests__/server.test.ts           | 44 ++++++++++++++++++++--
 packages/dashboard/src/server.ts                          |  8 ++--
 packages/engine/src/__tests__/concurrency.test.ts         | 38 +++++++++++++++++++
 5 files changed, 106 insertions(+), 9 deletions(-)

Fusion-Task-Id: FN-7205

Fusion-Task-Lineage: 957d6b2a-b7c4-4fd1-b83b-62f5cd51ebda

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>

.changeset/fn-7205-concurrency-counter-enforcement.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Concurrency panels now prefer live engine counts so running-agent totals stay accurate.
+category: fix
+dev: Prefer engine-manager task stores over stale registered/default fallback stores in the dashboard live-count source; add regressions for count normalization and scoped semaphore live-limit behavior.

packages/core/src/__tests__/live-agent-count.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { countRunningAgentTasks, isRunningAgentTask } from "../live-agent-count.js";
+import { countRunningAgentTasks, deriveRunningAgentCounts, isRunningAgentTask } from "../live-agent-count.js";
 import type { Task } from "../types.js";
 
 function task(overrides: Pick<Task, "column"> & Partial<Pick<Task, "status" | "paused">>): Pick<Task, "column" | "status" | "paused"> {
@@ -44,4 +44,20 @@ describe("live agent count predicates", () => {
       task({ column: "archived" }),
     ])).toBe(7);
   });
+
+  it("normalizes display counts for zero, one, multi-project, unopened, and oversubscribed states", () => {
+    expect(deriveRunningAgentCounts({})).toEqual({ currentlyActive: 0, projectsActive: {} });
+    expect(deriveRunningAgentCounts({ proj_zero: 0, proj_one: 1 })).toEqual({
+      currentlyActive: 1,
+      projectsActive: { proj_one: 1 },
+    });
+    expect(deriveRunningAgentCounts({ proj_a: 2, proj_b: 4, proj_unopened: 0 })).toEqual({
+      currentlyActive: 6,
+      projectsActive: { proj_a: 2, proj_b: 4 },
+    });
+    expect(deriveRunningAgentCounts({ proj_over_limit: 12, proj_negative: -3, proj_nan: Number.NaN })).toEqual({
+      currentlyActive: 12,
+      projectsActive: { proj_over_limit: 12 },
+    });
+  });
 });

packages/dashboard/src/__tests__/server.test.ts

@@ -203,7 +203,10 @@ describe("createServer options", () => {
 
   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" }]),
+      listTasks: vi.fn().mockResolvedValue([
+        { id: "FN-1", column: "in-progress" },
+        { id: "FN-2", column: "triage", status: "planning", paused: false },
+      ]),
     });
     const centralCore = {
       getDefaultProjectId: vi.fn().mockResolvedValue("proj_default"),
@@ -215,13 +218,13 @@ describe("createServer options", () => {
     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 });
+    expect(store.listTasks).toHaveBeenCalledWith({ 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" }]),
+      listTasks: vi.fn().mockResolvedValue([{ id: "FN-3", column: "in-review", status: "merging", paused: false }]),
     });
     const getEngine = vi.fn((projectId: string) => projectId === "proj_engine"
       ? { getTaskStore: vi.fn(() => engineStore) }
@@ -236,7 +239,40 @@ describe("createServer options", () => {
     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(engineStore.listTasks).toHaveBeenCalledWith({ slim: true });
+    expect(store.listTasks).not.toHaveBeenCalled();
+  });
+
+  it("prefers the live engine-manager store over the default fallback for the default project", async () => {
+    const store = createMockStore({
+      listTasks: vi.fn().mockResolvedValue([]),
+    });
+    const engineStore = createMockStore({
+      listTasks: vi.fn().mockResolvedValue([
+        { id: "FN-4", column: "in-progress" },
+        { id: "FN-5", column: "in-review", status: "reviewing", paused: false },
+      ]),
+    });
+    const getEngine = vi.fn((projectId: string) => projectId === "proj_default"
+      ? { getTaskStore: vi.fn(() => engineStore) }
+      : undefined);
+    const engineManager = { getEngine };
+    const centralCore = {
+      getDefaultProjectId: vi.fn().mockResolvedValue("proj_default"),
+    };
+
+    createServer(store, {
+      centralCore: centralCore as unknown as CentralCore,
+      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_default", "proj_unopened"])).resolves.toEqual({ proj_default: 2 });
+    expect(getEngine).toHaveBeenCalledWith("proj_default");
+    expect(getEngine).toHaveBeenCalledWith("proj_unopened");
+    expect(engineStore.listTasks).toHaveBeenCalledWith({ slim: true });
     expect(store.listTasks).not.toHaveBeenCalled();
   });
 });

packages/dashboard/src/server.ts

@@ -838,6 +838,9 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
 
   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.
+
+  FNXC:GlobalConcurrencyControls 2026-06-28-16:48:
+  FN-7205 requires the footer and Command Center counters to reflect actual running top-level agents from the live runtime store. Prefer engine-manager stores over registered/default fallback stores, and only use the default store when no live engine/registered source has supplied that project, so stale bootstrap stores cannot overwrite the semaphore-facing runtime count after a slider or engine lifecycle change.
   */
   setRunningAgentCountSource(async (projectIds) => {
     const requestedProjectIds = new Set(projectIds);
@@ -845,9 +848,6 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
 
     if (options?.engineManager) {
       await Promise.all(projectIds.map(async (projectId) => {
-        if (counts[projectId] !== undefined) {
-          return;
-        }
         const engine = options.engineManager?.getEngine(projectId);
         if (!engine) {
           return;
@@ -857,7 +857,7 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
     }
 
     const defaultProjectId = await options?.centralCore?.getDefaultProjectId?.();
-    if (defaultProjectId && requestedProjectIds.has(defaultProjectId)) {
+    if (defaultProjectId && requestedProjectIds.has(defaultProjectId) && counts[defaultProjectId] === undefined) {
       counts[defaultProjectId] = await countRunningAgentsInStore(store);
     }
 

packages/engine/src/__tests__/concurrency.test.ts

@@ -113,6 +113,44 @@ describe("ScopedAgentSemaphore", () => {
     expect(shared.activeCount).toBe(0);
   });
 
+  it("honors live global-limit changes across scoped project semaphores on the next acquire", async () => {
+    let globalLimit = 2;
+    const shared = new AgentSemaphore(() => globalLimit);
+    const projectA = new ScopedAgentSemaphore(shared);
+    const projectB = new ScopedAgentSemaphore(shared);
+
+    await projectA.acquire(PRIORITY_EXECUTE);
+    await projectB.acquire(PRIORITY_MERGE);
+    expect(shared.snapshot()).toEqual({ activeCount: 2, waitingCount: 0, availableCount: 0, limit: 2 });
+
+    globalLimit = 1;
+    let acquired = false;
+    const waiter = projectA.acquire(PRIORITY_EXECUTE).then(() => {
+      acquired = true;
+    });
+    await Promise.resolve();
+
+    expect(acquired).toBe(false);
+    expect(shared.snapshot()).toEqual({ activeCount: 2, waitingCount: 1, availableCount: 0, limit: 1 });
+
+    projectA.release();
+    await Promise.resolve();
+    expect(acquired).toBe(false);
+    expect(shared.snapshot()).toEqual({ activeCount: 1, waitingCount: 1, availableCount: 0, limit: 1 });
+
+    globalLimit = 2;
+    projectB.release();
+    await waiter;
+
+    expect(acquired).toBe(true);
+    expect(projectA.heldCount).toBe(1);
+    expect(projectB.heldCount).toBe(0);
+    expect(shared.snapshot()).toEqual({ activeCount: 1, waitingCount: 0, availableCount: 1, limit: 2 });
+
+    projectA.release();
+    expect(shared.activeCount).toBe(0);
+  });
+
   it("reconciles only this scope's slots when another project still holds global capacity", async () => {
     const shared = new AgentSemaphore(3);
     const idleProject = new ScopedAgentSemaphore(shared);