FN-7151: free scoped project concurrency slots

Free stopped or paused project engines' leaked slots without disturbing other projects.

- Add scoped semaphore accounting around the shared global agent pool.
- Return residual per-project slots after runtime stop drains aborted agents.
- Keep idle leak reconciliation scoped to the owning project so active projects retain their slots.
- Cover scoped release, reconciliation, project pause, and in-process runtime behavior with tests.

Files changed:
 .../fn-7151-free-concurrency-slots-on-stop.md      |   7 ++
 docs/architecture.md                               |   2 +-
 packages/engine/src/__tests__/concurrency.test.ts  | 126 +++++++++++++++++++++
 .../src/__tests__/project-engine-manager.test.ts   |  60 ++++++++++
 packages/engine/src/concurrency.ts                 | 123 +++++++++++++++++++-
 packages/engine/src/project-engine.ts              |   2 +-
 .../runtimes/__tests__/in-process-runtime.test.ts  |  51 +++++++++
 packages/engine/src/runtimes/in-process-runtime.ts |  22 +++-
 8 files changed, 385 insertions(+), 8 deletions(-)

Fusion-Task-Id: FN-7151

Fusion-Task-Lineage: e59e5e0c-67e8-4f80-a6b9-1ea4febaeb97

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>

.changeset/fn-7151-free-concurrency-slots-on-stop.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Stopping the engine or pausing a project now frees its global agent slots for other projects.
+category: fix
+dev: InProcessRuntime now returns the project's held slots back to the shared cross-project AgentSemaphore after abort+drain on stop, and ProjectEngineManager.pauseProject/stopAll return residual slots per project without clobbering slots held by other projects.

docs/architecture.md

@@ -669,7 +669,7 @@ Runtime action-gate flow (v1):
 - `require-approval` persists durable requests via `ApprovalRequestStore`, reusing pending requests by dedupe key in `targetAction.context.approvalDedupeKey`.
 
 ### Concurrency, recovery, and resiliency
-- `AgentSemaphore` (`concurrency.ts`) — slot acquisition
+- `AgentSemaphore` (`concurrency.ts`) — slot acquisition. Multi-project runtimes share a single manager-owned semaphore for the cross-project `globalMaxConcurrent` cap, while each `InProcessRuntime` wraps that pool in a scoped semaphore that tracks only that project's held slots. Engine stop, `pauseProject`, and `stopAll` abort in-flight agents, wait the configured stop drain window, then return any residual scoped slots to the shared pool without using a blanket `reconcileActiveCount(0)`, so other projects' active slots are preserved and stopped projects do not starve global capacity.
 - `RecoveryPolicy` (`recovery-policy.ts`) — retry/recovery decision policy
 - `StuckTaskDetector` (`stuck-task-detector.ts`) — inactivity/loop stall detection
 - `GridlockDetector` (`gridlock-detector.ts`) — detects all-blocked todo pipelines and emits notification events (plus explicit clear signals when gridlock resolves)

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

@@ -2,12 +2,138 @@ import { describe, it, expect, vi } from "vitest";
 import type { Task } from "@fusion/core";
 import {
   AgentSemaphore,
+  ScopedAgentSemaphore,
   PRIORITY_MERGE,
   PRIORITY_EXECUTE,
   PRIORITY_SPECIFY,
   recoverIdleSemaphoreLeakCandidate,
 } from "../concurrency.js";
 
+describe("ScopedAgentSemaphore", () => {
+  it("returns only this scope's residual slots without clobbering other scopes", async () => {
+    const shared = new AgentSemaphore(3);
+    const projectA = new ScopedAgentSemaphore(shared);
+    const projectB = new ScopedAgentSemaphore(shared);
+
+    await projectA.acquire(PRIORITY_EXECUTE);
+    await projectA.acquire(PRIORITY_MERGE);
+    await projectB.acquire(PRIORITY_SPECIFY);
+
+    expect(shared.activeCount).toBe(3);
+    expect(projectA.heldCount).toBe(2);
+    expect(projectB.heldCount).toBe(1);
+
+    expect(projectA.returnAllHeldSlots()).toBe(2);
+
+    expect(projectA.heldCount).toBe(0);
+    expect(projectB.heldCount).toBe(1);
+    expect(shared.activeCount).toBe(1);
+    expect(shared.availableCount).toBe(2);
+
+    projectB.release();
+    expect(shared.activeCount).toBe(0);
+  });
+
+  it("is idempotent for zero-held and double residual returns without excess warnings", async () => {
+    const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    try {
+      const shared = new AgentSemaphore(2);
+      const scope = new ScopedAgentSemaphore(shared);
+
+      expect(scope.returnAllHeldSlots()).toBe(0);
+      expect(scope.tryAcquire()).toBe(true);
+      expect(scope.heldCount).toBe(1);
+      expect(scope.returnAllHeldSlots()).toBe(1);
+      expect(scope.returnAllHeldSlots()).toBe(0);
+      scope.release();
+
+      expect(shared.activeCount).toBe(0);
+      expect(shared.availableCount).toBe(2);
+      expect(warnSpy).not.toHaveBeenCalled();
+    } finally {
+      warnSpy.mockRestore();
+    }
+  });
+
+  it("tracks run and runNested slots, and late finally releases are no-ops after residual return", async () => {
+    const shared = new AgentSemaphore(4);
+    const scope = new ScopedAgentSemaphore(shared);
+    let releaseRun!: () => void;
+    let releaseNested!: () => void;
+
+    const runPromise = scope.run(
+      () => new Promise<void>((resolve) => {
+        releaseRun = resolve;
+      }),
+      PRIORITY_EXECUTE,
+    );
+    await Promise.resolve();
+
+    const nestedPromise = scope.runNested(
+      () => new Promise<void>((resolve) => {
+        releaseNested = resolve;
+      }),
+    );
+    await Promise.resolve();
+
+    expect(scope.heldCount).toBe(2);
+    expect(shared.activeCount).toBe(2);
+    expect(scope.returnAllHeldSlots()).toBe(2);
+    expect(shared.activeCount).toBe(0);
+
+    releaseRun();
+    releaseNested();
+    await Promise.all([runPromise, nestedPromise]);
+
+    expect(scope.heldCount).toBe(0);
+    expect(shared.activeCount).toBe(0);
+  });
+
+  it("delegates queued acquisition priority through the shared pool", async () => {
+    const shared = new AgentSemaphore(1);
+    const low = new ScopedAgentSemaphore(shared);
+    const high = new ScopedAgentSemaphore(shared);
+    const order: string[] = [];
+
+    await low.acquire(PRIORITY_SPECIFY);
+    const lowWaiter = low.acquire(PRIORITY_SPECIFY).then(() => order.push("low"));
+    const highWaiter = high.acquire(PRIORITY_MERGE).then(() => order.push("high"));
+    await Promise.resolve();
+
+    low.release();
+    await highWaiter;
+    expect(order).toEqual(["high"]);
+    expect(high.heldCount).toBe(1);
+
+    high.release();
+    await lowWaiter;
+    expect(order).toEqual(["high", "low"]);
+    low.release();
+    expect(shared.activeCount).toBe(0);
+  });
+
+  it("reconciles only this scope's slots when another project still holds global capacity", async () => {
+    const shared = new AgentSemaphore(3);
+    const idleProject = new ScopedAgentSemaphore(shared);
+    const activeProject = new ScopedAgentSemaphore(shared);
+
+    await idleProject.acquire(PRIORITY_EXECUTE);
+    await idleProject.acquire(PRIORITY_EXECUTE);
+    await activeProject.acquire(PRIORITY_MERGE);
+
+    const result = idleProject.reconcileActiveCount(0);
+
+    expect(result).toEqual({ before: 2, after: 0, changed: true });
+    expect(idleProject.heldCount).toBe(0);
+    expect(activeProject.heldCount).toBe(1);
+    expect(shared.activeCount).toBe(1);
+    expect(shared.availableCount).toBe(2);
+
+    activeProject.release();
+    expect(shared.activeCount).toBe(0);
+  });
+});
+
 describe("AgentSemaphore", () => {
   it("allows immediate acquire when under limit", async () => {
     const sem = new AgentSemaphore(2);

packages/engine/src/__tests__/project-engine-manager.test.ts

@@ -42,6 +42,7 @@ import {
   EngineAlreadyRunningError,
 } from "../engine-singleton-lock.js";
 import type { RegisteredProject, CentralCore } from "@fusion/core";
+import { ScopedAgentSemaphore } from "../concurrency.js";
 
 function createMockCentralCore(projects: RegisteredProject[]): CentralCore {
   const projectMap = new Map(projects.map((p) => [p.id, p]));
@@ -267,6 +268,34 @@ describe("ProjectEngineManager", () => {
       expect(manager.getAllEngines().size).toBe(0);
     });
 
+    it("stopAll frees residual slots from each stopped project scope", async () => {
+      const manager = new ProjectEngineManager(centralCore);
+      await manager.startAll();
+      const engineA = manager.getEngine("proj_aaa")!;
+      const engineB = manager.getEngine("proj_bbb")!;
+      const sharedSemaphore = (manager as any).globalSemaphore;
+      const scopeA = new ScopedAgentSemaphore(sharedSemaphore);
+      const scopeB = new ScopedAgentSemaphore(sharedSemaphore);
+
+      await scopeA.acquire();
+      await scopeB.acquire();
+      expect(sharedSemaphore.activeCount).toBe(2);
+
+      (engineA.stop as ReturnType<typeof vi.fn>).mockImplementationOnce(async () => {
+        scopeA.returnAllHeldSlots();
+      });
+      (engineB.stop as ReturnType<typeof vi.fn>).mockImplementationOnce(async () => {
+        scopeB.returnAllHeldSlots();
+      });
+
+      await manager.stopAll();
+
+      expect(scopeA.heldCount).toBe(0);
+      expect(scopeB.heldCount).toBe(0);
+      expect(sharedSemaphore.activeCount).toBe(0);
+      expect(sharedSemaphore.availableCount).toBe(4);
+    });
+
     it("handles stop errors gracefully", async () => {
       const manager = new ProjectEngineManager(centralCore);
       await manager.startAll();
@@ -337,6 +366,37 @@ describe("ProjectEngineManager", () => {
       expect(manager.getEngine("proj_aaa")).toBeUndefined();
     });
 
+    it("frees only the paused project's residual shared semaphore slots", async () => {
+      const manager = new ProjectEngineManager(centralCore);
+      const engineA = await manager.ensureEngine("proj_aaa");
+      await manager.ensureEngine("proj_bbb");
+      const sharedSemaphore = (manager as any).globalSemaphore;
+      const scopeA = new ScopedAgentSemaphore(sharedSemaphore);
+      const scopeB = new ScopedAgentSemaphore(sharedSemaphore);
+
+      await scopeA.acquire();
+      await scopeA.acquire();
+      await scopeB.acquire();
+      expect(sharedSemaphore.activeCount).toBe(3);
+      expect(sharedSemaphore.availableCount).toBe(1);
+
+      (engineA.stop as ReturnType<typeof vi.fn>).mockImplementationOnce(async () => {
+        scopeA.returnAllHeldSlots();
+      });
+
+      await manager.pauseProject("proj_aaa");
+
+      expect(scopeA.heldCount).toBe(0);
+      expect(scopeB.heldCount).toBe(1);
+      expect(sharedSemaphore.activeCount).toBe(1);
+      expect(sharedSemaphore.availableCount).toBe(3);
+      expect(scopeB.tryAcquire()).toBe(true);
+      expect(scopeB.tryAcquire()).toBe(true);
+      expect(scopeB.tryAcquire()).toBe(true);
+      expect(scopeB.tryAcquire()).toBe(false);
+      scopeB.returnAllHeldSlots();
+    });
+
     it("removes from starting set to prevent stalled starts from completing", async () => {
       const manager = new ProjectEngineManager(centralCore);
 

packages/engine/src/concurrency.ts

@@ -261,14 +261,24 @@ export class AgentSemaphore {
    * deadlock that would occur if both parent and child needed a queued slot.
    */
   async runNested<T>(fn: () => Promise<T>): Promise<T> {
-    this._active++;
+    this.acquireNestedSlot();
     try {
       return await fn();
     } finally {
-      this.returnSlot("runNested");
+      this.releaseNestedSlot();
     }
   }
 
+  /** Reserve a nested helper-agent slot without queueing. */
+  acquireNestedSlot(): void {
+    this._active++;
+  }
+
+  /** Return a nested helper-agent slot. */
+  releaseNestedSlot(): void {
+    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.
@@ -320,3 +330,112 @@ export class AgentSemaphore {
     return bestIdx;
   }
 }
+
+/**
+ * FNXC:Scheduler-Concurrency 2026-06-27-19:50:
+ * Project engines share one global AgentSemaphore, so each runtime needs scope-local slot accounting. When a project stops or pauses after abort+drain, return only that project's residual held slots to the shared pool so other projects regain capacity without releasing slots held by still-running projects.
+ */
+export class ScopedAgentSemaphore extends AgentSemaphore {
+  private readonly delegate: AgentSemaphore;
+  private _held = 0;
+
+  constructor(delegate: AgentSemaphore) {
+    super(() => delegate.limit);
+    this.delegate = delegate;
+  }
+
+  /** Number of slots this scope currently owns in the delegated semaphore. */
+  get heldCount(): number {
+    return this._held;
+  }
+
+  /** Shared-pool active count, preserving scheduler/metrics semantics. */
+  override get activeCount(): number {
+    return this.delegate.activeCount;
+  }
+
+  override get waitingCount(): number {
+    return this.delegate.waitingCount;
+  }
+
+  override get availableCount(): number {
+    return this.delegate.availableCount;
+  }
+
+  override get limit(): number {
+    return this.delegate.limit;
+  }
+
+  override snapshot(): { activeCount: number; waitingCount: number; availableCount: number; limit: number } {
+    return this.delegate.snapshot();
+  }
+
+  override reconcileActiveCount(maxActive: number): { before: number; after: number; changed: boolean } {
+    const bounded = Math.max(0, Math.floor(maxActive));
+    const before = this._held;
+    if (before > bounded) {
+      const returned = before - bounded;
+      this._held = bounded;
+      for (let i = 0; i < returned; i++) {
+        this.delegate.release();
+      }
+    }
+    return { before, after: this._held, changed: before !== this._held };
+  }
+
+  override async acquire(priority: number = 0): Promise<void> {
+    await this.delegate.acquire(priority);
+    this._held++;
+  }
+
+  override tryAcquire(): boolean {
+    const acquired = this.delegate.tryAcquire();
+    if (acquired) this._held++;
+    return acquired;
+  }
+
+  override release(): void {
+    if (this._held <= 0) return;
+    this._held--;
+    this.delegate.release();
+  }
+
+  override async run<T>(fn: () => Promise<T>, priority: number = 0): Promise<T> {
+    await this.acquire(priority);
+    try {
+      return await fn();
+    } finally {
+      this.release();
+    }
+  }
+
+  override async runNested<T>(fn: () => Promise<T>): Promise<T> {
+    this.delegate.acquireNestedSlot();
+    this._held++;
+    try {
+      return await fn();
+    } finally {
+      if (this._held > 0) {
+        this._held--;
+        this.delegate.releaseNestedSlot();
+      }
+    }
+  }
+
+  /**
+   * Return every slot still attributed to this scope.
+   *
+   * Late normal releases from already-aborted agents become no-ops because the
+   * scope's held count is zeroed before returning residual slots to the pool.
+   */
+  returnAllHeldSlots(): number {
+    const residual = this._held;
+    if (residual <= 0) return 0;
+
+    this._held = 0;
+    for (let i = 0; i < residual; i++) {
+      this.delegate.release();
+    }
+    return residual;
+  }
+}

packages/engine/src/project-engine.ts

@@ -2346,7 +2346,7 @@ export class ProjectEngine {
             // Direct merge via AI agent, gated by semaphore
             runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge merging ${taskId}...`);
 
-            const semaphore = (this.runtime as any).globalSemaphore;
+            const semaphore = (this.runtime as any).projectSemaphore ?? (this.runtime as any).globalSemaphore;
 
             const pool = (this.runtime as any).worktreePool;
 

packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts

@@ -8,6 +8,7 @@ import type { Task, TaskStore, CentralCore, AgentStore, Agent } from "@fusion/co
 import { InProcessRuntime } from "../in-process-runtime.js";
 import type { ProjectRuntimeConfig } from "../../project-runtime.js";
 import { runtimeLog } from "../../logger.js";
+import { AgentSemaphore } from "../../concurrency.js";
 
 const {
   mockSelfHealingStart,
@@ -538,6 +539,56 @@ describe("InProcessRuntime", () => {
       expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("post-abort drain timeout"));
     }, 30000);
 
+    it("returns residual scoped semaphore slots after the post-abort drain", async () => {
+      const sharedSemaphore = new AgentSemaphore(2);
+      runtime = new InProcessRuntime(
+        { ...buildTestConfig(testDir), globalSemaphore: sharedSemaphore },
+        mockCentralCore,
+      );
+      await runtime.start();
+
+      const projectSemaphore = (runtime as any).projectSemaphore;
+      await projectSemaphore.acquire();
+      await projectSemaphore.acquire();
+      expect(projectSemaphore.heldCount).toBe(2);
+      expect(sharedSemaphore.availableCount).toBe(0);
+
+      await runtime.stop();
+
+      expect(projectSemaphore.heldCount).toBe(0);
+      expect(sharedSemaphore.activeCount).toBe(0);
+      expect(sharedSemaphore.availableCount).toBe(2);
+      await runtime.stop();
+      expect(sharedSemaphore.activeCount).toBe(0);
+    }, 30000);
+
+    it("returns residual slots in single-project local-semaphore mode without double-return warnings", async () => {
+      const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+      try {
+        await runtime.start();
+        const localSemaphore = (runtime as any).globalSemaphore;
+        const projectSemaphore = (runtime as any).projectSemaphore;
+
+        await projectSemaphore.acquire();
+        await projectSemaphore.acquire();
+        expect(projectSemaphore.heldCount).toBe(2);
+        expect(localSemaphore.activeCount).toBe(2);
+        expect(localSemaphore.availableCount).toBe(2);
+
+        await runtime.stop();
+        await runtime.stop();
+
+        expect(projectSemaphore.heldCount).toBe(0);
+        expect(localSemaphore.activeCount).toBe(0);
+        expect(localSemaphore.availableCount).toBe(4);
+        expect(warnSpy).not.toHaveBeenCalledWith(
+          expect.stringContaining("AgentSemaphore excess slot return ignored"),
+        );
+      } finally {
+        warnSpy.mockRestore();
+      }
+    }, 30000);
+
     it("continues stopping when abortAllInFlight throws", async () => {
       await runtime.start();
       const executor = (runtime as any).executor;

packages/engine/src/runtimes/in-process-runtime.ts

@@ -25,7 +25,7 @@ import { buildPrNodeDeps } from "../pr-nodes.js";
 import { isExperimentalFeatureEnabled } from "@fusion/core";
 import { createCliAgentRuntime, type BootstrappedCliAgentRuntime } from "../cli-agent/runtime.js";
 import { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js";
-import { AgentSemaphore } from "../concurrency.js";
+import { AgentSemaphore, ScopedAgentSemaphore } from "../concurrency.js";
 import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
 import { AutoClaimSnapshotManager } from "../auto-claim-snapshot.js";
 import { RoutineRunner, type RoutineRunnerOptions } from "../routine-runner.js";
@@ -178,6 +178,7 @@ export class InProcessRuntime
   private executor!: TaskExecutor;
   private worktreePool!: WorktreePool;
   private globalSemaphore?: AgentSemaphore;
+  private projectSemaphore?: ScopedAgentSemaphore;
   private stuckTaskDetector?: StuckTaskDetector;
   /**
    * Per-project CLI Agent Executor runtime bundle (PTY manager + telemetry hub +
@@ -378,6 +379,8 @@ export class InProcessRuntime
         }
       }
 
+      this.projectSemaphore = new ScopedAgentSemaphore(this.globalSemaphore);
+
       await yieldEventLoop();
 
       // 5a. Initialize AgentStore (required for scheduler assignment, reflection service, and heartbeat monitoring)
@@ -457,7 +460,7 @@ export class InProcessRuntime
       this.scheduler = new Scheduler(this.taskStore, {
         maxConcurrent: this.config.maxConcurrent,
         maxWorktrees: this.config.maxWorktrees,
-        semaphore: this.globalSemaphore,
+        semaphore: this.projectSemaphore,
         agentStore: this.agentStore,
         hasActiveAgentExecution: (agentId: string) => this.heartbeatMonitor?.getTrackedAgents().includes(agentId) ?? false,
         missionStore,
@@ -572,7 +575,7 @@ export class InProcessRuntime
       const prNodeGithubOps = this.config.prNodeGithubOps;
       const workflowAuthoritativeDriverRef: { current?: WorkflowAuthoritativeDriver } = {};
       const executorOptions: TaskExecutorOptions = {
-        semaphore: this.globalSemaphore,
+        semaphore: this.projectSemaphore,
         pool: this.worktreePool,
         usageLimitPauser: this.usageLimitPauser,
         stuckTaskDetector: this.stuckTaskDetector,
@@ -812,7 +815,7 @@ export class InProcessRuntime
         this.taskStore,
         this.config.workingDirectory,
         {
-          semaphore: this.globalSemaphore,
+          semaphore: this.projectSemaphore,
           stuckTaskDetector: this.stuckTaskDetector,
           agentStore: this.agentStore,
           pluginRunner: this.pluginRunner,
@@ -1193,6 +1196,17 @@ export class InProcessRuntime
         );
       }
 
+      /**
+       * FNXC:Scheduler-Concurrency 2026-06-27-20:05:
+       * After stop aborts a project's agents and waits the bounded drain window, any slots still attributed to this runtime are residual leaks from sessions that did not settle their normal finally path. Return only this project's scoped slots so pauseProject/stopAll promptly free shared global capacity without clobbering other projects' active slots.
+       */
+      const returnedResidualSlots = this.projectSemaphore?.returnAllHeldSlots() ?? 0;
+      if (returnedResidualSlots > 0) {
+        runtimeLog.warn(
+          `Returned ${returnedResidualSlots} residual global concurrency slot(s) after project stop drain`,
+        );
+      }
+
       // 8. Shutdown plugin runner
       if (this.pluginRunner) {
         await this.pluginRunner.shutdown();