diff --git a/packages/core/src/gateway/core-runtime/supervisor.ts b/packages/core/src/gateway/core-runtime/supervisor.ts index 89cf2433..15ae9e81 100644 --- a/packages/core/src/gateway/core-runtime/supervisor.ts +++ b/packages/core/src/gateway/core-runtime/supervisor.ts @@ -5,7 +5,7 @@ import { spawn, spawnSync, type ChildProcess } from "node:child_process"; import { randomBytes } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { networkInterfaces } from "node:os"; -import { existsSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { delimiter as pathDelimiter, join as pathJoin, resolve as pathResolve } from "node:path"; import { CONFIGDIR } from "@ccr/core/config/constants"; import { @@ -46,11 +46,11 @@ export function spawnGatewayProcess( coreAuthToken: string ): SpawnedGatewayProcess { const gatewayEntry = resolveGatewayEntry(); - const proxyPreloadFile = upstreamProxyUrl ? writeGatewayProxyPreloadFile() : undefined; + const fetchPreloadFile = writeGatewayFetchPreloadFile(); const nodeRuntime = resolveGatewayNodeRuntime(); const env = createGatewayProcessEnv(config, upstreamProxyUrl, runtimeId, coreAuthToken, nodeRuntime.electronRunAsNode); const gatewayBootstrapEntry = resolveGatewayBootstrapEntry(); - const args = proxyPreloadFile ? ["--require", proxyPreloadFile, gatewayBootstrapEntry] : [gatewayBootstrapEntry]; + const args = ["--require", fetchPreloadFile, gatewayBootstrapEntry]; const child = spawn(nodeRuntime.command, args, { cwd: CONFIGDIR, env, @@ -264,6 +264,8 @@ function createGatewayProcessEnv( AUTH_STATIC_API_KEY_ENV: coreGatewayAuthTokenEnv, AUTH_STATIC_API_KEY_HEADER: coreGatewayAuthHeader, CCR_GATEWAY_RUNTIME_ID: runtimeId, + CCR_UNDICI_MODULE: resolveUndiciProxyAgentModule(), + CCR_UPSTREAM_TIMEOUT_MS: String(gatewayUpstreamTimeoutMs(config)), [coreGatewayAuthTokenEnv]: coreAuthToken, HOST: config.gateway.coreHost, PORT: String(config.gateway.corePort) @@ -302,10 +304,14 @@ function createGatewayProcessEnv( env.https_proxy = upstreamProxyUrl; env.all_proxy = upstreamProxyUrl; env.CCR_UPSTREAM_PROXY_URL = upstreamProxyUrl; - env.CCR_UNDICI_MODULE = resolveUndiciProxyAgentModule(); return env; } +function gatewayUpstreamTimeoutMs(config: AppConfig): number { + const value = Number(config.API_TIMEOUT_MS); + return Number.isFinite(value) && value >= 0 ? Math.trunc(value) : 0; +} + function resolveGatewayNodeRuntime(): GatewayNodeRuntime { const candidates = uniqueGatewayNodeRuntimeCandidates([ ...configuredGatewayNodeRuntimeCandidates(), @@ -419,58 +425,77 @@ function appendGatewayChildOutput(child: ChildProcess, message: string): string return details ? `${message}\n${details}` : message; } -export function writeGatewayProxyPreloadFile(): string { +export function writeGatewayFetchPreloadFile(): string { const file = pathJoin(CONFIGDIR, "gateway-proxy-preload.cjs"); - writeFileSync( - file, - [ - "\"use strict\";", - "const up = process.env.CCR_UPSTREAM_PROXY_URL;", - "const um = process.env.CCR_UNDICI_MODULE;", - "if (up && um) {", - " const { ProxyAgent } = require(um);", - " const agent = new ProxyAgent(up);", - " const realFetch = globalThis.fetch.bind(globalThis);", - " const raw = (process.env.NO_PROXY || process.env.no_proxy || '').toLowerCase();", - " const byp = raw.split(',').map((s) => s.trim()).filter(Boolean);", - " const norm = (h) => h.replace(/^\\[/, '').replace(/\\]$/, '').replace(/\\.$/, '');", - " const isLP = (h) => h === 'localhost' || h === '127.0.0.1' || h === '::1' || h === '0:0:0:0:0:0:0:1' || h === '0.0.0.0' || h.startsWith('127.');", - " const shouldBypass = (input) => {", - " let h;", - " try {", - " const u = typeof input === 'string' ? new URL(input) : input instanceof URL ? input : new URL(input && input.url ? input.url : String(input));", - " h = norm(u.hostname);", - " } catch { return true; }", - " if (!h) return false;", - " if (isLP(h)) return true;", - " return byp.some((p) => {", - " if (p === '*') return true;", - " const s = p.split(':');", - " const ph = norm(s[0]);", - " if (s.length === 2 && s[1]) {", - " if (h !== ph) return false;", - " try { return new URL(input).port === s[1]; } catch { return false; }", - " }", - " if (ph.startsWith('*.')) return h.endsWith(ph.slice(1));", - " if (ph.startsWith('.')) return h.endsWith(ph) || h === ph.slice(1);", - " return h === ph;", - " });", - " };", - " const patched = function(input, init) {", - " if (init && init.dispatcher) return realFetch(input, init);", - " if (shouldBypass(input)) return realFetch(input, init);", - " return realFetch(input, Object.assign({}, init, { dispatcher: agent }));", - " };", - " if (Object.getOwnPropertyDescriptor(globalThis, 'fetch')?.writable) {", - " globalThis.fetch = patched;", - " }", - "}" - ].join("\n"), - "utf8" - ); + mkdirSync(CONFIGDIR, { recursive: true }); + writeFileSync(file, gatewayFetchPreloadScript(), "utf8"); return file; } +export function writeGatewayProxyPreloadFile(): string { + return writeGatewayFetchPreloadFile(); +} + +function gatewayFetchPreloadScript(): string { + return [ + "\"use strict\";", + "const up = process.env.CCR_UPSTREAM_PROXY_URL;", + "const um = process.env.CCR_UNDICI_MODULE;", + "const rawTimeout = process.env.CCR_UPSTREAM_TIMEOUT_MS;", + "const parsedTimeout = rawTimeout === undefined || rawTimeout === '' ? NaN : Number(rawTimeout);", + "const hasTimeout = Number.isFinite(parsedTimeout) && parsedTimeout >= 0;", + "if ((up || hasTimeout) && um) {", + " const { Agent, ProxyAgent } = require(um);", + " const timeoutOptions = hasTimeout ? { headersTimeout: Math.trunc(parsedTimeout), bodyTimeout: Math.trunc(parsedTimeout) } : {};", + " const directAgent = hasTimeout ? new Agent(timeoutOptions) : undefined;", + " const proxyAgent = up ? new ProxyAgent(Object.assign({ uri: up }, timeoutOptions)) : undefined;", + " const realFetch = globalThis.fetch.bind(globalThis);", + " const raw = (process.env.NO_PROXY || process.env.no_proxy || '').toLowerCase();", + " const byp = raw.split(',').map((s) => s.trim()).filter(Boolean);", + " const norm = (h) => h.replace(/^\\[/, '').replace(/\\]$/, '').replace(/\\.$/, '');", + " const isLP = (h) => h === 'localhost' || h === '127.0.0.1' || h === '::1' || h === '0:0:0:0:0:0:0:1' || h === '0.0.0.0' || h.startsWith('127.');", + " const requestUrl = (input) => {", + " try {", + " return typeof input === 'string' ? new URL(input) : input instanceof URL ? input : new URL(input && input.url ? input.url : String(input));", + " } catch { return undefined; }", + " };", + " const shouldBypass = (input) => {", + " const u = requestUrl(input);", + " if (!u) return true;", + " const h = norm(u.hostname);", + " if (!h) return false;", + " if (isLP(h)) return true;", + " return byp.some((p) => {", + " if (p === '*') return true;", + " const s = p.split(':');", + " const ph = norm(s[0]);", + " if (s.length === 2 && s[1]) return h === ph && u.port === s[1];", + " if (ph.startsWith('*.')) return h.endsWith(ph.slice(1));", + " if (ph.startsWith('.')) return h.endsWith(ph) || h === ph.slice(1);", + " return h === ph;", + " });", + " };", + " const dispatcherFor = (input) => {", + " if (!proxyAgent) return directAgent;", + " return shouldBypass(input) ? directAgent : proxyAgent;", + " };", + " const patched = function(input, init) {", + " if (init && init.dispatcher) return realFetch(input, init);", + " const dispatcher = dispatcherFor(input);", + " if (!dispatcher) return realFetch(input, init);", + " return realFetch(input, Object.assign({}, init, { dispatcher }));", + " };", + " if (Object.getOwnPropertyDescriptor(globalThis, 'fetch')?.writable) {", + " globalThis.fetch = patched;", + " }", + "}" + ].join("\n"); +} + +export function gatewayFetchPreloadScriptForTest(): string { + return gatewayFetchPreloadScript(); +} + function mergeNoProxy(current: string | undefined, values: string[]): string { const merged = new Set(); for (const value of [...(current || "").split(","), ...values]) { diff --git a/packages/core/src/gateway/features/hosted-web-search/request-transform.ts b/packages/core/src/gateway/features/hosted-web-search/request-transform.ts index b4992001..2e6238e7 100644 --- a/packages/core/src/gateway/features/hosted-web-search/request-transform.ts +++ b/packages/core/src/gateway/features/hosted-web-search/request-transform.ts @@ -2,6 +2,7 @@ import type { AppConfig } from "@ccr/core/contracts/app"; import { isRecord, rawStringValue, stringValue } from "@ccr/core/gateway/internal/value"; import type { AnthropicWebSearchProtocolContext, BrowserWebSearchProtocolRecord, ClaudeCodeWebSearchContinuationContext, HostedWebSearchProtocolContext } from "@ccr/core/gateway/internal/shared"; import { parseJsonObjectSafe, serializeJsonBody } from "@ccr/core/gateway/http/body"; +import { resolveGatewayPublicModelId } from "@ccr/core/gateway/features/model-discovery"; import { uniqueStrings } from "@ccr/core/gateway/internal/collections"; import { requestProtocolForPath } from "@ccr/core/routing/protocol-endpoints"; import { claudeCodeWebSearchToolResultTexts, extractAnthropicWebSearchQueryHint, extractClaudeCodeWebSearchToolResultQuery, extractHostedWebSearchQueryHint, fusionWebSearchToolNameForRequest, hasHostedWebSearchDeclaration, isAnthropicHostedWebSearchTool, isOpenAiHostedWebSearchTool, openAiToolChoiceNamesWebSearch, readHostedWebSearchMaxUses } from "@ccr/core/gateway/features/hosted-web-search/discovery"; @@ -24,7 +25,10 @@ export function createHostedWebSearchProtocolContext(input: { if (!body || !hasHostedWebSearchDeclaration(body, protocol)) { return undefined; } - const toolName = fusionWebSearchToolNameForRequest(input.config, stringValue(body.model) || input.routedModel); + const toolName = fusionWebSearchToolNameForRequest( + input.config, + hostedWebSearchModelForRequest(input.config, stringValue(body.model) || input.routedModel) + ); if (!toolName) { return undefined; } @@ -53,7 +57,10 @@ export function createClaudeCodeWebSearchContinuationContext(input: { if (!body || claudeCodeWebSearchToolResultTexts(body).length === 0) { return undefined; } - const toolName = fusionWebSearchToolNameForRequest(input.config, stringValue(body.model) || input.routedModel); + const toolName = fusionWebSearchToolNameForRequest( + input.config, + hostedWebSearchModelForRequest(input.config, stringValue(body.model) || input.routedModel) + ); if (!toolName) { return undefined; } @@ -64,6 +71,11 @@ export function createClaudeCodeWebSearchContinuationContext(input: { }; } + +function hostedWebSearchModelForRequest(config: AppConfig, model: string | undefined): string | undefined { + return resolveGatewayPublicModelId(model, config) ?? model; +} + export function prepareHostedWebSearchProtocolRequestBody( body: Buffer | undefined, records: BrowserWebSearchProtocolRecord[], diff --git a/packages/core/src/gateway/request/pipeline.ts b/packages/core/src/gateway/request/pipeline.ts index 47b44b74..d63f34ba 100644 --- a/packages/core/src/gateway/request/pipeline.ts +++ b/packages/core/src/gateway/request/pipeline.ts @@ -41,7 +41,7 @@ import { coreGatewayUsageAttributionConfig } from "@ccr/core/gateway/core-runtim import { providerModelPricingForUsage } from "@ccr/core/models/pricing-service"; import { clientClosedRequestStatusCode, clientDisconnectMessage, resolveStreamRequestLogOutcome, UpstreamRequestError } from "@ccr/core/gateway/internal/shared"; import type { BrowserWebSearchMcpIntegration, BrowserWebSearchProtocolRecord, UpstreamFetchResult } from "@ccr/core/gateway/internal/shared"; -import { applyProviderCapabilityRouting, cancelResponseBody, destroyResponseStreams, fetchUpstreamWithFallback, mergeFallbackResponseHeaders, rewriteCapabilityResponseHeaders, uniqueStreams, upstreamResponseHeaders } from "@ccr/core/gateway/upstream/executor"; +import { cancelResponseBody, destroyResponseStreams, fetchUpstreamWithFallback, mergeFallbackResponseHeaders, rewriteCapabilityResponseHeaders, uniqueStreams, upstreamResponseHeaders } from "@ccr/core/gateway/upstream/executor"; import { requestProtocolForPath, shouldApplyGatewayRouting } from "@ccr/core/routing/protocol-endpoints"; import { createClaudeCodeWebSearchContinuationContext, createHostedWebSearchProtocolContext, hostedWebSearchProtocolResponseStream, hostedWebSearchUnavailableMessage, prepareClaudeCodeWebSearchContinuationRequestBody, prepareHostedWebSearchProtocolRequestBody, selectClaudeCodeWebSearchContinuationRecords, selectHostedWebSearchProtocolRecords } from "@ccr/core/gateway/features/hosted-web-search/index"; @@ -380,43 +380,6 @@ export class GatewayRequestPipeline { }); } - const capabilityRoutingStartedAt = Date.now(); - const capabilityBodyBefore = bodyToForward; - const capabilityFallbackBefore = routeFallback; - const capabilityModelBefore = routedModel; - const capabilityProviderHeadersBefore = { - gateway: headers["x-gateway-target-provider"], - list: headers["x-target-providers"], - target: headers["x-target-provider"] - }; - const providerCapabilityRouting = applyProviderCapabilityRouting({ - body: bodyToForward, - config: this.config, - fallback: routeFallback, - headers, - path, - routedModel - }); - bodyToForward = providerCapabilityRouting.body; - routeFallback = providerCapabilityRouting.fallback; - routedModel = providerCapabilityRouting.routedModel; - routeTrace?.capture({ - changes: [ - ...(capabilityBodyBefore === bodyToForward ? [] : [{ operation: "replace" as const, path: "/body/model", scope: "body" as const }]), - reportedRouteChange("routing", "/routing/model", capabilityModelBefore, routedModel), - ...(capabilityFallbackBefore === routeFallback ? [] : [{ after: routeFallback, before: capabilityFallbackBefore, operation: "replace" as const, path: "/routing/fallback", scope: "routing" as const }]), - reportedRouteChange("headers", "/headers/x-target-provider", capabilityProviderHeadersBefore.target, headers["x-target-provider"]), - reportedRouteChange("headers", "/headers/x-target-providers", capabilityProviderHeadersBefore.list, headers["x-target-providers"]), - reportedRouteChange("headers", "/headers/x-gateway-target-provider", capabilityProviderHeadersBefore.gateway, headers["x-gateway-target-provider"]) - ].filter(isReportedRouteChange), - durationMs: Date.now() - capabilityRoutingStartedAt, - kind: "mutation", - name: "provider.capability-routing", - phase: "capability", - startedAtMs: capabilityRoutingStartedAt, - target: routedModel ? { model: routedModel } : undefined - }); - const hostedWebSearchProtocolContext = createHostedWebSearchProtocolContext({ body: bodyToForward, config: this.config, diff --git a/packages/core/src/gateway/upstream/executor.ts b/packages/core/src/gateway/upstream/executor.ts index c85f585b..d8ca72da 100644 --- a/packages/core/src/gateway/upstream/executor.ts +++ b/packages/core/src/gateway/upstream/executor.ts @@ -265,16 +265,50 @@ export async function fetchUpstreamWithFallback(input: { upstreamUrl: string; }): Promise { const fallbackMode = input.fallback.mode; + const planningHeaders = { ...input.headers }; + const planningRouting = applyProviderCapabilityRouting({ + body: input.body, + config: input.config, + fallback: input.fallback, + headers: planningHeaders, + path: input.path, + routedModel: input.routedModel + }); const attempts = buildUpstreamAttempts( input.config, - input.fallback, + planningRouting.fallback, input.method, input.path, - input.body, - input.routedModel + planningRouting.body, + planningRouting.routedModel ); const failedAttempts: UpstreamFailedAttempt[] = []; + const attemptRoutingCache = new Map; + routedModel?: string; + sourceBody?: Buffer; + sourceRoutedModel?: string; + }>(); + const primaryAttempt = attempts[0]; + const parsedInputBody = parseJsonObjectSafe(input.body); + const planningBodyCanSeedPrimary = requestProtocolForPath(input.path) === "gemini_generate_content" || + !parsedInputBody || + !primaryAttempt?.model || + stringValue(parsedInputBody.model) !== undefined; + if (primaryAttempt && planningBodyCanSeedPrimary) { + attemptRoutingCache.set(primaryAttempt.model, { + body: planningRouting.body, + headers: planningHeaders, + routedModel: primaryAttempt.model, + sourceBody: input.body, + sourceRoutedModel: input.routedModel + }); + } input.trace?.capture({ + changes: [ + routeTraceChange("routing", "/routing/fallback", input.fallback, planningRouting.fallback) + ].filter(isRouteTraceChange), decision: { reason: `fallback:${fallbackMode}`, source: "execution-plan" }, kind: "decision", name: "fallback.execution-plan", @@ -290,17 +324,69 @@ export async function fetchUpstreamWithFallback(input: { } const attemptNumber = index + 1; + const plannedAttempt = attempts[index]; + const capabilityRoutingStartedAt = Date.now(); + let cachedAttemptRouting = attemptRoutingCache.get(plannedAttempt.model); + if (!cachedAttemptRouting) { + const routedHeaders = { ...input.headers }; + const sourceBody = buildAttemptBody(input.body, input.path, plannedAttempt.model); + const routing = applyProviderCapabilityRouting({ + body: sourceBody, + config: input.config, + fallback: input.fallback, + headers: routedHeaders, + path: input.path, + routedModel: plannedAttempt.model + }); + cachedAttemptRouting = { + body: routing.body, + headers: routedHeaders, + routedModel: routing.routedModel, + sourceBody, + sourceRoutedModel: plannedAttempt.model + }; + attemptRoutingCache.set(plannedAttempt.model, cachedAttemptRouting); + } + const attemptHeaders = { ...cachedAttemptRouting.headers }; + const attemptSourceBody = cachedAttemptRouting.sourceBody; + const capabilityProviderHeadersBefore = { + gateway: input.headers["x-gateway-target-provider"], + list: input.headers["x-target-providers"], + target: input.headers["x-target-provider"] + }; + input.trace?.capture({ + attempt: attemptNumber, + changes: [ + ...(attemptSourceBody === cachedAttemptRouting.body + ? [] + : [{ operation: "replace" as const, path: "/body/model", scope: "body" as const }]), + routeTraceChange("routing", "/routing/model", cachedAttemptRouting.sourceRoutedModel, cachedAttemptRouting.routedModel), + routeTraceChange("headers", "/headers/x-target-provider", capabilityProviderHeadersBefore.target, attemptHeaders["x-target-provider"]), + routeTraceChange("headers", "/headers/x-target-providers", capabilityProviderHeadersBefore.list, attemptHeaders["x-target-providers"]), + routeTraceChange("headers", "/headers/x-gateway-target-provider", capabilityProviderHeadersBefore.gateway, attemptHeaders["x-gateway-target-provider"]) + ].filter(isRouteTraceChange), + durationMs: Date.now() - capabilityRoutingStartedAt, + kind: "mutation", + name: "provider.capability-routing", + phase: "capability", + startedAtMs: capabilityRoutingStartedAt, + target: cachedAttemptRouting.routedModel ? { model: cachedAttemptRouting.routedModel } : undefined + }); const attemptPreparationStartedAt = Date.now(); const attempt = prepareUpstreamCredentialAttempt({ - attempt: attempts[index], + attempt: { + ...plannedAttempt, + body: cachedAttemptRouting.body, + model: cachedAttemptRouting.routedModel ?? plannedAttempt.model + }, config: input.config, - headers: input.headers, + headers: attemptHeaders, method: input.method, path: input.path }); const hasNextAttempt = index < attempts.length - 1; const attemptUrl = rewriteRouteModelInUrl(input.upstreamUrl, attempt.model); - const attemptHeaders = { + const upstreamHeaders = { ...withCoreGatewayAuthHeader( omitLocalObservabilityHeaders(attempt.headers ?? input.headers), input.coreAuthToken @@ -346,13 +432,13 @@ export async function fetchUpstreamWithFallback(input: { }); releaseJsonObject(attempt.body); - releaseJsonObject(attempts[index].body); + releaseJsonObject(attemptSourceBody); releaseJsonObject(input.body); try { const response = await fetchWithSystemProxy(attemptUrl, { body: shouldSendBody(input.method) ? attempt.body?.toString("utf8") : undefined, - headers: attemptHeaders, + headers: upstreamHeaders, method: input.method, signal: input.signal }); @@ -904,9 +990,6 @@ function buildUpstreamAttempts( primaryModel: routedModel }); return plan.attempts.map((attempt) => ({ - body: parsedBody && !modelInPath && fallback.mode === "model-chain" && attempt.model - ? serializeJsonBodyWithModel(parsedBody, attempt.model) - : body, index: attempt.index, model: attempt.model, target: attempt.target @@ -914,6 +997,46 @@ function buildUpstreamAttempts( } +function buildAttemptBody( + body: Buffer | undefined, + path: string, + model: string | undefined +): Buffer | undefined { + if (!body || !model || requestProtocolForPath(path) === "gemini_generate_content") { + return body; + } + const parsedBody = parseJsonObjectSafe(body); + if (!parsedBody || stringValue(parsedBody.model) === model) { + return body; + } + return serializeJsonBodyWithModel(parsedBody, model); +} + + +function routeTraceChange( + scope: RequestRouteTraceChange["scope"], + path: string, + before: unknown, + after: unknown +): RequestRouteTraceChange | undefined { + if (before === after) { + return undefined; + } + return { + ...(before === undefined ? {} : { before }), + ...(after === undefined ? {} : { after }), + operation: before === undefined ? "add" : after === undefined ? "remove" : "replace", + path, + scope + }; +} + + +function isRouteTraceChange(value: RequestRouteTraceChange | undefined): value is RequestRouteTraceChange { + return Boolean(value); +} + + async function drainResponseBody(response: Response): Promise { try { await response.arrayBuffer(); diff --git a/packages/core/src/proxy/undici-proxy-agent.ts b/packages/core/src/proxy/undici-proxy-agent.ts index e94e364f..97c9a7b6 100644 --- a/packages/core/src/proxy/undici-proxy-agent.ts +++ b/packages/core/src/proxy/undici-proxy-agent.ts @@ -1 +1 @@ -export { ProxyAgent } from "undici"; +export { Agent, ProxyAgent } from "undici"; diff --git a/packages/core/test/integration/gateway/gateway-virtual-models.test.mjs b/packages/core/test/integration/gateway/gateway-virtual-models.test.mjs index 4db9c649..56a49ff6 100644 --- a/packages/core/test/integration/gateway/gateway-virtual-models.test.mjs +++ b/packages/core/test/integration/gateway/gateway-virtual-models.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { PassThrough, Readable } from "node:stream"; import test from "node:test"; +import { createHostedWebSearchProtocolContext } from "@ccr/core/gateway/features/hosted-web-search/index.ts"; import { fusionFallbackToolDefinitions, fusionWebSearchToolNameForRequest, @@ -504,6 +505,51 @@ test("gateway resolves normalized Fusion web search tool names for Anthropic pro assert.equal(fusionWebSearchToolNameForRequest(config, "Fusion/kimisearch"), "fusion_2_web_search"); }); +test("gateway resolves Claude discovery aliases before hosted web search matching", () => { + const config = { + Providers: [], + Router: { fallback: { mode: "off", models: [], retryCount: 0 } }, + gateway: {}, + virtualModelProfiles: [ + { + displayName: "Kimisearch", + enabled: true, + execution: { + clientToolsPolicy: "allow", + matchWebSearch: true, + maxToolCalls: 8, + maxTurns: 6, + mode: "tool_loop", + streamMode: "optimistic" + }, + id: "fusion-2", + key: "kimisearch", + match: { exactAliases: ["kimisearch", "Fusion/kimisearch"], prefixes: [], suffixes: [] }, + materialization: { enabled: true, includeInGatewayModels: true }, + metadata: { + fusionWebSearch: { provider: "browser", toolName: "web_search_fusion_2" } + }, + tools: [{ name: "web_search_fusion_2", visibility: "internal" }] + } + ] + }; + const context = createHostedWebSearchProtocolContext({ + body: Buffer.from(JSON.stringify({ + messages: [{ content: "weather", role: "user" }], + model: "claude-Fusion/kimisearch", + tools: [{ name: "web_search", type: "web_search_20250305" }] + })), + config, + method: "POST", + path: "/v1/messages", + requestId: "request-1", + routedModel: "claude-Fusion/kimisearch", + sinceMs: 0 + }); + + assert.equal(context?.toolName, "fusion_2_web_search"); +}); + test("gateway does not route hosted web search through an unrelated Fusion search profile", () => { const config = { Providers: [ diff --git a/packages/core/test/unit/gateway/gateway-fetch-preload.test.mjs b/packages/core/test/unit/gateway/gateway-fetch-preload.test.mjs new file mode 100644 index 00000000..97f2f216 --- /dev/null +++ b/packages/core/test/unit/gateway/gateway-fetch-preload.test.mjs @@ -0,0 +1,103 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import vm from "node:vm"; +import { gatewayFetchPreloadScriptForTest } from "@ccr/core/gateway/core-runtime/supervisor.ts"; + +test("gateway fetch preload applies API_TIMEOUT_MS to direct upstream fetches", async () => { + const harness = executePreload({ + CCR_UNDICI_MODULE: "mock-undici", + CCR_UPSTREAM_TIMEOUT_MS: "600000" + }); + + assert.equal(harness.agentOptions.length, 1); + assert.equal(harness.agentOptions[0].headersTimeout, 600000); + assert.equal(harness.agentOptions[0].bodyTimeout, 600000); + assert.equal(harness.proxyAgentOptions.length, 0); + + await harness.context.fetch("https://api.example.test/v1/chat/completions", { method: "POST" }); + + assert.equal(harness.fetchCalls.length, 1); + assert.equal(harness.fetchCalls[0].init.dispatcher.kind, "direct"); +}); + +test("gateway fetch preload combines proxy routing with timeout dispatcher options", async () => { + const harness = executePreload({ + CCR_UNDICI_MODULE: "mock-undici", + CCR_UPSTREAM_PROXY_URL: "http://127.0.0.1:8888", + CCR_UPSTREAM_TIMEOUT_MS: "600000", + NO_PROXY: "api.internal.test,.bypass.test" + }); + + assert.equal(harness.agentOptions.length, 1); + assert.equal(harness.proxyAgentOptions.length, 1); + assert.equal(harness.proxyAgentOptions[0].uri, "http://127.0.0.1:8888"); + assert.equal(harness.proxyAgentOptions[0].headersTimeout, 600000); + assert.equal(harness.proxyAgentOptions[0].bodyTimeout, 600000); + + await harness.context.fetch("https://api.external.test/v1/messages", {}); + await harness.context.fetch("https://api.internal.test/v1/messages", {}); + await harness.context.fetch("http://127.0.0.1:3456/health", {}); + await harness.context.fetch("https://service.bypass.test/v1/messages", {}); + + assert.equal(harness.fetchCalls[0].init.dispatcher.kind, "proxy"); + assert.equal(harness.fetchCalls[1].init.dispatcher.kind, "direct"); + assert.equal(harness.fetchCalls[2].init.dispatcher.kind, "direct"); + assert.equal(harness.fetchCalls[3].init.dispatcher.kind, "direct"); +}); + +test("gateway fetch preload preserves explicit fetch dispatchers", async () => { + const harness = executePreload({ + CCR_UNDICI_MODULE: "mock-undici", + CCR_UPSTREAM_TIMEOUT_MS: "600000" + }); + const dispatcher = { kind: "caller" }; + + await harness.context.fetch("https://api.example.test/v1/messages", { dispatcher }); + + assert.equal(harness.fetchCalls.length, 1); + assert.equal(harness.fetchCalls[0].init.dispatcher, dispatcher); +}); + +function executePreload(env) { + const agentOptions = []; + const proxyAgentOptions = []; + const fetchCalls = []; + + class Agent { + constructor(options) { + this.kind = "direct"; + this.options = options; + agentOptions.push(options); + } + } + + class ProxyAgent { + constructor(options) { + this.kind = "proxy"; + this.options = options; + proxyAgentOptions.push(options); + } + } + + const context = { + URL, + fetch: async (input, init) => { + fetchCalls.push({ init, input }); + return { ok: true }; + }, + process: { env }, + require: (moduleName) => { + assert.equal(moduleName, "mock-undici"); + return { Agent, ProxyAgent }; + } + }; + vm.createContext(context); + vm.runInContext(gatewayFetchPreloadScriptForTest(), context); + + return { + agentOptions, + context, + fetchCalls, + proxyAgentOptions + }; +} diff --git a/packages/core/test/unit/gateway/upstream-executor.test.mjs b/packages/core/test/unit/gateway/upstream-executor.test.mjs index edac51b8..7fc1cc00 100644 --- a/packages/core/test/unit/gateway/upstream-executor.test.mjs +++ b/packages/core/test/unit/gateway/upstream-executor.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import { fetchUpstreamWithFallback } from "@ccr/core/gateway/upstream/executor.ts"; +import { RequestRouteTraceRecorder } from "@ccr/core/observability/route-trace.ts"; const retryConfig = { Providers: [], @@ -64,3 +65,110 @@ test("retry backoff stops after client aborts a network error", async () => { throw new Error("upstream unavailable"); }); }); + +test("model-chain fallback rebuilds every protocol attempt from the canonical request", async () => { + const config = { + Providers: [ + { + capabilities: [{ baseUrl: "https://anthropic-primary.example", type: "anthropic_messages" }], + id: "anthropic-primary", + models: ["claude-primary"], + name: "Anthropic Primary" + }, + { + capabilities: [{ baseUrl: "https://openai-fallback.example", type: "openai_responses" }], + id: "openai-fallback", + models: ["gpt-fallback"], + name: "OpenAI Fallback" + }, + { + capabilities: [{ baseUrl: "https://anthropic-recovery.example", type: "anthropic_messages" }], + id: "anthropic-recovery", + models: ["claude-recovery"], + name: "Anthropic Recovery" + } + ], + Router: { fallback: { mode: "off", models: [], retryCount: 0 }, rules: [] }, + virtualModelProfiles: [] + }; + const fallback = { + mode: "model-chain", + models: ["OpenAI Fallback/gpt-fallback", "Anthropic Recovery/claude-recovery"], + retryCount: 0 + }; + const canonicalBody = { + context_management: { edits: [{ type: "clear_tool_uses_20250919" }] }, + messages: [{ content: "hello", role: "user" }], + model: "Anthropic Primary/claude-primary", + output_config: { effort: "high", verbosity: "medium" }, + system: [{ cache_control: { type: "ephemeral" }, text: "system", type: "text" }], + thinking: { type: "adaptive" } + }; + const captured = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = async (_url, init) => { + captured.push({ + body: JSON.parse(init.body), + headers: init.headers + }); + const status = captured.length === 1 ? 429 : captured.length === 2 ? 400 : 200; + return new Response('{"ok":true}', { + headers: { + "content-type": "application/json", + "retry-after": "0.001" + }, + status + }); + }; + + try { + const trace = new RequestRouteTraceRecorder(Date.now()); + const result = await fetchUpstreamWithFallback({ + body: Buffer.from(JSON.stringify(canonicalBody)), + config, + coreAuthToken: "core-token", + fallback, + headers: {}, + method: "POST", + path: "/v1/messages", + routedModel: canonicalBody.model, + trace, + upstreamUrl: "http://127.0.0.1:3456/v1/messages" + }); + + assert.equal(result.response.status, 200); + assert.equal(captured.length, 3); + assert.deepEqual(captured.map((attempt) => attempt.body.model), [ + "claude-primary", + "gpt-fallback", + "claude-recovery" + ]); + assert.deepEqual(captured[0].body.thinking, { type: "adaptive" }); + assert.equal(captured[1].body.thinking, undefined); + assert.deepEqual(captured[2].body.thinking, { type: "adaptive" }); + assert.deepEqual(captured[2].body.context_management, canonicalBody.context_management); + assert.deepEqual(captured[2].body.output_config, canonicalBody.output_config); + assert.equal(captured[0].headers["x-target-provider"], "anthropic-primary::anthropic_messages"); + assert.equal(captured[1].headers["x-target-provider"], "openai-fallback::openai_responses"); + assert.equal(captured[2].headers["x-target-provider"], "anthropic-recovery::anthropic_messages"); + const finishedTrace = trace.finish(); + const capabilityRoutingHops = finishedTrace.hops + .filter((hop) => hop.name === "provider.capability-routing"); + assert.deepEqual( + capabilityRoutingHops.map((hop) => hop.attempt), + [1, 2, 3] + ); + assert.deepEqual( + capabilityRoutingHops[0].changes.map((change) => change.path), + ["/body/model", "/routing/model"] + ); + assert.deepEqual( + finishedTrace.hops + .find((hop) => hop.name === "fallback.execution-plan") + ?.changes.map((change) => change.path), + ["/routing/fallback"] + ); + } finally { + globalThis.fetch = originalFetch; + } +});