diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d0c691e..d26b8c24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,7 @@ All notable changes to DreamCoder will be documented in this file. ### Fixed +- **Proxy** (#148): when the ChatGPT sign-in path streams a non-streaming request and the upstream ends with `response.failed` or `error`, the caller now gets an `api_error` (HTTP 502) with the upstream message, instead of an empty `end_turn` reply. Thanks @pcbeingused333. - **Sidebar re-renders** (#157, #158): the sidebar no longer re-renders on streamed text updates. Running indicators still follow session and tab status. Thanks @tedy69. - **Re-renders while a reply runs** (#156, #165): the session page no longer re-renders every second during a reply. Only the elapsed time updates, and it now counts from when the server starts the turn. Thanks @princecherry9905. - New sessions start on the active provider's main model. Before, activating a provider saved its main model in a hidden setting, and that saved value kept overriding the provider after its main model was edited. diff --git a/src/server/__tests__/proxy-streaming.test.ts b/src/server/__tests__/proxy-streaming.test.ts index 29c77c91..cf74b0a4 100644 --- a/src/server/__tests__/proxy-streaming.test.ts +++ b/src/server/__tests__/proxy-streaming.test.ts @@ -1,7 +1,10 @@ import { describe, test, expect } from 'bun:test' import { openaiChatStreamToAnthropic } from '../proxy/streaming/openaiChatStreamToAnthropic.js' import { openaiResponsesStreamToAnthropic } from '../proxy/streaming/openaiResponsesStreamToAnthropic.js' -import { openaiResponsesStreamToAnthropicResponse } from '../proxy/streaming/openaiResponsesStreamToAnthropicResponse.js' +import { + OpenAIResponsesStreamError, + openaiResponsesStreamToAnthropicResponse, +} from '../proxy/streaming/openaiResponsesStreamToAnthropicResponse.js' function makeStream(chunks: string[]): ReadableStream { const encoder = new TextEncoder() @@ -566,6 +569,47 @@ describe('openaiResponsesStreamToAnthropicResponse', () => { }) }) +describe('openaiResponsesStreamToAnthropicResponse upstream errors', () => { + test.each([ + [ + 'response.failed', + { type: 'response.failed', response: { id: 'resp_1', status: 'failed', error: { message: 'model overloaded' } } }, + 'model overloaded', + ], + ['error', { type: 'error', error: { message: 'rate limited' } }, 'rate limited'], + ['error', { type: 'error' }, 'Upstream Responses stream failed'], + ])('rejects on a %s event instead of returning an empty reply', async (event, payload, message) => { + const result = openaiResponsesStreamToAnthropicResponse( + makeStream([ + 'event: response.created\ndata: {"type":"response.created","id":"resp_1","status":"in_progress"}\n\n', + `event: ${event}\ndata: ${JSON.stringify(payload)}\n\n`, + ]), + 'm', + ) + + await expect(result).rejects.toBeInstanceOf(OpenAIResponsesStreamError) + await expect(result).rejects.toThrow(message) + }) + + test('keeps a completed response that follows an earlier error event', async () => { + const response = { + id: 'resp_1', + model: 'm', + status: 'completed', + output: [{ type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'ok' }] }], + } + const result = await openaiResponsesStreamToAnthropicResponse( + makeStream([ + 'event: error\ndata: {"type":"error","error":{"message":"transient"}}\n\n', + `event: response.completed\ndata: ${JSON.stringify({ type: 'response.completed', response })}\n\n`, + ]), + 'm', + ) + + expect(result.content).toEqual([{ type: 'text', text: 'ok' }]) + }) +}) + describe('stream cancellation', () => { test.each([ ['openaiChatStreamToAnthropic', openaiChatStreamToAnthropic, 'data: {"id":"c","object":"chat.completion.chunk","created":0,"model":"m","choices":[{"index":0,"delta":{"content":"x"},"finish_reason":null}]}\n\n'], diff --git a/src/server/proxy/streaming/openaiResponsesStreamToAnthropic.ts b/src/server/proxy/streaming/openaiResponsesStreamToAnthropic.ts index 57f7798b..f869f7c8 100644 --- a/src/server/proxy/streaming/openaiResponsesStreamToAnthropic.ts +++ b/src/server/proxy/streaming/openaiResponsesStreamToAnthropic.ts @@ -12,6 +12,14 @@ type StreamState = { messageStopped: boolean } +export function upstreamErrorMessage(data: Record): string { + const response = data.response as Record | undefined + const error = (response?.error ?? data.error ?? data) as Record | undefined + return typeof error?.message === 'string' && error.message + ? error.message + : 'Upstream Responses stream failed' +} + function formatSse(event: string, data: unknown): string { return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n` } @@ -298,11 +306,7 @@ function processEvent( case 'response.failed': case 'error': { - const response = data.response as Record | undefined - const error = (response?.error ?? data.error ?? data) as Record | undefined - const message = typeof error?.message === 'string' && error.message - ? error.message - : 'Upstream Responses stream failed' + const message = upstreamErrorMessage(data) state.messageStopped = true controller.enqueue(encoder.encode(formatSse('error', { type: 'error', diff --git a/src/server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.ts b/src/server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.ts index 42c43ec7..9ecd8b88 100644 --- a/src/server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.ts +++ b/src/server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.ts @@ -1,4 +1,5 @@ import { openaiResponsesToAnthropic } from '../transform/openaiResponsesToAnthropic.js' +import { upstreamErrorMessage } from './openaiResponsesStreamToAnthropic.js' import type { AnthropicResponse, OpenAIResponsesResponse, @@ -13,6 +14,8 @@ type StreamFallbackState = { usage: OpenAIResponsesResponse['usage'] } +export class OpenAIResponsesStreamError extends Error {} + export async function openaiResponsesStreamToAnthropicResponse( upstream: ReadableStream, model: string, @@ -22,6 +25,7 @@ export async function openaiResponsesStreamToAnthropicResponse( let buffer = '' let currentEvent = '' let completedResponse: OpenAIResponsesResponse | null = null + let upstreamError: string | null = null const fallback: StreamFallbackState = { id: `resp_${Date.now()}`, @@ -71,6 +75,8 @@ export async function openaiResponsesStreamToAnthropicResponse( if (isOpenAIResponsesResponse(response)) { completedResponse = response } + } else if (currentEvent === 'response.failed' || currentEvent === 'error') { + upstreamError = upstreamErrorMessage(data) } else { updateFallbackState(currentEvent, data, fallback) } @@ -78,6 +84,10 @@ export async function openaiResponsesStreamToAnthropicResponse( } } + if (!completedResponse && upstreamError !== null) { + throw new OpenAIResponsesStreamError(upstreamError) + } + return openaiResponsesToAnthropic( completedResponse ?? buildFallbackResponse(fallback), model, diff --git a/src/services/openaiAuth/fetch.test.ts b/src/services/openaiAuth/fetch.test.ts index ace92bb7..ff28642e 100644 --- a/src/services/openaiAuth/fetch.test.ts +++ b/src/services/openaiAuth/fetch.test.ts @@ -130,4 +130,31 @@ describe('buildOpenAICodexFetch', () => { content: [{ type: 'text', text: 'streamed ok' }], }) }) + + test('returns an error when a streamed Codex response fails for a non-streaming caller', async () => { + const fetchOverride: typeof fetch = async () => + new Response([ + 'event: response.failed', + 'data: {"type":"response.failed","response":{"id":"resp_789","status":"failed","error":{"message":"model overloaded"}}}', + '', + ].join('\n'), { + headers: { 'Content-Type': 'text/event-stream' }, + }) + + const openAIFetch = buildOpenAICodexFetch(fetchOverride, 'test') + const response = await openAIFetch('https://api.anthropic.com/v1/messages', { + method: 'POST', + body: JSON.stringify({ + model: 'gpt-5.5', + max_tokens: 64, + messages: [{ role: 'user', content: 'Say ok' }], + }), + }) + + expect(response.status).toBe(502) + await expect(response.json()).resolves.toEqual({ + type: 'error', + error: { type: 'api_error', message: 'model overloaded' }, + }) + }) }) diff --git a/src/services/openaiAuth/fetch.ts b/src/services/openaiAuth/fetch.ts index 35f44a27..81be5320 100644 --- a/src/services/openaiAuth/fetch.ts +++ b/src/services/openaiAuth/fetch.ts @@ -7,7 +7,10 @@ import { getOpenAIOAuthTokens } from './storage.js' import { anthropicToOpenaiResponses } from '../../server/proxy/transform/anthropicToOpenaiResponses.js' import { openaiResponsesToAnthropic } from '../../server/proxy/transform/openaiResponsesToAnthropic.js' import { openaiResponsesStreamToAnthropic } from '../../server/proxy/streaming/openaiResponsesStreamToAnthropic.js' -import { openaiResponsesStreamToAnthropicResponse } from '../../server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.js' +import { + OpenAIResponsesStreamError, + openaiResponsesStreamToAnthropicResponse, +} from '../../server/proxy/streaming/openaiResponsesStreamToAnthropicResponse.js' import type { AnthropicRequest } from '../../server/proxy/transform/types.js' import { logForDebugging } from '../../utils/debug.js' @@ -109,11 +112,22 @@ export function buildOpenAICodexFetch( } if (upstream.body && isEventStreamResponse(upstream)) { - const responseBody = await openaiResponsesStreamToAnthropicResponse( - upstream.body, - mappedModel, - ) - return Response.json(responseBody) + try { + const responseBody = await openaiResponsesStreamToAnthropicResponse( + upstream.body, + mappedModel, + ) + return Response.json(responseBody) + } catch (error) { + if (!(error instanceof OpenAIResponsesStreamError)) throw error + return Response.json( + { + type: 'error', + error: { type: 'api_error', message: error.message }, + }, + { status: 502 }, + ) + } } const responseBody = await upstream.json()