diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index c844003796..b5fdad1b61 100644 --- a/packages/engine/src/__tests__/concurrency.test.ts +++ b/packages/engine/src/__tests__/concurrency.test.ts @@ -443,12 +443,15 @@ describe("AgentSemaphore", () => { expect(hasPreHeldExecutorSlot("FN-1")).toBe(false); expect(takePreHeldExecutorSlot("FN-1")).toBe(false); - // Failed dispatch path releases both the registry entry and the semaphore slot. - expect(sem.tryAcquire()).toBe(true); + /* + FNXC:CapacityModel 2026-07-29-13:20: the failed-dispatch path releases the + REGISTRY ENTRY; the semaphore half moved to the call sites that own the + reference. The surviving contract is that a failed dispatch leaves nothing a + later pass could "take". + */ registerPreHeldExecutorSlot("FN-2"); - dropPreHeldExecutorSlot("FN-2", sem); + dropPreHeldExecutorSlot("FN-2"); expect(hasPreHeldExecutorSlot("FN-2")).toBe(false); - expect(sem.activeCount).toBe(1); sem.release(); clearPreHeldExecutorSlotsForTests(); }); @@ -471,8 +474,9 @@ describe("AgentSemaphore", () => { expect(sem.activeCount).toBe(1); // Authoritative / work-engine / heartbeat-defer early returns must drop, not leave the registration. - dropPreHeldExecutorSlot("FN-LEGACY-HANDOFF", sem); + dropPreHeldExecutorSlot("FN-LEGACY-HANDOFF"); expect(hasPreHeldExecutorSlot("FN-LEGACY-HANDOFF")).toBe(false); + sem.release(); expect(sem.activeCount).toBe(0); // Happy path: re-register then take + release (runWithExecutorSemaphore contract). @@ -483,8 +487,8 @@ describe("AgentSemaphore", () => { sem.release(); expect(sem.activeCount).toBe(0); // Second drop after take is a no-op — safe for execute()'s outer finally belt-and-suspenders. - dropPreHeldExecutorSlot("FN-LEGACY-TAKE", sem); - expect(sem.activeCount).toBe(0); + dropPreHeldExecutorSlot("FN-LEGACY-TAKE"); + expect(hasPreHeldExecutorSlot("FN-LEGACY-TAKE")).toBe(false); clearPreHeldExecutorSlotsForTests(); }); @@ -507,13 +511,12 @@ describe("AgentSemaphore", () => { const release = () => { if (released) return; released = true; - dropPreHeldExecutorSlot("FN-MOVE-FAIL", sem); + dropPreHeldExecutorSlot("FN-MOVE-FAIL"); }; release(); release(); // idempotent expect(hasPreHeldExecutorSlot("FN-MOVE-FAIL")).toBe(false); - expect(sem.activeCount).toBe(0); clearPreHeldExecutorSlotsForTests(); }); @@ -1207,8 +1210,10 @@ describe("ProjectAdmissionCoordinator", () => { expect(semaphore.activeCount).toBe(1); expect(hasPreHeldExecutorSlot("FN-TAKES")).toBe(true); - dropPreHeldExecutorSlot("FN-TAKES", semaphore); - expect(semaphore.activeCount).toBe(0); + dropPreHeldExecutorSlot("FN-TAKES"); + expect(hasPreHeldExecutorSlot("FN-TAKES")).toBe(false); + // The admitted candidate keeps its host slot until its own lane releases. + expect(semaphore.activeCount).toBe(1); clearPreHeldExecutorSlotsForTests(); }); diff --git a/packages/engine/src/__tests__/prehold-release-ownership.test.ts b/packages/engine/src/__tests__/prehold-release-ownership.test.ts new file mode 100644 index 0000000000..b220c1debd --- /dev/null +++ b/packages/engine/src/__tests__/prehold-release-ownership.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from "vitest"; +import { + AgentSemaphore, + registerPreHeldExecutorSlot, + takePreHeldExecutorSlot, + dropPreHeldExecutorSlot, + clearPreHeldExecutorSlotsForTests, +} from "../concurrency.js"; + +/* +FNXC:CapacityModel 2026-07-29-17:10 (PR #2574 review — greptile P1): +A drop AFTER a successful transfer must not release a slot it no longer owns. + +The two-argument form released the semaphore inside the "did I actually drop +anything?" guard, so a cleanup call following `takePreHeldExecutorSlot` was +intentionally inert — the transferred slot belongs to the lane, which releases it +through `semaphore.run`. Hoisting the release to the call site unconditionally +released a slot the call never held, INFLATING capacity: the opposite of the leak +the cleanup exists to prevent, and invisible because an over-released semaphore +simply admits more work. +*/ +describe("pre-held slot release ownership", () => { + it("reports false after a transfer, so the caller does not double-release", () => { + clearPreHeldExecutorSlotsForTests(); + const sem = new AgentSemaphore(2); + expect(sem.tryAcquire()).toBe(true); + registerPreHeldExecutorSlot("FN-XFER"); + + // The lane takes ownership; it will release via its own finally / semaphore.run. + expect(takePreHeldExecutorSlot("FN-XFER")).toBe(true); + + // The outer cleanup still runs. It must report that it dropped NOTHING. + const dropped = dropPreHeldExecutorSlot("FN-XFER"); + expect(dropped).toBe(false); + + // Mirrors the call-site guard: release only when the drop acted. + if (dropped) sem.release(); + expect(sem.activeCount, "the lane still owns its slot").toBe(1); + + sem.release(); + expect(sem.activeCount).toBe(0); + clearPreHeldExecutorSlotsForTests(); + }); + + it("reports true for an untransferred slot, so the caller does release", () => { + clearPreHeldExecutorSlotsForTests(); + const sem = new AgentSemaphore(2); + expect(sem.tryAcquire()).toBe(true); + registerPreHeldExecutorSlot("FN-EARLY-FAIL"); + + const dropped = dropPreHeldExecutorSlot("FN-EARLY-FAIL"); + expect(dropped).toBe(true); + if (dropped) sem.release(); + expect(sem.activeCount, "an early failure returns its untransferred slot").toBe(0); + clearPreHeldExecutorSlotsForTests(); + }); + + it("reports false when nothing was ever registered", () => { + clearPreHeldExecutorSlotsForTests(); + expect(dropPreHeldExecutorSlot("FN-NEVER")).toBe(false); + }); +}); diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 5f36d4ef5c..41d2e872e9 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -176,11 +176,18 @@ export class ProjectAdmissionCoordinator { */ const releaseAttempt = () => { if (!hasReservableHostSlot) return; + /* + FNXC:CapacityModel 2026-07-29-13:20: the host-slot release is now + UNCONDITIONAL across both branches. The pre-held branch used to release the + caller's semaphore through dropPreHeldExecutorSlot's second parameter; + with that parameter deleted it would otherwise unwind the registration and + the reservation while LEAKING the host slot this attempt acquired. + */ if (hasPreHeldExecutorSlot(winner.taskId)) { - dropPreHeldExecutorSlot(winner.taskId, params.semaphore); - return; + dropPreHeldExecutorSlot(winner.taskId); + } else { + this.releaseReservation(winner.taskId); } - this.releaseReservation(winner.taskId); params.semaphore?.release(); }; try { @@ -295,14 +302,40 @@ export function takePreHeldExecutorSlot(taskId: string): boolean { return taken; } -/** Drop a pre-held slot without transferring ownership (failed reserve / cancelled dispatch). Optionally releases the semaphore. */ -export function dropPreHeldExecutorSlot(taskId: string, semaphore?: { release(): void }): void { - if (!preHeldExecutorSlots.delete(taskId)) return; - // FNXC:ConcurrencyAdmission 2026-08-06-12:00: every rejection path funnels - // through this helper, so releasing the matching coordinator marker here - // prevents early scheduler/triage returns from permanently consuming a slot. +/* +FNXC:CapacityModel 2026-07-29-13:20 (drop the cross-project cap — pre-held slots): +Drop a pre-held slot without transferring ownership (failed reserve / cancelled +dispatch). + +The `semaphore` parameter is GONE. These slots are DUAL-PURPOSE — a cross-project +semaphore slot AND the FN-8453 per-project coordinator reservation — and only the +first half is deleted. All 16 production call sites passed `this.options.semaphore`, +which nothing wires any more, so the release was a no-op on an always-undefined +value: an optional parameter that reads as if it does something. + +The coordinator reservation is the half that MATTERS and stays: every rejection path +funnels through this helper so an early scheduler/triage return cannot permanently +consume a project slot (FNXC:ConcurrencyAdmission 2026-08-06-12:00). + +Sites that still hold a semaphore reference release it EXPLICITLY next to their +drop, so behaviour is unchanged for any caller that supplies one. +*/ +export function dropPreHeldExecutorSlot(taskId: string): boolean { + /* + FNXC:CapacityModel 2026-07-29-17:10 (PR #2574 review — greptile P1, double release): + RETURNS whether a registration was actually dropped, because callers that own a + semaphore reference must release it ONLY when this did something. + + The original two-argument form released the semaphore INSIDE this guard, so a call + after a successful `takePreHeldExecutorSlot` was "intentionally a no-op" — the + transferred slot belongs to the lane, which releases it via `semaphore.run`. + Hoisting the release to the call site unconditionally broke that: it released a + slot this call never held, INFLATING capacity — the opposite of the leak the + cleanup was guarding against. + */ + if (!preHeldExecutorSlots.delete(taskId)) return false; projectAdmissionCoordinator.releaseReservation(taskId); - semaphore?.release(); + return true; } /** Test/helper: whether a task currently has an unclaimed pre-held executor slot. */ diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3a62df97dc..1557736a70 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -11826,7 +11826,7 @@ export class TaskExecutor { try { await this.executeCore(task); } finally { - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); } } @@ -11854,7 +11854,7 @@ export class TaskExecutor { */ if (task.deletedAt) { executorLog.warn(`${task.id}: refusing execute — task is soft-deleted`); - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } /* @@ -11876,14 +11876,14 @@ export class TaskExecutor { await this.clearStalePauseAbortBeforeDispatch(task); if (await this.blockOuterDispatchWhenDependenciesUnmet(task)) { // FNXC:GlobalConcurrencyControls 2026-07-14-18:30: release any scheduler pre-held slot when outer dispatch aborts before agent work starts. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } // FNXC:EphemeralAgents 2026-07-01-00:00: gate ALL workflow dispatch paths // (graph/authoritative/work-engine) on ephemeralAgentsEnabled before any of // them can claim the task, so the single check covers all three entry points. if (await this.blockOuterDispatchWhenEphemeralDisabled(task)) { - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } /* @@ -11963,7 +11963,7 @@ export class TaskExecutor { executorLog.debug(`execute() called for ${task.id} (claimed=${claimed}, perInstanceExecuting=${this.executing.has(task.id)})`); if (!claimed) { // FNXC:GlobalConcurrencyControls 2026-07-15-02:55: graph fallback may have re-registered a pre-held slot; drop it when this process cannot claim the executor lock. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } @@ -11977,7 +11977,7 @@ export class TaskExecutor { executorLog.warn(`${task.id}: refusing execute — task is soft-deleted`); this.executing.delete(task.id); executingTaskLock.release(task.id); - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } @@ -11986,7 +11986,7 @@ export class TaskExecutor { this.executing.delete(task.id); executingTaskLock.release(task.id); // FNXC:GlobalConcurrencyControls 2026-07-15-02:55: work-engine ownership never take()s the legacy handoff registration — release the reserved global slot. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } @@ -12005,7 +12005,7 @@ export class TaskExecutor { this.executing.delete(task.id); executingTaskLock.release(task.id); // FNXC:GlobalConcurrencyControls 2026-07-15-02:55: heartbeat defer must free any re-registered pre-held global slot so capacity is not stranded until the next dispatch. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } @@ -12078,7 +12078,7 @@ export class TaskExecutor { await this.store.updateTask(task.id, { status: "needs-replan" }); await this.store.logEntry(task.id, staleness.reason, undefined, this.getRunContextFor(task.id)); // FNXC:GlobalConcurrencyControls 2026-07-15-02:55: replan handoff never starts agent work — free any re-registered pre-held slot before leaving execute(). - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } } @@ -12095,7 +12095,7 @@ export class TaskExecutor { if (await this.finalizeMergeConfirmedWorkflowGraphTask(task.id, "execute-preflight")) { this.executing.delete(task.id); executingTaskLock.release(task.id); - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); return; } } @@ -14954,7 +14954,7 @@ export class TaskExecutor { release any still-registered slot before lock/executing cleanup. execute()'s outer finally also drops (no-op once take/drop already cleared the registration). */ - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); this.executing.delete(task.id); executingTaskLock.release(task.id); diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 254287ed28..b758538916 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -2266,7 +2266,7 @@ export class Scheduler { SYMBOL_LOCK_LEASE_MS, ); if (!lockResult.acquired) { - dropPreHeldExecutorSlot(task.id, sem); + if (dropPreHeldExecutorSlot(task.id)) sem?.release(); if (reservedScope) { activeScopes.delete(task.id); activeScopeColumns.delete(task.id); @@ -2306,7 +2306,7 @@ export class Scheduler { reservedWorktreeSlots = Math.max(0, reservedWorktreeSlots - 1); reservedConcurrentSlots = Math.max(0, reservedConcurrentSlots - 1); dispatchPrepByTaskId.delete(task.id); - dropPreHeldExecutorSlot(task.id, sem); + if (dropPreHeldExecutorSlot(task.id)) sem?.release(); if (acquiredSymbols) { void this.store.releaseSymbolLocks(acquiredSymbols, task.id); } diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index f25e7ff48f..41910616bb 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -1124,7 +1124,7 @@ export class TriageProcessor { invariant holds either way. */ this.coordinatorAdmittedTaskIds.delete(taskId); - dropPreHeldExecutorSlot(taskId, this.options.semaphore); + if (dropPreHeldExecutorSlot(taskId)) this.options.semaphore?.release(); evicted.add(taskId); } @@ -2009,7 +2009,7 @@ export class TriageProcessor { // FNXC:ConcurrencyAdmission 2026-08-06-09:00: // A coordinator winner owns a real pre-held host slot. A duplicate/stale // planner handoff must return it instead of pinning max concurrency. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); this.coordinatorAdmittedTaskIds.delete(task.id); return; } @@ -2981,7 +2981,7 @@ export class TriageProcessor { // can exist before planner setup reaches takePreHeldExecutorSlot(). Every // early setup failure must return that untransferred host slot; after a // successful transfer this is intentionally a no-op. - dropPreHeldExecutorSlot(task.id, this.options.semaphore); + if (dropPreHeldExecutorSlot(task.id)) this.options.semaphore?.release(); /* FNXC:NodeWorktreeIsolation 2026-07-26-09:10: Release the planner's registry entry on EVERY exit path (success, planning failure, abort,