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

Track final connection IDs in failover logs (#5016)

Browse Source 
Integrated into release/v3.8.38
This commit is contained in:
Jan Leon
2026-06-27 03:04:39 +02:00
committed by GitHub
parent b8070ed3a8
commit 25b65ac34a
4 changed files with 349 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -390,6 +390,13 @@ export async function handleChatCore({
// call sites stay byte-identical.
const trace = (label: string, extra?: Record<string, unknown>) =>
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<string, string> | 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<Uint8Array>,
() => 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<string, string>
);
@@ -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<string, unknown> | 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,

40
open-sse/handlers/chatCore/attemptLogging.ts
View File

@@ -35,10 +35,7 @@ export type PersistAttemptLogsContext = {
model: string | null | undefined;
skillRequestId: string;
detailedLoggingEnabled: boolean;
reqLogger:
| { getPipelinePayloads?: () => Record<string, unknown> | undefined }
| null
| undefined;
reqLogger: { getPipelinePayloads?: () => Record<string, unknown> | 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<string, unknown>), {
...accountRotationMeta,
claudePromptCache: claudeCacheMeta,
})
),
responseBody: cloneBoundedChatLogPayload(
attachLogMeta(truncateForLog(responseBody as Record<string, unknown>), {
...accountRotationMeta,
claudePromptCache: claudeCacheMeta
? {
applied: claudeCacheMeta.applied,

109
src/shared/components/RequestLoggerDetail.tsx
View File

@@ -26,14 +26,18 @@ function PayloadSection({ title, json, onCopy, collapsible = true, defaultOpen =
<div>
<div className="flex items-center justify-between mb-2">
<div className="flex items-center gap-3">
<h3 className="text-[11px] text-text-muted uppercase tracking-wider font-bold">{title}</h3>
<h3 className="text-[11px] text-text-muted uppercase tracking-wider font-bold">
{title}
</h3>
{collapsible && (
<button
onClick={() => setOpen((v) => !v)}
className="p-1 rounded hover:bg-bg-subtle text-text-muted hover:text-text-primary transition-colors"
aria-label={open ? `Collapse ${title}` : `Expand ${title}`}
>
<span className="material-symbols-outlined text-[16px]">{open ? "expand_less" : "expand_more"}</span>
<span className="material-symbols-outlined text-[16px]">
{open ? "expand_less" : "expand_more"}
</span>
</button>
)}
</div>
@@ -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}`}
>
<span className="material-symbols-outlined text-[14px]">{copied ? "check" : "content_copy"}</span>
<span className="material-symbols-outlined text-[14px]">
{copied ? "check" : "content_copy"}
</span>
{copied ? "Copied!" : "Copy"}
</button>
</div>
@@ -136,6 +142,30 @@ function StreamSection({ title, json, onCopy }) {
type StreamChunks = Record<string, string | string[]>;
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 (
<div
className="fixed inset-0 z-50 flex items-start justify-center pt-[5vh]"
@@ -348,19 +379,27 @@ export default function RequestLoggerDetail({
{log.active ? (
<div className="flex flex-wrap gap-4 p-4 bg-bg-subtle rounded-xl border border-border">
<div className="min-w-[140px] flex-1">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Started At</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Started At
</div>
<div className="text-sm font-medium">{formatDate(log.timestamp)}</div>
</div>
<div className="min-w-[100px] flex-1">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Duration</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Duration
</div>
<div className="text-sm font-medium">{formatDuration(log.duration)}</div>
</div>
<div className="min-w-[140px] flex-1">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Model</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Model
</div>
<div className="text-sm font-medium text-primary font-mono">{log.model}</div>
</div>
<div className="min-w-[120px] flex-1">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Provider</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Provider
</div>
<span
className="inline-block px-2.5 py-1 rounded text-[10px] font-bold uppercase"
style={{ backgroundColor: providerColor.bg, color: providerColor.text }}
@@ -369,7 +408,9 @@ export default function RequestLoggerDetail({
</span>
</div>
<div className="min-w-[120px] flex-1">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Account</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Account
</div>
<div className="text-sm font-medium">{accountLabel}</div>
</div>
</div>
@@ -391,8 +432,13 @@ export default function RequestLoggerDetail({
<div className="text-sm font-medium">{formatDuration(log.duration)}</div>
</div>
<div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Input</div>
<div className="flex flex-wrap items-center gap-1.5" data-testid="token-group-input">
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Input
</div>
<div
className="flex flex-wrap items-center gap-1.5"
data-testid="token-group-input"
>
<span className="px-2 py-0.5 rounded bg-primary/20 text-primary text-xs font-bold">
Total In: {formatTokenValue(tokenStats.totalIn)}
</span>
@@ -402,22 +448,28 @@ export default function RequestLoggerDetail({
<span className="px-2 py-0.5 rounded bg-amber-500/20 text-amber-700 dark:text-amber-400 text-xs font-bold">
Cache Write: {formatTokenValue(tokenStats.cacheWrite)}
</span>
{tokenStats.compressed != null && tokenStats.compressed > 0 && (() => {
const fromTokens = tokenStats.totalIn + tokenStats.compressed;
const pct = Math.round((tokenStats.compressed / fromTokens) * 100);
return (
<span className="px-2 py-0.5 rounded bg-purple-500/20 text-purple-700 dark:text-purple-300 text-xs font-bold">
Compressed: {fromTokens.toLocaleString()} \u2192 {tokenStats.totalIn.toLocaleString()} (-{pct}%)
</span>
);
})()}
{tokenStats.compressed != null &&
tokenStats.compressed > 0 &&
(() => {
const fromTokens = tokenStats.totalIn + tokenStats.compressed;
const pct = Math.round((tokenStats.compressed / fromTokens) * 100);
return (
<span className="px-2 py-0.5 rounded bg-purple-500/20 text-purple-700 dark:text-purple-300 text-xs font-bold">
Compressed: {fromTokens.toLocaleString()} \u2192{" "}
{tokenStats.totalIn.toLocaleString()} (-{pct}%)
</span>
);
})()}
</div>
</div>
<div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Output
</div>
<div className="flex flex-wrap items-center gap-1.5" data-testid="token-group-output">
<div
className="flex flex-wrap items-center gap-1.5"
data-testid="token-group-output"
>
<span className="px-2 py-0.5 rounded bg-emerald-500/20 text-emerald-700 dark:text-emerald-400 text-xs font-bold">
Total Out: {formatTokenValue(tokenStats.totalOut)}
</span>
@@ -427,7 +479,9 @@ export default function RequestLoggerDetail({
</div>
</div>
<div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Model</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Model
</div>
<div className="text-sm font-medium text-primary font-mono">{log.model}</div>
</div>
<div>
@@ -482,6 +536,15 @@ export default function RequestLoggerDetail({
Account
</div>
<div className="text-sm font-medium">{accountLabel}</div>
{codexAccountRotation && (
<div
className="mt-1 text-[10px] text-amber-600 dark:text-amber-400 font-mono"
title={`${codexAccountRotation.initialConnectionId} -> ${codexAccountRotation.finalConnectionId}`}
>
Rotated: {formatConnectionId(codexAccountRotation.initialConnectionId)} -&gt;{" "}
{formatConnectionId(codexAccountRotation.finalConnectionId)}
</div>
)}
</div>
<div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
@@ -504,7 +567,9 @@ export default function RequestLoggerDetail({
</div>
</div>
<div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">Combo</div>
<div className="text-[10px] text-text-muted uppercase tracking-wider mb-1">
Combo
</div>
{detail?.comboName || log.comboName ? (
<span className="inline-block px-2.5 py-1 rounded-full text-[10px] font-bold bg-violet-500/20 text-violet-700 dark:text-violet-300 border border-violet-500/30">
{detail?.comboName || log.comboName}

53
tests/unit/chatcore-attempt-logging.test.ts
View File

@@ -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<string, unknown> = {}) {
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");