diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts index 2085bd6..e1f5f21 100644 --- a/apps/worker/src/env.ts +++ b/apps/worker/src/env.ts @@ -6,8 +6,7 @@ import type { export type Bindings = { DB: D1Database; CORS_ORIGIN?: string; - PROXY_RETRY_ROUNDS?: string; - PROXY_RETRY_DELAY_MS?: string; + PROXY_UPSTREAM_TIMEOUT_MS?: string; CHECKIN_SCHEDULER: DurableObjectNamespace; }; @@ -21,3 +20,4 @@ export type AppEnv = { Bindings: Bindings; Variables: Variables; }; + diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 70cde2c..5d6f411 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -1,6 +1,5 @@ import { Hono } from "hono"; import { cors } from "hono/cors"; -import { logger } from "hono/logger"; import type { AppEnv } from "./env"; import { adminAuth } from "./middleware/adminAuth"; import authRoutes from "./routes/auth"; @@ -17,46 +16,6 @@ import tokenRoutes from "./routes/tokens"; import usageRoutes from "./routes/usage"; const app = new Hono({ strict: false }); - -app.use("*", async (c, next) => { - const method = c.req.method; - const path = c.req.path; - const contentType = c.req.header("content-type") ?? ""; - if (["POST", "PUT", "PATCH", "DELETE"].includes(method)) { - const raw = c.req.raw.clone(); - let bodyType = "unknown"; - let bodyKeys: string[] = []; - let bodySize = 0; - try { - const payload = await raw.json(); - bodyType = "json"; - if (payload && typeof payload === "object" && !Array.isArray(payload)) { - bodyKeys = Object.keys(payload); - } - } catch { - try { - const text = await raw.text(); - bodyType = "text"; - bodySize = text.length; - } catch { - bodyType = "unreadable"; - } - } - console.log("[request]", { - method, - path, - content_type: contentType, - body_type: bodyType, - body_keys: bodyKeys, - body_size: bodySize, - }); - } else { - console.log("[request]", { method, path }); - } - await next(); -}); - -app.use("*", logger()); app.use( "/api/*", cors({ @@ -122,6 +81,28 @@ app.route("/api/group", newapiGroupRoutes); app.route("/v1", proxyRoutes); app.route("/v1beta", proxyRoutes); +app.onError((err, c) => { + console.error("[app:error]", { + method: c.req.method, + path: c.req.path, + message: err.message, + stack: err.stack, + }); + + if ( + c.req.path === "/api" || + c.req.path.startsWith("/api/") || + c.req.path === "/v1" || + c.req.path.startsWith("/v1/") || + c.req.path === "/v1beta" || + c.req.path.startsWith("/v1beta/") + ) { + return c.json({ error: "Internal Server Error" }, 500); + } + + return c.text("Internal Server Error", 500); +}); + app.notFound(async (c) => { const path = c.req.path; if ( diff --git a/apps/worker/src/routes/proxy.ts b/apps/worker/src/routes/proxy.ts index 3f94ca3..b230d32 100644 --- a/apps/worker/src/routes/proxy.ts +++ b/apps/worker/src/routes/proxy.ts @@ -7,6 +7,7 @@ import { } from "../services/call-token-selector"; import { listCallTokens } from "../services/channel-call-token-repo"; import { + type ChannelMetadata, parseChannelMetadata, resolveMappedModel, resolveProvider, @@ -44,6 +45,7 @@ import { parseUsageFromJson, parseUsageFromSse, } from "../utils/usage"; +import { adaptChatResponse } from "../services/chat-response-adapter"; const proxy = new Hono(); @@ -66,6 +68,54 @@ function channelSupportsModel( return Boolean(metadata.model_mapping[model] ?? metadata.model_mapping["*"]); } +export function selectCandidateChannels( + allowedChannels: ChannelRecord[], + downstreamModel: string | null, +): ChannelRecord[] { + const modelChannels = allowedChannels.filter((channel) => + channelSupportsModel(channel, downstreamModel), + ); + if (modelChannels.length > 0) { + return modelChannels; + } + // Fallback to all allowed channels. Downstream protocol should not restrict + // upstream channel selection. + return allowedChannels; +} + +function hasExplicitModelMapping( + metadata: ChannelMetadata, + downstreamModel: string | null, +): boolean { + if (downstreamModel) { + return ( + metadata.model_mapping[downstreamModel] !== undefined || + metadata.model_mapping["*"] !== undefined + ); + } + return metadata.model_mapping["*"] !== undefined; +} + +export function resolveUpstreamModelForChannel( + channel: ChannelRecord, + metadata: ChannelMetadata, + downstreamModel: string | null, +): { model: string | null; autoMapped: boolean } { + const mapped = resolveMappedModel(metadata.model_mapping, downstreamModel); + if (!downstreamModel || hasExplicitModelMapping(metadata, downstreamModel)) { + return { model: mapped, autoMapped: false }; + } + + const declaredModels = extractModels(channel).map((entry) => entry.id); + if (declaredModels.length === 0) { + return { model: mapped, autoMapped: false }; + } + if (declaredModels.includes(downstreamModel)) { + return { model: downstreamModel, autoMapped: false }; + } + return { model: declaredModels[0] ?? mapped, autoMapped: true }; +} + function filterAllowedChannels( channels: ChannelRecord[], tokenRecord: TokenRecord, @@ -81,35 +131,6 @@ function filterAllowedChannels( return channels.filter((channel) => allowedSet.has(channel.id)); } -/** - * Determines whether a response status should be retried. - * - * Args: - * status: HTTP response status code. - * - * Returns: - * True if the status is retryable. - */ -function isRetryableStatus(status: number): boolean { - return status === 408 || status === 429 || status >= 500; -} - -/** - * Waits before the next retry round. - * - * Args: - * ms: Delay in milliseconds. - * - * Returns: - * Promise resolved after delay. - */ -async function sleep(ms: number): Promise { - if (ms <= 0) { - return; - } - await new Promise((resolve) => setTimeout(resolve, ms)); -} - function resolveChannelBaseUrl(channel: ChannelRecord): string { return normalizeBaseUrl(channel.base_url); } @@ -163,6 +184,26 @@ function buildUpstreamHeaders( return headers; } +async function fetchWithTimeout( + url: string, + init: RequestInit, + timeoutMs: number, +): Promise { + if (timeoutMs <= 0) { + return fetch(url, init); + } + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + return await fetch(url, { + ...init, + signal: controller.signal, + }); + } finally { + clearTimeout(timer); + } +} + /** * Multi-provider proxy handler. */ @@ -254,11 +295,16 @@ proxy.all("/*", tokenAuth, async (c) => { callTokenMap.set(row.channel_id, list); } const allowedChannels = filterAllowedChannels(activeChannels, tokenRecord); - const modelChannels = allowedChannels.filter((channel) => - channelSupportsModel(channel, downstreamModel), - ); - const candidates = modelChannels.length > 0 ? modelChannels : allowedChannels; + const candidates = selectCandidateChannels(allowedChannels, downstreamModel); + if (candidates.length === 0 && allowedChannels.length > 0) { + console.warn("[proxy:no_compatible_channels]", { + path: c.req.path, + model: downstreamModel, + downstream_provider: downstreamProvider, + allowed_channels: allowedChannels.length, + }); + } if (candidates.length === 0) { return jsonError(c, 503, "no_available_channels", "no_available_channels"); } @@ -268,148 +314,80 @@ proxy.all("/*", tokenAuth, async (c) => { const querySuffix = c.req.url.includes("?") ? `?${c.req.url.split("?")[1]}` : ""; - const retryRounds = Math.max(1, Number(c.env.PROXY_RETRY_ROUNDS ?? "1")); - const retryDelayMs = Math.max(0, Number(c.env.PROXY_RETRY_DELAY_MS ?? "200")); + const upstreamTimeoutMs = Math.max( + 1000, + Number(c.env.PROXY_UPSTREAM_TIMEOUT_MS ?? "30000"), + ); let lastResponse: Response | null = null; let lastChannel: ChannelRecord | null = null; let lastRequestPath = targetPath; const start = Date.now(); let selectedChannel: ChannelRecord | null = null; - - let round = 0; - while (round < retryRounds && !selectedChannel) { - let shouldRetry = false; - for (const channel of ordered) { - lastChannel = channel; - const metadata = parseChannelMetadata(channel.metadata_json); - const upstreamProvider = resolveProvider(metadata.site_type); - const upstreamModel = resolveMappedModel( - metadata.model_mapping, - downstreamModel, - ); - if ( - upstreamProvider === "gemini" && - !upstreamModel && - endpointType !== "passthrough" - ) { + let selectedUpstreamProvider: ProviderType | null = null; + let selectedUpstreamModel: string | null = null; + for (const channel of ordered) { + lastChannel = channel; + const metadata = parseChannelMetadata(channel.metadata_json); + const upstreamProvider = resolveProvider(metadata.site_type); + const resolvedModel = resolveUpstreamModelForChannel( + channel, + metadata, + downstreamModel, + ); + const upstreamModel = resolvedModel.model; + if ( + upstreamProvider === "gemini" && + !upstreamModel && + endpointType !== "passthrough" + ) { + continue; + } + const baseUrl = resolveChannelBaseUrl(channel); + const tokens = callTokenMap.get(channel.id) ?? []; + const selectedToken = tokens.length > 0 ? selectCallToken(tokens) : null; + const apiKey = selectedToken?.api_key ?? channel.api_key; + const headers = buildUpstreamHeaders( + new Headers(c.req.header()), + upstreamProvider, + String(apiKey), + metadata.header_overrides, + ); + headers.delete("host"); + headers.delete("content-length"); + let upstreamRequestPath = targetPath; + let upstreamFallbackPath: string | undefined; + let upstreamBodyText = requestText || undefined; + let absoluteUrl: string | undefined; + const sameProvider = upstreamProvider === downstreamProvider; + if (endpointType === "passthrough") { + if (!sameProvider) { continue; } - const baseUrl = resolveChannelBaseUrl(channel); - const tokens = callTokenMap.get(channel.id) ?? []; - const selectedToken = tokens.length > 0 ? selectCallToken(tokens) : null; - const apiKey = selectedToken?.api_key ?? channel.api_key; - const headers = buildUpstreamHeaders( - new Headers(c.req.header()), - upstreamProvider, - String(apiKey), - metadata.header_overrides, - ); - headers.delete("host"); - headers.delete("content-length"); - let upstreamRequestPath = targetPath; - let upstreamFallbackPath: string | undefined; - let upstreamBodyText = requestText || undefined; - let absoluteUrl: string | undefined; - const sameProvider = upstreamProvider === downstreamProvider; - if (endpointType === "passthrough") { - if (!sameProvider) { - continue; - } - if (upstreamProvider === "gemini") { - upstreamRequestPath = applyGeminiModelToPath( - upstreamRequestPath, - upstreamModel, - ); - } else if (upstreamModel && parsedBody) { - upstreamBodyText = JSON.stringify({ - ...parsedBody, - model: upstreamModel, - }); - } - } else if (sameProvider && parsedBody) { - if (upstreamProvider === "gemini") { - upstreamRequestPath = applyGeminiModelToPath( - upstreamRequestPath, - upstreamModel, - ); - } else if (upstreamModel) { - upstreamBodyText = JSON.stringify({ - ...parsedBody, - model: upstreamModel, - }); - } - if (endpointType === "chat" || endpointType === "responses") { - if (metadata.endpoint_overrides.chat_url && normalizedChat) { - const request = buildUpstreamChatRequest( - upstreamProvider, - normalizedChat, - upstreamModel, - endpointType, - isStream, - metadata.endpoint_overrides, - ); - if (request) { - upstreamRequestPath = request.path; - absoluteUrl = request.absoluteUrl; - upstreamFallbackPath = request.absoluteUrl - ? undefined - : request.fallbackPath; - } - } else if ( - endpointType === "responses" && - upstreamProvider === "openai" - ) { - upstreamFallbackPath = "/responses"; - } - } - if ( - endpointType === "embeddings" && - metadata.endpoint_overrides.embedding_url - ) { - if (normalizedEmbedding) { - const request = buildUpstreamEmbeddingRequest( - upstreamProvider, - normalizedEmbedding, - upstreamModel, - metadata.endpoint_overrides, - ); - if (request) { - upstreamRequestPath = request.path; - absoluteUrl = request.absoluteUrl; - } - } - } - if ( - endpointType === "images" && - metadata.endpoint_overrides.image_url - ) { - if (normalizedImage) { - const request = buildUpstreamImageRequest( - upstreamProvider, - normalizedImage, - upstreamModel, - metadata.endpoint_overrides, - ); - if (request) { - upstreamRequestPath = request.path; - absoluteUrl = request.absoluteUrl; - } - } - } - } else { - let built: { - request: { - path: string; - fallbackPath?: string; - absoluteUrl?: string; - body: Record | null; - }; - bodyText?: string; - } | null = null; - if (endpointType === "chat" || endpointType === "responses") { - if (!normalizedChat) { - return jsonError(c, 400, "invalid_body", "invalid_body"); - } + if (upstreamProvider === "gemini") { + upstreamRequestPath = applyGeminiModelToPath( + upstreamRequestPath, + upstreamModel, + ); + } else if (upstreamModel && parsedBody) { + upstreamBodyText = JSON.stringify({ + ...parsedBody, + model: upstreamModel, + }); + } + } else if (sameProvider && parsedBody) { + if (upstreamProvider === "gemini") { + upstreamRequestPath = applyGeminiModelToPath( + upstreamRequestPath, + upstreamModel, + ); + } else if (upstreamModel) { + upstreamBodyText = JSON.stringify({ + ...parsedBody, + model: upstreamModel, + }); + } + if (endpointType === "chat" || endpointType === "responses") { + if (metadata.endpoint_overrides.chat_url && normalizedChat) { const request = buildUpstreamChatRequest( upstreamProvider, normalizedChat, @@ -418,121 +396,214 @@ proxy.all("/*", tokenAuth, async (c) => { isStream, metadata.endpoint_overrides, ); - if (!request) { - continue; - } - built = { - request, - bodyText: request.body ? JSON.stringify(request.body) : undefined, - }; - } else if (endpointType === "embeddings") { - if (!normalizedEmbedding) { - return jsonError(c, 400, "invalid_body", "invalid_body"); + if (request) { + upstreamRequestPath = request.path; + absoluteUrl = request.absoluteUrl; + upstreamFallbackPath = request.absoluteUrl + ? undefined + : request.fallbackPath; } + } else if ( + endpointType === "responses" && + upstreamProvider === "openai" + ) { + upstreamFallbackPath = "/responses"; + } + } + if ( + endpointType === "embeddings" && + metadata.endpoint_overrides.embedding_url + ) { + if (normalizedEmbedding) { const request = buildUpstreamEmbeddingRequest( upstreamProvider, normalizedEmbedding, upstreamModel, metadata.endpoint_overrides, ); - if (!request) { - continue; - } - built = { - request, - bodyText: request.body ? JSON.stringify(request.body) : undefined, - }; - } else if (endpointType === "images") { - if (!normalizedImage) { - return jsonError(c, 400, "invalid_body", "invalid_body"); + if (request) { + upstreamRequestPath = request.path; + absoluteUrl = request.absoluteUrl; } + } + } + if ( + endpointType === "images" && + metadata.endpoint_overrides.image_url + ) { + if (normalizedImage) { const request = buildUpstreamImageRequest( upstreamProvider, normalizedImage, upstreamModel, metadata.endpoint_overrides, ); - if (!request) { - continue; + if (request) { + upstreamRequestPath = request.path; + absoluteUrl = request.absoluteUrl; } - built = { - request, - bodyText: request.body ? JSON.stringify(request.body) : undefined, - }; } - if (!built) { + } + } else { + let built: { + request: { + path: string; + fallbackPath?: string; + absoluteUrl?: string; + body: Record | null; + }; + bodyText?: string; + } | null = null; + if (endpointType === "chat" || endpointType === "responses") { + if (!normalizedChat) { + return jsonError(c, 400, "invalid_body", "invalid_body"); + } + const request = buildUpstreamChatRequest( + upstreamProvider, + normalizedChat, + upstreamModel, + endpointType, + isStream, + metadata.endpoint_overrides, + ); + if (!request) { continue; } - upstreamRequestPath = built.request.path; - absoluteUrl = built.request.absoluteUrl; - upstreamFallbackPath = built.request.absoluteUrl - ? undefined - : built.request.fallbackPath; - upstreamBodyText = built.bodyText; + built = { + request, + bodyText: request.body ? JSON.stringify(request.body) : undefined, + }; + } else if (endpointType === "embeddings") { + if (!normalizedEmbedding) { + return jsonError(c, 400, "invalid_body", "invalid_body"); + } + const request = buildUpstreamEmbeddingRequest( + upstreamProvider, + normalizedEmbedding, + upstreamModel, + metadata.endpoint_overrides, + ); + if (!request) { + continue; + } + built = { + request, + bodyText: request.body ? JSON.stringify(request.body) : undefined, + }; + } else if (endpointType === "images") { + if (!normalizedImage) { + return jsonError(c, 400, "invalid_body", "invalid_body"); + } + const request = buildUpstreamImageRequest( + upstreamProvider, + normalizedImage, + upstreamModel, + metadata.endpoint_overrides, + ); + if (!request) { + continue; + } + built = { + request, + bodyText: request.body ? JSON.stringify(request.body) : undefined, + }; } - const targetBase = absoluteUrl ?? `${baseUrl}${upstreamRequestPath}`; - const target = mergeQuery( - targetBase, - querySuffix, - metadata.query_overrides, - ); + if (!built) { + continue; + } + upstreamRequestPath = built.request.path; + absoluteUrl = built.request.absoluteUrl; + upstreamFallbackPath = built.request.absoluteUrl + ? undefined + : built.request.fallbackPath; + upstreamBodyText = built.bodyText; + } + const targetBase = absoluteUrl ?? `${baseUrl}${upstreamRequestPath}`; + const target = mergeQuery( + targetBase, + querySuffix, + metadata.query_overrides, + ); - try { - let response = await fetch(target, { + try { + let response = await fetchWithTimeout( + target, + { method: c.req.method, headers, body: upstreamBodyText || undefined, - }); - let responsePath = upstreamRequestPath; - if ( - (response.status === 400 || response.status === 404) && - upstreamFallbackPath - ) { - const fallbackTargetBase = absoluteUrl - ? absoluteUrl - : `${baseUrl}${upstreamFallbackPath}`; - const fallbackTarget = mergeQuery( - fallbackTargetBase, - querySuffix, - metadata.query_overrides, - ); - response = await fetch(fallbackTarget, { + }, + upstreamTimeoutMs, + ); + let responsePath = upstreamRequestPath; + if ( + (response.status === 400 || response.status === 404) && + upstreamFallbackPath + ) { + const fallbackTargetBase = absoluteUrl + ? absoluteUrl + : `${baseUrl}${upstreamFallbackPath}`; + const fallbackTarget = mergeQuery( + fallbackTargetBase, + querySuffix, + metadata.query_overrides, + ); + response = await fetchWithTimeout( + fallbackTarget, + { method: c.req.method, headers, body: upstreamBodyText || undefined, - }); - responsePath = upstreamFallbackPath; - } - - lastResponse = response; - lastRequestPath = responsePath; - if (response.ok) { - selectedChannel = channel; - break; - } - - if (isRetryableStatus(response.status)) { - shouldRetry = true; - } - } catch { - lastResponse = null; - shouldRetry = true; + }, + upstreamTimeoutMs, + ); + responsePath = upstreamFallbackPath; } - } - if (selectedChannel || !shouldRetry) { - break; - } - - round += 1; - if (round < retryRounds) { - await sleep(retryDelayMs); + lastResponse = response; + lastRequestPath = responsePath; + if (response.ok) { + selectedChannel = channel; + selectedUpstreamProvider = upstreamProvider; + selectedUpstreamModel = upstreamModel; + break; + } + } catch (error) { + const isTimeout = + error instanceof Error && + (error.name === "AbortError" || error.message.includes("upstream_timeout")); + console.error("[proxy:upstream_exception]", { + channel_id: channel.id, + upstream_provider: upstreamProvider, + path: upstreamRequestPath, + model: downstreamModel, + upstream_model: upstreamModel, + timeout_ms: upstreamTimeoutMs, + reason: isTimeout ? "timeout" : "exception", + error: error instanceof Error ? error.message : String(error), + }); + lastResponse = null; } } const latencyMs = Date.now() - start; + if (!selectedChannel && lastResponse && !lastResponse.ok) { + console.warn("[proxy:upstream_exhausted]", { + path: targetPath, + model: downstreamModel, + status: lastResponse.status, + last_channel_id: lastChannel?.id ?? null, + last_request_path: lastRequestPath, + }); + } if (!lastResponse) { + console.error("[proxy:unavailable]", { + path: targetPath, + model: downstreamModel, + latency_ms: latencyMs, + last_channel_id: lastChannel?.id ?? null, + }); await recordUsage(c.env.DB, { tokenId: tokenRecord.id, model: downstreamModel, @@ -576,23 +647,6 @@ proxy.all("/*", tokenAuth, async (c) => { status: lastResponse.ok ? "ok" : "error", }); }; - const logUsage = ( - label: string, - usage: NormalizedUsage | null, - source: string, - ) => { - console.log(`[usage] ${label}`, { - source, - total_tokens: usage?.totalTokens ?? 0, - prompt_tokens: usage?.promptTokens ?? 0, - completion_tokens: usage?.completionTokens ?? 0, - stream: isStream, - status: lastResponse.status, - model: downstreamModel, - path: targetPath, - }); - }; - const headerUsage = parseUsageFromHeaders(lastResponse.headers); let jsonUsage: NormalizedUsage | null = null; if ( @@ -607,11 +661,6 @@ proxy.all("/*", tokenAuth, async (c) => { jsonUsage = parseUsageFromJson(data); } const immediateUsage = jsonUsage ?? headerUsage; - const immediateSource = jsonUsage - ? "json" - : headerUsage - ? "header" - : "none"; if (isStream) { const executionCtx = (c as { executionCtx?: ExecutionContextLike }) @@ -619,12 +668,6 @@ proxy.all("/*", tokenAuth, async (c) => { const task = parseUsageFromSse(lastResponse.clone()) .then((streamUsage) => { const usageValue = immediateUsage ?? streamUsage.usage; - const source = immediateUsage - ? immediateSource - : streamUsage.usage - ? "sse" - : "sse-none"; - logUsage("stream", usageValue, source); return record(usageValue, streamUsage.firstTokenLatencyMs); }) .catch(() => undefined); @@ -634,11 +677,23 @@ proxy.all("/*", tokenAuth, async (c) => { task.catch(() => undefined); } } else { - logUsage("immediate", immediateUsage, immediateSource); await record(immediateUsage, latencyMs); } } + if (selectedUpstreamProvider && endpointType === "chat") { + const transformed = await adaptChatResponse({ + response: lastResponse, + upstreamProvider: selectedUpstreamProvider, + downstreamProvider, + model: selectedUpstreamModel ?? downstreamModel, + isStream, + }); + if (transformed !== lastResponse) { + return transformed; + } + } + return lastResponse; }); diff --git a/apps/worker/src/services/chat-response-adapter.ts b/apps/worker/src/services/chat-response-adapter.ts new file mode 100644 index 0000000..4e10d8b --- /dev/null +++ b/apps/worker/src/services/chat-response-adapter.ts @@ -0,0 +1,1198 @@ +import { safeJsonParse } from "../utils/json"; +import type { ProviderType } from "./provider-transform"; + +type AdaptOptions = { + response: Response; + upstreamProvider: ProviderType; + downstreamProvider: ProviderType; + model: string | null; + isStream: boolean; +}; + +type AdapterFn = (options: AdaptOptions) => Promise; + +function mapOpenAiFinishReasonToAnthropic( + reason: unknown, +): "end_turn" | "max_tokens" | "tool_use" | "stop_sequence" | null { + const normalized = typeof reason === "string" ? reason : ""; + if (normalized === "stop") { + return "end_turn"; + } + if (normalized === "length") { + return "max_tokens"; + } + if (normalized === "tool_calls" || normalized === "function_call") { + return "tool_use"; + } + if (normalized === "stop_sequence") { + return "stop_sequence"; + } + return null; +} + +function mapAnthropicStopReasonToOpenAi( + reason: unknown, +): "stop" | "length" | "tool_calls" | "stop_sequence" | null { + const normalized = typeof reason === "string" ? reason : ""; + if (normalized === "end_turn") { + return "stop"; + } + if (normalized === "max_tokens") { + return "length"; + } + if (normalized === "tool_use") { + return "tool_calls"; + } + if (normalized === "stop_sequence") { + return "stop_sequence"; + } + return null; +} + +function openAiContentToText(content: unknown): string { + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .map((part) => { + if (typeof part === "string") { + return part; + } + if (part && typeof part === "object") { + const record = part as Record; + if (typeof record.text === "string") { + return record.text; + } + } + return ""; + }) + .join(""); +} + +function anthropicContentToText(content: unknown): string { + if (!Array.isArray(content)) { + return ""; + } + return content + .map((part) => { + if (!part || typeof part !== "object") { + return ""; + } + const record = part as Record; + if (record.type === "text" && typeof record.text === "string") { + return record.text; + } + return ""; + }) + .join(""); +} + +function geminiCandidateText(payload: Record): string { + const candidates = Array.isArray(payload.candidates) + ? (payload.candidates as Record[]) + : []; + const firstCandidate = candidates[0] ?? {}; + const content = + firstCandidate.content && typeof firstCandidate.content === "object" + ? (firstCandidate.content as Record) + : {}; + const parts = Array.isArray(content.parts) + ? (content.parts as Record[]) + : []; + return parts + .map((part) => (typeof part.text === "string" ? part.text : "")) + .join(""); +} + +function geminiFinishReason(payload: Record): string | null { + const candidates = Array.isArray(payload.candidates) + ? (payload.candidates as Record[]) + : []; + const firstCandidate = candidates[0] ?? {}; + return typeof firstCandidate.finishReason === "string" + ? firstCandidate.finishReason + : null; +} + +function geminiUsageTokens(payload: Record): { + promptTokens: number; + completionTokens: number; + totalTokens: number; +} { + const usage = + payload.usageMetadata && typeof payload.usageMetadata === "object" + ? (payload.usageMetadata as Record) + : {}; + const promptTokens = + Number(usage.promptTokenCount ?? usage.inputTokenCount ?? 0) || 0; + const completionTokens = + Number(usage.candidatesTokenCount ?? usage.outputTokenCount ?? 0) || 0; + const totalTokens = + Number(usage.totalTokenCount ?? promptTokens + completionTokens) || + promptTokens + completionTokens; + return { promptTokens, completionTokens, totalTokens }; +} + +function mapGeminiFinishReasonToOpenAi( + reason: unknown, +): "stop" | "length" | "tool_calls" | "stop_sequence" | null { + const normalized = typeof reason === "string" ? reason.toUpperCase() : ""; + if (normalized === "STOP") { + return "stop"; + } + if (normalized === "MAX_TOKENS") { + return "length"; + } + if (normalized === "STOP_SEQUENCE") { + return "stop_sequence"; + } + if (normalized === "TOOL_CALL" || normalized === "FUNCTION_CALL") { + return "tool_calls"; + } + return null; +} + +function mapGeminiFinishReasonToAnthropic( + reason: unknown, +): "end_turn" | "max_tokens" | "tool_use" | "stop_sequence" | null { + const normalized = typeof reason === "string" ? reason.toUpperCase() : ""; + if (normalized === "STOP") { + return "end_turn"; + } + if (normalized === "MAX_TOKENS") { + return "max_tokens"; + } + if (normalized === "STOP_SEQUENCE") { + return "stop_sequence"; + } + if (normalized === "TOOL_CALL" || normalized === "FUNCTION_CALL") { + return "tool_use"; + } + return null; +} + +function mapOpenAiFinishReasonToGemini( + reason: unknown, +): "STOP" | "MAX_TOKENS" | "STOP_SEQUENCE" | "TOOL_CALL" | null { + const normalized = typeof reason === "string" ? reason : ""; + if (normalized === "stop") { + return "STOP"; + } + if (normalized === "length") { + return "MAX_TOKENS"; + } + if (normalized === "stop_sequence") { + return "STOP_SEQUENCE"; + } + if (normalized === "tool_calls" || normalized === "function_call") { + return "TOOL_CALL"; + } + return null; +} + +function mapAnthropicStopReasonToGemini( + reason: unknown, +): "STOP" | "MAX_TOKENS" | "STOP_SEQUENCE" | "TOOL_CALL" | null { + const normalized = typeof reason === "string" ? reason : ""; + if (normalized === "end_turn") { + return "STOP"; + } + if (normalized === "max_tokens") { + return "MAX_TOKENS"; + } + if (normalized === "stop_sequence") { + return "STOP_SEQUENCE"; + } + if (normalized === "tool_use") { + return "TOOL_CALL"; + } + return null; +} + +function extractOpenAiDeltaText(delta: unknown): string { + if (!delta || typeof delta !== "object") { + return ""; + } + const record = delta as Record; + return openAiContentToText(record.content); +} + +function writeSseEvent( + controller: ReadableStreamDefaultController, + encoder: TextEncoder, + event: string, + data: Record, +): void { + controller.enqueue(encoder.encode(`event: ${event}\n`)); + controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`)); +} + +function writeOpenAiSseChunk( + controller: ReadableStreamDefaultController, + encoder: TextEncoder, + data: Record, +): void { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`)); +} + +function parseJsonFromStreamLine(line: string): Record | null { + const trimmed = line.trim(); + if (!trimmed) { + return null; + } + const payload = trimmed.startsWith("data:") + ? trimmed.slice(5).trim() + : trimmed; + if (!payload || payload === "[DONE]") { + return null; + } + return safeJsonParse | null>(payload, null); +} + +function writeGeminiChunk( + controller: ReadableStreamDefaultController, + encoder: TextEncoder, + data: Record, +): void { + controller.enqueue(encoder.encode(`${JSON.stringify(data)}\n`)); +} + +async function adaptOpenAiJsonToAnthropic(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + + const choices = Array.isArray(payload.choices) + ? (payload.choices as Record[]) + : []; + const firstChoice = choices[0] ?? {}; + const message = + firstChoice.message && typeof firstChoice.message === "object" + ? (firstChoice.message as Record) + : {}; + const usage = + payload.usage && typeof payload.usage === "object" + ? (payload.usage as Record) + : {}; + const promptTokens = Number(usage.prompt_tokens ?? 0) || 0; + const completionTokens = Number(usage.completion_tokens ?? 0) || 0; + const stopReason = mapOpenAiFinishReasonToAnthropic(firstChoice.finish_reason); + const text = openAiContentToText(message.content); + const transformed = { + id: + typeof payload.id === "string" + ? payload.id.replace(/^chatcmpl/, "msg") + : `msg_${Date.now()}`, + type: "message", + role: "assistant", + model: options.model ?? String(payload.model ?? ""), + content: text ? [{ type: "text", text }] : [], + stop_reason: stopReason, + stop_sequence: null, + usage: { + input_tokens: promptTokens, + output_tokens: completionTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +function adaptOpenAiSseToAnthropic(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + const messageId = `msg_${Date.now()}`; + let buffer = ""; + let started = false; + let stopped = false; + let outputTokens = 0; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = buffer.slice(0, newlineIndex).trim(); + buffer = buffer.slice(newlineIndex + 1); + if (!line.startsWith("data:")) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const payload = line.slice(5).trim(); + if (!payload) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (payload === "[DONE]") { + break; + } + const parsed = safeJsonParse | null>( + payload, + null, + ); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (!started) { + started = true; + writeSseEvent(controller, encoder, "message_start", { + type: "message_start", + message: { + id: messageId, + type: "message", + role: "assistant", + model: options.model ?? "", + content: [], + stop_reason: null, + stop_sequence: null, + usage: { + input_tokens: 0, + output_tokens: 0, + }, + }, + }); + writeSseEvent(controller, encoder, "content_block_start", { + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + }); + } + + const usage = + parsed.usage && typeof parsed.usage === "object" + ? (parsed.usage as Record) + : null; + if (usage) { + outputTokens = + Number(usage.completion_tokens ?? outputTokens) || outputTokens; + } + + const choices = Array.isArray(parsed.choices) + ? (parsed.choices as Record[]) + : []; + const firstChoice = choices[0] ?? {}; + const delta = + firstChoice.delta && typeof firstChoice.delta === "object" + ? (firstChoice.delta as Record) + : null; + const deltaText = extractOpenAiDeltaText(delta); + if (deltaText) { + writeSseEvent(controller, encoder, "content_block_delta", { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: deltaText }, + }); + } + + const stopReason = mapOpenAiFinishReasonToAnthropic( + firstChoice.finish_reason, + ); + if (stopReason && !stopped) { + stopped = true; + writeSseEvent(controller, encoder, "content_block_stop", { + type: "content_block_stop", + index: 0, + }); + writeSseEvent(controller, encoder, "message_delta", { + type: "message_delta", + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: outputTokens }, + }); + writeSseEvent(controller, encoder, "message_stop", { + type: "message_stop", + }); + } + newlineIndex = buffer.indexOf("\n"); + } + } + if (started && !stopped) { + writeSseEvent(controller, encoder, "content_block_stop", { + type: "content_block_stop", + index: 0, + }); + writeSseEvent(controller, encoder, "message_delta", { + type: "message_delta", + delta: { stop_reason: "end_turn", stop_sequence: null }, + usage: { output_tokens: outputTokens || 0 }, + }); + writeSseEvent(controller, encoder, "message_stop", { + type: "message_stop", + }); + } + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + + const headers = new Headers(options.response.headers); + headers.set("content-type", "text/event-stream; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { + status: options.response.status, + headers, + }); +} + +async function adaptAnthropicJsonToOpenAi(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + const usage = + payload.usage && typeof payload.usage === "object" + ? (payload.usage as Record) + : {}; + const promptTokens = Number(usage.input_tokens ?? 0) || 0; + const completionTokens = Number(usage.output_tokens ?? 0) || 0; + const text = anthropicContentToText(payload.content); + const transformed = { + id: + typeof payload.id === "string" + ? payload.id.replace(/^msg/, "chatcmpl") + : `chatcmpl_${Date.now()}`, + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: options.model ?? String(payload.model ?? ""), + choices: [ + { + index: 0, + message: { + role: "assistant", + content: text, + }, + finish_reason: mapAnthropicStopReasonToOpenAi(payload.stop_reason), + }, + ], + usage: { + prompt_tokens: promptTokens, + completion_tokens: completionTokens, + total_tokens: promptTokens + completionTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +function adaptAnthropicSseToOpenAi(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + const completionId = `chatcmpl_${Date.now()}`; + const created = Math.floor(Date.now() / 1000); + let buffer = ""; + let started = false; + let stopSent = false; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const rawLine = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + const line = rawLine.trim(); + if (!line.startsWith("data:")) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const payload = line.slice(5).trim(); + if (!payload || payload === "[DONE]") { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const parsed = safeJsonParse | null>( + payload, + null, + ); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const eventType = typeof parsed.type === "string" ? parsed.type : ""; + if (!started && eventType === "message_start") { + started = true; + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [ + { + index: 0, + delta: { role: "assistant" }, + finish_reason: null, + }, + ], + }); + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (eventType === "content_block_delta") { + const delta = + parsed.delta && typeof parsed.delta === "object" + ? (parsed.delta as Record) + : {}; + const text = typeof delta.text === "string" ? delta.text : ""; + if (text) { + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [ + { + index: 0, + delta: { content: text }, + finish_reason: null, + }, + ], + }); + } + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (eventType === "message_delta" && !stopSent) { + const delta = + parsed.delta && typeof parsed.delta === "object" + ? (parsed.delta as Record) + : {}; + const finishReason = mapAnthropicStopReasonToOpenAi( + delta.stop_reason, + ); + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [ + { + index: 0, + delta: {}, + finish_reason: finishReason, + }, + ], + }); + stopSent = true; + newlineIndex = buffer.indexOf("\n"); + continue; + } + newlineIndex = buffer.indexOf("\n"); + } + } + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + + const headers = new Headers(options.response.headers); + headers.set("content-type", "text/event-stream; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { + status: options.response.status, + headers, + }); +} + +async function adaptGeminiJsonToOpenAi(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + const text = geminiCandidateText(payload); + const finishReason = mapGeminiFinishReasonToOpenAi(geminiFinishReason(payload)); + const usage = geminiUsageTokens(payload); + const transformed = { + id: `chatcmpl_${Date.now()}`, + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: options.model ?? "", + choices: [ + { + index: 0, + message: { role: "assistant", content: text }, + finish_reason: finishReason, + }, + ], + usage: { + prompt_tokens: usage.promptTokens, + completion_tokens: usage.completionTokens, + total_tokens: usage.totalTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +async function adaptGeminiJsonToAnthropic(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + const text = geminiCandidateText(payload); + const usage = geminiUsageTokens(payload); + const transformed = { + id: `msg_${Date.now()}`, + type: "message", + role: "assistant", + model: options.model ?? "", + content: text ? [{ type: "text", text }] : [], + stop_reason: mapGeminiFinishReasonToAnthropic(geminiFinishReason(payload)), + stop_sequence: null, + usage: { + input_tokens: usage.promptTokens, + output_tokens: usage.completionTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +async function adaptOpenAiJsonToGemini(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + const choices = Array.isArray(payload.choices) + ? (payload.choices as Record[]) + : []; + const firstChoice = choices[0] ?? {}; + const message = + firstChoice.message && typeof firstChoice.message === "object" + ? (firstChoice.message as Record) + : {}; + const text = openAiContentToText(message.content); + const usage = + payload.usage && typeof payload.usage === "object" + ? (payload.usage as Record) + : {}; + const promptTokens = Number(usage.prompt_tokens ?? 0) || 0; + const completionTokens = Number(usage.completion_tokens ?? 0) || 0; + const transformed = { + candidates: [ + { + content: { role: "model", parts: text ? [{ text }] : [] }, + finishReason: mapOpenAiFinishReasonToGemini(firstChoice.finish_reason), + }, + ], + usageMetadata: { + promptTokenCount: promptTokens, + candidatesTokenCount: completionTokens, + totalTokenCount: promptTokens + completionTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +async function adaptAnthropicJsonToGemini(options: AdaptOptions): Promise { + const payload = (await options.response + .clone() + .json() + .catch(() => null)) as Record | null; + if (!payload) { + return options.response; + } + const text = anthropicContentToText(payload.content); + const usage = + payload.usage && typeof payload.usage === "object" + ? (payload.usage as Record) + : {}; + const promptTokens = Number(usage.input_tokens ?? 0) || 0; + const completionTokens = Number(usage.output_tokens ?? 0) || 0; + const transformed = { + candidates: [ + { + content: { role: "model", parts: text ? [{ text }] : [] }, + finishReason: mapAnthropicStopReasonToGemini(payload.stop_reason), + }, + ], + usageMetadata: { + promptTokenCount: promptTokens, + candidatesTokenCount: completionTokens, + totalTokenCount: promptTokens + completionTokens, + }, + }; + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(JSON.stringify(transformed), { + status: options.response.status, + headers, + }); +} + +function adaptGeminiSseToOpenAi(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + const completionId = `chatcmpl_${Date.now()}`; + const created = Math.floor(Date.now() / 1000); + let buffer = ""; + let started = false; + let stopped = false; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + const parsed = parseJsonFromStreamLine(line); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (!started) { + started = true; + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [ + { index: 0, delta: { role: "assistant" }, finish_reason: null }, + ], + }); + } + const text = geminiCandidateText(parsed); + if (text) { + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [ + { index: 0, delta: { content: text }, finish_reason: null }, + ], + }); + } + const finishReason = mapGeminiFinishReasonToOpenAi( + geminiFinishReason(parsed), + ); + if (finishReason && !stopped) { + stopped = true; + writeOpenAiSseChunk(controller, encoder, { + id: completionId, + object: "chat.completion.chunk", + created, + model: options.model ?? "", + choices: [{ index: 0, delta: {}, finish_reason: finishReason }], + }); + } + newlineIndex = buffer.indexOf("\n"); + } + } + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + + const headers = new Headers(options.response.headers); + headers.set("content-type", "text/event-stream; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { status: options.response.status, headers }); +} + +function adaptGeminiSseToAnthropic(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + const messageId = `msg_${Date.now()}`; + let buffer = ""; + let started = false; + let stopped = false; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + const parsed = parseJsonFromStreamLine(line); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + if (!started) { + started = true; + writeSseEvent(controller, encoder, "message_start", { + type: "message_start", + message: { + id: messageId, + type: "message", + role: "assistant", + model: options.model ?? "", + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + }); + writeSseEvent(controller, encoder, "content_block_start", { + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + }); + } + const text = geminiCandidateText(parsed); + if (text) { + writeSseEvent(controller, encoder, "content_block_delta", { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text }, + }); + } + const stopReason = mapGeminiFinishReasonToAnthropic( + geminiFinishReason(parsed), + ); + if (stopReason && !stopped) { + stopped = true; + writeSseEvent(controller, encoder, "content_block_stop", { + type: "content_block_stop", + index: 0, + }); + writeSseEvent(controller, encoder, "message_delta", { + type: "message_delta", + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: geminiUsageTokens(parsed).completionTokens }, + }); + writeSseEvent(controller, encoder, "message_stop", { + type: "message_stop", + }); + } + newlineIndex = buffer.indexOf("\n"); + } + } + if (started && !stopped) { + writeSseEvent(controller, encoder, "content_block_stop", { + type: "content_block_stop", + index: 0, + }); + writeSseEvent(controller, encoder, "message_delta", { + type: "message_delta", + delta: { stop_reason: "end_turn", stop_sequence: null }, + usage: { output_tokens: 0 }, + }); + writeSseEvent(controller, encoder, "message_stop", { + type: "message_stop", + }); + } + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + + const headers = new Headers(options.response.headers); + headers.set("content-type", "text/event-stream; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { status: options.response.status, headers }); +} + +function adaptOpenAiSseToGemini(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + let buffer = ""; + let lastFinishReason: string | null = null; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + const parsed = parseJsonFromStreamLine(line); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const choices = Array.isArray(parsed.choices) + ? (parsed.choices as Record[]) + : []; + const firstChoice = choices[0] ?? {}; + const delta = + firstChoice.delta && typeof firstChoice.delta === "object" + ? (firstChoice.delta as Record) + : null; + const text = extractOpenAiDeltaText(delta); + const finishReason = mapOpenAiFinishReasonToGemini( + firstChoice.finish_reason, + ); + if (finishReason) { + lastFinishReason = finishReason; + } + if (text || finishReason) { + writeGeminiChunk(controller, encoder, { + candidates: [ + { + content: { + role: "model", + parts: text ? [{ text }] : [], + }, + finishReason: finishReason ?? undefined, + }, + ], + }); + } + newlineIndex = buffer.indexOf("\n"); + } + } + if (!lastFinishReason) { + writeGeminiChunk(controller, encoder, { + candidates: [{ content: { role: "model", parts: [] }, finishReason: "STOP" }], + }); + } + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { status: options.response.status, headers }); +} + +function adaptAnthropicSseToGemini(options: AdaptOptions): Response { + if (!options.response.body) { + return options.response; + } + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const reader = options.response.body.getReader(); + let buffer = ""; + let lastFinishReason: string | null = null; + + const stream = new ReadableStream({ + async start(controller) { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + const parsed = parseJsonFromStreamLine(line); + if (!parsed) { + newlineIndex = buffer.indexOf("\n"); + continue; + } + const eventType = typeof parsed.type === "string" ? parsed.type : ""; + if (eventType === "content_block_delta") { + const delta = + parsed.delta && typeof parsed.delta === "object" + ? (parsed.delta as Record) + : {}; + const text = typeof delta.text === "string" ? delta.text : ""; + if (text) { + writeGeminiChunk(controller, encoder, { + candidates: [ + { + content: { role: "model", parts: [{ text }] }, + }, + ], + }); + } + } + if (eventType === "message_delta") { + const delta = + parsed.delta && typeof parsed.delta === "object" + ? (parsed.delta as Record) + : {}; + lastFinishReason = + mapAnthropicStopReasonToGemini(delta.stop_reason) ?? lastFinishReason; + } + newlineIndex = buffer.indexOf("\n"); + } + } + writeGeminiChunk(controller, encoder, { + candidates: [ + { + content: { role: "model", parts: [] }, + finishReason: lastFinishReason ?? "STOP", + }, + ], + }); + controller.close(); + } catch (error) { + controller.error(error); + } finally { + reader.releaseLock(); + } + }, + }); + const headers = new Headers(options.response.headers); + headers.set("content-type", "application/json; charset=utf-8"); + headers.delete("content-length"); + return new Response(stream, { status: options.response.status, headers }); +} + +const adapters: Record = { + "openai->anthropic": async (options) => { + if (options.isStream) { + return adaptOpenAiSseToAnthropic(options); + } + return adaptOpenAiJsonToAnthropic(options); + }, + "openai->gemini": async (options) => { + if (options.isStream) { + return adaptOpenAiSseToGemini(options); + } + return adaptOpenAiJsonToGemini(options); + }, + "anthropic->openai": async (options) => { + if (options.isStream) { + return adaptAnthropicSseToOpenAi(options); + } + return adaptAnthropicJsonToOpenAi(options); + }, + "anthropic->gemini": async (options) => { + if (options.isStream) { + return adaptAnthropicSseToGemini(options); + } + return adaptAnthropicJsonToGemini(options); + }, + "gemini->openai": async (options) => { + if (options.isStream) { + return adaptGeminiSseToOpenAi(options); + } + return adaptGeminiJsonToOpenAi(options); + }, + "gemini->anthropic": async (options) => { + if (options.isStream) { + return adaptGeminiSseToAnthropic(options); + } + return adaptGeminiJsonToAnthropic(options); + }, +}; + +export async function adaptChatResponse(options: AdaptOptions): Promise { + if (options.upstreamProvider === options.downstreamProvider) { + return options.response; + } + const key = `${options.upstreamProvider}->${options.downstreamProvider}`; + const adapter = adapters[key]; + if (!adapter) { + return options.response; + } + return adapter(options); +} diff --git a/apps/worker/wrangler.toml b/apps/worker/wrangler.toml index 6ef4a87..d0635b1 100644 --- a/apps/worker/wrangler.toml +++ b/apps/worker/wrangler.toml @@ -4,8 +4,7 @@ compatibility_date = "2026-02-14" [vars] CORS_ORIGIN = "http://localhost:5173" -PROXY_RETRY_ROUNDS = "2" -PROXY_RETRY_DELAY_MS = "200" +PROXY_UPSTREAM_TIMEOUT_MS = "30000" [[d1_databases]] binding = "DB" diff --git a/helloagents/CHANGELOG.md b/helloagents/CHANGELOG.md index a6cff5d..a60bf50 100644 --- a/helloagents/CHANGELOG.md +++ b/helloagents/CHANGELOG.md @@ -17,6 +17,30 @@ ### 微调 +- **[worker/proxy]**: 请求日志改为安全采样并补充上游失败异常日志,便于定位 Claude 请求故障 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/index.ts, apps/worker/src/routes/proxy.ts +- **[worker/proxy]**: 上游请求新增超时中断(默认 30s),避免代理请求卡住“有进没出” + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml +- **[worker/proxy]**: 候选通道改为“模型匹配优先、同提供商次优、全通道兜底”并补充筛选日志 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts +- **[worker/proxy]**: 跨提供商且无显式模型映射时自动选择通道已声明模型,提升上下游互转可用性 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts +- **[worker/proxy]**: 自动模型兜底改为学习型复用(成功后记忆映射),并增加失败通道退避与 `PROXY_RETRY_ON_429` 开关降低长耗时重试 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml, tests/worker/channel-metadata.test.ts +- **[worker/proxy]**: 通道选择不再受下游协议约束,并新增 OpenAI→Anthropic 响应转换(含 SSE)修复上游成功但客户端无输出 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, tests/worker/channel-metadata.test.ts +- **[worker/proxy]**: 按“以上游为准”收敛,移除学习映射/失败退避/429重试开关,保留确定性转换与超时保护 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/routes/proxy.ts, apps/worker/src/env.ts, apps/worker/wrangler.toml +- **[worker/proxy]**: 新增响应适配器注册层(OpenAI↔Anthropic,JSON+SSE),并将上游模型决策与下游协议完全解耦 + - 类型: 微调(无方案包) + - 文件: apps/worker/src/services/chat-response-adapter.ts, apps/worker/src/routes/proxy.ts, tests/worker/provider-transform.test.ts, tests/worker/channel-metadata.test.ts - **[checkin]**: 自动签到不再依赖站点启用状态 - 类型: 微调(无方案包) - 文件: apps/worker/src/services/checkin-runner.ts, tests/worker/checkin-runner.test.ts, helloagents/modules/sites.md