From 8672ff8dff60522d9e72eb2a6004cf63bdc2a8d5 Mon Sep 17 00:00:00 2001 From: cita Date: Thu, 19 Mar 2026 23:20:40 +0800 Subject: [PATCH] Handle mislabeled responses SSE bodies --- src/server/proxy-core/surfaces/chatSurface.ts | 50 ++++++++++++++++-- .../surfaces/openAiResponsesSurface.ts | 52 +++++++++++++++++-- .../routes/proxy/chat.codex-oauth.test.ts | 34 ++++++++++++ .../proxy/responses.codex-oauth.test.ts | 35 +++++++++++++ src/server/routes/proxy/responsesSseFinal.ts | 47 +++++++++++++++-- 5 files changed, 204 insertions(+), 14 deletions(-) diff --git a/src/server/proxy-core/surfaces/chatSurface.ts b/src/server/proxy-core/surfaces/chatSurface.ts index 9828151..4c619fc 100644 --- a/src/server/proxy-core/surfaces/chatSurface.ts +++ b/src/server/proxy-core/surfaces/chatSurface.ts @@ -39,7 +39,12 @@ import { import { getOauthInfoFromExtraConfig } from '../../services/oauth/oauthAccount.js'; import { recordOauthQuotaResetHint } from '../../services/oauth/quota.js'; import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js'; -import { collectResponsesFinalPayloadFromSse } from '../../routes/proxy/responsesSseFinal.js'; +import { + collectResponsesFinalPayloadFromSse, + collectResponsesFinalPayloadFromSseText, + createSingleChunkStreamReader, + looksLikeResponsesSseText, +} from '../../routes/proxy/responsesSseFinal.js'; import { createGeminiCliStreamReader, unwrapGeminiCliPayload, @@ -394,6 +399,37 @@ export async function handleChatSurfaceRequest( if (!upstreamContentType.includes('text/event-stream')) { const fallbackText = await upstream.text(); rawText = fallbackText; + if (looksLikeResponsesSseText(fallbackText)) { + startSseResponse(); + const streamResult = await streamSession.run( + createSingleChunkStreamReader(fallbackText), + reply.raw, + ); + const latency = Date.now() - startTime; + if (streamResult.status === 'failed') { + tokenRouter.recordFailure(selected.channel.id); + logProxy( + selected, + requestedModel, + 'failed', + 200, + latency, + streamResult.errorMessage, + retryCount, + downstreamPath, + parsedUsage.promptTokens, + parsedUsage.completionTokens, + parsedUsage.totalTokens, + 0, + null, + successfulUpstreamPath, + clientContext, + logDownstreamApiKeyId ? downstreamApiKeyId : null, + ); + return; + } + return; + } let fallbackData: unknown = null; try { fallbackData = JSON.parse(fallbackText); @@ -580,11 +616,15 @@ export async function handleChatSurfaceRequest( upstreamData = collected.payload; } else { rawText = await upstream.text(); - upstreamData = rawText; - try { - upstreamData = JSON.parse(rawText); - } catch { + if (looksLikeResponsesSseText(rawText)) { + upstreamData = collectResponsesFinalPayloadFromSseText(rawText, modelName).payload; + } else { upstreamData = rawText; + try { + upstreamData = JSON.parse(rawText); + } catch { + upstreamData = rawText; + } } } if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') { diff --git a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts index d35aec1..98d9537 100644 --- a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts +++ b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts @@ -34,7 +34,12 @@ import { import { getOauthInfoFromExtraConfig } from '../../services/oauth/oauthAccount.js'; import { recordOauthQuotaResetHint } from '../../services/oauth/quota.js'; import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js'; -import { collectResponsesFinalPayloadFromSse } from '../../routes/proxy/responsesSseFinal.js'; +import { + collectResponsesFinalPayloadFromSse, + collectResponsesFinalPayloadFromSseText, + createSingleChunkStreamReader, + looksLikeResponsesSseText, +} from '../../routes/proxy/responsesSseFinal.js'; import { createGeminiCliStreamReader, unwrapGeminiCliPayload, @@ -541,6 +546,39 @@ export async function handleOpenAiResponsesSurfaceRequest( }); if (!upstreamContentType.includes('text/event-stream')) { const rawText = await upstream.text(); + if (looksLikeResponsesSseText(rawText)) { + startSseResponse(); + const streamResult = await streamSession.run( + createSingleChunkStreamReader(rawText), + reply.raw, + ); + const latency = Date.now() - startTime; + if (streamResult.status === 'failed') { + tokenRouter.recordFailure(selected.channel.id); + logProxy( + selected, + requestedModel, + 'failed', + 200, + latency, + streamResult.errorMessage, + retryCount, + downstreamPath, + parsedUsage.promptTokens, + parsedUsage.completionTokens, + parsedUsage.totalTokens, + 0, + null, + successfulUpstreamPath, + clientContext, + logDownstreamApiKeyId ? downstreamApiKeyId : null, + ); + return; + } + + await finalizeStreamSuccess(parsedUsage, latency); + return; + } let upstreamData: unknown = rawText; try { upstreamData = JSON.parse(rawText); @@ -692,11 +730,15 @@ export async function handleOpenAiResponsesSurfaceRequest( upstreamData = collected.payload; } else { rawText = await upstream.text(); - upstreamData = rawText; - try { - upstreamData = JSON.parse(rawText); - } catch { + if (looksLikeResponsesSseText(rawText)) { + upstreamData = collectResponsesFinalPayloadFromSseText(rawText, modelName).payload; + } else { upstreamData = rawText; + try { + upstreamData = JSON.parse(rawText); + } catch { + upstreamData = rawText; + } } } if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') { diff --git a/src/server/routes/proxy/chat.codex-oauth.test.ts b/src/server/routes/proxy/chat.codex-oauth.test.ts index 60f6bc1..fb7177b 100644 --- a/src/server/routes/proxy/chat.codex-oauth.test.ts +++ b/src/server/routes/proxy/chat.codex-oauth.test.ts @@ -245,4 +245,38 @@ describe('chat proxy codex oauth compatibility', () => { expect(response.body).not.toContain('event: response.created'); expect(response.body).not.toContain('"type":"response.output_text.delta"'); }); + + it('still translates raw codex SSE into OpenAI chat completion chunks when upstream mislabels the content-type', async () => { + fetchMock.mockResolvedValue(new Response([ + 'event: response.created\n', + 'data: {"type":"response.created","response":{"id":"resp_codex_openai_stream_header_miss","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n', + 'event: response.output_item.added\n', + 'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_codex_openai_stream_header_miss","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n', + 'event: response.output_text.delta\n', + 'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_codex_openai_stream_header_miss","delta":"pong from codex"}\n\n', + 'event: response.completed\n', + 'data: {"type":"response.completed","response":{"id":"resp_codex_openai_stream_header_miss","model":"gpt-5.4","status":"completed","usage":{"input_tokens":9,"output_tokens":3,"total_tokens":12}}}\n\n', + 'data: [DONE]\n\n', + ].join(''), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/chat/completions', + payload: { + model: 'gpt-5.4', + stream: true, + messages: [{ role: 'user', content: 'hello codex' }], + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.headers['content-type']).toContain('text/event-stream'); + expect(response.body).toContain('"object":"chat.completion.chunk"'); + expect(response.body).toContain('pong from codex'); + expect(response.body).not.toContain('event: response.created'); + expect(response.body).not.toContain('"type":"response.output_text.delta"'); + }); }); diff --git a/src/server/routes/proxy/responses.codex-oauth.test.ts b/src/server/routes/proxy/responses.codex-oauth.test.ts index b2f3ecd..4384990 100644 --- a/src/server/routes/proxy/responses.codex-oauth.test.ts +++ b/src/server/routes/proxy/responses.codex-oauth.test.ts @@ -615,4 +615,39 @@ describe('responses proxy codex oauth refresh', () => { expect(recordFailureMock).not.toHaveBeenCalled(); expect(recordSuccessMock).toHaveBeenCalledTimes(1); }); + + it('replays raw codex SSE when upstream mislabels a streaming responses body as application/json', async () => { + fetchMock.mockResolvedValue(new Response([ + 'event: response.created\n', + 'data: {"type":"response.created","response":{"id":"resp_codex_stream_header_miss","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n', + 'event: response.output_item.added\n', + 'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_codex_stream_header_miss","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n', + 'event: response.output_text.delta\n', + 'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_codex_stream_header_miss","delta":"pong"}\n\n', + 'event: response.completed\n', + 'data: {"type":"response.completed","response":{"id":"resp_codex_stream_header_miss","model":"gpt-5.4","status":"completed","usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n', + 'data: [DONE]\n\n', + ].join(''), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.4', + input: 'hello codex', + stream: true, + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.body).toContain('event: response.output_text.delta'); + expect(response.body).toContain('"delta":"pong"'); + expect(response.body).not.toContain('"output_text":"event: response.created'); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(recordFailureMock).not.toHaveBeenCalled(); + expect(recordSuccessMock).toHaveBeenCalledTimes(1); + }); }); diff --git a/src/server/routes/proxy/responsesSseFinal.ts b/src/server/routes/proxy/responsesSseFinal.ts index 12a8e43..3fee5ac 100644 --- a/src/server/routes/proxy/responsesSseFinal.ts +++ b/src/server/routes/proxy/responsesSseFinal.ts @@ -120,11 +120,43 @@ function materializeCompletedPayloadFromAggregate( return null; } -export async function collectResponsesFinalPayloadFromSse( - upstream: { text(): Promise }, +export function looksLikeResponsesSseText(rawText: string): boolean { + const { events, rest } = openAiResponsesTransformer.pullSseEvents(rawText); + if (events.length === 0 || rest.trim().length > 0) return false; + return events.some((event) => { + if (event.data === '[DONE]') return true; + if (event.event === 'error' || event.event.startsWith('response.')) return true; + const payload = parseResponsesSsePayload(event.data); + const payloadType = typeof payload?.type === 'string' ? payload.type : ''; + return payloadType === 'error' || payloadType.startsWith('response.'); + }); +} + +export function createSingleChunkStreamReader(rawText: string): { + read(): Promise<{ done: boolean; value?: Uint8Array }>; + cancel(reason?: unknown): Promise; + releaseLock(): void; +} { + const chunk = Buffer.from(rawText, 'utf8'); + let done = false; + return { + async read() { + if (done) return { done: true }; + done = true; + return { done: false, value: chunk }; + }, + async cancel() { + done = true; + return undefined; + }, + releaseLock() {}, + }; +} + +export function collectResponsesFinalPayloadFromSseText( + rawText: string, modelName: string, -): Promise<{ payload: Record; rawText: string }> { - const rawText = await upstream.text(); +): { payload: Record; rawText: string } { const { events } = openAiResponsesTransformer.pullSseEvents(rawText); const streamContext = openAiResponsesTransformer.createStreamContext(modelName); const aggregateState = openAiResponsesTransformer.aggregator.createState(modelName); @@ -223,3 +255,10 @@ export async function collectResponsesFinalPayloadFromSse( throw new Error('stream disconnected before response.completed'); } + +export async function collectResponsesFinalPayloadFromSse( + upstream: { text(): Promise }, + modelName: string, +): Promise<{ payload: Record; rawText: string }> { + return collectResponsesFinalPayloadFromSseText(await upstream.text(), modelName); +}