Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
46 changes: 45 additions & 1 deletion src/server/__tests__/proxy-streaming.test.ts
Original file line number Diff line number Diff line change
@@ -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<Uint8Array> {
const encoder = new TextEncoder()
Expand Down Expand Up @@ -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'],
Expand Down
14 changes: 9 additions & 5 deletions src/server/proxy/streaming/openaiResponsesStreamToAnthropic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,14 @@ type StreamState = {
messageStopped: boolean
}

export function upstreamErrorMessage(data: Record<string, unknown>): string {
const response = data.response as Record<string, unknown> | undefined
const error = (response?.error ?? data.error ?? data) as Record<string, unknown> | 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`
}
Expand Down Expand Up @@ -298,11 +306,7 @@ function processEvent(

case 'response.failed':
case 'error': {
const response = data.response as Record<string, unknown> | undefined
const error = (response?.error ?? data.error ?? data) as Record<string, unknown> | 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',
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { openaiResponsesToAnthropic } from '../transform/openaiResponsesToAnthropic.js'
import { upstreamErrorMessage } from './openaiResponsesStreamToAnthropic.js'
import type {
AnthropicResponse,
OpenAIResponsesResponse,
Expand All @@ -13,6 +14,8 @@ type StreamFallbackState = {
usage: OpenAIResponsesResponse['usage']
}

export class OpenAIResponsesStreamError extends Error {}

export async function openaiResponsesStreamToAnthropicResponse(
upstream: ReadableStream<Uint8Array>,
model: string,
Expand All @@ -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()}`,
Expand Down Expand Up @@ -71,13 +75,19 @@ export async function openaiResponsesStreamToAnthropicResponse(
if (isOpenAIResponsesResponse(response)) {
completedResponse = response
}
} else if (currentEvent === 'response.failed' || currentEvent === 'error') {
upstreamError = upstreamErrorMessage(data)
} else {
updateFallbackState(currentEvent, data, fallback)
}
currentEvent = ''
}
}

if (!completedResponse && upstreamError !== null) {
throw new OpenAIResponsesStreamError(upstreamError)
}

return openaiResponsesToAnthropic(
completedResponse ?? buildFallbackResponse(fallback),
model,
Expand Down
27 changes: 27 additions & 0 deletions src/services/openaiAuth/fetch.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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' },
})
})
})
26 changes: 20 additions & 6 deletions src/services/openaiAuth/fetch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'

Expand Down Expand Up @@ -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()
Expand Down
Loading