1
0
forked from boos/BoosAPI

fix responses eof false success logging

This commit is contained in:
cita
2026-03-19 20:30:01 +08:00
parent 36e9a85664
commit 8f0b5c81f4
3 changed files with 64 additions and 10 deletions
@@ -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({
@@ -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');
});
});
@@ -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;
}