feat(provider-concurrency): add release cooldown (200ms) to prevent N+1 overshoot from provider accounting lag

3 files changed, 107 insertions, 3 deletions

diff --git a/packages/provider-concurrency/src/concurrency-manager.test.ts b/packages/provider-concurrency/src/concurrency-manager.test.ts
index bfefb4a..6583a34 100644
--- a/packages/provider-concurrency/src/concurrency-manager.test.ts
+++ b/packages/provider-concurrency/src/concurrency-manager.test.ts
@@ -59,7 +59,7 @@ function createFakeTimers() {
   };
 }
 
-function createManager(): {
+function createManager(opts?: { releaseCooldownMs?: number }): {
   manager: ConcurrencyService;
   timers: ReturnType<typeof createFakeTimers>;
 } {
@@ -69,6 +69,7 @@ function createManager(): {
     slotTimeoutMs: 5000,
     watchdogIntervalMs: 1000,
     defaultPauseMs: 30000,
+    ...(opts?.releaseCooldownMs !== undefined ? { releaseCooldownMs: opts.releaseCooldownMs } : {}),
     setTimeout: timers.setTimeout,
     clearTimeout: timers.clearTimeout,
     setInterval: timers.setInterval,
@@ -383,4 +384,67 @@ describe("createConcurrencyManager", () => {
     }
     expect(manager.getStatus("umans")?.inFlight).toBe(0);
   });
+
+  it("release cooldown delays slot recycling (inFlight stays incremented during cooldown)", async () => {
+    const { manager, timers } = createManager({ releaseCooldownMs: 200 });
+    manager.setLimit("umans", 1);
+
+    const release1 = await manager.acquire("umans", "conv1", 0);
+    expect(manager.getStatus("umans")?.inFlight).toBe(1);
+
+    // Queue a waiter.
+    let resolved = false;
+    const promise2 = manager.acquire("umans", "conv2", 100).then((r) => {
+      resolved = true;
+      return r;
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(resolved).toBe(false);
+    expect(manager.getStatus("umans")?.queued).toBe(1);
+
+    // Release the slot — inFlight should stay 1 (cooldown active).
+    release1();
+    expect(manager.getStatus("umans")?.inFlight).toBe(1);
+    expect(resolved).toBe(false); // waiter NOT granted yet
+
+    // Advance past the cooldown.
+    timers.advance(200);
+
+    // Now the slot is recycled and the waiter is granted.
+    const release2 = await promise2;
+    expect(resolved).toBe(true);
+    expect(manager.getStatus("umans")?.inFlight).toBe(1);
+    expect(manager.getStatus("umans")?.queued).toBe(0);
+    release2();
+  });
+
+  it("release cooldown is idempotent (double-release only schedules one cooldown)", async () => {
+    const { manager, timers } = createManager({ releaseCooldownMs: 200 });
+    manager.setLimit("umans", 2);
+
+    const release = await manager.acquire("umans", "conv1", 0);
+    expect(manager.getStatus("umans")?.inFlight).toBe(1);
+
+    release();
+    expect(manager.getStatus("umans")?.inFlight).toBe(1); // still 1 (cooldown)
+
+    // Double-release should not schedule a second cooldown.
+    release();
+
+    // After cooldown, inFlight should drop by exactly 1 (to 0), not 2.
+    timers.advance(200);
+    expect(manager.getStatus("umans")?.inFlight).toBe(0);
+  });
+
+  it("destroy clears cooldown timers without error", () => {
+    const { manager } = createManager({ releaseCooldownMs: 200 });
+    manager.setLimit("umans", 1);
+    // Acquire + release to schedule a cooldown timer.
+    manager.acquire("umans", "conv1", 0).then((release) => {
+      release();
+      // Now there's a pending cooldown timer — destroy should clean it up.
+      expect(() => manager.destroy()).not.toThrow();
+    });
+  });
 });
diff --git a/packages/provider-concurrency/src/concurrency-manager.ts b/packages/provider-concurrency/src/concurrency-manager.ts
index e7535cf..49575f7 100644
--- a/packages/provider-concurrency/src/concurrency-manager.ts
+++ b/packages/provider-concurrency/src/concurrency-manager.ts
@@ -114,6 +114,15 @@ export interface ConcurrencyManagerOpts {
   readonly watchdogIntervalMs: number;
   /** Default pause duration when a 429 arrives without Retry-After (ms). */
   readonly defaultPauseMs: number;
+  /**
+   * Delay after a slot is released before the slot is recycled (ms). During
+   * this window `inFlight` stays incremented — a new `acquire` sees the slot
+   * as still held and queues. This covers the upstream provider's accounting
+   * lag: the provider's `concurrent_sessions` counter may not decrement the
+   * instant our stream completes, so re-admitting immediately risks an N+1
+   * overshoot. 0 = instant re-admission (no cooldown). Default: 0.
+   */
+  readonly releaseCooldownMs?: number;
   /** Injected timers (default: global). Override in tests for deterministic time. */
   readonly setTimeout?: typeof setTimeout;
   readonly clearTimeout?: typeof clearTimeout;
@@ -132,12 +141,14 @@ export function createConcurrencyManager(opts: ConcurrencyManagerOpts): Concurre
   const now = opts.now;
   const slotTimeoutMs = opts.slotTimeoutMs;
   const defaultPauseMs = opts.defaultPauseMs;
+  const releaseCooldownMs = opts.releaseCooldownMs ?? 0;
   const setTimeout = opts.setTimeout ?? globalThis.setTimeout.bind(globalThis);
   const clearTimeout = opts.clearTimeout ?? globalThis.clearTimeout.bind(globalThis);
   const setInterval = opts.setInterval ?? globalThis.setInterval.bind(globalThis);
   const clearInterval = opts.clearInterval ?? globalThis.clearInterval.bind(globalThis);
 
   const states = new Map<string, ProviderState>();
+  const cooldownTimers = new Set<ReturnType<typeof setTimeout>>();
   let slotIdCounter = 0;
 
   // ── Slot granting ──────────────────────────────────────────────────────────
@@ -149,8 +160,25 @@ export function createConcurrencyManager(opts: ConcurrencyManagerOpts): Concurre
       if (released) return;
       released = true;
       state.slots.delete(id);
-      state.inFlight--;
-      tryGrantNext(providerId);
+
+      // Recycle the slot: decrement inFlight + grant the next waiter.
+      // With a release cooldown > 0, defer this by the cooldown duration so
+      // the upstream provider has time to decrement its concurrent_sessions
+      // counter — preventing an N+1 overshoot from accounting lag. During the
+      // cooldown, inFlight stays incremented, so new acquires queue.
+      const recycle = () => {
+        state.inFlight--;
+        tryGrantNext(providerId);
+      };
+      if (releaseCooldownMs > 0) {
+        const timer = setTimeout(() => {
+          cooldownTimers.delete(timer);
+          recycle();
+        }, releaseCooldownMs);
+        cooldownTimers.add(timer);
+      } else {
+        recycle();
+      }
     };
     state.slots.set(id, {
       conversationId,
@@ -314,6 +342,10 @@ export function createConcurrencyManager(opts: ConcurrencyManagerOpts): Concurre
 
     destroy() {
       clearInterval(watchdogTimer);
+      for (const timer of cooldownTimers) {
+        clearTimeout(timer);
+      }
+      cooldownTimers.clear();
       for (const state of states.values()) {
         if (state.pauseTimer !== undefined) {
           clearTimeout(state.pauseTimer);
diff --git a/packages/provider-concurrency/src/extension.ts b/packages/provider-concurrency/src/extension.ts
index c741173..fc2d578 100644
--- a/packages/provider-concurrency/src/extension.ts
+++ b/packages/provider-concurrency/src/extension.ts
@@ -23,10 +23,17 @@ export const manifest: Manifest = {
  *   Umans docs note each concurrency 429 deprioritizes the account for ~30 min,
  *   but a 30s queue pause prevents immediate re-overshoot while still allowing
  *   recovery.
+ * - `RELEASE_COOLDOWN_MS` (200ms): after a slot is released, hold it for this
+ *   duration before recycling it to the next waiter. Covers the upstream
+ *   provider's accounting lag — the provider's concurrent_sessions counter
+ *   may not decrement the instant our stream completes, so re-admitting
+ *   immediately risks an N+1 overshoot that triggers a 429. 200ms is the
+ *   default most concurrency proxies use for AI/LLM APIs.
  */
 const SLOT_TIMEOUT_MS = 5 * 60 * 1000;
 const WATCHDOG_INTERVAL_MS = 30 * 1000;
 const DEFAULT_PAUSE_MS = 30 * 1000;
+const RELEASE_COOLDOWN_MS = 200;
 
 export function activate(host: HostAPI): void {
   const logger = host.logger;
@@ -36,6 +43,7 @@ export function activate(host: HostAPI): void {
     slotTimeoutMs: SLOT_TIMEOUT_MS,
     watchdogIntervalMs: WATCHDOG_INTERVAL_MS,
     defaultPauseMs: DEFAULT_PAUSE_MS,
+    releaseCooldownMs: RELEASE_COOLDOWN_MS,
     onWatchdogReclaim: (providerId, conversationId, heldMs) => {
       logger.warn("provider-concurrency: watchdog reclaimed stale slot", {
         providerId,