diff --git a/src/server/routes/proxy/responses.codex-oauth.test.ts b/src/server/routes/proxy/responses.codex-oauth.test.ts index 3e692ec..b2f3ecd 100644 --- a/src/server/routes/proxy/responses.codex-oauth.test.ts +++ b/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[] = []; +const originalProxyEmptyContentFailEnabled = config.proxyEmptyContentFailEnabled; const dbInsertMock = vi.fn((_arg?: any) => ({ values: (values: Record) => { 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({ diff --git a/src/server/routes/proxy/responsesSseFinal.test.ts b/src/server/routes/proxy/responsesSseFinal.test.ts index 9db829c..64c96e5 100644 --- a/src/server/routes/proxy/responsesSseFinal.test.ts +++ b/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, + }, + }, + }); + }); }); diff --git a/src/server/routes/proxy/responsesSseFinal.ts b/src/server/routes/proxy/responsesSseFinal.ts index b22eb48..12a8e43 100644 --- a/src/server/routes/proxy/responsesSseFinal.ts +++ b/src/server/routes/proxy/responsesSseFinal.ts @@ -24,6 +24,36 @@ function getResponsesFailureMessage(payload: Record): 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): boolean { return ( payload.object === 'response.compaction' @@ -32,6 +62,64 @@ function hasCompleteFinalResponsesPayload(payload: Record): boo ); } +function hasMeaningfulFinalResponsesPayload(payload: Record): 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, + payload: Record, +): 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, + streamContext: ReturnType, + usage: ReturnType, +): Record | 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 }, 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( - aggregateState, - streamContext, - usage, - )); + const materialized = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage); + if (materialized) { + completedPayload = materialized; + } } if (completedPayload) { diff --git a/src/server/transformers/openai/responses/proxyStream.ts b/src/server/transformers/openai/responses/proxyStream.ts index b938bed..3f0e67d 100644 --- a/src/server/transformers/openai/responses/proxyStream.ts +++ b/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 { 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): boolean { + return state.outputItems.some((item) => hasMeaningfulOutputItem(item)); +} + +function shouldFailEmptyResponsesCompletion(input: { + payload: unknown; + state: ReturnType; + 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',