From e13496932c8a49b09ff3b5f91733ec6c40f564f6 Mon Sep 17 00:00:00 2001 From: Randi <55005611+rdself@users.noreply.github.com> Date: Mon, 15 Jun 2026 19:58:15 -0400 Subject: [PATCH] fix(stream): persist mid-stream failures (#3937) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.26 — mid-stream failure persistence (follow-up to #3879). Validated locally: typecheck:core clean, 75/75 stream/usage tests, eslint 0 errors, file-size + any-budget OK. --- open-sse/handlers/chatCore.ts | 86 +++++----- .../translator/response/claude-to-openai.ts | 30 +++- open-sse/utils/stream.ts | 76 +++++++-- open-sse/utils/streamFailureFinalization.ts | 151 ++++++++++++++++++ open-sse/utils/streamHandler.ts | 60 ++++++- open-sse/utils/usageTracking.ts | 6 + src/app/api/logs/[id]/route.ts | 3 +- src/app/api/usage/call-logs/route.ts | 4 +- src/lib/usage/usageHistory.ts | 16 ++ tests/unit/stream-handler.test.ts | 38 +++++ tests/unit/stream-utils.test.ts | 6 +- 11 files changed, 403 insertions(+), 73 deletions(-) create mode 100644 open-sse/utils/streamFailureFinalization.ts diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index fe28551589..25488919fa 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -20,6 +20,7 @@ import { ensureStreamReadiness } from "../utils/streamReadiness.ts"; import { synthesizeOpenAiSseFromJson } from "../utils/jsonToSse.ts"; import { resolveStreamReadinessTimeout } from "../utils/streamReadinessPolicy.ts"; import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts"; +import * as streamFailure from "../utils/streamFailureFinalization.ts"; import { createSseHeartbeatTransform, shapeForClientFormat } from "../utils/sseHeartbeat.ts"; import { addBufferToUsage, filterUsageForFormat, estimateUsage } from "../utils/usageTracking.ts"; import { @@ -295,10 +296,6 @@ function cloneBoundedChatLogPayload(value: unknown, depth = 0): unknown { } import { estimateSizeFast, isSmallEnoughForSemanticCache } from "../utils/estimateSize.ts"; -import { - finalizeMostRecentPendingRequest, - finalizePendingRequestById, -} from "@/lib/usage/usageHistory.ts"; const MAX_LOG_BODY_CHARS = 8 * 1024; // 8KB cap for logged request/response bodies /** @@ -3688,9 +3685,12 @@ export async function handleChatCore({ }; }; + let onPipelineStreamError: streamFailure.PipelineStreamErrorHandler | null = null; + // Create stream controller for disconnect detection const streamController = createStreamController({ onDisconnect, + onError: (event) => onPipelineStreamError?.(event), provider, model, connectionId, @@ -5495,6 +5495,8 @@ export async function handleChatCore({ (finalBody as Record | null | undefined) ?? null ); + let streamFailureCompletionRecorded = false; + // Callback to save call log when stream completes (include responseBody when provided by stream) const onStreamComplete = ({ status: streamStatus, @@ -5502,11 +5504,18 @@ export async function handleChatCore({ responseBody: streamResponseBody, providerPayload, clientPayload, + error: streamError, + errorCode: streamErrorCode, ttft, }) => { + const normalizedStreamStatus = streamStatus || 200; + if (normalizedStreamStatus !== 200) { + if (streamFailureCompletionRecorded) return; + streamFailureCompletionRecorded = true; + } const cacheUsageLogMeta = buildCacheUsageLogMeta(streamUsage); - if (streamStatus === 200) { + if (normalizedStreamStatus === 200) { void maybeSyncClaudeExtraUsageState({ provider, connectionId, @@ -5517,7 +5526,7 @@ export async function handleChatCore({ // Reasoning Replay Cache (#1628): Capture reasoning_content from streaming responses // with tool_calls so it can be replayed on subsequent turns (DeepSeek V4, Kimi K2, etc.) - if (streamStatus === 200 && streamResponseBody) { + if (normalizedStreamStatus === 200 && streamResponseBody) { try { const body = streamResponseBody as Record; const choices = body.choices as { message?: Record }[] | undefined; @@ -5532,23 +5541,17 @@ export async function handleChatCore({ } effectiveServiceTier = resolveReportedServiceTier(streamResponseBody) ?? effectiveServiceTier; - try { - const finalizedConnId = connectionId || credentials?.connectionId || null; - const completedById = finalizePendingRequestById(pendingRequestId, { - providerResponse: providerPayload ?? streamResponseBody ?? undefined, - clientResponse: clientPayload ?? streamResponseBody ?? undefined, - }); - if (!completedById) { - finalizeMostRecentPendingRequest(model, provider, finalizedConnId, { - providerResponse: providerPayload ?? streamResponseBody ?? undefined, - clientResponse: clientPayload ?? streamResponseBody ?? undefined, - }); - } - } catch (e) { - try { - console.warn("finalizeMostRecentPendingRequest failed:", e && (e.message || e)); - } catch {} - } + streamFailure.finalizeStreamRequestLog({ + pendingRequestId, + model, + provider, + connectionId: connectionId || credentials?.connectionId || null, + providerResponse: providerPayload ?? streamResponseBody ?? undefined, + clientResponse: clientPayload ?? streamResponseBody ?? undefined, + status: normalizedStreamStatus, + error: streamError, + errorCode: streamErrorCode, + }); // Track cache token metrics for streaming responses if (streamUsage && typeof streamUsage === "object") { @@ -5558,11 +5561,12 @@ export async function handleChatCore({ provider: provider || "unknown", model: model || "unknown", tokens: streamUsage, - status: String(streamStatus || 200), - success: streamStatus === 200, + status: String(normalizedStreamStatus), + success: normalizedStreamStatus === 200, latencyMs: Date.now() - startTime, timeToFirstTokenMs: ttft, - errorCode: null, + errorCode: + normalizedStreamStatus === 200 ? null : streamErrorCode || String(normalizedStreamStatus), timestamp: new Date().toISOString(), connectionId: connectionId || undefined, apiKeyId: apiKeyInfo?.id || undefined, @@ -5573,7 +5577,7 @@ export async function handleChatCore({ console.error("Failed to save usage stats:", err.message); }); - if (apiKeyInfo?.id && streamStatus === 200) { + if (apiKeyInfo?.id && normalizedStreamStatus === 200) { try { const billable = computeBillableTokens(streamUsage); if (billable > 0) @@ -5585,7 +5589,8 @@ export async function handleChatCore({ } persistAttemptLogs({ - status: streamStatus || 200, + status: normalizedStreamStatus, + error: streamError || undefined, tokens: streamUsage || {}, responseBody: streamResponseBody ?? undefined, providerRequest: finalBody || translatedBody, @@ -5608,7 +5613,7 @@ export async function handleChatCore({ // Resolve the real per-request cost (calculateCost) so USD-unit pools accrue // on streaming traffic too; this previously recorded usd:0 hardcoded, which // meant DeepSeek-style `usd/monthly` shared pools never blocked on streams. - if (apiKeyInfo?.id && credentials?.connectionId && streamStatus === 200) { + if (apiKeyInfo?.id && credentials?.connectionId && normalizedStreamStatus === 200) { const quotaApiKeyId = apiKeyInfo.id; const quotaConnectionId = credentials.connectionId; // onStreamComplete is sync — use .then() (fire-and-forget, fail-open) instead of await @@ -5621,7 +5626,7 @@ export async function handleChatCore({ provider, model, streamUsage, - streamStatus, + streamStatus: normalizedStreamStatus, serviceTier: effectiveServiceTier, }, { calculateCost, log } @@ -5681,19 +5686,14 @@ export async function handleChatCore({ } }; - const handleStreamFailure = (failure: { - status: number; - message: string; - code?: string; - type?: string; - }) => { - persistFailureUsage(failure.status || HTTP_STATUS.BAD_GATEWAY, failure.code || failure.type); - try { - onStreamFailure?.(failure); - } catch { - // Best-effort fallback state update only. - } - }; + const streamFailureFinalizers = streamFailure.createStreamFailureFinalizers({ + isFailureCompletionRecorded: () => streamFailureCompletionRecorded, + onStreamComplete, + persistFailureUsage, + onStreamFailure, + }); + const handleStreamFailure = streamFailureFinalizers.handleStreamFailure; + onPipelineStreamError = streamFailureFinalizers.onPipelineStreamError; // For providers using Responses API format, translate stream back to openai (Chat Completions) format // UNLESS client is Droid CLI which expects openai-responses format back diff --git a/open-sse/translator/response/claude-to-openai.ts b/open-sse/translator/response/claude-to-openai.ts index 76126cface..22af8213b6 100644 --- a/open-sse/translator/response/claude-to-openai.ts +++ b/open-sse/translator/response/claude-to-openai.ts @@ -115,6 +115,21 @@ export function claudeToOpenAIResponse(chunk, state) { // Extract usage from message_delta event (Claude native format) // Normalize to OpenAI format (prompt_tokens/completion_tokens) for consistent logging if (chunk.usage && typeof chunk.usage === "object") { + const previousUsage = state.usage && typeof state.usage === "object" ? state.usage : {}; + const previousInputTokens = + typeof previousUsage.input_tokens === "number" + ? previousUsage.input_tokens + : typeof previousUsage.prompt_tokens === "number" + ? previousUsage.prompt_tokens + : 0; + const previousCacheReadTokens = + typeof previousUsage.cache_read_input_tokens === "number" + ? previousUsage.cache_read_input_tokens + : 0; + const previousCacheCreationTokens = + typeof previousUsage.cache_creation_input_tokens === "number" + ? previousUsage.cache_creation_input_tokens + : 0; const inputTokens = typeof chunk.usage.input_tokens === "number" ? chunk.usage.input_tokens : 0; const outputTokens = @@ -136,7 +151,10 @@ export function claudeToOpenAIResponse(chunk, state) { // minimum, so a 2-token "hi" can be reported as ~2008 prompt_tokens and // inflate downstream billing ~250x. cache_creation is still exposed // separately via prompt_tokens_details.cache_creation_tokens below. - const billableInputTokens = inputTokens + cacheReadTokens; + const billableInputTokens = + inputTokens > 0 || cacheReadTokens > 0 || cacheCreationTokens > 0 + ? inputTokens + cacheReadTokens + : previousInputTokens; state.usage = { prompt_tokens: billableInputTokens, completion_tokens: outputTokens, @@ -145,11 +163,13 @@ export function claudeToOpenAIResponse(chunk, state) { }; // Store cache tokens if present (needed for prompt_tokens_details in final chunk) - if (cacheReadTokens > 0) { - state.usage.cache_read_input_tokens = cacheReadTokens; + const effectiveCacheReadTokens = cacheReadTokens || previousCacheReadTokens; + const effectiveCacheCreationTokens = cacheCreationTokens || previousCacheCreationTokens; + if (effectiveCacheReadTokens > 0) { + state.usage.cache_read_input_tokens = effectiveCacheReadTokens; } - if (cacheCreationTokens > 0) { - state.usage.cache_creation_input_tokens = cacheCreationTokens; + if (effectiveCacheCreationTokens > 0) { + state.usage.cache_creation_input_tokens = effectiveCacheCreationTokens; } } diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 52641d1501..d88c161772 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -93,6 +93,9 @@ type StreamCompletePayload = { responseBody?: unknown; providerPayload?: unknown; clientPayload?: unknown; + error?: string | null; + errorCode?: string | null; + ttft?: number | null; }; type StreamFailurePayload = { @@ -116,7 +119,7 @@ type StreamOptions = { apiKeyInfo?: unknown; body?: unknown; onComplete?: ((payload: StreamCompletePayload) => void) | null; - onFailure?: ((payload: StreamFailurePayload) => void | Promise) | null; + onFailure?: ((payload: StreamFailurePayload) => boolean | void | Promise) | null; }; type TranslateState = ReturnType & { @@ -846,12 +849,13 @@ export function createSSEStream(options: StreamOptions = {}) { reqLogger?.appendConvertedChunk?.(errOutput); clientPayloadCollector.push(errorEvent); controller.enqueue(encoder.encode(errOutput)); + let failureHandled = false; if (onFailure) { try { - void onFailure({ status: 502, message: msg, code: "empty_response" }); + failureHandled = onFailure({ status: 502, message: msg, code: "empty_response" }) === true; } catch {} } - if (decrementPendingRequest) { + if (decrementPendingRequest && !failureHandled) { clearPendingRequestFromStream(); } controller.error(markPendingRequestCleared(new Error(msg))); @@ -1063,7 +1067,21 @@ export function createSSEStream(options: StreamOptions = {}) { clearIdleTimer(); const timeoutMsg = `[STREAM] Idle timeout: no data from ${provider || "provider"} for ${STREAM_IDLE_TIMEOUT_MS}ms (model: ${model || "unknown"})`; console.warn(timeoutMsg); - clearPendingRequestFromStream(); + let failureHandled = false; + if (onFailure) { + try { + failureHandled = + onFailure({ + status: HTTP_STATUS.GATEWAY_TIMEOUT, + message: timeoutMsg, + code: "stream_idle_timeout", + type: "timeout_error", + }) === true; + } catch {} + } + if (!failureHandled) { + clearPendingRequestFromStream(); + } appendRequestLog({ model, provider, @@ -1766,13 +1784,16 @@ export function createSSEStream(options: StreamOptions = {}) { reqLogger?.appendConvertedChunk?.(output); controller.enqueue(encoder.encode(output)); if (failurePayload) { + let failureHandled = false; if (onFailure) { try { - void onFailure(failurePayload); + failureHandled = onFailure(failurePayload) === true; } catch {} } clearIdleTimer(); - clearPendingRequestFromStream(); + if (!failureHandled) { + clearPendingRequestFromStream(); + } controller.error( markPendingRequestCleared(new Error(failurePayload.message || "Upstream failure")) ); @@ -1929,7 +1950,25 @@ export function createSSEStream(options: StreamOptions = {}) { // Extract usage const extracted = extractUsage(parsed); - if (extracted) state.usage = extracted; // Keep original usage for logging + if (extracted) { + if (!state.usage) { + state.usage = extracted; + } else { + const su = state.usage as Record; + const eu = extracted as Record; + if (eu.prompt_tokens > 0) su.prompt_tokens = eu.prompt_tokens; + if (eu.completion_tokens > 0) su.completion_tokens = eu.completion_tokens; + if (eu.total_tokens > 0) su.total_tokens = eu.total_tokens; + if (eu.input_tokens > 0) su.input_tokens = eu.input_tokens; + if (eu.output_tokens > 0) su.output_tokens = eu.output_tokens; + if (eu.cache_read_input_tokens > 0) + su.cache_read_input_tokens = eu.cache_read_input_tokens; + if (eu.cache_creation_input_tokens > 0) + su.cache_creation_input_tokens = eu.cache_creation_input_tokens; + if (eu.cached_tokens > 0) su.cached_tokens = eu.cached_tokens; + if (eu.reasoning_tokens > 0) su.reasoning_tokens = eu.reasoning_tokens; + } + } // Translate: targetFormat -> openai -> sourceFormat const translated = translateResponse(targetFormat, sourceFormat, parsed, state); @@ -2309,6 +2348,8 @@ export function createSSEStream(options: StreamOptions = {}) { if (eu.prompt_tokens > 0) su.prompt_tokens = eu.prompt_tokens; if (eu.completion_tokens > 0) su.completion_tokens = eu.completion_tokens; if (eu.total_tokens > 0) su.total_tokens = eu.total_tokens; + if (eu.input_tokens > 0) su.input_tokens = eu.input_tokens; + if (eu.output_tokens > 0) su.output_tokens = eu.output_tokens; if (eu.cache_read_input_tokens > 0) su.cache_read_input_tokens = eu.cache_read_input_tokens; if (eu.cache_creation_input_tokens > 0) @@ -2336,15 +2377,16 @@ export function createSSEStream(options: StreamOptions = {}) { if (state?.upstreamError) { const err = state.upstreamError; - clearPendingRequestFromStream(); + let failureHandled = false; if (onFailure) { try { - void onFailure({ - status: err.status, - message: err.message, - code: err.code, - type: err.type, - }); + failureHandled = + onFailure({ + status: err.status, + message: err.message, + code: err.code, + type: err.type, + }) === true; } catch {} } @@ -2355,6 +2397,8 @@ export function createSSEStream(options: StreamOptions = {}) { status: err.status, usage: state?.usage, responseBody: errorBody, + error: err.message, + errorCode: err.code, providerPayload: providerPayloadCollector.build( buildStreamSummaryFromEvents( providerPayloadCollector.getEvents(), @@ -2367,10 +2411,14 @@ export function createSSEStream(options: StreamOptions = {}) { includeEvents: false, }), }); + failureHandled = true; } catch {} } clearIdleTimer(); + if (!failureHandled) { + clearPendingRequestFromStream(); + } controller.error( markPendingRequestCleared(new Error(err.message || "Upstream failure")) ); diff --git a/open-sse/utils/streamFailureFinalization.ts b/open-sse/utils/streamFailureFinalization.ts new file mode 100644 index 0000000000..fc8a339ce7 --- /dev/null +++ b/open-sse/utils/streamFailureFinalization.ts @@ -0,0 +1,151 @@ +import { + finalizeMostRecentPendingRequest, + finalizePendingRequestById, +} from "@/lib/usage/usageHistory.ts"; + +import { HTTP_STATUS } from "../config/constants.ts"; +import { buildErrorBody } from "./error.ts"; + +export type StreamCompletionPayload = { + status: number; + usage: unknown; + responseBody?: unknown; + providerPayload?: unknown; + clientPayload?: unknown; + error?: string | null; + errorCode?: string | null; + ttft?: number | null; +}; + +export type StreamFailurePayload = { + status: number; + message: string; + code?: string; + type?: string; +}; + +export type PipelineStreamErrorHandler = (event: { + message: string; + statusCode: number; +}) => boolean; + +export function finalizeStreamRequestLog({ + pendingRequestId, + model, + provider, + connectionId, + providerResponse, + clientResponse, + status, + error, + errorCode, + onWarn, +}: { + pendingRequestId: string; + model: string; + provider: string; + connectionId: string | null; + providerResponse?: unknown; + clientResponse?: unknown; + status: number; + error?: string | null; + errorCode?: string | null; + onWarn?: (error: unknown) => void; +}) { + try { + const completedById = finalizePendingRequestById(pendingRequestId, { + providerResponse, + clientResponse, + status, + error: error || null, + errorCode: errorCode || null, + }); + if (!completedById) { + finalizeMostRecentPendingRequest(model, provider, connectionId, { + providerResponse, + clientResponse, + status, + error: error || null, + errorCode: errorCode || null, + }); + } + } catch (error) { + try { + if (onWarn) { + onWarn(error); + } else { + console.warn( + "finalizeMostRecentPendingRequest failed:", + error && typeof error === "object" && "message" in error + ? (error as { message?: unknown }).message + : error + ); + } + } catch {} + } +} + +export function createStreamFailureFinalizers({ + isFailureCompletionRecorded, + onStreamComplete, + persistFailureUsage, + onStreamFailure, +}: { + isFailureCompletionRecorded: () => boolean; + onStreamComplete: (payload: StreamCompletionPayload) => void; + persistFailureUsage: (status: number, errorCode?: string) => void; + onStreamFailure?: ((failure: StreamFailurePayload) => void) | null; +}) { + const handleStreamFailure = (failure: StreamFailurePayload) => { + const status = failure.status || HTTP_STATUS.BAD_GATEWAY; + const message = failure.message || "Upstream stream error"; + const code = failure.code || failure.type || String(status); + + if (!isFailureCompletionRecorded()) { + const errorBody = buildErrorBody(status, message); + onStreamComplete({ + status, + usage: null, + responseBody: errorBody, + providerPayload: errorBody, + clientPayload: errorBody, + error: message, + errorCode: code, + ttft: 0, + }); + } + + persistFailureUsage(status, code); + try { + onStreamFailure?.(failure); + } catch { + // Best-effort fallback state update only. + } + return true; + }; + + let pipelineStreamFailureFinalized = false; + const onPipelineStreamError: PipelineStreamErrorHandler = ({ message, statusCode }) => { + if (pipelineStreamFailureFinalized) return true; + pipelineStreamFailureFinalized = true; + + const status = + Number.isFinite(statusCode) && statusCode >= 400 && statusCode <= 599 + ? statusCode + : HTTP_STATUS.BAD_GATEWAY; + const normalizedMessage = message || "Upstream stream error"; + const code = normalizedMessage.toLowerCase().includes("terminated") + ? "stream_terminated" + : "stream_pipeline_error"; + + handleStreamFailure({ + status, + message: normalizedMessage, + code, + type: "stream_error", + }); + return true; + }; + + return { handleStreamFailure, onPipelineStreamError }; +} diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 762b2747ae..634eb61592 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -11,8 +11,16 @@ type StreamDisconnectEvent = { duration: number; }; +type StreamErrorEvent = { + error: unknown; + message: string; + statusCode: number; + duration: number; +}; + type StreamControllerOptions = { onDisconnect?: (event: StreamDisconnectEvent) => void; + onError?: (event: StreamErrorEvent) => boolean | void; provider?: string; model?: string; connectionId?: string | null; @@ -116,6 +124,30 @@ function getTimeString() { }); } +function isPendingRequestClearedError(error: unknown): boolean { + return ( + !!error && + typeof error === "object" && + (error as Record)[PENDING_REQUEST_CLEARED_MARKER] === true + ); +} + +function getErrorMessage(error: unknown): string { + if (error instanceof Error && error.message) return error.message; + if (typeof error === "string" && error.trim().length > 0) return error; + return "Upstream stream error"; +} + +function getErrorStatusCode(error: unknown): number { + if (error && typeof error === "object" && "statusCode" in error) { + const statusCode = Number((error as { statusCode?: unknown }).statusCode); + if (Number.isFinite(statusCode) && statusCode >= 400 && statusCode <= 599) { + return statusCode; + } + } + return 502; +} + /** * Create stream controller with abort and disconnect detection * @param {object} options @@ -127,6 +159,7 @@ function getTimeString() { /** @param {StreamControllerOptions} options */ export function createStreamController({ onDisconnect, + onError, provider, model, connectionId, @@ -209,7 +242,25 @@ export function createStreamController({ abortTimeout = null; } - clearPendingRequest(error); + const alreadyCleared = isPendingRequestClearedError(error); + let handled = false; + if (!alreadyCleared) { + try { + handled = + onError?.({ + error, + message: getErrorMessage(error), + statusCode: getErrorStatusCode(error), + duration: Date.now() - startTime, + }) === true; + } catch {} + } + + if (!handled) { + clearPendingRequest(error); + } else { + pendingRequestCleared = true; + } if (error instanceof Error && error.name === "AbortError") { logStream("aborted"); @@ -312,11 +363,8 @@ export function createDisconnectAwareStream(transformStream, streamController) { // T35: Encapsulate mid-stream errors as SSE events instead of abruptly aborting // This prevents TransferEncodingError on the client side - const errorMsg = error instanceof Error ? error.message : "Upstream stream error"; - const statusCode = - typeof error === "object" && error !== null && "statusCode" in error - ? Number((error as { statusCode?: unknown }).statusCode) || 500 - : 500; + const errorMsg = getErrorMessage(error); + const statusCode = getErrorStatusCode(error); for (const chunk of buildStreamErrorChunks( errorMsg, diff --git a/open-sse/utils/usageTracking.ts b/open-sse/utils/usageTracking.ts index b70dcd631c..8e3e969c4e 100644 --- a/open-sse/utils/usageTracking.ts +++ b/open-sse/utils/usageTracking.ts @@ -264,6 +264,8 @@ export function normalizeUsage(usage) { assignNumber("prompt_tokens", usage?.prompt_tokens); assignNumber("completion_tokens", usage?.completion_tokens); assignNumber("total_tokens", usage?.total_tokens); + assignNumber("input_tokens", usage?.input_tokens); + assignNumber("output_tokens", usage?.output_tokens); assignNumber("cache_read_input_tokens", usage?.cache_read_input_tokens); assignNumber("cache_creation_input_tokens", usage?.cache_creation_input_tokens); assignNumber("cached_tokens", usage?.cached_tokens); @@ -323,6 +325,8 @@ export function extractUsage(chunk) { return normalizeUsage({ prompt_tokens: inputTokens + cacheRead + cacheCreation, completion_tokens: u.output_tokens || u.completion_tokens || 0, + input_tokens: inputTokens + cacheRead + cacheCreation, + output_tokens: u.output_tokens || u.completion_tokens || 0, cache_read_input_tokens: u.cache_read_input_tokens, cache_creation_input_tokens: u.cache_creation_input_tokens, }); @@ -337,6 +341,8 @@ export function extractUsage(chunk) { return normalizeUsage({ prompt_tokens: deltaInput + deltaCacheRead + deltaCacheCreation, completion_tokens: chunk.usage.output_tokens || 0, + input_tokens: deltaInput + deltaCacheRead + deltaCacheCreation, + output_tokens: chunk.usage.output_tokens || 0, cache_read_input_tokens: chunk.usage.cache_read_input_tokens, cache_creation_input_tokens: chunk.usage.cache_creation_input_tokens, }); diff --git a/src/app/api/logs/[id]/route.ts b/src/app/api/logs/[id]/route.ts index 5fb9c62bbe..d88ee3c3f1 100644 --- a/src/app/api/logs/[id]/route.ts +++ b/src/app/api/logs/[id]/route.ts @@ -75,13 +75,14 @@ export async function GET( id: inMem.id, timestamp: new Date(inMem.startedAt).toISOString(), path: inMem.clientEndpoint || "", - status: 0, + status: typeof inMem.status === "number" ? inMem.status : inMem.error ? 502 : 0, model: inMem.model, provider: inMem.provider, connectionId: inMem.connectionId, duration: Date.now() - inMem.startedAt, detailState: "in-memory", active: false, + error: inMem.error || null, pipelinePayloads, hasPipelineDetails: true, }; diff --git a/src/app/api/usage/call-logs/route.ts b/src/app/api/usage/call-logs/route.ts index 987e7478b3..5a469cb59e 100644 --- a/src/app/api/usage/call-logs/route.ts +++ b/src/app/api/usage/call-logs/route.ts @@ -81,7 +81,7 @@ export function buildCallLogListRows({ timestamp: new Date(detail.startedAt).toISOString(), method: "", path: detail.clientEndpoint || "", - status: 200, + status: typeof detail.status === "number" ? detail.status : detail.error ? 502 : 200, model: detail.model, requestedModel: null, provider: detail.provider, @@ -95,7 +95,7 @@ export function buildCallLogListRows({ apiKeyId: null, apiKeyName: null, comboName: null, - error: null, + error: detail.error || null, active: false, completed: true, completedAt: completedAt ? new Date(completedAt).toISOString() : null, diff --git a/src/lib/usage/usageHistory.ts b/src/lib/usage/usageHistory.ts index c3818b0686..3a30df601d 100644 --- a/src/lib/usage/usageHistory.ts +++ b/src/lib/usage/usageHistory.ts @@ -33,6 +33,9 @@ type PendingRequestMetadata = { providerUrl?: string | null; providerResponse?: unknown; clientResponse?: unknown; + status?: number | null; + error?: string | null; + errorCode?: string | null; stage?: string | null; stageUpdatedAt?: number | null; }; @@ -48,6 +51,9 @@ export type PendingRequestDetail = { providerUrl?: string | null; providerResponse?: unknown; clientResponse?: unknown; + status?: number | null; + error?: string | null; + errorCode?: string | null; completedAt?: number | null; durationMs?: number | null; stage?: string | null; @@ -177,6 +183,16 @@ function normalizePendingMetadata(metadata?: PendingRequestMetadata): PendingReq protectPayloadForLog(metadata.clientResponse) ); } + if (metadata.status !== undefined) { + const status = Number(metadata.status); + normalized.status = Number.isFinite(status) ? status : null; + } + if (metadata.error !== undefined) { + normalized.error = toStringOrNull(metadata.error) || null; + } + if (metadata.errorCode !== undefined) { + normalized.errorCode = toStringOrNull(metadata.errorCode) || null; + } return normalized; } diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index c0b323a9ab..a64423425e 100644 --- a/tests/unit/stream-handler.test.ts +++ b/tests/unit/stream-handler.test.ts @@ -399,6 +399,44 @@ test("pipeWithDisconnect clears pending requests when the upstream stream errors assert.equal(pending.details[connectionId], undefined); }); +test("pipeWithDisconnect lets controller onError own pending cleanup", async () => { + clearPendingRequests(); + const provider = "openai"; + const model = "gpt-stream-error-owned"; + const connectionId = "conn-stream-error-owned"; + const modelKey = `${model} (${provider})`; + let errorEvent = null; + + trackPendingRequest(model, provider, connectionId, true); + + const source = new ReadableStream({ + start(controller) { + controller.error(Object.assign(new Error("terminated"), { statusCode: 502 })); + }, + }); + const stream = pipeWithDisconnect( + new Response(source), + new TransformStream(), + createStreamController({ + provider, + model, + connectionId, + onError(event) { + errorEvent = event; + return true; + }, + }) + ); + + const text = await readStreamText(stream); + const pending = getPendingRequests(); + + assert.match(text, /"message":"terminated"/); + assert.equal(errorEvent?.statusCode, 502); + assert.equal(pending.byModel[modelKey], 1); + assert.equal(pending.byAccount[connectionId][modelKey], 1); +}); + test("pipeWithDisconnect does not double-clear transform errors already accounted for", async () => { clearPendingRequests(); const provider = "openai"; diff --git a/tests/unit/stream-utils.test.ts b/tests/unit/stream-utils.test.ts index ce5ba7988d..6159cdc12f 100644 --- a/tests/unit/stream-utils.test.ts +++ b/tests/unit/stream-utils.test.ts @@ -798,8 +798,9 @@ test("createSSEStream translate mode converts Claude SSE into OpenAI chunks and assert.match(text, /\[DONE\]/); assert.equal(onCompletePayload.status, 200); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello Claude"); + assert.equal(onCompletePayload.responseBody.usage.prompt_tokens, 3); assert.equal(onCompletePayload.responseBody.usage.completion_tokens, 4); - assert.equal(onCompletePayload.responseBody.usage.total_tokens, 4); + assert.equal(onCompletePayload.responseBody.usage.total_tokens, 7); }); test("createSSEStream translate mode preserves Claude text_delta thinking tags as content", async () => { @@ -1807,8 +1808,9 @@ test("createSSETransformStreamWithLogger flushes a trailing Claude usage event w assert.match(text, /Buffered tail/); assert.match(text, /\[DONE\]/); assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Buffered tail"); + assert.equal(onCompletePayload.responseBody.usage.prompt_tokens, 3); assert.equal(onCompletePayload.responseBody.usage.completion_tokens, 5); - assert.equal(onCompletePayload.responseBody.usage.total_tokens, 5); + assert.equal(onCompletePayload.responseBody.usage.total_tokens, 8); }); test("buildStreamSummaryFromEvents compacts Responses API deltas into a synthetic response", () => {