Apply provider capability routing per upstream attempt

This commit is contained in:
musistudio
2026-08-04 10:21:51 +08:00
parent ccf63088ea
commit 4cba6b51a8
8 changed files with 484 additions and 104 deletions
@@ -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<string>();
for (const value of [...(current || "").split(","), ...values]) {
@@ -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[],
+1 -38
View File
@@ -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,
+134 -11
View File
@@ -265,16 +265,50 @@ export async function fetchUpstreamWithFallback(input: {
upstreamUrl: string;
}): Promise<UpstreamFetchResult> {
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<string | undefined, {
body?: Buffer;
headers: Record<string, string>;
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<void> {
try {
await response.arrayBuffer();
@@ -1 +1 @@
export { ProxyAgent } from "undici";
export { Agent, ProxyAgent } from "undici";
@@ -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: [
@@ -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
};
}
@@ -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;
}
});