diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 0a3bc7b399..1c5116ee3e 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -13,7 +13,7 @@ import { resolveCompressionHeader, } from "./chatCore/headers.ts"; import { markCodexScopeRateLimited } from "./chatCore/codexFailover.ts"; -import { getCombosCached, getUpstreamProxyConfigCached } from "./chatCore/comboContextCache.ts"; +import { getCombosCached } from "./chatCore/comboContextCache.ts"; export { clearCombosCache, clearUpstreamProxyConfigCache } from "./chatCore/comboContextCache.ts"; import { resolveAccountSemaphoreKey, @@ -136,6 +136,8 @@ import { import { wrapReadableStreamWithFinalize } from "./chatCore/streamFinalize.ts"; import { buildCacheUsageLogMeta, attachLogMeta } from "./chatCore/cacheUsageMeta.ts"; import { buildExecutorClientHeaders } from "./chatCore/executorClientHeaders.ts"; +import { resolveExecutionCredentials as resolveExecutionCredentialsFor } from "./chatCore/executionCredentials.ts"; +import { resolveExecutorWithProxy as resolveExecutorWithProxyFor } from "./chatCore/executorProxy.ts"; import { getCallLogPipelineCaptureStreamChunks, @@ -180,7 +182,6 @@ import { import { getModelNormalizeToolCallId, getModelPreserveOpenAIDeveloperRole } from "@/lib/localDb"; import { getProviderCredentials, extractSessionAffinityKey } from "@/sse/services/auth"; import { deleteSessionAccountAffinity } from "@/lib/db/sessionAccountAffinity"; -import { getExecutor } from "../executors/index.ts"; import { getCacheControlSettings } from "@/lib/cacheControlSettings"; import { guardrailRegistry, resolveDisabledGuardrails } from "@/lib/guardrails"; import { @@ -2227,80 +2228,7 @@ export async function handleChatCore({ // 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"); - - // Read custom fallback codes from settings. Default: 5xx + 429 + network errors. - let fallbackCodes: number[] = [429, 500, 502, 503, 504]; - try { - const allSettings = await getCachedSettings(); - if ( - typeof allSettings.cliproxyapi_fallback_codes === "string" && - allSettings.cliproxyapi_fallback_codes.trim() - ) { - const parsed = allSettings.cliproxyapi_fallback_codes - .split(",") - .map((s: string) => Number.parseInt(s.trim(), 10)) - .filter((n: number) => !Number.isNaN(n)); - if (parsed.length > 0) fallbackCodes = parsed; - } - } catch { - /* use defaults */ - } - const isRetryableStatus = (s: number) => fallbackCodes.includes(s) || 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; - }) => { - let result; - try { - result = await nativeExec.execute(input); - } catch (err) { - const errMsg = err instanceof Error ? err.message : String(err); - log?.info?.("UPSTREAM_PROXY", `${prov} native error (${errMsg}), retrying via CLIProxyAPI`); - try { - return await proxyExec.execute(input); - } catch (proxyErr) { - const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr); - log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`); - throw proxyErr; - } - } - - if (!isRetryableStatus(result.response.status)) { - return result; - } - log?.info?.( - "UPSTREAM_PROXY", - `${prov} native failed (${result.response.status}), retrying via CLIProxyAPI` - ); - try { - return await proxyExec.execute(input); - } catch (proxyErr) { - const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr); - log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`); - throw proxyErr; - } - }; - return wrapper; - }; + const resolveExecutorWithProxy = (prov: string) => resolveExecutorWithProxyFor(prov, log); // === Quota Share enforcement PRE-hook (B/F7) === // Runs after provider/model/credentials/apiKeyInfo are fully resolved, @@ -2369,50 +2297,15 @@ export async function handleChatCore({ // Get executor for this provider (with optional upstream proxy routing) const executor = await resolveExecutorWithProxy(provider); - const getExecutionCredentials = () => { - const nextCredentials = nativeCodexPassthrough - ? { ...credentials, requestEndpointPath: endpointPath } - : credentials; - - const providerSpecificData = - nextCredentials?.providerSpecificData && - typeof nextCredentials.providerSpecificData === "object" - ? { ...nextCredentials.providerSpecificData } - : {}; - - // Some providers (Azure AI Foundry, OCI OpenAI-compatible) choose upstream - // endpoint path from providerSpecificData.apiType. When a model routes to - // OpenAI Responses format, force apiType=responses unless explicitly set. - if ( - targetFormat === FORMATS.OPENAI_RESPONSES && - (provider === "azure-ai" || provider === "oci") && - providerSpecificData.apiType !== "responses" - ) { - providerSpecificData.apiType = "responses"; - } - - if ( - targetFormat === FORMATS.OPENAI_RESPONSES && - (provider === "azure-ai" || provider === "oci") - ) { - providerSpecificData._omnirouteForceResponsesUpstream = true; - } - - const withApiType = { - ...nextCredentials, - providerSpecificData, - }; - - if (!ccSessionId) return withApiType; - - return { - ...withApiType, - providerSpecificData: { - ...(withApiType?.providerSpecificData || {}), - ccSessionId, - }, - }; - }; + const getExecutionCredentials = () => + resolveExecutionCredentialsFor({ + credentials, + nativeCodexPassthrough, + endpointPath, + targetFormat, + provider, + ccSessionId, + }); let onPipelineStreamError: streamFailure.PipelineStreamErrorHandler | null = null; diff --git a/open-sse/handlers/chatCore/executionCredentials.ts b/open-sse/handlers/chatCore/executionCredentials.ts new file mode 100644 index 0000000000..3d0d7dcf93 --- /dev/null +++ b/open-sse/handlers/chatCore/executionCredentials.ts @@ -0,0 +1,72 @@ +/** + * chatCore execution-credentials resolver (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Pure builder extracted from handleChatCore: derives the per-execution credentials object from the + * resolved request context. Applies the native-Codex passthrough endpoint override, forces + * apiType=responses (and the responses-upstream marker) for Azure AI Foundry / OCI when the model + * routes to the OpenAI Responses format, and threads the Claude Code session id when present. + * Side-effect-free; behaviour is byte-identical to the previous inline closure. + */ + +import { FORMATS } from "../../translator/formats.ts"; + +type CredentialsLike = + | { + providerSpecificData?: Record | null; + [key: string]: unknown; + } + | null + | undefined; + +export function resolveExecutionCredentials(opts: { + credentials: CredentialsLike; + nativeCodexPassthrough: boolean; + endpointPath: string; + targetFormat: string; + provider: string | null | undefined; + ccSessionId: string | null; +}) { + const { credentials, nativeCodexPassthrough, endpointPath, targetFormat, provider, ccSessionId } = + opts; + + const nextCredentials = nativeCodexPassthrough + ? { ...credentials, requestEndpointPath: endpointPath } + : credentials; + + const providerSpecificData = + nextCredentials?.providerSpecificData && + typeof nextCredentials.providerSpecificData === "object" + ? { ...nextCredentials.providerSpecificData } + : {}; + + // Some providers (Azure AI Foundry, OCI OpenAI-compatible) choose upstream + // endpoint path from providerSpecificData.apiType. When a model routes to + // OpenAI Responses format, force apiType=responses unless explicitly set. + if ( + targetFormat === FORMATS.OPENAI_RESPONSES && + (provider === "azure-ai" || provider === "oci") && + providerSpecificData.apiType !== "responses" + ) { + providerSpecificData.apiType = "responses"; + } + + if (targetFormat === FORMATS.OPENAI_RESPONSES && (provider === "azure-ai" || provider === "oci")) { + providerSpecificData._omnirouteForceResponsesUpstream = true; + } + + const withApiType = { + ...nextCredentials, + providerSpecificData, + }; + + if (!ccSessionId) return withApiType; + + return { + ...withApiType, + providerSpecificData: { + ...(withApiType?.providerSpecificData || {}), + ccSessionId, + }, + }; +} diff --git a/open-sse/handlers/chatCore/executorProxy.ts b/open-sse/handlers/chatCore/executorProxy.ts new file mode 100644 index 0000000000..bb9e327ea9 --- /dev/null +++ b/open-sse/handlers/chatCore/executorProxy.ts @@ -0,0 +1,98 @@ +/** + * chatCore upstream-proxy executor resolver (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Extracted from handleChatCore: resolves the executor for a provider honoring the configured + * upstream proxy mode. `native` / disabled → the provider's own executor; `cliproxyapi` → the + * CLIProxyAPI passthrough executor; `fallback` → a wrapper that tries the native executor first and + * retries via CLIProxyAPI on configured failure codes (default 5xx + 429 + network) or on a thrown + * error. Behaviour is byte-identical to the previous inline closure (it only captured `log`). + */ + +import { getExecutor } from "../../executors/index.ts"; +import { getCachedSettings } from "@/lib/db/readCache"; +import { getUpstreamProxyConfigCached } from "./comboContextCache.ts"; + +type LoggerLike = + | { + info?: (...args: unknown[]) => void; + error?: (...args: unknown[]) => void; + warn?: (...args: unknown[]) => void; + } + | null + | undefined; + +export async function resolveExecutorWithProxy(prov: string, log?: LoggerLike) { + 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"); + + // Read custom fallback codes from settings. Default: 5xx + 429 + network errors. + let fallbackCodes: number[] = [429, 500, 502, 503, 504]; + try { + const allSettings = await getCachedSettings(); + if ( + typeof allSettings.cliproxyapi_fallback_codes === "string" && + allSettings.cliproxyapi_fallback_codes.trim() + ) { + const parsed = allSettings.cliproxyapi_fallback_codes + .split(",") + .map((s: string) => Number.parseInt(s.trim(), 10)) + .filter((n: number) => !Number.isNaN(n)); + if (parsed.length > 0) fallbackCodes = parsed; + } + } catch { + /* use defaults */ + } + const isRetryableStatus = (s: number) => fallbackCodes.includes(s) || 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; + }) => { + let result; + try { + result = await nativeExec.execute(input); + } catch (err) { + const errMsg = err instanceof Error ? err.message : String(err); + log?.info?.("UPSTREAM_PROXY", `${prov} native error (${errMsg}), retrying via CLIProxyAPI`); + try { + return await proxyExec.execute(input); + } catch (proxyErr) { + const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr); + log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`); + throw proxyErr; + } + } + + if (!isRetryableStatus(result.response.status)) { + return result; + } + log?.info?.( + "UPSTREAM_PROXY", + `${prov} native failed (${result.response.status}), retrying via CLIProxyAPI` + ); + try { + return await proxyExec.execute(input); + } catch (proxyErr) { + const proxyMsg = proxyErr instanceof Error ? proxyErr.message : String(proxyErr); + log?.error?.("UPSTREAM_PROXY", `${prov} CLIProxyAPI fallback also failed: ${proxyMsg}`); + throw proxyErr; + } + }; + return wrapper; +} diff --git a/tests/unit/chatcore-execution-credentials.test.ts b/tests/unit/chatcore-execution-credentials.test.ts new file mode 100644 index 0000000000..8eec6ee344 --- /dev/null +++ b/tests/unit/chatcore-execution-credentials.test.ts @@ -0,0 +1,98 @@ +// tests/unit/chatcore-execution-credentials.test.ts +// Characterization of resolveExecutionCredentials — the per-execution credentials builder extracted +// from handleChatCore (chatCore god-file decomposition, #3501). Locks: the native-Codex passthrough +// endpoint override, the Azure AI / OCI apiType=responses forcing (+ responses-upstream marker) only +// under the OpenAI Responses target format, respecting an explicit apiType, and the Claude Code +// session-id threading. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { resolveExecutionCredentials } from "../../open-sse/handlers/chatCore/executionCredentials.ts"; + +const RESPONSES = "openai-responses"; + +const base = { + credentials: { providerSpecificData: { foo: "bar" } } as Record, + nativeCodexPassthrough: false, + endpointPath: "/v1/responses", + targetFormat: "openai", + provider: "openai", + ccSessionId: null, +}; + +test("non-passthrough leaves credentials without a requestEndpointPath", () => { + const out = resolveExecutionCredentials({ ...base }) as Record; + assert.equal("requestEndpointPath" in out, false); + assert.deepEqual(out.providerSpecificData, { foo: "bar" }); +}); + +test("native Codex passthrough injects requestEndpointPath", () => { + const out = resolveExecutionCredentials({ + ...base, + nativeCodexPassthrough: true, + }) as Record; + assert.equal(out.requestEndpointPath, "/v1/responses"); +}); + +test("azure-ai + responses target forces apiType=responses and the upstream marker", () => { + const out = resolveExecutionCredentials({ + ...base, + provider: "azure-ai", + targetFormat: RESPONSES, + }) as Record; + const psd = out.providerSpecificData as Record; + assert.equal(psd.apiType, "responses"); + assert.equal(psd._omnirouteForceResponsesUpstream, true); +}); + +test("a non-responses apiType is forced to responses under the responses target", () => { + const out = resolveExecutionCredentials({ + ...base, + provider: "oci", + targetFormat: RESPONSES, + credentials: { providerSpecificData: { apiType: "chat" } }, + }) as Record; + const psd = out.providerSpecificData as Record; + assert.equal(psd.apiType, "responses"); + assert.equal(psd._omnirouteForceResponsesUpstream, true); +}); + +test("an explicit apiType=responses is preserved (guard short-circuits the reassignment)", () => { + const out = resolveExecutionCredentials({ + ...base, + provider: "oci", + targetFormat: RESPONSES, + credentials: { providerSpecificData: { apiType: "responses" } }, + }) as Record; + const psd = out.providerSpecificData as Record; + assert.equal(psd.apiType, "responses"); + assert.equal(psd._omnirouteForceResponsesUpstream, true); +}); + +test("non azure/oci providers never get apiType forcing", () => { + const out = resolveExecutionCredentials({ + ...base, + provider: "openai", + targetFormat: RESPONSES, + }) as Record; + const psd = out.providerSpecificData as Record; + assert.equal(psd.apiType, undefined); + assert.equal(psd._omnirouteForceResponsesUpstream, undefined); +}); + +test("ccSessionId is threaded into providerSpecificData when present", () => { + const out = resolveExecutionCredentials({ + ...base, + ccSessionId: "sess-123", + }) as Record; + const psd = out.providerSpecificData as Record; + assert.equal(psd.ccSessionId, "sess-123"); + assert.equal(psd.foo, "bar"); +}); + +test("missing providerSpecificData defaults to an empty object", () => { + const out = resolveExecutionCredentials({ + ...base, + credentials: { connectionId: "c1" }, + }) as Record; + assert.deepEqual(out.providerSpecificData, {}); +}); diff --git a/tests/unit/chatcore-executor-proxy.test.ts b/tests/unit/chatcore-executor-proxy.test.ts new file mode 100644 index 0000000000..98e5700565 --- /dev/null +++ b/tests/unit/chatcore-executor-proxy.test.ts @@ -0,0 +1,80 @@ +// tests/unit/chatcore-executor-proxy.test.ts +// Characterization of resolveExecutorWithProxy — the upstream-proxy executor resolver extracted from +// handleChatCore (chatCore god-file decomposition, #3501). Exercises the REAL config path through a +// temp DB: disabled/native → the provider's own executor; cliproxyapi → the passthrough executor; +// fallback → a distinct wrapper that owns its own execute(). The wrapper's retry behaviour is not +// invoked here (it would hit the network); the existing cliproxyapi-fallback-wiring.test.ts covers +// the surrounding wiring. +import { test, before, after, beforeEach } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const testDataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omni-executor-proxy-test-")); +process.env.DATA_DIR = testDataDir; + +// Dynamic imports AFTER DATA_DIR is set so core.ts picks up the temp path. +const coreDb = await import("../../src/lib/db/core.ts"); +const upstreamProxyDb = await import("../../src/lib/db/upstreamProxy.ts"); +const { resolveExecutorWithProxy } = await import( + "../../open-sse/handlers/chatCore/executorProxy.ts" +); +const { getExecutor } = await import("../../open-sse/executors/index.ts"); +const { clearUpstreamProxyConfigCache } = await import( + "../../open-sse/handlers/chatCore/comboContextCache.ts" +); + +before(async () => { + await coreDb.ensureDbInitialized(); +}); + +beforeEach(() => { + clearUpstreamProxyConfigCache(); +}); + +after(() => { + coreDb.resetDbInstance(); + fs.rmSync(testDataDir, { recursive: true, force: true }); +}); + +test("no config (disabled by default) returns the provider's own executor", async () => { + clearUpstreamProxyConfigCache("openai"); + const exec = await resolveExecutorWithProxy("openai"); + assert.equal(exec, getExecutor("openai")); +}); + +test("mode 'native' returns the provider's own executor", async () => { + await upstreamProxyDb.upsertUpstreamProxyConfig({ + providerId: "openai", + mode: "native", + enabled: true, + }); + clearUpstreamProxyConfigCache("openai"); + const exec = await resolveExecutorWithProxy("openai"); + assert.equal(exec, getExecutor("openai")); +}); + +test("mode 'cliproxyapi' returns the CLIProxyAPI passthrough executor", async () => { + await upstreamProxyDb.upsertUpstreamProxyConfig({ + providerId: "anthropic", + mode: "cliproxyapi", + enabled: true, + }); + clearUpstreamProxyConfigCache("anthropic"); + const exec = await resolveExecutorWithProxy("anthropic"); + assert.equal(exec, getExecutor("cliproxyapi")); +}); + +test("mode 'fallback' returns a distinct wrapper owning its own execute()", async () => { + await upstreamProxyDb.upsertUpstreamProxyConfig({ + providerId: "openai", + mode: "fallback", + enabled: true, + }); + clearUpstreamProxyConfigCache("openai"); + const exec = await resolveExecutorWithProxy("openai"); + assert.notEqual(exec, getExecutor("openai")); + assert.notEqual(exec, getExecutor("cliproxyapi")); + assert.equal(typeof exec.execute, "function"); +});