fix: stop agents exceeding the global concurrency cap#2107
Merged
Conversation
Triage and scheduling could admit planners and an executor together so live running counts exceeded Global Max Concurrent. Reserve a shared semaphore slot before todo→in-progress, hand it to the executor/graph without double-acquire, and admit planners against the same live running-agent claim the dashboard shows.
Contributor
📝 WalkthroughWalkthroughThe scheduler and triage processor now account for live top-level agent claims, pending planner demand, and semaphore activity. Pre-held semaphore slots transfer from scheduling to execution without nested acquisition, with explicit cleanup and regression coverage. ChangesGlobal concurrency control
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
When workflow graph falls back and re-registers a scheduler pre-held concurrency slot, every execute() path that never reaches runWithExecutorSemaphore must drop the registration. Authoritative dispatch accept, work-engine claim, heartbeat defer, lock contention, and outer finally now release the slot so global capacity cannot leak.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/engine/src/executor.ts (1)
10329-10396: 🩺 Stability & Availability | 🟠 Major | ⚖️ Poor tradeoffSeed the outer concurrency claim before constructing
StepSessionExecutor— on the legacy fallback path,maybeExecuteWorkflowGraph()dropsouterConcurrencyClaimsbefore returning, andrunWithExecutorSemaphore()only restores it after this constructor has already run. That leaves step sessions with a real semaphore on stores that lack workflow-selection APIs, so per-stepacquire()can still double-book the top-level slot and deadlock under a full cap.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/src/executor.ts` around lines 10329 - 10396, Seed this task’s outer concurrency claim before constructing StepSessionExecutor, including the legacy fallback path where maybeExecuteWorkflowGraph() removes it before returning. Ensure runWithExecutorSemaphore() restores or establishes the claim before the StepSessionExecutor semaphore selection executes, so the existing outerConcurrencyClaims check passes and per-step sessions receive no semaphore when the top-level slot is already owned.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@packages/engine/src/executor.ts`:
- Around line 10329-10396: Seed this task’s outer concurrency claim before
constructing StepSessionExecutor, including the legacy fallback path where
maybeExecuteWorkflowGraph() removes it before returning. Ensure
runWithExecutorSemaphore() restores or establishes the claim before the
StepSessionExecutor semaphore selection executes, so the existing
outerConcurrencyClaims check passes and per-step sessions receive no semaphore
when the top-level slot is already owned.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 8c5eff92-4f7c-43e9-b377-0e589760813c
📒 Files selected for processing (7)
.changeset/fix-global-concurrency-over-cap.mdpackages/engine/src/__tests__/concurrency.test.tspackages/engine/src/__tests__/triage.test.tspackages/engine/src/concurrency.tspackages/engine/src/executor.tspackages/engine/src/scheduler.tspackages/engine/src/triage.ts
Graph fallback re-registers a scheduler concurrency slot for the legacy path. If workflowAuthoritativeDispatch throws, execute exited before the accept-path drop and before the main finally, leaking the semaphore. Drop the registration before rethrowing and cover the rejection surface.
Keep global concurrency pre-held slot helpers and the planner overseer session-advisor log-flush setter from main.
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
packages/engine/src/__tests__/executor-preheld-legacy-handoff.test.ts (1)
3-13: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the redundant import.
The side-effect import of
"./executor-test-helpers.js"on line 3 is redundant sincecreateMockStoreandresetExecutorMocksare already imported from the exact same file on line 12.♻️ Proposed refactor
-import "./executor-test-helpers.js"; import { AgentSemaphore, clearPreHeldExecutorSlotsForTests, hasPreHeldExecutorSlot, registerPreHeldExecutorSlot, } from "../concurrency.js"; import { TaskExecutor } from "../executor.js"; import { executingTaskLock } from "../active-session-registry.js"; import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js";🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/src/__tests__/executor-preheld-legacy-handoff.test.ts` around lines 3 - 13, Remove the standalone side-effect import of "./executor-test-helpers.js" and retain the existing named import of createMockStore and resetExecutorMocks from that module.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@packages/engine/src/__tests__/executor-preheld-legacy-handoff.test.ts`:
- Around line 3-13: Remove the standalone side-effect import of
"./executor-test-helpers.js" and retain the existing named import of
createMockStore and resetExecutorMocks from that module.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c8a87375-b0cb-4858-9138-8891695f8279
📒 Files selected for processing (3)
packages/engine/src/__tests__/concurrency.test.tspackages/engine/src/__tests__/executor-preheld-legacy-handoff.test.tspackages/engine/src/executor.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/engine/src/executor.ts
Wrap execute() in try/finally that always dropPreHeldExecutorSlot so new early-return paths cannot permanently shrink global capacity. Document the register-only-after-tryAcquire invariant and cover scheduler prep.release on move failure.
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
packages/engine/src/__tests__/concurrency.test.ts (1)
502-510: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTest the idempotency of
dropPreHeldExecutorSlotdirectly.The
releasedboolean guard short-circuits the second call, meaning the test verifies the local wrapper's logic rather than the underlying utility. SincedropPreHeldExecutorSlotis inherently idempotent (it checksSet.delete()before releasing), you can simplify the test by calling it twice directly. This ensures the concurrency primitives themselves are robust against duplicate calls.♻️ Proposed refactor
- // Mirrors scheduler.ts prep.release() after a failed/aborted hold release. - let released = false; - const release = () => { - if (released) return; - released = true; - dropPreHeldExecutorSlot("FN-MOVE-FAIL", sem); - }; - release(); - release(); // idempotent + // dropPreHeldExecutorSlot is inherently idempotent; calling it twice is safe + dropPreHeldExecutorSlot("FN-MOVE-FAIL", sem); + dropPreHeldExecutorSlot("FN-MOVE-FAIL", sem); // idempotent🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/src/__tests__/concurrency.test.ts` around lines 502 - 510, Update the test around dropPreHeldExecutorSlot to call the utility directly twice with the same executor and semaphore, removing the local released guard and wrapper. Keep the assertions that verify only one slot is released, so the test covers the utility’s own idempotency.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@packages/engine/src/__tests__/concurrency.test.ts`:
- Around line 502-510: Update the test around dropPreHeldExecutorSlot to call
the utility directly twice with the same executor and semaphore, removing the
local released guard and wrapper. Keep the assertions that verify only one slot
is released, so the test covers the utility’s own idempotency.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 9dced772-783a-479e-b777-b9e25eff98f7
📒 Files selected for processing (4)
packages/engine/src/__tests__/concurrency.test.tspackages/engine/src/__tests__/executor-preheld-legacy-handoff.test.tspackages/engine/src/concurrency.tspackages/engine/src/executor.ts
🚧 Files skipped from review as they are similar to previous changes (3)
- packages/engine/src/concurrency.ts
- packages/engine/src/tests/executor-preheld-legacy-handoff.test.ts
- packages/engine/src/executor.ts
gsxdsm
added a commit
that referenced
this pull request
Jul 15, 2026
## Summary - Operators could see more agents running than Global Max Concurrent (e.g. 5 running with cap 4: 4 planners + 1 executor). - Scheduler now `tryAcquire`s a shared semaphore slot before todo→in-progress and hands that pre-held slot to the executor/graph run. - Triage admits planners against the live top-level running-agent claim (planning + in-progress + active in-review), not only `semaphore.availableCount`. - Executor claims the pre-held slot for the full run and avoids a second top-level acquire on step/seam re-entry (deadlock under a full cap). ## Test plan - [x] `pnpm --filter @fusion/engine exec vitest run src/__tests__/concurrency.test.ts src/__tests__/triage.test.ts` - [x] Regression: triage leaves room when 1 in-progress agent is live under global cap 4 - [x] Regression: pre-held executor slot register/take/drop handoff - [ ] Manual: set Global Max Concurrent and Max triage concurrent to 4, fill Planning + run 1 In Progress; footer should not show 5 running under a full steady state <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **Bug Fixes** - Improved global concurrency enforcement so the scheduler and executor never start more agents than the configured limit, including tighter top-level “claimed capacity” accounting. - Updated triage admission control to consider global top-level utilization, factoring processing tasks and agents already running to prevent over-admitting planners. - Added safer pre-held concurrency-slot handoff behavior to avoid capacity leaks and drift during graph routing, step execution, and legacy fallback. - Ensured reserved capacity is reliably released on early exits, failed dispatches, and other aborted paths (with idempotent cleanup). - Refreshed concurrency diagnostics to better explain whether throttling is due to project or global limits, with clearer claimed/processing visibility. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
tryAcquires a shared semaphore slot before todo→in-progress and hands that pre-held slot to the executor/graph run.semaphore.availableCount.Test plan
pnpm --filter @fusion/engine exec vitest run src/__tests__/concurrency.test.ts src/__tests__/triage.test.tsSummary by CodeRabbit