diff --git a/open-sse/executors/base.ts b/open-sse/executors/base.ts index 543375916e..fd261cec74 100644 --- a/open-sse/executors/base.ts +++ b/open-sse/executors/base.ts @@ -75,7 +75,7 @@ export function mergeUpstreamExtraHeaders( } } -function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal { +export function mergeAbortSignals(primary: AbortSignal, secondary: AbortSignal): AbortSignal { const controller = new AbortController(); const abortBoth = () => { diff --git a/open-sse/executors/cliproxyapi.ts b/open-sse/executors/cliproxyapi.ts new file mode 100644 index 0000000000..6e87684168 --- /dev/null +++ b/open-sse/executors/cliproxyapi.ts @@ -0,0 +1,92 @@ +import { BaseExecutor, mergeUpstreamExtraHeaders, mergeAbortSignals } from "./base.ts"; +import { HTTP_STATUS, FETCH_TIMEOUT_MS } from "../config/constants.ts"; + +const DEFAULT_PORT = 8317; +const DEFAULT_HOST = "127.0.0.1"; + +function resolveCliproxyapiBaseUrl(): string { + const host = process.env.CLIPROXYAPI_HOST || DEFAULT_HOST; + const port = parseInt(process.env.CLIPROXYAPI_PORT || String(DEFAULT_PORT), 10); + return `http://${host}:${port}`; +} + +export class CliproxyapiExecutor extends BaseExecutor { + private readonly upstreamBaseUrl: string; + + constructor() { + super("cliproxyapi", { + id: "cliproxyapi", + baseUrl: resolveCliproxyapiBaseUrl() + "/v1/chat/completions", + headers: { "Content-Type": "application/json" }, + }); + this.upstreamBaseUrl = resolveCliproxyapiBaseUrl(); + } + + buildUrl(_model: string, _stream: boolean, _urlIndex = 0): string { + return `${this.upstreamBaseUrl}/v1/chat/completions`; + } + + buildHeaders(credentials: any, stream = true): Record { + const headers: Record = { + "Content-Type": "application/json", + }; + + const key = credentials?.apiKey || credentials?.accessToken; + if (key) { + headers["Authorization"] = `Bearer ${key}`; + } + + if (stream) { + headers["Accept"] = "text/event-stream"; + } + + return headers; + } + + transformRequest(model: string, body: any, _stream: boolean, _credentials: any): any { + if (body && typeof body === "object" && body.model !== model) { + return { ...body, model }; + } + return body; + } + + async execute(input: { + model: string; + body: unknown; + stream: boolean; + credentials: any; + signal?: AbortSignal | null; + log?: any; + upstreamExtraHeaders?: Record | null; + }) { + const url = this.buildUrl(input.model, input.stream); + const headers = this.buildHeaders(input.credentials, input.stream); + const transformedBody = this.transformRequest( + input.model, + input.body, + input.stream, + input.credentials + ); + mergeUpstreamExtraHeaders(headers, input.upstreamExtraHeaders); + + const timeoutSignal = AbortSignal.timeout(FETCH_TIMEOUT_MS); + const combinedSignal = input.signal + ? mergeAbortSignals(input.signal, timeoutSignal) + : timeoutSignal; + + const response = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify(transformedBody), + signal: combinedSignal, + }); + + if (response.status === HTTP_STATUS.RATE_LIMITED) { + input.log?.warn?.("CPA", `CLIProxyAPI rate limited: ${response.status}`); + } + + return { response, url, headers, transformedBody }; + } +} + +export default CliproxyapiExecutor; diff --git a/open-sse/executors/index.ts b/open-sse/executors/index.ts index 878706bf4f..b3bbb1347d 100644 --- a/open-sse/executors/index.ts +++ b/open-sse/executors/index.ts @@ -11,6 +11,7 @@ import { CloudflareAIExecutor } from "./cloudflare-ai.ts"; import { OpencodeExecutor } from "./opencode.ts"; import { PuterExecutor } from "./puter.ts"; import { VertexExecutor } from "./vertex.ts"; +import { CliproxyapiExecutor } from "./cliproxyapi.ts"; const executors = { antigravity: new AntigravityExecutor(), @@ -30,6 +31,8 @@ const executors = { puter: new PuterExecutor(), pu: new PuterExecutor(), // Alias vertex: new VertexExecutor(), + cliproxyapi: new CliproxyapiExecutor(), + cpa: new CliproxyapiExecutor(), // Alias }; const defaultCache = new Map(); @@ -57,4 +60,5 @@ export { PollinationsExecutor } from "./pollinations.ts"; export { CloudflareAIExecutor } from "./cloudflare-ai.ts"; export { OpencodeExecutor } from "./opencode.ts"; export { PuterExecutor } from "./puter.ts"; +export { CliproxyapiExecutor } from "./cliproxyapi.ts"; export { VertexExecutor } from "./vertex.ts"; diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 7a2887b641..1707f12cb3 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -44,6 +44,7 @@ import { getModelNormalizeToolCallId, getModelPreserveOpenAIDeveloperRole, getModelUpstreamExtraHeaders, + getUpstreamProxyConfig, } from "@/lib/localDb"; import { getExecutor } from "../executors/index.ts"; import { getCacheControlSettings } from "@/lib/cacheControlSettings"; @@ -329,6 +330,25 @@ function attachLogMeta( * @param {boolean} options.isCombo - Whether this request is from a combo * @param {string} options.connectionId - Connection ID for settings lookup */ + +/** + * Module-level cache for upstream proxy config (shared across all requests). + * 10s TTL prevents per-request DB lookups while staying fresh enough for setting changes. + */ +const _proxyConfigCache = new Map(); +const PROXY_CONFIG_CACHE_TTL = 10_000; + +async function getUpstreamProxyConfigCached(providerId: string) { + const cached = _proxyConfigCache.get(providerId); + if (cached && Date.now() - cached.ts < PROXY_CONFIG_CACHE_TTL) return cached; + const cfg = await getUpstreamProxyConfig(providerId).catch(() => null); + const result = cfg + ? { mode: cfg.mode, enabled: cfg.enabled, ts: Date.now() } + : { mode: "native" as const, enabled: false, ts: Date.now() }; + _proxyConfigCache.set(providerId, result); + return result; +} + export async function handleChatCore({ body, modelInfo, @@ -1010,8 +1030,55 @@ export async function handleChatCore({ } } - // Get executor for this provider - const executor = getExecutor(provider); + // Resolve executor with optional upstream proxy (CLIProxyAPI) routing. + // mode="native" (default): returns the native executor unchanged. + // mode="cliproxyapi": returns the CLIProxyAPI executor instead. + // mode="fallback": returns a wrapper that tries native first, falls back to CLIProxyAPI on 5xx/network errors. + + const resolveExecutorWithProxy = async (prov: string) => { + const cfg = await getUpstreamProxyConfigCached(prov); + if (!cfg.enabled || cfg.mode === "native") return getExecutor(prov); + + if (cfg.mode === "cliproxyapi") { + log?.info?.("UPSTREAM_PROXY", `${prov} routed through CLIProxyAPI (passthrough)`); + return getExecutor("cliproxyapi"); + } + + // mode === "fallback": try native first, retry via CLIProxyAPI on specific failures + const nativeExec = getExecutor(prov); + const proxyExec = getExecutor("cliproxyapi"); + const isRetryableStatus = (s: number) => s >= 500 || s === 429 || s === 0; + + const wrapper = Object.create(nativeExec); + wrapper.execute = async (input: { + model: string; + body: unknown; + stream: boolean; + credentials: unknown; + signal?: AbortSignal | null; + log?: unknown; + upstreamExtraHeaders?: Record | null; + }) => { + try { + const result = await nativeExec.execute(input); + if (isRetryableStatus(result.response.status)) { + log?.info?.( + "UPSTREAM_PROXY", + `${prov} native failed (${result.response.status}), retrying via CLIProxyAPI` + ); + return proxyExec.execute(input); + } + return result; + } catch (err) { + log?.info?.("UPSTREAM_PROXY", `${prov} native error, retrying via CLIProxyAPI`); + return proxyExec.execute(input); + } + }; + return wrapper; + }; + + // Get executor for this provider (with optional upstream proxy routing) + const executor = await resolveExecutorWithProxy(provider); const getExecutionCredentials = () => { const nextCredentials = nativeCodexPassthrough ? { ...credentials, requestEndpointPath: endpointPath } diff --git a/tests/unit/cliproxyapi-executor.test.mjs b/tests/unit/cliproxyapi-executor.test.mjs new file mode 100644 index 0000000000..cc37ccb8a3 --- /dev/null +++ b/tests/unit/cliproxyapi-executor.test.mjs @@ -0,0 +1,232 @@ +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; + +const originalFetch = globalThis.fetch; +const originalEnv = { ...process.env }; + +afterEach(() => { + globalThis.fetch = originalFetch; + process.env.CLIPROXYAPI_HOST = originalEnv.CLIPROXYAPI_HOST; + process.env.CLIPROXYAPI_PORT = originalEnv.CLIPROXYAPI_PORT; +}); + +describe("CliproxyapiExecutor", () => { + let CliproxyapiExecutor; + + beforeEach(async () => { + process.env.CLIPROXYAPI_HOST = ""; + process.env.CLIPROXYAPI_PORT = ""; + const mod = await import("../../open-sse/executors/cliproxyapi.ts"); + CliproxyapiExecutor = mod.CliproxyapiExecutor; + }); + + describe("constructor", () => { + it("should default to 127.0.0.1:8317", () => { + const exec = new CliproxyapiExecutor(); + assert.equal(exec.getProvider(), "cliproxyapi"); + }); + + it("should respect CLIPROXYAPI_HOST env", () => { + process.env.CLIPROXYAPI_HOST = "192.168.1.1"; + const exec = new CliproxyapiExecutor(); + assert.equal(exec.getProvider(), "cliproxyapi"); + }); + + it("should respect CLIPROXYAPI_PORT env", () => { + process.env.CLIPROXYAPI_PORT = "9999"; + const exec = new CliproxyapiExecutor(); + assert.equal(exec.getProvider(), "cliproxyapi"); + }); + }); + + describe("buildUrl", () => { + it("should always return /v1/chat/completions", () => { + process.env.CLIPROXYAPI_HOST = "127.0.0.1"; + process.env.CLIPROXYAPI_PORT = "8317"; + const exec = new CliproxyapiExecutor(); + const url = exec.buildUrl("any-model", true); + assert.equal(url, "http://127.0.0.1:8317/v1/chat/completions"); + }); + + it("should ignore model parameter", () => { + const exec = new CliproxyapiExecutor(); + const url = exec.buildUrl("gpt-4", false); + assert.equal(url, "http://127.0.0.1:8317/v1/chat/completions"); + }); + + it("should use custom host/port", () => { + process.env.CLIPROXYAPI_HOST = "10.0.0.1"; + process.env.CLIPROXYAPI_PORT = "9090"; + const exec = new CliproxyapiExecutor(); + const url = exec.buildUrl("model", true); + assert.equal(url, "http://10.0.0.1:9090/v1/chat/completions"); + }); + }); + + describe("buildHeaders", () => { + it("should return content-type without auth when no credentials", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({}); + assert.equal(headers["Content-Type"], "application/json"); + assert.equal(headers["Authorization"], undefined); + }); + + it("should add Authorization with apiKey", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({ apiKey: "test-key" }); + assert.equal(headers["Authorization"], "Bearer test-key"); + }); + + it("should add Authorization with accessToken", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({ accessToken: "test-token" }); + assert.equal(headers["Authorization"], "Bearer test-token"); + }); + + it("should prefer apiKey over accessToken", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({ apiKey: "key", accessToken: "token" }); + assert.equal(headers["Authorization"], "Bearer key"); + }); + + it("should add Accept header for streaming", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({}, true); + assert.equal(headers["Accept"], "text/event-stream"); + }); + + it("should not add Accept header for non-streaming", () => { + const exec = new CliproxyapiExecutor(); + const headers = exec.buildHeaders({}, false); + assert.equal(headers["Accept"], undefined); + }); + }); + + describe("transformRequest", () => { + it("should update model if body.model differs", () => { + const exec = new CliproxyapiExecutor(); + const body = { model: "old-model", messages: [] }; + const result = exec.transformRequest("new-model", body, true, {}); + assert.equal(result.model, "new-model"); + assert.deepEqual(result.messages, []); + }); + + it("should return body unchanged if model matches", () => { + const exec = new CliproxyapiExecutor(); + const body = { model: "same-model", messages: [] }; + const result = exec.transformRequest("same-model", body, true, {}); + assert.equal(result.model, "same-model"); + }); + + it("should handle non-object body", () => { + const exec = new CliproxyapiExecutor(); + const result = exec.transformRequest("model", "not-an-object", true, {}); + assert.equal(result, "not-an-object"); + }); + + it("should handle null body", () => { + const exec = new CliproxyapiExecutor(); + const result = exec.transformRequest("model", null, true, {}); + assert.equal(result, null); + }); + }); + + describe("execute", () => { + it("should make fetch request with correct URL, headers, and body", async () => { + let capturedUrl, capturedOptions; + globalThis.fetch = async (url, options) => { + capturedUrl = url; + capturedOptions = options; + return { status: 200, ok: true }; + }; + + const exec = new CliproxyapiExecutor(); + const result = await exec.execute({ + model: "test-model", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: true, + credentials: {}, + }); + + assert.equal(capturedUrl, "http://127.0.0.1:8317/v1/chat/completions"); + assert.equal(capturedOptions.method, "POST"); + assert.ok(capturedOptions.signal); + const parsed = JSON.parse(capturedOptions.body); + assert.equal(parsed.messages[0].content, "hi"); + assert.ok(result.response); + }); + + it("should pass credentials to headers", async () => { + let capturedHeaders; + globalThis.fetch = async (_url, options) => { + capturedHeaders = options.headers; + return { status: 200, ok: true }; + }; + + const exec = new CliproxyapiExecutor(); + await exec.execute({ + model: "test", + body: {}, + stream: false, + credentials: { apiKey: "secret-key" }, + }); + + assert.equal(capturedHeaders["Authorization"], "Bearer secret-key"); + }); + + it("should merge upstream extra headers", async () => { + let capturedHeaders; + globalThis.fetch = async (_url, options) => { + capturedHeaders = options.headers; + return { status: 200, ok: true }; + }; + + const exec = new CliproxyapiExecutor(); + await exec.execute({ + model: "test", + body: {}, + stream: false, + credentials: {}, + upstreamExtraHeaders: { "X-Custom": "value" }, + }); + + assert.equal(capturedHeaders["X-Custom"], "value"); + }); + + it("should handle rate limited response", async () => { + globalThis.fetch = async () => ({ status: 429, ok: false }); + const log = { warn: (tag, msg) => {} }; + let logged = false; + log.warn = () => { + logged = true; + }; + + const exec = new CliproxyapiExecutor(); + const result = await exec.execute({ + model: "test", + body: {}, + stream: false, + credentials: {}, + log, + }); + + assert.equal(result.response.status, 429); + }); + + it("should return url, headers, and transformedBody", async () => { + globalThis.fetch = async () => ({ status: 200, ok: true }); + + const exec = new CliproxyapiExecutor(); + const result = await exec.execute({ + model: "test", + body: { messages: [] }, + stream: true, + credentials: {}, + }); + + assert.ok(result.url); + assert.ok(result.headers); + assert.ok(result.transformedBody); + }); + }); +});