Merge pull request #177 from cita-777/codex/fix-responses-eof-false-success-clean

[codex] Handle mislabeled responses SSE bodies

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,6 +616,9 @@ export async function handleChatSurfaceRequest(
         upstreamData = collected.payload;
       } else {
         rawText = await upstream.text();
+        if (looksLikeResponsesSseText(rawText)) {
+          upstreamData = collectResponsesFinalPayloadFromSseText(rawText, modelName).payload;
+        } else {
           upstreamData = rawText;
           try {
             upstreamData = JSON.parse(rawText);
@@ -587,6 +626,7 @@ export async function handleChatSurfaceRequest(
             upstreamData = rawText;
           }
         }
+      }
       if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
         upstreamData = unwrapGeminiCliPayload(upstreamData);
       }

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,6 +730,9 @@ export async function handleOpenAiResponsesSurfaceRequest(
           upstreamData = collected.payload;
         } else {
           rawText = await upstream.text();
+          if (looksLikeResponsesSseText(rawText)) {
+            upstreamData = collectResponsesFinalPayloadFromSseText(rawText, modelName).payload;
+          } else {
             upstreamData = rawText;
             try {
               upstreamData = JSON.parse(rawText);
@@ -699,6 +740,7 @@ export async function handleOpenAiResponsesSurfaceRequest(
               upstreamData = rawText;
             }
           }
+        }
         if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
           upstreamData = unwrapGeminiCliPayload(upstreamData);
         }

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"');
+  });
 });

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);
+  });
 });

src/server/routes/proxy/responsesSseFinal.ts

@@ -120,11 +120,43 @@ function materializeCompletedPayloadFromAggregate(
   return null;
 }
 
-export async function collectResponsesFinalPayloadFromSse(
-  upstream: { text(): Promise<string> },
+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<unknown>;
+  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<string, unknown>; rawText: string }> {
-  const rawText = await upstream.text();
+): { payload: Record<string, unknown>; 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<string> },
+  modelName: string,
+): Promise<{ payload: Record<string, unknown>; rawText: string }> {
+  return collectResponsesFinalPayloadFromSseText(await upstream.text(), modelName);
+}