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

[codex] fix responses empty completion handling

src/server/routes/proxy/responses.codex-oauth.test.ts

@@ -1,5 +1,6 @@
 import Fastify, { type FastifyInstance } from 'fastify';
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { config } from '../../config.js';
 
 const fetchMock = vi.fn();
 const selectChannelMock = vi.fn();
@@ -17,6 +18,7 @@ const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) =>
 const refreshOauthAccessTokenSingleflightMock = vi.fn();
 const recordOauthQuotaResetHintMock = vi.fn();
 const insertedProxyLogs: Record<string, unknown>[] = [];
+const originalProxyEmptyContentFailEnabled = config.proxyEmptyContentFailEnabled;
 const dbInsertMock = vi.fn((_arg?: any) => ({
   values: (values: Record<string, unknown>) => {
     insertedProxyLogs.push(values);
@@ -109,6 +111,7 @@ describe('responses proxy codex oauth refresh', () => {
   });
 
   beforeEach(() => {
+    config.proxyEmptyContentFailEnabled = false;
     fetchMock.mockReset();
     selectChannelMock.mockReset();
     selectNextChannelMock.mockReset();
@@ -152,6 +155,7 @@ describe('responses proxy codex oauth refresh', () => {
   });
 
   afterAll(async () => {
+    config.proxyEmptyContentFailEnabled = originalProxyEmptyContentFailEnabled;
     await app.close();
   });
 
@@ -501,6 +505,39 @@ describe('responses proxy codex oauth refresh', () => {
     expect(String(insertedProxyLogs.at(-1)?.errorMessage || '')).toContain('stream closed before response.completed');
   });
 
+  it('does not record success when a native responses stream completes with empty content and empty usage while empty-content failure is enabled', async () => {
+    config.proxyEmptyContentFailEnabled = true;
+
+    fetchMock.mockResolvedValue(createSseResponse([
+      'event: response.created\n',
+      'data: {"type":"response.created","response":{"id":"resp_codex_empty","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
+      'event: response.completed\n',
+      'data: {"type":"response.completed","response":{"id":"resp_codex_empty","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}\n\n',
+      'data: [DONE]\n\n',
+    ]));
+
+    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.failed');
+    expect(response.body).not.toContain('event: response.completed');
+    expect(recordSuccessMock).not.toHaveBeenCalled();
+    expect(recordFailureMock).toHaveBeenCalledTimes(1);
+    expect(insertedProxyLogs.at(-1)).toMatchObject({
+      status: 'failed',
+      httpStatus: 200,
+    });
+    expect(String(insertedProxyLogs.at(-1)?.errorMessage || '')).toContain('empty content');
+  });
+
   it('does not retry or mark failure after converting a non-stream upstream payload into SSE when post-stream usage accounting fails', async () => {
     resolveProxyUsageWithSelfLogFallbackMock.mockRejectedValueOnce(new Error('usage accounting failed'));
     fetchMock.mockResolvedValue(new Response(JSON.stringify({

src/server/routes/proxy/responsesSseFinal.test.ts

@@ -20,4 +20,56 @@ describe('collectResponsesFinalPayloadFromSse', () => {
       .rejects
       .toThrow('quota exceeded');
   });
+
+  it('prefers aggregated stream content when response.completed only carries an empty output array', async () => {
+    const upstream = {
+      async text() {
+        return [
+          'event: response.created',
+          'data: {"type":"response.created","response":{"id":"resp_empty_completed","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
+          '',
+          'event: response.output_item.added',
+          'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_empty_completed","type":"message","role":"assistant","status":"in_progress","content":[]}}',
+          '',
+          'event: response.output_text.delta',
+          'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_empty_completed","delta":"pong"}',
+          '',
+          'event: response.completed',
+          'data: {"type":"response.completed","response":{"id":"resp_empty_completed","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
+          '',
+          'data: [DONE]',
+          '',
+        ].join('\n');
+      },
+    };
+
+    await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
+      .resolves
+      .toMatchObject({
+        payload: {
+          id: 'resp_empty_completed',
+          status: 'completed',
+          output: [
+            {
+              id: 'msg_empty_completed',
+              type: 'message',
+              role: 'assistant',
+              status: 'completed',
+              content: [
+                {
+                  type: 'output_text',
+                  text: 'pong',
+                },
+              ],
+            },
+          ],
+          output_text: 'pong',
+          usage: {
+            input_tokens: 3,
+            output_tokens: 1,
+            total_tokens: 4,
+          },
+        },
+      });
+  });
 });

src/server/routes/proxy/responsesSseFinal.ts

@@ -24,6 +24,36 @@ function getResponsesFailureMessage(payload: Record<string, unknown>): string {
   return 'upstream stream failed';
 }
 
+function hasMeaningfulMessageContent(content: unknown): boolean {
+  if (!Array.isArray(content)) return false;
+  return content.some((part) => {
+    if (!isRecord(part)) return false;
+    const partType = typeof part.type === 'string' ? part.type.trim().toLowerCase() : '';
+    if (partType === 'output_text' || partType === 'text') {
+      return typeof part.text === 'string' && part.text.length > 0;
+    }
+    return true;
+  });
+}
+
+function hasMeaningfulResponsesOutput(output: unknown): boolean {
+  if (!Array.isArray(output)) return false;
+  return output.some((item) => {
+    if (!isRecord(item)) return false;
+    const itemType = typeof item.type === 'string' ? item.type.trim().toLowerCase() : '';
+    if (itemType === 'message') {
+      return hasMeaningfulMessageContent(item.content);
+    }
+    if (itemType === 'reasoning') {
+      return (
+        (Array.isArray(item.summary) && item.summary.length > 0)
+        || (typeof item.encrypted_content === 'string' && item.encrypted_content.trim().length > 0)
+      );
+    }
+    return itemType.length > 0;
+  });
+}
+
 function hasCompleteFinalResponsesPayload(payload: Record<string, unknown>): boolean {
   return (
     payload.object === 'response.compaction'
@@ -32,6 +62,64 @@ function hasCompleteFinalResponsesPayload(payload: Record<string, unknown>): boo
   );
 }
 
+function hasMeaningfulFinalResponsesPayload(payload: Record<string, unknown>): boolean {
+  if (payload.object === 'response.compaction') {
+    return Array.isArray(payload.output) && payload.output.length > 0;
+  }
+  if (typeof payload.output_text === 'string' && payload.output_text.length > 0) {
+    return true;
+  }
+  return hasMeaningfulResponsesOutput(payload.output);
+}
+
+function rememberStreamResponseEnvelope(
+  streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
+  payload: Record<string, unknown>,
+): void {
+  const responsePayload = isRecord(payload.response) ? payload.response : payload;
+  if (typeof responsePayload.id === 'string' && responsePayload.id.trim().length > 0) {
+    streamContext.id = responsePayload.id;
+  }
+  if (typeof responsePayload.model === 'string' && responsePayload.model.trim().length > 0) {
+    streamContext.model = responsePayload.model;
+  }
+  const createdAt = (
+    typeof responsePayload.created_at === 'number' && Number.isFinite(responsePayload.created_at)
+      ? responsePayload.created_at
+      : (typeof responsePayload.created === 'number' && Number.isFinite(responsePayload.created)
+        ? responsePayload.created
+        : null)
+  );
+  if (createdAt !== null) {
+    streamContext.created = createdAt;
+  }
+}
+
+function materializeCompletedPayloadFromAggregate(
+  aggregateState: ReturnType<typeof openAiResponsesTransformer.aggregator.createState>,
+  streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
+  usage: ReturnType<typeof parseProxyUsage>,
+): Record<string, unknown> | null {
+  const lines = openAiResponsesTransformer.aggregator.complete(
+    aggregateState,
+    streamContext,
+    usage,
+  );
+  const { events } = openAiResponsesTransformer.pullSseEvents(lines.join(''));
+  for (const event of events) {
+    if (event.data === '[DONE]') continue;
+    const payload = parseResponsesSsePayload(event.data);
+    if (!payload) continue;
+    if (event.event === 'response.completed' && isRecord(payload.response)) {
+      return payload.response;
+    }
+    if (payload.type === 'response.completed' && hasCompleteFinalResponsesPayload(payload)) {
+      return payload;
+    }
+  }
+  return null;
+}
+
 export async function collectResponsesFinalPayloadFromSse(
   upstream: { text(): Promise<string> },
   modelName: string,
@@ -92,12 +180,17 @@ export async function collectResponsesFinalPayloadFromSse(
 
     const payloadType = typeof payload.type === 'string' ? payload.type : '';
     const eventType = payloadType || event.event;
+    rememberStreamResponseEnvelope(streamContext, payload);
     usage = mergeProxyUsage(usage, parseProxyUsage(payload));
     captureCompletedPayloadFromEvent(eventType, payload);
     if (completedPayload) {
       continue;
     }
-    const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(payload, streamContext, modelName);
+    const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(
+      payload,
+      streamContext,
+      modelName,
+    );
     captureCompletedPayloadFromLines(openAiResponsesTransformer.aggregator.serialize({
       state: aggregateState,
       streamContext,
@@ -106,12 +199,19 @@ export async function collectResponsesFinalPayloadFromSse(
     }));
   }
 
+  if (
+    completedPayload
+    && !hasMeaningfulFinalResponsesPayload(completedPayload)
+    && hasMeaningfulResponsesOutput(aggregateState.outputItems)
+  ) {
+    completedPayload = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage);
+  }
+
   if (!completedPayload) {
-    captureCompletedPayloadFromLines(openAiResponsesTransformer.aggregator.complete(
+    const materialized = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage);
-      aggregateState,
+    if (materialized) {
-      streamContext,
+      completedPayload = materialized;
-      usage,
+    }
-    ));
   }
 
   if (completedPayload) {

src/server/transformers/openai/responses/proxyStream.ts

@@ -3,6 +3,7 @@ import { type ParsedSseEvent } from '../../shared/normalized.js';
 import { completeResponsesStream, createOpenAiResponsesAggregateState, failResponsesStream, serializeConvertedResponsesEvents } from './aggregator.js';
 import { openAiResponsesOutbound } from './outbound.js';
 import { openAiResponsesStream } from './stream.js';
+import { config } from '../../../config.js';
 
 type StreamReader = {
   read(): Promise<{ done: boolean; value?: Uint8Array }>;
@@ -40,6 +41,62 @@ function isRecord(value: unknown): value is Record<string, unknown> {
   return !!value && typeof value === 'object';
 }
 
+function hasNonEmptyString(value: unknown): boolean {
+  return typeof value === 'string' && value.trim().length > 0;
+}
+
+function hasMeaningfulContentPart(part: unknown): boolean {
+  if (!isRecord(part)) return false;
+  const partType = typeof part.type === 'string' ? part.type.trim().toLowerCase() : '';
+  if (partType === 'output_text' || partType === 'text') {
+    return hasNonEmptyString(part.text);
+  }
+  return partType.length > 0;
+}
+
+function hasMeaningfulOutputItem(item: unknown): boolean {
+  if (!isRecord(item)) return false;
+  const itemType = typeof item.type === 'string' ? item.type.trim().toLowerCase() : '';
+  if (itemType === 'message') {
+    return Array.isArray(item.content) && item.content.some((part) => hasMeaningfulContentPart(part));
+  }
+  if (itemType === 'reasoning') {
+    return (
+      (Array.isArray(item.summary) && item.summary.some((part) => hasMeaningfulContentPart(part)))
+      || hasNonEmptyString(item.encrypted_content)
+    );
+  }
+  return itemType.length > 0;
+}
+
+function hasMeaningfulResponsesPayloadOutput(payload: unknown): boolean {
+  if (!isRecord(payload)) return false;
+  if (hasNonEmptyString(payload.output_text)) return true;
+  return Array.isArray(payload.output) && payload.output.some((item) => hasMeaningfulOutputItem(item));
+}
+
+function hasMeaningfulAggregateOutput(state: ReturnType<typeof createOpenAiResponsesAggregateState>): boolean {
+  return state.outputItems.some((item) => hasMeaningfulOutputItem(item));
+}
+
+function shouldFailEmptyResponsesCompletion(input: {
+  payload: unknown;
+  state: ReturnType<typeof createOpenAiResponsesAggregateState>;
+  usage: {
+    promptTokens: number;
+    completionTokens: number;
+    totalTokens: number;
+  };
+}): boolean {
+  if (!config.proxyEmptyContentFailEnabled) return false;
+  const responsePayload = isRecord(input.payload) && isRecord(input.payload.response)
+    ? input.payload.response
+    : null;
+  if (hasMeaningfulAggregateOutput(input.state)) return false;
+  if (responsePayload && hasMeaningfulResponsesPayloadOutput(responsePayload)) return false;
+  return input.usage.completionTokens <= 0 && input.usage.totalTokens <= 0;
+}
+
 function getResponsesStreamFailureMessage(payload: unknown, fallback = 'upstream stream failed'): string {
   if (isRecord(payload)) {
     if (isRecord(payload.error) && typeof payload.error.message === 'string' && payload.error.message.trim()) {
@@ -145,12 +202,29 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
 
     if (parsedPayload && typeof parsedPayload === 'object') {
       const normalizedEvent = openAiResponsesStream.normalizeEvent(parsedPayload, streamContext, input.modelName);
-      input.writeLines(serializeConvertedResponsesEvents({
+      const convertedLines = serializeConvertedResponsesEvents({
         state: responsesState,
         streamContext,
         event: normalizedEvent,
         usage: input.getUsage(),
-      }));
+      });
+      if (
+        (eventBlock.event === 'response.completed' || payloadType === 'response.completed')
+        && shouldFailEmptyResponsesCompletion({
+          payload: parsedPayload,
+          state: responsesState,
+          usage: input.getUsage(),
+        })
+      ) {
+        fail({
+          type: 'response.failed',
+          error: {
+            message: 'Upstream returned empty content',
+          },
+        }, 'Upstream returned empty content');
+        return true;
+      }
+      input.writeLines(convertedLines);
       if (eventBlock.event === 'response.completed' || payloadType === 'response.completed') {
         complete();
       }
@@ -192,6 +266,20 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
         usage: input.getUsage(),
         serializationMode: 'response',
       });
+      if (shouldFailEmptyResponsesCompletion({
+        payload: { type: 'response.completed', response: streamPayload },
+        state: responsesState,
+        usage: input.getUsage(),
+      })) {
+        fail({
+          type: 'response.failed',
+          error: {
+            message: 'Upstream returned empty content',
+          },
+        }, 'Upstream returned empty content');
+        response?.end();
+        return terminalResult;
+      }
       const createdPayload = {
         ...streamPayload,
         status: 'in_progress',