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
2026-06-13 21:23:39 -07:00
parent b1ba87e599
commit be2773b412
6 changed files with 249 additions and 68 deletions
--- a/.changeset/fn-6423-scheduler-capacity.md
+++ b/.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.
--- a/packages/engine/src/tests/concurrency.test.ts
+++ b/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
--- a/packages/engine/src/tests/executor-step-session.test.ts
+++ b/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({
-      ...mockedCreateFnAgent.mock.calls[0][0].taskEnv,
-      FUSION_WORKFLOW_STEP: "1",
-    });
+    const withoutWorkflowStep = (env: Record<string, string | undefined>) => {
+      const { FUSION_WORKFLOW_STEP: _workflowStep, ...stableEnv } = env;
+      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");
--- a/packages/engine/src/tests/scheduler.test.ts
+++ b/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 () => {
--- a/packages/engine/src/concurrency.ts
+++ b/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._drain();
+    this.returnSlot("release");
  }

  /**
@@ -262,11 +265,29 @@ export class AgentSemaphore {
    try {
      return await fn();
    } finally {
-      this._active--;
-      this._drain();
+      this.returnSlot("runNested");
    }
  }

+  /**
+   * 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.
   *
--- a/packages/engine/src/scheduler.ts
+++ b/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,
-      limit: params.semaphore.limit,
-      slack: params.semaphore.availableCount,
-    }
+    ? (() => {
+      const used = Math.max(0, params.semaphore.activeCount, params.agentSlots) + startedThisTick;
+      return {
+        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 (maxWorktreesGate.slack === params.available) bindingGates.push("maxWorktrees");
-  if (semaphoreGate && semaphoreGate.slack === params.available) bindingGates.push("semaphore");
+  if (maxConcurrentGate.used >= maxConcurrentGate.limit) bindingGates.push("maxConcurrent");
+  if (maxWorktreesGate.used >= maxWorktreesGate.limit) bindingGates.push("maxWorktrees");
+  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
-      // workflow's in-progress capacity as a per-column gate — the generalization
-      // of the legacy maxConcurrent gate (which reads through to the same value).
-      // Additive: omitted flag-OFF so the three-gate report shape is unchanged.
-      const perColumnGates = isWorkflowColumnsEnabled(settings)
-        ? [{
-          workflowId: DEFAULT_WORKFLOW_POOL_ID,
-          columnId: "in-progress",
-          used: agentSlots,
-          limit: maxConcurrent,
-          slack: maxConcurrent - agentSlots,
-        }]
-        : undefined;
-      const concurrencyGateDiagnostic = computeConcurrencyGateDiagnostic({
-        agentSlots,
-        maxConcurrent,
-        activeWorktrees,
-        maxWorktrees,
-        semaphore: this.options.semaphore,
-        inProgressTaskIds,
-        available,
-        perColumnGates,
-      });
-      if (available <= 0) return;
+      const computeDispatchCapacityDiagnostic = (startedThisTick: number): ConcurrencyGateDiagnostic => {
+        const started = Math.max(0, Math.floor(startedThisTick));
+        // U6 (KTD-10): when the workflowColumns flag is ON, report the default
+        // workflow's in-progress capacity as a per-column gate — the generalization
+        // of the legacy maxConcurrent gate (which reads through to the same value).
+        // Additive: omitted flag-OFF so the three-gate report shape is unchanged.
+        const perColumnGates = isWorkflowColumnsEnabled(settings)
+          ? [{
+            workflowId: DEFAULT_WORKFLOW_POOL_ID,
+            columnId: "in-progress",
+            used: agentSlots + started,
+            limit: maxConcurrent,
+            slack: maxConcurrent - (agentSlots + started),
+          }]
+          : undefined;
+        return computeConcurrencyGateDiagnostic({
+          agentSlots,
+          maxConcurrent,
+          activeWorktrees,
+          maxWorktrees,
+          semaphore: this.options.semaphore,
+          inProgressTaskIds,
+          startedThisTick: started,
+          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) {
-          const reason = formatConcurrencyLimitReason(concurrencyGateDiagnostic);
-          const concurrencySignature = formatConcurrencyLimitMemoKey(concurrencyGateDiagnostic);
+        /**
+         * FNXC:Scheduler-Concurrency 2026-06-13-20:08:
+         * 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 queuePointCapacity = computeDispatchCapacityDiagnostic(started);
+        if (queuePointCapacity.available <= 0) {
+          const reason = formatConcurrencyLimitReason(queuePointCapacity);
+          const concurrencySignature = formatConcurrencyLimitMemoKey(queuePointCapacity);
          await this.logDispatchQueuedReason(
            task.id,
            reason,