From be2773b412949373d718dce14e93af09dbe73457 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 13 Jun 2026 21:23:39 -0700 Subject: [PATCH] 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 | 5 + .../engine/src/__tests__/concurrency.test.ts | 32 +++++ .../__tests__/executor-step-session.test.ts | 10 +- .../engine/src/__tests__/scheduler.test.ts | 122 +++++++++++++++++- packages/engine/src/concurrency.ts | 31 ++++- packages/engine/src/scheduler.ts | 117 +++++++++-------- 6 files changed, 249 insertions(+), 68 deletions(-) create mode 100644 .changeset/fn-6423-scheduler-capacity.md diff --git a/.changeset/fn-6423-scheduler-capacity.md b/.changeset/fn-6423-scheduler-capacity.md new file mode 100644 index 0000000000..e94a1d6e62 --- /dev/null +++ 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. diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index 9d8e51ba20..0a0b60334e 100644 --- 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 diff --git a/packages/engine/src/__tests__/executor-step-session.test.ts b/packages/engine/src/__tests__/executor-step-session.test.ts index a8158d3351..db254e6a21 100644 --- 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) => { + 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"); diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index 80396cfb06..1236d84cde 100644 --- 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).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).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).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).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).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).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 () => { diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 651d3032b2..2861ae65c8 100644 --- 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. * diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index a21a2cd5aa..a18a6e62a5 100644 --- 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,