fix(combo): fail-fast concurrency gate and execute-mode overflow (#8890)

Validated in post-merge-train sweep (boards clean on release/v3.8.50 tip)

open-sse/services/__tests__/fail-fast-concurrency-gate.test.ts

@@ -0,0 +1,30 @@
+import { describe, it, expect, beforeEach } from "vitest";
+import { acquire, isAccountSemaphoreFull, resetAll } from "../accountSemaphore.ts";
+
+describe("isAccountSemaphoreFull fail-fast concurrency gate", () => {
+  beforeEach(() => {
+    resetAll();
+  });
+
+  it("returns false when no semaphore gate exists", () => {
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(false);
+  });
+
+  it("returns false when maxConcurrency is null, <= 0, or bypassed", () => {
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", null)).toBe(false);
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 0)).toBe(false);
+  });
+
+  it("returns false when running < maxConcurrency", async () => {
+    const release = await acquire("featherless-ai:conn-1", { maxConcurrency: 2 });
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 2)).toBe(false);
+    release();
+  });
+
+  it("returns true immediately when running >= maxConcurrency", async () => {
+    const release = await acquire("featherless-ai:conn-1", { maxConcurrency: 1 });
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(true);
+    release();
+    expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(false);
+  });
+});

open-sse/services/accountFallback.ts

@@ -1631,7 +1631,11 @@ export function checkFallbackError(
       !errorStr.toLowerCase().includes("hour quota") &&
       !errorStr.toLowerCase().includes("quota has been exceeded")
     ) {
-      return resolveApiKeyForbiddenFallback(errorStr, buildRetryableFallback, RateLimitReason.AUTH_ERROR);
+      return resolveApiKeyForbiddenFallback(
+        errorStr,
+        buildRetryableFallback,
+        RateLimitReason.AUTH_ERROR
+      );
     }
   }
 
@@ -1953,6 +1957,8 @@ export function applyErrorState<T extends AccountState | null | undefined>(
   return nextState;
 }
 
+export { isAccountSemaphoreFull } from "./accountSemaphore.ts";
+
 /**
  * Get account health score (0-100) for P2C selection (Phase 9)
  * @param {object} account

open-sse/services/accountSemaphore.ts

@@ -342,6 +342,24 @@ export function getStats(): Record<string, AccountSemaphoreStatsEntry> {
   return stats;
 }
 
+/**
+ * Check if an account semaphore key is currently at or over its max concurrency limit.
+ * Returns true if running >= maxConcurrency or blocked.
+ */
+export function isAccountSemaphoreFull(
+  provider: string,
+  accountKey: string,
+  maxConcurrency?: number | null
+): boolean {
+  if (isBypassed(maxConcurrency)) return false;
+  const key = buildAccountSemaphoreKey({ provider, accountKey });
+  const gate = gates.get(key);
+  if (!gate) return false;
+  const effectiveCap = maxConcurrency ?? gate.maxConcurrency;
+  if (isBypassed(effectiveCap)) return false;
+  return gate.running >= effectiveCap || isBlocked(gate);
+}
+
 /**
  * Reset a single key and reject queued waiters.
  */

open-sse/services/combo.ts

@@ -14,6 +14,7 @@ import {
   getModelLockoutInfo,
   getRuntimeProviderProfile,
   hasPerModelQuota,
+  isAccountSemaphoreFull,
   isModelLocked,
   MODEL_ACCESS_DENIED_PATTERNS,
   recordModelLockoutFailure,
@@ -1012,6 +1013,20 @@ export async function handleComboChat({
             if (i > 0) fallbackCount++;
             return null;
           }
+
+          // Concurrency gate: fail-fast skip when connection is at max_concurrent capacity (e.g. Featherless 1/1)
+          const maxConcurrentCap = await lookupPositiveCap(connectionId);
+          if (
+            maxConcurrentCap &&
+            isAccountSemaphoreFull(provider, connectionId, maxConcurrentCap)
+          ) {
+            log.info(
+              "COMBO",
+              `Skipping ${modelStr} — connection ${connectionId} is at max concurrency cap (${maxConcurrentCap})`
+            );
+            if (i > 0) fallbackCount++;
+            return null;
+          }
         }
 
         // Retry loop for transient errors

open-sse/services/combo/runtimeUnitCapacity.ts

@@ -0,0 +1,80 @@
+/**
+ * @file runtimeUnitCapacity.ts
+ * @description Concurrency-capacity checks for nested combo execute-mode units so
+ *   ordered strategies overflow to the next slot instead of queueing on a full connection.
+ *
+ * @changes
+ * - [2026-07-24] [Composer] - Initial capacity pre-check for execute-mode runtime units
+ */
+import { isAccountSemaphoreFull } from "../accountSemaphore.ts";
+import { resolveComboTargets } from "./comboStructure.ts";
+import { lookupPositiveCap } from "./concurrencyCaps.ts";
+import type { ComboCollectionLike, ComboLike, ResolvedComboUnit } from "./types.ts";
+
+type CapLookup = (connectionId: string) => Promise<number | null>;
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return !!value && typeof value === "object" && !Array.isArray(value);
+}
+
+function getCombosList(allCombos: ComboCollectionLike): ComboLike[] {
+  const combos = Array.isArray(allCombos) ? allCombos : allCombos?.combos || [];
+  return combos.filter(
+    (combo): combo is ComboLike => isRecord(combo) && typeof combo.name === "string"
+  );
+}
+
+function findComboByName(allCombos: ComboCollectionLike, name: string): ComboLike | null {
+  return getCombosList(allCombos).find((combo) => combo.name === name) || null;
+}
+
+async function isConnectionAtConcurrencyCap(
+  provider: string,
+  connectionId: string,
+  lookupCap: CapLookup
+): Promise<boolean> {
+  const cap = await lookupCap(connectionId);
+  if (!cap) return false;
+  return isAccountSemaphoreFull(provider, connectionId, cap);
+}
+
+/**
+ * Returns true when the runtime unit should be skipped because every limited
+ * connection it would use is already at max_concurrent.
+ */
+export async function isRuntimeUnitAtConcurrencyCap(
+  unit: ResolvedComboUnit,
+  allCombos: ComboCollectionLike,
+  lookupCap: CapLookup = lookupPositiveCap
+): Promise<boolean> {
+  if (unit.kind === "model") {
+    if (!unit.connectionId || !unit.provider) return false;
+    return isConnectionAtConcurrencyCap(unit.provider, unit.connectionId, lookupCap);
+  }
+
+  const childCombo = findComboByName(allCombos, unit.comboName);
+  if (!childCombo) return false;
+
+  const targets = resolveComboTargets(childCombo, allCombos, 1);
+  const byConnection = new Map<string, { provider: string; connectionId: string }>();
+  for (const target of targets) {
+    if (!target.connectionId || !target.provider) continue;
+    byConnection.set(target.connectionId, {
+      provider: target.provider,
+      connectionId: target.connectionId,
+    });
+  }
+  if (byConnection.size === 0) return false;
+
+  let sawLimitedConnection = false;
+  for (const { provider, connectionId } of byConnection.values()) {
+    const cap = await lookupCap(connectionId);
+    if (!cap) continue;
+    sawLimitedConnection = true;
+    if (!isAccountSemaphoreFull(provider, connectionId, cap)) {
+      return false;
+    }
+  }
+
+  return sawLimitedConnection;
+}

open-sse/services/combo/runtimeUnits.ts

@@ -1,7 +1,14 @@
-// Nested combo runtime unit execution — see combo.ts for integration.
+/**
+ * @file runtimeUnits.ts
+ * @description Nested combo runtime unit execution — see combo.ts for integration.
+ *
+ * @changes
+ * - [2026-07-24] [Composer] - Skip execute-mode units at concurrency cap before dispatch
+ */
 import { errorResponse } from "../../utils/error.ts";
 import { recordComboRequest } from "../comboMetrics.ts";
 import { resolveDelayMs } from "./comboPredicates.ts";
+import { isRuntimeUnitAtConcurrencyCap } from "./runtimeUnitCapacity.ts";
 import { validateResponseQuality, releaseQualityClone } from "./validateQuality.ts";
 import type { ResponseValidationConfig } from "./responseValidation.ts";
 import type {
@@ -190,6 +197,15 @@ export async function executeRuntimeUnitCombo(args: {
   let fallbackCount = 0;
 
   for (const unit of orderedUnits) {
+    if (await isRuntimeUnitAtConcurrencyCap(unit, args.allCombos)) {
+      args.log.info(
+        "COMBO",
+        `Skipping ${unit.kind} ${unitDisplayName(unit)} — concurrency cap reached`
+      );
+      fallbackCount += 1;
+      continue;
+    }
+
     for (let retry = 0; retry <= maxRetries; retry += 1) {
       if (args.signal?.aborted)
         return { response: errorResponse(499, "Client disconnected"), unit };

tests/unit/combo-runtime-unit-concurrency.test.ts

@@ -0,0 +1,145 @@
+/**
+ * @file combo-runtime-unit-concurrency.test.ts
+ * @description Regression tests for execute-mode concurrency overflow (skip full units).
+ *
+ * @changes
+ * - [2026-07-24] [Composer] - Initial execute-mode capacity overflow coverage
+ */
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import {
+  acquire,
+  buildAccountSemaphoreKey,
+  resetAll,
+} from "../../open-sse/services/accountSemaphore.ts";
+import { isRuntimeUnitAtConcurrencyCap } from "../../open-sse/services/combo/runtimeUnitCapacity.ts";
+import type { ResolvedComboUnit } from "../../open-sse/services/combo/types.ts";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-runtime-unit-cap-"));
+const ORIGINAL_DATA_DIR = process.env.DATA_DIR;
+process.env.DATA_DIR = TEST_DATA_DIR;
+
+const { handleComboChat } = await import("../../open-sse/services/combo.ts");
+const { getDbInstance, resetDbInstance } = await import("../../src/lib/db/core.ts");
+
+function createLog() {
+  return {
+    info: () => {},
+    warn: () => {},
+    error: () => {},
+    debug: () => {},
+  };
+}
+
+function okResponse() {
+  return new Response(JSON.stringify({ choices: [{ message: { content: "ok" } }] }), {
+    status: 200,
+    headers: { "content-type": "application/json" },
+  });
+}
+
+function seedConnection(id: string, provider: string, maxConcurrent: number) {
+  const db = getDbInstance();
+  db.prepare(
+    `INSERT INTO provider_connections
+      (id, provider, auth_type, is_active, max_concurrent, created_at, updated_at)
+     VALUES (?, ?, 'apikey', 1, ?, datetime('now'), datetime('now'))`
+  ).run(id, provider, maxConcurrent);
+}
+
+test.after(() => {
+  resetAll();
+  resetDbInstance();
+  if (ORIGINAL_DATA_DIR === undefined) delete process.env.DATA_DIR;
+  else process.env.DATA_DIR = ORIGINAL_DATA_DIR;
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+});
+
+test("isRuntimeUnitAtConcurrencyCap returns true for a model unit at cap", async () => {
+  const connectionId = "conn-feather-cap";
+  const provider = "featherless-ai";
+  const key = buildAccountSemaphoreKey({ provider, accountKey: connectionId });
+  const release1 = await acquire(key, { maxConcurrency: 2 });
+  const release2 = await acquire(key, { maxConcurrency: 2 });
+
+  const unit: ResolvedComboUnit = {
+    kind: "model",
+    stepId: "step-1",
+    executionKey: "step-1",
+    modelStr: "featherless-ai/deepseek-ai/DeepSeek-V4-Pro",
+    provider,
+    providerId: provider,
+    connectionId,
+    weight: 0,
+    label: null,
+  };
+
+  assert.equal(
+    await isRuntimeUnitAtConcurrencyCap(unit, [], async () => 2),
+    true,
+    "unit should be at cap when two slots are in use"
+  );
+
+  release1();
+  assert.equal(
+    await isRuntimeUnitAtConcurrencyCap(unit, [], async () => 2),
+    false,
+    "unit should have headroom after one slot frees"
+  );
+  release2();
+});
+
+test("execute fill-first overflows to the next unit when the first connection is at cap", async () => {
+  const primaryConnectionId = "conn-primary-overflow";
+  const backupConnectionId = "conn-backup-overflow";
+  const provider = "featherless-ai";
+  seedConnection(primaryConnectionId, provider, 2);
+  seedConnection(backupConnectionId, "alibaba", 2);
+
+  const key = buildAccountSemaphoreKey({ provider, accountKey: primaryConnectionId });
+  const release1 = await acquire(key, { maxConcurrency: 2 });
+  const release2 = await acquire(key, { maxConcurrency: 2 });
+
+  const calls: string[] = [];
+  const combo = {
+    name: "overflow-execute",
+    strategy: "fill-first",
+    models: [
+      {
+        kind: "model",
+        model: "featherless-ai/deepseek-ai/DeepSeek-V4-Pro",
+        providerId: provider,
+        connectionId: primaryConnectionId,
+      },
+      {
+        kind: "model",
+        model: "alibaba/qwen3.7-max-preview",
+        providerId: "alibaba",
+        connectionId: backupConnectionId,
+      },
+    ],
+    config: { nestedComboMode: "execute", maxRetries: 0, retryDelayMs: 0 },
+  };
+
+  const result = await handleComboChat({
+    body: {},
+    combo,
+    handleSingleModel: async (_body, modelStr) => {
+      calls.push(modelStr);
+      return okResponse();
+    },
+    isModelAvailable: async () => true,
+    log: createLog(),
+    settings: null,
+    allCombos: [combo],
+  });
+
+  assert.equal(result.ok, true);
+  assert.deepEqual(calls, ["alibaba/qwen3.7-max-preview"]);
+
+  release1();
+  release2();
+});