FN-6423: fix scheduler capacity accounting

Correct scheduler dispatch diagnostics so capacity decisions use consistent non-negative slot counts.

- Clamp excess semaphore releases at zero and warn once when a slot is returned without an active holder.
- Recompute dispatch capacity at each queue decision, including tasks started earlier in the same scheduler tick.
- Update scheduler and semaphore tests for true binding gates, non-negative diagnostics, and workflow-step env stability.
- Add a patch changeset for the scheduler capacity fix.

Files changed:
 .changeset/fn-6423-scheduler-capacity.md           |   5 +
 packages/engine/src/__tests__/concurrency.test.ts  |  32 ++++++
 .../src/__tests__/executor-step-session.test.ts    |  10 +-
 packages/engine/src/__tests__/scheduler.test.ts    | 122 ++++++++++++++++++++-
 packages/engine/src/concurrency.ts                 |  29 ++++-
 packages/engine/src/scheduler.ts                   | 117 ++++++++++----------
 6 files changed, 248 insertions(+), 67 deletions(-)

Fusion-Task-Id: FN-6423

Fusion-Task-Lineage: a6b2e668-a822-46e9-9cd8-ac267fbde804

.changeset/fn-6423-scheduler-capacity.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Fix scheduler concurrency diagnostics and semaphore slot accounting so queued tasks are not held behind contradictory or negative capacity readings.

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

@@ -19,6 +19,38 @@ describe("AgentSemaphore", () => {
     expect(sem.availableCount).toBe(2);
   });
 
+  it("FN-6423: clamps excess slot returns without breaking future acquires", async () => {
+    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    try {
+      const sem = new AgentSemaphore(2);
+      await sem.acquire();
+      sem.release();
+      sem.release();
+      sem.release();
+
+      expect(sem.activeCount).toBe(0);
+      expect(sem.availableCount).toBe(2);
+      expect(sem.snapshot()).toEqual({
+        activeCount: 0,
+        waitingCount: 0,
+        availableCount: 2,
+        limit: 2,
+      });
+      expect(warnSpy).toHaveBeenCalledTimes(1);
+      expect(String(warnSpy.mock.calls[0]?.[0])).toContain("AgentSemaphore excess slot return ignored from release");
+
+      expect(sem.tryAcquire()).toBe(true);
+      expect(sem.activeCount).toBe(1);
+      sem.release();
+      await sem.acquire();
+      expect(sem.activeCount).toBe(1);
+      sem.release();
+      expect(sem.activeCount).toBe(0);
+    } finally {
+      warnSpy.mockRestore();
+    }
+  });
+
   it("queues waiters when at capacity and unblocks FIFO", async () => {
     const sem = new AgentSemaphore(1);
     await sem.acquire(); // slot taken

packages/engine/src/__tests__/executor-step-session.test.ts

@@ -475,10 +475,12 @@ describe("Workflow Steps Execution", () => {
     expect(secondCall[0].tools).toBe("readonly");
     expect(secondCall[0].systemPrompt).toContain("Docs Review");
     expect(secondCall[0].systemPrompt).toContain("Review all docs and verify they are complete.");
-    expect(secondCall[0].taskEnv).toEqual({
+    const withoutWorkflowStep = (env: Record<string, string | undefined>) => {
-      ...mockedCreateFnAgent.mock.calls[0][0].taskEnv,
+      const { FUSION_WORKFLOW_STEP: _workflowStep, ...stableEnv } = env;
-      FUSION_WORKFLOW_STEP: "1",
+      return stableEnv;
-    });
+    };
+    expect(secondCall[0].taskEnv.FUSION_WORKFLOW_STEP).toBe("1");
+    expect(withoutWorkflowStep(secondCall[0].taskEnv)).toEqual(withoutWorkflowStep(mockedCreateFnAgent.mock.calls[0][0].taskEnv));
 
     // Task should move to in-review
     expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");

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

@@ -1815,7 +1815,7 @@ describe("Scheduler", () => {
 
       const call = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.find((c: unknown[]) => c[0] === "FN-C");
       expect(String(call?.[1])).toContain("gate=maxConcurrent");
-      expect(String(call?.[1])).toContain("maxConcurrent used=1/2");
+      expect(String(call?.[1])).toContain("maxConcurrent used=2/2");
       expect(String(call?.[1])).toContain("holders: FN-A");
     });
 
@@ -1840,7 +1840,7 @@ describe("Scheduler", () => {
 
       const call = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.find((c: unknown[]) => c[0] === "FN-D");
       expect(String(call?.[1])).toContain("gate=maxWorktrees");
-      expect(String(call?.[1])).toContain("maxWorktrees used=2/3");
+      expect(String(call?.[1])).toContain("maxWorktrees used=3/3");
       expect(String(call?.[1])).toContain("holders: FN-A, FN-B");
     });
 
@@ -1864,10 +1864,124 @@ describe("Scheduler", () => {
 
       const call = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.find((c: unknown[]) => c[0] === "FN-B");
       expect(String(call?.[1])).toContain("gate=semaphore");
-      expect(String(call?.[1])).toContain("semaphore used=0/1");
+      expect(String(call?.[1])).toContain("semaphore used=1/1");
       expect(String(call?.[1])).toContain("semaphore slots may include triage/merge agents outside in-progress");
     });
 
+    it.each([false, true])("FN-6423: logs queue-point capacity without negative semaphore usage (workflowColumns=%s)", async (workflowColumns) => {
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+
+      const semaphore = new AgentSemaphore(3);
+      (semaphore as any)._active = -9;
+      const tasks = [
+        createMockTask({ id: "FN-6412", column: "in-progress" }),
+        createMockTask({ id: "FN-B", column: "todo", dependencies: [] }),
+        createMockTask({ id: "FN-C", column: "todo", dependencies: [] }),
+        createMockTask({ id: "FN-D", column: "todo", dependencies: [] }),
+      ];
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue(tasks),
+        getSettings: vi.fn().mockResolvedValue({
+          maxConcurrent: 15,
+          maxWorktrees: 3,
+          experimentalFeatures: { workflowColumns },
+        }),
+      });
+
+      const scheduler = new Scheduler(store, { semaphore });
+      (scheduler as any).running = true;
+      await scheduler.schedule();
+
+      expect(store.moveTask).toHaveBeenCalledWith(
+        "FN-B",
+        "in-progress",
+        expect.objectContaining({ moveSource: "scheduler" }),
+      );
+      expect(store.moveTask).toHaveBeenCalledWith(
+        "FN-C",
+        "in-progress",
+        expect.objectContaining({ moveSource: "scheduler" }),
+      );
+      const call = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.find((c: unknown[]) => c[0] === "FN-D");
+      const reason = String(call?.[1]);
+      expect(reason).toContain("queued — concurrency limit reached");
+      expect(reason).not.toMatch(/semaphore used=-/);
+      expect(reason).not.toContain("maxWorktrees used=1/3");
+      expect(reason).toContain("maxWorktrees used=3/3");
+      expect(reason).toContain("semaphore used=3/3");
+      const gateLabel = reason.match(/gate=([^;]+)/)?.[1] ?? "";
+      for (const gate of gateLabel.split(", ").filter(Boolean)) {
+        const usedLimit = reason.match(new RegExp(`${gate} used=(\\d+)/(\\d+)`));
+        expect(usedLimit, `${gate} must have used/limit details`).not.toBeNull();
+        expect(Number(usedLimit?.[1])).toBeGreaterThanOrEqual(Number(usedLimit?.[2]));
+      }
+    });
+
+    it("FN-6423: dispatches ready tasks while maxWorktrees still has slack", async () => {
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+
+      const tasks = [
+        createMockTask({ id: "FN-A", column: "in-progress" }),
+        createMockTask({ id: "FN-B", column: "todo", dependencies: [] }),
+      ];
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue(tasks),
+        getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 15, maxWorktrees: 3 }),
+      });
+
+      const scheduler = new Scheduler(store);
+      (scheduler as any).running = true;
+      await scheduler.schedule();
+
+      expect(store.moveTask).toHaveBeenCalledWith(
+        "FN-B",
+        "in-progress",
+        expect.objectContaining({ moveSource: "scheduler" }),
+      );
+      const concurrencyReasonCalls = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.filter(
+        (call: unknown[]) => call[0] === "FN-B" && String(call[1]).includes("queued — concurrency limit reached"),
+      );
+      expect(concurrencyReasonCalls).toHaveLength(0);
+    });
+
+    it("FN-6423: preserves legitimate maxWorktrees queueing at the true limit", async () => {
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+
+      const tasks = [
+        createMockTask({ id: "FN-A", column: "in-progress" }),
+        createMockTask({ id: "FN-B", column: "todo", dependencies: [] }),
+        createMockTask({ id: "FN-C", column: "todo", dependencies: [] }),
+      ];
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue(tasks),
+        getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 15, maxWorktrees: 2 }),
+      });
+
+      const scheduler = new Scheduler(store);
+      (scheduler as any).running = true;
+      await scheduler.schedule();
+
+      expect(store.moveTask).toHaveBeenCalledWith(
+        "FN-B",
+        "in-progress",
+        expect.objectContaining({ moveSource: "scheduler" }),
+      );
+      const call = (store.logEntry as ReturnType<typeof vi.fn>).mock.calls.find((c: unknown[]) => c[0] === "FN-C");
+      const reason = String(call?.[1]);
+      expect(reason).toContain("gate=maxWorktrees");
+      expect(reason).toContain("maxWorktrees used=2/2");
+      expect(formatConcurrencyLimitMemoKey({
+        available: 0,
+        bindingGates: ["maxWorktrees"],
+        maxConcurrentGate: { used: 2, limit: 15, slack: 13 },
+        maxWorktreesGate: { used: 2, limit: 2, slack: 0 },
+        holders: { maxConcurrent: ["FN-A"], maxWorktrees: ["FN-A"] },
+      })).toBe("queued-concurrency:maxWorktrees");
+    });
+
     it("recovers an idle leaked semaphore slot before dispatching", async () => {
       vi.mocked(existsSync).mockReturnValue(true);
       vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
@@ -2005,7 +2119,7 @@ describe("Scheduler", () => {
         (call: unknown[]) => call[0] === "FN-C" && String(call[1]).includes("queued — concurrency limit reached"),
       );
       expect(concurrencyReasonCalls).toHaveLength(1);
-      expect(String(concurrencyReasonCalls[0]?.[1])).toContain("semaphore used=1/2");
+      expect(String(concurrencyReasonCalls[0]?.[1])).toContain("semaphore used=2/2");
     });
 
     it("suppresses re-log and re-audit when only binding holder identity changes", async () => {

packages/engine/src/concurrency.ts

@@ -1,4 +1,7 @@
 import type { Task } from "@fusion/core";
+import { createLogger } from "./logger.js";
+
+const concurrencyLog = createLogger("concurrency");
 
 /** Priority level for merge agents — served first. */
 export const PRIORITY_MERGE = 2;
@@ -106,6 +109,7 @@ export class AgentSemaphore {
   private _active = 0;
   private _waiters: PriorityWaiter[] = [];
   private _getLimit: () => number;
+  private _excessReleaseWarned = false;
 
   /**
    * @param limit - Either a static number or a getter that returns the current
@@ -118,7 +122,7 @@ export class AgentSemaphore {
 
   /** Number of slots currently held by running agents. */
   get activeCount(): number {
-    return this._active;
+    return Math.max(0, this._active);
   }
 
   /** Number of callers currently queued for a semaphore slot. */
@@ -219,8 +223,7 @@ export class AgentSemaphore {
    * (if any).
    */
   release(): void {
-    this._active--;
+    this.returnSlot("release");
-    this._drain();
   }
 
   /**
@@ -262,11 +265,29 @@ export class AgentSemaphore {
     try {
       return await fn();
     } finally {
-      this._active--;
+      this.returnSlot("runNested");
-      this._drain();
     }
   }
 
+  /**
+   * FNXC:Scheduler-Concurrency 2026-06-13-19:58:
+   * FN-6423 requires excess slot returns to remain observable without corrupting scheduler capacity accounting. Clamp the active slot count at zero and warn once so a release leak cannot surface as negative `activeCount` or a negative `semaphore used=` diagnostic.
+   */
+  private returnSlot(source: "release" | "runNested"): void {
+    if (this._active <= 0) {
+      this._active = 0;
+      if (!this._excessReleaseWarned) {
+        this._excessReleaseWarned = true;
+        concurrencyLog.warn(`AgentSemaphore excess slot return ignored from ${source}; activeCount already 0`);
+      }
+      this._drain();
+      return;
+    }
+
+    this._active--;
+    this._drain();
+  }
+
   /**
    * Unblock waiters while slots are available.
    *

packages/engine/src/scheduler.ts

@@ -343,36 +343,47 @@ function computeConcurrencyGateDiagnostic(params: {
   maxWorktrees: number;
   semaphore?: AgentSemaphore;
   inProgressTaskIds: string[];
-  available: number;
+  startedThisTick?: number;
   /** U6: additive per-column capacity gates (flag-ON only). Omitted → the legacy
    *  three-gate report is byte-identical. */
   perColumnGates?: PerColumnCapacityGate[];
 }): ConcurrencyGateDiagnostic {
+  const startedThisTick = Math.max(0, Math.floor(params.startedThisTick ?? 0));
+  const maxConcurrentUsed = params.agentSlots + startedThisTick;
+  const maxWorktreesUsed = params.activeWorktrees + startedThisTick;
   const maxConcurrentGate: ConcurrencyGateSnapshot = {
-    used: params.agentSlots,
+    used: maxConcurrentUsed,
     limit: params.maxConcurrent,
-    slack: params.maxConcurrent - params.agentSlots,
+    slack: params.maxConcurrent - maxConcurrentUsed,
   };
   const maxWorktreesGate: ConcurrencyGateSnapshot = {
-    used: params.activeWorktrees,
+    used: maxWorktreesUsed,
     limit: params.maxWorktrees,
-    slack: params.maxWorktrees - params.activeWorktrees,
+    slack: params.maxWorktrees - maxWorktreesUsed,
   };
   const semaphoreGate = params.semaphore
-    ? {
+    ? (() => {
-      used: params.semaphore.activeCount,
+      const used = Math.max(0, params.semaphore.activeCount, params.agentSlots) + startedThisTick;
-      limit: params.semaphore.limit,
+      return {
-      slack: params.semaphore.availableCount,
+        used,
-    }
+        limit: params.semaphore.limit,
+        slack: params.semaphore.limit - used,
+      };
+    })()
     : undefined;
+  const available = Math.min(
+    maxConcurrentGate.slack,
+    maxWorktreesGate.slack,
+    semaphoreGate?.slack ?? Infinity,
+  );
 
   const bindingGates: ConcurrencyGateName[] = [];
-  if (maxConcurrentGate.slack === params.available) bindingGates.push("maxConcurrent");
+  if (maxConcurrentGate.used >= maxConcurrentGate.limit) bindingGates.push("maxConcurrent");
-  if (maxWorktreesGate.slack === params.available) bindingGates.push("maxWorktrees");
+  if (maxWorktreesGate.used >= maxWorktreesGate.limit) bindingGates.push("maxWorktrees");
-  if (semaphoreGate && semaphoreGate.slack === params.available) bindingGates.push("semaphore");
+  if (semaphoreGate && semaphoreGate.used >= semaphoreGate.limit) bindingGates.push("semaphore");
 
   return {
-    available: params.available,
+    available,
     bindingGates,
     maxConcurrentGate,
     maxWorktreesGate,
@@ -398,8 +409,9 @@ function formatConcurrencyLimitReason(diagnostic: ConcurrencyGateDiagnostic): st
     `maxWorktrees used=${diagnostic.maxWorktreesGate.used}/${diagnostic.maxWorktreesGate.limit} (holders: ${holdersText("maxWorktrees")})`,
   ];
   if (diagnostic.semaphoreGate) {
+    const semaphoreUsed = Math.max(0, diagnostic.semaphoreGate.used);
     details.push(
-      `semaphore used=${diagnostic.semaphoreGate.used}/${diagnostic.semaphoreGate.limit} (holders: ${holdersText("semaphore")}; note: semaphore slots may include triage/merge agents outside in-progress)`,
+      `semaphore used=${semaphoreUsed}/${diagnostic.semaphoreGate.limit} (holders: ${holdersText("semaphore")}; note: semaphore slots may include triage/merge agents outside in-progress)`,
     );
   }
   return `queued — concurrency limit reached: gate=${gateLabel}; ${details.join("; ")}`;
@@ -1243,43 +1255,34 @@ export class Scheduler {
 
       // When a semaphore is provided, factor in its available slots so we
       // don't schedule more tasks than the global limit allows.
-      const semaphoreAvailable = this.options.semaphore
-        ? Math.min(
-          this.options.semaphore.availableCount,
-          this.options.semaphore.limit - agentSlots,
-        )
-        : Infinity;
-
-      const available = Math.min(
-        maxConcurrent - agentSlots,
-        maxWorktrees - activeWorktrees,
-        semaphoreAvailable,
-      );
       const inProgressTaskIds = inProgress.map((task) => task.id);
-      // U6 (KTD-10): when the workflowColumns flag is ON, report the default
+      const computeDispatchCapacityDiagnostic = (startedThisTick: number): ConcurrencyGateDiagnostic => {
-      // workflow's in-progress capacity as a per-column gate — the generalization
+        const started = Math.max(0, Math.floor(startedThisTick));
-      // of the legacy maxConcurrent gate (which reads through to the same value).
+        // U6 (KTD-10): when the workflowColumns flag is ON, report the default
-      // Additive: omitted flag-OFF so the three-gate report shape is unchanged.
+        // workflow's in-progress capacity as a per-column gate — the generalization
-      const perColumnGates = isWorkflowColumnsEnabled(settings)
+        // of the legacy maxConcurrent gate (which reads through to the same value).
-        ? [{
+        // Additive: omitted flag-OFF so the three-gate report shape is unchanged.
-          workflowId: DEFAULT_WORKFLOW_POOL_ID,
+        const perColumnGates = isWorkflowColumnsEnabled(settings)
-          columnId: "in-progress",
+          ? [{
-          used: agentSlots,
+            workflowId: DEFAULT_WORKFLOW_POOL_ID,
-          limit: maxConcurrent,
+            columnId: "in-progress",
-          slack: maxConcurrent - agentSlots,
+            used: agentSlots + started,
-        }]
+            limit: maxConcurrent,
-        : undefined;
+            slack: maxConcurrent - (agentSlots + started),
-      const concurrencyGateDiagnostic = computeConcurrencyGateDiagnostic({
+          }]
-        agentSlots,
+          : undefined;
-        maxConcurrent,
+        return computeConcurrencyGateDiagnostic({
-        activeWorktrees,
+          agentSlots,
-        maxWorktrees,
+          maxConcurrent,
-        semaphore: this.options.semaphore,
+          activeWorktrees,
-        inProgressTaskIds,
+          maxWorktrees,
-        available,
+          semaphore: this.options.semaphore,
-        perColumnGates,
+          inProgressTaskIds,
-      });
+          startedThisTick: started,
-      if (available <= 0) return;
+          perColumnGates,
+        });
+      };
+      if (computeDispatchCapacityDiagnostic(0).available <= 0) return;
 
       const now = Date.now();
       let todo = tasks.filter((t) => {
@@ -1626,10 +1629,14 @@ export class Scheduler {
           }
         }
 
-        // Dependencies met — check concurrency
+        /**
-        if (started >= available) {
+         * FNXC:Scheduler-Concurrency 2026-06-13-20:08:
-          const reason = formatConcurrencyLimitReason(concurrencyGateDiagnostic);
+         * FN-6423 fixes the FN-6420 evidence where queue logs reported `gate=maxWorktrees` with `maxWorktrees used=1/3` and `semaphore used=-9/3`. Recompute capacity at the queue decision point, including tasks already started this tick, so the gate label, memo key, and `started` decision share one authoritative snapshot.
-          const concurrencySignature = formatConcurrencyLimitMemoKey(concurrencyGateDiagnostic);
+         */
+        const queuePointCapacity = computeDispatchCapacityDiagnostic(started);
+        if (queuePointCapacity.available <= 0) {
+          const reason = formatConcurrencyLimitReason(queuePointCapacity);
+          const concurrencySignature = formatConcurrencyLimitMemoKey(queuePointCapacity);
           await this.logDispatchQueuedReason(
             task.id,
             reason,