1
0
forked from boos/BoosAPI

fix responses empty completion handling

This commit is contained in:
cita
2026-03-19 22:22:02 +08:00
parent 8f0b5c81f4
commit 2f0ee4d648
4 changed files with 285 additions and 8 deletions
@@ -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({
@@ -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,
},
},
});
});
});
+106 -6
View File
@@ -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(
aggregateState,
streamContext,
usage,
));
const materialized = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage);
if (materialized) {
completedPayload = materialized;
}
}
if (completedPayload) {
@@ -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',