diff --git a/src/server/routes/proxy/responses.codex-oauth.test.ts b/src/server/routes/proxy/responses.codex-oauth.test.ts index 9a4a6d1..3e692ec 100644 --- a/src/server/routes/proxy/responses.codex-oauth.test.ts +++ b/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({ diff --git a/src/server/routes/proxy/responses.websocket.test.ts b/src/server/routes/proxy/responses.websocket.test.ts index a88c896..540c6f3 100644 --- a/src/server/routes/proxy/responses.websocket.test.ts +++ b/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'); }); }); diff --git a/src/server/transformers/openai/responses/proxyStream.ts b/src/server/transformers/openai/responses/proxyStream.ts index d27191d..b938bed 100644 --- a/src/server/transformers/openai/responses/proxyStream.ts +++ b/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; }