From 9a32f3949b327b409c168f5654ae41b437676d10 Mon Sep 17 00:00:00 2001 From: orangeboyChen Date: Thu, 20 Aug 2026 17:47:11 +0800 Subject: [PATCH 1/3] feat: support responses upstream protocol --- AGENTS.md | 10 + app/credentials/credential-card.tsx | 42 +- app/credentials/credential-group.tsx | 8 +- app/credentials/credentials.tsx | 61 +- app/page-shell.tsx | 14 +- bun.lock | 1 + lib/server/domain/credentials.ts | 33 +- lib/server/proxy/anthropic.ts | 73 +- lib/server/proxy/codebuddy.ts | 1176 +++++++++++++++++++++++++- lib/server/proxy/responses.ts | 54 +- messages/en-US.json | 9 +- messages/ja-JP.json | 9 +- messages/zh-CN.json | 9 +- package.json | 1 + tests/admin/initial-state.test.ts | 2 + tests/server/anthropic.test.ts | 154 ++++ tests/server/units.test.ts | 730 +++++++++++++++- 17 files changed, 2297 insertions(+), 89 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0519126..204cdc1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,3 +32,13 @@ - Run `bun run test:patch-branches --base "$(git merge-base HEAD origin/main)"` after coverage; changed branch coverage must be at least 90%. - Confirm Codecov's `patch` status is successful and meets the target configured in `codecov.yml`; do not lower the patch target or threshold to bypass a coverage failure. - Use the Codecov PR report as the source of truth for patch coverage because it compares the uploaded `lcov.info` against the PR base commit. + + + +# This is NOT the Next.js you know + +This version has breaking changes — APIs, conventions, and file structure may all differ from your training data. Read the relevant guide in `node_modules/next/dist/docs/` (resolved from this file's directory; in monorepos the `next` package may not be visible from the repo root) before writing any code. Heed deprecation notices. + +This block is written and re-added by `next dev` — verify at `node_modules/next/dist/server/lib/generate-agent-files.js`. Removing it from a diff only re-creates the uncommitted change; committing it with your work keeps the tree clean. + + diff --git a/app/credentials/credential-card.tsx b/app/credentials/credential-card.tsx index fde8373..7966011 100644 --- a/app/credentials/credential-card.tsx +++ b/app/credentials/credential-card.tsx @@ -1,5 +1,5 @@ import { Avatar, Block, Flexbox, Tag } from '@lobehub/ui'; -import { Button } from '@lobehub/ui/base-ui'; +import { Button, Select } from '@lobehub/ui/base-ui'; import { CalendarDays, Clock3, @@ -24,7 +24,7 @@ interface CredentialCardProps { current: CurrentCredentialInfo | null; form: CredentialFormState; onCredentialFirstMessageRoleToSystemChange: (value: boolean) => void; - onCredentialResponsesPassthroughChange: (value: boolean) => void; + onCredentialUpstreamProtocolChange: (value: 'chat' | 'responses') => void; onDelete: () => void; onEdit: () => void; onResetCredentialForm: () => void; @@ -36,7 +36,7 @@ export const CredentialCard = ({ current, form, onCredentialFirstMessageRoleToSystemChange, - onCredentialResponsesPassthroughChange, + onCredentialUpstreamProtocolChange, onDelete, onEdit, onResetCredentialForm, @@ -104,9 +104,9 @@ export const CredentialCard = ({
- {credential.responses_passthrough - ? text('credentials.credentialResponsesDirect') - : text('credentials.credentialResponsesProxyTag')} + {credential.upstream_protocol === 'responses' + ? text('credentials.credentialUpstreamResponsesTag') + : text('credentials.credentialUpstreamChatTag')} {credential.first_message_role_to_system @@ -130,12 +130,30 @@ export const CredentialCard = ({ {text('credentials.credentialEditTitle')}
- + { onCredentialFirstMessageRoleToSystemChange={ onCredentialFirstMessageRoleToSystemChange } - onCredentialResponsesPassthroughChange={ - onCredentialResponsesPassthroughChange + onCredentialUpstreamProtocolChange={ + onCredentialUpstreamProtocolChange } onDelete={onDeleteCredential} onEdit={onEditCredential} @@ -608,8 +635,8 @@ const Credentials = () => { onCredentialFirstMessageRoleToSystemChange={ onCredentialFirstMessageRoleToSystemChange } - onCredentialResponsesPassthroughChange={ - onCredentialResponsesPassthroughChange + onCredentialUpstreamProtocolChange={ + onCredentialUpstreamProtocolChange } onDelete={onDeleteCredential} onEdit={onEditCredential} diff --git a/app/page-shell.tsx b/app/page-shell.tsx index 1b8eab1..cb8e454 100644 --- a/app/page-shell.tsx +++ b/app/page-shell.tsx @@ -984,14 +984,14 @@ const AdminPageLayoutContent = ({ index: credentials.form.editingIndex, first_message_role_to_system: credentials.form.firstMessageRoleToSystem, - responses_passthrough: credentials.form.responsesPassthrough, + upstream_protocol: credentials.form.upstreamProtocol, } : { access_token: credentials.form.bearerToken.trim(), bearer_token: credentials.form.bearerToken.trim(), first_message_role_to_system: credentials.form.firstMessageRoleToSystem, - responses_passthrough: credentials.form.responsesPassthrough, + upstream_protocol: credentials.form.upstreamProtocol, user_id: credentials.form.userId.trim() || undefined, }, ), @@ -1021,7 +1021,7 @@ const AdminPageLayoutContent = ({ bearerToken: '', editingIndex: null, firstMessageRoleToSystem: false, - responsesPassthrough: false, + upstreamProtocol: 'chat', userId: '', }, })); @@ -1736,12 +1736,12 @@ const AdminPageLayoutContent = ({ }, })); }, - onCredentialResponsesPassthroughChange: (value) => { + onCredentialUpstreamProtocolChange: (value) => { setCredentials((current) => ({ ...current, form: { ...current.form, - responsesPassthrough: value, + upstreamProtocol: value, }, })); }, @@ -1774,7 +1774,7 @@ const AdminPageLayoutContent = ({ editingIndex: credential.index, firstMessageRoleToSystem: credential.first_message_role_to_system, - responsesPassthrough: credential.responses_passthrough, + upstreamProtocol: credential.upstream_protocol, userId: credential.user_id ?? '', }, })); @@ -1827,7 +1827,7 @@ const AdminPageLayoutContent = ({ bearerToken: '', editingIndex: null, firstMessageRoleToSystem: false, - responsesPassthrough: false, + upstreamProtocol: 'chat', userId: '', }, })); diff --git a/bun.lock b/bun.lock index c046030..f88f904 100644 --- a/bun.lock +++ b/bun.lock @@ -31,6 +31,7 @@ "@eslint/compat": "^2.1.0", "@next/eslint-plugin-next": "^16.3.0", "@tailwindcss/postcss": "^4.3.3", + "@testing-library/dom": "^10.4.1", "@testing-library/jest-dom": "^7.0.1", "@testing-library/react": "^16.3.0", "@types/better-sqlite3": "^9.6.0", diff --git a/lib/server/domain/credentials.ts b/lib/server/domain/credentials.ts index 2192fe4..1179380 100644 --- a/lib/server/domain/credentials.ts +++ b/lib/server/domain/credentials.ts @@ -30,6 +30,7 @@ export type CredentialData = Record & { user_id?: string; user_info?: Record; responses_passthrough?: boolean; + upstream_protocol?: 'chat' | 'responses'; first_message_role_to_system?: boolean; supported_models?: string; }; @@ -473,6 +474,8 @@ export const listCredentials = async (): Promise<{ responses_passthrough: getBooleanSetting( record.data.responses_passthrough, ), + upstream_protocol: getCredentialProxySettings(record.data) + .upstreamProtocol, first_message_role_to_system: getBooleanSetting( record.data.first_message_role_to_system, ), @@ -559,6 +562,21 @@ export const addCredential = async ( existingPayload = storedPayload; } + const upstreamProtocol = + credentialData.upstream_protocol === 'responses' + ? 'responses' + : credentialData.upstream_protocol === 'chat' + ? 'chat' + : credentialData.responses_passthrough !== undefined + ? getBooleanSetting(credentialData.responses_passthrough) + ? 'responses' + : 'chat' + : existingPayload.upstream_protocol === 'responses' + ? 'responses' + : getBooleanSetting(existingPayload.responses_passthrough) + ? 'responses' + : 'chat'; + const payload = { ...existingPayload, ...credentialData, @@ -566,10 +584,8 @@ export const addCredential = async ( typeof existingPayload.created_at === 'number' ? existingPayload.created_at : now, - responses_passthrough: getBooleanSetting( - credentialData.responses_passthrough ?? - existingPayload.responses_passthrough, - ), + responses_passthrough: upstreamProtocol === 'responses', + upstream_protocol: upstreamProtocol, first_message_role_to_system: getBooleanSetting( credentialData.first_message_role_to_system ?? existingPayload.first_message_role_to_system, @@ -779,12 +795,17 @@ export const getCredentialProxySettings = ( credential: CredentialData | null | undefined, ): { firstMessageRoleToSystem: boolean; - responsesPassthrough: boolean; + upstreamProtocol: 'chat' | 'responses'; } => { return { firstMessageRoleToSystem: getBooleanSetting( credential?.first_message_role_to_system, ), - responsesPassthrough: getBooleanSetting(credential?.responses_passthrough), + upstreamProtocol: + credential?.upstream_protocol === 'responses' || + (credential?.upstream_protocol !== 'chat' && + getBooleanSetting(credential?.responses_passthrough)) + ? 'responses' + : 'chat', }; }; diff --git a/lib/server/proxy/anthropic.ts b/lib/server/proxy/anthropic.ts index bea0c15..be28486 100644 --- a/lib/server/proxy/anthropic.ts +++ b/lib/server/proxy/anthropic.ts @@ -392,6 +392,11 @@ const buildChatRequestBody = async ( const chatMessages = mapAnthropicMessagesToChat(body.messages ?? []); const messages: ChatMessage[] = []; + const disableParallelToolUse = + body.tool_choice && typeof body.tool_choice === 'object' + ? (body.tool_choice as { disable_parallel_tool_use?: unknown }) + .disable_parallel_tool_use + : undefined; if (systemText) { messages.push({ role: 'system', content: systemText }); @@ -412,6 +417,10 @@ const buildChatRequestBody = async ( stop: body.stop_sequences, tools: mapAnthropicToolsToChat(body.tools), tool_choice: mapAnthropicToolChoiceToChat(body.tool_choice), + parallel_tool_calls: + typeof disableParallelToolUse === 'boolean' + ? !disableParallelToolUse + : undefined, }; // Pass through thinking/reasoning config so upstream models that support @@ -832,13 +841,15 @@ const mapOpenAIStreamToAnthropicSSE = ( const upstreamReader = upstreamResponse.body!.getReader(); reader = upstreamReader; let buffer = ''; - const rejectStream = (): void => { + const rejectStream = ( + message = 'Upstream SSE frame exceeds the maximum size', + ): void => { streamRejected = true; enqueueEvent({ type: 'error', error: { type: 'invalid_request_error', - message: 'Upstream SSE frame exceeds the maximum size', + message, }, }); }; @@ -867,7 +878,7 @@ const mapOpenAIStreamToAnthropicSSE = ( const chunk = JSON.parse(raw) as OpenAIStreamChunk; const upstreamError = chunk as OpenAIStreamError; if (upstreamError.error?.message) { - rejectStream(); + rejectStream(upstreamError.error.message); return; } processChunk(chunk); @@ -939,6 +950,38 @@ const mapOpenAIStreamToAnthropicSSE = ( }); }; +const extractErrorMessage = (value: unknown): string | null => { + if (typeof value === 'string') { + try { + return extractErrorMessage(JSON.parse(value) as unknown) ?? value; + } catch { + return value; + } + } + if (!value || typeof value !== 'object') return null; + + const payload = value as { + detail?: unknown; + error?: unknown; + message?: unknown; + }; + const detail = extractErrorMessage(payload.detail); + if (detail) return detail; + if (typeof payload.message === 'string') return payload.message; + return extractErrorMessage(payload.error); +}; + +const getUpstreamErrorMessage = async (response: Response): Promise => { + const text = await response.text(); + if (!text) return 'Upstream CodeBuddy request failed'; + + try { + return extractErrorMessage(JSON.parse(text) as unknown) ?? text; + } catch { + return text; + } +}; + // --------------------------------------------------------------------------- // Main handler // --------------------------------------------------------------------------- @@ -963,7 +1006,10 @@ export const handleMessagesRequest = async ( ); if (!upstreamResponse.ok) { - return upstreamResponse; + return createAnthropicError( + upstreamResponse.status, + await getUpstreamErrorMessage(upstreamResponse), + ); } const model = String(chatBody.model ?? 'unknown'); @@ -984,11 +1030,28 @@ export const handleMessagesRequest = async ( }; const createAnthropicError = (status: number, message: string): Response => { + const type = + status === 401 + ? 'authentication_error' + : status === 403 + ? 'permission_error' + : status === 404 + ? 'not_found_error' + : status === 413 + ? 'request_too_large' + : status === 429 + ? 'rate_limit_error' + : status === 529 + ? 'overloaded_error' + : status >= 500 + ? 'api_error' + : 'invalid_request_error'; + return Response.json( { type: 'error', error: { - type: 'invalid_request_error', + type, message, }, }, diff --git a/lib/server/proxy/codebuddy.ts b/lib/server/proxy/codebuddy.ts index 460508a..6ef7d0f 100644 --- a/lib/server/proxy/codebuddy.ts +++ b/lib/server/proxy/codebuddy.ts @@ -42,6 +42,9 @@ export interface ChatRequestBody { model?: string; messages?: OpenAIMessage[]; stream?: boolean; + stream_options?: { + include_usage?: boolean; + }; temperature?: number; max_tokens?: number; max_completion_tokens?: number; @@ -52,6 +55,7 @@ export interface ChatRequestBody { stop?: string | string[]; tools?: unknown[]; tool_choice?: unknown; + parallel_tool_calls?: boolean; thinking?: Record; reasoning_effort?: string; } @@ -111,7 +115,7 @@ export interface ProxyContext { credentialFilename: string | null; preferences: { firstMessageRoleToSystem: boolean; - responsesPassthrough: boolean; + upstreamProtocol: 'chat' | 'responses'; }; } @@ -182,6 +186,65 @@ const extractResponsesUsage = (value: unknown): unknown => { return payload.response?.usage ?? payload.usage ?? null; }; +const mapResponsesUsageToChat = ( + usage: unknown, +): Record | null => { + if (!usage || typeof usage !== 'object') return null; + + const value = usage as { + cache_creation_input_tokens?: unknown; + cache_read_input_tokens?: unknown; + input_tokens?: unknown; + input_tokens_details?: { + cache_creation_tokens?: unknown; + cached_tokens?: unknown; + }; + output_tokens?: unknown; + output_tokens_details?: { + reasoning_tokens?: unknown; + }; + total_tokens?: unknown; + }; + const inputTokens = Number(value.input_tokens ?? 0); + const outputTokens = Number(value.output_tokens ?? 0); + const cachedTokens = Number( + value.input_tokens_details?.cached_tokens ?? + value.cache_read_input_tokens ?? + 0, + ); + const cacheCreationTokens = Number( + value.input_tokens_details?.cache_creation_tokens ?? + value.cache_creation_input_tokens ?? + 0, + ); + const reasoningTokens = Number( + value.output_tokens_details?.reasoning_tokens ?? 0, + ); + + return { + completion_tokens: outputTokens, + completion_tokens_details: { + reasoning_tokens: reasoningTokens, + }, + prompt_tokens: inputTokens, + prompt_tokens_details: { + cache_creation_tokens: cacheCreationTokens, + cached_tokens: cachedTokens, + }, + total_tokens: Number(value.total_tokens ?? inputTokens + outputTokens), + }; +}; + +const extractResponsesId = (value: unknown): string | null => { + if (!value || typeof value !== 'object') return null; + const payload = value as { + id?: unknown; + response?: { id?: unknown }; + }; + const id = payload.response?.id ?? payload.id; + return typeof id === 'string' && id ? id : null; +}; + const parseUsageHeader = (response: Response): unknown => { const usageHeader = response.headers.get('x-codebuddy-usage'); @@ -199,11 +262,13 @@ const parseUsageHeader = (response: Response): unknown => { const trackResponsesUsageStream = async ({ fallbackUsage, model, + onResponseId, proxyContext, upstreamResponse, }: { fallbackUsage: unknown; model: string; + onResponseId?: (responseId: string) => Promise; proxyContext: ProxyContext; upstreamResponse: Response; }): Promise => { @@ -225,16 +290,48 @@ const trackResponsesUsageStream = async ({ const encoder = new TextEncoder(); let reader: ReadableStreamDefaultReader | null = null; let cancelled = false; + let latestUsage = fallbackUsage; + let responseBinding: Promise | null = null; + let usageRecorded = false; const releaseReader = (): void => { reader?.releaseLock(); reader = null; }; + const recordStreamUsage = async (): Promise => { + if (usageRecorded) return; + usageRecorded = true; + try { + await recordProxyUsage({ + model, + proxyContext, + route: '/v1/responses', + usage: latestUsage, + }); + } catch (error) { + console.error('[CodeBuddy2API] Failed to record Responses stream usage', { + error, + route: '/v1/responses', + }); + } + }; + const bindResponseId = (id: string): void => { + if (!onResponseId || responseBinding) return; + responseBinding = onResponseId(id).catch((error) => { + console.error( + '[CodeBuddy2API] Failed to bind upstream Responses session', + { + error, + responseId: id, + }, + ); + }); + }; const stream = new ReadableStream({ start: (controller) => { const upstreamReader = upstreamResponse.body!.getReader(); reader = upstreamReader; let buffer = ''; - let latestUsage = fallbackUsage; + let responseId: string | null = null; const inspectFrame = (frame: string): void => { frame.split('\n').forEach((line) => { @@ -249,8 +346,10 @@ const trackResponsesUsageStream = async ({ } try { - latestUsage = - extractResponsesUsage(JSON.parse(raw) as unknown) ?? latestUsage; + const event = JSON.parse(raw) as unknown; + latestUsage = extractResponsesUsage(event) ?? latestUsage; + responseId = extractResponsesId(event) ?? responseId; + if (responseId) bindResponseId(responseId); } catch { // Preserve malformed upstream frames without recording them. } @@ -271,12 +370,8 @@ const trackResponsesUsageStream = async ({ controller.enqueue(encoder.encode(buffer)); } - await recordProxyUsage({ - model, - proxyContext, - route: '/v1/responses', - usage: latestUsage, - }); + await recordStreamUsage(); + await responseBinding; releaseReader(); controller.close(); return; @@ -300,13 +395,25 @@ const trackResponsesUsageStream = async ({ } }; - void pump(); + void pump().catch(async (error) => { + if (cancelled) return; + console.error('[CodeBuddy2API] Responses upstream stream failed', { + error, + route: '/v1/responses', + }); + await responseBinding; + await recordStreamUsage(); + releaseReader(); + controller.error(error); + }); }, async cancel(reason): Promise { cancelled = true; try { await reader?.cancel(reason); } finally { + await responseBinding; + await recordStreamUsage(); releaseReader(); } }, @@ -707,13 +814,931 @@ const buildUpstreamBody = async ( frequency_penalty: body.frequency_penalty, presence_penalty: body.presence_penalty, stop: body.stop, + stream_options: body.stream_options, tools: body.tools, tool_choice: body.tool_choice, + parallel_tool_calls: body.parallel_tool_calls, thinking: body.thinking, reasoning_effort: body.reasoning_effort, }; }; +const stringifyResponsesInputContent = (content: unknown): string => { + if (typeof content === 'string') return content; + if (content === null || content === undefined) return ''; + if (Array.isArray(content)) { + return content + .map((part) => { + if (typeof part === 'string') return part; + if (part && typeof part === 'object' && 'text' in part) { + return String((part as { text?: unknown }).text ?? ''); + } + return JSON.stringify(part); + }) + .join(''); + } + return JSON.stringify(content); +}; + +const mapChatContentToResponses = ( + content: unknown, +): Array> => { + if (!Array.isArray(content)) { + return [ + { + text: stringifyResponsesInputContent(content), + type: 'input_text', + }, + ]; + } + + return content.flatMap((part): Array> => { + if (typeof part === 'string') { + return [{ text: part, type: 'input_text' }]; + } + if (!part || typeof part !== 'object') { + return [{ text: JSON.stringify(part), type: 'input_text' }]; + } + const value = part as { + image_url?: string | { detail?: unknown; url?: unknown }; + text?: unknown; + type?: unknown; + }; + if (value.type === 'image_url') { + const imageUrl = + typeof value.image_url === 'string' + ? value.image_url + : value.image_url?.url; + if (typeof imageUrl === 'string' && imageUrl) { + const detail = + typeof value.image_url === 'object' && + typeof value.image_url.detail === 'string' + ? value.image_url.detail + : undefined; + return [ + { + image_url: imageUrl, + ...(detail ? { detail } : {}), + type: 'input_image', + }, + ]; + } + } + if (value.type === 'input_image' && typeof value.image_url === 'string') { + return [{ image_url: value.image_url, type: 'input_image' }]; + } + if (typeof value.text === 'string') { + return [{ text: value.text, type: 'input_text' }]; + } + return [{ text: JSON.stringify(value), type: 'input_text' }]; + }); +}; + +const translateChatToolChoiceToResponses = (toolChoice: unknown): unknown => { + if (typeof toolChoice === 'string') return toolChoice; + if (!toolChoice || typeof toolChoice !== 'object') return undefined; + const value = toolChoice as { + function?: { name?: unknown }; + name?: unknown; + type?: unknown; + }; + if (value.type !== 'function') return toolChoice; + const name = value.function?.name ?? value.name; + return typeof name === 'string' ? { name, type: 'function' } : toolChoice; +}; + +const translateChatResponseFormatToResponses = ( + responseFormat: unknown, +): Record | undefined => { + if (!responseFormat || typeof responseFormat !== 'object') return undefined; + const value = responseFormat as { + json_schema?: Record; + type?: unknown; + }; + if (value.type === 'json_object') { + return { format: { type: 'json_object' } }; + } + if (value.type !== 'json_schema' || !value.json_schema) return undefined; + const schema = value.json_schema; + if (typeof schema.name !== 'string' || !schema.name) return undefined; + return { + format: { + ...(schema.description ? { description: schema.description } : {}), + name: schema.name, + schema: schema.schema ?? { type: 'object', properties: {} }, + ...(typeof schema.strict === 'boolean' ? { strict: schema.strict } : {}), + type: 'json_schema', + }, + }; +}; + +const translateChatThinkingToResponses = ( + thinking: Record | undefined, + reasoningEffort: string | undefined, +): Record | undefined => { + if (!thinking) + return reasoningEffort ? { effort: reasoningEffort } : undefined; + + if (thinking.type === 'disabled') return { effort: 'none' }; + if (thinking.type !== 'adaptive' && thinking.type !== 'enabled') { + return undefined; + } + + const budgetTokens = + typeof thinking.budget_tokens === 'number' + ? thinking.budget_tokens + : Number.NaN; + const effort = reasoningEffort + ? reasoningEffort + : Number.isFinite(budgetTokens) + ? budgetTokens <= 2_048 + ? 'low' + : budgetTokens <= 8_192 + ? 'medium' + : 'high' + : undefined; + + return { + ...(effort ? { effort } : {}), + summary: 'auto', + }; +}; + +const normalizeStopSequences = ( + stop: string | string[] | undefined, +): string[] => { + return (Array.isArray(stop) ? stop : stop ? [stop] : []).filter(Boolean); +}; + +const findFirstStopSequence = ( + text: string, + stopSequences: string[], +): number | null => { + return stopSequences.reduce((earliest, stopSequence) => { + const index = text.indexOf(stopSequence); + if (index < 0) return earliest; + return earliest === null ? index : Math.min(earliest, index); + }, null); +}; + +const getPendingStopPrefixLength = ( + text: string, + stopSequences: string[], +): number => { + const maximumLength = Math.min( + text.length, + Math.max( + 0, + ...stopSequences.map((stopSequence) => stopSequence.length - 1), + ), + ); + + for (let length = maximumLength; length > 0; length -= 1) { + const suffix = text.slice(-length); + if (stopSequences.some((stopSequence) => stopSequence.startsWith(suffix))) { + return length; + } + } + + return 0; +}; + +const normalizeResponsesUpstreamBody = ( + body: Record, +): Record => { + const { messages, ...rest } = body; + + if (rest.input !== undefined || !Array.isArray(messages)) { + return rest; + } + + const systemInstructions = messages + .filter((message) => { + return ( + message && + typeof message === 'object' && + ((message as { role?: unknown }).role === 'system' || + (message as { role?: unknown }).role === 'developer') + ); + }) + .map((message) => { + return stringifyResponsesInputContent( + (message as { content?: unknown }).content, + ); + }) + .filter(Boolean) + .join('\n\n'); + const input = messages.flatMap((message) => { + if (!message || typeof message !== 'object') return []; + const value = message as { content?: unknown; role?: unknown }; + if (value.role === 'system' || value.role === 'developer') return []; + const role = value.role === 'assistant' ? 'assistant' : 'user'; + return [ + { + content: mapChatContentToResponses(value.content), + role, + }, + ]; + }); + + const existingInstructions = + typeof rest.instructions === 'string' ? rest.instructions.trim() : ''; + const instructions = [existingInstructions, systemInstructions] + .filter(Boolean) + .join('\n\n'); + + return { ...rest, ...(instructions ? { instructions } : {}), input }; +}; + +const buildResponsesBodyFromChat = ( + body: ChatRequestBody, +): Record => { + const instructions = body.messages + ?.filter( + (message) => message.role === 'system' || message.role === 'developer', + ) + .map((message) => stringifyResponsesInputContent(message.content)) + .filter(Boolean) + .join('\n\n'); + const input = + body.messages + ?.filter( + (message) => message.role !== 'system' && message.role !== 'developer', + ) + .map((message) => { + if (message.role === 'tool') { + return { + call_id: message.tool_call_id, + output: stringifyResponsesInputContent(message.content), + type: 'function_call_output', + }; + } + const toolCalls = Array.isArray(message.tool_calls) + ? message.tool_calls + : []; + const functionCalls = toolCalls.flatMap((toolCall) => { + if (!toolCall || typeof toolCall !== 'object') return []; + const call = toolCall as { + function?: { arguments?: unknown; name?: unknown }; + id?: unknown; + }; + if (typeof call.function?.name !== 'string') return []; + return [ + { + arguments: String(call.function.arguments ?? ''), + call_id: String(call.id ?? crypto.randomUUID()), + name: call.function.name, + type: 'function_call', + }, + ]; + }); + const content = mapChatContentToResponses(message.content); + const hasContent = content.some((part) => { + return ( + (part.type === 'input_text' && Boolean(part.text)) || + (part.type === 'input_image' && Boolean(part.image_url)) + ); + }); + const shouldOmitMessage = + message.role === 'assistant' && + functionCalls.length > 0 && + !hasContent; + + return [ + ...(shouldOmitMessage + ? [] + : [ + { + content, + role: message.role === 'assistant' ? 'assistant' : 'user', + }, + ]), + ...functionCalls, + ]; + }) + .flat() ?? []; + const tools = body.tools?.flatMap((tool) => { + if (!tool || typeof tool !== 'object') return []; + const value = tool as { + function?: Record; + type?: unknown; + }; + const definition: Record = + value.type === 'function' && value.function ? value.function : value; + if (typeof definition.name !== 'string') return []; + return [ + { + ...definition, + parameters: definition.parameters ?? { type: 'object', properties: {} }, + type: 'function', + }, + ]; + }); + const text = translateChatResponseFormatToResponses(body.response_format); + const reasoning = translateChatThinkingToResponses( + body.thinking, + body.reasoning_effort, + ); + + return { + ...(instructions ? { instructions } : {}), + input, + max_output_tokens: body.max_tokens ?? body.max_completion_tokens, + model: body.model, + parallel_tool_calls: body.parallel_tool_calls, + reasoning, + stream: Boolean(body.stream), + temperature: body.temperature, + top_p: body.top_p, + ...(tools?.length ? { tools } : {}), + ...(body.tool_choice + ? { tool_choice: translateChatToolChoiceToResponses(body.tool_choice) } + : {}), + ...(text ? { text } : {}), + }; +}; + +const getUnsupportedResponsesChatOptions = ( + body: ChatRequestBody, +): string[] => { + return [ + body.frequency_penalty !== undefined ? 'frequency_penalty' : null, + body.presence_penalty !== undefined ? 'presence_penalty' : null, + body.thinking !== undefined && + !translateChatThinkingToResponses(body.thinking, body.reasoning_effort) + ? 'thinking' + : null, + ].filter((name): name is string => Boolean(name)); +}; + +const extractResponsesReasoningText = (output: unknown[]): string => { + return output + .flatMap((item) => { + if (!item || typeof item !== 'object') return []; + const value = item as { + content?: unknown; + summary?: unknown; + type?: unknown; + }; + if (value.type !== 'reasoning') return []; + return [value.summary, value.content].flatMap((parts) => { + if (!Array.isArray(parts)) return []; + return parts.flatMap((part) => { + if (!part || typeof part !== 'object') return []; + const text = (part as { text?: unknown }).text; + return typeof text === 'string' ? [text] : []; + }); + }); + }) + .join(''); +}; + +const mapResponsesPayloadToChat = ( + payload: Record, + model: string, + stop: string | string[] | undefined, +): Record => { + const output = Array.isArray(payload.output) ? payload.output : []; + const toolCalls = output.flatMap((item) => { + if (!item || typeof item !== 'object') return []; + const value = item as Record; + if (value.type !== 'function_call') return []; + return [ + { + function: { + arguments: String(value.arguments ?? ''), + name: String(value.name ?? 'function'), + }, + id: String(value.call_id ?? value.id ?? crypto.randomUUID()), + type: 'function', + }, + ]; + }); + const usage = + payload.usage && typeof payload.usage === 'object' + ? (payload.usage as Record) + : undefined; + const inputTokens = Number(usage?.input_tokens ?? 0); + const outputTokens = Number(usage?.output_tokens ?? 0); + + const rawOutputText = + typeof payload.output_text === 'string' + ? payload.output_text + : output + .flatMap((item) => { + if (!item || typeof item !== 'object') return []; + const content = (item as { content?: unknown }).content; + if (!Array.isArray(content)) return []; + return content.flatMap((part) => { + if (!part || typeof part !== 'object') return []; + const value = part as { text?: unknown; type?: unknown }; + return value.type === 'output_text' && + typeof value.text === 'string' + ? [value.text] + : []; + }); + }) + .join(''); + const stopIndex = findFirstStopSequence( + rawOutputText, + normalizeStopSequences(stop), + ); + const outputText = + stopIndex === null ? rawOutputText : rawOutputText.slice(0, stopIndex); + const reasoningText = extractResponsesReasoningText(output); + const incompleteReason = + payload.incomplete_details && typeof payload.incomplete_details === 'object' + ? (payload.incomplete_details as { reason?: unknown }).reason + : undefined; + const finishReason = + payload.status === 'incomplete' + ? incompleteReason === 'content_filter' + ? 'content_filter' + : 'length' + : toolCalls.length + ? 'tool_calls' + : 'stop'; + + return { + choices: [ + { + finish_reason: finishReason, + index: 0, + message: { + content: outputText || null, + role: 'assistant', + ...(reasoningText ? { reasoning_content: reasoningText } : {}), + ...(toolCalls.length ? { tool_calls: toolCalls } : {}), + }, + }, + ], + created: Number(payload.created_at ?? Math.floor(Date.now() / 1000)), + id: String(payload.id ?? `chatcmpl-${crypto.randomUUID()}`), + model, + object: 'chat.completion', + usage: { + completion_tokens: outputTokens, + prompt_tokens: inputTokens, + total_tokens: Number(usage?.total_tokens ?? inputTokens + outputTokens), + }, + }; +}; + +const mapResponsesStreamToChat = ( + upstreamResponse: Response, + model: string, + proxyContext: ProxyContext, + route: string, + stop: string | string[] | undefined, + includeUsage: boolean, +): Response => { + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const responseId = `chatcmpl-${crypto.randomUUID()}`; + let reader: ReadableStreamDefaultReader | null = + upstreamResponse.body?.getReader() ?? null; + const fallbackUsage = parseUsageHeader(upstreamResponse); + let buffer = ''; + let emittedFinish = false; + let emittedUsage = false; + let hasToolCalls = false; + let latestUsage = fallbackUsage; + let usageRecorded = false; + const stopSequences = normalizeStopSequences(stop); + let pendingStopText = ''; + const toolIndexes = new Map(); + const toolCallIds = new Map(); + let nextToolIndex = 0; + let stoppedLocally = false; + + const getToolIndex = (itemId: string): number => { + const existing = toolIndexes.get(itemId); + if (existing !== undefined) return existing; + const index = nextToolIndex; + nextToolIndex += 1; + toolIndexes.set(itemId, index); + return index; + }; + + const encodeChunk = (choice: Record): Uint8Array => { + return encoder.encode( + `data: ${JSON.stringify({ + choices: [choice], + created: Math.floor(Date.now() / 1000), + id: responseId, + model, + object: 'chat.completion.chunk', + })}\n\n`, + ); + }; + + const enqueueUsage = ( + controller: ReadableStreamDefaultController, + ): void => { + if (!includeUsage || emittedUsage) return; + const usage = mapResponsesUsageToChat(latestUsage); + if (!usage) return; + + emittedUsage = true; + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + choices: [], + created: Math.floor(Date.now() / 1000), + id: responseId, + model, + object: 'chat.completion.chunk', + usage, + })}\n\n`, + ), + ); + }; + + const recordStreamUsage = async (): Promise => { + if (usageRecorded) return; + usageRecorded = true; + try { + await recordProxyUsage({ + model, + proxyContext, + route, + usage: latestUsage, + }); + } catch (error) { + console.error('[CodeBuddy2API] Failed to record Responses stream usage', { + error, + route, + }); + } + }; + + const cancelAndReleaseReader = async (reason?: unknown): Promise => { + try { + await reader?.cancel(reason); + } catch (error) { + console.error('[CodeBuddy2API] Failed to cancel Responses stream', { + error, + route, + }); + } finally { + reader?.releaseLock(); + reader = null; + } + }; + + const stream = new ReadableStream({ + async pull(controller) { + if (!reader) { + await recordStreamUsage(); + controller.close(); + return; + } + while (true) { + let readResult: ReadableStreamReadResult; + try { + readResult = await reader.read(); + } catch (error) { + await recordStreamUsage(); + reader.releaseLock(); + reader = null; + controller.error(error); + return; + } + const { done, value } = readResult; + if (done) { + if (pendingStopText) { + controller.enqueue( + encodeChunk({ + delta: { content: pendingStopText }, + index: 0, + }), + ); + pendingStopText = ''; + } + if (!emittedFinish) { + controller.enqueue( + encodeChunk({ + delta: {}, + finish_reason: hasToolCalls ? 'tool_calls' : 'stop', + index: 0, + }), + ); + } + enqueueUsage(controller); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await recordStreamUsage(); + reader.releaseLock(); + reader = null; + controller.close(); + return; + } + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + if (buffer.length > MAX_STREAM_FRAME_LENGTH) { + controller.enqueue( + encoder.encode( + 'data: {"error":{"message":"Upstream SSE frame exceeds the maximum size"}}\n\n', + ), + ); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await cancelAndReleaseReader(); + await recordStreamUsage(); + controller.close(); + return; + } + let emitted = false; + for (const frame of frames) { + if (frame.length > MAX_STREAM_FRAME_LENGTH) { + controller.enqueue( + encoder.encode( + 'data: {"error":{"message":"Upstream SSE frame exceeds the maximum size"}}\n\n', + ), + ); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await cancelAndReleaseReader(); + await recordStreamUsage(); + controller.close(); + return; + } + const dataLine = frame + .split(/\r?\n/) + .find((line) => line.startsWith('data: ')); + if (!dataLine || dataLine === 'data: [DONE]') continue; + try { + const event = JSON.parse(dataLine.slice(6)) as { + delta?: unknown; + item?: unknown; + item_id?: unknown; + output_index?: unknown; + error?: unknown; + response?: unknown; + type?: unknown; + }; + latestUsage = extractResponsesUsage(event) ?? latestUsage; + if ( + stoppedLocally && + event.type !== 'response.completed' && + event.type !== 'response.incomplete' + ) { + continue; + } + if (event.type === 'response.output_text.delta') { + const delta = String(event.delta ?? ''); + if (stopSequences.length) { + pendingStopText += delta; + const stopIndex = findFirstStopSequence( + pendingStopText, + stopSequences, + ); + if (stopIndex !== null) { + const content = pendingStopText.slice(0, stopIndex); + if (content) { + controller.enqueue( + encodeChunk({ delta: { content }, index: 0 }), + ); + } + pendingStopText = ''; + controller.enqueue( + encodeChunk({ + delta: {}, + finish_reason: 'stop', + index: 0, + }), + ); + emittedFinish = true; + stoppedLocally = true; + emitted = true; + continue; + } + + const pendingLength = getPendingStopPrefixLength( + pendingStopText, + stopSequences, + ); + const content = pendingStopText.slice( + 0, + pendingStopText.length - pendingLength, + ); + pendingStopText = pendingLength + ? pendingStopText.slice(-pendingLength) + : ''; + if (!content) continue; + controller.enqueue( + encodeChunk({ delta: { content }, index: 0 }), + ); + emitted = true; + continue; + } + controller.enqueue( + encodeChunk({ + delta: { content: delta }, + index: 0, + }), + ); + emitted = true; + continue; + } + if ( + event.type === 'response.reasoning_summary_text.delta' || + event.type === 'response.reasoning_text.delta' + ) { + controller.enqueue( + encodeChunk({ + delta: { reasoning_content: String(event.delta ?? '') }, + index: 0, + }), + ); + emitted = true; + continue; + } + if ( + event.type === 'response.output_item.added' && + event.item && + typeof event.item === 'object' + ) { + const item = event.item as { + arguments?: unknown; + call_id?: unknown; + id?: unknown; + name?: unknown; + type?: unknown; + }; + if (item.type !== 'function_call') continue; + const itemId = String(item.id ?? item.call_id ?? nextToolIndex); + const index = getToolIndex(itemId); + const callId = String(item.call_id ?? item.id ?? itemId); + toolCallIds.set(itemId, callId); + hasToolCalls = true; + controller.enqueue( + encodeChunk({ + delta: { + tool_calls: [ + { + function: { + arguments: String(item.arguments ?? ''), + name: String(item.name ?? 'function'), + }, + id: callId, + index, + type: 'function', + }, + ], + }, + index: 0, + }), + ); + emitted = true; + continue; + } + if (event.type === 'response.function_call_arguments.delta') { + const itemId = String( + event.item_id ?? event.output_index ?? nextToolIndex, + ); + const index = getToolIndex(itemId); + const callId = toolCallIds.get(itemId) ?? `call_${index + 1}`; + toolCallIds.set(itemId, callId); + hasToolCalls = true; + controller.enqueue( + encodeChunk({ + delta: { + tool_calls: [ + { + function: { arguments: String(event.delta ?? '') }, + id: callId, + index, + }, + ], + }, + index: 0, + }), + ); + emitted = true; + continue; + } + if (event.type === 'response.completed') { + if (pendingStopText) { + controller.enqueue( + encodeChunk({ + delta: { content: pendingStopText }, + index: 0, + }), + ); + pendingStopText = ''; + } + if (!emittedFinish) { + controller.enqueue( + encodeChunk({ + delta: {}, + finish_reason: hasToolCalls ? 'tool_calls' : 'stop', + index: 0, + }), + ); + } + enqueueUsage(controller); + emittedFinish = true; + emitted = true; + if (stoppedLocally) { + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await cancelAndReleaseReader('Stop sequence matched'); + await recordStreamUsage(); + controller.close(); + return; + } + continue; + } + if (event.type === 'response.incomplete') { + if (pendingStopText) { + controller.enqueue( + encodeChunk({ + delta: { content: pendingStopText }, + index: 0, + }), + ); + pendingStopText = ''; + } + const incompleteReason = + event.response && typeof event.response === 'object' + ? ( + (event.response as { incomplete_details?: unknown }) + .incomplete_details as { reason?: unknown } | undefined + )?.reason + : undefined; + if (!emittedFinish) { + controller.enqueue( + encodeChunk({ + delta: {}, + finish_reason: + incompleteReason === 'content_filter' + ? 'content_filter' + : 'length', + index: 0, + }), + ); + } + enqueueUsage(controller); + emittedFinish = true; + emitted = true; + if (stoppedLocally) { + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await cancelAndReleaseReader('Stop sequence matched'); + await recordStreamUsage(); + controller.close(); + return; + } + continue; + } + if ( + event.type === 'response.failed' || + event.type === 'response.error' || + event.type === 'error' + ) { + const failure = + event.error ?? + (event.response && typeof event.response === 'object' + ? (event.response as { error?: unknown }).error + : undefined); + const message = + failure && typeof failure === 'object' + ? String( + (failure as { message?: unknown }).message ?? failure, + ) + : String(failure ?? 'Upstream Responses stream failed'); + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ error: { message } })}\n\n`, + ), + ); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + await cancelAndReleaseReader(); + await recordStreamUsage(); + controller.close(); + return; + } + } catch { + // Ignore malformed upstream events and continue reading. + } + } + if (stoppedLocally) continue; + if (emitted) return; + } + }, + async cancel(reason) { + await cancelAndReleaseReader(reason); + }, + }); + + return new Response(stream, { + headers: { + 'Access-Control-Allow-Origin': '*', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'Content-Type': 'text/event-stream; charset=utf-8', + }, + status: upstreamResponse.status, + }); +}; + const aggregateToolCalls = ( toolCalls: NonNullable, ): Array<{ @@ -1383,6 +2408,107 @@ export const proxyChatCompletions = async ( context ?? (await resolveProxyContext(request, body.model)); setDebugTraceCredential(debugTrace, resolvedContext.credentialFilename); const upstreamBody = await buildUpstreamBody(body, resolvedContext); + + if (resolvedContext.preferences.upstreamProtocol === 'responses') { + const unsupportedOptions = getUnsupportedResponsesChatOptions(body); + if (unsupportedOptions.length) { + return createErrorResponse( + 400, + `Unsupported Chat options for Responses upstream: ${unsupportedOptions.join(', ')}`, + ); + } + const apiEndpoint = await getCodeBuddyApiEndpoint(); + const upstreamUrl = `${apiEndpoint}/responses`; + const upstreamHeaders = new Headers( + await buildUpstreamHeaders(request, resolvedContext.auth), + ); + upstreamHeaders.set('User-Agent', 'TCodex/0.0.16 CLI/0.144.5'); + upstreamHeaders.set('X-IDE-Name', 'TCodex'); + upstreamHeaders.set('X-IDE-Version', '0.144.5'); + upstreamHeaders.set('X-Product-Version', '0.0.16'); + const responsesBody = { + ...buildResponsesBodyFromChat(upstreamBody), + stream: Boolean(body.stream), + }; + + setDebugUpstreamRequest(debugTrace, { + body: responsesBody, + headers: headersToRecord(upstreamHeaders), + method: 'POST', + url: upstreamUrl, + }); + + let upstreamResponse = await fetch(upstreamUrl, { + method: 'POST', + headers: upstreamHeaders, + body: JSON.stringify(responsesBody), + cache: 'no-store', + }); + upstreamResponse = enqueueUpstreamResponseSnapshot( + debugTrace, + upstreamResponse, + ); + + if (!upstreamResponse.ok) { + const detail = await upstreamResponse.text(); + logUpstreamFailure({ + detail, + route: usageRoute, + status: upstreamResponse.status, + url: upstreamUrl, + }); + setDebugTraceError(debugTrace, detail); + return createErrorResponse( + upstreamResponse.status, + 'Upstream CodeBuddy request failed', + detail, + ); + } + + if (body.stream) { + return mapResponsesStreamToChat( + upstreamResponse, + String(upstreamBody.model ?? 'unknown'), + resolvedContext, + usageRoute, + body.stop, + Boolean(body.stream_options?.include_usage) || + usageRoute === '/v1/messages', + ); + } + + const payload = (await upstreamResponse.json()) as Record< + string, + unknown + >; + await recordProxyUsage({ + model: String(upstreamBody.model ?? 'unknown'), + proxyContext: resolvedContext, + route: usageRoute, + usage: payload.usage ?? null, + }); + if (payload.status === 'failed' || payload.error) { + const error = + payload.error && typeof payload.error === 'object' + ? (payload.error as { message?: unknown }) + : undefined; + return createErrorResponse( + 502, + typeof error?.message === 'string' + ? error.message + : 'Upstream Responses request failed', + payload.error, + ); + } + return Response.json( + mapResponsesPayloadToChat( + payload, + String(upstreamBody.model ?? 'unknown'), + body.stop, + ), + ); + } + const apiEndpoint = await getCodeBuddyApiEndpoint(); const upstreamUrl = `${apiEndpoint}/v2/chat/completions`; const upstreamHeaders = await buildUpstreamHeaders( @@ -1493,6 +2619,7 @@ export const proxyResponsesUpstream = async ( body: Record, context?: ProxyContext, debugTrace?: DebugTrace, + onResponseId?: (responseId: string) => Promise, ): Promise => { try { const resolvedContext = @@ -1503,18 +2630,21 @@ export const proxyResponsesUpstream = async ( )); setDebugTraceCredential(debugTrace, resolvedContext.credentialFilename); const upstreamBody = { - ...body, + ...normalizeResponsesUpstreamBody(body), model: typeof body.model === 'string' && body.model.trim() ? body.model : await getDefaultModel(), }; const apiEndpoint = await getCodeBuddyApiEndpoint(); - const upstreamUrl = `${apiEndpoint}/v1/responses`; - const upstreamHeaders = await buildUpstreamHeaders( - request, - resolvedContext.auth, + const upstreamUrl = `${apiEndpoint}/responses`; + const upstreamHeaders = new Headers( + await buildUpstreamHeaders(request, resolvedContext.auth), ); + upstreamHeaders.set('User-Agent', 'TCodex/0.0.16 CLI/0.144.5'); + upstreamHeaders.set('X-IDE-Name', 'TCodex'); + upstreamHeaders.set('X-IDE-Version', '0.144.5'); + upstreamHeaders.set('X-Product-Version', '0.0.16'); setDebugUpstreamRequest(debugTrace, { body: upstreamBody, @@ -1558,15 +2688,20 @@ export const proxyResponsesUpstream = async ( if (contentType.toLowerCase().includes('application/json')) { const payloadText = await upstreamResponse.text(); let usage = fallbackUsage; + let responseId: string | null = null; try { - usage = - extractResponsesUsage(JSON.parse(payloadText) as unknown) ?? - fallbackUsage; + const payload = JSON.parse(payloadText) as unknown; + usage = extractResponsesUsage(payload) ?? fallbackUsage; + responseId = extractResponsesId(payload); } catch { // Preserve malformed upstream JSON while retaining header usage. } + if (responseId && onResponseId) { + await onResponseId(responseId); + } + await recordProxyUsage({ model, proxyContext: resolvedContext, @@ -1584,6 +2719,7 @@ export const proxyResponsesUpstream = async ( return trackResponsesUsageStream({ fallbackUsage, model, + onResponseId, proxyContext: resolvedContext, upstreamResponse, }); @@ -1605,7 +2741,7 @@ export const proxyResponsesUpstream = async ( logUpstreamFailure({ error, route: '/v1/responses', - url: `${await getCodeBuddyApiEndpoint()}/v1/responses`, + url: `${await getCodeBuddyApiEndpoint()}/responses`, }); return createErrorResponse( 500, diff --git a/lib/server/proxy/responses.ts b/lib/server/proxy/responses.ts index 504162e..e4ed402 100644 --- a/lib/server/proxy/responses.ts +++ b/lib/server/proxy/responses.ts @@ -68,6 +68,7 @@ interface ResponseSession { model: string; transcript: TranscriptMessage[]; defaults: ResponseSessionDefaults; + upstreamProtocol?: 'chat' | 'responses'; } interface ChatResponseToolCall { @@ -310,6 +311,27 @@ const storeResponseSession = async ( pruneResponseSessions(); }; +const storeUpstreamResponseBinding = async ({ + model, + proxyContext, + responseId, +}: { + model: string; + proxyContext: ProxyContext; + responseId: string; +}): Promise => { + await storeResponseSession({ + accessKeyId: proxyContext.accessKeyId, + credentialFilename: proxyContext.credentialFilename, + createdAt: Date.now(), + defaults: {}, + id: responseId, + model, + transcript: [], + upstreamProtocol: 'responses', + }); +}; + const flattenNamespaceToolName = (namespace: string, name: string): string => { return `${namespace}__${name}`; }; @@ -1083,6 +1105,7 @@ const mapChatResponseToResponsesPayload = async ( }, ], defaults, + upstreamProtocol: 'chat', }); return { @@ -1307,6 +1330,7 @@ const createResponsesEventStream = async ( }, ], defaults, + upstreamProtocol: 'chat', }); } catch (error) { console.error( @@ -1624,7 +1648,7 @@ export const handleResponsesRequest = async ( throw new Error('Unknown or expired previous_response_id'); } - const proxyContext = storedPreviousSession?.credentialFilename + const resolvedProxyContext = storedPreviousSession?.credentialFilename ? await resolveProxyContextByCredentialFilename( storedPreviousSession.credentialFilename, { @@ -1642,25 +1666,45 @@ export const handleResponsesRequest = async ( request, typeof body.model === 'string' ? body.model : undefined, ); + const proxyContext = storedPreviousSession?.upstreamProtocol + ? { + ...resolvedProxyContext, + preferences: { + ...resolvedProxyContext.preferences, + upstreamProtocol: storedPreviousSession.upstreamProtocol, + }, + } + : resolvedProxyContext; const scopedBody = - !storedPreviousSession && - (typeof body.model !== 'string' || !body.model.trim()) + typeof body.model !== 'string' || !body.model.trim() ? { ...body, model: + storedPreviousSession?.model ?? getCredentialSupportedModels( proxyContext.auth.credentialData, - )[0] ?? (await getDefaultModel()), + )[0] ?? + (await getDefaultModel()), } : body; - if (proxyContext.preferences.responsesPassthrough) { + if (proxyContext.preferences.upstreamProtocol === 'responses') { + const model = String( + (scopedBody as ResponsesRequestBody).model ?? 'unknown', + ); return proxyResponsesUpstream( request, scopedBody as Record, proxyContext, debugTrace, + async (responseId) => { + await storeUpstreamResponseBinding({ + model, + proxyContext, + responseId, + }); + }, ); } diff --git a/messages/en-US.json b/messages/en-US.json index 70dd76e..fcb1a8c 100644 --- a/messages/en-US.json +++ b/messages/en-US.json @@ -229,9 +229,12 @@ "credentialEditTitle": "Edit credential settings", "credentialEmpty": "No credentials yet. Authenticate or add one manually first.", "credentialExpired": "Expired credentials", - "credentialResponsesDirect": "Send Responses requests upstream directly", - "credentialResponsesDirectHelp": "When enabled, `/v1/responses` requests using this credential are sent upstream directly instead of being converted into Chat Completions.", - "credentialResponsesProxyTag": "Responses → Chat", + "credentialUpstreamProtocol": "Upstream protocol", + "credentialUpstreamProtocolHelp": "Chat sends requests to Chat Completions. Responses converts every supported downstream API to the upstream Responses protocol.", + "credentialUpstreamChat": "Chat", + "credentialUpstreamResponses": "Responses", + "credentialUpstreamChatTag": "Upstream: Chat", + "credentialUpstreamResponsesTag": "Upstream: Responses", "credentialRoleAsSystem": "Normalize developer messages for upstream", "credentialRoleAsSystemHelp": "Sends every developer message as user.", "credentialRoleAsSystemTag": "developer → user", diff --git a/messages/ja-JP.json b/messages/ja-JP.json index d34eb60..4344a78 100644 --- a/messages/ja-JP.json +++ b/messages/ja-JP.json @@ -229,9 +229,12 @@ "credentialEditTitle": "認証情報設定を編集", "credentialEmpty": "認証情報がありません。先に認証するか手動で追加してください。", "credentialExpired": "期限切れの認証情報", - "credentialResponsesDirect": "Responses リクエストを上流へ直接送信", - "credentialResponsesDirectHelp": "有効にすると、この認証情報で処理される `/v1/responses` リクエストは Chat Completions へ変換せず上流へ直接送信されます。", - "credentialResponsesProxyTag": "Responses → Chat", + "credentialUpstreamProtocol": "上流プロトコル", + "credentialUpstreamProtocolHelp": "Chat は Chat Completions に送信します。Responses は対応するすべての下流 API を上流の Responses プロトコルに変換します。", + "credentialUpstreamChat": "Chat", + "credentialUpstreamResponses": "Responses", + "credentialUpstreamChatTag": "上流: Chat", + "credentialUpstreamResponsesTag": "上流: Responses", "credentialRoleAsSystem": "developer メッセージを上流向けに正規化", "credentialRoleAsSystemHelp": "すべての developer メッセージを user として送信します。", "credentialRoleAsSystemTag": "developer → user", diff --git a/messages/zh-CN.json b/messages/zh-CN.json index aa4987c..f9fee34 100644 --- a/messages/zh-CN.json +++ b/messages/zh-CN.json @@ -225,9 +225,12 @@ "credentialEditTitle": "编辑凭证配置", "credentialEmpty": "暂无凭证，请先认证或手动添加。", "credentialExpired": "已过期凭证", - "credentialResponsesDirect": "直接转发 Responses 请求至上游", - "credentialResponsesDirectHelp": "启用后，使用此凭证的 `/v1/responses` 请求将直接发送到上游，不再转换为 Chat Completions。", - "credentialResponsesProxyTag": "Responses → Chat", + "credentialUpstreamProtocol": "上游协议", + "credentialUpstreamProtocolHelp": "Chat 将请求发送到 Chat Completions；Responses 会将所有受支持的下游接口转换为上游 Responses 协议。", + "credentialUpstreamChat": "Chat", + "credentialUpstreamResponses": "Responses", + "credentialUpstreamChatTag": "上游：Chat", + "credentialUpstreamResponsesTag": "上游：Responses", "credentialRoleAsSystem": "转换 developer 消息角色以兼容上游", "credentialRoleAsSystemHelp": "所有 developer 消息均作为 user 发送。", "credentialRoleAsSystemTag": "developer → user", diff --git a/package.json b/package.json index 0bc8b17..a71c677 100644 --- a/package.json +++ b/package.json @@ -45,6 +45,7 @@ "@eslint/compat": "^2.1.0", "@next/eslint-plugin-next": "^16.3.0", "@tailwindcss/postcss": "^4.3.3", + "@testing-library/dom": "^10.4.1", "@testing-library/jest-dom": "^7.0.1", "@testing-library/react": "^16.3.0", "@types/better-sqlite3": "^9.6.0", diff --git a/tests/admin/initial-state.test.ts b/tests/admin/initial-state.test.ts index 43bd86e..be21055 100644 --- a/tests/admin/initial-state.test.ts +++ b/tests/admin/initial-state.test.ts @@ -60,6 +60,7 @@ describe('admin initial state', () => { is_expired: false, name: null, responses_passthrough: false, + upstream_protocol: 'chat', scope: null, session_state: null, tenant_id: null, @@ -82,6 +83,7 @@ describe('admin initial state', () => { is_expired: false, name: null, responses_passthrough: false, + upstream_protocol: 'chat', scope: null, session_state: null, tenant_id: null, diff --git a/tests/server/anthropic.test.ts b/tests/server/anthropic.test.ts index b28c5eb..6ced151 100644 --- a/tests/server/anthropic.test.ts +++ b/tests/server/anthropic.test.ts @@ -4,6 +4,7 @@ import path from 'node:path'; import { NextRequest } from 'next/server'; import { handleMessagesRequest } from '@/lib/server/proxy/anthropic'; +import { createAccessKey } from '@/lib/server/domain/access-keys'; import { addCredential } from '@/lib/server/domain/credentials'; const repoRoot = process.cwd(); @@ -72,6 +73,137 @@ describe('anthropic messages api', () => { expect(json.type).toBe('error'); }); + it('uses Anthropic errors and converts thinking for Responses upstream', async () => { + const credential = await addCredential({ + bearer_token: 'anthropic-responses-token', + upstream_protocol: 'responses', + user_id: 'anthropic-responses@example.com', + }); + const accessKey = await createAccessKey({ + credentialFilenames: [credential.filename], + name: 'Anthropic Responses Key', + }); + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + makeJsonResponse({ + id: 'resp_anthropic_reasoning', + output: [ + { + type: 'reasoning', + summary: [{ text: 'Think first', type: 'summary_text' }], + }, + ], + output_text: 'answerENDignored', + status: 'completed', + }), + ) + .mockResolvedValueOnce( + makeJsonResponse({ + error: { message: 'Responses model failed' }, + id: 'resp_anthropic_failed', + status: 'failed', + }), + ) + .mockResolvedValueOnce( + makeJsonResponse( + { error: { message: 'Responses tenant is rate limited' } }, + 429, + ), + ); + const request = makeNextRequest('http://localhost/v1/messages', { + method: 'POST', + headers: { authorization: `Bearer ${accessKey.secret}` }, + }); + + const response = await handleMessagesRequest(request, { + model: 'hy3', + max_tokens: 4096, + messages: [{ role: 'user', content: 'Think' }], + stop_sequences: ['END'], + thinking: { type: 'adaptive' }, + tool_choice: { disable_parallel_tool_use: true, type: 'auto' }, + }); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + content: [ + { thinking: 'Think first', type: 'thinking' }, + { text: 'answer', type: 'text' }, + ], + type: 'message', + }); + expect( + JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body)), + ).toMatchObject({ + parallel_tool_calls: false, + reasoning: { summary: 'auto' }, + }); + + const failedResponse = await handleMessagesRequest(request, { + model: 'hy3', + max_tokens: 4096, + messages: [{ role: 'user', content: 'Fail' }], + }); + expect(failedResponse.status).toBe(502); + expect(await failedResponse.json()).toMatchObject({ + error: { + message: 'Responses model failed', + type: 'api_error', + }, + type: 'error', + }); + + const limitedResponse = await handleMessagesRequest(request, { + model: 'hy3', + max_tokens: 4096, + messages: [{ role: 'user', content: 'Retry later' }], + }); + expect(limitedResponse.status).toBe(429); + expect(await limitedResponse.json()).toMatchObject({ + error: { + message: 'Responses tenant is rate limited', + type: 'rate_limit_error', + }, + type: 'error', + }); + }); + + it('preserves Responses usage in Anthropic streams', async () => { + const credential = await addCredential({ + bearer_token: 'anthropic-responses-stream-token', + upstream_protocol: 'responses', + user_id: 'anthropic-responses-stream@example.com', + }); + const accessKey = await createAccessKey({ + credentialFilenames: [credential.filename], + name: 'Anthropic Responses Stream Key', + }); + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + makeSseResponse([ + 'data: {"type":"response.output_text.delta","delta":"answer"}', + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":2,"total_tokens":5}}}', + 'data: [DONE]', + ]), + ); + + const response = await handleMessagesRequest( + makeNextRequest('http://localhost/v1/messages', { + method: 'POST', + headers: { authorization: `Bearer ${accessKey.secret}` }, + }), + { + model: 'hy3', + max_tokens: 1024, + stream: true, + messages: [{ role: 'user', content: 'Count usage' }], + }, + ); + const payload = await response.text(); + + expect(payload).toContain('"text":"answer"'); + expect(payload).toContain('"usage":{"input_tokens":3,"output_tokens":2'); + }); + it('translates a simple non-streaming request and response', async () => { const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( makeJsonResponse({ @@ -588,6 +720,28 @@ describe('anthropic messages api', () => { expect(text).not.toContain('event: message_stop'); }); + it('preserves upstream Chat stream errors', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + makeSseResponse([ + 'data: {"error":{"message":"Responses upstream is unavailable"}}', + ]), + ); + + const response = await handleMessagesRequest( + makeNextRequest('http://localhost/v1/messages', { method: 'POST' }), + { + model: 'claude-sonnet-4.6', + max_tokens: 1024, + stream: true, + messages: [{ role: 'user', content: 'Hi' }], + }, + ); + + const text = await response.text(); + expect(text).toContain('event: error'); + expect(text).toContain('Responses upstream is unavailable'); + }); + it('cancels the upstream Chat stream when an Anthropic client disconnects', async () => { const cancel = vi.fn(); vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( diff --git a/tests/server/units.test.ts b/tests/server/units.test.ts index ea23881..d987145 100644 --- a/tests/server/units.test.ts +++ b/tests/server/units.test.ts @@ -1320,6 +1320,7 @@ describe('server units', () => { '00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01', ); expect(upstreamHeaders.get('tracestate')).toBe('vendor=value'); + expect(upstreamHeaders.get('Authorization')).toBe('Bearer token-a'); expect( JSON.parse(String((fetchMock.mock.calls[2]?.[1] as RequestInit).body)) .max_tokens, @@ -1334,6 +1335,514 @@ describe('server units', () => { ).toBeUndefined(); }); + it('uses the Responses upstream protocol for Chat Completions requests', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-token', + upstream_protocol: 'responses', + user_id: 'responses@example.com', + }, + filePath: '/tmp/responses.json', + filename: 'responses.json', + }); + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + makeJsonResponse({ + created_at: 123, + id: 'resp_123', + output: [], + output_text: 'hello from responses', + usage: { input_tokens: 2, output_tokens: 3, total_tokens: 5 }, + }), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [ + { content: 'Follow instructions', role: 'system' }, + { + content: [ + { text: 'Say hello', type: 'text' }, + { + image_url: { detail: 'high', url: 'https://example.com/a.png' }, + type: 'image_url', + }, + ], + role: 'user', + }, + ], + model: 'hy3', + reasoning_effort: 'high', + parallel_tool_calls: false, + response_format: { + json_schema: { + name: 'answer', + schema: { properties: {}, type: 'object' }, + strict: true, + }, + type: 'json_schema', + }, + tool_choice: { + function: { name: 'lookup_weather' }, + type: 'function', + }, + temperature: 0.2, + top_p: 0.8, + tools: [ + { + function: { + name: 'lookup_weather', + parameters: { properties: {}, type: 'object' }, + }, + type: 'function', + }, + ], + }, + context, + ); + + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + choices: [ + { message: { content: 'hello from responses', role: 'assistant' } }, + ], + model: 'hy3', + object: 'chat.completion', + }); + expect(String(fetchMock.mock.calls[0]?.[0])).toBe( + 'https://copilot.tencent.com/responses', + ); + expect( + JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body)), + ).toMatchObject({ + input: [ + { + content: [ + { text: 'Say hello', type: 'input_text' }, + { + detail: 'high', + image_url: 'https://example.com/a.png', + type: 'input_image', + }, + ], + role: 'user', + }, + ], + instructions: 'Follow instructions', + model: 'hy3', + parallel_tool_calls: false, + reasoning: { effort: 'high' }, + stream: false, + temperature: 0.2, + text: { + format: { + name: 'answer', + schema: { properties: {}, type: 'object' }, + strict: true, + type: 'json_schema', + }, + }, + tool_choice: { name: 'lookup_weather', type: 'function' }, + tools: [ + { + name: 'lookup_weather', + parameters: { properties: {}, type: 'object' }, + type: 'function', + }, + ], + top_p: 0.8, + }); + }); + + it('applies Chat stop sequences locally for the Responses upstream', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-options-token', + upstream_protocol: 'responses', + user_id: 'responses-options@example.com', + }, + filePath: '/tmp/responses-options.json', + filename: 'responses-options.json', + }); + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + makeJsonResponse({ + id: 'resp_stopped', + output: [], + output_text: 'beforeENDafter', + status: 'completed', + }), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Stop early', role: 'user' }], + model: 'hy3', + stop: ['END'], + }, + context, + ); + + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + choices: [ + { + finish_reason: 'stop', + message: { content: 'before' }, + }, + ], + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect( + JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body)), + ).not.toHaveProperty('stop'); + }); + + it('maps Responses reasoning and semantic failures to Chat', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-reasoning-token', + upstream_protocol: 'responses', + user_id: 'responses-reasoning@example.com', + }, + filePath: '/tmp/responses-reasoning.json', + filename: 'responses-reasoning.json', + }); + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + makeJsonResponse({ + id: 'resp_reasoning', + output: [ + { + type: 'reasoning', + summary: [{ text: 'Reasoning summary', type: 'summary_text' }], + }, + ], + output_text: 'answer', + status: 'completed', + }), + ) + .mockResolvedValueOnce( + makeJsonResponse({ + error: { code: 'server_error', message: 'model failed' }, + id: 'resp_failed', + status: 'failed', + }), + ); + + const reasoningResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Think', role: 'user' }], + model: 'hy3', + thinking: { type: 'adaptive' }, + }, + context, + ); + expect(await reasoningResponse.json()).toMatchObject({ + choices: [ + { + message: { + content: 'answer', + reasoning_content: 'Reasoning summary', + }, + }, + ], + }); + expect( + JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body)), + ).toMatchObject({ reasoning: { summary: 'auto' } }); + + const failedResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Fail', role: 'user' }], + model: 'hy3', + }, + context, + ); + expect(failedResponse.status).toBe(502); + expect(await failedResponse.text()).toContain('model failed'); + }); + + it('maps Responses content filtering and split stop sequences in streams', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-stream-stop-token', + upstream_protocol: 'responses', + user_id: 'responses-stream-stop@example.com', + }, + filePath: '/tmp/responses-stream-stop.json', + filename: 'responses-stream-stop.json', + }); + vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.output_text.delta","delta":"beforeEN"}\n\n' + + 'data: {"type":"response.output_text.delta","delta":"Dafter"}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"content_filter"}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const stoppedResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Stop', role: 'user' }], + model: 'hy3', + stop: 'END', + stream: true, + }, + context, + ); + const stoppedText = await stoppedResponse.text(); + expect(stoppedText).toContain('"content":"before"'); + expect(stoppedText).not.toContain('after'); + expect(stoppedText).toContain('"finish_reason":"stop"'); + + const filteredResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Filter', role: 'user' }], + model: 'hy3', + stream: true, + }, + context, + ); + expect(await filteredResponse.text()).toContain( + '"finish_reason":"content_filter"', + ); + }); + + it('emits Responses usage for Chat streams and preserves it after local stops', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-stream-usage-token', + upstream_protocol: 'responses', + user_id: 'responses-stream-usage@example.com', + }, + filePath: '/tmp/responses-stream-usage.json', + filename: 'responses-stream-usage.json', + }); + vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.output_text.delta","delta":"normal"}\n\n' + + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":2,"total_tokens":5}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.output_text.delta","delta":"beforeENDafter"}\n\n' + + 'data: {"type":"response.output_text.delta","delta":"ignored"}\n\n' + + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":4,"output_tokens":3,"total_tokens":7}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const normalResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Normal usage', role: 'user' }], + model: 'hy3', + stream: true, + stream_options: { include_usage: true }, + }, + context, + ); + const normalPayload = await normalResponse.text(); + expect(normalPayload).toContain('"choices":[]'); + expect(normalPayload).toContain( + '"prompt_tokens":3,"prompt_tokens_details"', + ); + expect(normalPayload).toContain('"completion_tokens":2'); + + const stoppedResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Stopped usage', role: 'user' }], + model: 'hy3', + stop: 'END', + stream: true, + stream_options: { include_usage: true }, + }, + context, + ); + const stoppedPayload = await stoppedResponse.text(); + expect(stoppedPayload).toContain('"content":"before"'); + expect(stoppedPayload).not.toContain('after'); + expect(stoppedPayload).not.toContain('ignored'); + expect(stoppedPayload.match(/"finish_reason":"stop"/g)).toHaveLength(1); + expect(stoppedPayload).toContain('"completion_tokens":3'); + expect(stoppedPayload).toContain('"total_tokens":7'); + + expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ + { + callCount: 2, + cacheHitTokens: 0, + model: 'hy3', + totalTokens: 12, + }, + ]); + }); + + it('maps Responses function call events back to Chat Completions SSE', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-stream-token', + upstream_protocol: 'responses', + user_id: 'responses-stream@example.com', + }, + filePath: '/tmp/responses-stream.json', + filename: 'responses-stream.json', + }); + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response( + 'event: response.output_item.added\n' + + 'data: {"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"lookup_weather","arguments":""}}\n\n' + + 'event: response.function_call_arguments.delta\n' + + 'data: {"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{\\"city\\":\\"Shanghai\\"}"}\n\n' + + 'event: response.incomplete\n' + + 'data: {"type":"response.incomplete"}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Use the weather tool', role: 'user' }], + model: 'hy3', + stream: true, + }, + context, + ); + const payload = await response.text(); + + expect(payload).toContain('"name":"lookup_weather"'); + expect(payload).toContain('"arguments":"{\\"city\\":\\"Shanghai\\"}"'); + expect(payload.match(/"id":"call_1"/g)).toHaveLength(2); + expect(payload).toContain('"finish_reason":"length"'); + expect(payload).toContain('data: [DONE]'); + }); + + it('surfaces Responses stream failures as Chat Completions errors', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-error-token', + upstream_protocol: 'responses', + user_id: 'responses-error@example.com', + }, + filePath: '/tmp/responses-error.json', + filename: 'responses-error.json', + }); + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response( + 'event: response.failed\n' + + 'data: {"type":"response.failed","response":{"error":{"message":"upstream failed"}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Fail upstream', role: 'user' }], + model: 'hy3', + stream: true, + }, + context, + ); + const payload = await response.text(); + + expect(payload).toContain('"message":"upstream failed"'); + expect(payload).not.toContain('"finish_reason":"stop"'); + expect(payload).toContain('data: [DONE]'); + }); + + it('records Responses usage when a Chat stream reader fails', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-reader-failure-token', + upstream_protocol: 'responses', + user_id: 'responses-reader-failure@example.com', + }, + filePath: '/tmp/responses-reader-failure.json', + filename: 'responses-reader-failure.json', + }); + const encoder = new TextEncoder(); + let pullCount = 0; + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response( + new ReadableStream({ + pull(controller) { + if (pullCount === 0) { + pullCount += 1; + controller.enqueue( + encoder.encode( + 'data: {"type":"response.output_text.delta","delta":"partial","response":{"usage":{"input_tokens":4,"output_tokens":3,"total_tokens":7}}}\n\n', + ), + ); + return; + } + + controller.error(new Error('Responses reader failed')); + }, + }), + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Fail after usage', role: 'user' }], + model: 'hy3', + stream: true, + }, + context, + ); + + await expect(response.text()).rejects.toThrow('Responses reader failed'); + expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ + { + callCount: 1, + cacheHitTokens: 0, + model: 'hy3', + totalTokens: 7, + }, + ]); + }); + it('persists successful proxy calls without upstream usage across runtime restarts', async () => { const credential = (await listCredentials()).credentials[0]; expect(credential).toBeDefined(); @@ -2116,7 +2625,8 @@ describe('server units', () => { }, }, ), - ); + ) + .mockResolvedValueOnce(makeJsonResponse({ output_text: 'normalized' })); const tools = [ { @@ -2458,11 +2968,34 @@ describe('server units', () => { }); await streamResponse.text(); - expect(fetchMock).toHaveBeenCalledTimes(2); + const normalizedResponse = await proxyResponsesUpstream(request, { + messages: [ + { content: 'System compatibility instruction', role: 'system' }, + { content: 'messages compatibility input', role: 'user' }, + ], + model: 'gpt-5.5', + }); + await normalizedResponse.text(); + + expect(fetchMock).toHaveBeenCalledTimes(3); + expect( + JSON.parse(String((fetchMock.mock.calls[2]?.[1] as RequestInit).body)), + ).toMatchObject({ + input: [ + { + content: [ + { text: 'messages compatibility input', type: 'input_text' }, + ], + role: 'user', + }, + ], + instructions: 'System compatibility instruction', + model: 'gpt-5.5', + }); await waitForAsync(async () => { expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ { - callCount: 2, + callCount: 3, cacheHitTokens: 0, model: 'gpt-5.5', totalTokens: 14, @@ -2496,6 +3029,30 @@ describe('server units', () => { expect(cancel).toHaveBeenCalledWith('client disconnected'); }); + it('errors the downstream Responses stream when the upstream reader fails', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response( + new ReadableStream({ + pull(controller) { + controller.error(new Error('upstream connection reset')); + }, + }), + { + headers: { + 'Content-Type': 'text/event-stream; charset=utf-8', + }, + }, + ), + ); + + const response = await proxyResponsesUpstream( + makeNextRequest('http://localhost/v1/responses', { method: 'POST' }), + { input: 'fail while streaming', model: 'gpt-5.5', stream: true }, + ); + + await expect(response.text()).rejects.toThrow('upstream connection reset'); + }); + it('covers responses passthrough header fallback, raw body passthrough, and upstream errors', async () => { const createdCredential = await addCredential({ bearer_token: 'token-responses', @@ -2718,6 +3275,137 @@ describe('server units', () => { ).toBe('resp_from_upstream'); }); + it('pins upstream Responses follow-ups to the original credential', async () => { + const firstCredential = await addCredential({ + bearer_token: 'token-response-binding-a', + upstream_protocol: 'responses', + user_id: 'response-binding-a@example.com', + }); + const secondCredential = await addCredential({ + bearer_token: 'token-response-binding-b', + upstream_protocol: 'responses', + user_id: 'response-binding-b@example.com', + }); + const accessKey = await createAccessKey({ + credentialFilenames: [ + firstCredential.filename, + secondCredential.filename, + ], + name: 'Responses Binding Key', + }); + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + makeJsonResponse({ + id: 'resp_bound_upstream', + object: 'response', + output_text: 'first', + }), + ) + .mockResolvedValueOnce( + makeJsonResponse({ + id: 'resp_bound_follow_up', + object: 'response', + output_text: 'second', + }), + ); + const request = makeNextRequest('http://localhost/v1/responses', { + method: 'POST', + headers: { authorization: `Bearer ${accessKey.secret}` }, + }); + + const firstResponse = await handleResponsesRequest(request, { + input: 'first', + model: 'hy3', + }); + expect(firstResponse.status).toBe(200); + await firstResponse.text(); + + await addCredential( + { upstream_protocol: 'chat' }, + firstCredential.filename, + ); + + const followUpResponse = await handleResponsesRequest(request, { + input: 'second', + previous_response_id: 'resp_bound_upstream', + }); + expect(followUpResponse.status).toBe(200); + await followUpResponse.text(); + + const firstAuthorization = new Headers( + (fetchMock.mock.calls[0]?.[1] as RequestInit).headers, + ).get('authorization'); + const secondAuthorization = new Headers( + (fetchMock.mock.calls[1]?.[1] as RequestInit).headers, + ).get('authorization'); + expect(secondAuthorization).toBe(firstAuthorization); + expect(String(fetchMock.mock.calls[1]?.[0])).toBe( + 'https://copilot.tencent.com/responses', + ); + expect( + JSON.parse(String((fetchMock.mock.calls[1]?.[1] as RequestInit).body)), + ).toMatchObject({ model: 'hy3' }); + }); + + it('keeps Chat-backed Responses sessions on Chat after protocol changes', async () => { + const credential = await addCredential({ + bearer_token: 'token-chat-session-binding', + upstream_protocol: 'chat', + user_id: 'chat-session-binding@example.com', + }); + const accessKey = await createAccessKey({ + credentialFilenames: [credential.filename], + name: 'Chat Session Binding Key', + }); + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + makeJsonResponse({ + choices: [{ message: { content: 'first answer' } }], + }), + ) + .mockResolvedValueOnce( + makeJsonResponse({ + choices: [{ message: { content: 'second answer' } }], + }), + ); + const request = makeNextRequest('http://localhost/v1/responses', { + method: 'POST', + headers: { authorization: `Bearer ${accessKey.secret}` }, + }); + + const firstResponse = await handleResponsesRequest(request, { + input: 'first question', + model: 'hy3', + }); + const firstPayload = (await firstResponse.json()) as { id: string }; + + await addCredential( + { upstream_protocol: 'responses' }, + credential.filename, + ); + + const followUpResponse = await handleResponsesRequest(request, { + input: 'second question', + previous_response_id: firstPayload.id, + }); + expect(followUpResponse.status).toBe(200); + expect(String(fetchMock.mock.calls[1]?.[0])).toBe( + 'https://copilot.tencent.com/v2/chat/completions', + ); + expect( + JSON.parse(String((fetchMock.mock.calls[1]?.[1] as RequestInit).body)), + ).toMatchObject({ + messages: [ + { content: 'first question', role: 'user' }, + { content: 'first answer', role: 'assistant' }, + { content: 'second question', role: 'user' }, + ], + model: 'hy3', + }); + }); + it('covers proxy context helpers for saved credentials', async () => { const createdCredential = await addCredential({ bearer_token: 'token-from-bearer-token', @@ -2938,7 +3626,41 @@ describe('server units', () => { ); expect(resolved.auth.bearerToken).toBe('token-updated'); expect(resolved.preferences.firstMessageRoleToSystem).toBe(true); - expect(resolved.preferences.responsesPassthrough).toBe(false); + expect(resolved.preferences.upstreamProtocol).toBe('chat'); + }); + + it('keeps legacy and current upstream protocol fields synchronized', async () => { + const created = await addCredential({ + bearer_token: 'token-protocol-sync', + upstream_protocol: 'responses', + user_id: 'protocol-sync@example.com', + }); + + let record = (await readCredentialRecords()).find( + (credential) => credential.filename === created.filename, + ); + expect(record?.data).toMatchObject({ + responses_passthrough: true, + upstream_protocol: 'responses', + }); + + await addCredential({ upstream_protocol: 'chat' }, created.filename); + record = (await readCredentialRecords()).find( + (credential) => credential.filename === created.filename, + ); + expect(record?.data).toMatchObject({ + responses_passthrough: false, + upstream_protocol: 'chat', + }); + + await addCredential({ responses_passthrough: true }, created.filename); + record = (await readCredentialRecords()).find( + (credential) => credential.filename === created.filename, + ); + expect(record?.data).toMatchObject({ + responses_passthrough: true, + upstream_protocol: 'responses', + }); }); it('discovers models per credential without live upstream requests', async () => { From 1edbc5d88119638a78de9eb52698eddccbff9b07 Mon Sep 17 00:00:00 2001 From: orangeboyChen Date: Thu, 20 Aug 2026 18:23:54 +0800 Subject: [PATCH 2/3] test: cover responses protocol branches --- tests/server/anthropic.test.ts | 31 +++ tests/server/units.test.ts | 400 +++++++++++++++++++++++++++++++++ 2 files changed, 431 insertions(+) diff --git a/tests/server/anthropic.test.ts b/tests/server/anthropic.test.ts index 6ced151..ff92c45 100644 --- a/tests/server/anthropic.test.ts +++ b/tests/server/anthropic.test.ts @@ -168,6 +168,37 @@ describe('anthropic messages api', () => { }); }); + it('maps upstream failures to Anthropic error types', async () => { + vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(new Response('unauthorized', { status: 401 })) + .mockResolvedValueOnce(new Response('forbidden', { status: 403 })) + .mockResolvedValueOnce(new Response('missing', { status: 404 })) + .mockResolvedValueOnce(new Response('too large', { status: 413 })) + .mockResolvedValueOnce(new Response('overloaded', { status: 529 })) + .mockResolvedValueOnce(new Response('bad request', { status: 400 })); + const request = makeNextRequest('http://localhost/v1/messages', { + method: 'POST', + }); + const expected = [ + [401, 'authentication_error'], + [403, 'permission_error'], + [404, 'not_found_error'], + [413, 'request_too_large'], + [529, 'overloaded_error'], + [400, 'invalid_request_error'], + ] as const; + + for (const [status, type] of expected) { + const response = await handleMessagesRequest(request, { + model: 'hy3', + max_tokens: 32, + messages: [{ role: 'user', content: 'Fail' }], + }); + expect(response.status).toBe(status); + expect(await response.json()).toMatchObject({ error: { type } }); + } + }); + it('preserves Responses usage in Anthropic streams', async () => { const credential = await addCredential({ bearer_token: 'anthropic-responses-stream-token', diff --git a/tests/server/units.test.ts b/tests/server/units.test.ts index d987145..01e9e88 100644 --- a/tests/server/units.test.ts +++ b/tests/server/units.test.ts @@ -1456,6 +1456,289 @@ describe('server units', () => { }); }); + it('covers Responses upstream compatibility input variants', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-compatibility-token', + upstream_protocol: 'responses', + user_id: 'responses-compatibility@example.com', + }, + filePath: '/tmp/responses-compatibility.json', + filename: 'responses-compatibility.json', + }); + const upstreamPayload = { + incomplete_details: { reason: 'content_filter' }, + output: [ + null, + 1, + { type: 'other' }, + { type: 'function_call' }, + { content: null, type: 'message' }, + { + content: [ + null, + 1, + { text: 1, type: 'output_text' }, + { text: 'from output', type: 'output_text' }, + ], + type: 'message', + }, + { + content: [{ text: 'summary content' }], + summary: [null, 1, { text: 1 }, { text: 'summary' }], + type: 'reasoning', + }, + ], + status: 'incomplete', + usage: { input_tokens: 2, output_tokens: 3 }, + }; + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockImplementation(async () => makeJsonResponse(upstreamPayload)); + const request = async ( + body: Parameters[1], + ) => { + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + body, + context, + ); + await response.text(); + return response.status; + }; + + expect( + await request({ + messages: [ + { content: null, role: 'system' }, + { + content: [ + 'plain', + null, + { text: null }, + { unknown: true }, + { + image_url: 'https://example.com/string.png', + type: 'image_url', + }, + { + image_url: { url: 'https://example.com/object.png' }, + type: 'image_url', + }, + { + image_url: 'https://example.com/input.png', + type: 'input_image', + }, + { text: 'text part' }, + { image_url: { url: 1 }, type: 'image_url' }, + ], + role: 'user', + }, + ], + model: 'hy3', + response_format: { type: 'json_object' }, + thinking: { type: 'disabled' }, + tool_choice: 'auto', + tools: [ + null, + 1, + { name: 'direct_tool' }, + { function: { description: 'missing name' }, type: 'function' }, + ], + }), + ).toBe(200); + + expect( + await request({ + messages: [ + { + content: null, + role: 'assistant', + tool_calls: [ + null, + 1, + { function: {} }, + { function: { name: 'lookup' } }, + ], + }, + { content: { result: true }, role: 'tool', tool_call_id: 'call_1' }, + ], + model: 'hy3', + response_format: 1, + thinking: { budget_tokens: 1_000, type: 'adaptive' }, + tool_choice: 1, + }), + ).toBe(200); + + expect( + await request({ + messages: [ + { + content: 'assistant text', + role: 'assistant', + tool_calls: [ + { + function: { arguments: '{}', name: 'lookup' }, + id: 'call_lookup', + }, + ], + }, + ], + model: 'hy3', + response_format: { + json_schema: { description: 'schema', name: 'answer' }, + type: 'json_schema', + }, + thinking: { budget_tokens: 5_000, type: 'enabled' }, + tool_choice: { name: 'lookup', type: 'function' }, + }), + ).toBe(200); + + expect( + await request({ + messages: [{ content: undefined, role: 'user' }], + model: 'hy3', + response_format: { json_schema: {}, type: 'json_schema' }, + thinking: { budget_tokens: 10_000, type: 'adaptive' }, + tool_choice: { type: 'function' }, + }), + ).toBe(200); + + expect( + await request({ + messages: [{ content: 'reason', role: 'user' }], + model: 'hy3', + reasoning_effort: 'medium', + response_format: { type: 'text' }, + thinking: { type: 'enabled' }, + tool_choice: { type: 'required' }, + }), + ).toBe(200); + + expect( + await request({ + messages: [{ content: 'unsupported', role: 'user' }], + model: 'hy3', + thinking: { type: 'unknown' }, + }), + ).toBe(400); + expect( + await request({ + frequency_penalty: 0, + messages: [{ content: 'unsupported', role: 'user' }], + model: 'hy3', + presence_penalty: 0, + }), + ).toBe(400); + + expect(fetchMock).toHaveBeenCalledTimes(5); + const firstBody = JSON.parse( + String((fetchMock.mock.calls[0]?.[1] as RequestInit).body), + ) as Record; + expect(firstBody).toMatchObject({ + reasoning: { effort: 'none' }, + text: { format: { type: 'json_object' } }, + tool_choice: 'auto', + }); + }); + + it('covers Responses payload fallback and stop variants', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-payload-token', + upstream_protocol: 'responses', + user_id: 'responses-payload@example.com', + }, + filePath: '/tmp/responses-payload.json', + filename: 'responses-payload.json', + }); + vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + makeJsonResponse({ + output: [ + null, + 1, + { content: null, type: 'message' }, + { + content: [ + null, + 1, + { text: 1, type: 'output_text' }, + { text: 'zSTOPaEND', type: 'output_text' }, + ], + type: 'message', + }, + ], + status: 'incomplete', + }), + ) + .mockResolvedValueOnce( + makeJsonResponse({ + output: [ + { + arguments: undefined, + id: 'fc_fallback', + name: undefined, + type: 'function_call', + }, + ], + output_text: '', + status: 'completed', + usage: {}, + }), + ); + + const incompleteResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'fallback', role: 'user' }], + model: 'hy3', + stop: ['', 'END', 'STOP'], + }, + context, + ); + expect(await incompleteResponse.json()).toMatchObject({ + choices: [ + { + finish_reason: 'length', + message: { content: 'z' }, + }, + ], + usage: { completion_tokens: 0, prompt_tokens: 0, total_tokens: 0 }, + }); + + const toolResponse = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'tool fallback', role: 'user' }], + model: 'hy3', + }, + context, + ); + expect(await toolResponse.json()).toMatchObject({ + choices: [ + { + finish_reason: 'tool_calls', + message: { + content: null, + tool_calls: [ + { + function: { arguments: '', name: 'function' }, + id: 'fc_fallback', + }, + ], + }, + }, + ], + }); + }); + it('applies Chat stop sequences locally for the Responses upstream', async () => { const context = createProxyContextFromCredential({ data: { @@ -1709,6 +1992,123 @@ describe('server units', () => { ]); }); + it('covers Responses Chat stream terminal variants', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-stream-terminal-token', + upstream_protocol: 'responses', + user_id: 'responses-stream-terminal@example.com', + }, + filePath: '/tmp/responses-stream-terminal.json', + filename: 'responses-stream-terminal.json', + }); + vi.spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.reasoning_summary_text.delta"}\n\n' + + 'data: {"type":"response.reasoning_text.delta","delta":"reason"}\n\n' + + 'data: {"type":"response.output_item.added","item":{"type":"function_call"}}\n\n' + + 'data: {"type":"response.function_call_arguments.delta","output_index":0}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.output_text.delta","delta":"tailZ"}\n\n' + + 'data: {"type":"response.completed","response":{"usage":{}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.output_text.delta","delta":"partialQ"}\n\n' + + 'data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"content_filter"}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.error","response":{"error":"response error"}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response('data: {"type":"error","error":{}}\n\n', { + headers: { 'Content-Type': 'text/event-stream' }, + }), + ) + .mockResolvedValueOnce( + new Response('data: {"type":"response.failed"}\n\n', { + headers: { 'Content-Type': 'text/event-stream' }, + }), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":4,"output_tokens":3,"input_tokens_details":{"cached_tokens":1,"cache_creation_tokens":2},"output_tokens_details":{"reasoning_tokens":2}}}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ) + .mockResolvedValueOnce( + new Response( + 'data: {"type":"response.completed","response":{"usage":"invalid"}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + const stream = async ( + options: Partial[1]> = {}, + ): Promise => { + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Stream terminal', role: 'user' }], + model: 'hy3', + stream: true, + ...options, + }, + context, + ); + return response.text(); + }; + + const toolPayload = await stream(); + expect(toolPayload).toContain('"reasoning_content":""'); + expect(toolPayload).toContain('"reasoning_content":"reason"'); + expect(toolPayload).toContain('"name":"function"'); + expect(toolPayload).toContain('"finish_reason":"tool_calls"'); + + const completedPayload = await stream({ + stop: 'ZZ', + stream_options: { include_usage: true }, + }); + expect(completedPayload).toContain('"content":"tail"'); + expect(completedPayload).toContain('"content":"Z"'); + expect(completedPayload).toContain('"total_tokens":0'); + + const incompletePayload = await stream({ stop: 'QQ' }); + expect(incompletePayload).toContain('"content":"partial"'); + expect(incompletePayload).toContain('"content":"Q"'); + expect(incompletePayload).toContain('"finish_reason":"content_filter"'); + + expect(await stream()).toContain('response error'); + expect(await stream()).toContain('[object Object]'); + expect(await stream()).toContain('Upstream Responses stream failed'); + + const detailedUsagePayload = await stream({ + stream_options: { include_usage: true }, + }); + expect(detailedUsagePayload).toContain('"cached_tokens":1'); + expect(detailedUsagePayload).toContain('"cache_creation_tokens":2'); + expect(detailedUsagePayload).toContain('"reasoning_tokens":2'); + + const invalidUsagePayload = await stream({ + stream_options: { include_usage: true }, + }); + expect(invalidUsagePayload).not.toContain('"usage":'); + expect(invalidUsagePayload).toContain('data: [DONE]'); + }); + it('maps Responses function call events back to Chat Completions SSE', async () => { const context = createProxyContextFromCredential({ data: { From 79311c7029af45efa3711341ef60dbc655f7e315 Mon Sep 17 00:00:00 2001 From: orangeboyChen Date: Thu, 20 Aug 2026 19:38:28 +0800 Subject: [PATCH 3/3] fix: address responses stream review feedback --- lib/server/domain/usage.ts | 11 ++- lib/server/proxy/codebuddy.ts | 31 +++++---- tests/server/units.test.ts | 123 +++++++++++++++++++++++++++++++++- 3 files changed, 147 insertions(+), 18 deletions(-) diff --git a/lib/server/domain/usage.ts b/lib/server/domain/usage.ts index 75fac45..d3a6c9b 100644 --- a/lib/server/domain/usage.ts +++ b/lib/server/domain/usage.ts @@ -22,6 +22,10 @@ export type UsageRange = export interface UsageSnapshot { cache_creation_input_tokens?: number | null; cache_read_input_tokens?: number | null; + input_tokens_details?: { + cached_tokens?: number | null; + cache_creation_tokens?: number | null; + } | null; prompt_cache_hit_tokens?: number | null; prompt_cache_miss_tokens?: number | null; prompt_cache_write_tokens?: number | null; @@ -155,14 +159,17 @@ const getLargestTokenCount = (...values: unknown[]): number => { const normalizeUsage = (usage: UsageSnapshot): UsageEventRecord => { const inputTokens = toNumber(usage.input_tokens ?? usage.prompt_tokens); const outputTokens = toNumber(usage.output_tokens ?? usage.completion_tokens); + const inputTokenDetails = usage.input_tokens_details; const promptTokenDetails = usage.prompt_tokens_details; const cacheReadTokens = getLargestTokenCount( usage.cache_read_input_tokens, + inputTokenDetails?.cached_tokens, promptTokenDetails?.cached_tokens, usage.prompt_cache_hit_tokens, ); const cacheCreationTokens = getLargestTokenCount( usage.cache_creation_input_tokens, + inputTokenDetails?.cache_creation_tokens, promptTokenDetails?.cache_creation_tokens, usage.prompt_cache_write_tokens, ); @@ -171,7 +178,9 @@ const normalizeUsage = (usage: UsageSnapshot): UsageEventRecord => { explicitTotal || inputTokens + outputTokens + - (promptTokenDetails ? 0 : cacheReadTokens + cacheCreationTokens); + (inputTokenDetails || promptTokenDetails + ? 0 + : cacheReadTokens + cacheCreationTokens); return { accessKeyId: null, diff --git a/lib/server/proxy/codebuddy.ts b/lib/server/proxy/codebuddy.ts index 6ef7d0f..0408657 100644 --- a/lib/server/proxy/codebuddy.ts +++ b/lib/server/proxy/codebuddy.ts @@ -314,9 +314,9 @@ const trackResponsesUsageStream = async ({ }); } }; - const bindResponseId = (id: string): void => { - if (!onResponseId || responseBinding) return; - responseBinding = onResponseId(id).catch((error) => { + const bindResponseId = (id: string): Promise => { + if (!onResponseId) return Promise.resolve(); + responseBinding ??= onResponseId(id).catch((error) => { console.error( '[CodeBuddy2API] Failed to bind upstream Responses session', { @@ -325,6 +325,7 @@ const trackResponsesUsageStream = async ({ }, ); }); + return responseBinding; }; const stream = new ReadableStream({ start: (controller) => { @@ -333,27 +334,27 @@ const trackResponsesUsageStream = async ({ let buffer = ''; let responseId: string | null = null; - const inspectFrame = (frame: string): void => { - frame.split('\n').forEach((line) => { + const inspectFrame = async (frame: string): Promise => { + for (const line of frame.split('\n')) { if (!line.startsWith('data:')) { - return; + continue; } const raw = line.slice(5).trim(); if (!raw || raw === '[DONE]') { - return; + continue; } try { const event = JSON.parse(raw) as unknown; latestUsage = extractResponsesUsage(event) ?? latestUsage; responseId = extractResponsesId(event) ?? responseId; - if (responseId) bindResponseId(responseId); + if (responseId) await bindResponseId(responseId); } catch { // Preserve malformed upstream frames without recording them. } - }); + } }; const pump = async (): Promise => { @@ -366,7 +367,7 @@ const trackResponsesUsageStream = async ({ if (done) { if (buffer) { - inspectFrame(buffer); + await inspectFrame(buffer); controller.enqueue(encoder.encode(buffer)); } @@ -385,13 +386,14 @@ const trackResponsesUsageStream = async ({ buffer = ''; } - frames.forEach((frame) => { + for (const frame of frames) { if (frame.length > MAX_STREAM_FRAME_LENGTH) { - return; + continue; } - inspectFrame(frame); + await inspectFrame(frame); + if (cancelled) return; controller.enqueue(encoder.encode(`${frame}\n\n`)); - }); + } } }; @@ -1725,6 +1727,7 @@ const mapResponsesStreamToChat = ( }, async cancel(reason) { await cancelAndReleaseReader(reason); + await recordStreamUsage(); }, }); diff --git a/tests/server/units.test.ts b/tests/server/units.test.ts index 01e9e88..9dea633 100644 --- a/tests/server/units.test.ts +++ b/tests/server/units.test.ts @@ -1351,7 +1351,15 @@ describe('server units', () => { id: 'resp_123', output: [], output_text: 'hello from responses', - usage: { input_tokens: 2, output_tokens: 3, total_tokens: 5 }, + usage: { + input_tokens: 2, + input_tokens_details: { + cached_tokens: 1, + cache_creation_tokens: 1, + }, + output_tokens: 3, + total_tokens: 5, + }, }), ); @@ -1454,6 +1462,21 @@ describe('server units', () => { ], top_p: 0.8, }); + expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ + { + callCount: 1, + cacheHitTokens: 1, + model: 'hy3', + totalTokens: 5, + }, + ]); + const usageStore = JSON.parse( + fs.readFileSync(path.join(tempDataDir, 'usage-history.json'), 'utf8'), + ) as { events: Array> }; + expect(usageStore.events[0]).toMatchObject({ + cacheCreationTokens: 1, + cacheReadTokens: 1, + }); }); it('covers Responses upstream compatibility input variants', async () => { @@ -1992,6 +2015,56 @@ describe('server units', () => { ]); }); + it('records Responses usage when a Chat stream is cancelled', async () => { + const context = createProxyContextFromCredential({ + data: { + bearer_token: 'responses-cancel-usage-token', + upstream_protocol: 'responses', + user_id: 'responses-cancel-usage@example.com', + }, + filePath: '/tmp/responses-cancel-usage.json', + filename: 'responses-cancel-usage.json', + }); + const cancel = vi.fn(); + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response(new ReadableStream({ cancel }), { + headers: { + 'Content-Type': 'text/event-stream', + 'x-codebuddy-usage': JSON.stringify({ + input_tokens: 6, + input_tokens_details: { cached_tokens: 4 }, + output_tokens: 2, + total_tokens: 8, + }), + }, + }), + ); + + const response = await proxyChatCompletions( + makeNextRequest('http://localhost/v1/chat/completions', { + method: 'POST', + }), + { + messages: [{ content: 'Cancel this stream', role: 'user' }], + model: 'hy3', + stream: true, + }, + context, + ); + + await response.body?.cancel('client disconnected'); + + expect(cancel).toHaveBeenCalledWith('client disconnected'); + expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ + { + callCount: 1, + cacheHitTokens: 4, + model: 'hy3', + totalTokens: 8, + }, + ]); + }); + it('covers Responses Chat stream terminal variants', async () => { const context = createProxyContextFromCredential({ data: { @@ -3328,6 +3401,10 @@ describe('server units', () => { response: { usage: { input_tokens: 4, + input_tokens_details: { + cached_tokens: 1, + cache_creation_tokens: 1, + }, output_tokens: 2, total_tokens: 6, }, @@ -3341,7 +3418,7 @@ describe('server units', () => { 'data: {"type":"response.created","response":{"id":"resp_1"}}', '', 'event: response.completed', - 'data: {"type":"response.completed","response":{"usage":{"input_tokens":5,"output_tokens":3,"total_tokens":8}}}', + 'data: {"type":"response.completed","response":{"usage":{"input_tokens":5,"input_tokens_details":{"cached_tokens":2,"cache_creation_tokens":1},"output_tokens":3,"total_tokens":8}}}', '', ].join('\n'), { @@ -3396,7 +3473,7 @@ describe('server units', () => { expect((await getUsageAnalytics({ range: 'today' })).tableRows).toEqual([ { callCount: 3, - cacheHitTokens: 0, + cacheHitTokens: 3, model: 'gpt-5.5', totalTokens: 14, }, @@ -3404,6 +3481,46 @@ describe('server units', () => { }); }); + it('waits for streamed response bindings before forwarding response ids', async () => { + let resolveBinding: (() => void) | undefined; + const binding = new Promise((resolve) => { + resolveBinding = resolve; + }); + const onResponseId = vi.fn(async () => binding); + vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response( + 'data: {"type":"response.created","response":{"id":"resp_delayed_binding"}}\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ), + ); + + const response = await proxyResponsesUpstream( + makeNextRequest('http://localhost/v1/responses', { method: 'POST' }), + { input: 'bind before forwarding', model: 'gpt-5.5', stream: true }, + undefined, + undefined, + onResponseId, + ); + const reader = response.body!.getReader(); + let readSettled = false; + const readPromise = reader.read().then((result) => { + readSettled = true; + return result; + }); + + await waitForAsync(async () => { + expect(onResponseId).toHaveBeenCalledWith('resp_delayed_binding'); + }); + expect(readSettled).toBe(false); + + resolveBinding?.(); + const firstChunk = await readPromise; + expect(new TextDecoder().decode(firstChunk.value)).toContain( + 'resp_delayed_binding', + ); + expect((await reader.read()).done).toBe(true); + }); + it('cancels the upstream Responses stream when the client disconnects', async () => { const cancel = vi.fn(); vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce(