From 28aa2d2645f84ec0561e7ee77dddcecb85333707 Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 22 Aug 2026 00:34:10 +0800 Subject: [PATCH 1/3] feat(proxy): disable memory-heavy features under high-concurrency mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit High-concurrency mode previously only reduced Redis debug snapshots and session observability writes. Memory-heavy coordination features — Replay, stream content gating, hedge-loser billing, client-abort retention, and response diagnostics — continued to run, undermining the CPU and IO savings the mode was designed to provide. ProxySession now exposes policy methods that return false when high-concurrency mode is active, causing the proxy pipeline to skip these features entirely. Forwarding, core billing, and quota enforcement remain enabled. Redis retention TTLs for circuit-breaker state and public-status projections are capped at 24 hours while the mode is active. The settings UI shows a toast warning listing the disabled features. --- messages/en/settings/config.json | 3 +- messages/ja/settings/config.json | 3 +- messages/ru/settings/config.json | 3 +- messages/zh-CN/settings/config.json | 3 +- messages/zh-TW/settings/config.json | 3 +- .../_components/system-settings-form.tsx | 5 +- src/app/v1/_lib/proxy/forwarder.ts | 57 +++++++-- .../v1/_lib/proxy/provider-request-filter.ts | 6 +- src/app/v1/_lib/proxy/replay/replay-guard.ts | 4 + src/app/v1/_lib/proxy/replay/replay-spool.ts | 4 + src/app/v1/_lib/proxy/request-filter.ts | 6 +- src/app/v1/_lib/proxy/response-handler.ts | 111 ++++++++++++++++-- src/app/v1/_lib/proxy/session.ts | 25 ++++ src/lib/public-status/rebuild-worker.ts | 25 +++- .../vendor-type-circuit-breaker-state.ts | 7 +- src/lib/system-settings/proxy-runtime.ts | 7 ++ .../proxy-runtime-high-concurrency.test.ts | 45 +++++++ tests/unit/proxy/replay-guard.test.ts | 11 ++ tests/unit/proxy/session.test.ts | 24 ++++ ...tings-form-upstream-error-message.test.tsx | 13 ++ 20 files changed, 330 insertions(+), 35 deletions(-) create mode 100644 tests/unit/lib/system-settings/proxy-runtime-high-concurrency.test.ts diff --git a/messages/en/settings/config.json b/messages/en/settings/config.json index c6f6a1832..551c4e9bb 100644 --- a/messages/en/settings/config.json +++ b/messages/en/settings/config.json @@ -59,7 +59,8 @@ "enableOpenaiResponsesWebsocket": "Enable OpenAI Responses WebSocket", "enableOpenaiResponsesWebsocketDesc": "When enabled, if a client opens a WebSocket connection to /v1/responses and the selected provider is a Codex type, CCH will attempt a sibling WebSocket to the upstream. If the upstream does not support WebSocket or the handshake fails, CCH gracefully falls back to standard HTTP Responses while keeping the client WebSocket open; the fallback is not counted toward circuit breakers. Non-WebSocket clients and non-Codex providers are unaffected.", "enableHighConcurrencyMode": "Enable High-Concurrency Mode", - "enableHighConcurrencyModeDesc": "When enabled, CCH disables part of the Redis debug snapshots and real-time session observability writes to reduce CPU and IO pressure under high RPM. Forwarding, rectifiers, fake-200 detection, billing, and quota enforcement remain unchanged, but Sessions debugging details may be reduced or delayed.", + "enableHighConcurrencyModeDesc": "When enabled, CCH disables memory-heavy Replay, stream gating, provider-racing loser billing, client-abort retention, and session diagnostics, in addition to Redis debug and observability writes. Forwarding, core billing, and quota enforcement remain enabled.", + "highConcurrencyModeWarning": "High-concurrency mode disables Replay, stream gating, racing-loser billing, client-abort retention, and session diagnostics.", "enableResponseFixer": "Enable Response Fixer", "enableResponseFixerDesc": "Automatically repairs common upstream response issues (encoding, SSE, truncated JSON). Enabled by default.", "enableThinkingSignatureRectifier": "Enable Thinking Signature Rectifier", diff --git a/messages/ja/settings/config.json b/messages/ja/settings/config.json index 27cb07459..c66612082 100644 --- a/messages/ja/settings/config.json +++ b/messages/ja/settings/config.json @@ -59,7 +59,8 @@ "enableOpenaiResponsesWebsocket": "OpenAI Responses WebSocket を有効化", "enableOpenaiResponsesWebsocketDesc": "有効にすると、クライアントが /v1/responses に WebSocket 接続し、かつ Codex タイプのプロバイダーが選択された場合、CCH は上流にも WebSocket 接続を試みます。上流が WebSocket をサポートしない、またはハンドシェイクに失敗した場合は、クライアント WebSocket を開いたまま通常の HTTP Responses に優雅にフォールバックします。このフォールバックはサーキットブレーカーにカウントされません。非 WebSocket クライアントと非 Codex プロバイダーの動作は変わりません。", "enableHighConcurrencyMode": "高並行モードを有効化", - "enableHighConcurrencyModeDesc": "有効にすると、高 RPM 時の CPU / IO 負荷を下げるため、Redis の一部デバッグスナップショットとリアルタイム Session 観測書き込みを停止します。転送、整流、fake 200 検知、課金、制限処理は維持されますが、Sessions のデバッグ詳細は減少または遅延する場合があります。", + "enableHighConcurrencyModeDesc": "有効にすると、Redis のデバッグスナップショットとリアルタイム Session 観測に加え、メモリ負荷の高い Replay、ストリームゲート、競合敗者の課金、クライアント中断保持、Session 診断を停止します。転送、基本課金、制限処理は維持されます。", + "highConcurrencyModeWarning": "高並行モードでは Replay、ストリームゲート、競合敗者の課金、クライアント中断保持、Session 診断を無効化します。", "enableResponseFixer": "レスポンス整流を有効化", "enableResponseFixerDesc": "上流応答の一般的な形式問題（エンコーディング、SSE、途切れた JSON）を自動修復します（既定で有効）。", "enableThinkingSignatureRectifier": "thinking 署名整流を有効化", diff --git a/messages/ru/settings/config.json b/messages/ru/settings/config.json index 3171cce69..e2e476459 100644 --- a/messages/ru/settings/config.json +++ b/messages/ru/settings/config.json @@ -59,7 +59,8 @@ "enableOpenaiResponsesWebsocket": "Включить OpenAI Responses WebSocket", "enableOpenaiResponsesWebsocketDesc": "Если включено, то когда клиент открывает WebSocket-соединение с /v1/responses и выбирается провайдер типа Codex, CCH попытается установить WebSocket-соединение с вышестоящим сервером. Если сервер не поддерживает WebSocket или рукопожатие не удастся, CCH плавно переключится на обычный HTTP Responses, сохраняя WebSocket клиента открытым; этот fallback не учитывается в circuit breaker. Клиенты без WebSocket и провайдеры, отличные от Codex, работают без изменений.", "enableHighConcurrencyMode": "Включить режим высокой нагрузки", - "enableHighConcurrencyModeDesc": "Если включено, CCH отключит часть Redis-снимков для отладки и записи real-time Session-наблюдения, чтобы снизить нагрузку на CPU и IO при высоком RPM. Пересылка, rectifier-логика, обнаружение fake 200, биллинг и лимиты сохраняются, но детализация отладки в Sessions может уменьшиться или запаздывать.", + "enableHighConcurrencyModeDesc": "При включении CCH отключает Redis-снимки для отладки и real-time-наблюдение Session, а также ресурсоёмкие Replay, stream-gate, тарификацию проигравших в гонке, сохранение при отмене клиентом и диагностику Session. Пересылка, базовый биллинг и лимиты сохраняются.", + "highConcurrencyModeWarning": "Режим высокой нагрузки отключает Replay, stream-gate, тарификацию проигравших в гонке, сохранение при отмене клиентом и диагностику Session.", "enableResponseFixer": "Включить исправление ответов", "enableResponseFixerDesc": "Автоматически исправляет распространённые проблемы ответа у провайдеров (кодировка, SSE, обрезанный JSON). Включено по умолчанию.", "enableThinkingSignatureRectifier": "Включить исправление thinking-signature", diff --git a/messages/zh-CN/settings/config.json b/messages/zh-CN/settings/config.json index 6b3e4ed08..a00ba9aa9 100644 --- a/messages/zh-CN/settings/config.json +++ b/messages/zh-CN/settings/config.json @@ -70,7 +70,8 @@ "enableOpenaiResponsesWebsocket": "启用 OpenAI Responses WebSocket", "enableOpenaiResponsesWebsocketDesc": "启用后，当客户端以 WebSocket 连接 /v1/responses 且选中 Codex 类型供应商时，CCH 会尝试与上游建立 WebSocket。若上游不支持或握手失败，将优雅降级到普通 HTTP Responses，客户端 WebSocket 保持打开；降级不计入熔断。非 WebSocket 客户端与非 Codex 供应商行为不变。", "enableHighConcurrencyMode": "启用高并发模式", - "enableHighConcurrencyModeDesc": "开启后，将关闭部分 Redis 调试快照与实时会话观测写入，以降低高并发下的 CPU 与 IO 开销。不会影响转发、整流、fake 200 检测、计费与限额，但 Sessions 调试详情会减少或延后。", + "enableHighConcurrencyModeDesc": "开启后，除 Redis 调试快照与实时会话观测写入外，还会关闭高内存占用的 Replay、流式门禁、竞速输家计费、客户端中断保留计费和会话诊断。转发、基础计费与限额仍会执行。", + "highConcurrencyModeWarning": "高并发模式将关闭 Replay、流式门禁、竞速输家计费、客户端中断保留计费和会话诊断。", "interceptAnthropicWarmupRequests": "拦截 Warmup 请求（Anthropic）", "interceptAnthropicWarmupRequestsDesc": "开启后，识别到 Claude Code 的 Warmup 探测请求将由 CCH 直接抢答短响应，避免访问上游供应商；该请求会记录在日志中，但不计费、不限流、不计入统计。", "enableThinkingSignatureRectifier": "启用 thinking 签名整流器", diff --git a/messages/zh-TW/settings/config.json b/messages/zh-TW/settings/config.json index 45b9debf2..cca77ae4f 100644 --- a/messages/zh-TW/settings/config.json +++ b/messages/zh-TW/settings/config.json @@ -59,7 +59,8 @@ "enableOpenaiResponsesWebsocket": "啟用 OpenAI Responses WebSocket", "enableOpenaiResponsesWebsocketDesc": "啟用後，當客戶端以 WebSocket 連線 /v1/responses 且命中 Codex 類型供應商時，CCH 會嘗試與上游建立 WebSocket 連線。若上游不支援或握手失敗，將優雅降級為一般 HTTP Responses，客戶端 WebSocket 保持開啟；降級不計入熔斷。非 WebSocket 客戶端與非 Codex 供應商行為不變。", "enableHighConcurrencyMode": "啟用高並發模式", - "enableHighConcurrencyModeDesc": "開啟後，將關閉部分 Redis 除錯快照與即時 Session 觀測寫入，以降低高並發下的 CPU 與 IO 開銷。轉發、整流、fake 200 偵測、計費與限額不受影響，但 Sessions 除錯詳情會減少或延後。", + "enableHighConcurrencyModeDesc": "開啟後，除 Redis 除錯快照與即時 Session 觀測寫入外，也會關閉高記憶體用量的 Replay、串流門控、競速輸家計費、客戶端中斷保留計費與 Session 診斷。轉發、基礎計費與限額仍會執行。", + "highConcurrencyModeWarning": "高並發模式將關閉 Replay、串流門控、競速輸家計費、客戶端中斷保留計費與 Session 診斷。", "enableResponseFixer": "啟用回應整流", "enableResponseFixerDesc": "自動修復上游回應中常見的編碼、SSE 與 JSON 格式問題（預設開啟）。", "enableThinkingSignatureRectifier": "啟用 thinking 簽名整流器", diff --git a/src/app/[locale]/settings/config/_components/system-settings-form.tsx b/src/app/[locale]/settings/config/_components/system-settings-form.tsx index 789c01b02..244b37a06 100644 --- a/src/app/[locale]/settings/config/_components/system-settings-form.tsx +++ b/src/app/[locale]/settings/config/_components/system-settings-form.tsx @@ -947,7 +947,10 @@ export function SystemSettingsForm({ setEnableHighConcurrencyMode(checked)} + onCheckedChange={(checked) => { + setEnableHighConcurrencyMode(checked); + if (checked) toast.warning(t("highConcurrencyModeWarning")); + }} disabled={isPending} /> diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 57390be3a..c9f4e56d1 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -1714,7 +1714,9 @@ export class ProxyForwarder { let gateChainAudit: ProviderChainItem["streamGate"]; const gateMode = resolveStreamGateMode(); const shouldRunPrecommitGate = - gateMode === "enforce" || session.replayState?.role === "owner"; + (typeof session.shouldRunStreamContentGate !== "function" || + session.shouldRunStreamContentGate()) && + (gateMode === "enforce" || session.replayState?.role === "owner"); if ( shouldRunPrecommitGate && response.body && @@ -1891,7 +1893,10 @@ export class ProxyForwarder { isJson && hasValidContentLength && contentLengthBytes <= NON_STREAM_BODY_INSPECTION_MAX_BYTES; - const shouldInspectBody = isHtml || !hasValidContentLength || shouldInspectJson; + const shouldInspectBody = + (typeof session.shouldParseResponseDiagnostics !== "function" || + session.shouldParseResponseDiagnostics()) && + (isHtml || !hasValidContentLength || shouldInspectJson); if (shouldStrictValidateReplayJson) { const validationLimit = getEnvConfig().REPLAY_MAX_PAYLOAD_BYTES; if (contentLengthBytes !== null && contentLengthBytes > validationLimit) { @@ -2927,7 +2932,11 @@ export class ProxyForwarder { // Final-phase request filter for Gemini: after headers built, before body serialization // Clone body to prevent in-place mutation of session.request.message on retries - if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { + if ( + !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); const bodyForFinal = structuredClone(bodyToSerialize); await requestFilterEngine.applyFinal(session, bodyForFinal, processedHeaders); @@ -2966,7 +2975,11 @@ export class ProxyForwarder { ); // Final-phase request filter for no-body requests (header-only operations) - if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { + if ( + !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal( session, @@ -3250,7 +3263,11 @@ export class ProxyForwarder { structuredClone(session.request.message) ) as Record; - if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { + if ( + !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal(session, logicalBody, processedHeaders); } @@ -3314,7 +3331,11 @@ export class ProxyForwarder { } // Final-phase request filter: after all provider overrides, before serialization - if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { + if ( + !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal(session, messageToSend, processedHeaders); } @@ -3354,7 +3375,11 @@ export class ProxyForwarder { } } else { // No body (GET/HEAD): still run final-phase for header-only filter operations - if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { + if ( + !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal( session, @@ -4388,7 +4413,9 @@ export class ProxyForwarder { const rawCrossProviderFallbackEnabled = session.isRawCrossProviderFallbackEnabled(); // 竞速输家计费开关：开启时落败供应商不被直接掐断，而是后台 drain 并计费。 - const billHedgeLosers = (await getCachedSystemSettings()).billHedgeLosers === true; + const billHedgeLosers = + (typeof session.shouldBillHedgeLosers !== "function" || session.shouldBillHedgeLosers()) && + (await getCachedSystemSettings()).billHedgeLosers === true; const launchedProviderIds = new Set(); let launchedProviderCount = 0; let settled = false; @@ -4455,6 +4482,13 @@ export class ProxyForwarder { // 不取消连接：读到流自然结束（或超时/容量上限）后，复用赢家相同的计费链， // 把费用异步累加回原请求行。幂等（loserBillingStarted 守卫），失败静默。 const startLoserBilling = (attempt: StreamingHedgeAttempt) => { + if (typeof session.shouldBillHedgeLosers === "function" && !session.shouldBillHedgeLosers()) { + const cancel = attempt.reader?.cancel("high_concurrency_loser_billing_disabled"); + cancel?.catch(() => undefined); + releaseAttemptAgent(attempt); + return; + } + if (attempt.loserBillingStarted) return; attempt.loserBillingStarted = true; @@ -4779,6 +4813,8 @@ export class ProxyForwarder { // 「首个有效内容帧」。 // 级联阈值计时器保持不动——内容慢的 attempt 不提交，自动触发下一候选竞速。 const hedgeGateFamily = + (typeof session.shouldRunStreamContentGate !== "function" || + session.shouldRunStreamContentGate()) && (resolveStreamGateMode() === "enforce" || session.replayState?.role === "owner") && session.getEndpointPolicy().kind !== "raw_passthrough" ? mapProviderTypeToFamily(attempt.provider.providerType) @@ -5436,7 +5472,10 @@ export class ProxyForwarder { // Discovery uses the same opt-in loser billing switch as legacy Hedge. The // attempt is only kept alive after a winner commits when it already has a // protocol-valid prefix and a readable response body (see cancelLosers). - const billHedgeLosers = settings.billHedgeLosers === true && session.messageContext?.id != null; + const billHedgeLosers = + (typeof session.shouldBillHedgeLosers !== "function" || session.shouldBillHedgeLosers()) && + settings.billHedgeLosers === true && + session.messageContext?.id != null; const coordinator = new DiscoveryCoordinator({ concurrency, maxRounds }); const discoveryMetrics = new DiscoveryRequestMetrics( { diff --git a/src/app/v1/_lib/proxy/provider-request-filter.ts b/src/app/v1/_lib/proxy/provider-request-filter.ts index 68242b029..211bed97d 100644 --- a/src/app/v1/_lib/proxy/provider-request-filter.ts +++ b/src/app/v1/_lib/proxy/provider-request-filter.ts @@ -9,7 +9,11 @@ import type { ProxySession } from "./session"; */ export class ProxyProviderRequestFilter { static async ensure(session: ProxySession): Promise { - if (session.getEndpointPolicy().bypassRequestFilters) { + if ( + session.getEndpointPolicy().bypassRequestFilters || + (typeof session.shouldApplyContentTransforms === "function" && + session.shouldApplyContentTransforms() === false) + ) { return; } diff --git a/src/app/v1/_lib/proxy/replay/replay-guard.ts b/src/app/v1/_lib/proxy/replay/replay-guard.ts index 6b7415a9a..bb69c8269 100644 --- a/src/app/v1/_lib/proxy/replay/replay-guard.ts +++ b/src/app/v1/_lib/proxy/replay/replay-guard.ts @@ -39,6 +39,10 @@ const ATTACH_MAX_WAIT_MS = 10 * 60 * 1000; export class ProxyReplayGuard { static async ensure(session: ProxySession): Promise { + if (typeof session.shouldUseRequestReplay === "function" && !session.shouldUseRequestReplay()) { + return null; + } + try { // guard 位于 provider 步骤之前：先刷新运行时覆写快照，管理端刚保存的 // replayEnabled 首个请求即生效（底层系统设置缓存有 TTL，常态为缓存命中） diff --git a/src/app/v1/_lib/proxy/replay/replay-spool.ts b/src/app/v1/_lib/proxy/replay/replay-spool.ts index 23318602f..41ce0f65b 100644 --- a/src/app/v1/_lib/proxy/replay/replay-spool.ts +++ b/src/app/v1/_lib/proxy/replay/replay-spool.ts @@ -570,6 +570,10 @@ export function createReplaySpoolIfOwner( delivery: ReplayDelivery = "stream", options: ReplaySpoolOptions = {} ): ReplaySpool | null { + if (typeof session.shouldUseRequestReplay === "function" && !session.shouldUseRequestReplay()) { + return null; + } + const replayState = session.replayState; if (replayState?.role !== "owner") return null; const declineOwnership = (): null => { diff --git a/src/app/v1/_lib/proxy/request-filter.ts b/src/app/v1/_lib/proxy/request-filter.ts index 68e45aae7..89757009e 100644 --- a/src/app/v1/_lib/proxy/request-filter.ts +++ b/src/app/v1/_lib/proxy/request-filter.ts @@ -12,7 +12,11 @@ import type { ProxySession } from "./session"; */ export class ProxyRequestFilter { static async ensure(session: ProxySession): Promise { - if (session.getEndpointPolicy().bypassRequestFilters) { + if ( + session.getEndpointPolicy().bypassRequestFilters || + (typeof session.shouldApplyContentTransforms === "function" && + session.shouldApplyContentTransforms() === false) + ) { return; } diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index 33aa95cfd..ccafb895f 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -57,6 +57,7 @@ import { recordAffinityWinner, tombstoneAffinityOnFailure } from "./affinity/aff import { bindClientAbortListener } from "./client-abort-listener"; import { CLIENT_ABORT_METER_MAX_RETAINED_BYTES, + type ClientAbortMeteringObserver, createClientAbortMeteringObserver, } from "./client-abort-metering"; import { @@ -1871,7 +1872,12 @@ function finalizeDeferredStreamingFinalizationIfNeeded( const billHedgeLosers = meta?.billHedgeLosers === true; const hasDiscoveryBindingIntent = meta?.bindingIntent === "create" || meta?.bindingIntent === "renew"; - const completionInspection = inspectStreamCompletion(allContent, session.originalFormat); + const parseResponseDiagnostics = + typeof session.shouldParseResponseDiagnostics !== "function" || + session.shouldParseResponseDiagnostics(); + const completionInspection = parseResponseDiagnostics + ? inspectStreamCompletion(allContent, session.originalFormat) + : { hasMarker: false, hasProtocolError: false }; const completionMarkerMissingForBinding = meta?.requiresCompletionMarkerForBinding === true && hasDiscoveryBindingIntent && @@ -1954,6 +1960,11 @@ function finalizeDeferredStreamingFinalizationIfNeeded( : bodyDetected; let clientAbortGateUsage: FinalizeDeferredStreamingResult["clientAbortGateUsage"]; const clientAbortCompleteSuccess = (() => { + if ( + typeof session.shouldRetainClientAbortBilling === "function" && + !session.shouldRetainClientAbortBilling() + ) + return false; if (!clientAborted || upstreamStatusCode < 200 || upstreamStatusCode >= 300) { return false; } @@ -1991,7 +2002,9 @@ function finalizeDeferredStreamingFinalizationIfNeeded( let statusCodeInferred = false; let statusCodeInferenceMatcherId: string | undefined; if (detected.isError) { - const inferred = inferUpstreamErrorStatusCodeFromText(allContent); + const inferred = parseResponseDiagnostics + ? inferUpstreamErrorStatusCodeFromText(allContent) + : null; if (inferred) { effectiveStatusCode = inferred.statusCode; statusCodeInferred = true; @@ -2005,8 +2018,12 @@ function finalizeDeferredStreamingFinalizationIfNeeded( errorMessage = null; } else if (streamEndedNormally && upstreamStatusCode >= 400) { effectiveStatusCode = upstreamStatusCode; - const upstreamError = detectUpstreamErrorFromSseOrJsonText(allContent); - errorMessage = upstreamError.isError ? upstreamError.code : `HTTP ${upstreamStatusCode}`; + if (parseResponseDiagnostics) { + const upstreamError = detectUpstreamErrorFromSseOrJsonText(allContent); + errorMessage = upstreamError.isError ? upstreamError.code : `HTTP ${upstreamStatusCode}`; + } else { + errorMessage = `HTTP ${upstreamStatusCode}`; + } } else if (clientAborted) { effectiveStatusCode = 499; errorMessage = "CLIENT_ABORTED"; @@ -2018,9 +2035,12 @@ function finalizeDeferredStreamingFinalizationIfNeeded( effectiveStatusCode = upstreamStatusCode; if (upstreamStatusCode >= 400) { - // 非200错误状态码：解析JSON错误响应 - const detected = detectUpstreamErrorFromSseOrJsonText(allContent); - errorMessage = detected.isError ? detected.code : `HTTP ${upstreamStatusCode}`; + if (parseResponseDiagnostics) { + const detected = detectUpstreamErrorFromSseOrJsonText(allContent); + errorMessage = detected.isError ? detected.code : `HTTP ${upstreamStatusCode}`; + } else { + errorMessage = `HTTP ${upstreamStatusCode}`; + } } else { // 2xx 成功状态码 errorMessage = null; @@ -2470,7 +2490,11 @@ export class ProxyResponseHandler { } let fixedResponse = response; - if (!session.getEndpointPolicy().bypassResponseRectifier) { + if ( + !session.getEndpointPolicy().bypassResponseRectifier && + (typeof session.shouldApplyContentTransforms !== "function" || + session.shouldApplyContentTransforms()) + ) { try { // raw passthrough 端点跳过 ResponseFixer，也跳过其中的 Responses 输出归一化。 fixedResponse = await ResponseFixer.process(session, response); @@ -3572,10 +3596,26 @@ export class ProxyResponseHandler { session.getEndpointPolicy().kind === "raw_passthrough" ? null : mapProviderTypeToFamily(provider.providerType); - let streamProtocolObserver = nativeStreamProtocolFamily - ? createStreamProtocolObserver(nativeStreamProtocolFamily) - : null; - const clientAbortMeter = createClientAbortMeteringObserver(session.originalFormat); + let streamProtocolObserver = + nativeStreamProtocolFamily && + (typeof session.shouldParseResponseDiagnostics !== "function" || + session.shouldParseResponseDiagnostics()) + ? createStreamProtocolObserver(nativeStreamProtocolFamily) + : null; + const clientAbortMeter: ClientAbortMeteringObserver = + typeof session.shouldRetainClientAbortBilling !== "function" || + session.shouldRetainClientAbortBilling() + ? createClientAbortMeteringObserver(session.originalFormat) + : { + observe: () => ({ billingComplete: false }), + finish: () => ({ + text: "", + billingComplete: false, + retainedBytes: 0, + skippedOversizedFrames: 0, + protocolFailure: null, + }), + }; let protocolObservedBeforeProcessing = false; // --- GEMINI STREAM HANDLING --- @@ -3600,6 +3640,11 @@ export class ProxyResponseHandler { // F1 shadow 遥测：enforce 已在 forwarder 作用于该流量，shadow 观察同样不留盲区 let passthroughShadowObserver = (() => { + if ( + typeof session.shouldParseResponseDiagnostics === "function" && + !session.shouldParseResponseDiagnostics() + ) + return null; if (resolveStreamGateMode() !== "shadow") return null; if (session.getEndpointPolicy().kind === "raw_passthrough") return null; const family = mapProviderTypeToFamily(provider.providerType); @@ -3634,6 +3679,21 @@ export class ProxyResponseHandler { }; const startPassthroughDrain = (reason?: unknown) => { if (passthroughPump.getState() === "closed") return; + if ( + typeof session.shouldRetainClientAbortBilling === "function" && + !session.shouldRetainClientAbortBilling() + ) { + passthroughClientDetached = true; + const abortError = + reason instanceof Error ? reason : new Error("client_detached_high_concurrency"); + streamTextAccumulator.discardRetainedBytes(); + streamProtocolObserver = null; + passthroughShadowObserver = null; + abortPassthroughTransport(abortError); + passthroughPump.startDrain(abortError); + passthroughPump.cancelSource(abortError); + return; + } if (passthroughClientDetached) { passthroughPump.startDrain(reason); return; @@ -4409,6 +4469,15 @@ export class ProxyResponseHandler { }; const handleClientAbort = (reason?: unknown) => { if (responsePump?.getState() === "closed") return; + if ( + typeof session.shouldRetainClientAbortBilling === "function" && + !session.shouldRetainClientAbortBilling() + ) { + clientDetachHandled = true; + responsePump?.startDrain(reason ?? "client_detached_high_concurrency"); + responsePump?.cancelSource(reason ?? "client_detached_high_concurrency"); + return; + } if (clientDetachHandled) { responsePump?.startDrain(reason ?? "client_detached"); return; @@ -4589,6 +4658,11 @@ export class ProxyResponseHandler { const awaitFinalization = (promise: Promise): Promise => raceWithDeadline(promise, finalizationDeadlineAtMs, "stream_finalization_timeout"); const detachedProtocolObservation: StreamProtocolObservation | null = (() => { + if ( + typeof session.shouldParseResponseDiagnostics === "function" && + !session.shouldParseResponseDiagnostics() + ) + return null; if (!clientDetachHandled || streamProtocolObserver) return null; const metering = clientAbortMeter.finish(); return { @@ -4686,6 +4760,8 @@ export class ProxyResponseHandler { | undefined; if ( provider.providerType === "codex" && + (typeof session.shouldParseResponseDiagnostics !== "function" || + session.shouldParseResponseDiagnostics()) && effectiveStatusCode >= 200 && effectiveStatusCode < 300 && session.sessionId && @@ -5017,6 +5093,11 @@ export class ProxyResponseHandler { // F1 shadow 模式：旁路逐帧分类，记录「首非空字节 vs 首有效内容」的分歧与延迟差， // 不缓冲、不 failover，仅用于 enforce 灰度前评估误判率。 shadowGateObserver = (() => { + if ( + typeof session.shouldParseResponseDiagnostics === "function" && + !session.shouldParseResponseDiagnostics() + ) + return null; if (resolveStreamGateMode() !== "shadow") return null; if (session.getEndpointPolicy().kind === "raw_passthrough") return null; const family = mapProviderTypeToFamily(provider.providerType); @@ -6384,6 +6465,12 @@ export async function finalizeHedgeLoserBilling(params: { billingContext, } = params; + if ( + typeof loserSession.shouldBillHedgeLosers === "function" && + !loserSession.shouldBillHedgeLosers() + ) + return null; + try { if (isNonBillingUsageEndpoint(loserSession)) { return null; diff --git a/src/app/v1/_lib/proxy/session.ts b/src/app/v1/_lib/proxy/session.ts index d0d4443f7..af6a8cea3 100644 --- a/src/app/v1/_lib/proxy/session.ts +++ b/src/app/v1/_lib/proxy/session.ts @@ -582,6 +582,31 @@ export class ProxySession { return !this.highConcurrencyModeEnabled; } + /** High-concurrency mode disables optional body-heavy coordination features. */ + shouldUseRequestReplay(): boolean { + return !this.highConcurrencyModeEnabled; + } + + shouldRunStreamContentGate(): boolean { + return !this.highConcurrencyModeEnabled; + } + + shouldRetainClientAbortBilling(): boolean { + return !this.highConcurrencyModeEnabled; + } + + shouldBillHedgeLosers(): boolean { + return !this.highConcurrencyModeEnabled; + } + + shouldParseResponseDiagnostics(): boolean { + return !this.highConcurrencyModeEnabled; + } + + shouldApplyContentTransforms(): boolean { + return !this.highConcurrencyModeEnabled; + } + addSpecialSetting(setting: SpecialSetting): void { this.specialSettings.push(setting); } diff --git a/src/lib/public-status/rebuild-worker.ts b/src/lib/public-status/rebuild-worker.ts index 95e36e5ee..67d5f9c10 100644 --- a/src/lib/public-status/rebuild-worker.ts +++ b/src/lib/public-status/rebuild-worker.ts @@ -1,4 +1,8 @@ import { getRedisClient } from "@/lib/redis"; +import { + getProxyRuntimeSettings, + resolveRedisRetentionTtlSeconds, +} from "@/lib/system-settings/proxy-runtime"; import { publishCurrentPublicStatusConfigProjection } from "./config-publisher"; import { readCurrentInternalPublicStatusConfigSnapshot } from "./config-snapshot"; import { @@ -120,6 +124,7 @@ async function publishPublicStatusProjection(input: { rollupSampleCount: number; groups: unknown; }): Promise { + await getProxyRuntimeSettings(); const snapshotKey = buildPublicStatusCurrentSnapshotKey({ intervalMinutes: input.intervalMinutes, rangeHours: input.rangeHours, @@ -195,19 +200,19 @@ async function publishPublicStatusProjection(input: { input.redis, snapshotKey, JSON.stringify(snapshotRecord), - GENERATION_PROJECTION_TTL_SECONDS + resolveRedisRetentionTtlSeconds(GENERATION_PROJECTION_TTL_SECONDS) ); await setWithTtl( input.redis, seriesKey, JSON.stringify(seriesRecord), - GENERATION_PROJECTION_TTL_SECONDS + resolveRedisRetentionTtlSeconds(GENERATION_PROJECTION_TTL_SECONDS) ); await setWithTtl( input.redis, versionedManifestKey, JSON.stringify(manifestRecord), - GENERATION_PROJECTION_TTL_SECONDS + resolveRedisRetentionTtlSeconds(GENERATION_PROJECTION_TTL_SECONDS) ); if (typeof input.redis.get === "function") { let existingCurrentManifest: { configVersion?: string; coveredTo?: string } | null = null; @@ -221,10 +226,20 @@ async function publishPublicStatusProjection(input: { } if (shouldPromoteCurrentManifest(existingCurrentManifest, manifestRecord)) { - await input.redis.set(currentManifestKey, JSON.stringify(manifestRecord)); + await setWithTtl( + input.redis, + currentManifestKey, + JSON.stringify(manifestRecord), + resolveRedisRetentionTtlSeconds(GENERATION_PROJECTION_TTL_SECONDS) + ); } } else { - await input.redis.set(currentManifestKey, JSON.stringify(manifestRecord)); + await setWithTtl( + input.redis, + currentManifestKey, + JSON.stringify(manifestRecord), + resolveRedisRetentionTtlSeconds(GENERATION_PROJECTION_TTL_SECONDS) + ); } if (input.redis.del) { await input.redis.del(snapshotTempKey, seriesTempKey); diff --git a/src/lib/redis/vendor-type-circuit-breaker-state.ts b/src/lib/redis/vendor-type-circuit-breaker-state.ts index 1c893db73..efe3578a8 100644 --- a/src/lib/redis/vendor-type-circuit-breaker-state.ts +++ b/src/lib/redis/vendor-type-circuit-breaker-state.ts @@ -1,6 +1,10 @@ import "server-only"; import { logger } from "@/lib/logger"; +import { + getProxyRuntimeSettings, + resolveRedisRetentionTtlSeconds, +} from "@/lib/system-settings/proxy-runtime"; import type { ProviderType } from "@/types/provider"; import { getRedisClient } from "./client"; @@ -85,10 +89,11 @@ export async function saveVendorTypeCircuitState( } try { + await getProxyRuntimeSettings(); const key = getStateKey(vendorId, providerType); const data = serializeState(state); await redis.hset(key, data); - await redis.expire(key, STATE_TTL_SECONDS); + await redis.expire(key, resolveRedisRetentionTtlSeconds(STATE_TTL_SECONDS)); } catch (error) { logger.warn("[VendorTypeCircuitState] Failed to save to Redis", { vendorId, diff --git a/src/lib/system-settings/proxy-runtime.ts b/src/lib/system-settings/proxy-runtime.ts index 48c17e3d6..a6e81697b 100644 --- a/src/lib/system-settings/proxy-runtime.ts +++ b/src/lib/system-settings/proxy-runtime.ts @@ -27,6 +27,7 @@ export interface ProxyRuntimeSettings { // 最近一次成功读取的快照；同步热路径消费，异步读取与开机预热负责保鲜。 let lastKnown: ProxyRuntimeSettings | null = null; +let highConcurrencyModeEnabled = false; function envReplayDefault(): boolean { try { @@ -76,6 +77,7 @@ export async function getProxyRuntimeSettings(): Promise { cacheEffectivenessEnabled: settings.cacheEffectivenessEnabled ?? envCacheEffectivenessDefault(), }; + highConcurrencyModeEnabled = settings.enableHighConcurrencyMode === true; return lastKnown; } catch { // getCachedSystemSettings 自身已 fail-safe；此处兜底其意外异常 @@ -94,3 +96,8 @@ export function getCachedProxyRuntimeSettings(): ProxyRuntimeSettings | null { export function isCacheEffectivenessEnabled(): boolean { return lastKnown?.cacheEffectivenessEnabled ?? envCacheEffectivenessDefault(); } + +/** Shrinks long-lived Redis projections while high-concurrency mode is active. */ +export function resolveRedisRetentionTtlSeconds(defaultTtlSeconds: number): number { + return highConcurrencyModeEnabled ? Math.min(defaultTtlSeconds, 24 * 60 * 60) : defaultTtlSeconds; +} diff --git a/tests/unit/lib/system-settings/proxy-runtime-high-concurrency.test.ts b/tests/unit/lib/system-settings/proxy-runtime-high-concurrency.test.ts new file mode 100644 index 000000000..a23aaffce --- /dev/null +++ b/tests/unit/lib/system-settings/proxy-runtime-high-concurrency.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const settingsMock = vi.hoisted(() => ({ + current: { + enableHighConcurrencyMode: false, + streamGateMode: "enforce" as const, + affinityIgnoreClientSessionId: true, + replayEnabled: true, + replayCacheTtlMinutes: 30, + cacheEffectivenessEnabled: true, + }, +})); + +vi.mock("@/lib/config/system-settings-cache", () => ({ + getCachedSystemSettings: vi.fn(async () => settingsMock.current), +})); + +describe("proxy runtime high-concurrency Redis retention", () => { + beforeEach(() => { + vi.resetModules(); + settingsMock.current.enableHighConcurrencyMode = false; + }); + + it("keeps the normal retention when high-concurrency mode is disabled", async () => { + const { getProxyRuntimeSettings, resolveRedisRetentionTtlSeconds } = await import( + "@/lib/system-settings/proxy-runtime" + ); + + await getProxyRuntimeSettings(); + + expect(resolveRedisRetentionTtlSeconds(2_592_000)).toBe(2_592_000); + }); + + it("caps long retention at one day when high-concurrency mode is enabled", async () => { + settingsMock.current.enableHighConcurrencyMode = true; + const { getProxyRuntimeSettings, resolveRedisRetentionTtlSeconds } = await import( + "@/lib/system-settings/proxy-runtime" + ); + + await getProxyRuntimeSettings(); + + expect(resolveRedisRetentionTtlSeconds(2_592_000)).toBe(86_400); + expect(resolveRedisRetentionTtlSeconds(300)).toBe(300); + }); +}); diff --git a/tests/unit/proxy/replay-guard.test.ts b/tests/unit/proxy/replay-guard.test.ts index 2842e9d3d..a5322d466 100644 --- a/tests/unit/proxy/replay-guard.test.ts +++ b/tests/unit/proxy/replay-guard.test.ts @@ -98,6 +98,7 @@ vi.mock("@/repository/message", () => ({ interface GuardSessionOverrides { message?: Record; headers?: Record; + highConcurrency?: boolean; apiKey?: string | null; sessionIdentity?: { identity: string; @@ -133,6 +134,7 @@ function makeSession(overrides: GuardSessionOverrides = {}): ProxySession { getOriginalModel: () => "claude-sonnet-4", getEndpoint: () => "/v1/messages", getMessagesLength: () => 1, + shouldUseRequestReplay: () => !overrides.highConcurrency, getSessionIdentityMetadata: () => overrides.sessionIdentity ?? { identity: "sess-1", @@ -187,6 +189,15 @@ describe("ProxyReplayGuard：放行路径", () => { expect(session.replayState).toBeNull(); }); + it("高并发模式直接放行，不计算 identity 或触碰 Replay 存储", async () => { + const session = makeSession({ highConcurrency: true }); + + await expect(ProxyReplayGuard.ensure(session)).resolves.toBeNull(); + expect(storeControl.getMeta).not.toHaveBeenCalled(); + expect(storeControl.findCompleted).not.toHaveBeenCalled(); + expect(storeControl.tryClaimOwner).not.toHaveBeenCalled(); + }); + it("非流式请求不参与 replay", async () => { const session = makeSession({ message: { stream: false, model: "claude-sonnet-4" } }); diff --git a/tests/unit/proxy/session.test.ts b/tests/unit/proxy/session.test.ts index 5c77a0a29..ec58e49ee 100644 --- a/tests/unit/proxy/session.test.ts +++ b/tests/unit/proxy/session.test.ts @@ -163,6 +163,30 @@ describe("ProxySession endpoint policy", () => { }); }); +describe("ProxySession high-concurrency policy", () => { + it("closes optional body-heavy features while preserving the base session", () => { + const session = createSession({ redirectedModel: null }); + + expect(session.shouldUseRequestReplay()).toBe(true); + expect(session.shouldRunStreamContentGate()).toBe(true); + expect(session.shouldRetainClientAbortBilling()).toBe(true); + expect(session.shouldBillHedgeLosers()).toBe(true); + expect(session.shouldParseResponseDiagnostics()).toBe(true); + expect(session.shouldApplyContentTransforms()).toBe(true); + + session.setHighConcurrencyModeEnabled(true); + + expect(session.shouldUseRequestReplay()).toBe(false); + expect(session.shouldRunStreamContentGate()).toBe(false); + expect(session.shouldRetainClientAbortBilling()).toBe(false); + expect(session.shouldBillHedgeLosers()).toBe(false); + expect(session.shouldParseResponseDiagnostics()).toBe(false); + expect(session.shouldApplyContentTransforms()).toBe(false); + expect(session.shouldPersistSessionDebugArtifacts()).toBe(false); + expect(session.shouldTrackSessionObservability()).toBe(false); + }); +}); + describe("ProxySession.getCachedPriceDataByBillingSource", () => { it("配置 = original 时应优先使用原始模型", async () => { const originalPriceData: ModelPriceData = { input_cost_per_token: 1, output_cost_per_token: 2 }; diff --git a/tests/unit/settings/system-settings-form-upstream-error-message.test.tsx b/tests/unit/settings/system-settings-form-upstream-error-message.test.tsx index 465e00561..36b170a30 100644 --- a/tests/unit/settings/system-settings-form-upstream-error-message.test.tsx +++ b/tests/unit/settings/system-settings-form-upstream-error-message.test.tsx @@ -205,6 +205,19 @@ describe("SystemSettingsForm upstream error message toggles", () => { unmount(); }); + test("开启高并发模式时提示将停用的高开销功能", () => { + const { unmount } = renderForm(buildSettings()); + + clickSwitch("enable-high-concurrency-mode"); + + expect(sonnerMocks.toast.warning).toHaveBeenCalledWith( + "High-concurrency mode disables Replay, stream gating, racing-loser billing, client-abort retention, and session diagnostics." + ); + expect(getSwitch("enable-high-concurrency-mode").getAttribute("aria-checked")).toBe("true"); + + unmount(); + }); + test("旧开关仍可独立提交，不会连带修改新开关", async () => { systemConfigActionMocks.saveSystemSettings.mockResolvedValueOnce({ ok: true, From 4492c36a0543a55fbfc8810b34e2557db8f9d509 Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 22 Aug 2026 01:01:53 +0800 Subject: [PATCH 2/3] fix(proxy): keep fake-200 detection active in high-concurrency mode Body inspection for fake-200 HTML responses was gated on shouldParseResponseDiagnostics, which returns false in high-concurrency mode. This inadvertently disabled the core failover guard that triggers provider switching on blocked responses. The diagnostics gate is removed so fake-200 detection remains active regardless of concurrency mode. --- src/app/v1/_lib/proxy/forwarder.ts | 7 ++-- .../proxy-forwarder-fake-200-html.test.ts | 40 +++++++++++++++++++ 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index c9f4e56d1..99b682bf3 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -1893,10 +1893,9 @@ export class ProxyForwarder { isJson && hasValidContentLength && contentLengthBytes <= NON_STREAM_BODY_INSPECTION_MAX_BYTES; - const shouldInspectBody = - (typeof session.shouldParseResponseDiagnostics !== "function" || - session.shouldParseResponseDiagnostics()) && - (isHtml || !hasValidContentLength || shouldInspectJson); + // Fake-200 detection is a core failover guard, so it remains active in + // high-concurrency mode even though optional diagnostics are disabled. + const shouldInspectBody = isHtml || !hasValidContentLength || shouldInspectJson; if (shouldStrictValidateReplayJson) { const validationLimit = getEnvConfig().REPLAY_MAX_PAYLOAD_BYTES; if (contentLengthBytes !== null && contentLengthBytes > validationLimit) { diff --git a/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts b/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts index 02a5bf89b..ce5177e10 100644 --- a/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts +++ b/tests/unit/proxy/proxy-forwarder-fake-200-html.test.ts @@ -392,6 +392,46 @@ describe("ProxyForwarder - fake 200 HTML body", () => { expect(mocks.recordSuccess).not.toHaveBeenCalledWith(1); }); + test("高并发模式仍保留非流式 fake-200 核心故障切换", async () => { + const provider1 = createProvider({ id: 1, name: "p1", key: "k1", maxRetryAttempts: 1 }); + const provider2 = createProvider({ id: 2, name: "p2", key: "k2", maxRetryAttempts: 1 }); + const session = createSession(); + session.setHighConcurrencyModeEnabled(true); + session.setProvider(provider1); + + mocks.pickRandomProviderWithExclusion.mockResolvedValueOnce(provider2); + const doForward = vi.spyOn(ProxyForwarder as any, "doForward"); + const htmlBody = "blocked"; + const okJson = JSON.stringify({ type: "message", content: [{ type: "text", text: "ok" }] }); + + doForward + .mockResolvedValueOnce( + new Response(htmlBody, { + status: 200, + headers: { + "content-type": "text/html; charset=utf-8", + "content-length": String(htmlBody.length), + }, + }) + ) + .mockResolvedValueOnce( + new Response(okJson, { + status: 200, + headers: { "content-type": "application/json", "content-length": String(okJson.length) }, + }) + ); + + const response = await ProxyForwarder.send(session); + + await expect(response.text()).resolves.toBe(okJson); + expect(doForward).toHaveBeenCalledTimes(2); + expect(mocks.recordFailure).toHaveBeenCalledWith( + provider1.id, + expect.objectContaining({ message: "FAKE_200_HTML_BODY" }) + ); + expect(mocks.recordSuccess).toHaveBeenCalledWith(provider2.id); + }); + test("200 + text/html 但 body 是 JSON error 也应视为失败并切换供应商", async () => { const provider1 = createProvider({ id: 1, name: "p1", key: "k1", maxRetryAttempts: 1 }); const provider2 = createProvider({ id: 2, name: "p2", key: "k2", maxRetryAttempts: 1 }); From a22d3379515009978399f1a07c9d35d2c0c0bd6f Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 22 Aug 2026 01:11:17 +0800 Subject: [PATCH 3/3] fix(proxy): keep request filters active in high-concurrency mode Request filters and the response rectifier were gated on shouldApplyContentTransforms, which returns false in high-concurrency mode. This inadvertently disabled content transformation and response normalization that providers rely on. The gate is removed so filters and the response fixer run regardless of concurrency mode. --- src/app/v1/_lib/proxy/forwarder.ts | 30 ++++--------------- .../v1/_lib/proxy/provider-request-filter.ts | 6 +--- src/app/v1/_lib/proxy/request-filter.ts | 6 +--- src/app/v1/_lib/proxy/response-handler.ts | 6 +--- src/app/v1/_lib/proxy/session.ts | 4 --- tests/unit/proxy/session.test.ts | 2 -- 6 files changed, 8 insertions(+), 46 deletions(-) diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 99b682bf3..93118f906 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -2931,11 +2931,7 @@ export class ProxyForwarder { // Final-phase request filter for Gemini: after headers built, before body serialization // Clone body to prevent in-place mutation of session.request.message on retries - if ( - !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); const bodyForFinal = structuredClone(bodyToSerialize); await requestFilterEngine.applyFinal(session, bodyForFinal, processedHeaders); @@ -2974,11 +2970,7 @@ export class ProxyForwarder { ); // Final-phase request filter for no-body requests (header-only operations) - if ( - !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal( session, @@ -3262,11 +3254,7 @@ export class ProxyForwarder { structuredClone(session.request.message) ) as Record; - if ( - !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal(session, logicalBody, processedHeaders); } @@ -3330,11 +3318,7 @@ export class ProxyForwarder { } // Final-phase request filter: after all provider overrides, before serialization - if ( - !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal(session, messageToSend, processedHeaders); } @@ -3374,11 +3358,7 @@ export class ProxyForwarder { } } else { // No body (GET/HEAD): still run final-phase for header-only filter operations - if ( - !ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!ProxyForwarder.getEndpointPolicy(session).bypassRequestFilters) { const { requestFilterEngine } = await import("@/lib/request-filter-engine"); await requestFilterEngine.applyFinal( session, diff --git a/src/app/v1/_lib/proxy/provider-request-filter.ts b/src/app/v1/_lib/proxy/provider-request-filter.ts index 211bed97d..68242b029 100644 --- a/src/app/v1/_lib/proxy/provider-request-filter.ts +++ b/src/app/v1/_lib/proxy/provider-request-filter.ts @@ -9,11 +9,7 @@ import type { ProxySession } from "./session"; */ export class ProxyProviderRequestFilter { static async ensure(session: ProxySession): Promise { - if ( - session.getEndpointPolicy().bypassRequestFilters || - (typeof session.shouldApplyContentTransforms === "function" && - session.shouldApplyContentTransforms() === false) - ) { + if (session.getEndpointPolicy().bypassRequestFilters) { return; } diff --git a/src/app/v1/_lib/proxy/request-filter.ts b/src/app/v1/_lib/proxy/request-filter.ts index 89757009e..68e45aae7 100644 --- a/src/app/v1/_lib/proxy/request-filter.ts +++ b/src/app/v1/_lib/proxy/request-filter.ts @@ -12,11 +12,7 @@ import type { ProxySession } from "./session"; */ export class ProxyRequestFilter { static async ensure(session: ProxySession): Promise { - if ( - session.getEndpointPolicy().bypassRequestFilters || - (typeof session.shouldApplyContentTransforms === "function" && - session.shouldApplyContentTransforms() === false) - ) { + if (session.getEndpointPolicy().bypassRequestFilters) { return; } diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index ccafb895f..e218fbb12 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -2490,11 +2490,7 @@ export class ProxyResponseHandler { } let fixedResponse = response; - if ( - !session.getEndpointPolicy().bypassResponseRectifier && - (typeof session.shouldApplyContentTransforms !== "function" || - session.shouldApplyContentTransforms()) - ) { + if (!session.getEndpointPolicy().bypassResponseRectifier) { try { // raw passthrough 端点跳过 ResponseFixer，也跳过其中的 Responses 输出归一化。 fixedResponse = await ResponseFixer.process(session, response); diff --git a/src/app/v1/_lib/proxy/session.ts b/src/app/v1/_lib/proxy/session.ts index af6a8cea3..fff19da82 100644 --- a/src/app/v1/_lib/proxy/session.ts +++ b/src/app/v1/_lib/proxy/session.ts @@ -603,10 +603,6 @@ export class ProxySession { return !this.highConcurrencyModeEnabled; } - shouldApplyContentTransforms(): boolean { - return !this.highConcurrencyModeEnabled; - } - addSpecialSetting(setting: SpecialSetting): void { this.specialSettings.push(setting); } diff --git a/tests/unit/proxy/session.test.ts b/tests/unit/proxy/session.test.ts index ec58e49ee..ac7195370 100644 --- a/tests/unit/proxy/session.test.ts +++ b/tests/unit/proxy/session.test.ts @@ -172,7 +172,6 @@ describe("ProxySession high-concurrency policy", () => { expect(session.shouldRetainClientAbortBilling()).toBe(true); expect(session.shouldBillHedgeLosers()).toBe(true); expect(session.shouldParseResponseDiagnostics()).toBe(true); - expect(session.shouldApplyContentTransforms()).toBe(true); session.setHighConcurrencyModeEnabled(true); @@ -181,7 +180,6 @@ describe("ProxySession high-concurrency policy", () => { expect(session.shouldRetainClientAbortBilling()).toBe(false); expect(session.shouldBillHedgeLosers()).toBe(false); expect(session.shouldParseResponseDiagnostics()).toBe(false); - expect(session.shouldApplyContentTransforms()).toBe(false); expect(session.shouldPersistSessionDebugArtifacts()).toBe(false); expect(session.shouldTrackSessionObservability()).toBe(false); });