Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 13 additions & 120 deletions open-sse/handlers/chatCore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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<string, string> | 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,
Expand Down Expand Up @@ -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;

Expand Down
72 changes: 72 additions & 0 deletions open-sse/handlers/chatCore/executionCredentials.ts
Original file line number Diff line number Diff line change
@@ -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<string, unknown> | 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,
},
};
}
98 changes: 98 additions & 0 deletions open-sse/handlers/chatCore/executorProxy.ts
Original file line number Diff line number Diff line change
@@ -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<string, string> | null;
}) => {
Comment on lines +57 to +66

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The current type signature for input is incomplete and hardcodes only a subset of the properties actually passed at runtime (missing contextEditing, clientHeaders, extendedContext, etc.). To ensure complete type safety and prevent this from going out of sync when new parameters are added to executors, dynamically derive the parameter type from the executor's execute method.

  const wrapper = Object.create(nativeExec) as ReturnType<typeof getExecutor>;
  wrapper.execute = async (input: Parameters<ReturnType<typeof getExecutor>["execute"]>[0]) => {

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;
}
Comment on lines +82 to +84

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To prevent potential runtime TypeError exceptions if result or result.response is undefined, use optional chaining to safely access the status code.

Suggested change
if (!isRetryableStatus(result.response.status)) {
return result;
}
if (!result?.response || !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;
}
Loading
Loading