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

fix(stream): persist mid-stream failures (#3937)

Browse Source 
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.
This commit is contained in:
Randi
2026-06-15 19:58:15 -04:00
committed by GitHub
parent fe182d102f
commit e13496932c
11 changed files with 403 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<string, unknown> | 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<string, unknown>;
const choices = body.choices as { message?: Record<string, unknown> }[] | 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

30
open-sse/translator/response/claude-to-openai.ts
View File

@@ -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;
}
}

76
open-sse/utils/stream.ts
View File

@@ -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<void>) | null;
onFailure?: ((payload: StreamFailurePayload) => boolean | void | Promise<void>) | null;
};
type TranslateState = ReturnType<typeof initState> & {
@@ -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<string, number>;
const eu = extracted as Record<string, number>;
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"))
);

151
open-sse/utils/streamFailureFinalization.ts Normal file
View File

@@ -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 };
}

60
open-sse/utils/streamHandler.ts
View File

@@ -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<string, unknown>)[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,

6
open-sse/utils/usageTracking.ts
View File

@@ -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,
});

3
src/app/api/logs/[id]/route.ts
View File

@@ -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,
};

4
src/app/api/usage/call-logs/route.ts
View File

@@ -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,

16
src/lib/usage/usageHistory.ts
View File

@@ -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;
}

38
tests/unit/stream-handler.test.ts
View File

@@ -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";

6
tests/unit/stream-utils.test.ts
View File

@@ -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", () => {