diff --git a/src/server/routes/proxy/chat.stream.test.ts b/src/server/routes/proxy/chat.stream.test.ts index f566a57..ed1e39b 100644 --- a/src/server/routes/proxy/chat.stream.test.ts +++ b/src/server/routes/proxy/chat.stream.test.ts @@ -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 () => { diff --git a/src/server/routes/proxy/chat.ts b/src/server/routes/proxy/chat.ts index b7886ad..60ed69c 100644 --- a/src/server/routes/proxy/chat.ts +++ b/src/server/routes/proxy/chat.ts @@ -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, diff --git a/src/server/routes/proxy/endpointFlow.test.ts b/src/server/routes/proxy/endpointFlow.test.ts index fc621b0..8ae89a3 100644 --- a/src/server/routes/proxy/endpointFlow.test.ts +++ b/src/server/routes/proxy/endpointFlow.test.ts @@ -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' }, diff --git a/src/server/routes/proxy/endpointFlow.ts b/src/server/routes/proxy/endpointFlow.ts index 1ed1bc4..03e5e42 100644 --- a/src/server/routes/proxy/endpointFlow.ts +++ b/src/server/routes/proxy/endpointFlow.ts @@ -63,6 +63,12 @@ type ExecuteEndpointFlowInput = { onDowngrade?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise; }; +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 { 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) diff --git a/src/server/routes/proxy/responses.codex-oauth.test.ts b/src/server/routes/proxy/responses.codex-oauth.test.ts index 483ce51..cc7d66a 100644 --- a/src/server/routes/proxy/responses.codex-oauth.test.ts +++ b/src/server/routes/proxy/responses.codex-oauth.test.ts @@ -16,10 +16,14 @@ const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) => })); const refreshOauthAccessTokenSingleflightMock = vi.fn(); const recordOauthQuotaResetHintMock = vi.fn(); +const insertedProxyLogs: Record[] = []; const dbInsertMock = vi.fn((_arg?: any) => ({ - values: () => ({ - run: () => undefined, - }), + values: (values: Record) => { + 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'); + }); }); diff --git a/src/server/routes/proxy/responses.ts b/src/server/routes/proxy/responses.ts index fe73c68..6edc5ff 100644 --- a/src/server/routes/proxy/responses.ts +++ b/src/server/routes/proxy/responses.ts @@ -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, diff --git a/src/server/transformers/openai/chat/proxyStream.ts b/src/server/transformers/openai/chat/proxyStream.ts index 15af7ad..54161ae 100644 --- a/src/server/transformers/openai/chat/proxyStream.ts +++ b/src/server/transformers/openai/chat/proxyStream.ts @@ -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; + if (record.error && typeof record.error === 'object' && !Array.isArray(record.error)) { + const message = (record.error as Record).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).error; + if (responseError && typeof responseError === 'object' && !Array.isArray(responseError)) { + const message = (responseError as Record).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).type === 'string' + ? String((parsedPayload as Record).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).type === 'string' + ? String((payload as Record).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 { const lifecycle = createProxyStreamLifecycle({ reader, response, @@ -166,6 +222,7 @@ export function createChatProxyStreamSession(input: ChatProxyStreamSessionInput) onEof: finalize, }); await lifecycle.run(); + return terminalResult; }, }; } diff --git a/src/server/transformers/openai/responses/proxyStream.ts b/src/server/transformers/openai/responses/proxyStream.ts index 14ebc71..7fbc9e3 100644 --- a/src/server/transformers/openai/responses/proxyStream.ts +++ b/src/server/transformers/openai/responses/proxyStream.ts @@ -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 { 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 { const lifecycle = createProxyStreamLifecycle({ reader, response, pullEvents: (buffer) => openAiResponsesStream.pullSseEvents(buffer), handleEvent: handleEventBlock, - onEof: finalize, + onEof: closeOut, }); await lifecycle.run(); + return terminalResult; }, }; }