From 8e19f91acde95712061fe4283453b99f8e238733 Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 1 Aug 2026 22:12:38 +0800 Subject: [PATCH 1/2] fix(proxy): enforce structured stream fallback boundaries --- messages/en/dashboard.json | 2 + messages/ja/dashboard.json | 2 + messages/ru/dashboard.json | 2 + messages/zh-CN/dashboard.json | 2 + messages/zh-TW/dashboard.json | 2 + src/app/v1/_lib/proxy/discovery-validity.ts | 213 +++------ src/app/v1/_lib/proxy/forwarder.ts | 4 +- src/app/v1/_lib/proxy/response-handler.ts | 173 ++++++- .../proxy/stream-gate/frame-classifier.ts | 43 +- .../v1/_lib/proxy/stream-gate/sse-frames.ts | 133 +++++- .../stream-gate/stream-protocol-observer.ts | 47 +- .../integration/proxy-hedge-lifecycle.test.ts | 236 ++++++++++ tests/unit/i18n/pricing-source-keys.test.ts | 45 ++ tests/unit/proxy/discovery-validity.test.ts | 45 +- ...handler-endpoint-circuit-isolation.test.ts | 11 +- .../response-handler-stream-terminal.test.ts | 433 ++++++++++++++++++ .../stream-gate-frame-classifier.test.ts | 65 ++- .../stream-gate-protocol-observer.test.ts | 89 +++- .../unit/proxy/stream-gate-sse-frames.test.ts | 45 ++ 19 files changed, 1358 insertions(+), 234 deletions(-) create mode 100644 tests/unit/i18n/pricing-source-keys.test.ts diff --git a/messages/en/dashboard.json b/messages/en/dashboard.json index 5fbc5d622..01be843ed 100644 --- a/messages/en/dashboard.json +++ b/messages/en/dashboard.json @@ -319,6 +319,7 @@ "local_manual": "Local manual pricing", "cloud_exact": "Exact cloud provider pricing", "cloud_model_fallback": "Fallback model provider pricing", + "cloud_official": "Official cloud pricing", "priority_fallback": "Priority provider fallback pricing", "single_provider_top_level": "Top-level single-provider pricing", "official_fallback": "Official pricing" @@ -594,6 +595,7 @@ "local_manual": "Local manual pricing", "cloud_exact": "Exact cloud provider pricing", "cloud_model_fallback": "Fallback model provider pricing", + "cloud_official": "Official cloud pricing", "priority_fallback": "Priority provider fallback pricing", "single_provider_top_level": "Top-level single-provider pricing", "official_fallback": "Official pricing" diff --git a/messages/ja/dashboard.json b/messages/ja/dashboard.json index 2176634b5..b61809d75 100644 --- a/messages/ja/dashboard.json +++ b/messages/ja/dashboard.json @@ -319,6 +319,7 @@ "local_manual": "ローカル手動価格", "cloud_exact": "クラウドの正確なプロバイダー価格", "cloud_model_fallback": "フォールバックモデルのプロバイダー価格", + "cloud_official": "クラウド公式価格", "priority_fallback": "優先フォールバック価格", "single_provider_top_level": "トップレベル単一プロバイダー価格", "official_fallback": "公式価格" @@ -594,6 +595,7 @@ "local_manual": "ローカル手動価格", "cloud_exact": "クラウドの正確なプロバイダー価格", "cloud_model_fallback": "フォールバックモデルのプロバイダー価格", + "cloud_official": "クラウド公式価格", "priority_fallback": "優先フォールバック価格", "single_provider_top_level": "トップレベル単一プロバイダー価格", "official_fallback": "公式価格" diff --git a/messages/ru/dashboard.json b/messages/ru/dashboard.json index 454fe0803..aab97a56f 100644 --- a/messages/ru/dashboard.json +++ b/messages/ru/dashboard.json @@ -319,6 +319,7 @@ "local_manual": "Локальная ручная цена", "cloud_exact": "Точная облачная цена провайдера", "cloud_model_fallback": "Цена провайдера из резервной модели", + "cloud_official": "Официальная облачная цена", "priority_fallback": "Приоритетная резервная цена", "single_provider_top_level": "Цена верхнего уровня для одного провайдера", "official_fallback": "Официальная цена" @@ -594,6 +595,7 @@ "local_manual": "Локальная ручная цена", "cloud_exact": "Точная облачная цена провайдера", "cloud_model_fallback": "Цена провайдера из резервной модели", + "cloud_official": "Официальная облачная цена", "priority_fallback": "Приоритетная резервная цена", "single_provider_top_level": "Цена верхнего уровня для одного провайдера", "official_fallback": "Официальная цена" diff --git a/messages/zh-CN/dashboard.json b/messages/zh-CN/dashboard.json index f314c58ec..ff6a925c1 100644 --- a/messages/zh-CN/dashboard.json +++ b/messages/zh-CN/dashboard.json @@ -319,6 +319,7 @@ "local_manual": "本地手动价格", "cloud_exact": "云端精确提供商价格", "cloud_model_fallback": "回退模型的提供商价格", + "cloud_official": "云端官方价格", "priority_fallback": "优先级回退价格", "single_provider_top_level": "顶层单提供商价格", "official_fallback": "官方价格" @@ -594,6 +595,7 @@ "local_manual": "本地手动价格", "cloud_exact": "云端精确提供商价格", "cloud_model_fallback": "回退模型的提供商价格", + "cloud_official": "云端官方价格", "priority_fallback": "优先级回退价格", "single_provider_top_level": "顶层单提供商价格", "official_fallback": "官方价格" diff --git a/messages/zh-TW/dashboard.json b/messages/zh-TW/dashboard.json index e8e5fee6f..e3825d2d2 100644 --- a/messages/zh-TW/dashboard.json +++ b/messages/zh-TW/dashboard.json @@ -319,6 +319,7 @@ "local_manual": "本地手動價格", "cloud_exact": "雲端精確供應商價格", "cloud_model_fallback": "回退模型的供應商價格", + "cloud_official": "雲端官方價格", "priority_fallback": "優先級回退價格", "single_provider_top_level": "頂層單供應商價格", "official_fallback": "官方價格" @@ -594,6 +595,7 @@ "local_manual": "本地手動價格", "cloud_exact": "雲端精確供應商價格", "cloud_model_fallback": "回退模型的供應商價格", + "cloud_official": "雲端官方價格", "priority_fallback": "優先級回退價格", "single_provider_top_level": "頂層單供應商價格", "official_fallback": "官方價格" diff --git a/src/app/v1/_lib/proxy/discovery-validity.ts b/src/app/v1/_lib/proxy/discovery-validity.ts index 119cd6b3a..c85b594fd 100644 --- a/src/app/v1/_lib/proxy/discovery-validity.ts +++ b/src/app/v1/_lib/proxy/discovery-validity.ts @@ -1,3 +1,9 @@ +import { + classifyFrame, + type FrameVerdict, + type ProtocolFamily, +} from "./stream-gate/frame-classifier"; + export type DiscoveryProtocol = | "anthropic" | "openai-chat" @@ -16,76 +22,12 @@ export const DISCOVERY_PREFIX_MAX_BYTES = 1024 * 1024; export const DISCOVERY_EVENT_MAX_COUNT = 1024; const DISCOVERY_TEXT_ENCODER = new TextEncoder(); -function hasContent(value: unknown): boolean { - if (typeof value === "string") return value.trim().length > 0; - if (!value || typeof value !== "object") return false; - if (Array.isArray(value)) return value.some(hasContent); - const object = value as Record; - return [ - "text", - "content", - "delta", - "output_text", - "thinking", - "tool_use", - "tool_calls", - "functionCall", - "function_call", - "function", - "arguments", - "partial_json", - "id", - "name", - "input", - "parts", - ].some((key) => hasContent(object[key])); -} - -function hasAnthropicContentBlock(value: unknown): boolean { - if (!value || typeof value !== "object" || Array.isArray(value)) return false; - const block = value as Record; - if (typeof block.type !== "string" || block.type.length === 0) return false; - // Text blocks need non-empty text; tool_use/thinking/image blocks are - // deliverable as soon as their typed block starts, even with empty input. - return block.type === "text" ? hasContent(block.text) : true; -} - -function hasOpenAIResponsesOutputItem(value: unknown): boolean { - if (!value || typeof value !== "object" || Array.isArray(value)) return false; - const item = value as Record; - if (typeof item.type !== "string") return false; - - switch (item.type) { - case "message": - return hasContent(item.content); - case "reasoning": - return hasContent(item.summary) || hasContent(item.content); - case "function_call": - case "mcp_call": - return hasContent(item.name) || hasContent(item.arguments); - case "custom_tool_call": - return hasContent(item.name) || hasContent(item.input); - case "computer_call": - case "web_search_call": - case "file_search_call": - case "code_interpreter_call": - case "local_shell_call": - case "shell_call": - case "apply_patch_call": - return [ - item.action, - item.arguments, - item.input, - item.queries, - item.query, - item.code, - item.command, - item.operation, - ].some(hasContent); - default: - return false; - } -} +const DISCOVERY_PROTOCOL_FAMILIES: Partial> = { + anthropic: "anthropic", + "openai-chat": "openai-chat", + "openai-responses": "openai-responses", + gemini: "gemini", +}; /** * Protocol-level error signals that must remain terminal even if a provider @@ -114,71 +56,53 @@ export function isDiscoveryProtocolErrorPayload(value: unknown): boolean { ); } -function classifyJson(value: unknown, protocol: DiscoveryProtocol): DiscoveryValidity { - if (!value || typeof value !== "object") return { ready: false, terminal: false, error: true }; - const object = value as Record; - if (isDiscoveryProtocolErrorPayload(value)) { - return { ready: false, terminal: true, error: true }; - } - if (protocol === "openai-chat") { - const choices = Array.isArray(object.choices) ? object.choices : []; - const ready = choices.some((choice) => { - if (!choice || typeof choice !== "object") return false; - const choiceObject = choice as Record; - const delta = choiceObject.delta; - return hasContent(delta) || hasContent(choiceObject.message); - }); - return { ready, terminal: false, error: false }; - } - if (protocol === "openai-responses") { - if (object.type === "response.completed" || object.type === "response.done") { - return { ready: false, terminal: true, error: false }; - } - return { - ready: - (object.type === "response.output_text.delta" && hasContent(object.delta)) || - (object.type === "response.function_call_arguments.delta" && hasContent(object.delta)) || - (object.type === "response.reasoning_summary_text.delta" && hasContent(object.delta)) || - (object.type === "response.output_item.added" && hasOpenAIResponsesOutputItem(object.item)), - terminal: false, - error: false, - }; - } - if (protocol === "gemini") { - const response = - object.response && typeof object.response === "object" && !Array.isArray(object.response) - ? (object.response as Record) - : null; - const candidatesValue = response?.candidates ?? object.candidates; - const candidates = Array.isArray(candidatesValue) ? candidatesValue : []; - return { - ready: candidates.some((candidate) => hasContent(candidate)), - terminal: false, - error: false, - }; - } - // Anthropic SSE data events: message_start/message_delta are metadata; a - // content_block_delta or tool use is the first deliverable event. - if ( - object.type === "message_start" || - object.type === "message_delta" || - object.type === "ping" - ) { - return { ready: false, terminal: false, error: false }; - } - if (object.type === "message_stop") { - return { ready: false, terminal: true, error: false }; - } +function validityFromVerdict(verdict: FrameVerdict): DiscoveryValidity { return { - ready: - (object.type === "content_block_delta" && hasContent(object.delta)) || - (object.type === "content_block_start" && hasAnthropicContentBlock(object.content_block)) || - hasContent(object.content), - terminal: false, - error: false, + ready: verdict === "content", + terminal: verdict === "terminal" || verdict === "error" || verdict === "malformed", + error: verdict === "error" || verdict === "malformed", }; } +function classifyProtocolFrame( + data: string, + protocol: DiscoveryProtocol, + eventName: string | null +): DiscoveryValidity { + const family = DISCOVERY_PROTOCOL_FAMILIES[protocol]; + if (!family) return { ready: false, terminal: false, error: false }; + + let verdict = classifyFrame(family, eventName, data); + if (verdict !== "neutral") return validityFromVerdict(verdict); + + try { + const value = JSON.parse(data) as unknown; + if (isDiscoveryProtocolErrorPayload(value)) { + return { ready: false, terminal: true, error: true }; + } + + // Gemini SDK wrappers may expose the native candidate chunk under response. + if ( + family === "gemini" && + value && + typeof value === "object" && + !Array.isArray(value) && + (value as Record).response && + typeof (value as Record).response === "object" + ) { + verdict = classifyFrame( + family, + eventName, + JSON.stringify((value as Record).response) + ); + } + } catch { + return validityFromVerdict("malformed"); + } + + return validityFromVerdict(verdict); +} + export function classifyDiscoveryChunk( chunk: Uint8Array | string, protocol: DiscoveryProtocol @@ -189,6 +113,7 @@ export function classifyDiscoveryChunk( export class DiscoveryValidityParser { private buffered = ""; private dataLines: string[] = []; + private eventName: string | null = null; private readonly decoder = new TextDecoder(); private _ready = false; private _terminal = false; @@ -211,6 +136,7 @@ export class DiscoveryValidityParser { this._limitExceeded = true; this.buffered = ""; this.dataLines = []; + this.eventName = null; return this.result; } } @@ -229,6 +155,7 @@ export class DiscoveryValidityParser { if (this._error) { this.buffered = ""; this.dataLines = []; + this.eventName = null; return this.result; } } @@ -263,6 +190,10 @@ export class DiscoveryValidityParser { const colonIndex = line.indexOf(":"); const field = colonIndex === -1 ? line : line.slice(0, colonIndex); + if (field === "event") { + this.eventName = (colonIndex === -1 ? "" : line.slice(colonIndex + 1)).trim(); + return; + } if (field === "data") { let value = colonIndex === -1 ? "" : line.slice(colonIndex + 1); if (value.startsWith(" ")) value = value.slice(1); @@ -273,7 +204,7 @@ export class DiscoveryValidityParser { // event/id/retry and unknown SSE fields carry framing metadata only. A // bare JSON line is supported for providers returning non-SSE JSON, but // never while an SSE data event is pending. - if (field === "event" || field === "id" || field === "retry" || this.dataLines.length > 0) { + if (field === "id" || field === "retry" || this.dataLines.length > 0) { return; } const candidate = line.trim(); @@ -287,24 +218,18 @@ export class DiscoveryValidityParser { } private flushSseEvent(): void { + const eventName = this.eventName; + this.eventName = null; if (this.dataLines.length === 0) return; const candidate = this.dataLines.join("\n"); this.dataLines = []; if (!this.beginEvent()) return; - if (candidate.trim() === "[DONE]") { - this._terminal = true; - return; - } - try { - this.consumeValue(JSON.parse(candidate) as unknown); - } catch { - // A complete but non-JSON SSE event cannot establish protocol validity. - } + this.consumeFrame(candidate, eventName); } private consumeEventValue(value: unknown): void { if (!this.beginEvent()) return; - this.consumeValue(value); + this.consumeFrame(JSON.stringify(value), null); } private beginEvent(): boolean { @@ -317,8 +242,8 @@ export class DiscoveryValidityParser { return true; } - private consumeValue(value: unknown): void { - const result = classifyJson(value, this.protocol); + private consumeFrame(data: string, eventName: string | null): void { + const result = classifyProtocolFrame(data, this.protocol, eventName); this._ready ||= result.ready; this._terminal ||= result.terminal; this._error ||= result.error; diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 8e78cfdea..1dfc34fcd 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -6449,7 +6449,9 @@ export class ProxyForwarder { kind: effectiveKind, finalRescue: effectiveFinalRescue, controller, - parser: new DiscoveryValidityParser(protocol), + parser: new DiscoveryValidityParser( + mapProviderTypeToFamily(provider.providerType) ?? protocol + ), chunks: [], pending: true, ready: false, diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index d143eaf82..0144d9f28 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -79,7 +79,6 @@ import { createShadowGateObserver, resolveStreamGateMode } from "./stream-gate/s import { createStreamProtocolObserver, type StreamProtocolObservation, - type StreamProtocolObserver, } from "./stream-gate/stream-protocol-observer"; const CLIENT_ABORT_DRAIN_MAX_MS = 60_000; @@ -1289,6 +1288,11 @@ function hasPositiveBillableTokens(usage: UsageMetrics | null): boolean { return tokens > 0; } +function hasPositiveOutputTokens(usage: UsageMetrics | null): boolean { + if (!usage) return false; + return (usage.output_tokens ?? 0) + (usage.output_image_tokens ?? 0) > 0; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -1392,6 +1396,66 @@ export function hasStreamCompletionMarker( return inspectStreamCompletion(text, format).hasMarker; } +function hasTerminalStreamUsageEvidence( + text: string, + format: ProxySession["originalFormat"] +): boolean { + const events = parseSSEData(text); + const hasPositiveUsage = (value: unknown): boolean => + hasPositiveOutputTokens(extractUsageMetrics(value)); + + switch (format) { + case "response": + return events.some((event) => { + if (!isRecord(event.data)) return false; + const type = event.data.type; + if (type !== "response.completed" && type !== "response.done") return false; + const response = isRecord(event.data.response) ? event.data.response : null; + return hasPositiveUsage(event.data.usage) || hasPositiveUsage(response?.usage); + }); + case "claude": { + const hasTerminalUsage = events.some((event) => { + if (!isRecord(event.data)) return false; + const type = event.data.type; + if (event.event !== "message_delta" && type !== "message_delta") return false; + const delta = isRecord(event.data.delta) ? event.data.delta : null; + return hasPositiveUsage(event.data.usage) || hasPositiveUsage(delta?.usage); + }); + return hasTerminalUsage && inspectStreamCompletion(text, format).hasMarker; + } + case "openai": { + const hasTerminalUsage = events.some((event) => { + if (!isRecord(event.data) || !hasPositiveUsage(event.data.usage)) return false; + const choices = event.data.choices; + return ( + (Array.isArray(choices) && choices.length === 0) || + hasOpenAIChatCompletionMarker(event.data) + ); + }); + return hasTerminalUsage && inspectStreamCompletion(text, format).hasMarker; + } + case "gemini": + case "gemini-cli": { + const payloads: unknown[] = events.map((event) => event.data); + for (const candidate of extractJsonChunks(text)) { + try { + payloads.push(JSON.parse(candidate) as unknown); + } catch { + // 不完整 JSON 不能建立终态 usage 证据。 + } + } + return payloads.some((value) => { + if (!isRecord(value)) return false; + const payload = isRecord(value.response) ? value.response : value; + return ( + hasGeminiCompletionMarker(payload) && + (hasPositiveUsage(payload.usageMetadata) || hasPositiveUsage(payload.usage)) + ); + }); + } + } +} + export async function resolveBillableUsageMetricsForCost( session: ProxySession, provider: Provider | null, @@ -1519,6 +1583,8 @@ type FinalizeDeferredStreamingResult = { allowAuxiliarySessionBinding: boolean; /** Discovery auxiliary bindings must wait for, and depend on, the primary generation CAS. */ confirmAuxiliarySessionBinding: () => Promise; + /** 协议层已确认不可作为 Replay 来源，但不一定改变本次请求的成功/计费终态。 */ + replayIneligibleReason?: "protocol_malformed"; }; /** @@ -1801,8 +1867,22 @@ function finalizeDeferredStreamingFinalizationIfNeeded( ? detectUpstreamErrorFromSseOrJsonText(allContent) : ({ isError: false } as const); const protocolFailure = protocolObservation?.failure ?? null; + const replayIneligibleReason = + protocolFailure?.verdict === "malformed" ? ("protocol_malformed" as const) : undefined; + const postcommitMalformed = + streamEndedNormally && + upstreamStatusCode >= 200 && + upstreamStatusCode < 300 && + protocolFailure?.verdict === "malformed" && + protocolFailure.afterContent; + const successfulPostcommitMalformed = + postcommitMalformed && hasTerminalStreamUsageEvidence(allContent, session.originalFormat); const successfulHttpProtocolFailure = - streamEndedNormally && upstreamStatusCode >= 200 && upstreamStatusCode < 300 && protocolFailure; + streamEndedNormally && + upstreamStatusCode >= 200 && + upstreamStatusCode < 300 && + protocolFailure && + !successfulPostcommitMalformed; const detected = !bodyDetected.isError && ((shouldDetectFake200 && completionInspection.hasProtocolError) || @@ -1928,6 +2008,7 @@ function finalizeDeferredStreamingFinalizationIfNeeded( finalizeAttemptResources: finalizeProviderSessionRef, allowAuxiliarySessionBinding, confirmAuxiliarySessionBinding, + replayIneligibleReason, }; } @@ -2005,6 +2086,7 @@ function finalizeDeferredStreamingFinalizationIfNeeded( finalizeAttemptResources: finalizeProviderSessionRef, allowAuxiliarySessionBinding, confirmAuxiliarySessionBinding, + replayIneligibleReason, }; } @@ -2076,6 +2158,7 @@ function finalizeDeferredStreamingFinalizationIfNeeded( finalizeAttemptResources: finalizeProviderSessionRef, allowAuxiliarySessionBinding, confirmAuxiliarySessionBinding, + replayIneligibleReason, }; } @@ -2140,6 +2223,7 @@ function finalizeDeferredStreamingFinalizationIfNeeded( finalizeAttemptResources: finalizeProviderSessionRef, allowAuxiliarySessionBinding, confirmAuxiliarySessionBinding, + replayIneligibleReason, }; } @@ -2316,6 +2400,7 @@ function finalizeDeferredStreamingFinalizationIfNeeded( finalizeAttemptResources: finalizeProviderSessionRef, allowAuxiliarySessionBinding, confirmAuxiliarySessionBinding, + replayIneligibleReason, }; } @@ -3451,6 +3536,14 @@ export class ProxyResponseHandler { startHedgeBindingHeartbeat(session); let processedStream: ReadableStream = response.body; + const nativeStreamProtocolFamily = + session.getEndpointPolicy().kind === "raw_passthrough" + ? null + : mapProviderTypeToFamily(provider.providerType); + const streamProtocolObserver = nativeStreamProtocolFamily + ? createStreamProtocolObserver(nativeStreamProtocolFamily) + : null; + let protocolObservedBeforeProcessing = false; // --- GEMINI STREAM HANDLING --- if (provider.providerType === "gemini" || provider.providerType === "gemini-cli") { @@ -3490,6 +3583,8 @@ export class ProxyResponseHandler { // 若在 headers 阶段就 clearResponseTimeout，会导致首字节超时失效，客户端与服务端都会表现为一直“请求中”。 // 透传场景下，我们在后台 stats 读取到第一块数据时再清除超时（与非透传路径口径一致）。 + const streamTextAccumulator = new BoundedStreamTextAccumulator(); + let lastStreamTextSnapshot: BoundedStreamTextSnapshot | null = null; let observePassthroughChunk = (_value: Uint8Array) => {}; let observePassthroughReadStart = () => {}; let observePassthroughDrainStart = () => {}; @@ -3519,6 +3614,8 @@ export class ProxyResponseHandler { onReadStart: () => observePassthroughReadStart(), onChunk: (value) => { passthroughShadowObserver?.observe(value); + streamProtocolObserver?.observe(value); + streamTextAccumulator.pushBytes(value); observePassthroughChunk(value); }, onClientCancel: (reason) => { @@ -3550,8 +3647,6 @@ export class ProxyResponseHandler { responseController?: AbortController; }; - const streamTextAccumulator = new BoundedStreamTextAccumulator(); - let lastStreamTextSnapshot: BoundedStreamTextSnapshot | null = null; const getCollectedChunkCount = () => lastStreamTextSnapshot?.chunkCount ?? streamTextAccumulator.chunkCount; let isFirstChunk = true; @@ -3675,7 +3770,6 @@ export class ProxyResponseHandler { session.recordTfft(); clearResponseTimeoutOnce(value.byteLength); } - streamTextAccumulator.pushBytes(value); AsyncTaskManager.touch(taskId); }; @@ -3779,7 +3873,7 @@ export class ProxyResponseHandler { streamEndedNormally, clientAborted, discoveryLeaseLifecycle, - null, + streamProtocolObserver?.finish() ?? null, abortReason ); latestCommitSideEffects = finalized.commitSideEffects; @@ -3960,8 +4054,10 @@ export class ProxyResponseHandler { }); let buffer = ""; + protocolObservedBeforeProcessing = true; const transformStream = new TransformStream({ transform(chunk, controller) { + streamProtocolObserver?.observe(chunk); const decoder = new TextDecoder(); const text = decoder.decode(chunk, { stream: true }); buffer += text; @@ -4246,7 +4342,6 @@ export class ProxyResponseHandler { }; let streamFinalizationPromise: Promise | null = null; - let replayProtocolObserver: StreamProtocolObserver | null = null; const finalizeStream = ( allContent: string, streamEndedNormally: boolean, @@ -4265,7 +4360,7 @@ export class ProxyResponseHandler { streamEndedNormally, clientAborted, discoveryLeaseLifecycle, - replayProtocolObserver?.finish() ?? null, + streamProtocolObserver?.finish() ?? null, abortReason ); latestStreamCommitSideEffects = finalized.commitSideEffects @@ -4578,6 +4673,7 @@ export class ProxyResponseHandler { finalized.commitSideEffects !== undefined && effectiveStatusCode >= 200 && effectiveStatusCode < 300 && + !finalized.replayIneligibleReason && hasStreamCompletionMarker(allContent, session.originalFormat); if (isReplayableSuccess) { postTerminalSideEffects.push(async () => { @@ -4588,7 +4684,11 @@ export class ProxyResponseHandler { } }); } else { - void replaySpool.abort(streamErrorMessage ?? `status_${effectiveStatusCode}`); + void replaySpool.abort( + finalized.replayIneligibleReason ?? + streamErrorMessage ?? + `status_${effectiveStatusCode}` + ); } } @@ -4696,13 +4796,6 @@ export class ProxyResponseHandler { // write-behind 喂入 Redis 热层，供并发/断线的相同请求 attach 跟尾。 const replaySpool = createReplaySpoolIfOwner(session, response); if (replaySpool) { - const replayProtocolFamily = - session.getEndpointPolicy().kind === "raw_passthrough" - ? null - : mapProviderTypeToFamily(provider.providerType); - replayProtocolObserver = replayProtocolFamily - ? createStreamProtocolObserver(replayProtocolFamily) - : null; try { clientAbortDrainTimeoutMs = getEnvConfig().REPLAY_MAX_DETACHED_MS; } catch { @@ -4716,8 +4809,17 @@ export class ProxyResponseHandler { streamTextAccumulator.pushBytes(value); AsyncTaskManager.touch(taskId); shadowGateObserver?.observe(value); - replayProtocolObserver?.observe(value); - replaySpool?.observe(value); + const protocolFailure = protocolObservedBeforeProcessing + ? null + : (streamProtocolObserver?.observe(value) ?? null); + if (protocolFailure && replaySpool && !replaySpool.isTerminal) { + void replaySpool.abort( + `stream_protocol_${protocolFailure.verdict}_${ + protocolFailure.afterContent ? "after" : "before" + }_content` + ); + } + if (!replaySpool?.isTerminal) replaySpool?.observe(value); logger.trace("ResponseHandler: Upstream stream chunk received", { taskId, @@ -5490,6 +5592,41 @@ export function parseUsageFromResponseText( // Fallback to SSE parsing when body is not valid JSON } + // Native Gemini passthrough may be NDJSON: the whole body is not one JSON value, + // while each line is an independent response chunk. Usage is cumulative, so the + // last valid usageMetadata/usage object is authoritative. + if (!usageMetrics && (providerType === "gemini" || providerType === "gemini-cli")) { + let lastGeminiUsageRecord: Record | null = null; + let lastGeminiUsageMetrics: UsageMetrics | null = null; + for (const candidate of extractJsonChunks(responseText)) { + try { + const parsed = JSON.parse(candidate) as unknown; + if (!isRecord(parsed)) continue; + const payload = isRecord(parsed.response) ? parsed.response : parsed; + const usageValue = isRecord(payload.usageMetadata) + ? payload.usageMetadata + : isRecord(payload.usage) + ? payload.usage + : null; + if (!usageValue) continue; + const extracted = extractUsageMetrics(usageValue); + if (!extracted) continue; + lastGeminiUsageRecord = usageValue; + lastGeminiUsageMetrics = extracted; + } catch { + // malformed line 不影响后续独立 NDJSON chunk 的 usage 提取。 + } + } + if (lastGeminiUsageMetrics) { + usageRecord = lastGeminiUsageRecord; + usageMetrics = adjustUsageForProviderType( + lastGeminiUsageMetrics, + providerType, + lastGeminiUsageRecord + ); + } + } + // SSE 解析：支持两种格式 // 1. 标准 SSE (event: + data:) - Claude/OpenAI // 2. 纯 data: 格式 - Gemini diff --git a/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts b/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts index 2bbfb65b6..3d8891c3e 100644 --- a/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts +++ b/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts @@ -57,15 +57,21 @@ const STREAM_SIGNALS: Record = { ], }, { - // start 帧即携带实体 payload 的内容块；text/thinking 空启动块等 delta + // start 帧即携带完整实体 payload 的内容块。tool_use 系列只提供 id/name， + // Anthropic SDK 需要后续 input_json_delta 才能得到可执行 input，不能在这里提交。 eventTypes: ["content_block_start"], + anyPaths: [ + "content_block.data", + "content_block.content", + "content_block.file_id", + "content_block.fileId", + "content_block.url", + "content_block.result", + ], valueMatches: [ { path: "content_block.type", values: [ - "tool_use", - "server_tool_use", - "mcp_tool_use", "redacted_thinking", "web_search_tool_result", "web_fetch_tool_result", @@ -82,7 +88,7 @@ const STREAM_SIGNALS: Record = { ], errorRules: [ // SSE event: error + data {"type":"error","error":{...}} - { eventTypes: ["error"] }, + { eventTypes: ["error", "response.error"] }, // 任意帧携带非空顶层 error 对象（非规范上游 fake-200 兜底） { anyPaths: ["error"] }, ], @@ -94,8 +100,8 @@ const STREAM_SIGNALS: Record = { // chunk 无事件名；delta 携带 content/tool_calls/refusal/audio 即内容 anyPaths: [ "choices.#.delta.content", - "choices.#.delta.tool_calls", - "choices.#.delta.function_call", + "choices.#.delta.tool_calls.#.function.arguments", + "choices.#.delta.function_call.arguments", "choices.#.delta.refusal", "choices.#.delta.audio.data", "choices.#.delta.audio.transcript", @@ -161,21 +167,34 @@ const STREAM_SIGNALS: Record = { anyPaths: ["code"], }, { - // function_call / mcp_call output item 携带工具名 = 模型已决定调用工具 - eventTypes: ["response.output_item.added"], - anyPaths: ["item.name"], + // output_item.added 的 name/id/status 只是结构元数据；真实 payload 到达前不能提交， + // 否则紧随其后的 response.failed / 断流将失去透明 fallback 机会。 + // fake-streaming 的完整 item 会在 added/done 中携带 arguments/input/action，仍可提交。 + eventTypes: ["response.output_item.added", "response.output_item.done"], + anyPaths: [ + "item.content.#.text", + "item.summary.#.text", + "item.arguments", + "item.input", + "item.action", + "item.queries", + "item.query", + "item.code", + "item.command", + "item.operation", + ], }, ], errorRules: [ // 顶层 error 事件（code/message/param） - { eventTypes: ["error"] }, + { eventTypes: ["error", "response.error"] }, // 整个 response 失败（response.error 已填充）； // 子工具失败（mcp_call.failed 等）模型可继续，为中性 { eventTypes: ["response.failed"] }, // 任意帧携带非空 error 对象（response.* 事件的 error:null 不命中） { anyPaths: ["error", "response.error"] }, ], - terminalEvents: ["response.completed", "response.incomplete"], + terminalEvents: ["response.completed", "response.incomplete", "response.done"], }, gemini: { contentRules: [ diff --git a/src/app/v1/_lib/proxy/stream-gate/sse-frames.ts b/src/app/v1/_lib/proxy/stream-gate/sse-frames.ts index 904fe27ef..1365e8445 100644 --- a/src/app/v1/_lib/proxy/stream-gate/sse-frames.ts +++ b/src/app/v1/_lib/proxy/stream-gate/sse-frames.ts @@ -21,8 +21,17 @@ export interface SseFrame { export interface SseFrameParserOptions { maxBufferedCharacters?: number; + bufferLimitExemption?: { + /** 豁免帧仍受独立硬上限约束，避免特殊协议帧造成无界 retained buffer。 */ + maxBufferedCharacters: number; + /** 只接收固定长度 data 头部，避免为了判定豁免复制完整大型帧。 */ + matches: (eventName: string | null, dataHead: string) => boolean; + }; } +const DATA_HEAD_MAX_CHARACTERS = 64; +const LINE_HEAD_MAX_CHARACTERS = DATA_HEAD_MAX_CHARACTERS + "data: ".length; + export class SseFrameBufferLimitError extends Error { constructor(maxBufferedCharacters: number) { super(`SSE parser buffered data exceeded ${maxBufferedCharacters} characters`); @@ -32,10 +41,14 @@ export class SseFrameBufferLimitError extends Error { export class SseFrameParser { private readonly decoder = new TextDecoder("utf-8"); - private lineTail = ""; + private lineParts: string[] = []; + private lineCharacters = 0; + private lineHead = ""; + private skipLeadingLf = false; private currentEvent: string | null = null; private dataLines: string[] = []; private dataCharacters = 0; + private dataHead = ""; constructor(private readonly options: SseFrameParserOptions = {}) {} @@ -51,12 +64,11 @@ export class SseFrameParser { /** 流终止：冲刷尾部未换行的行与未 dispatch 的帧。 */ finish(): SseFrame[] { - const frames: SseFrame[] = []; - const tail = this.lineTail + this.decoder.decode(); - this.lineTail = ""; - if (tail.length > 0) { + const frames = this.consume(this.decoder.decode()); + this.skipLeadingLf = false; + if (this.lineCharacters > 0) { // 尾部残行按一行处理（与既有校验器对无终止空行的流的行为一致） - const frame = this.handleLine(stripTrailingCr(tail)); + const frame = this.handleLine(this.takeLine()); if (frame) frames.push(frame); } const last = this.flush(); @@ -66,24 +78,53 @@ export class SseFrameParser { private consume(text: string): SseFrame[] { const frames: SseFrame[] = []; - let buffer = this.lineTail + text; - // CR 落在末尾时可能是被切开的 CRLF，留到下一个 chunk 再判 - let holdCr = false; - if (buffer.endsWith("\r")) { - buffer = buffer.slice(0, -1); - holdCr = true; + let start = 0; + if (this.skipLeadingLf) { + if (text.length === 0) return frames; + if (text.charCodeAt(0) === 10) start = 1; + this.skipLeadingLf = false; } - const lines = buffer.split(/\r\n|\n|\r/); - // 最后一段是未完成行，保留 - this.lineTail = (lines.pop() ?? "") + (holdCr ? "\r" : ""); - this.assertBufferLimit(); - for (const line of lines) { - const frame = this.handleLine(line); + + for (let index = start; index < text.length; index += 1) { + const code = text.charCodeAt(index); + if (code !== 10 && code !== 13) continue; + + this.appendLinePart(text.slice(start, index)); + const frame = this.handleLine(this.takeLine()); if (frame) frames.push(frame); + + if (code === 13) { + if (index + 1 < text.length && text.charCodeAt(index + 1) === 10) { + index += 1; + } else if (index === text.length - 1) { + this.skipLeadingLf = true; + } + } + start = index + 1; } + + this.appendLinePart(text.slice(start)); + this.assertBufferLimit(); return frames; } + private appendLinePart(part: string): void { + if (part.length === 0) return; + this.lineParts.push(part); + this.lineCharacters += part.length; + if (this.lineHead.length < LINE_HEAD_MAX_CHARACTERS) { + this.lineHead += part.slice(0, LINE_HEAD_MAX_CHARACTERS - this.lineHead.length); + } + } + + private takeLine(): string { + const line = this.lineParts.length === 1 ? this.lineParts[0] : this.lineParts.join(""); + this.lineParts = []; + this.lineCharacters = 0; + this.lineHead = ""; + return line ?? ""; + } + private handleLine(line: string): SseFrame | null { if (line.length === 0) { return this.flush(); @@ -101,9 +142,18 @@ export class SseFrameParser { if (this.dataLines.length > 0) this.dataCharacters += 1; this.dataCharacters += data.length; this.dataLines.push(data); + this.appendDataHead(data); this.assertBufferLimit(); return null; } + const candidate = line.trim(); + if ( + this.currentEvent === null && + this.dataLines.length === 0 && + (candidate.startsWith("{") || candidate.startsWith("[")) + ) { + return { eventName: null, data: candidate }; + } // id: / retry: / 未知字段：忽略 return null; } @@ -112,30 +162,63 @@ export class SseFrameParser { const event = this.currentEvent; this.currentEvent = null; if (this.dataLines.length === 0) { + this.dataHead = ""; return null; } const data = this.dataLines.join("\n"); this.dataLines = []; this.dataCharacters = 0; + this.dataHead = ""; return { eventName: event, data }; } + private appendDataHead(data: string): void { + if (this.dataHead.length >= DATA_HEAD_MAX_CHARACTERS) return; + if (this.dataLines.length > 1) this.dataHead += "\n"; + this.dataHead += data.slice(0, DATA_HEAD_MAX_CHARACTERS - this.dataHead.length); + } + + private currentDataHead(): string { + if (this.dataHead.length >= DATA_HEAD_MAX_CHARACTERS) return this.dataHead; + if (!this.lineHead.startsWith("data:")) return this.dataHead; + + const tailData = this.lineHead.slice(5).replace(/^\s/, ""); + const separator = this.dataLines.length > 0 && this.dataHead.length > 0 ? "\n" : ""; + return `${this.dataHead}${separator}${tailData}`.slice(0, DATA_HEAD_MAX_CHARACTERS); + } + + private resetRetainedState(): void { + this.lineParts = []; + this.lineCharacters = 0; + this.lineHead = ""; + this.currentEvent = null; + this.dataLines = []; + this.dataCharacters = 0; + this.dataHead = ""; + this.skipLeadingLf = false; + } + private assertBufferLimit(): void { const maxBufferedCharacters = this.options.maxBufferedCharacters; if (maxBufferedCharacters === undefined) return; const bufferedCharacters = - this.lineTail.length + (this.currentEvent?.length ?? 0) + this.dataCharacters; - if (bufferedCharacters > maxBufferedCharacters) { - throw new SseFrameBufferLimitError(maxBufferedCharacters); + this.lineCharacters + (this.currentEvent?.length ?? 0) + this.dataCharacters; + if (bufferedCharacters <= maxBufferedCharacters) return; + + const exemption = this.options.bufferLimitExemption; + if ( + exemption && + bufferedCharacters <= exemption.maxBufferedCharacters && + exemption.matches(this.currentEvent, this.currentDataHead()) + ) { + return; } + this.resetRetainedState(); + throw new SseFrameBufferLimitError(maxBufferedCharacters); } } -function stripTrailingCr(line: string): string { - return line.endsWith("\r") ? line.slice(0, -1) : line; -} - /** 对完整 SSE body 一次性解析出全部帧。 */ export function parseSseBody(body: string): SseFrame[] { const parser = new SseFrameParser(); diff --git a/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts b/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts index d78c50c46..930b676cd 100644 --- a/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts +++ b/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts @@ -1,9 +1,12 @@ -import { classifyFrame, type ProtocolFamily } from "./frame-classifier"; +import { classifyFrame, isRequestEchoFrame, type ProtocolFamily } from "./frame-classifier"; import { type SseFrame, SseFrameParser } from "./sse-frames"; +import { resolveStreamGateCaps } from "./stream-content-gate"; -export const REPLAY_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS = 1024 * 1024; +export const STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS = 10 * 1024 * 1024; +const DEFAULT_STREAM_GATE_PREBUFFER_CHARACTERS = 10 * 1024 * 1024; export interface StreamProtocolFailure { + afterContent: boolean; verdict: "error" | "malformed"; eventName: string | null; } @@ -11,29 +14,42 @@ export interface StreamProtocolFailure { export interface StreamProtocolObservation { sawContent: boolean; sawTerminal: boolean; + observationIncomplete: boolean; failure: StreamProtocolFailure | null; } export interface StreamProtocolObserver { - observe(chunk: Uint8Array): void; + observe(chunk: Uint8Array): StreamProtocolFailure | null; finish(): StreamProtocolObservation; } export function createStreamProtocolObserver(family: ProtocolFamily): StreamProtocolObserver { + const { prebufferByteCap } = resolveStreamGateCaps(); + const streamGatePrebufferCharacters = + Number.isSafeInteger(prebufferByteCap) && prebufferByteCap > 0 + ? prebufferByteCap + : DEFAULT_STREAM_GATE_PREBUFFER_CHARACTERS; const parser = new SseFrameParser({ - maxBufferedCharacters: REPLAY_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS, + bufferLimitExemption: { + // 门禁对 request echo 的豁免额度最多把总缓冲抬到 2x cap；observer 采用同一边界， + // 允许合法的大请求回显，同时继续阻止伪装 echo 的无界单帧。 + maxBufferedCharacters: streamGatePrebufferCharacters * 2, + matches: (eventName, dataHead) => isRequestEchoFrame(family, eventName, dataHead), + }, + maxBufferedCharacters: STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS, }); const observation: StreamProtocolObservation = { sawContent: false, sawTerminal: false, + observationIncomplete: false, failure: null, }; let finished = false; let disabled = false; - const disableWithMalformedFailure = (): void => { + const disableIncompleteObservation = (): void => { disabled = true; - observation.failure ??= { verdict: "malformed", eventName: null }; + observation.observationIncomplete = true; }; const record = (frame: SseFrame): void => { @@ -41,19 +57,25 @@ export function createStreamProtocolObserver(family: ProtocolFamily): StreamProt if (verdict === "content") observation.sawContent = true; if (verdict === "terminal") observation.sawTerminal = true; if ((verdict === "error" || verdict === "malformed") && !observation.failure) { - observation.failure = { verdict, eventName: frame.eventName }; + observation.failure = { + afterContent: observation.sawContent, + verdict, + eventName: frame.eventName, + }; } }; return { - observe(chunk: Uint8Array): void { - if (finished || disabled || chunk.byteLength === 0) return; + observe(chunk: Uint8Array): StreamProtocolFailure | null { + if (finished || disabled || chunk.byteLength === 0) return observation.failure; try { for (const frame of parser.push(chunk)) record(frame); } catch { - // 旁路观察器异常不得改变客户端流或计费热路径。 - disableWithMalformedFailure(); + // parser 容量保护和本地观察异常只能说明观察不完整，不能伪造上游 malformed。 + // 旁路 observer 必须 fail-open，避免本地资源或实现问题改写客户端流与计费终态。 + disableIncompleteObservation(); } + return observation.failure; }, finish(): StreamProtocolObservation { @@ -63,13 +85,14 @@ export function createStreamProtocolObserver(family: ProtocolFamily): StreamProt try { for (const frame of parser.finish()) record(frame); } catch { - disableWithMalformedFailure(); + disableIncompleteObservation(); } } } return { sawContent: observation.sawContent, sawTerminal: observation.sawTerminal, + observationIncomplete: observation.observationIncomplete, failure: observation.failure ? { ...observation.failure } : null, }; }, diff --git a/tests/integration/proxy-hedge-lifecycle.test.ts b/tests/integration/proxy-hedge-lifecycle.test.ts index 50b2d2700..566c84dfc 100644 --- a/tests/integration/proxy-hedge-lifecycle.test.ts +++ b/tests/integration/proxy-hedge-lifecycle.test.ts @@ -375,6 +375,7 @@ function createProvider(id: number, url: string, firstByteTimeoutStreamingMs: nu } type Upstream = { + readonly abort: (error?: Error) => Promise; readonly abortCount: () => number; readonly baseUrl: string; readonly close: () => Promise; @@ -418,6 +419,11 @@ async function startUpstream(): Promise { }); }); return { + abort: async (error = new Error("upstream stream aborted")) => { + const response = await responseGate.promise; + response.destroy(error); + await terminationGate.promise; + }, abortCount: () => aborts, baseUrl, close: async () => { @@ -764,6 +770,236 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p } }); + it.each([ + ...[false, true].flatMap((discoveryEnabled) => [ + { + discoveryEnabled, + failureBody: responsesFrame("response.failed", { + response: { + error: { message: "upstream connection closed before completion" }, + id: "resp_failed", + status: "failed", + }, + type: "response.failed", + }), + failureName: "response.failed", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + { + discoveryEnabled, + failureBody: responsesFrame("response.error", { + code: "stream_read_error", + message: "stream_read_error", + type: "response.error", + }), + failureName: "response.error", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + { + discoveryEnabled, + failureBody: + 'event: response.in_progress\ndata: {"type":"response.in_progress","response":\n\n', + failureName: "malformed frame", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + ]), + ])( + "keeps Responses tool metadata precommit and falls back on $failureName ($pathName)", + async ({ discoveryEnabled, failureBody }) => { + const [failed, fallback] = await Promise.all([startUpstream(), startUpstream()]); + const client = new AbortController(); + try { + state.discoveryEnabled = discoveryEnabled; + state.streamGateMode = "enforce"; + const initialProvider = createProvider(1, failed.baseUrl, 0); + initialProvider.providerType = "codex"; + const fallbackProvider = createProvider(2, fallback.baseUrl, 0); + fallbackProvider.priority = initialProvider.priority; + fallbackProvider.providerType = "codex"; + state.providers.push(fallbackProvider); + const session = await createSession(initialProvider, "/v1/responses", client.signal); + session.sessionId = `integration-responses-precommit-${discoveryEnabled}`; + + const forwarded = ProxyForwarder.send(session); + await failed.response; + if (discoveryEnabled) await fallback.response; + + await failed.write( + responsesFrame("response.output_item.added", { + item: { + arguments: "", + id: "fc_failed", + name: "lookup", + status: "in_progress", + type: "function_call", + }, + output_index: 0, + type: "response.output_item.added", + }) + ); + + const precommitState = await Promise.race([ + forwarded.then( + () => "resolved" as const, + () => "rejected" as const + ), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 25)), + ]); + expect(precommitState).toBe("pending"); + + await failed.send(failureBody); + await fallback.response; + const fallbackStream = responsesStreamFixture("resp_fallback", "msg_fallback"); + await fallback.send(`${fallbackStream.firstContent}${fallbackStream.completed}`); + + const downstream = await ProxyResponseHandler.dispatch(session, await forwarded); + const body = await downstream.text(); + await settleTasks(); + + expect(body).toContain('"delta":"我"'); + expect(body).not.toContain("fc_failed"); + expect(session.provider?.id).toBe(fallbackProvider.id); + if (discoveryEnabled) { + expect(session.getRoutingTrace()?.mode).toBe("discovery"); + } + expect(session.getProviderChain()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ id: initialProvider.id }), + expect.objectContaining({ id: fallbackProvider.id, statusCode: 200 }), + ]) + ); + } finally { + client.abort(new Error("fixture cleanup")); + await Promise.all([failed.close(), fallback.close()]); + } + } + ); + + it.each( + [false, true].flatMap((discoveryEnabled) => [ + { + discoveryEnabled, + metadataBody: + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"tu_failed","name":"lookup","input":{}}}\n\n', + metadataName: "tool start", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + { + discoveryEnabled, + metadataBody: + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}\n\n', + metadataName: "empty thinking start", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + { + discoveryEnabled, + metadataBody: + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"redacted_thinking","data":""}}\n\n', + metadataName: "empty redacted thinking start", + pathName: discoveryEnabled ? "Discovery" : "enforced gate", + }, + ]) + )( + "keeps Anthropic $metadataName precommit and falls back on transport abort ($pathName)", + async ({ discoveryEnabled, metadataBody }) => { + const [failed, fallback] = await Promise.all([startUpstream(), startUpstream()]); + const client = new AbortController(); + try { + state.discoveryEnabled = discoveryEnabled; + state.streamGateMode = "enforce"; + const initialProvider = createProvider(1, failed.baseUrl, 0); + const fallbackProvider = createProvider(2, fallback.baseUrl, 0); + fallbackProvider.priority = initialProvider.priority; + state.providers.push(fallbackProvider); + const session = await createSession(initialProvider, "/v1/messages", client.signal); + session.sessionId = `integration-anthropic-precommit-${discoveryEnabled}`; + + const forwarded = ProxyForwarder.send(session); + await failed.response; + if (discoveryEnabled) await fallback.response; + + await failed.write(metadataBody); + + const precommitState = await Promise.race([ + forwarded.then( + () => "resolved" as const, + () => "rejected" as const + ), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 25)), + ]); + expect(precommitState).toBe("pending"); + + await failed.abort(new Error("stream interrupted before completion")); + await fallback.response; + await fallback.send( + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"fallback"}}\n\n' + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n' + ); + + const downstream = await ProxyResponseHandler.dispatch(session, await forwarded); + const body = await downstream.text(); + await settleTasks(); + + expect(body).toContain('"text":"fallback"'); + expect(body).not.toContain("tu_failed"); + expect(session.provider?.id).toBe(fallbackProvider.id); + expect(session.getProviderChain()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ id: initialProvider.id }), + expect.objectContaining({ id: fallbackProvider.id, statusCode: 200 }), + ]) + ); + } finally { + client.abort(new Error("fixture cleanup")); + await Promise.all([failed.close(), fallback.close()]); + } + } + ); + + it("does not splice a fallback stream after real Responses content is committed", async () => { + const [committed, fallback] = await Promise.all([startUpstream(), startUpstream()]); + const client = new AbortController(); + try { + state.streamGateMode = "enforce"; + const initialProvider = createProvider(1, committed.baseUrl, 0); + initialProvider.providerType = "codex"; + const fallbackProvider = createProvider(2, fallback.baseUrl, 0); + fallbackProvider.providerType = "codex"; + state.providers.push(fallbackProvider); + const session = await createSession(initialProvider, "/v1/responses", client.signal); + const stream = responsesStreamFixture("resp_committed", "msg_committed"); + + const forwarded = ProxyForwarder.send(session); + await committed.response; + await committed.write(stream.firstContent); + + const downstream = await ProxyResponseHandler.dispatch(session, await forwarded); + expect(fallback.requestCount()).toBe(0); + + const malformed = + 'event: response.in_progress\ndata: {"type":"response.in_progress","response":\n\n'; + await committed.send(`${malformed}${stream.completed}`); + const body = await downstream.text(); + await settleTasks(); + + expect(body).toContain('"delta":"我"'); + expect(body).toContain(malformed); + expect(fallback.requestCount()).toBe(0); + expect(state.durableTerminal).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ + inputTokens: 1, + outputTokens: 1, + statusCode: 200, + }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + } finally { + client.abort(new Error("fixture cleanup")); + await Promise.all([committed.close(), fallback.close()]); + } + }); + it("runs a leased Discovery race over real loopback transports and cancels the loser", async () => { const [loser, winner] = await Promise.all([startUpstream(), startUpstream()]); const client = new AbortController(); diff --git a/tests/unit/i18n/pricing-source-keys.test.ts b/tests/unit/i18n/pricing-source-keys.test.ts new file mode 100644 index 000000000..82def973a --- /dev/null +++ b/tests/unit/i18n/pricing-source-keys.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; + +import enDashboard from "../../../messages/en/dashboard.json"; +import jaDashboard from "../../../messages/ja/dashboard.json"; +import ruDashboard from "../../../messages/ru/dashboard.json"; +import zhCNDashboard from "../../../messages/zh-CN/dashboard.json"; +import zhTWDashboard from "../../../messages/zh-TW/dashboard.json"; + +const PRICING_SOURCES = [ + "local_manual", + "cloud_exact", + "cloud_model_fallback", + "cloud_official", + "priority_fallback", + "single_provider_top_level", + "official_fallback", +] as const; + +const dashboards = { + en: enDashboard, + ja: jaDashboard, + ru: ruDashboard, + "zh-CN": zhCNDashboard, + "zh-TW": zhTWDashboard, +}; + +describe("dashboard pricing source translations", () => { + for (const [locale, dashboard] of Object.entries(dashboards)) { + it(`${locale} covers every pricing source in list and detail views`, () => { + const maps = [ + dashboard.logs.billingDetails.pricingSource, + dashboard.logs.details.billingDetails.pricingSource, + ]; + + for (const pricingSource of maps) { + expect(Object.keys(pricingSource).sort(), `${locale} pricing source keys`).toEqual( + [...PRICING_SOURCES].sort() + ); + for (const source of PRICING_SOURCES) { + expect(pricingSource[source].trim(), `${locale} ${source}`).not.toBe(""); + } + } + }); + } +}); diff --git a/tests/unit/proxy/discovery-validity.test.ts b/tests/unit/proxy/discovery-validity.test.ts index 00f9532a1..f769d99f9 100644 --- a/tests/unit/proxy/discovery-validity.test.ts +++ b/tests/unit/proxy/discovery-validity.test.ts @@ -95,7 +95,7 @@ describe("discovery validity", () => { ).toBe(false); expect( classifyDiscoveryChunk( - 'data: {"type":"response.output_text.delta","delta":" "}\n\n', + 'data: {"type":"response.output_text.delta","delta":""}\n\n', "openai-responses" ).ready ).toBe(false); @@ -154,12 +154,24 @@ describe("discovery validity", () => { "openai-responses" ).ready ).toBe(false); + expect( + classifyDiscoveryChunk( + 'data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","name":"lookup","arguments":""}}\n\n', + "openai-responses" + ).ready + ).toBe(false); expect( classifyDiscoveryChunk( 'data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","name":"lookup","arguments":"{}"}}\n\n', "openai-responses" ).ready ).toBe(true); + expect( + classifyDiscoveryChunk( + 'data: {"type":"response.output_item.done","item":{"id":"fc_1","type":"function_call","name":"lookup","arguments":"{}"}}\n\n', + "openai-responses" + ).ready + ).toBe(true); }); it("consumes split SSE lines incrementally without waiting for the full stream", () => { @@ -256,13 +268,13 @@ describe("discovery validity", () => { ).toMatchObject({ ready: true, error: false }); }); - it("accepts Anthropic tool-use starts and partial JSON deltas", () => { + it("holds Anthropic tool-use starts until partial JSON deltas arrive", () => { expect( classifyDiscoveryChunk( 'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"tu_1","name":"search","input":{}}}\n\n', "anthropic" ).ready - ).toBe(true); + ).toBe(false); expect( classifyDiscoveryChunk( 'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"q\\":1}"}}\n\n', @@ -271,6 +283,33 @@ describe("discovery validity", () => { ).toBe(true); }); + it("shares the enforced gate content boundary for structured protocol frames", () => { + expect( + classifyDiscoveryChunk( + 'event: content_block_start\ndata: {"type":"content_block_start","content_block":{"type":"thinking","thinking":""}}\n\n', + "anthropic" + ).ready + ).toBe(false); + expect( + classifyDiscoveryChunk( + 'event: content_block_start\ndata: {"type":"content_block_start","content_block":{"type":"redacted_thinking","data":"opaque"}}\n\n', + "anthropic" + ).ready + ).toBe(true); + expect( + classifyDiscoveryChunk( + 'event: response.refusal.delta\ndata: {"type":"response.refusal.delta","delta":"blocked"}\n\n', + "openai-responses" + ).ready + ).toBe(true); + expect( + classifyDiscoveryChunk( + 'event: response.output_item.done\ndata: {"type":"response.output_item.done","item":{"type":"computer_call","action":{"type":"click","x":10,"y":20}}}\n\n', + "openai-responses" + ).ready + ).toBe(true); + }); + it("fails a metadata-only prefix after the byte limit", () => { const parser = new DiscoveryValidityParser("openai-chat"); const result = parser.push(`:${"x".repeat(DISCOVERY_PREFIX_MAX_BYTES + 1)}`); diff --git a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts b/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts index 169ecfdb9..7a4483048 100644 --- a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts +++ b/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts @@ -1236,9 +1236,14 @@ describe("Endpoint circuit breaker isolation", () => { ])("accepts a structurally valid $label completion marker", async ({ format, body }) => { const session = createSession(); session.originalFormat = format; - if (format === "gemini" || format === "gemini-cli") { - session.provider = { ...session.provider!, providerType: format }; - } + const providerTypeByFormat = { + claude: "claude", + gemini: "gemini", + "gemini-cli": "gemini-cli", + openai: "openai-compatible", + response: "codex", + } as const; + session.provider = { ...session.provider!, providerType: providerTypeByFormat[format] }; const snapshot = { sessionId: "fake-session", keyId: 456, diff --git a/tests/unit/proxy/response-handler-stream-terminal.test.ts b/tests/unit/proxy/response-handler-stream-terminal.test.ts index 260fa97a3..68f2f8b48 100644 --- a/tests/unit/proxy/response-handler-stream-terminal.test.ts +++ b/tests/unit/proxy/response-handler-stream-terminal.test.ts @@ -298,6 +298,439 @@ describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { expect(releaseAgent).toHaveBeenCalledOnce(); }); + it("fails closed on a late Gemini passthrough NDJSON error", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "gemini" }); + session.originalFormat = "gemini"; + const body = [ + '{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}', + '{"error":{"message":"late upstream failure"}}', + "", + ].join("\n"); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + await returned.text(); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + + it("observes native Gemini failures before transforming the client stream", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "gemini" }); + session.originalFormat = "claude"; + const body = [ + 'data: {"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}\n\n', + 'data: {"error":{"message":"late upstream failure"}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + await returned.text(); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + + it("bills a naturally completed postcommit malformed stream with terminal usage but rejects Replay", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "codex" }); + session.originalFormat = "response"; + session.replayState = { + role: "owner", + ownerToken: "owner-token-malformed-usage", + identity: { + replayId: "replay-malformed-usage", + verifier: "verifier", + scopeTag: "scope-tag", + keyId: KEY.id, + userId: USER.id, + format: "response", + model: "gpt-test", + endpoint: "/v1/responses", + }, + }; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + const malformed = "event: response.in_progress\ndata: not-json\n\n"; + const body = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + malformed, + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":10,"output_tokens":2}}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toContain(malformed); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ inputTokens: 10, outputTokens: 2, statusCode: 200 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + expect(mocks.replayAbort).toHaveBeenCalledWith(expect.stringContaining("protocol_malformed")); + expect(mocks.replayComplete).not.toHaveBeenCalled(); + }); + + it.each([ + { + label: "Anthropic", + providerType: "claude" as const, + format: "claude" as const, + body: [ + 'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":10}}}\n\n', + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"text":"ok"}}\n\n', + "event: ping\ndata: not-json\n\n", + 'event: message_delta\ndata: {"type":"message_delta","usage":{"output_tokens":2}}\n\n', + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ].join(""), + }, + { + label: "OpenAI Chat", + providerType: "openai-compatible" as const, + format: "openai" as const, + body: [ + 'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n', + "data: not-json\n\n", + 'data: {"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":2}}\n\n', + "data: [DONE]\n\n", + ].join(""), + }, + { + label: "Gemini", + providerType: "gemini" as const, + format: "gemini" as const, + body: [ + '{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}', + "{not-json}", + '{"candidates":[{"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2}}', + "", + ].join("\n"), + }, + ])( + "records a naturally completed $label postcommit malformed stream as successful with terminal usage", + async ({ providerType, format, body }) => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType }); + session.originalFormat = format; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ inputTokens: 10, outputTokens: 2, statusCode: 200 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + } + ); + + it("does not treat Anthropic message_start usage as terminal usage evidence", async () => { + const { session } = await createSession({}); + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + const body = [ + 'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":10}}}\n\n', + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"text":"ok"}}\n\n', + "event: ping\ndata: not-json\n\n", + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + + it("does not treat zero terminal usage as billable success after malformed", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "codex" }); + session.originalFormat = "response"; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + const body = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":0,"output_tokens":0}}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + + it.each([ + { + label: "OpenAI Responses", + providerType: "codex" as const, + format: "response" as const, + body: [ + 'event: response.in_progress\ndata: {"type":"response.in_progress","response":{"status":"in_progress","usage":{"input_tokens":10}}}\n\n', + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{}}}\n\n', + ].join(""), + }, + { + label: "Anthropic", + providerType: "claude" as const, + format: "claude" as const, + body: [ + 'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":10}}}\n\n', + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"text":"ok"}}\n\n', + "event: ping\ndata: not-json\n\n", + 'event: message_delta\ndata: {"type":"message_delta","usage":{}}\n\n', + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ].join(""), + }, + { + label: "OpenAI Chat", + providerType: "openai-compatible" as const, + format: "openai" as const, + body: [ + 'data: {"choices":[{"delta":{"role":"assistant"}}],"usage":{"prompt_tokens":10}}\n\n', + 'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n', + "data: not-json\n\n", + 'data: {"choices":[],"usage":{}}\n\n', + "data: [DONE]\n\n", + ].join(""), + }, + { + label: "Gemini", + providerType: "gemini" as const, + format: "gemini" as const, + body: [ + '{"usageMetadata":{"promptTokenCount":10}}', + '{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}', + "{not-json}", + '{"candidates":[{"finishReason":"STOP"}],"usageMetadata":{}}', + "", + ].join("\n"), + }, + ])( + "does not combine early $label usage with empty terminal usage after malformed", + async ({ providerType, format, body }) => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType }); + session.originalFormat = format; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + } + ); + + it.each([ + { + label: "OpenAI Responses", + providerType: "codex" as const, + format: "response" as const, + body: [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":10,"output_tokens":0}}}\n\n', + ].join(""), + }, + { + label: "Anthropic", + providerType: "claude" as const, + format: "claude" as const, + body: [ + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"text":"ok"}}\n\n', + "event: ping\ndata: not-json\n\n", + 'event: message_delta\ndata: {"type":"message_delta","usage":{"input_tokens":10,"output_tokens":0}}\n\n', + 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ].join(""), + }, + { + label: "OpenAI Chat", + providerType: "openai-compatible" as const, + format: "openai" as const, + body: [ + 'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n', + "data: not-json\n\n", + 'data: {"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":0}}\n\n', + "data: [DONE]\n\n", + ].join(""), + }, + { + label: "Gemini", + providerType: "gemini" as const, + format: "gemini" as const, + body: [ + '{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}', + "{not-json}", + '{"candidates":[{"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":0}}', + "", + ].join("\n"), + }, + ])( + "does not treat terminal input-only $label usage as successful after malformed", + async ({ providerType, format, body }) => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType }); + session.originalFormat = format; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + } + ); + + it("allows observation overflow to complete a successful Replay entry", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "codex" }); + session.originalFormat = "response"; + session.replayState = { + role: "owner", + ownerToken: "owner-token-overflow", + identity: { + replayId: "replay-overflow", + verifier: "verifier", + scopeTag: "scope-tag", + keyId: KEY.id, + userId: USER.id, + format: "response", + model: "gpt-test", + endpoint: "/v1/responses", + }, + }; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + const oversizedDelta = "x".repeat(10 * 1024 * 1024 + 1); + const body = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + `event: response.output_text.delta\ndata: ${JSON.stringify({ type: "response.output_text.delta", delta: oversizedDelta })}\n\n`, + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":10,"output_tokens":2}}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect((await returned.text()).length).toBe(body.length); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ inputTokens: 10, outputTokens: 2, statusCode: 200 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + expect(mocks.replayAbort).not.toHaveBeenCalled(); + expect(mocks.replayComplete).toHaveBeenCalledWith(MESSAGE.id); + }); + it("aborts Replay when a late protocol error is outside the bounded text snapshot", async () => { const { session } = await createSession({}); session.setProvider({ ...createProvider(), providerType: "codex" }); diff --git a/tests/unit/proxy/stream-gate-frame-classifier.test.ts b/tests/unit/proxy/stream-gate-frame-classifier.test.ts index ecc3f4d7e..aed7fcab8 100644 --- a/tests/unit/proxy/stream-gate-frame-classifier.test.ts +++ b/tests/unit/proxy/stream-gate-frame-classifier.test.ts @@ -57,14 +57,14 @@ describe("classifyFrame: anthropic", () => { ).toBe("neutral"); }); - it("content: content_block_start carrying entity payload (tool_use)", () => { + it("neutral: content_block_start carrying only tool metadata", () => { expect( classifyFrame( "anthropic", "content_block_start", '{"type":"content_block_start","content_block":{"type":"tool_use","id":"t1","name":"f"}}' ) - ).toBe("content"); + ).toBe("neutral"); }); it("neutral: content_block_start for empty text block", () => { @@ -77,6 +77,37 @@ describe("classifyFrame: anthropic", () => { ).toBe("neutral"); }); + it("requires entity payload before structured content_block_start commits", () => { + expect( + classifyFrame( + "anthropic", + "content_block_start", + '{"type":"content_block_start","content_block":{"type":"redacted_thinking","data":""}}' + ) + ).toBe("neutral"); + expect( + classifyFrame( + "anthropic", + "content_block_start", + '{"type":"content_block_start","content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1","content":[]}}' + ) + ).toBe("neutral"); + expect( + classifyFrame( + "anthropic", + "content_block_start", + '{"type":"content_block_start","content_block":{"type":"redacted_thinking","data":"opaque"}}' + ) + ).toBe("content"); + expect( + classifyFrame( + "anthropic", + "content_block_start", + '{"type":"content_block_start","content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1","content":[{"type":"web_search_result","url":"https://example.com"}]}}' + ) + ).toBe("content"); + }); + it("neutral: message_start / ping / message_delta bookkeeping", () => { expect( classifyFrame("anthropic", "message_start", '{"type":"message_start","message":{"id":"m"}}') @@ -128,7 +159,7 @@ describe("classifyFrame: anthropic", () => { }); describe("classifyFrame: openai-chat", () => { - it("content: delta content / tool_calls / refusal / audio", () => { + it("content: delta content / tool arguments / refusal / audio", () => { expect(classifyFrame("openai-chat", null, '{"choices":[{"delta":{"content":"hi"}}]}')).toBe( "content" ); @@ -136,7 +167,7 @@ describe("classifyFrame: openai-chat", () => { classifyFrame( "openai-chat", null, - '{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"f"}}]}}]}' + '{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"f","arguments":"{}"}}]}}]}' ) ).toBe("content"); expect(classifyFrame("openai-chat", null, '{"choices":[{"delta":{"refusal":"no"}}]}')).toBe( @@ -157,6 +188,13 @@ describe("classifyFrame: openai-chat", () => { expect(classifyFrame("openai-chat", null, '{"choices":[],"usage":{"total_tokens":10}}')).toBe( "neutral" ); + expect( + classifyFrame( + "openai-chat", + null, + '{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"f","arguments":""}}]}}]}' + ) + ).toBe("neutral"); }); it("neutral: empty string content delta", () => { @@ -222,13 +260,23 @@ describe("classifyFrame: openai-responses", () => { ).toBe("content"); }); - it("content: output_item.added carrying tool name", () => { + it("neutral: output_item.added carrying only tool metadata", () => { expect( classifyFrame( "openai-responses", "response.output_item.added", '{"type":"response.output_item.added","item":{"type":"function_call","name":"get_x"}}' ) + ).toBe("neutral"); + }); + + it("content: output_item tool payload with complete arguments", () => { + expect( + classifyFrame( + "openai-responses", + "response.output_item.done", + '{"type":"response.output_item.done","item":{"type":"function_call","name":"get_x","arguments":"{}"}}' + ) ).toBe("content"); }); @@ -266,6 +314,13 @@ describe("classifyFrame: openai-responses", () => { expect( classifyFrame("openai-responses", "error", '{"type":"error","code":"x","message":"m"}') ).toBe("error"); + expect( + classifyFrame( + "openai-responses", + "response.error", + '{"type":"response.error","code":"stream_read_error","message":"stream_read_error"}' + ) + ).toBe("error"); expect( classifyFrame( "openai-responses", diff --git a/tests/unit/proxy/stream-gate-protocol-observer.test.ts b/tests/unit/proxy/stream-gate-protocol-observer.test.ts index a7e8ff1cc..08431abd2 100644 --- a/tests/unit/proxy/stream-gate-protocol-observer.test.ts +++ b/tests/unit/proxy/stream-gate-protocol-observer.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "vitest"; import { createStreamProtocolObserver, - REPLAY_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS, + STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS, } from "@/app/v1/_lib/proxy/stream-gate/stream-protocol-observer"; const encoder = new TextEncoder(); @@ -15,6 +15,7 @@ function observeAtEveryBoundary(stream: string): void { expect(observer.finish()).toEqual({ sawContent: true, sawTerminal: true, + observationIncomplete: false, failure: null, }); } @@ -47,9 +48,11 @@ describe("StreamProtocolObserver", () => { expect(result.sawContent).toBe(true); expect(result.sawTerminal).toBe(true); expect(result.failure).toEqual({ + afterContent: true, verdict: "error", eventName: "response.failed", }); + expect(result.observationIncomplete).toBe(false); }); test("把 malformed data 记录为终态失败", () => { @@ -66,11 +69,12 @@ describe("StreamProtocolObserver", () => { expect(observer.finish()).toEqual({ sawContent: true, sawTerminal: false, - failure: { verdict: "malformed", eventName: "message_stop" }, + observationIncomplete: false, + failure: { afterContent: true, verdict: "malformed", eventName: "message_stop" }, }); }); - test("成功 Anthropic tool_use/message_stop 可 completed", () => { + test("Anthropic tool metadata alone does not count as committed content", () => { const observer = createStreamProtocolObserver("anthropic"); observer.observe( encoder.encode( @@ -82,8 +86,9 @@ describe("StreamProtocolObserver", () => { ); expect(observer.finish()).toEqual({ - sawContent: true, + sawContent: false, sawTerminal: true, + observationIncomplete: false, failure: null, }); }); @@ -95,6 +100,7 @@ describe("StreamProtocolObserver", () => { expect(observer.finish()).toEqual({ sawContent: true, sawTerminal: false, + observationIncomplete: false, failure: null, }); }); @@ -110,6 +116,7 @@ describe("StreamProtocolObserver", () => { expect(observer.finish()).toEqual({ sawContent: true, sawTerminal: true, + observationIncomplete: false, failure: null, }); }); @@ -121,24 +128,83 @@ describe("StreamProtocolObserver", () => { expect(observer.finish()).toEqual({ sawContent: false, sawTerminal: false, - failure: { verdict: "malformed", eventName: "response.completed" }, + observationIncomplete: false, + failure: { afterContent: false, verdict: "malformed", eventName: "response.completed" }, }); }); - test("超长未终止 SSE 行会 fail closed，而不是被视为干净流", () => { + test("observer 为单个未完成协议帧保留 10 MiB 观察空间", () => { + expect(STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS).toBe(10 * 1024 * 1024); + }); + + test("超长未终止 SSE 行只会使观察不完整，不会伪造 malformed", () => { const observer = createStreamProtocolObserver("openai-responses"); observer.observe( - encoder.encode(`data: ${"x".repeat(REPLAY_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1)}`) + encoder.encode(`data: ${"x".repeat(STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1)}`) ); expect(observer.finish()).toEqual({ sawContent: false, sawTerminal: false, - failure: { verdict: "malformed", eventName: null }, + observationIncomplete: true, + failure: null, }); }); - test("首内容后 observer 超限仍保留 content，并阻止 Replay completed", () => { + test("大型 Responses request echo 后的正常内容不会被误判为 malformed", () => { + const observer = createStreamProtocolObserver("openai-responses"); + const requestEcho = [ + "event: response.created\n", + `data: ${JSON.stringify({ + response: { + id: "resp_large_echo", + instructions: "x".repeat(STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1), + status: "in_progress", + }, + type: "response.created", + })}\n\n`, + ].join(""); + const suffix = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n', + ].join(""); + const bytes = encoder.encode(`${requestEcho}${suffix}`); + for (let offset = 0; offset < bytes.length; offset += 16 * 1024) { + observer.observe(bytes.subarray(offset, offset + 16 * 1024)); + } + + expect(observer.finish()).toEqual({ + sawContent: true, + sawTerminal: true, + observationIncomplete: false, + failure: null, + }); + }); + + test("大型非 request echo 帧达到资源上限时标记 observation incomplete", () => { + const observer = createStreamProtocolObserver("openai-responses"); + observer.observe( + encoder.encode( + `event: response.output_item.added\ndata: ${JSON.stringify({ + item: { + id: "msg_oversized", + payload: "x".repeat(STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1), + type: "message", + }, + type: "response.output_item.added", + })}\n\n` + ) + ); + + expect(observer.finish()).toEqual({ + sawContent: false, + sawTerminal: false, + observationIncomplete: true, + failure: null, + }); + }); + + test("首内容后 observer 超限保留 content，并把资源上限与协议失败分开", () => { const observer = createStreamProtocolObserver("openai-responses"); observer.observe( encoder.encode( @@ -146,13 +212,14 @@ describe("StreamProtocolObserver", () => { ) ); observer.observe( - encoder.encode(`data: ${"x".repeat(REPLAY_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1)}`) + encoder.encode(`data: ${"x".repeat(STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + 1)}`) ); expect(observer.finish()).toEqual({ sawContent: true, sawTerminal: false, - failure: { verdict: "malformed", eventName: null }, + observationIncomplete: true, + failure: null, }); }); }); diff --git a/tests/unit/proxy/stream-gate-sse-frames.test.ts b/tests/unit/proxy/stream-gate-sse-frames.test.ts index 80339ba71..a27341ca6 100644 --- a/tests/unit/proxy/stream-gate-sse-frames.test.ts +++ b/tests/unit/proxy/stream-gate-sse-frames.test.ts @@ -34,6 +34,13 @@ describe("SseFrameParser", () => { expect(frames).toEqual([{ eventName: null, data: "x" }]); }); + it("parses newline-delimited raw JSON frames", () => { + expect(parseSseBody('{"candidates":[]}\n{"error":{"message":"failed"}}\n')).toEqual([ + { eventName: null, data: '{"candidates":[]}' }, + { eventName: null, data: '{"error":{"message":"failed"}}' }, + ]); + }); + it("event without data emits no frame and resets event name", () => { const frames = parseSseBody("event: orphan\n\ndata: y\n\n"); expect(frames).toEqual([{ eventName: null, data: "y" }]); @@ -116,6 +123,44 @@ describe("SseFrameParser", () => { ); }); + it("supports a separately bounded buffer exemption for recognized frames", () => { + const parser = new SseFrameParser({ + bufferLimitExemption: { + maxBufferedCharacters: 128, + matches: (eventName, dataHead) => + eventName === "response.created" && dataHead.includes('"type":"response.created"'), + }, + maxBufferedCharacters: 16, + }); + const data = '{"type":"response.created","response":{"input":"large"}}'; + + expect( + collectAll(parser, [new TextEncoder().encode(`event: response.created\ndata: ${data}\n\n`)]) + ).toEqual([{ eventName: "response.created", data }]); + }); + + it("keeps the exemption bounded and resets it after the matching frame", () => { + const parser = new SseFrameParser({ + bufferLimitExemption: { + maxBufferedCharacters: 80, + matches: (eventName) => eventName === "response.created", + }, + maxBufferedCharacters: 16, + }); + const encoder = new TextEncoder(); + + expect(() => + parser.push(encoder.encode(`event: response.created\ndata: ${"x".repeat(81)}\n\n`)) + ).toThrow(SseFrameBufferLimitError); + + expect( + parser.push(encoder.encode("event: response.created\ndata: 12345678901234567890\n\n")) + ).toEqual([{ eventName: "response.created", data: "12345678901234567890" }]); + expect(() => parser.push(encoder.encode(`data: ${"x".repeat(20)}`))).toThrow( + SseFrameBufferLimitError + ); + }); + it("keeps the existing unlimited behavior when no retained buffer limit is configured", () => { const data = "x".repeat(1024 * 1024 + 1); From 947036d422d3dd9a21bdc3475807cccb84c99a3e Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 1 Aug 2026 23:10:03 +0800 Subject: [PATCH 2/2] fix(proxy): address stream review edge cases --- src/app/v1/_lib/proxy/discovery-validity.ts | 40 ++++--- src/app/v1/_lib/proxy/response-handler.ts | 35 +++--- .../proxy/stream-gate/frame-classifier.ts | 21 +++- .../stream-gate/stream-protocol-observer.ts | 22 +++- tests/unit/proxy/discovery-validity.test.ts | 9 ++ .../response-handler-stream-terminal.test.ts | 100 +++++++++++++++++- .../stream-gate-frame-classifier.test.ts | 10 ++ .../stream-gate-protocol-observer.test.ts | 25 +++++ 8 files changed, 229 insertions(+), 33 deletions(-) diff --git a/src/app/v1/_lib/proxy/discovery-validity.ts b/src/app/v1/_lib/proxy/discovery-validity.ts index c85b594fd..acc9aa335 100644 --- a/src/app/v1/_lib/proxy/discovery-validity.ts +++ b/src/app/v1/_lib/proxy/discovery-validity.ts @@ -72,32 +72,38 @@ function classifyProtocolFrame( const family = DISCOVERY_PROTOCOL_FAMILIES[protocol]; if (!family) return { ready: false, terminal: false, error: false }; - let verdict = classifyFrame(family, eventName, data); - if (verdict !== "neutral") return validityFromVerdict(verdict); - + let parsed: unknown; try { - const value = JSON.parse(data) as unknown; - if (isDiscoveryProtocolErrorPayload(value)) { + parsed = JSON.parse(data) as unknown; + // 同帧的通用失败标志优先于 content；fake-200 失败响应不能赢得 Discovery。 + if (isDiscoveryProtocolErrorPayload(parsed)) { return { ready: false, terminal: true, error: true }; } + } catch { + parsed = undefined; + } - // Gemini SDK wrappers may expose the native candidate chunk under response. - if ( - family === "gemini" && - value && - typeof value === "object" && - !Array.isArray(value) && - (value as Record).response && - typeof (value as Record).response === "object" - ) { + let verdict = classifyFrame(family, eventName, data); + if (verdict !== "neutral") return validityFromVerdict(verdict); + + // Gemini SDK wrappers may expose the native candidate chunk under response. + if ( + family === "gemini" && + parsed && + typeof parsed === "object" && + !Array.isArray(parsed) && + (parsed as Record).response && + typeof (parsed as Record).response === "object" + ) { + try { verdict = classifyFrame( family, eventName, - JSON.stringify((value as Record).response) + JSON.stringify((parsed as Record).response) ); + } catch { + return validityFromVerdict("malformed"); } - } catch { - return validityFromVerdict("malformed"); } return validityFromVerdict(verdict); diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index 0144d9f28..a0eca2cdb 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -1398,7 +1398,8 @@ export function hasStreamCompletionMarker( function hasTerminalStreamUsageEvidence( text: string, - format: ProxySession["originalFormat"] + format: ProxySession["originalFormat"], + hasCompletionMarker: boolean ): boolean { const events = parseSSEData(text); const hasPositiveUsage = (value: unknown): boolean => @@ -1406,13 +1407,16 @@ function hasTerminalStreamUsageEvidence( switch (format) { case "response": - return events.some((event) => { - if (!isRecord(event.data)) return false; - const type = event.data.type; - if (type !== "response.completed" && type !== "response.done") return false; - const response = isRecord(event.data.response) ? event.data.response : null; - return hasPositiveUsage(event.data.usage) || hasPositiveUsage(response?.usage); - }); + return ( + hasCompletionMarker && + events.some((event) => { + if (!isRecord(event.data)) return false; + const type = event.data.type; + if (type !== "response.completed" && type !== "response.done") return false; + const response = isRecord(event.data.response) ? event.data.response : null; + return hasPositiveUsage(event.data.usage) || hasPositiveUsage(response?.usage); + }) + ); case "claude": { const hasTerminalUsage = events.some((event) => { if (!isRecord(event.data)) return false; @@ -1421,7 +1425,7 @@ function hasTerminalStreamUsageEvidence( const delta = isRecord(event.data.delta) ? event.data.delta : null; return hasPositiveUsage(event.data.usage) || hasPositiveUsage(delta?.usage); }); - return hasTerminalUsage && inspectStreamCompletion(text, format).hasMarker; + return hasTerminalUsage && hasCompletionMarker; } case "openai": { const hasTerminalUsage = events.some((event) => { @@ -1432,7 +1436,7 @@ function hasTerminalStreamUsageEvidence( hasOpenAIChatCompletionMarker(event.data) ); }); - return hasTerminalUsage && inspectStreamCompletion(text, format).hasMarker; + return hasTerminalUsage && hasCompletionMarker; } case "gemini": case "gemini-cli": { @@ -1868,7 +1872,9 @@ function finalizeDeferredStreamingFinalizationIfNeeded( : ({ isError: false } as const); const protocolFailure = protocolObservation?.failure ?? null; const replayIneligibleReason = - protocolFailure?.verdict === "malformed" ? ("protocol_malformed" as const) : undefined; + protocolFailure?.verdict === "malformed" || protocolFailure?.sawMalformed + ? ("protocol_malformed" as const) + : undefined; const postcommitMalformed = streamEndedNormally && upstreamStatusCode >= 200 && @@ -1876,7 +1882,12 @@ function finalizeDeferredStreamingFinalizationIfNeeded( protocolFailure?.verdict === "malformed" && protocolFailure.afterContent; const successfulPostcommitMalformed = - postcommitMalformed && hasTerminalStreamUsageEvidence(allContent, session.originalFormat); + postcommitMalformed && + hasTerminalStreamUsageEvidence( + allContent, + session.originalFormat, + completionInspection.hasMarker + ); const successfulHttpProtocolFailure = streamEndedNormally && upstreamStatusCode >= 200 && diff --git a/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts b/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts index 3d8891c3e..0db984c66 100644 --- a/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts +++ b/src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts @@ -316,13 +316,14 @@ export function classifyFrame( data: string ): FrameVerdict { try { - return classifyFrameInner(STREAM_SIGNALS[family], eventName, data); + return classifyFrameInner(family, STREAM_SIGNALS[family], eventName, data); } catch { return "neutral"; } } function classifyFrameInner( + family: ProtocolFamily, signal: StreamSignal, eventName: string | null, data: string @@ -348,6 +349,24 @@ function classifyFrameInner( return "malformed"; } + const outerVerdict = classifyParsedFrame(signal, eventName, parsed); + if (outerVerdict !== "neutral" || family !== "gemini" || Array.isArray(parsed)) { + return outerVerdict; + } + + // Gemini SDK 可能把原生 chunk 包在 response 中；先保留 envelope 外层错误优先级， + // 只有外层中性时才解包，供所有门控与 observer 共用同一分类结果。 + const response = (parsed as Record).response; + return response && typeof response === "object" && !Array.isArray(response) + ? classifyParsedFrame(signal, eventName, response) + : outerVerdict; +} + +function classifyParsedFrame( + signal: StreamSignal, + eventName: string | null, + parsed: object +): FrameVerdict { let effective = (eventName ?? "").trim(); if (effective === "" && !Array.isArray(parsed)) { const typeField = (parsed as Record).type; diff --git a/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts b/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts index 930b676cd..734b2be83 100644 --- a/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts +++ b/src/app/v1/_lib/proxy/stream-gate/stream-protocol-observer.ts @@ -9,6 +9,7 @@ export interface StreamProtocolFailure { afterContent: boolean; verdict: "error" | "malformed"; eventName: string | null; + sawMalformed?: true; } export interface StreamProtocolObservation { @@ -33,7 +34,10 @@ export function createStreamProtocolObserver(family: ProtocolFamily): StreamProt bufferLimitExemption: { // 门禁对 request echo 的豁免额度最多把总缓冲抬到 2x cap；observer 采用同一边界， // 允许合法的大请求回显，同时继续阻止伪装 echo 的无界单帧。 - maxBufferedCharacters: streamGatePrebufferCharacters * 2, + maxBufferedCharacters: Math.max( + streamGatePrebufferCharacters * 2, + STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS + ), matches: (eventName, dataHead) => isRequestEchoFrame(family, eventName, dataHead), }, maxBufferedCharacters: STREAM_PROTOCOL_OBSERVER_MAX_BUFFER_CHARACTERS, @@ -56,12 +60,26 @@ export function createStreamProtocolObserver(family: ProtocolFamily): StreamProt const verdict = classifyFrame(family, frame.eventName, frame.data); if (verdict === "content") observation.sawContent = true; if (verdict === "terminal") observation.sawTerminal = true; - if ((verdict === "error" || verdict === "malformed") && !observation.failure) { + if (verdict !== "error" && verdict !== "malformed") return; + + if (!observation.failure) { + observation.failure = { + afterContent: observation.sawContent, + verdict, + eventName: frame.eventName, + }; + return; + } + + if (verdict === "error" && observation.failure.verdict === "malformed") { observation.failure = { afterContent: observation.sawContent, verdict, eventName: frame.eventName, + sawMalformed: true, }; + } else if (verdict === "malformed" && observation.failure.verdict === "error") { + observation.failure = { ...observation.failure, sawMalformed: true }; } }; diff --git a/tests/unit/proxy/discovery-validity.test.ts b/tests/unit/proxy/discovery-validity.test.ts index f769d99f9..9ff586c06 100644 --- a/tests/unit/proxy/discovery-validity.test.ts +++ b/tests/unit/proxy/discovery-validity.test.ts @@ -80,6 +80,15 @@ describe("discovery validity", () => { }); }); + it("keeps a Responses error terminal when the same frame also carries content", () => { + expect( + classifyDiscoveryChunk( + '{"type":"response.output_text.delta","delta":"must not win","failed":true}', + "openai-responses" + ) + ).toEqual({ ready: false, terminal: true, error: true }); + }); + it("does not promote empty tool or content events", () => { expect( classifyDiscoveryChunk( diff --git a/tests/unit/proxy/response-handler-stream-terminal.test.ts b/tests/unit/proxy/response-handler-stream-terminal.test.ts index 68f2f8b48..8486af410 100644 --- a/tests/unit/proxy/response-handler-stream-terminal.test.ts +++ b/tests/unit/proxy/response-handler-stream-terminal.test.ts @@ -339,6 +339,28 @@ describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { ); }); + it("treats malformed after wrapped Gemini content as postcommit", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "gemini" }); + session.originalFormat = "gemini"; + const body = [ + '{"response":{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}}', + "{not-json}", + '{"response":{"candidates":[{"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2}}}', + "", + ].join("\n"); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ inputTokens: 10, outputTokens: 2, statusCode: 200 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + it("bills a naturally completed postcommit malformed stream with terminal usage but rejects Replay", async () => { const { session } = await createSession({}); session.setProvider({ ...createProvider(), providerType: "codex" }); @@ -386,10 +408,44 @@ describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { expect.objectContaining({ inputTokens: 10, outputTokens: 2, statusCode: 200 }), expect.objectContaining({ onCommitted: expect.any(Function) }) ); - expect(mocks.replayAbort).toHaveBeenCalledWith(expect.stringContaining("protocol_malformed")); + expect(mocks.replayAbort).toHaveBeenLastCalledWith("protocol_malformed"); expect(mocks.replayComplete).not.toHaveBeenCalled(); }); + it("rejects a Responses usage payload without a valid completion marker after malformed", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "codex" }); + session.originalFormat = "response"; + setDeferredStreamingFinalization(session, { + providerId: session.provider?.id ?? 0, + providerName: session.provider?.name ?? "provider", + providerPriority: session.provider?.priority ?? 0, + attemptNumber: 1, + totalProvidersAttempted: 1, + isFirstAttempt: true, + isFailoverSuccess: false, + endpointId: null, + endpointUrl: session.provider?.url ?? "", + upstreamStatusCode: 200, + bindingIntent: "none", + }); + const body = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + 'event: unrelated.event\ndata: {"type":"response.completed","usage":{"output_tokens":2}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect(await returned.text()).toBe(body); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + }); + it.each([ { label: "Anthropic", @@ -788,6 +844,48 @@ describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { ); }); + it("does not recover malformed when a later protocol error is outside the bounded snapshot", async () => { + const { session } = await createSession({}); + session.setProvider({ ...createProvider(), providerType: "codex" }); + session.originalFormat = "response"; + session.replayState = { + role: "owner", + ownerToken: "owner-token-malformed-then-error", + identity: { + replayId: "replay-malformed-then-error", + verifier: "verifier", + scopeTag: "scope-tag", + keyId: KEY.id, + userId: USER.id, + format: "response", + model: "gpt-test", + endpoint: "/v1/responses", + }, + }; + const headFiller = "x".repeat(2 * 1024 * 1024); + const tailFiller = "y".repeat(10 * 1024 * 1024); + const body = [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + `event: response.in_progress\ndata: ${JSON.stringify({ type: "response.in_progress", headFiller })}\n\n`, + 'event: response.failed\ndata: {"type":"response.failed","response":{"status":"failed"}}\n\n', + `event: response.in_progress\ndata: ${JSON.stringify({ type: "response.in_progress", tailFiller })}\n\n`, + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":10,"output_tokens":2}}}\n\n', + ].join(""); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(body)); + expect((await returned.text()).length).toBe(body.length); + await settleTasks(); + + expect(mocks.durable).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ statusCode: 502 }), + expect.objectContaining({ onCommitted: expect.any(Function) }) + ); + expect(mocks.replayAbort).toHaveBeenLastCalledWith("protocol_malformed"); + expect(mocks.replayComplete).not.toHaveBeenCalled(); + }); + it("aborts Replay when a successful non-200 2xx stream contains a late protocol error", async () => { const { session } = await createSession({}); session.setProvider({ ...createProvider(), providerType: "codex" }); diff --git a/tests/unit/proxy/stream-gate-frame-classifier.test.ts b/tests/unit/proxy/stream-gate-frame-classifier.test.ts index aed7fcab8..b3b4ae08b 100644 --- a/tests/unit/proxy/stream-gate-frame-classifier.test.ts +++ b/tests/unit/proxy/stream-gate-frame-classifier.test.ts @@ -420,6 +420,16 @@ describe("classifyFrame: gemini", () => { ).toBe("error"); }); + it("error precedence: wrapper error beats response content", () => { + expect( + classifyFrame( + "gemini", + null, + '{"error":{"message":"failed"},"response":{"candidates":[{"content":{"parts":[{"text":"must not commit"}]}}]}}' + ) + ).toBe("error"); + }); + it("content precedence: normal STOP with text is content, not terminal", () => { expect( classifyFrame( diff --git a/tests/unit/proxy/stream-gate-protocol-observer.test.ts b/tests/unit/proxy/stream-gate-protocol-observer.test.ts index 08431abd2..0a5e705cc 100644 --- a/tests/unit/proxy/stream-gate-protocol-observer.test.ts +++ b/tests/unit/proxy/stream-gate-protocol-observer.test.ts @@ -74,6 +74,31 @@ describe("StreamProtocolObserver", () => { }); }); + test("显式 protocol error 覆盖较早的 malformed，但保留 Replay 禁用证据", () => { + const observer = createStreamProtocolObserver("openai-responses"); + observer.observe( + encoder.encode( + [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"ok"}\n\n', + "event: response.in_progress\ndata: not-json\n\n", + 'event: response.failed\ndata: {"type":"response.failed","response":{"status":"failed"}}\n\n', + ].join("") + ) + ); + + expect(observer.finish()).toEqual({ + sawContent: true, + sawTerminal: false, + observationIncomplete: false, + failure: { + afterContent: true, + verdict: "error", + eventName: "response.failed", + sawMalformed: true, + }, + }); + }); + test("Anthropic tool metadata alone does not count as committed content", () => { const observer = createStreamProtocolObserver("anthropic"); observer.observe(