aibo/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests Activity

fix responses stream failure logging

Browse Source
This commit is contained in:
cita
2026-03-18 18:17:06 +08:00
parent 38e46e165b
commit 332007d4fb
8 changed files with 248 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/proxy/chat.stream.test.ts
+2
View File
@@ -3193,6 +3193,8 @@ describe('chat proxy stream behavior', () => {
expect(response.statusCode).toBe(200);
expect(response.body).toContain('"finish_reason":"error"');
expect(response.body).toContain('[DONE]');
expect(recordSuccessMock).not.toHaveBeenCalled();
expect(recordFailureMock).toHaveBeenCalledTimes(1);
});
it('preserves non-stream function_call output when /v1/chat/completions falls back to /v1/responses', async () => {
src/server/routes/proxy/chat.ts
+49 -2
View File
@@ -366,6 +366,7 @@ async function handleChatProxyRequest(
const streamSession = openAiChatTransformer.proxyStream.createSession({
downstreamFormat,
modelName,
successfulUpstreamPath,
onParsedPayload: (payload) => {
if (payload && typeof payload === 'object') {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(payload));
@@ -389,9 +390,32 @@ async function handleChatProxyRequest(
if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
fallbackData = unwrapGeminiCliPayload(fallbackData);
}
streamSession.consumeUpstreamFinalPayload(fallbackData, fallbackText, reply.raw);
const streamResult = streamSession.consumeUpstreamFinalPayload(fallbackData, fallbackText, reply.raw);
const latency = Date.now() - startTime;
if (streamResult.status === 'failed') {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
200,
latency,
streamResult.errorMessage,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return;
}
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
@@ -443,9 +467,32 @@ async function handleChatProxyRequest(
const reader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
? createGeminiCliStreamReader(upstreamReader)
: upstreamReader;
await streamSession.run(reader, reply.raw);
const streamResult = await streamSession.run(reader, reply.raw);
const latency = Date.now() - startTime;
if (streamResult.status === 'failed') {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
200,
latency,
streamResult.errorMessage,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return;
}
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
src/server/routes/proxy/endpointFlow.test.ts
+16
View File
@@ -185,6 +185,22 @@ describe('executeEndpointFlow', () => {
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it('uses proxyUrl for the default fetch path when no dispatch hook is provided', async () => {
fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { 'content-type': 'application/json' },
})));
await executeEndpointFlow({
siteUrl: 'https://example.com',
proxyUrl: 'https://proxy.internal/base',
endpointCandidates: ['responses'],
buildRequest: () => requestFor('/v1/responses'),
});
expect(fetchMock.mock.calls[0]?.[0]).toBe('https://proxy.internal/base/v1/responses');
});
it('returns normalized final error when all endpoints fail', async () => {
fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({
error: { message: 'upstream_error', type: 'upstream_error' },
src/server/routes/proxy/endpointFlow.ts
+10 -1
View File
@@ -63,6 +63,12 @@ type ExecuteEndpointFlowInput = {
onDowngrade?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise<void>;
};
function buildAbsoluteUrl(base: string, path: string): string {
const normalizedBase = base.replace(/\/+$/, '');
const normalizedPath = path.replace(/^\/+/, '');
return normalizedPath ? `${normalizedBase}/${normalizedPath}` : normalizedBase;
}
export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Promise<EndpointFlowResult> {
const endpointCount = input.endpointCandidates.length;
if (endpointCount <= 0) {
@@ -80,7 +86,10 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
for (let endpointIndex = 0; endpointIndex < endpointCount; endpointIndex += 1) {
const endpoint = input.endpointCandidates[endpointIndex] as UpstreamEndpoint;
const request = input.buildRequest(endpoint, endpointIndex);
const targetUrl = buildUpstreamUrl(input.siteUrl, request.path);
const defaultTarget = buildUpstreamUrl(input.siteUrl, request.path);
const targetUrl = input.proxyUrl
? buildAbsoluteUrl(input.proxyUrl, request.path)
: defaultTarget;
let response = input.dispatchRequest
? await input.dispatchRequest(request, targetUrl)
src/server/routes/proxy/responses.codex-oauth.test.ts
+39 -3
View File
@@ -16,10 +16,14 @@ const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) =>
}));
const refreshOauthAccessTokenSingleflightMock = vi.fn();
const recordOauthQuotaResetHintMock = vi.fn();
const insertedProxyLogs: Record<string, unknown>[] = [];
const dbInsertMock = vi.fn((_arg?: any) => ({
values: () => ({
run: () => undefined,
}),
values: (values: Record<string, unknown>) => {
insertedProxyLogs.push(values);
return {
run: () => undefined,
};
},
}));
vi.mock('undici', () => ({
@@ -117,6 +121,7 @@ describe('responses proxy codex oauth refresh', () => {
refreshOauthAccessTokenSingleflightMock.mockReset();
recordOauthQuotaResetHintMock.mockReset();
dbInsertMock.mockClear();
insertedProxyLogs.length = 0;
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
@@ -431,4 +436,35 @@ describe('responses proxy codex oauth refresh', () => {
expect(secondBody.store).toBe(false);
expect(secondBody.stream).toBe(true);
});
it('does not record success when a streaming responses request ends with response.failed', async () => {
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_codex_failed","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.failed\n',
'data: {"type":"response.failed","response":{"id":"resp_codex_failed","model":"gpt-5.4","status":"failed","error":{"message":"tool execution failed"}}}\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('response.failed');
expect(response.body).not.toContain('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('tool execution failed');
});
});
src/server/routes/proxy/responses.ts
+24 -1
View File
@@ -457,9 +457,32 @@ export async function responsesProxyRoute(app: FastifyInstance) {
reply.raw.write(chunk);
},
});
await streamSession.run(reader, reply.raw);
const streamResult = await streamSession.run(reader, reply.raw);
const latency = Date.now() - startTime;
if (streamResult.status === 'failed') {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
200,
latency,
streamResult.errorMessage,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return;
}
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
src/server/transformers/openai/chat/proxyStream.ts
+60 -3
View File
@@ -14,6 +14,7 @@ type StreamReader = {
type ChatProxyStreamSessionInput = {
downstreamFormat: DownstreamFormat;
modelName: string;
successfulUpstreamPath: string;
onParsedPayload?: (payload: unknown) => void;
writeLines: (lines: string[]) => void;
writeRaw: (chunk: string) => void;
@@ -23,6 +24,11 @@ type ResponseSink = {
end(): void;
};
type ChatProxyStreamResult = {
status: 'completed' | 'failed';
errorMessage: string | null;
};
export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput) {
const downstreamTransformer = input.downstreamFormat === 'claude'
? anthropicMessagesTransformer
@@ -39,6 +45,36 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
? createOpenAiChatAggregateState()
: null;
let finalized = false;
let terminalResult: ChatProxyStreamResult = {
status: 'completed',
errorMessage: null,
};
const extractFailureMessage = (payload: unknown, fallback = 'upstream stream failed'): string => {
if (payload && typeof payload === 'object' && !Array.isArray(payload)) {
const record = payload as Record<string, unknown>;
if (record.error && typeof record.error === 'object' && !Array.isArray(record.error)) {
const message = (record.error as Record<string, unknown>).message;
if (typeof message === 'string' && message.trim()) return message.trim();
}
if (typeof record.message === 'string' && record.message.trim()) return record.message.trim();
if (record.response && typeof record.response === 'object' && !Array.isArray(record.response)) {
const responseError = (record.response as Record<string, unknown>).error;
if (responseError && typeof responseError === 'object' && !Array.isArray(responseError)) {
const message = (responseError as Record<string, unknown>).message;
if (typeof message === 'string' && message.trim()) return message.trim();
}
}
}
return fallback;
};
const markFailed = (payload: unknown, fallbackMessage?: string) => {
terminalResult = {
status: 'failed',
errorMessage: extractFailureMessage(payload, fallbackMessage),
};
};
const finalize = () => {
if (finalized) return;
@@ -51,7 +87,12 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
return;
}
if (input.downstreamFormat === 'openai' && chatAggregateState && chatAggregateState.choices.size > 0) {
if (
input.downstreamFormat === 'openai'
&& terminalResult.status !== 'failed'
&& chatAggregateState
&& chatAggregateState.choices.size > 0
) {
const needsTerminalFinishChunk = Array.from(chatAggregateState.choices.values())
.some((choice) => !choice.finishReason);
if (needsTerminalFinishChunk) {
@@ -109,6 +150,12 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
}
if (parsedPayload && typeof parsedPayload === 'object') {
const payloadType = typeof (parsedPayload as Record<string, unknown>).type === 'string'
? String((parsedPayload as Record<string, unknown>).type)
: '';
if (payloadType === 'response.failed' || payloadType === 'error') {
markFailed(parsedPayload);
}
const normalizedEvent = downstreamTransformer.transformStreamEvent(parsedPayload, streamContext, input.modelName);
if (input.downstreamFormat === 'openai' && chatAggregateState) {
applyOpenAiChatStreamEvent(chatAggregateState, normalizedEvent);
@@ -129,10 +176,18 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
};
return {
consumeUpstreamFinalPayload(payload: unknown, fallbackText: string, response?: ResponseSink) {
consumeUpstreamFinalPayload(payload: unknown, fallbackText: string, response?: ResponseSink): ChatProxyStreamResult {
if (payload && typeof payload === 'object') {
input.onParsedPayload?.(payload);
}
if (payload && typeof payload === 'object' && !Array.isArray(payload)) {
const payloadType = typeof (payload as Record<string, unknown>).type === 'string'
? String((payload as Record<string, unknown>).type)
: '';
if (payloadType === 'response.failed' || payloadType === 'error') {
markFailed(payload);
}
}
if (input.downstreamFormat === 'openai') {
const normalizedFinal = openAiChatOutbound.normalizeFinal(payload, input.modelName, fallbackText);
streamContext.id = normalizedFinal.id;
@@ -156,8 +211,9 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
}
finalize();
response?.end();
return terminalResult;
},
async run(reader: StreamReader | null | undefined, response: ResponseSink) {
async run(reader: StreamReader | null | undefined, response: ResponseSink): Promise<ChatProxyStreamResult> {
const lifecycle = createProxyStreamLifecycle<ParsedSseEvent>({
reader,
response,
@@ -166,6 +222,7 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput)
onEof: finalize,
});
await lifecycle.run();
return terminalResult;
},
};
}
src/server/transformers/openai/responses/proxyStream.ts
+48 -5
View File
@@ -13,6 +13,11 @@ type ResponseSink = {
end(): void;
};
type ResponsesProxyStreamResult = {
status: 'completed' | 'failed';
errorMessage: string | null;
};
type ResponsesProxyStreamSessionInput = {
modelName: string;
successfulUpstreamPath: string;
@@ -33,20 +38,58 @@ function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
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()) {
return payload.error.message.trim();
}
if (typeof payload.message === 'string' && payload.message.trim()) {
return payload.message.trim();
}
if (isRecord(payload.response) && isRecord(payload.response.error) && typeof payload.response.error.message === 'string' && payload.response.error.message.trim()) {
return payload.response.error.message.trim();
}
}
return fallback;
}
export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSessionInput) {
const streamContext = openAiResponsesStream.createContext(input.modelName);
const responsesState = createOpenAiResponsesAggregateState(input.modelName);
let finalized = false;
let terminalResult: ResponsesProxyStreamResult = {
status: 'completed',
errorMessage: null,
};
const finalize = () => {
if (finalized) return;
finalized = true;
terminalResult = {
status: 'completed',
errorMessage: null,
};
input.writeLines(completeResponsesStream(responsesState, streamContext, input.getUsage()));
};
const fail = (payload: unknown, fallbackMessage?: string) => {
if (finalized) return;
finalized = true;
terminalResult = {
status: 'failed',
errorMessage: getResponsesStreamFailureMessage(payload, fallbackMessage),
};
input.writeLines(failResponsesStream(responsesState, streamContext, input.getUsage(), payload));
};
const closeOut = () => {
if (finalized) return;
finalize();
};
const handleEventBlock = (eventBlock: ParsedSseEvent): boolean => {
if (eventBlock.data === '[DONE]') {
finalize();
closeOut();
return true;
}
@@ -71,8 +114,7 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
|| payloadType === 'response.failed'
);
if (isFailureEvent) {
input.writeLines(failResponsesStream(responsesState, streamContext, input.getUsage(), parsedPayload));
finalized = true;
fail(parsedPayload);
return true;
}
@@ -97,15 +139,16 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
};
return {
async run(reader: StreamReader | null | undefined, response: ResponseSink) {
async run(reader: StreamReader | null | undefined, response: ResponseSink): Promise<ResponsesProxyStreamResult> {
const lifecycle = createProxyStreamLifecycle<ParsedSseEvent>({
reader,
response,
pullEvents: (buffer) => openAiResponsesStream.pullSseEvents(buffer),
handleEvent: handleEventBlock,
onEof: finalize,
onEof: closeOut,
});
await lifecycle.run();
return terminalResult;
},
};
}