diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index aa0fdadda3..037e788059 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -390,6 +390,13 @@ export async function handleChatCore({ // call sites stay byte-identical. const trace = (label: string, extra?: Record) => stageTrace(label, extra, { traceEnabled, startTime, traceId, log }); + const getCurrentConnectionId = () => { + const credentialConnectionId = + typeof credentials?.connectionId === "string" && credentials.connectionId.trim().length > 0 + ? credentials.connectionId.trim() + : null; + return credentialConnectionId || connectionId || null; + }; let tokensCompressed: number | null = null; body = injectSystemPrompt(body); // ── Per-endpoint custom system prompt (port of upstream #2063) ── @@ -441,7 +448,7 @@ export async function handleChatCore({ buildFailureUsageRecord({ provider, model, - connectionId, + connectionId: getCurrentConnectionId(), apiKeyInfo, effectiveServiceTier, isCombo, @@ -462,7 +469,8 @@ export async function handleChatCore({ ): void => recordKeyHealthStatusFor(status, creds, log); const persistCodexQuotaState = async (headers: Record | null, status = 0) => { - if (provider !== "codex" || !connectionId || !headers) return; + const currentConnectionId = getCurrentConnectionId(); + if (provider !== "codex" || !currentConnectionId || !headers) return; try { const existingProviderData = @@ -486,10 +494,10 @@ export async function handleChatCore({ // Invalidate the preflight cache for this connection so the next // isModelAvailable check fetches fresh quota data. if (status === 429) { - invalidateCodexQuotaCache(connectionId); + invalidateCodexQuotaCache(currentConnectionId); } - await updateProviderConnection(connectionId, { + await updateProviderConnection(currentConnectionId, { providerSpecificData: built.nextProviderData, }); @@ -2069,17 +2077,6 @@ export async function handleChatCore({ const executeProviderRequest = async (modelToCall = effectiveModel, allowDedup = false) => { const execute = async () => { - const executionCredentials = getExecutionCredentials(); - // Track execution credentials for key health recording (to capture selectedKeyId) - let lastExecCreds = executionCredentials; - const accountSemaphoreMaxConcurrency = - resolveAccountSemaphoreMaxConcurrency(executionCredentials); - const accountSemaphoreKey = resolveAccountSemaphoreKey({ - provider, - model: modelToCall, - connectionId, - credentials: executionCredentials, - }); // Upstream body preparation extracted to chatCore/upstreamBody.ts (#3501 — first internal // sub-slice of executeProviderRequest); produces the body sent upstream (payload rules + // tool-limit truncation + qwen oauth user backfill + prompt_cache_key injection). @@ -2097,90 +2094,105 @@ export async function handleChatCore({ stage: "payload_prepared", }); - trace("pre_semaphore", { - semaphoreKey: accountSemaphoreKey, - max: accountSemaphoreMaxConcurrency, - }); - if (accountSemaphoreKey && accountSemaphoreMaxConcurrency != null) { - updatePendingScope(pendingScope, { - stage: "waiting_account_slot", - }); - } - const acquireAccountSemaphoreRelease = - accountSemaphoreKey && accountSemaphoreMaxConcurrency != null - ? await acquireAccountSemaphore(accountSemaphoreKey, { - maxConcurrency: accountSemaphoreMaxConcurrency, - signal: streamController.signal, - }) - : () => {}; - trace("post_semaphore"); - updatePendingScope(pendingScope, { - stage: "waiting_rate_limit", - }); - + let releaseRawResultAccountSemaphore = () => {}; try { - trace("pre_rate_limit"); - const rawResult = await withRateLimit( - provider, - connectionId, - modelToCall, - async () => { - trace("inside_rate_limit"); - updatePendingScope(pendingScope, { - stage: "rate_limit_slot_acquired", - }); - let attempts = 0; - const isModelScopeForRequest = isModelScope(); - const maxAttempts = isModelScopeForRequest + const rawResult = await (async () => { + let attempts = 0; + const isModelScopeForRequest = isModelScope(); + const maxAttempts = isModelScopeForRequest + ? 3 + : provider === "qwen" ? 3 - : provider === "qwen" + : provider === "codex" ? 3 - : provider === "codex" - ? 3 - : 1; + : 1; - // ── Codex 429 account-rotation state ───────────────────────────────── - // Track excluded connection IDs for codex failover across attempts. - const codexExcludedIds: string[] = []; - // Derive session affinity key once for codex failover (used to clear affinity on 429). - const codexSessionAffinityKey = - provider === "codex" - ? (extractSessionAffinityKey(body, clientRawRequest?.headers) ?? null) - : null; + // ── Codex 429 account-rotation state ───────────────────────────────── + // Track excluded connection IDs for codex failover across attempts. + const codexExcludedIds: string[] = []; + // Derive session affinity key once for codex failover (used to clear affinity on 429). + const codexSessionAffinityKey = + provider === "codex" + ? (extractSessionAffinityKey(body, clientRawRequest?.headers) ?? null) + : null; - while (attempts < maxAttempts) { - trace("pre_executor", { attempt: attempts }); + while (attempts < maxAttempts) { + trace("pre_executor", { attempt: attempts }); + updatePendingScope(pendingScope, { + stage: "sending_to_provider", + }); + const execCreds = getExecutionCredentials(); + const attemptConnectionId = execCreds?.connectionId || connectionId; + const accountSemaphoreMaxConcurrency = resolveAccountSemaphoreMaxConcurrency(execCreds); + const accountSemaphoreKey = resolveAccountSemaphoreKey({ + provider, + model: modelToCall, + connectionId: attemptConnectionId, + credentials: execCreds, + }); + + trace("pre_semaphore", { + semaphoreKey: accountSemaphoreKey, + max: accountSemaphoreMaxConcurrency, + }); + if (accountSemaphoreKey && accountSemaphoreMaxConcurrency != null) { updatePendingScope(pendingScope, { - stage: "sending_to_provider", + stage: "waiting_account_slot", }); - const execCreds = getExecutionCredentials(); - const rawExecutorResult = await executeWithUpstreamStartTimeout({ - executor, + } + const releaseAccountSemaphore = + accountSemaphoreKey && accountSemaphoreMaxConcurrency != null + ? await acquireAccountSemaphore(accountSemaphoreKey, { + maxConcurrency: accountSemaphoreMaxConcurrency, + signal: streamController.signal, + }) + : () => {}; + trace("post_semaphore"); + updatePendingScope(pendingScope, { + stage: "waiting_rate_limit", + }); + + try { + trace("pre_rate_limit", { connectionId: attemptConnectionId }); + const rawExecutorResult = await withRateLimit( provider, - model: modelToCall, - signal: streamController.signal, - log, - execute: (signal) => - runWithCapture(providerRequestCapture, () => - executor.execute({ - model: modelToCall, - body: bodyToSend, - stream: upstreamStream, - credentials: execCreds, - signal, - log, - extendedContext, - upstreamExtraHeaders: buildUpstreamHeadersForExecute(modelToCall), - clientHeaders: buildExecutorClientHeaders( - clientRawRequest?.headers, - userAgent + attemptConnectionId, + modelToCall, + async () => { + trace("inside_rate_limit", { connectionId: attemptConnectionId }); + updatePendingScope(pendingScope, { + stage: "rate_limit_slot_acquired", + }); + return executeWithUpstreamStartTimeout({ + executor, + provider, + model: modelToCall, + signal: streamController.signal, + log, + execute: (signal) => + runWithCapture(providerRequestCapture, () => + executor.execute({ + model: modelToCall, + body: bodyToSend, + stream: upstreamStream, + credentials: execCreds, + signal, + log, + extendedContext, + upstreamExtraHeaders: buildUpstreamHeadersForExecute(modelToCall), + clientHeaders: buildExecutorClientHeaders( + clientRawRequest?.headers, + userAgent + ), + onCredentialsRefreshed, + skipUpstreamRetry, + contextEditing: { enabled: contextEditingEnabled }, + }) ), - onCredentialsRefreshed, - skipUpstreamRetry, - contextEditing: { enabled: contextEditingEnabled }, - }) - ), - }); + }); + }, + streamController.signal + ); const res = normalizeExecutorResult(rawExecutorResult); trace("post_executor", { status: res?.response?.status }); @@ -2210,6 +2222,7 @@ export async function handleChatCore({ if (bodyPeek.toLowerCase().includes("exceeded your current quota")) { const delay = 1500 * (attempts + 1); log?.warn?.("QWEN_RETRY", `Quota 429 hit. Retrying in ${delay}ms...`); + releaseAccountSemaphore(); await new Promise((r) => setTimeout(r, delay)); attempts++; continue; @@ -2229,6 +2242,7 @@ export async function handleChatCore({ "MODELSCOPE_RETRY", `429 ${decision.kind}; retrying in ${delay}ms (model remaining: ${decision.snapshot.modelRemaining ?? "unknown"})` ); + releaseAccountSemaphore(); await new Promise((r) => setTimeout(r, delay)); attempts++; continue; @@ -2242,7 +2256,8 @@ export async function handleChatCore({ res.response.status === 429 && attempts < maxAttempts - 1 ) { - const failedConnectionId = credentials?.connectionId || connectionId; + const failedConnectionId = + execCreds?.connectionId || credentials?.connectionId || connectionId; const normalizedHeaders = normalizeHeaders(res.response.headers); const retryAfterHeader = normalizedHeaders["retry-after"] ?? null; const retryAfterMs = retryAfterHeader @@ -2289,7 +2304,18 @@ export async function handleChatCore({ if (!nextCreds || nextCreds.allRateLimited) { log?.warn?.("CODEX_FAILOVER", "No more codex accounts available — returning 429"); - return res; + if (stream) { + releaseAccountSemaphore(); + return { + ...res, + _executionCredentials: execCreds, + }; + } + return { + ...res, + _accountSemaphoreRelease: releaseAccountSemaphore, + _executionCredentials: execCreds, + }; } const newConnectionId = nextCreds.connectionId; @@ -2313,6 +2339,7 @@ export async function handleChatCore({ // Update credentials in-place so getExecutionCredentials() picks up the new account Object.assign(credentials, nextCreds); + releaseAccountSemaphore(); attempts++; continue; } @@ -2321,7 +2348,7 @@ export async function handleChatCore({ if (stream) { const originalBody = res.response.body; if (!originalBody) { - acquireAccountSemaphoreRelease(); + releaseAccountSemaphore(); return res; } @@ -2413,7 +2440,7 @@ export async function handleChatCore({ originalBody as ReadableStream, () => runUpstreamStream(bodyToSend), { - finalize: acquireAccountSemaphoreRelease, + finalize: releaseAccountSemaphore, onRetry: (attempt, err) => log?.warn?.( "STREAM_RECOVERY", @@ -2432,7 +2459,7 @@ export async function handleChatCore({ } else { clientBody = wrapReadableStreamWithFinalize( originalBody, - acquireAccountSemaphoreRelease + releaseAccountSemaphore ); } @@ -2451,11 +2478,14 @@ export async function handleChatCore({ return { ...res, _executionCredentials: execCreds, + _accountSemaphoreRelease: releaseAccountSemaphore, }; + } catch (error) { + releaseAccountSemaphore(); + throw error; } - }, - streamController.signal - ); + } + })(); if (stream) { return rawResult; @@ -2471,6 +2501,10 @@ export async function handleChatCore({ ) { recordKeyHealthStatus(status, rawResult._executionCredentials); } + releaseRawResultAccountSemaphore = + typeof rawResult._accountSemaphoreRelease === "function" + ? rawResult._accountSemaphoreRelease + : () => {}; const statusText = rawResult.response.statusText; const headersObj = normalizeHeaders(rawResult.response.headers); @@ -2482,7 +2516,8 @@ export async function handleChatCore({ contentType, upstreamStream ); - acquireAccountSemaphoreRelease(); + releaseRawResultAccountSemaphore(); + releaseRawResultAccountSemaphore = () => {}; return { ...rawResult, @@ -2500,7 +2535,7 @@ export async function handleChatCore({ }, }; } catch (error) { - acquireAccountSemaphoreRelease(); + releaseRawResultAccountSemaphore(); throw error; } }; @@ -2597,6 +2632,7 @@ export async function handleChatCore({ providerUrl = result.url; providerHeaders = result.headers; finalBody = providerRequestCapture.body(result.transformedBody); + const responseConnectionId = getCurrentConnectionId(); effectiveServiceTier = resolveEffectiveServiceTier(finalBody); claudePromptCacheLogMeta = buildClaudePromptCacheLogMeta( targetFormat, @@ -2615,7 +2651,7 @@ export async function handleChatCore({ // Update rate limiter from response headers (learn limits dynamically) updateFromHeaders( provider, - connectionId, + responseConnectionId, providerResponse.headers, providerResponse.status, model @@ -2625,7 +2661,7 @@ export async function handleChatCore({ try { const { storeRateLimitHeaders } = await import("@/lib/quota/saturationSignals"); storeRateLimitHeaders( - connectionId, + responseConnectionId, provider, providerResponse.headers as Record ); @@ -2946,10 +2982,11 @@ export async function handleChatCore({ `${decision.kind} (model remaining: ${decision.snapshot.modelRemaining ?? "unknown"}, total remaining: ${decision.snapshot.totalRemaining ?? "unknown"})` ); } - if (connectionId && errorType) { + const errorConnectionId = getCurrentConnectionId(); + if (errorConnectionId && errorType) { try { if (errorType === PROVIDER_ERROR_TYPES.FORBIDDEN) { - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { isActive: false, testStatus: "banned", lastErrorType: errorType, @@ -2957,28 +2994,28 @@ export async function handleChatCore({ errorCode: statusCode, }); console.warn( - `[provider] Node ${connectionId} banned (${statusCode}) — disabling permanently` + `[provider] Node ${errorConnectionId} banned (${statusCode}) — disabling permanently` ); } else if (errorType === PROVIDER_ERROR_TYPES.ACCOUNT_DEACTIVATED) { // 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( - connectionId, + errorConnectionId, (credentials?.providerSpecificData as Record | undefined) ?.extraApiKeys as string[] | undefined ) ) { - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { lastErrorType: errorType, lastError: message, errorCode: statusCode, }); console.warn( - `[provider] Node ${connectionId} account deactivated (${statusCode}) — has extra keys, keeping connection active` + `[provider] Node ${errorConnectionId} account deactivated (${statusCode}) — has extra keys, keeping connection active` ); } else { - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { isActive: false, testStatus: "deactivated", lastErrorType: errorType, @@ -2986,7 +3023,7 @@ export async function handleChatCore({ errorCode: statusCode, }); console.warn( - `[provider] Node ${connectionId} account deactivated (${statusCode}) — disabling permanently` + `[provider] Node ${errorConnectionId} account deactivated (${statusCode}) — disabling permanently` ); } } else if (errorType === PROVIDER_ERROR_TYPES.QUOTA_EXHAUSTED) { @@ -2995,75 +3032,64 @@ export async function handleChatCore({ const accountSemaphoreKey = resolveAccountSemaphoreKey({ provider, model: currentModel, - connectionId, + connectionId: errorConnectionId, credentials, }); if (accountSemaphoreKey) { markAccountSemaphoreBlocked(accountSemaphoreKey, quotaCooldownMs); } - if (isModelScope() && connectionId) { - lockModel(provider, connectionId, model, "quota_exhausted", quotaCooldownMs); + if (isModelScope() && errorConnectionId) { + lockModel(provider, errorConnectionId, model, "quota_exhausted", quotaCooldownMs); console.warn( - `[provider] Node ${connectionId} ModelScope model quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)` + `[provider] Node ${errorConnectionId} ModelScope model quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)` ); } else if ( lockModelIfPerModelQuota( provider, - connectionId, + errorConnectionId, model, "quota_exhausted", quotaCooldownMs ) ) { console.warn( - `[provider] Node ${connectionId} model-only quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)` + `[provider] Node ${errorConnectionId} model-only quota exhausted (${statusCode}) for ${model} - ${Math.ceil(quotaCooldownMs / 1000)}s (connection stays active)` ); } else { - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { testStatus: "credits_exhausted", lastErrorType: errorType, lastError: message, errorCode: statusCode, }); - console.warn(`[provider] Node ${connectionId} exhausted quota (${statusCode})`); + console.warn(`[provider] Node ${errorConnectionId} exhausted quota (${statusCode})`); } - } else if (errorType === PROVIDER_ERROR_TYPES.ACCOUNT_DEACTIVATED) { - await updateProviderConnection(connectionId, { - isActive: false, - testStatus: "expired", - lastErrorType: errorType, - lastError: message, - errorCode: statusCode, - }); - console.warn( - `[provider] Node ${connectionId} account deactivated (${statusCode}) — marked expired` - ); } else if (errorType === PROVIDER_ERROR_TYPES.UNAUTHORIZED) { // Normal 401 (token/session auth issue): keep account active for refresh/re-auth. - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { lastErrorType: errorType, lastError: message, errorCode: statusCode, }); } else if (errorType === PROVIDER_ERROR_TYPES.OAUTH_INVALID_TOKEN) { // OAuth 401 with invalid credentials - token refresh can recover - await updateProviderConnection(connectionId, { + await updateProviderConnection(errorConnectionId, { lastErrorType: errorType, lastError: message, errorCode: statusCode, }); console.warn( - `[provider] Node ${connectionId} OAuth token invalid (${statusCode}) — token refresh available` + `[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(connectionId, { + await updateProviderConnection(errorConnectionId, { lastErrorType: errorType, lastError: message, errorCode: statusCode, }); console.warn( - `[provider] Node ${connectionId} project routing error (${statusCode}) — not banning` + `[provider] Node ${errorConnectionId} project routing error (${statusCode}) — not banning` ); } } catch { @@ -3071,9 +3097,12 @@ export async function handleChatCore({ } } - appendRequestLog({ model, provider, connectionId, status: `FAILED ${statusCode}` }).catch( - () => {} - ); + appendRequestLog({ + model, + provider, + connectionId: errorConnectionId, + status: `FAILED ${statusCode}`, + }).catch(() => {}); const errMsg = formatProviderError(new Error(message), provider, model, statusCode); console.log(`${COLORS.red}[ERROR] ${errMsg}${COLORS.reset}`); @@ -3094,9 +3123,9 @@ export async function handleChatCore({ ); // Update rate limiter from error response headers - updateFromHeaders(provider, connectionId, providerResponse.headers, statusCode, model); - if (connectionId && upstreamErrorBody !== null && upstreamErrorBody !== undefined) { - updateFromResponseBody(provider, connectionId, upstreamErrorBody, statusCode, model); + updateFromHeaders(provider, errorConnectionId, providerResponse.headers, statusCode, model); + if (errorConnectionId && upstreamErrorBody !== null && upstreamErrorBody !== undefined) { + updateFromResponseBody(provider, errorConnectionId, upstreamErrorBody, statusCode, model); } // ── T5: Intra-family model fallback ────────────────────────────────────── @@ -3449,9 +3478,10 @@ export async function handleChatCore({ if (onRequestSuccess) { await onRequestSuccess(); } + const successConnectionId = getCurrentConnectionId(); await maybeSyncClaudeExtraUsageState({ provider, - connectionId, + connectionId: successConnectionId, providerSpecificData: credentials?.providerSpecificData, log, }); @@ -3472,16 +3502,20 @@ export async function handleChatCore({ skillRequestId, log, }); - appendRequestLog({ model, provider, connectionId, tokens: usage, status: "200 OK" }).catch( - () => {} - ); + appendRequestLog({ + model, + provider, + connectionId: successConnectionId, + tokens: usage, + status: "200 OK", + }).catch(() => {}); // Save structured call log with full payloads const cacheUsageLogMeta = buildCacheUsageLogMeta(usage); recordNonStreamingUsageStats(usage, { traceEnabled, provider, - connectionId, + connectionId: successConnectionId, model, startTime, apiKeyInfo, @@ -3878,11 +3912,12 @@ export async function handleChatCore({ streamFailureCompletionRecorded = true; } const cacheUsageLogMeta = buildCacheUsageLogMeta(streamUsage); + const streamConnectionId = getCurrentConnectionId(); if (normalizedStreamStatus === 200) { void maybeSyncClaudeExtraUsageState({ provider, - connectionId, + connectionId: streamConnectionId, providerSpecificData: credentials?.providerSpecificData, log, }); @@ -3909,7 +3944,7 @@ export async function handleChatCore({ pendingRequestId, model, provider, - connectionId: connectionId || credentials?.connectionId || null, + connectionId: streamConnectionId, providerResponse: providerPayload ?? streamResponseBody ?? undefined, clientResponse: clientPayload ?? streamResponseBody ?? undefined, status: normalizedStreamStatus, @@ -3928,7 +3963,7 @@ export async function handleChatCore({ startTime, ttft, streamErrorCode, - connectionId, + connectionId: streamConnectionId, apiKeyInfo, effectiveServiceTier, isCombo, diff --git a/open-sse/handlers/chatCore/attemptLogging.ts b/open-sse/handlers/chatCore/attemptLogging.ts index 7414b951da..86947eecbd 100644 --- a/open-sse/handlers/chatCore/attemptLogging.ts +++ b/open-sse/handlers/chatCore/attemptLogging.ts @@ -35,10 +35,7 @@ export type PersistAttemptLogsContext = { model: string | null | undefined; skillRequestId: string; detailedLoggingEnabled: boolean; - reqLogger: - | { getPipelinePayloads?: () => Record | undefined } - | null - | undefined; + reqLogger: { getPipelinePayloads?: () => Record | undefined } | null | undefined; pendingRequestId: unknown; clientRawRequest: { endpoint?: string } | null | undefined; requestedModel: unknown; @@ -55,6 +52,26 @@ export type PersistAttemptLogsContext = { noLogEnabled: unknown; }; +function toConnectionId(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function buildAccountRotationMeta( + provider: string | null | undefined, + initialConnectionId: string | null, + finalConnectionId: string | null +) { + if (provider !== "codex" || !initialConnectionId || !finalConnectionId) return null; + if (initialConnectionId === finalConnectionId) return null; + + return { + codexAccountRotation: { + initialConnectionId, + finalConnectionId, + }, + }; +} + export function persistAttemptLogs(args: PersistAttemptLogsArgs, ctx: PersistAttemptLogsContext) { const { status, @@ -90,20 +107,27 @@ export function persistAttemptLogs(args: PersistAttemptLogsArgs, ctx: PersistAtt apiKeyInfo, noLogEnabled, } = ctx; + const initialConnectionId = toConnectionId(connectionId); + const finalConnectionId = toConnectionId(credentials?.connectionId) || initialConnectionId; + const accountRotationMeta = buildAccountRotationMeta( + provider, + initialConnectionId, + finalConnectionId + ); const providerWarnings = extractProviderWarnings(providerResponse, clientResponse, responseBody); if (providerWarnings.length > 0) { logAuditEvent({ action: "provider.warning", actor: "system", - target: [provider, connectionId].filter(Boolean).join(":") || provider || model, + target: [provider, finalConnectionId].filter(Boolean).join(":") || provider || model, resourceType: "provider_warning", status: "warning", requestId: skillRequestId, details: { provider, model, - connectionId, + connectionId: finalConnectionId, httpStatus: status, warnings: providerWarnings, }, @@ -142,16 +166,18 @@ export function persistAttemptLogs(args: PersistAttemptLogsArgs, ctx: PersistAtt model, requestedModel, provider, - connectionId: connectionId || credentials?.connectionId || undefined, + connectionId: finalConnectionId || undefined, duration: Date.now() - startTime, tokens: tokens || {}, requestBody: cloneBoundedChatLogPayload( attachLogMeta(truncateForLog(body as Record), { + ...accountRotationMeta, claudePromptCache: claudeCacheMeta, }) ), responseBody: cloneBoundedChatLogPayload( attachLogMeta(truncateForLog(responseBody as Record), { + ...accountRotationMeta, claudePromptCache: claudeCacheMeta ? { applied: claudeCacheMeta.applied, diff --git a/src/shared/components/RequestLoggerDetail.tsx b/src/shared/components/RequestLoggerDetail.tsx index 51a0612897..840a8c3b65 100644 --- a/src/shared/components/RequestLoggerDetail.tsx +++ b/src/shared/components/RequestLoggerDetail.tsx @@ -26,14 +26,18 @@ function PayloadSection({ title, json, onCopy, collapsible = true, defaultOpen =
-

{title}

+

+ {title} +

{collapsible && ( )}
@@ -117,7 +121,9 @@ function StreamSection({ title, json, onCopy }) { className="flex items-center gap-1 px-2 py-1 text-xs text-text-muted hover:text-text-primary transition-colors" aria-label={`Copy ${title}`} > - {copied ? "check" : "content_copy"} + + {copied ? "check" : "content_copy"} + {copied ? "Copied!" : "Copy"}
@@ -136,6 +142,30 @@ function StreamSection({ title, json, onCopy }) { type StreamChunks = Record; +function getCodexAccountRotation(detail) { + const sources = [detail?.requestBody, detail?.responseBody]; + + for (const source of sources) { + const meta = source?._omniroute; + const rotation = meta?.codexAccountRotation; + if ( + rotation && + typeof rotation.initialConnectionId === "string" && + typeof rotation.finalConnectionId === "string" && + rotation.initialConnectionId !== rotation.finalConnectionId + ) { + return rotation; + } + } + + return null; +} + +function formatConnectionId(value) { + if (typeof value !== "string" || value.length === 0) return "-"; + return value.length > 8 ? `${value.slice(0, 8)}...` : value; +} + export default function RequestLoggerDetail({ log, detail, @@ -262,6 +292,7 @@ export default function RequestLoggerDetail({ ? "bg-emerald-500/20 text-emerald-700 dark:text-emerald-300 border-emerald-500/30" : "bg-sky-500/20 text-sky-700 dark:text-sky-300 border-sky-500/30"; const accountLabel = maskAccount(detail?.account || log.account, emailsVisible); + const codexAccountRotation = getCodexAccountRotation(detail); return (
-
Started At
+
+ Started At +
{formatDate(log.timestamp)}
-
Duration
+
+ Duration +
{formatDuration(log.duration)}
-
Model
+
+ Model +
{log.model}
-
Provider
+
+ Provider +
-
Account
+
+ Account +
{accountLabel}
@@ -391,8 +432,13 @@ export default function RequestLoggerDetail({
{formatDuration(log.duration)}
-
Input
-
+
+ Input +
+
Total In: {formatTokenValue(tokenStats.totalIn)} @@ -402,22 +448,28 @@ export default function RequestLoggerDetail({ Cache Write: {formatTokenValue(tokenStats.cacheWrite)} - {tokenStats.compressed != null && tokenStats.compressed > 0 && (() => { - const fromTokens = tokenStats.totalIn + tokenStats.compressed; - const pct = Math.round((tokenStats.compressed / fromTokens) * 100); - return ( - - Compressed: {fromTokens.toLocaleString()} \u2192 {tokenStats.totalIn.toLocaleString()} (-{pct}%) - - ); - })()} + {tokenStats.compressed != null && + tokenStats.compressed > 0 && + (() => { + const fromTokens = tokenStats.totalIn + tokenStats.compressed; + const pct = Math.round((tokenStats.compressed / fromTokens) * 100); + return ( + + Compressed: {fromTokens.toLocaleString()} \u2192{" "} + {tokenStats.totalIn.toLocaleString()} (-{pct}%) + + ); + })()}
Output
-
+
Total Out: {formatTokenValue(tokenStats.totalOut)} @@ -427,7 +479,9 @@ export default function RequestLoggerDetail({
-
Model
+
+ Model +
{log.model}
@@ -482,6 +536,15 @@ export default function RequestLoggerDetail({ Account
{accountLabel}
+ {codexAccountRotation && ( +
${codexAccountRotation.finalConnectionId}`} + > + Rotated: {formatConnectionId(codexAccountRotation.initialConnectionId)} ->{" "} + {formatConnectionId(codexAccountRotation.finalConnectionId)} +
+ )}
@@ -504,7 +567,9 @@ export default function RequestLoggerDetail({
-
Combo
+
+ Combo +
{detail?.comboName || log.comboName ? ( {detail?.comboName || log.comboName} diff --git a/tests/unit/chatcore-attempt-logging.test.ts b/tests/unit/chatcore-attempt-logging.test.ts index d4871162f3..e41c58befa 100644 --- a/tests/unit/chatcore-attempt-logging.test.ts +++ b/tests/unit/chatcore-attempt-logging.test.ts @@ -2,8 +2,8 @@ // Characterization of persistAttemptLogs — the per-attempt call-log persistence extracted from // handleChatCore (chatCore god-file decomposition, #3501). Uses a real temp DB and polls the // persisted row (saveCallLog is async + fire-and-forget). Locks: the field mapping, the -// cacheSource semantic/upstream normalization, the connectionId → credentials.connectionId -// fallback, and error persistence. +// cacheSource semantic/upstream normalization, final credentials.connectionId attribution, +// credentials fallback, and error persistence. import { test, before, after } from "node:test"; import assert from "node:assert/strict"; import fs from "node:fs"; @@ -17,6 +17,15 @@ const coreDb = await import("../../src/lib/db/core.ts"); const { getCallLogById } = await import("../../src/lib/usage/callLogs.ts"); const { persistAttemptLogs } = await import("../../open-sse/handlers/chatCore/attemptLogging.ts"); +type CodexRotationEnvelope = { + _omniroute?: { + codexAccountRotation?: { + initialConnectionId: unknown; + finalConnectionId: unknown; + }; + }; +}; + function baseCtx(overrides: Record = {}) { return { provider: "openai", @@ -52,6 +61,11 @@ async function pollForCallLog(id: string, tries = 120) { return null; } +function getCodexAccountRotation(value: unknown) { + if (!value || typeof value !== "object") return undefined; + return (value as CodexRotationEnvelope)._omniroute?.codexAccountRotation; +} + before(async () => { await coreDb.ensureDbInitialized(); }); @@ -63,7 +77,10 @@ after(() => { test("persists a call log row with the mapped fields (default cacheSource=upstream)", async () => { const id = "attempt-basic-1"; - persistAttemptLogs({ status: 200, tokens: { input: 1, output: 2 } }, baseCtx({ pendingRequestId: id })); + persistAttemptLogs( + { status: 200, tokens: { input: 1, output: 2 } }, + baseCtx({ pendingRequestId: id, credentials: { connectionId: "conn-1" } }) + ); const row = await pollForCallLog(id); assert.ok(row, "call log row should be persisted"); assert.equal(row.status, 200); @@ -74,12 +91,34 @@ test("persists a call log row with the mapped fields (default cacheSource=upstre assert.equal(row.cacheSource, "upstream"); }); +test("uses final credentials connectionId when Codex failover rotates the account", async () => { + const id = "attempt-codex-rotation-1"; + persistAttemptLogs( + { status: 200, tokens: { input: 1, output: 2 }, responseBody: { id: "response-1" } }, + baseCtx({ + pendingRequestId: id, + provider: "codex", + connectionId: "initial-conn", + credentials: { connectionId: "final-conn" }, + }) + ); + + const row = await pollForCallLog(id); + assert.ok(row); + assert.equal(row.connectionId, "final-conn"); + assert.deepEqual(getCodexAccountRotation(row.requestBody), { + initialConnectionId: "initial-conn", + finalConnectionId: "final-conn", + }); + assert.deepEqual(getCodexAccountRotation(row.responseBody), { + initialConnectionId: "initial-conn", + finalConnectionId: "final-conn", + }); +}); + test("cacheSource 'semantic' is preserved", async () => { const id = "attempt-semantic-1"; - persistAttemptLogs( - { status: 200, cacheSource: "semantic" }, - baseCtx({ pendingRequestId: id }) - ); + persistAttemptLogs({ status: 200, cacheSource: "semantic" }, baseCtx({ pendingRequestId: id })); const row = await pollForCallLog(id); assert.ok(row); assert.equal(row.cacheSource, "semantic");