diff --git a/changelog.d/fixes/13043-non-streaming-failure-classification.md b/changelog.d/fixes/13043-non-streaming-failure-classification.md new file mode 100644 index 0000000000..e6774f9825 --- /dev/null +++ b/changelog.d/fixes/13043-non-streaming-failure-classification.md @@ -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). diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 20078c522d..d14035a49e 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -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).stream_options; + } + + const executeRefreshCredentials = async ( + currentCreds: Record + ): Promise | null> => { + if (typeof executor.refreshCredentials !== "function") { + return null; + } + if (hadStreamOptions) { + return null; + } + if (await shouldIsolateProbeFailures()) { + return null; + } + + const targetCredentials = (currentCreds || credentials || {}) as Record; + const attemptedRefreshToken = + typeof targetCredentials?.refreshToken === "string" ? targetCredentials.refreshToken : null; + credentialRefreshPersistRan = false; + const persistFn = onCredentialsRefreshed + ? async (refreshResult: Record) => { + 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; + + 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) => { + 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 | 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 | 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) { diff --git a/open-sse/handlers/chatCore/nonStreamingProviderLeg.ts b/open-sse/handlers/chatCore/nonStreamingProviderLeg.ts index 7656939533..ba17639af9 100644 --- a/open-sse/handlers/chatCore/nonStreamingProviderLeg.ts +++ b/open-sse/handlers/chatCore/nonStreamingProviderLeg.ts @@ -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, diff --git a/open-sse/handlers/chatCore/providerExecutionPipeline.ts b/open-sse/handlers/chatCore/providerExecutionPipeline.ts index 077ff7cfe1..b058d8b54c 100644 --- a/open-sse/handlers/chatCore/providerExecutionPipeline.ts +++ b/open-sse/handlers/chatCore/providerExecutionPipeline.ts @@ -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) ?? attempt.headers, + headers: + (signatureRecovery.execution.headers as Record) ?? 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)); } diff --git a/src/lib/skills/toolLoopTypes.ts b/src/lib/skills/toolLoopTypes.ts index 5ebf575518..f3af5a3871 100644 --- a/src/lib/skills/toolLoopTypes.ts +++ b/src/lib/skills/toolLoopTypes.ts @@ -44,6 +44,8 @@ export interface ChatCoreErrorResult { retryAfterMs?: number; originalError?: unknown; rawMessage?: string; + upstreamHeaders?: Headers; + upstreamErrorBody?: unknown; } export type NonStreamingProviderLegResult =