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

fix: stop false success logging when responses streams end early

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

@@ -468,6 +468,39 @@ describe('responses proxy codex oauth refresh', () => {
     expect(String(insertedProxyLogs.at(-1)?.errorMessage || '')).toContain('tool execution failed');
   });
 
+  it('does not record success when a native responses stream closes before response.completed', async () => {
+    fetchMock.mockResolvedValue(createSseResponse([
+      'event: response.created\n',
+      'data: {"type":"response.created","response":{"id":"resp_codex_truncated","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_truncated","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_truncated","delta":"partial"}\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('stream closed before response.completed');
+  });
+
   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/responses.websocket.test.ts

@@ -526,7 +526,7 @@ describe('responses websocket transport', () => {
 
     const socket = new WebSocket(`${baseUrl}/v1/responses`);
     await waitForSocketOpen(socket);
-    const messagesPromise = waitForSocketMessages(socket, 3, 400);
+    const messagesPromise = waitForSocketMessages(socket, 6, 400);
 
     socket.send(JSON.stringify({
       type: 'response.create',
@@ -546,9 +546,12 @@ describe('responses websocket transport', () => {
     expect(messages.map((message) => message.type)).toEqual([
       'response.created',
       'response.output_text.delta',
-      'error',
+      'response.output_text.done',
+      'response.content_part.done',
+      'response.output_item.done',
+      'response.failed',
     ]);
-    expect(messages[2]?.status).toBe(408);
-    expect(messages[2]?.error?.message).toContain('stream closed before response.completed');
+    expect(messages[5]?.response?.status).toBe('failed');
+    expect(messages[5]?.error?.message).toContain('stream closed before response.completed');
   });
 });

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

@@ -58,6 +58,9 @@ function getResponsesStreamFailureMessage(payload: unknown, fallback = 'upstream
 export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSessionInput) {
   const streamContext = openAiResponsesStream.createContext(input.modelName);
   const responsesState = createOpenAiResponsesAggregateState(input.modelName);
+  const requiresExplicitTerminalEvent = input.strictTerminalEvents
+    || input.successfulUpstreamPath.endsWith('/responses')
+    || input.successfulUpstreamPath.endsWith('/responses/compact');
   let finalized = false;
   let terminalResult: ResponsesProxyStreamResult = {
     status: 'completed',
@@ -84,14 +87,24 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
     input.writeLines(failResponsesStream(responsesState, streamContext, input.getUsage(), payload));
   };
 
+  const complete = () => {
+    if (finalized) return;
+    finalized = true;
+    terminalResult = {
+      status: 'completed',
+      errorMessage: null,
+    };
+  };
+
   const closeOut = () => {
     if (finalized) return;
-    if (input.strictTerminalEvents) {
-      finalized = true;
-      terminalResult = {
-        status: 'failed',
-        errorMessage: 'stream closed before response.completed',
-      };
+    if (requiresExplicitTerminalEvent) {
+      fail({
+        type: 'response.failed',
+        error: {
+          message: 'stream closed before response.completed',
+        },
+      }, 'stream closed before response.completed');
       return;
     }
     finalize();
@@ -120,8 +133,10 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
     const isFailureEvent = (
       eventBlock.event === 'error'
       || eventBlock.event === 'response.failed'
+      || eventBlock.event === 'response.incomplete'
       || payloadType === 'error'
       || payloadType === 'response.failed'
+      || payloadType === 'response.incomplete'
     );
     if (isFailureEvent) {
       fail(parsedPayload);
@@ -136,6 +151,9 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
         event: normalizedEvent,
         usage: input.getUsage(),
       }));
+      if (eventBlock.event === 'response.completed' || payloadType === 'response.completed') {
+        complete();
+      }
       return false;
     }