pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-09-21 14:22:14 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(chatcore): restore provider failure classification and credential refresh on non-streaming leg (#13043) (#13069)

Browse Source 
This is the one that mattered most. The non-streaming leg returning error outcomes without classification meant 402/429 never reached model lockout or rate-limit bookkeeping, and a 401 on a refreshable provider failed instead of refreshing — on a leg that serves real traffic. Extracting `applyProviderFailureClassification` and wiring both legs through it is the right shape: the asymmetry was the bug, so the fix has to remove the asymmetry rather than patch one side.

---

Validated in one consolidated worktree cut from `release/v3.8.51`, boarded with the other 19 PRs of this batch. Two in-batch conflicts, both additive and resolved by keeping each side: the `ENVIRONMENT.md` table (#13035 + #13011) and the `chatHelpers.ts` import block (#12975 on the tip + #13017).

- `typecheck:core` clean; `check:dashboard-typecheck` OK (206 pre-existing, within baseline); `check:changelog-integrity` OK; `check:docs-counts` migrations ✓
- complexity 2816 / baseline 3218 and cognitive-complexity 1271 / baseline 1437 — both under baseline
- 531 of 532 focused assertions green across the batch's 46 test files
- `check-file-size` rebaselined for the batch's real growth (annotation `_rebaseline_2026_09_11_mergebatch_v3851_houminxi`, landed on #13038), attributed per PR

The single red is **not this batch**: `tests/unit/combo/quota-weighted-strategy.test.ts` → "A/B isolation: 7 hard-empty + 2 at 0.5% + 1 at 40%, floor=1" asserts an order between two connections of identical weight and flakes on the pure tip too — 2 failures in 4 runs at `origin/release/v3.8.51` with nothing from this batch applied.

⚠️ base-red inherited: #12732 — `Docs Gates`, `Merge integrity`, `No new ESLint warnings`, `Unit Tests fast-path` and `Fast Quality Gates` reproduce on the pure tip (provider count 356 vs the 358 the modules define, SKILL.md drift, and `open-sse/utils/stream.ts` at 3115 > frozen 3098, untouched here).

Thanks @HouMinXi — the live evidence on these (X500 logs, `storage.sqlite` state, real `/v1/models` probes, the 36-minute outage write-up) is what let a 20-PR batch be reviewed as a unit.
This commit is contained in:
Bob.Hou
2026-09-11 18:27:36 -04:00
committed by GitHub
parent c75e293a47
commit a31c7880c4
5 changed files with 488 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/fixes/13043-non-streaming-failure-classification.md Normal file
View File

@@ -0,0 +1 @@
- Restore provider failure classification and credential refresh on non-streaming requests: classify non-2xx failures to lock models on per-model quota exhaustion, update connection rate limits from headers and body, and pass credential refresh handlers to pipeline execution so 401 tokens can be refreshed and retried (#13043).

783
open-sse/handlers/chatCore.ts
View File

@@ -3615,6 +3615,418 @@ export async function handleChatCore({
let finalBody;
let claudePromptCacheLogMeta = null;
let credentialRefreshPersistRan = false;
const hadStreamOptions =
targetFormat === FORMATS.OPENAI_RESPONSES &&
translatedBody &&
typeof translatedBody === "object" &&
"stream_options" in translatedBody;
if (hadStreamOptions) {
delete (translatedBody as Record<string, unknown>).stream_options;
}
const executeRefreshCredentials = async (
currentCreds: Record<string, unknown>
): Promise<Record<string, unknown> | null> => {
if (typeof executor.refreshCredentials !== "function") {
return null;
}
if (hadStreamOptions) {
return null;
}
if (await shouldIsolateProbeFailures()) {
return null;
}
const targetCredentials = (currentCreds || credentials || {}) as Record<string, unknown>;
const attemptedRefreshToken =
typeof targetCredentials?.refreshToken === "string" ? targetCredentials.refreshToken : null;
credentialRefreshPersistRan = false;
const persistFn = onCredentialsRefreshed
? async (refreshResult: Record<string, unknown>) => {
credentialRefreshPersistRan = true;
Object.assign(targetCredentials, refreshResult);
Object.assign(credentials, refreshResult);
await onCredentialsRefreshed(refreshResult);
}
: undefined;
const casConnectionId =
typeof targetCredentials?.connectionId === "string"
? targetCredentials.connectionId.trim()
: "";
const casReread = casConnectionId
? async () => {
const latest = await getProviderConnectionById(casConnectionId);
return typeof latest?.refreshToken === "string" ? latest.refreshToken : null;
}
: null;
const newCredentials = (await refreshWithRetry(
() =>
runWithCasGuard(
casReread ? { expectedRefreshToken: attemptedRefreshToken, reread: casReread } : null,
() =>
runWithOnPersist(persistFn, () => executor.refreshCredentials(targetCredentials, log))
),
3,
log,
provider
)) as null | Record<string, unknown>;
if (newCredentials?.accessToken || newCredentials?.copilotToken) {
log?.info?.("TOKEN", `${provider?.toUpperCase()} | refreshed`);
if (!credentialRefreshPersistRan) {
Object.assign(targetCredentials, newCredentials);
Object.assign(credentials, newCredentials);
}
const errorConnectionId = String(getCurrentConnectionId() || connectionId || "");
if (errorConnectionId) {
updateProviderConnection(errorConnectionId, newCredentials).catch(() => {});
}
return newCredentials;
}
return null;
};
const handleCredentialsRefreshed = async (refreshed: Record<string, unknown>) => {
Object.assign(credentials, refreshed);
if (!credentialRefreshPersistRan && onCredentialsRefreshed) {
credentialRefreshPersistRan = true;
const targetConnectionId =
(credentials as { connectionId?: string })?.connectionId ||
(credentials as { id?: string })?.id ||
getCurrentConnectionId() ||
connectionId;
try {
await onCredentialsRefreshed({
...refreshed,
provider,
connectionId: targetConnectionId,
});
} catch (refreshErr) {
log?.warn?.(
"REFRESH",
`onCredentialsRefreshed persistence callback failed for connection ${targetConnectionId}: ${refreshErr}`
);
}
}
};
const applyProviderFailureClassification = async ({
statusCode,
message,
headers,
upstreamErrorBody,
retryAfterMs,
targetModel,
}: {
statusCode: number;
message: string;
headers?: Headers | null;
upstreamErrorBody?: unknown;
retryAfterMs?: number | null;
targetModel: string;
}) => {
let errorType = classifyProviderError(statusCode, message, provider);
if (statusCode === 429 && isModelScope()) {
const decision = classifyModelScope429(message, normalizeHeaders(headers));
errorType =
decision.kind === "quota_exhausted"
? PROVIDER_ERROR_TYPES.QUOTA_EXHAUSTED
: PROVIDER_ERROR_TYPES.RATE_LIMITED;
log?.warn?.(
"MODELSCOPE_429",
`${decision.kind} (model remaining: ${decision.snapshot.modelRemaining ?? "unknown"}, total remaining: ${decision.snapshot.totalRemaining ?? "unknown"})`
);
}
const persistentMessage = sanitizeErrorMessage(message) || "Provider request failed";
const errorConnectionId = getCurrentConnectionId() || connectionId;
if (errorConnectionId && errorType) {
try {
if (errorType === PROVIDER_ERROR_TYPES.FORBIDDEN) {
const probeIsolated = await shouldIsolateProbeFailures();
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "banned",
isActive: false,
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
probeIsolated ? "probe" : "production"
);
if (probeIsolated) {
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) -- connection stays active`
);
} else {
console.warn(
`[provider] Node ${errorConnectionId} banned (${statusCode}) -- disabling permanently`
);
}
} else if (errorType === PROVIDER_ERROR_TYPES.ACCOUNT_DEACTIVATED) {
if (
connectionHasExtraKeys(
errorConnectionId,
(credentials?.providerSpecificData as Record<string, unknown> | undefined)
?.extraApiKeys as string[] | undefined
)
) {
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} account deactivated (${statusCode}) -- has extra keys, keeping connection active`
);
} else {
const probeIsolated2 = await shouldIsolateProbeFailures();
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "deactivated",
isActive: false,
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
probeIsolated2 ? "probe" : "production"
);
if (probeIsolated2) {
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) -- connection stays active`
);
} else {
console.warn(
`[provider] Node ${errorConnectionId} account deactivated (${statusCode}) -- disabling permanently`
);
}
}
} else if (errorType === PROVIDER_ERROR_TYPES.QUOTA_EXHAUSTED) {
const probeIsolated3 = await shouldIsolateProbeFailures();
if (probeIsolated3) {
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "credits_exhausted",
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
"probe"
);
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) -- connection stays active`
);
} else {
let kimiRateLimitResetAt: string | null = null;
if (provider === "kimi-coding") {
try {
const { fetchAndPersistProviderLimits } =
await import("@/lib/usage/providerLimits");
const { usage } = await fetchAndPersistProviderLimits(errorConnectionId, "manual");
kimiRateLimitResetAt = getKimiTemporaryRateLimitResetAt(usage);
} catch {}
}
let quotaCooldownMs = kimiRateLimitResetAt
? Math.max(new Date(kimiRateLimitResetAt).getTime() - Date.now(), 0)
: retryAfterMs || COOLDOWN_MS.rateLimit;
const deferAntigravityQuotaStateToCaller = shouldDeferAntigravityQuotaStateToCaller(
provider,
typeof onStreamFailure === "function"
);
const isAntigravityQuotaFamily = shouldDeferAntigravityQuotaStateToCaller(
provider,
true
);
let coreOwnedAntigravityLockout: {
cooldownMs: number;
failureCount: number;
} | null = null;
if (isAntigravityQuotaFamily && !deferAntigravityQuotaStateToCaller) {
const quotaErrorText =
typeof upstreamErrorBody === "string"
? upstreamErrorBody
: upstreamErrorBody == null
? message
: JSON.stringify(upstreamErrorBody);
coreOwnedAntigravityLockout = await recordCoreOwnedAntigravityQuotaState({
provider,
connectionId: errorConnectionId,
model,
status: statusCode,
errorText: quotaErrorText,
headers: headers ?? undefined,
});
quotaCooldownMs = coreOwnedAntigravityLockout.cooldownMs;
}
const accountSemaphoreKey = resolveAccountSemaphoreKey({
provider,
model: targetModel,
connectionId: errorConnectionId,
credentials,
});
if (accountSemaphoreKey && !deferAntigravityQuotaStateToCaller) {
markAccountSemaphoreBlocked(accountSemaphoreKey, quotaCooldownMs);
}
if (deferAntigravityQuotaStateToCaller) {
} else if (coreOwnedAntigravityLockout) {
console.warn(
`[provider] Node ${errorConnectionId} Antigravity model quota exhausted (${statusCode}) for ${model} - ${Math.ceil(coreOwnedAntigravityLockout.cooldownMs / 1000)}s (failureCount=${coreOwnedAntigravityLockout.failureCount}, owner=core)`
);
} else if (kimiRateLimitResetAt) {
await updateProviderConnection(errorConnectionId, {
testStatus: "unavailable",
rateLimitedUntil: kimiRateLimitResetAt,
backoffLevel: 0,
lastErrorType: PROVIDER_ERROR_TYPES.RATE_LIMITED,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} Kimi request window exhausted (${statusCode}) -- retrying after ${kimiRateLimitResetAt}`
);
} else if (isModelScope() && errorConnectionId) {
lockModel(provider, errorConnectionId, model, "quota_exhausted", quotaCooldownMs);
if (targetModel && targetModel !== model) {
lockModel(
provider,
errorConnectionId,
targetModel,
"quota_exhausted",
quotaCooldownMs
);
}
console.warn(
`[provider] Node ${errorConnectionId} ModelScope model quota exhausted (${statusCode}) for ${targetModel} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)`
);
} else if (
lockModelIfPerModelQuota(
provider,
errorConnectionId,
model,
"quota_exhausted",
quotaCooldownMs
) ||
(targetModel &&
targetModel !== model &&
lockModelIfPerModelQuota(
provider,
errorConnectionId,
targetModel,
"quota_exhausted",
quotaCooldownMs
))
) {
const quotaScope = getQuotaScopeLabelForProvider(provider, targetModel);
console.warn(
`[provider] Node ${errorConnectionId} ${quotaScope}-only quota exhausted (${statusCode}) for ${targetModel} - ${Math.ceil(quotaCooldownMs / 1000)}s (cooldown_scope=${quotaScope}, ttl_source=${retryAfterMs ? "upstream" : "inferred"}, connection stays active)`
);
} else {
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "credits_exhausted",
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
"production"
);
console.warn(`[provider] Node ${errorConnectionId} exhausted quota (${statusCode})`);
}
}
} else if (errorType === PROVIDER_ERROR_TYPES.UNAUTHORIZED) {
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
} else if (errorType === PROVIDER_ERROR_TYPES.OAUTH_INVALID_TOKEN) {
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} OAuth token invalid (${statusCode}) -- token refresh available`
);
} else if (errorType === PROVIDER_ERROR_TYPES.PROJECT_ROUTE_ERROR) {
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} project routing error (${statusCode}) -- not banning`
);
} else if (errorType === PROVIDER_ERROR_TYPES.GEO_BLOCKED) {
const geoCooldownMs = COOLDOWN_MS.geoBlocked ?? 24 * 60 * 60 * 1000;
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
if (!(await shouldIsolateProbeFailures())) {
try {
const { setConnectionRateLimitUntil } = await import("@/lib/db/providers");
setConnectionRateLimitUntil(errorConnectionId, Date.now() + geoCooldownMs);
} catch {}
}
console.warn(
`[provider] Node ${errorConnectionId} geo-blocked (${statusCode}) -- excluded for ${Math.ceil(geoCooldownMs / 1000)}s, trying other accounts`
);
} else if (errorType === PROVIDER_ERROR_TYPES.GCP_PROJECT_REQUIRED) {
const byopCooldownMs = COOLDOWN_MS.gcpProjectRequired ?? 24 * 60 * 60 * 1000;
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
try {
const { setConnectionRateLimitUntil } = await import("@/lib/db/providers");
setConnectionRateLimitUntil(errorConnectionId, Date.now() + byopCooldownMs);
} catch {}
console.warn(
`[provider] Node ${errorConnectionId} GCP project required (${statusCode}) -- excluded for ${Math.ceil(byopCooldownMs / 1000)}s, routing to other accounts (enter a Project ID to restore)`
);
} else if (errorType === PROVIDER_ERROR_TYPES.MODEL_NOT_FOUND) {
const notFoundCooldownMs = COOLDOWN_MS.notFound;
if (!(await shouldIsolateProbeFailures())) {
const modelToLock = targetModel || model;
lockModel(
provider,
errorConnectionId,
modelToLock,
"model_not_found",
notFoundCooldownMs
);
console.warn(
`[provider] Node ${errorConnectionId} model not found (${statusCode}) for ${modelToLock} - locking model for ${Math.ceil(notFoundCooldownMs / 1000)}s (connection stays active)`
);
}
}
} catch {}
}
if (headers) {
updateFromHeaders(provider, errorConnectionId, headers, statusCode, targetModel);
}
if (errorConnectionId && upstreamErrorBody !== null && upstreamErrorBody !== undefined) {
updateFromResponseBody(
provider,
errorConnectionId,
upstreamErrorBody,
statusCode,
targetModel
);
}
};
let pipelineRecovered = false;
if (stream) {
try {
@@ -3640,7 +4052,8 @@ export async function handleChatCore({
replaceCredentials: (next) => {
Object.assign(credentials, next);
},
onCredentialsRefreshed: async () => {},
onCredentialsRefreshed: handleCredentialsRefreshed,
refreshCredentials: executeRefreshCredentials,
assertManagedLeaseFence: (id) => {
assertManagedLeaseFence(id);
},
@@ -4187,339 +4600,15 @@ export async function handleChatCore({
break providerFailure;
}
// T06/T10/T36: classify provider errors and persist terminal account states.
let errorType = classifyProviderError(statusCode, message, provider);
if (statusCode === 429 && isModelScope()) {
const decision = classifyModelScope429(message, normalizeHeaders(providerResponse.headers));
errorType =
decision.kind === "quota_exhausted"
? PROVIDER_ERROR_TYPES.QUOTA_EXHAUSTED
: PROVIDER_ERROR_TYPES.RATE_LIMITED;
log?.warn?.(
"MODELSCOPE_429",
`${decision.kind} (model remaining: ${decision.snapshot.modelRemaining ?? "unknown"}, total remaining: ${decision.snapshot.totalRemaining ?? "unknown"})`
);
}
// Classifiers and recovery paths above consume the raw provider wording.
// Project a separate value only at persistent connection-state boundaries.
const persistentMessage = sanitizeErrorMessage(message) || "Provider request failed";
const errorConnectionId = getCurrentConnectionId();
if (errorConnectionId && errorType) {
try {
if (errorType === PROVIDER_ERROR_TYPES.FORBIDDEN) {
{
const probeIsolated = await shouldIsolateProbeFailures();
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "banned",
isActive: false,
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
probeIsolated ? "probe" : "production"
);
if (probeIsolated) {
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) — connection stays active`
);
} else if (hasPerModelQuota(provider, model)) {
// Compatible / passthrough gateways: a 402 without a model id
// still must not terminalize the whole connection. Record the
// error for operators; sibling models stay selectable.
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} per-model quota exhausted (${statusCode}) — connection stays active`
);
} else {
console.warn(
`[provider] Node ${errorConnectionId} banned (${statusCode}) — disabling permanently`
);
}
}
} else if (errorType === PROVIDER_ERROR_TYPES.ACCOUNT_DEACTIVATED) {
// T-PROBE: probe-origin failures (test-all) never deactivate —
// record but stay active; Plan A (extra keys) stays first so the
// real path keeps its existing priority (#9817).
// Plan A: if connection has extra API keys, don't disable — only the failing key is affected.
// Single-key connections still get disabled as before.
if (
connectionHasExtraKeys(
errorConnectionId,
(credentials?.providerSpecificData as Record<string, unknown> | undefined)
?.extraApiKeys as string[] | undefined
)
) {
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} account deactivated (${statusCode}) — has extra keys, keeping connection active`
);
} else {
const probeIsolated2 = await shouldIsolateProbeFailures();
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "deactivated",
isActive: false,
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
probeIsolated2 ? "probe" : "production"
);
if (probeIsolated2) {
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) — connection stays active`
);
} else {
console.warn(
`[provider] Node ${errorConnectionId} account deactivated (${statusCode}) — disabling permanently`
);
}
}
} else if (errorType === PROVIDER_ERROR_TYPES.QUOTA_EXHAUSTED) {
{
const probeIsolated3 = await shouldIsolateProbeFailures();
if (probeIsolated3) {
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "credits_exhausted",
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
"probe"
);
console.warn(
`[provider] Node ${errorConnectionId} probe ${errorType} (${statusCode}) — connection stays active`
);
} else {
// Kimi's 403 says "billing cycle" for both an exhausted subscription and a
// temporary request window. Read its official usage endpoint before making
// the connection terminal: a non-zero Weekly quota plus an empty Ratelimit
// window must recover automatically at the reported reset time.
let kimiRateLimitResetAt: string | null = null;
if (provider === "kimi-coding") {
try {
const { fetchAndPersistProviderLimits } =
await import("@/lib/usage/providerLimits");
const { usage } = await fetchAndPersistProviderLimits(
errorConnectionId,
"manual"
);
kimiRateLimitResetAt = getKimiTemporaryRateLimitResetAt(usage);
} catch {
// Preserve the existing quota handling when Kimi's usage endpoint is unavailable.
}
}
// Providers with per-model quotas — lock the model only, not the connection
let quotaCooldownMs = kimiRateLimitResetAt
? Math.max(new Date(kimiRateLimitResetAt).getTime() - Date.now(), 0)
: retryAfterMs || COOLDOWN_MS.rateLimit;
const deferAntigravityQuotaStateToCaller = shouldDeferAntigravityQuotaStateToCaller(
provider,
typeof onStreamFailure === "function"
);
const isAntigravityQuotaFamily = shouldDeferAntigravityQuotaStateToCaller(
provider,
true
);
let coreOwnedAntigravityLockout: {
cooldownMs: number;
failureCount: number;
} | null = null;
if (isAntigravityQuotaFamily && !deferAntigravityQuotaStateToCaller) {
const quotaErrorText =
typeof upstreamErrorBody === "string"
? upstreamErrorBody
: upstreamErrorBody == null
? message
: JSON.stringify(upstreamErrorBody);
coreOwnedAntigravityLockout = await recordCoreOwnedAntigravityQuotaState({
provider,
connectionId: errorConnectionId,
model,
status: statusCode,
errorText: quotaErrorText,
headers: providerResponse.headers,
});
quotaCooldownMs = coreOwnedAntigravityLockout.cooldownMs;
}
const accountSemaphoreKey = resolveAccountSemaphoreKey({
provider,
model: currentModel,
connectionId: errorConnectionId,
credentials,
});
if (accountSemaphoreKey && !deferAntigravityQuotaStateToCaller) {
markAccountSemaphoreBlocked(accountSemaphoreKey, quotaCooldownMs);
}
if (deferAntigravityQuotaStateToCaller) {
// Defer both model and account-semaphore cooldowns to
// markAccountUnavailable, where header/body provenance and the
// configured maxCooldownMs are available. Direct consumers such
// as Responses pass no owner callback and retain core ownership.
} else if (coreOwnedAntigravityLockout) {
console.warn(
`[provider] Node ${errorConnectionId} Antigravity model quota exhausted (${statusCode}) for ${model} - ${Math.ceil(coreOwnedAntigravityLockout.cooldownMs / 1000)}s (failureCount=${coreOwnedAntigravityLockout.failureCount}, owner=core)`
);
} else if (kimiRateLimitResetAt) {
await updateProviderConnection(errorConnectionId, {
testStatus: "unavailable",
rateLimitedUntil: kimiRateLimitResetAt,
backoffLevel: 0,
lastErrorType: PROVIDER_ERROR_TYPES.RATE_LIMITED,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} Kimi request window exhausted (${statusCode}) — retrying after ${kimiRateLimitResetAt}`
);
} else if (isModelScope() && errorConnectionId) {
lockModel(provider, errorConnectionId, model, "quota_exhausted", quotaCooldownMs);
console.warn(
`[provider] Node ${errorConnectionId} ModelScope model quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)`
);
} else if (
lockModelIfPerModelQuota(
provider,
errorConnectionId,
model,
"quota_exhausted",
quotaCooldownMs
)
) {
const quotaScope = getQuotaScopeLabelForProvider(provider, model);
console.warn(
`[provider] Node ${errorConnectionId} ${quotaScope}-only quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (cooldown_scope=${quotaScope}, ttl_source=${retryAfterMs ? "upstream" : "inferred"}, connection stays active)`
);
} else {
await writeTerminalStatus(
errorConnectionId,
{
testStatus: "credits_exhausted",
lastError: persistentMessage,
lastErrorType: errorType,
errorCode: String(statusCode),
},
"production"
);
console.warn(
`[provider] Node ${errorConnectionId} exhausted quota (${statusCode})`
);
}
} // close probeIsolated3 else
}
} else if (errorType === PROVIDER_ERROR_TYPES.UNAUTHORIZED) {
// Normal 401 (token/session auth issue): keep account active for refresh/re-auth.
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
} else if (errorType === PROVIDER_ERROR_TYPES.OAUTH_INVALID_TOKEN) {
// OAuth 401 with invalid credentials - token refresh can recover
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} OAuth token invalid (${statusCode}) — token refresh available`
);
} else if (errorType === PROVIDER_ERROR_TYPES.PROJECT_ROUTE_ERROR) {
// Cloud Code 403 with stale project: not a ban, keep account active.
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
console.warn(
`[provider] Node ${errorConnectionId} project routing error (${statusCode}) — not banning`
);
} else if (errorType === PROVIDER_ERROR_TYPES.GEO_BLOCKED) {
// Google regional-availability refusal (e.g. "User location is not
// supported for the API use."). Account-independent and non-terminal:
// exclude the connection for the cooldown window so routing moves to
// other accounts instead of re-selecting this one on every request,
// and never mark it banned/expired. It becomes usable again once
// egress is routed through a supported-region proxy.
const geoCooldownMs = COOLDOWN_MS.geoBlocked ?? 24 * 60 * 60 * 1000;
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
// T-PROBE: the 24h exclusion is a routing mutation — a probe must
// not push a connection into a day-long cooldown (#9817).
if (!(await shouldIsolateProbeFailures())) {
try {
const { setConnectionRateLimitUntil } = await import("@/lib/db/providers");
setConnectionRateLimitUntil(errorConnectionId, Date.now() + geoCooldownMs);
} catch {
// DB write failure must never break the fallback loop
}
}
console.warn(
`[provider] Node ${errorConnectionId} geo-blocked (${statusCode}) — excluded for ${Math.ceil(geoCooldownMs / 1000)}s, trying other accounts`
);
} else if (errorType === PROVIDER_ERROR_TYPES.GCP_PROJECT_REQUIRED) {
// Antigravity BYOP: the account must Bring Its Own GCP Project.
// Account-specific and fixable by entering a Project ID — never a
// model lockout, never a ban. Exclude the connection for the
// cooldown window so selection prefers sibling accounts; the 422
// body carries the actionable message when no sibling is available.
const byopCooldownMs = COOLDOWN_MS.gcpProjectRequired ?? 24 * 60 * 60 * 1000;
await updateProviderConnection(errorConnectionId, {
lastErrorType: errorType,
lastError: persistentMessage,
errorCode: statusCode,
});
try {
const { setConnectionRateLimitUntil } = await import("@/lib/db/providers");
setConnectionRateLimitUntil(errorConnectionId, Date.now() + byopCooldownMs);
} catch {
// best-effort — never break the error path
}
console.warn(
`[provider] Node ${errorConnectionId} GCP project required (${statusCode}) — excluded for ${Math.ceil(byopCooldownMs / 1000)}s, routing to other accounts (enter a Project ID to restore)`
);
} else if (errorType === PROVIDER_ERROR_TYPES.MODEL_NOT_FOUND) {
// 404 — model/endpoint does not exist upstream. Lock the model so the
// retry/backoff loop stops hammering the dead endpoint (which would
// otherwise degenerate into a 429 rate-limit storm). Connection stays
// active since only the specific model is unavailable. (#6827)
const notFoundCooldownMs = COOLDOWN_MS.notFound;
// T-PROBE: the model lockout is a routing mutation — a probe must
// not lock a model for the cooldown window (#9817).
if (!(await shouldIsolateProbeFailures())) {
lockModel(
provider,
errorConnectionId,
currentModel,
"model_not_found",
notFoundCooldownMs
);
console.warn(
`[provider] Node ${errorConnectionId} model not found (${statusCode}) for ${currentModel} - locking model for ${Math.ceil(notFoundCooldownMs / 1000)}s (connection stays active)`
);
}
}
} catch {
// Best-effort state update; request flow should continue with fallback handling.
}
}
const errorConnectionId = getCurrentConnectionId() || connectionId;
await applyProviderFailureClassification({
statusCode,
message,
headers: providerResponse.headers,
upstreamErrorBody,
retryAfterMs,
targetModel: currentModel,
});
appendRequestLog({
model,
@@ -4546,11 +4635,7 @@ export async function handleChatCore({
upstreamErrorBody
);
// Update rate limiter from error response headers
updateFromHeaders(provider, errorConnectionId, providerResponse.headers, statusCode, model);
if (errorConnectionId && upstreamErrorBody !== null && upstreamErrorBody !== undefined) {
updateFromResponseBody(provider, errorConnectionId, upstreamErrorBody, statusCode, model);
}
// Rate limiter updated in applyProviderFailureClassification
// ── T5: Intra-family model fallback ──────────────────────────────────────
// Before returning a model-unavailable error upstream, try sibling models
@@ -4795,7 +4880,8 @@ export async function handleChatCore({
replaceCredentials: (next) => {
Object.assign(credentials, next);
},
onCredentialsRefreshed: async () => {},
onCredentialsRefreshed: handleCredentialsRefreshed,
refreshCredentials: executeRefreshCredentials,
assertManagedLeaseFence: (id) => {
assertManagedLeaseFence(id);
},
@@ -4908,6 +4994,23 @@ export async function handleChatCore({
if (legResult.kind === "error") {
const err = legResult.result;
const errMessage =
err?.rawMessage ||
(err?.originalError instanceof Error ? err.originalError.message : err?.error) ||
"";
const errHeaders = err?.upstreamHeaders || err?.response?.headers;
const errUpstreamBody = err?.upstreamErrorBody;
if (err) {
await applyProviderFailureClassification({
statusCode: err.status,
message: errMessage,
headers: errHeaders,
upstreamErrorBody: errUpstreamBody,
retryAfterMs: err.retryAfterMs ?? null,
targetModel: currentModel,
});
}
const captured = providerRequestCapture.latest?.() ?? null;
finalBody = captured?.body ?? finalBody ?? translatedBody;
if (captured) {

6
open-sse/handlers/chatCore/nonStreamingProviderLeg.ts
View File

@@ -435,6 +435,9 @@ export async function runNonStreamingProviderLeg(
{ passthrough: input.sourceFormat === "claude" }
),
response: outcome.result.response,
rawMessage: outcome.result.rawMessage || outcome.result.error,
upstreamErrorBody: outcome.result.upstreamErrorBody,
upstreamHeaders: outcome.result.upstreamHeaders ?? outcome.result.response?.headers,
},
receipt,
usage: outcome.providerUsage,
@@ -758,6 +761,9 @@ export async function runNonStreamingProviderLeg(
upstreamErrorType,
{ passthrough: sourceFormat === FORMATS.CLAUDE }
);
errorResult.rawMessage = message;
errorResult.upstreamHeaders = providerResponse.headers;
errorResult.upstreamErrorBody = parsedErrorBody;
return {
kind: "error",
result: errorResult as ChatCoreErrorResult,

48
open-sse/handlers/chatCore/providerExecutionPipeline.ts
View File

@@ -3,11 +3,17 @@ import type { getProviderCredentials } from "@/sse/services/auth.ts";
import type { updateFromHeaders, updateFromResponseBody } from "../../services/rateLimitManager.ts";
import type { writeTerminalStatus } from "@/shared/utils/terminalStatus.ts";
import type { updateProviderConnection } from "@/lib/db/providers.ts";
import type { lockModel, recordCoreOwnedAntigravityQuotaState } from "../../services/accountFallback.ts";
import type {
lockModel,
recordCoreOwnedAntigravityQuotaState,
} from "../../services/accountFallback.ts";
import { createErrorResult } from "../../utils/error.ts";
import { applyStatusRestatement } from "../../config/upstreamStatusRestatement.ts";
import { recoverAnthropicThinkingSignature } from "./thinkingSignatureRecovery.ts";
import { isModelUnavailableError, getNextFamilyFallback as defaultGetNextFamilyFallback } from "../../services/modelFamilyFallback.ts";
import {
isModelUnavailableError,
getNextFamilyFallback as defaultGetNextFamilyFallback,
} from "../../services/modelFamilyFallback.ts";
import { COOLDOWN_MS } from "../../config/errorConfig.ts";
import { normalizeHeaders } from "../../utils/headers.ts";
@@ -196,11 +202,7 @@ async function toOutcome(
body,
retryAfterMs: null,
});
const result = createErrorResult(
restatement.status,
message,
restatement.retryAfterMs
);
const result = createErrorResult(restatement.status, message, restatement.retryAfterMs);
return {
kind: "error",
result: {
@@ -210,6 +212,9 @@ async function toOutcome(
error: result.error,
errorCode: result.errorCode,
errorType: result.errorType,
rawMessage: message,
upstreamErrorBody: body,
upstreamHeaders: attempt.response.headers,
},
providerUsage: null,
model,
@@ -273,7 +278,12 @@ export async function runProviderExecutionPipeline(
const status = attempt.response.status;
if (status >= 200 && status < 300) {
return toOutcome(attempt, wire.currentModel, currentConnectionId(connection), target.provider);
return toOutcome(
attempt,
wire.currentModel,
currentConnectionId(connection),
target.provider
);
}
const isolateProbe = await state.isolateProbeFailures();
@@ -401,11 +411,16 @@ export async function runProviderExecutionPipeline(
};
},
});
if (signatureRecovery.attempted && signatureRecovery.succeeded && signatureRecovery.execution) {
if (
signatureRecovery.attempted &&
signatureRecovery.succeeded &&
signatureRecovery.execution
) {
lastAttempt = {
response: signatureRecovery.execution.response,
url: signatureRecovery.execution.url ?? attempt.url,
headers: (signatureRecovery.execution.headers as Record<string, string>) ?? attempt.headers,
headers:
(signatureRecovery.execution.headers as Record<string, string>) ?? attempt.headers,
transformedBody: signatureRecovery.execution.transformedBody ?? attempt.transformedBody,
};
return toOutcome(
@@ -430,7 +445,11 @@ export async function runProviderExecutionPipeline(
// keep statusText
}
if (isModelUnavailableError(status, fallbackMessage, target.provider)) {
const nextModel = resolveFamilyFallback(wire.currentModel, wire.triedModels, target.provider);
const nextModel = resolveFamilyFallback(
wire.currentModel,
wire.triedModels,
target.provider
);
if (nextModel) {
wire.setBodyAndModel({ ...wire.body, model: nextModel }, nextModel);
modelFallbackPending = true;
@@ -443,7 +462,12 @@ export async function runProviderExecutionPipeline(
}
if (lastAttempt) {
return toOutcome(lastAttempt, wire.currentModel, currentConnectionId(connection), target.provider);
return toOutcome(
lastAttempt,
wire.currentModel,
currentConnectionId(connection),
target.provider
);
}
return leaseMismatch(wire.currentModel, currentConnectionId(connection));
}

2
src/lib/skills/toolLoopTypes.ts
View File

@@ -44,6 +44,8 @@ export interface ChatCoreErrorResult {
retryAfterMs?: number;
originalError?: unknown;
rawMessage?: string;
upstreamHeaders?: Headers;
upstreamErrorBody?: unknown;
}
export type NonStreamingProviderLegResult =