Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 66 additions & 5 deletions open-sse/services/combo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3298,6 +3298,15 @@ export async function handleComboChat({
// Reset each retry so providers excluded in a previous attempt get another chance.
const exhaustedProviders = new Set<string>();
const transientRateLimitedProviders = new Set<string>();
// #3200: Track consecutive failures per provider across targets to break infinite
// loops where a single failing provider exhausts all its targets before the combo
// moves to a different provider. Once a provider fails >= threshold times in a row,
// all remaining targets from that provider are skipped for this set iteration.
const consecutiveProviderFailures = new Map<string, number>();
const providerFailureThreshold =
(config as Record<string, unknown>).providerFailureThreshold
? Number((config as Record<string, unknown>).providerFailureThreshold) || 2
: 2;
if (setTry > 0) {
log.info("COMBO", `All targets failed — retrying set (${setTry}/${maxSetRetries})`);
await new Promise((resolve) => {
Expand Down Expand Up @@ -3360,6 +3369,12 @@ export async function handleComboChat({
return null;
}

if (provider && (consecutiveProviderFailures.get(provider) ?? 0) >= providerFailureThreshold) {
log.info("COMBO", `Skipping ${modelStr} — provider ${provider} hit consecutive failure threshold (#3200)`);
if (i > 0) fallbackCount++;
return null;
}

// Pre-check: skip models where no credentials are available (excluded, rate-limited, or unavailable)
if (isModelAvailable) {
const available = await isModelAvailable(modelStr, targetForAttempt);
Expand Down Expand Up @@ -3537,6 +3552,10 @@ export async function handleComboChat({
});
return null;
}
// #3200: Reset consecutive failure counter for this provider on success.
if (provider && provider !== "unknown") {
consecutiveProviderFailures.delete(provider);
}
const latencyMs = Date.now() - startTime;
emit("combo.target.succeeded", {
comboName: combo.name,
Expand Down Expand Up @@ -3827,22 +3846,32 @@ export async function handleComboChat({
if (!lastStatus) lastStatus = result.status;
if (i > 0) fallbackCount++;
log.warn("COMBO", `Model ${modelStr} failed with body-specific error, stopping combo`);
break; // Break out of the target loop to avoid trying other models
if (provider && provider !== "unknown") {
const prev = consecutiveProviderFailures.get(provider) ?? 0;
const next = prev + 1;
consecutiveProviderFailures.set(provider, next);
if (next >= providerFailureThreshold && !exhaustedProviders.has(provider)) {
exhaustedProviders.add(provider);
log.info("COMBO", `Provider ${provider} consecutive failure threshold reached (${next}/${providerFailureThreshold}) — skipping remaining targets (#3200)`);
}
}
break;
}

// Trigger shared provider circuit breaker for 5xx errors and connection failures.
// If the next target in the combo is on the same provider, don't mark the provider
// as failed — different models on the same provider may still succeed.
// G-02: when fallbackResult.skipProviderBreaker is set (embedded service supervisor
// outage signalled via X-Omni-Fallback-Hint: connection_cooldown) apply connection
// cooldown only — do NOT trip the whole-provider breaker.
const nextTarget = orderedTargets[i + 1];
const sameProviderNext =
typeof nextTarget?.provider === "string" && nextTarget.provider === provider;
const providerHitThreshold =
provider &&
provider !== "unknown" &&
(consecutiveProviderFailures.get(provider) ?? 0) >= providerFailureThreshold;
if (
!isStreamReadinessFailure &&
isProviderFailureCode(result.status) &&
!sameProviderNext &&
(!sameProviderNext || providerHitThreshold) &&
!fallbackResult.skipProviderBreaker
) {
recordProviderFailure(provider, log, target.connectionId, profile);
Expand Down Expand Up @@ -3872,6 +3901,16 @@ export async function handleComboChat({
if (i > 0) fallbackCount++;
log.warn("COMBO", `Model ${modelStr} failed, trying next`, { status: result.status });

if (provider && provider !== "unknown") {
const prev = consecutiveProviderFailures.get(provider) ?? 0;
const next = prev + 1;
consecutiveProviderFailures.set(provider, next);
if (next >= providerFailureThreshold && !exhaustedProviders.has(provider)) {
exhaustedProviders.add(provider);
log.info("COMBO", `Provider ${provider} consecutive failure threshold reached (${next}/${providerFailureThreshold}) — skipping remaining targets (#3200)`);
}
}

const fallbackWaitMs =
fallbackDelayMs > 0 && cooldownMs > 0 && cooldownMs <= MAX_FALLBACK_WAIT_MS
? Math.min(cooldownMs, fallbackDelayMs)
Expand Down Expand Up @@ -4086,6 +4125,11 @@ async function handleRoundRobinCombo({
// provider are skipped to avoid the cascade through N same-provider targets.
const exhaustedProviders = new Set<string>();
const transientRateLimitedProviders = new Set<string>();
const consecutiveProviderFailures = new Map<string, number>();
const providerFailureThreshold =
(config as Record<string, unknown>).providerFailureThreshold
? Number((config as Record<string, unknown>).providerFailureThreshold) || 2
: 2;

// Try each model starting from the round-robin target
for (let offset = 0; offset < modelCount; offset++) {
Expand Down Expand Up @@ -4122,6 +4166,13 @@ async function handleRoundRobinCombo({
continue;
}

const rrConsecutiveFails = consecutiveProviderFailures.get(provider) ?? 0;
if (provider && rrConsecutiveFails >= providerFailureThreshold) {
log.info("COMBO-RR", `Skipping ${modelStr} — provider ${provider} consecutive failure threshold (${rrConsecutiveFails}/${providerFailureThreshold}) (#3200)`);
if (offset > 0) fallbackCount++;
continue;
}

// Acquire semaphore slot (may wait in queue)
let release: () => void;
try {
Expand Down Expand Up @@ -4398,6 +4449,16 @@ async function handleRoundRobinCombo({
if (offset > 0) fallbackCount++;
log.warn("COMBO-RR", `${modelStr} failed, trying next model`, { status: result.status });

if (provider && provider !== "unknown") {
const prev = consecutiveProviderFailures.get(provider) ?? 0;
const next = prev + 1;
consecutiveProviderFailures.set(provider, next);
if (next >= providerFailureThreshold && !exhaustedProviders.has(provider)) {
exhaustedProviders.add(provider);
log.info("COMBO-RR", `Provider ${provider} consecutive failure threshold reached (${next}/${providerFailureThreshold}) — skipping remaining targets (#3200)`);
}
}

const fallbackWaitMs =
fallbackDelayMs > 0 && cooldownMs > 0 && cooldownMs <= MAX_FALLBACK_WAIT_MS
? Math.min(cooldownMs, fallbackDelayMs)
Expand Down
158 changes: 158 additions & 0 deletions tests/integration/combo-consecutive-provider-failures-3145.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
// #3145 — Consecutive same-provider failures should skip remaining targets
// When a combo has multiple targets from the same provider and they fail
// consecutively (even with 429 which is NOT in PROVIDER_FAILURE_ERROR_CODES),
// the combo should skip remaining targets from that provider to prevent loops.
//
// Uses two different providers (opencode-zen, opencode-go) in separate tests
// with distinct API key values so there is zero data overlap.
import test from "node:test";
import assert from "node:assert/strict";

import { createChatPipelineHarness } from "../integration/_chatPipelineHarness.ts";

const harness = await createChatPipelineHarness("consecutive-fail-3145");
const {
buildClaudeResponse,
buildRequest,
combosDb,
handleChat,
resetStorage,
seedConnection,
settingsDb,
} = harness;

function toPlain(obj: any): Record<string, string> {
if (!obj) return {};
if (obj instanceof Headers) return Object.fromEntries(obj.entries());
return Object.fromEntries(
Object.entries(obj).map(([k, v]) => [k, v == null ? "" : String(v)])
);
}

test.beforeEach(async () => {
await resetStorage();
});

test.afterEach(async () => {
await resetStorage();
});

test.after(async () => {
await harness.cleanup();
});

/* ── Test 1: opencode-zen → 429 → fallback to anthropic ── */
test("test1: 429 on opencode-zen falls back to anthropic", async () => {
await seedConnection("opencode-zen", { apiKey: "sk-zen-alpha" });
await seedConnection("opencode-zen", { apiKey: "sk-zen-beta" });
await seedConnection("anthropic", { apiKey: "sk-anthro-001" });

await settingsDb.updateSettings({
requestRetry: 0,
maxRetryIntervalSec: 0,
});

await combosDb.createCombo({
name: "c1",
strategy: "priority",
config: { maxRetries: 0, retryDelayMs: 0, fallbackDelayMs: 0 },
models: [
"opencode-zen/m-a",
"opencode-zen/m-b",
"anthropic/claude-sonnet",
],
});

let zen = 0;
let anthro = 0;

globalThis.fetch = async (_url: string, init: any = {}) => {
const h = toPlain(init.headers);
const key = h["x-api-key"];
if (key === "sk-zen-alpha" || key === "sk-zen-beta") {
zen++;
return new Response(
JSON.stringify({ error: { message: "no quota" } }),
{ status: 429, headers: { "Content-Type": "application/json" } }
);
}
if (key === "sk-anthro-001") {
anthro++;
return buildClaudeResponse("ok from anthro");
}
throw new Error(`U1 key=${key} hdrs=${JSON.stringify(h)}`);
};

const res = await handleChat(
buildRequest({
body: {
model: "c1",
stream: false,
messages: [{ role: "user", content: "hi" }],
},
})
);
assert.equal(res.status, 200, "should 200");
const body: any = await res.json();
assert.equal(body.choices[0].message.content, "ok from anthro");
assert.ok(zen <= 2, `zen calls=${zen}, expected ≤2`);
assert.equal(anthro, 1, "anthropic fallback exact once");
});

/* ── Test 2: opencode-go → 502 → falls back to anthro ── */
test("test2: 502 on opencode-go falls back to anthropic", async () => {
await seedConnection("opencode-go", { apiKey: "sk-go-a1" });
await seedConnection("opencode-go", { apiKey: "sk-go-a2" });
await seedConnection("anthropic", { apiKey: "sk-anthro-002" });

await settingsDb.updateSettings({
requestRetry: 0,
maxRetryIntervalSec: 0,
});

await combosDb.createCombo({
name: "c2",
strategy: "priority",
config: { maxRetries: 0, retryDelayMs: 0, fallbackDelayMs: 0 },
models: [
"opencode-go/m-x",
"opencode-go/m-y",
"anthropic/claude-sonnet",
],
});

let go = 0;
let anthro = 0;

globalThis.fetch = async (_url: string, init: any = {}) => {
const h = toPlain(init.headers);
const key = h["x-api-key"];
if (key === "sk-go-a1" || key === "sk-go-a2") {
go++;
return new Response(
JSON.stringify({ error: { message: "fetch failed" } }),
{ status: 502, headers: { "Content-Type": "application/json" } }
);
}
if (key === "sk-anthro-002") {
anthro++;
return buildClaudeResponse("ok anthro 2");
}
throw new Error(`U2 key=${key} hdrs=${JSON.stringify(h)}`);
};

const res = await handleChat(
buildRequest({
body: {
model: "c2",
stream: false,
messages: [{ role: "user", content: "hi" }],
},
})
);
assert.equal(res.status, 200, "should 200");
const body: any = await res.json();
assert.equal(body.choices[0].message.content, "ok anthro 2");
assert.ok(go <= 2, `go calls=${go}, expected ≤2`);
assert.equal(anthro, 1, "anthropic fallback exact once");
});
Loading