+
+ {t("slotSaturation", { count: saturationEvents.length })}
+
+
+ {saturationEvents.map((event, index) => {
+ const provider = asRecord(event.provider);
+ return (
+
+ {t("slotSaturationEvent", {
+ provider: asString(provider.name) ?? "-",
+ active: asNumber(event.activeAttemptCount) ?? 0,
+ cap: asNumber(event.configuredCap) ?? 0,
+ elapsed: Math.round(
+ asNumber(event.elapsedMs) ?? asNumber(event.durationMs) ?? 0
+ ),
+ })}
+
+ );
+ })}
+
+
diff --git a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/LogicTraceTab.tsx b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/LogicTraceTab.tsx
index 807190702..c0b6ab95b 100644
--- a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/LogicTraceTab.tsx
+++ b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/LogicTraceTab.tsx
@@ -64,7 +64,8 @@ function getRequestStatus(item: ProviderChainItem): StepStatus {
item.reason === "concurrent_limit_failed" ||
item.reason === "hedge_loser_cancelled" ||
item.reason === "hedge_loser_billed" ||
- item.reason === "client_abort"
+ item.reason === "client_abort" ||
+ item.reason === "client_abort_no_first_byte"
) {
return "failure";
}
@@ -222,7 +223,10 @@ export function LogicTraceTab({
// Calculate step offset for session reuse flow
const sessionReuseStepOffset = isSessionReuseFlow ? 1 : 0;
- if (normalizedRoutingTrace?.mode === "discovery") {
+ if (
+ normalizedRoutingTrace?.mode === "discovery" ||
+ normalizedRoutingTrace?.mode === "legacy_hedge"
+ ) {
return (
@@ -927,7 +931,8 @@ export function LogicTraceTab({
const isHedgeWinner = item.reason === "hedge_winner";
const isHedgeLoser = item.reason === "hedge_loser_cancelled";
const isHedgeLoserBilled = item.reason === "hedge_loser_billed";
- const isClientAbort = item.reason === "client_abort";
+ const isClientAbort =
+ item.reason === "client_abort" || item.reason === "client_abort_no_first_byte";
// Resolved hedge losers (cancelled or billed) carry billing detail when
// their reclaimed upstream response was charged to the request total.
const hedgeLoserBilling =
@@ -963,7 +968,9 @@ export function LogicTraceTab({
: isHedgeLoserBilled
? tChain("timeline.hedgeLoserBilled")
: isClientAbort
- ? tChain("timeline.clientAbort")
+ ? item.reason === "client_abort_no_first_byte"
+ ? tChain("reasons.client_abort_no_first_byte")
+ : tChain("timeline.clientAbort")
: isRetry
? t("logicTrace.retryAttempt", { number: item.attemptNumber ?? 1 })
: item.reason === "hedge_winner"
diff --git a/src/app/[locale]/dashboard/logs/_components/provider-chain-popover.tsx b/src/app/[locale]/dashboard/logs/_components/provider-chain-popover.tsx
index 676edde53..3c16f6c94 100644
--- a/src/app/[locale]/dashboard/logs/_components/provider-chain-popover.tsx
+++ b/src/app/[locale]/dashboard/logs/_components/provider-chain-popover.tsx
@@ -281,7 +281,7 @@ function getItemStatus(item: ProviderChainItem): {
bgColor: "bg-slate-50 dark:bg-slate-800/50",
};
}
- if (item.reason === "client_abort") {
+ if (item.reason === "client_abort" || item.reason === "client_abort_no_first_byte") {
return {
icon: MinusCircle,
color: "text-amber-600",
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 a749d211e..09212da22 100644
--- a/src/app/[locale]/settings/config/_components/system-settings-form.tsx
+++ b/src/app/[locale]/settings/config/_components/system-settings-form.tsx
@@ -80,6 +80,7 @@ interface SystemSettingsFormProps {
| "codexPriorityBillingSource"
| "billNonSuccessfulRequests"
| "billHedgeLosers"
+ | "legacyHedgeMaxInFlight"
| "discoveryEnabled"
| "discoveryConcurrency"
| "maxDiscoveryRounds"
@@ -163,6 +164,9 @@ export function SystemSettingsForm({
initialSettings.billNonSuccessfulRequests
);
const [billHedgeLosers, setBillHedgeLosers] = useState(initialSettings.billHedgeLosers);
+ const [legacyHedgeMaxInFlight, setLegacyHedgeMaxInFlight] = useState(
+ initialSettings.legacyHedgeMaxInFlight ?? 2
+ );
const [discoveryEnabled, setDiscoveryEnabled] = useState(initialSettings.discoveryEnabled);
const [discoveryConcurrency, setDiscoveryConcurrency] = useState(
initialSettings.discoveryConcurrency
@@ -295,6 +299,16 @@ export function SystemSettingsForm({
return;
}
+ const legacyHedgeMaxInFlightValue = Number(legacyHedgeMaxInFlight);
+ if (
+ !Number.isSafeInteger(legacyHedgeMaxInFlightValue) ||
+ legacyHedgeMaxInFlightValue < 1 ||
+ legacyHedgeMaxInFlightValue > 4
+ ) {
+ toast.error(t("legacyHedgeMaxInFlightInvalid"));
+ return;
+ }
+
const discoveryConfig = {
discoveryConcurrency: Number(discoveryConcurrency),
maxDiscoveryRounds: Number(maxDiscoveryRounds),
@@ -402,6 +416,7 @@ export function SystemSettingsForm({
codexPriorityBillingSource,
billNonSuccessfulRequests,
billHedgeLosers,
+ legacyHedgeMaxInFlight: legacyHedgeMaxInFlightValue,
discoveryEnabled,
...(discoveryEnabled ? discoveryConfig : {}),
timezone,
@@ -459,6 +474,7 @@ export function SystemSettingsForm({
setCodexPriorityBillingSource(result.data.codexPriorityBillingSource);
setBillNonSuccessfulRequests(result.data.billNonSuccessfulRequests);
setBillHedgeLosers(result.data.billHedgeLosers);
+ setLegacyHedgeMaxInFlight(result.data.legacyHedgeMaxInFlight);
setDiscoveryEnabled(result.data.discoveryEnabled);
setDiscoveryConcurrency(result.data.discoveryConcurrency);
setMaxDiscoveryRounds(result.data.maxDiscoveryRounds);
@@ -754,7 +770,51 @@ export function SystemSettingsForm({
/>
- {/* Bounded Streaming Discovery */}
+ {/* Legacy streaming hedge concurrency */}
+
+
+
+
+
+
+
+
+
+
+ {t("legacyHedgeMaxInFlightTooltip")}
+
+
+
+
+ {t("legacyHedgeMaxInFlightDesc")}
+
+
+
+ setLegacyHedgeMaxInFlight(
+ event.target.value === "" ? "" : Number(event.target.value)
+ )
+ }
+ disabled={isPending}
+ className={`${inputClassName} w-24 shrink-0`}
+ />
+
+
+
diff --git a/src/app/[locale]/settings/config/page.tsx b/src/app/[locale]/settings/config/page.tsx
index c32f4e579..37a02e503 100644
--- a/src/app/[locale]/settings/config/page.tsx
+++ b/src/app/[locale]/settings/config/page.tsx
@@ -57,6 +57,7 @@ async function SettingsConfigContent({ locale }: { locale: string }) {
codexPriorityBillingSource: settings.codexPriorityBillingSource,
billNonSuccessfulRequests: settings.billNonSuccessfulRequests,
billHedgeLosers: settings.billHedgeLosers,
+ legacyHedgeMaxInFlight: settings.legacyHedgeMaxInFlight,
discoveryEnabled: settings.discoveryEnabled,
discoveryConcurrency: settings.discoveryConcurrency,
maxDiscoveryRounds: settings.maxDiscoveryRounds,
diff --git a/src/app/api/admin/system-config/route.ts b/src/app/api/admin/system-config/route.ts
index b98c02390..a5f56127a 100644
--- a/src/app/api/admin/system-config/route.ts
+++ b/src/app/api/admin/system-config/route.ts
@@ -83,6 +83,7 @@ export async function POST(req: Request) {
currencyDisplay: validated.currencyDisplay,
billingModelSource: validated.billingModelSource,
codexPriorityBillingSource: validated.codexPriorityBillingSource,
+ legacyHedgeMaxInFlight: validated.legacyHedgeMaxInFlight,
discoveryEnabled: validated.discoveryEnabled,
discoveryConcurrency: validated.discoveryConcurrency,
maxDiscoveryRounds: validated.maxDiscoveryRounds,
diff --git a/src/app/api/v1/resources/system/handlers.ts b/src/app/api/v1/resources/system/handlers.ts
index f238060b8..3ce448569 100644
--- a/src/app/api/v1/resources/system/handlers.ts
+++ b/src/app/api/v1/resources/system/handlers.ts
@@ -10,6 +10,7 @@ import { jsonResponse } from "@/lib/api/v1/_shared/response-helpers";
import { SystemSettingsUpdateSchema } from "@/lib/api/v1/schemas/system-config";
import { getDiscoveryValidationErrorCode } from "@/lib/validation/discovery-settings";
import { getReplayCacheTtlValidationErrorCode } from "@/lib/validation/replay-settings";
+import { getLegacyHedgeMaxInFlightValidationErrorCode } from "@/lib/validation/schemas";
export async function getSystemSettings(c: Context): Promise
{
const actions = await import("@/actions/system-config");
@@ -32,7 +33,8 @@ export async function updateSystemSettings(c: Context): Promise {
const body = await parseHonoJsonBody(c, SystemSettingsUpdateSchema, {
validationErrorCode: (error) =>
getDiscoveryValidationErrorCode(error.issues) ??
- getReplayCacheTtlValidationErrorCode(error.issues),
+ getReplayCacheTtlValidationErrorCode(error.issues) ??
+ getLegacyHedgeMaxInFlightValidationErrorCode(error.issues),
});
if (!body.ok) return body.response;
const actions = await import("@/actions/system-config");
diff --git a/src/app/api/v1/resources/system/router.ts b/src/app/api/v1/resources/system/router.ts
index b3962aa50..a90048090 100644
--- a/src/app/api/v1/resources/system/router.ts
+++ b/src/app/api/v1/resources/system/router.ts
@@ -11,6 +11,7 @@ import {
} from "@/lib/api/v1/schemas/system-config";
import { getDiscoveryValidationErrorCode } from "@/lib/validation/discovery-settings";
import { getReplayCacheTtlValidationErrorCode } from "@/lib/validation/replay-settings";
+import { getLegacyHedgeMaxInFlightValidationErrorCode } from "@/lib/validation/schemas";
import {
getSystemDisplaySettings,
getSystemSettings,
@@ -25,7 +26,8 @@ export const systemRouter = new OpenAPIHono({
result.error,
new URL(c.req.url).pathname,
getDiscoveryValidationErrorCode(result.error.issues) ??
- getReplayCacheTtlValidationErrorCode(result.error.issues)
+ getReplayCacheTtlValidationErrorCode(result.error.issues) ??
+ getLegacyHedgeMaxInFlightValidationErrorCode(result.error.issues)
);
}
},
diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts
index fbe719d21..f70155d35 100644
--- a/src/app/v1/_lib/proxy/forwarder.ts
+++ b/src/app/v1/_lib/proxy/forwarder.ts
@@ -164,7 +164,19 @@ import {
export const DEFAULT_CODEX_USER_AGENT =
"codex_cli_rs/0.93.0 (Windows 10.0.26200; x86_64) vscode/1.108.1";
const EMPTY_PREFIX_CHUNK = new Uint8Array(0);
-const LEGACY_STREAMING_HEDGE_MAX_CONCURRENCY = 2;
+const LEGACY_STREAMING_HEDGE_DEFAULT_MAX_IN_FLIGHT = 2;
+const LEGACY_STREAMING_HEDGE_MIN_MAX_IN_FLIGHT = 1;
+const LEGACY_STREAMING_HEDGE_MAX_MAX_IN_FLIGHT = 4;
+const CLIENT_ABORT_HEALTH_FALLBACK_THRESHOLD_MS = 30_000;
+
+function clampLegacyHedgeMaxInFlight(value: unknown): number {
+ const numeric = typeof value === "number" ? value : Number(value);
+ if (!Number.isFinite(numeric)) return LEGACY_STREAMING_HEDGE_DEFAULT_MAX_IN_FLIGHT;
+ return Math.min(
+ LEGACY_STREAMING_HEDGE_MAX_MAX_IN_FLIGHT,
+ Math.max(LEGACY_STREAMING_HEDGE_MIN_MAX_IN_FLIGHT, Math.floor(numeric))
+ );
+}
async function runStreamContentGateWithAbortSignals(
reader: ReadableStreamDefaultReader,
@@ -586,6 +598,23 @@ type StreamingHedgeAttempt = {
gateAudit?: ProviderChainItem["streamGate"];
/** 该 attempt 首字节到达时刻（epoch ms）；只有赢家的值会被记为 session TTFB。 */
firstByteAt?: number | null;
+ /** Stable identity for routing trace and per-attempt health attribution. */
+ attemptId: string;
+ /** Monotonic dispatch timestamp used for client-abort threshold comparisons. */
+ startedAtMonotonic: number;
+ /** Monotonic timestamp for the current hedge threshold window, including pre-dispatch setup. */
+ thresholdStartedAtMonotonic: number;
+ /** Effective health-attribution threshold; independent from request timeout behavior. */
+ healthAttributionThresholdMs: number;
+ /** Set immediately when the upstream dispatch starts. */
+ dispatched: boolean;
+ /** Exactly-once guard for provider health/circuit settlement. */
+ healthSettlementClaimed: boolean;
+ healthOutcome: "client_abort_no_first_byte" | "provider_failure" | "other_failure" | null;
+ healthPausedAtMonotonic: number | null;
+ healthPausedDurationMs: number;
+ /** Avoid duplicate saturation events for a threshold trigger. */
+ hedgeSaturationRecorded: boolean;
/**
* Billing context snapshot for the INITIAL provider's losing attempt, captured BEFORE
* commitWinner overwrites the shared session's model/context with the winner's. Null for
@@ -670,7 +699,8 @@ const NON_STREAM_BODY_INSPECTION_MAX_BYTES = 32 * 1024; // 32 KiB
*/
async function readResponseTextUpTo(
response: Response,
- maxBytes: number
+ maxBytes: number,
+ onChunk?: (value: Uint8Array) => void
): Promise<{ text: string; truncated: boolean }> {
const reader = response.body?.getReader();
if (!reader) {
@@ -687,6 +717,7 @@ async function readResponseTextUpTo(
const { done, value } = await reader.read();
if (done) break;
if (!value || value.byteLength === 0) continue;
+ onChunk?.(value);
const remaining = maxBytes - bytesRead;
// 注意：remaining<=0 发生在“已经读到下一块 chunk”之后。
@@ -1648,6 +1679,9 @@ export class ProxyForwarder {
1,
discoverySettings.stickyTimeoutCooldownMs ?? 300_000
),
+ legacyHedgeMaxInFlight: clampLegacyHedgeMaxInFlight(
+ discoverySettings.legacyHedgeMaxInFlight
+ ),
sessionTtlSeconds,
},
});
@@ -1660,6 +1694,9 @@ export class ProxyForwarder {
}
const useStreamingHedge = ProxyForwarder.shouldUseStreamingHedge(session);
+ const legacyHedgeMaxInFlight = clampLegacyHedgeMaxInFlight(
+ discoverySettings.legacyHedgeMaxInFlight
+ );
const singleUpstream =
discoveryPreparation.reason === "binding_conflict" ||
discoveryPreparation.reason === "lease_conflict" ||
@@ -1676,10 +1713,24 @@ export class ProxyForwarder {
eligible: false,
bypassReason: discoveryPreparation.reason,
startedAt: requestStartedAt,
+ config: {
+ discoveryConcurrency: Math.max(2, Math.floor(discoverySettings.discoveryConcurrency ?? 2)),
+ maxDiscoveryRounds: Math.max(1, Math.floor(discoverySettings.maxDiscoveryRounds ?? 2)),
+ discoverySlaMs: Math.max(1, discoverySettings.discoverySlaMs ?? 10_000),
+ stickySlaMs: Math.max(1, discoverySettings.stickySlaMs ?? 20_000),
+ racingTotalTimeoutMs: Math.max(1, discoverySettings.racingTotalTimeoutMs ?? 60_000),
+ stickyTimeoutCooldownMs: Math.max(1, discoverySettings.stickyTimeoutCooldownMs ?? 300_000),
+ legacyHedgeMaxInFlight,
+ sessionTtlSeconds,
+ },
});
if (useStreamingHedge) {
- const hedgePromise = ProxyForwarder.sendStreamingWithHedge(session);
+ const hedgePromise = ProxyForwarder.sendStreamingWithHedge(
+ session,
+ discoverySettings,
+ legacyHedgeMaxInFlight
+ );
void hedgePromise.catch(() => undefined);
return await hedgePromise;
}
@@ -1908,6 +1959,11 @@ export class ProxyForwarder {
// ========== 内层循环：重试当前供应商（根据配置最多尝试 maxAttemptsPerProvider 次）==========
while (attemptCount < maxAttemptsPerProvider) {
attemptCount++;
+ let attemptStartedAtMonotonic = 0;
+ let attemptFirstByteSeen = false;
+ let attemptDispatched = false;
+ let healthPausedAtMonotonic: number | null = null;
+ let healthPausedDurationMs = 0;
// Use currentEndpointIndex for endpoint selection (sticky behavior)
// - currentEndpointIndex is advanced only on SYSTEM_ERROR (network errors)
@@ -1929,7 +1985,14 @@ export class ProxyForwarder {
currentProvider,
activeEndpoint.baseUrl,
endpointAudit,
- attemptCount
+ attemptCount,
+ false,
+ undefined,
+ () => {
+ attemptDispatched = true;
+ attemptStartedAtMonotonic = performance.now();
+ attemptFirstByteSeen = false;
+ }
);
// ========== 空响应检测（仅非流式）==========
@@ -2001,6 +2064,7 @@ export class ProxyForwarder {
// 首字节到达即清除首字节计时器，保持「首字节超时」的原始语义——
// 思考型模型可在首个内容帧前长时间输出中性帧，不应触发该计时器
onFirstByte: () => {
+ attemptFirstByteSeen = true;
gateFirstByteAt ??= Date.now();
runtime.clearResponseTimeout?.();
},
@@ -2008,8 +2072,20 @@ export class ProxyForwarder {
idleTimeoutMs: currentProvider.streamingIdleTimeoutMs,
captureCommitMarker: !session.isHighConcurrencyModeEnabled(),
prebufferBudget: getStreamGatePrebufferBudget(),
- onBudgetWaitStart: runtime.pauseResponseTimeout,
- onBudgetWaitEnd: runtime.resumeResponseTimeout,
+ onBudgetWaitStart: () => {
+ runtime.pauseResponseTimeout?.();
+ healthPausedAtMonotonic ??= performance.now();
+ },
+ onBudgetWaitEnd: () => {
+ runtime.resumeResponseTimeout?.();
+ if (healthPausedAtMonotonic !== null) {
+ healthPausedDurationMs += Math.max(
+ 0,
+ performance.now() - healthPausedAtMonotonic
+ );
+ healthPausedAtMonotonic = null;
+ }
+ },
},
[runtime.responseController?.signal, session.clientAbortSignal]
);
@@ -2107,6 +2183,15 @@ export class ProxyForwarder {
endpointUrl: endpointAudit.endpointUrl,
upstreamStatusCode: response.status,
bindingIntent: session.isSessionBindingAllowed() ? undefined : "none",
+ healthAttemptId: `legacy-serial-${totalProvidersAttempted}-${attemptCount}`,
+ healthAttemptStartedAtMonotonic: attemptStartedAtMonotonic,
+ healthAttributionThresholdMs:
+ currentProvider.firstByteTimeoutStreamingMs > 0
+ ? currentProvider.firstByteTimeoutStreamingMs
+ : CLIENT_ABORT_HEALTH_FALLBACK_THRESHOLD_MS,
+ healthFirstByteSeen: attemptFirstByteSeen,
+ healthPausedDurationMs,
+ healthOutcomeSettled: false,
});
logger.info("ProxyForwarder: Streaming response received, deferring finalization", {
@@ -2185,7 +2270,10 @@ export class ProxyForwarder {
const clonedResponse = response.clone();
const inspected = await readResponseTextUpTo(
clonedResponse,
- NON_STREAM_BODY_INSPECTION_MAX_BYTES
+ NON_STREAM_BODY_INSPECTION_MAX_BYTES,
+ (value) => {
+ if (value.byteLength > 0) attemptFirstByteSeen = true;
+ }
);
inspectedText = inspected.text;
inspectedTruncated = inspected.truncated;
@@ -2453,12 +2541,61 @@ export class ProxyForwarder {
totalProvidersAttempted,
});
+ const now = performance.now();
+ const elapsedMs = Math.max(
+ 0,
+ now -
+ attemptStartedAtMonotonic -
+ healthPausedDurationMs -
+ (healthPausedAtMonotonic === null ? 0 : now - healthPausedAtMonotonic)
+ );
+ const thresholdMs =
+ currentProvider.firstByteTimeoutStreamingMs > 0
+ ? currentProvider.firstByteTimeoutStreamingMs
+ : CLIENT_ABORT_HEALTH_FALLBACK_THRESHOLD_MS;
+ const qualifiesForHealth =
+ attemptDispatched &&
+ !attemptFirstByteSeen &&
+ elapsedMs >= thresholdMs &&
+ endpointPolicy.allowCircuitBreakerAccounting;
+
+ if (qualifiesForHealth) {
+ const abortFailure = new ProxyError(
+ "Client aborted while provider was waiting for the first byte",
+ 499,
+ undefined,
+ true
+ );
+ await recordFailure(currentProvider.id, abortFailure).catch((healthError) => {
+ logger.warn("ProxyForwarder: Failed to account serial client abort health", {
+ providerId: currentProvider.id,
+ error: healthError instanceof Error ? healthError.message : String(healthError),
+ });
+ });
+ session.appendRoutingTraceEvent({
+ type: "client_abort_no_first_byte",
+ attemptId: `legacy-serial-${totalProvidersAttempted}-${attemptCount}`,
+ provider: {
+ id: currentProvider.id,
+ name: currentProvider.name,
+ priority: currentProvider.priority || 0,
+ },
+ outcome: "provider_failure",
+ cancellationKind: "client_abort",
+ reason: "external_client_abort",
+ effectiveThresholdMs: thresholdMs,
+ circuitAccountingApplied: true,
+ availabilityAccountingApplied: true,
+ durationMs: Math.round(elapsedMs),
+ });
+ }
+
await ProxyForwarder.clearSessionProviderBinding(session, currentProvider.id);
// 记录到决策链（标记为客户端中断）
session.addProviderToChain(currentProvider, {
...endpointAudit,
- reason: "client_abort",
+ reason: qualifiesForHealth ? "client_abort_no_first_byte" : "client_abort",
circuitState: getCircuitState(currentProvider.id),
attemptNumber: attemptCount,
errorMessage: "Client aborted request",
@@ -3085,7 +3222,8 @@ export class ProxyForwarder {
endpointAudit?: { endpointId: number | null; endpointUrl: string },
attemptNumber?: number,
deferDetailSnapshotPersistence: boolean = false,
- externalAbortSignal?: AbortSignal
+ externalAbortSignal?: AbortSignal,
+ onUpstreamDispatch?: () => void
): Promise {
if (!provider) {
throw new Error("Provider is required");
@@ -3661,6 +3799,10 @@ export class ProxyForwarder {
interface UndiciFetchOptions extends RequestInit {
dispatcher?: Dispatcher;
}
+ const fetchWithDispatch = async (url: string, requestInit: UndiciFetchOptions) => {
+ onUpstreamDispatch?.();
+ return await fetch(url, requestInit);
+ };
// ⭐ 双路超时控制（first-byte / total）
// 注意：由于 undici fetch API 的限制，无法精确分离 DNS/TCP/TLS 连接阶段和响应头接收阶段
@@ -3878,6 +4020,7 @@ export class ProxyForwarder {
const requestBodyJson = decodeRequestBodyAsJson(requestBody);
if (requestBodyJson) {
+ onUpstreamDispatch?.();
const wsResult = await tryResponsesWebsocketUpstream({
provider,
upstreamUrl: proxyUrl,
@@ -3965,9 +4108,10 @@ export class ProxyForwarder {
provider.id,
provider.name,
session,
- deferDetailSnapshotPersistence
+ deferDetailSnapshotPersistence,
+ onUpstreamDispatch
)
- : await fetch(proxyUrl, init);
+ : await fetchWithDispatch(proxyUrl, init);
// ⭐ fetch 成功：收到 HTTP 响应头，保留响应超时继续监控
// 注意：undici 的 fetch 在收到 HTTP 响应头后就 resolve，但实际数据（SSE 首字节 / 完整 JSON）
// 还没到达。responseTimeoutId 需要延续到 response-handler 中才能真正控制"首字节"或"总耗时"
@@ -4227,9 +4371,10 @@ export class ProxyForwarder {
provider.id,
provider.name,
session,
- deferDetailSnapshotPersistence
+ deferDetailSnapshotPersistence,
+ onUpstreamDispatch
)
- : await fetch(proxyUrl, http1FallbackInit);
+ : await fetchWithDispatch(proxyUrl, http1FallbackInit);
logger.info("ProxyForwarder: HTTP/1.1 fallback succeeded", {
providerId: provider.id,
@@ -4304,9 +4449,10 @@ export class ProxyForwarder {
provider.id,
provider.name,
session,
- deferDetailSnapshotPersistence
+ deferDetailSnapshotPersistence,
+ onUpstreamDispatch
)
- : await fetch(proxyUrl, fallbackInit);
+ : await fetchWithDispatch(proxyUrl, fallbackInit);
logger.info("ProxyForwarder: Direct connection succeeded after proxy failure", {
providerId: provider.id,
providerName: provider.name,
@@ -4682,7 +4828,11 @@ export class ProxyForwarder {
return resolveEndpointPolicy(policySession.requestUrl?.pathname ?? "/");
}
- private static async sendStreamingWithHedge(session: ProxySession): Promise {
+ private static async sendStreamingWithHedge(
+ session: ProxySession,
+ settings: SystemSettings,
+ maxInFlight: number
+ ): Promise {
const initialProvider = session.provider;
if (!initialProvider) {
throw new Error("代理上下文缺少供应商");
@@ -4690,7 +4840,7 @@ export class ProxyForwarder {
const rawCrossProviderFallbackEnabled = session.isRawCrossProviderFallbackEnabled();
// 竞速输家计费开关：开启时落败供应商不被直接掐断，而是后台 drain 并计费。
- const billHedgeLosers = (await getCachedSystemSettings()).billHedgeLosers === true;
+ const billHedgeLosers = settings.billHedgeLosers === true;
const launchedProviderIds = new Set();
let launchedProviderCount = 0;
let settled = false;
@@ -4895,6 +5045,38 @@ export class ProxyForwarder {
attempt.thresholdRemainingMs = 0;
if (settled || attempt.settled || attempt.thresholdTriggered) return;
attempt.thresholdTriggered = true;
+ if (attempts.size >= maxInFlight && !attempt.hedgeSaturationRecorded) {
+ attempt.hedgeSaturationRecorded = true;
+ const now = performance.now();
+ const thresholdStartedAt =
+ attempt.startedAtMonotonic > 0
+ ? attempt.startedAtMonotonic
+ : attempt.thresholdStartedAtMonotonic;
+ const elapsedMs = Math.max(
+ 0,
+ Math.round(
+ now -
+ thresholdStartedAt -
+ attempt.healthPausedDurationMs -
+ (attempt.healthPausedAtMonotonic === null ? 0 : now - attempt.healthPausedAtMonotonic)
+ )
+ );
+ session.appendRoutingTraceEvent({
+ type: "hedge_slot_saturated",
+ attemptId: attempt.attemptId,
+ provider: {
+ id: attempt.provider.id,
+ name: attempt.provider.name,
+ priority: attempt.provider.priority || 0,
+ },
+ outcome: "slot_saturated",
+ reason: "hedge_threshold",
+ activeAttemptCount: attempts.size,
+ configuredCap: maxInFlight,
+ durationMs: elapsedMs,
+ elapsedMs,
+ });
+ }
session.addProviderToChain(attempt.provider, {
...attempt.endpointAudit,
reason: "hedge_triggered",
@@ -4929,6 +5111,9 @@ export class ProxyForwarder {
attempt.thresholdPaused = false;
attempt.thresholdDeadlineAt = null;
attempt.thresholdRemainingMs = attempt.firstByteTimeoutMs;
+ attempt.thresholdStartedAtMonotonic = performance.now();
+ attempt.healthPausedAtMonotonic = null;
+ attempt.healthPausedDurationMs = 0;
scheduleAttemptThreshold(attempt);
};
@@ -4944,6 +5129,9 @@ export class ProxyForwarder {
if (attempt.thresholdDeadlineAt !== null) {
attempt.thresholdRemainingMs = Math.max(1, attempt.thresholdDeadlineAt - Date.now());
}
+ if (attempt.healthPausedAtMonotonic === null) {
+ attempt.healthPausedAtMonotonic = performance.now();
+ }
clearTimeout(attempt.thresholdTimer);
attempt.thresholdTimer = null;
attempt.thresholdDeadlineAt = null;
@@ -4952,6 +5140,13 @@ export class ProxyForwarder {
const resumeAttemptThreshold = (attempt: StreamingHedgeAttempt) => {
if (!attempt.thresholdPaused) return;
+ if (attempt.healthPausedAtMonotonic !== null) {
+ attempt.healthPausedDurationMs += Math.max(
+ 0,
+ performance.now() - attempt.healthPausedAtMonotonic
+ );
+ attempt.healthPausedAtMonotonic = null;
+ }
attempt.thresholdPaused = false;
scheduleAttemptThreshold(attempt);
};
@@ -4971,7 +5166,7 @@ export class ProxyForwarder {
const launchAlternative = async () => {
if (settled || winnerCommitted || noMoreProviders) return;
- if (attempts.size >= LEGACY_STREAMING_HEDGE_MAX_CONCURRENCY) return;
+ if (attempts.size >= maxInFlight) return;
if (launchingAlternative) {
await launchingAlternative;
return;
@@ -5022,17 +5217,34 @@ export class ProxyForwarder {
const runAttempt = (attempt: StreamingHedgeAttempt) => {
const providerForRequest =
- attempt.firstByteTimeoutMs > 0
+ attempt.firstByteTimeoutMs > 0 && maxInFlight > 1
? { ...attempt.provider, firstByteTimeoutStreamingMs: 0 }
: attempt.provider;
+ let dispatchMarked = false;
+
+ const markUpstreamDispatch = () => {
+ if (dispatchMarked) return;
+ dispatchMarked = true;
+ attempt.dispatched = true;
+ attempt.startedAtMonotonic = performance.now();
+ attempt.healthPausedAtMonotonic = null;
+ attempt.healthPausedDurationMs = 0;
+ armAttemptThreshold(attempt);
+ };
+ // Arm the hedge threshold when the attempt enters the transport call. The health clock
+ // remains gated by `attempt.dispatched` and is reset by the transport callback below, so
+ // setup time can trigger a hedge without being eligible for provider-failure attribution.
+ armAttemptThreshold(attempt);
void ProxyForwarder.doForward(
attempt.session,
providerForRequest,
attempt.baseUrl,
attempt.endpointAudit,
attempt.requestAttemptCount,
- true
+ true,
+ undefined,
+ markUpstreamDispatch
)
.then(async (response) => {
if (settled || winnerCommitted || attempt.settled) {
@@ -5171,6 +5383,8 @@ export class ProxyForwarder {
return;
}
+ attempt.firstByteAt ??= Date.now();
+
// 保留首块：若本 attempt 落败且需要计费，drain 时需要补回首块的 usage。
attempt.billingPrefixChunks = [firstChunk.value];
acceptedAsWinner = await commitWinner(attempt, [firstChunk.value], false);
@@ -5236,6 +5450,12 @@ export class ProxyForwarder {
}
if (settled || winnerCommitted || attempt.settled) return;
+ // Claim the attempt's terminal race before awaiting asynchronous error classification. If
+ // the downstream abort arrives while classification is in flight, the upstream error that
+ // reached this handler first remains authoritative. A rectifier retry below reopens this
+ // claim for the same logical attempt.
+ attempt.healthSettlementClaimed = true;
+ attempt.healthOutcome = "other_failure";
lastError = error;
let errorCategory = await categorizeErrorAsync(error);
@@ -5387,7 +5607,13 @@ export class ProxyForwarder {
attempt.thresholdTimer = null;
}
attempt.requestAttemptCount += 1;
- armAttemptThreshold(attempt);
+ attempt.dispatched = false;
+ attempt.startedAtMonotonic = 0;
+ attempt.firstByteAt = null;
+ attempt.attemptId = `legacy-hedge-${attempt.sequence}-${attempt.requestAttemptCount}`;
+ attempt.healthSettlementClaimed = false;
+ attempt.healthOutcome = null;
+ attempt.hedgeSaturationRecorded = false;
runAttempt(attempt);
return;
}
@@ -5408,6 +5634,8 @@ export class ProxyForwarder {
});
}
+ attempt.healthSettlementClaimed = true;
+ attempt.healthOutcome = "other_failure";
attempt.settled = true;
if (attempt.thresholdTimer) {
clearTimeout(attempt.thresholdTimer);
@@ -5421,6 +5649,7 @@ export class ProxyForwarder {
statusCode !== 404 &&
!isRequestScopedGateFailure(error)
) {
+ attempt.healthOutcome = "provider_failure";
await recordFailure(attempt.provider.id, error);
}
@@ -5700,6 +5929,19 @@ export class ProxyForwarder {
clearResponseTimeout: null,
firstByteTimeoutMs:
provider.firstByteTimeoutStreamingMs > 0 ? provider.firstByteTimeoutStreamingMs : 0,
+ attemptId: `legacy-hedge-${launchedProviderCount}-1`,
+ startedAtMonotonic: 0,
+ thresholdStartedAtMonotonic: 0,
+ healthAttributionThresholdMs:
+ provider.firstByteTimeoutStreamingMs > 0
+ ? provider.firstByteTimeoutStreamingMs
+ : CLIENT_ABORT_HEALTH_FALLBACK_THRESHOLD_MS,
+ dispatched: false,
+ healthSettlementClaimed: false,
+ healthOutcome: null,
+ healthPausedAtMonotonic: null,
+ healthPausedDurationMs: 0,
+ hedgeSaturationRecorded: false,
sequence: launchedProviderCount,
requestAttemptCount: 1,
reactiveRectifierRetryState: {
@@ -5742,28 +5984,103 @@ export class ProxyForwarder {
});
}
- armAttemptThreshold(attempt);
-
runAttempt(attempt);
return true;
};
+ const settleClientAbortHealth = (attempt: StreamingHedgeAttempt): boolean => {
+ if (
+ !attempt.dispatched ||
+ attempt.settled ||
+ attempt.firstByteAt != null ||
+ winnerCommitted ||
+ attempt.healthSettlementClaimed
+ ) {
+ return false;
+ }
+
+ const now = performance.now();
+ const elapsedMs = Math.max(
+ 0,
+ now -
+ attempt.startedAtMonotonic -
+ attempt.healthPausedDurationMs -
+ (attempt.healthPausedAtMonotonic === null ? 0 : now - attempt.healthPausedAtMonotonic)
+ );
+ if (elapsedMs < attempt.healthAttributionThresholdMs) return false;
+
+ attempt.healthSettlementClaimed = true;
+ attempt.healthOutcome = "client_abort_no_first_byte";
+ const roundedElapsedMs = Math.round(elapsedMs);
+ const failure = new ProxyError(
+ "Client aborted while provider was waiting for the first byte",
+ 499,
+ undefined,
+ true
+ );
+
+ session.appendRoutingTraceEvent({
+ type: "client_abort_no_first_byte",
+ attemptId: attempt.attemptId,
+ provider: {
+ id: attempt.provider.id,
+ name: attempt.provider.name,
+ priority: attempt.provider.priority || 0,
+ },
+ outcome: "provider_failure",
+ cancellationKind: "client_abort",
+ reason: "external_client_abort",
+ effectiveThresholdMs: attempt.healthAttributionThresholdMs,
+ circuitAccountingApplied: true,
+ availabilityAccountingApplied: true,
+ durationMs: roundedElapsedMs,
+ elapsedMs: roundedElapsedMs,
+ });
+
+ // Do not inherit the downstream abort signal: health and trace side effects must finish
+ // independently after the client-facing response has become HTTP 499.
+ void recordFailure(attempt.provider.id, failure).catch((healthError) => {
+ logger.warn("ProxyForwarder: Failed to account client abort provider health", {
+ error: healthError instanceof Error ? healthError.message : String(healthError),
+ attemptId: attempt.attemptId,
+ providerId: attempt.provider.id,
+ });
+ });
+ return true;
+ };
+
const cleanupClientAbortListener = bindClientAbortListener(session.clientAbortSignal, () => {
if (settled || winnerCommitted) return;
noMoreProviders = true;
lastError = new ProxyError("Request aborted by client", 499, undefined, true);
lastErrorCategory = ErrorCategory.CLIENT_ABORT;
+ const attributedAttempts: StreamingHedgeAttempt[] = [];
for (const attempt of Array.from(attempts)) {
if (!attempt.settled) {
- session.addProviderToChain(attempt.provider, {
- ...attempt.endpointAudit,
- reason: "client_abort",
- attemptNumber: attempt.sequence,
- errorMessage: "Client aborted request",
- modelRedirect: getAttemptModelRedirect(attempt),
- });
+ const attributed = settleClientAbortHealth(attempt);
+ if (!attributed) {
+ session.addProviderToChain(attempt.provider, {
+ ...attempt.endpointAudit,
+ reason: "client_abort",
+ attemptNumber: attempt.sequence,
+ errorMessage: "Client aborted request",
+ modelRedirect: getAttemptModelRedirect(attempt),
+ });
+ } else {
+ attributedAttempts.push(attempt);
+ }
}
}
+ for (const attempt of attributedAttempts) {
+ session.addProviderToChain(attempt.provider, {
+ ...attempt.endpointAudit,
+ reason: "client_abort_no_first_byte",
+ attemptNumber: attempt.sequence,
+ errorMessage: "Client aborted before provider first byte threshold",
+ circuitState: getCircuitState(attempt.provider.id),
+ modelRedirect: getAttemptModelRedirect(attempt),
+ });
+ }
abortAllAttempts(undefined, "client_abort");
void finishIfExhausted();
});
@@ -8598,7 +8915,8 @@ export class ProxyForwarder {
providerId: number,
providerName: string,
session?: ProxySession,
- deferDetailSnapshotPersistence: boolean = false
+ deferDetailSnapshotPersistence: boolean = false,
+ onUpstreamDispatch?: () => void
): Promise {
const { FETCH_HEADERS_TIMEOUT: headersTimeout, FETCH_BODY_TIMEOUT: bodyTimeout } =
getEnvConfig();
@@ -8636,6 +8954,7 @@ export class ProxyForwarder {
return undefined;
};
+ onUpstreamDispatch?.();
const undiciRes = await undiciRequest(url, {
method: init.method as string,
headers: headersObj,
diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts
index 26d34a17f..d2a545bae 100644
--- a/src/app/v1/_lib/proxy/response-handler.ts
+++ b/src/app/v1/_lib/proxy/response-handler.ts
@@ -1794,6 +1794,8 @@ type FinalizeDeferredStreamingResult = {
* @param streamEndedNormally - 必须是 reader 读到 done=true 的“自然结束”；超时/中断等异常结束由其它逻辑处理。
* @param clientAborted - 标记是否为客户端主动中断（用于内部状态码映射，避免把中断记为 200 completed）
* @param abortReason - 非自然结束时的原因码（用于内部记录/熔断归因；不会影响客户端响应）
+ * @param firstByteSeen - Authoritative body-byte observation. Undefined means the caller cannot
+ * report first-byte state and therefore cannot qualify a no-first-byte attribution.
*/
function finalizeDeferredStreamingFinalizationIfNeeded(
session: ProxySession,
@@ -1803,7 +1805,8 @@ function finalizeDeferredStreamingFinalizationIfNeeded(
clientAborted: boolean,
discoveryLeaseLifecycle: DiscoveryLeaseLifecycle,
protocolObservation: StreamProtocolObservation | null,
- abortReason?: string
+ abortReason?: string,
+ firstByteSeen?: boolean
): FinalizeDeferredStreamingResult {
const meta = consumeDeferredStreamingFinalization(session);
const provider = session.provider;
@@ -2149,6 +2152,48 @@ function finalizeDeferredStreamingFinalizationIfNeeded(
return true;
})();
+ const healthAttributionElapsedMs =
+ meta?.healthAttemptStartedAtMonotonic == null
+ ? null
+ : Math.max(
+ 0,
+ (meta.healthAbortAtMonotonic ?? performance.now()) -
+ meta.healthAttemptStartedAtMonotonic -
+ (meta.healthPausedDurationMs ?? 0)
+ );
+ const clientAbortNoFirstByte =
+ !clientAbortCompleteSuccess &&
+ clientAborted &&
+ meta?.healthAttemptId != null &&
+ meta.healthFirstByteSeen !== true &&
+ firstByteSeen === false &&
+ meta.healthAttributionThresholdMs != null &&
+ healthAttributionElapsedMs != null &&
+ healthAttributionElapsedMs >= meta.healthAttributionThresholdMs &&
+ meta.healthOutcomeSettled !== true &&
+ session.getEndpointPolicy().allowCircuitBreakerAccounting;
+ if (clientAbortNoFirstByte && meta) {
+ meta.healthOutcomeSettled = true;
+ const elapsedMs = Math.round(healthAttributionElapsedMs ?? 0);
+ session.appendRoutingTraceEvent({
+ type: "client_abort_no_first_byte",
+ attemptId: meta.healthAttemptId,
+ provider: {
+ id: meta.providerId,
+ name: meta.providerName,
+ priority: meta.providerPriority,
+ },
+ outcome: "provider_failure",
+ cancellationKind: "client_abort",
+ reason: "external_client_abort",
+ effectiveThresholdMs: meta.healthAttributionThresholdMs,
+ circuitAccountingApplied: true,
+ availabilityAccountingApplied: true,
+ durationMs: elapsedMs,
+ elapsedMs,
+ });
+ }
+
// “内部结算用”的状态码（不会改变客户端实际 HTTP 状态码）。
// - 假 200：优先映射为“推断得到的 4xx/5xx”（未命中则回退 502），确保内部统计/熔断/会话绑定把它当作失败。
// - 未自然结束：也应映射为失败（避免把中断/部分流误记为 200 completed）。
@@ -2324,13 +2369,13 @@ function finalizeDeferredStreamingFinalizationIfNeeded(
// 未自然结束：不更新 session 绑定（避免把会话粘到不稳定 provider），但要避免把它误记为 200 completed。
//
// 同时，为了让故障转移/熔断能正确工作：
- // - 客户端主动中断：不计入熔断器（这通常不是供应商问题）
+ // - 客户端主动中断：默认不计入熔断器；仅 legacy serial 的静默首字节阈值命中会归因供应商
// - 非客户端中断：计入 provider/endpoint 熔断失败（与 timeout 路径保持一致）
if ((clientAborted || !streamEndedNormally) && !clientAbortCompleteSuccess) {
session.addProviderToChain(providerForChain, {
endpointId: meta.endpointId,
endpointUrl: meta.endpointUrl,
- reason: "system_error",
+ reason: clientAbortNoFirstByte ? "client_abort_no_first_byte" : "system_error",
attemptNumber: meta.attemptNumber,
statusCode: effectiveStatusCode,
errorMessage: errorMessage ?? undefined,
@@ -2340,7 +2385,10 @@ function finalizeDeferredStreamingFinalizationIfNeeded(
try {
await clearSessionBinding();
- if (!clientAborted && session.getEndpointPolicy().allowCircuitBreakerAccounting) {
+ if (
+ (!clientAborted || clientAbortNoFirstByte) &&
+ session.getEndpointPolicy().allowCircuitBreakerAccounting
+ ) {
try {
const { recordFailure } = await import("@/lib/circuit-breaker");
await recordFailure(meta.providerId, new Error(errorMessage ?? "STREAM_ABORTED"));
@@ -3889,6 +3937,7 @@ export class ProxyResponseHandler {
const streamTextAccumulator = new BoundedStreamTextAccumulator();
let lastStreamTextSnapshot: BoundedStreamTextSnapshot | null = null;
+ let passthroughFirstByteSeen = false;
let observePassthroughChunk = (_value: Uint8Array) => {};
let observePassthroughReadStart = () => {};
let observePassthroughDrainStart = () => {};
@@ -3910,6 +3959,11 @@ export class ProxyResponseHandler {
return;
}
passthroughClientDetached = true;
+ const deferredMeta = peekDeferredStreamingFinalization(session);
+ if (deferredMeta) {
+ deferredMeta.healthAbortAtMonotonic = performance.now();
+ deferredMeta.healthFirstByteSeen = passthroughFirstByteSeen;
+ }
clientAbortMeter?.switchToDetachedMode();
if (!clientAbortMeter) {
const rejection = new Error("client_detached_without_metering");
@@ -3963,6 +4017,7 @@ export class ProxyResponseHandler {
source: response.body,
onReadStart: () => observePassthroughReadStart(),
onChunk: (value) => {
+ if (value.byteLength > 0) passthroughFirstByteSeen = true;
const metering = clientAbortMeter?.observe(value);
passthroughShadowObserver?.observe(value);
streamProtocolObserver?.observe(value);
@@ -4251,7 +4306,8 @@ export class ProxyResponseHandler {
discoveryLeaseLifecycle,
streamProtocolObserver?.finish() ??
(meteringSnapshot ? protocolObservationFromMetering(meteringSnapshot) : null),
- abortReason
+ abortReason,
+ passthroughFirstByteSeen
);
latestCommitSideEffects = finalized.commitSideEffects;
latestFinalizeAttemptResources = finalized.finalizeAttemptResources;
@@ -4326,7 +4382,8 @@ export class ProxyResponseHandler {
clientAborted,
discoveryLeaseLifecycle,
meteringSnapshot ? protocolObservationFromMetering(meteringSnapshot) : null,
- abortReason
+ abortReason,
+ passthroughFirstByteSeen
);
latestCommitSideEffects = finalized.commitSideEffects;
latestFinalizeAttemptResources = finalized.finalizeAttemptResources;
@@ -4704,6 +4761,9 @@ export class ProxyResponseHandler {
responsePump?.startDrain(reason ?? "client_detached");
return;
}
+ upstreamFirstByteSeenAtAbort = upstreamFirstByteSeen;
+ const deferredMeta = peekDeferredStreamingFinalization(session);
+ if (deferredMeta) deferredMeta.healthAbortAtMonotonic = performance.now();
clientDetachHandled = true;
clientAbortMeter?.switchToDetachedMode();
const activeReplaySpool = replaySpool && !replaySpool.isTerminal ? replaySpool : null;
@@ -4767,6 +4827,8 @@ export class ProxyResponseHandler {
// 统计/结算只保留有界的“头 + 尾”文本快照，避免长流式响应把进程堆撑满。
let usageForCost: UsageMetrics | null = null;
let isFirstChunk = true; // 标记是否为第一块数据
+ let upstreamFirstByteSeen = false;
+ let upstreamFirstByteSeenAtAbort: boolean | null = null;
// 不在首次读取前启动 idle timer（避免与首字节超时职责重叠）
// idle timer 仅在首块数据到达后启动，用于检测流中途静默。
@@ -4894,7 +4956,10 @@ export class ProxyResponseHandler {
clientAborted,
discoveryLeaseLifecycle,
streamProtocolObserver?.finish() ?? compactProtocolObservation,
- abortReason
+ abortReason,
+ clientAborted && upstreamFirstByteSeenAtAbort !== null
+ ? upstreamFirstByteSeenAtAbort
+ : upstreamFirstByteSeen
);
latestStreamCommitSideEffects = finalized.commitSideEffects
? [finalized.commitSideEffects]
@@ -5333,6 +5398,7 @@ export class ProxyResponseHandler {
const observeChunk = (value: Uint8Array) => {
const chunkSize = value.length;
+ if (chunkSize > 0) upstreamFirstByteSeen = true;
clearIdleTimer();
const metering = clientAbortMeter?.observe(value);
AsyncTaskManager.touch(taskId);
diff --git a/src/app/v1/_lib/proxy/session.ts b/src/app/v1/_lib/proxy/session.ts
index f4062eee7..babac2496 100644
--- a/src/app/v1/_lib/proxy/session.ts
+++ b/src/app/v1/_lib/proxy/session.ts
@@ -829,6 +829,7 @@ export class ProxySession {
| "hedge_loser_cancelled" // 该供应商输掉 Hedge 竞速，请求被取消（未计费）
| "hedge_loser_billed" // 该供应商输掉 Hedge 竞速，但其响应被后台拿回并计费
| "client_abort" // 客户端在响应完成前断开连接
+ | "client_abort_no_first_byte" // 客户端阈值后断开且供应商未返回首字节
| "affinity_hit"; // 最长前缀亲和命中（软提名，已通过全套硬校验）
selectionMethod?:
| "session_reuse"
@@ -1067,7 +1068,9 @@ export class ProxySession {
const resolvedOutcome =
outcome ??
(statusCode === 499
- ? "client_abort"
+ ? this.providerChain.at(-1)?.reason === "client_abort_no_first_byte"
+ ? "failed"
+ : "client_abort"
: this.routingTraceSummaryDraft?.outcome === "deadline" ||
this.routingTrace.summary?.outcome === "deadline"
? "deadline"
diff --git a/src/app/v1/_lib/proxy/stream-finalization.ts b/src/app/v1/_lib/proxy/stream-finalization.ts
index 6f34eb6db..746d5b9b4 100644
--- a/src/app/v1/_lib/proxy/stream-finalization.ts
+++ b/src/app/v1/_lib/proxy/stream-finalization.ts
@@ -73,6 +73,14 @@ export type DeferredStreamingFinalization = {
hedgeBindingHeartbeat?: DeferredStreamingBindingHeartbeat;
/** F1 门控提交标记：随成功链条目落库（高并发模式下为空）。 */
streamGate?: ProviderChainItem["streamGate"];
+ /** Optional attempt-scoped health attribution metadata for serial streaming requests. */
+ healthAttemptId?: string;
+ healthAttemptStartedAtMonotonic?: number;
+ healthAttributionThresholdMs?: number;
+ healthFirstByteSeen?: boolean;
+ healthAbortAtMonotonic?: number;
+ healthPausedDurationMs?: number;
+ healthOutcomeSettled?: boolean;
};
const deferredMeta = new WeakMap();
diff --git a/src/drizzle/schema.ts b/src/drizzle/schema.ts
index 08088d73d..3e9282ee2 100644
--- a/src/drizzle/schema.ts
+++ b/src/drizzle/schema.ts
@@ -11,6 +11,7 @@ import {
jsonb,
index,
uniqueIndex,
+ check,
pgEnum,
} from 'drizzle-orm/pg-core';
import { relations, sql } from 'drizzle-orm';
@@ -899,6 +900,9 @@ export const systemSettings = pgTable('system_settings', {
// 关闭：竞速输家直接取消连接，不计费（旧行为）
billHedgeLosers: boolean('bill_hedge_losers').notNull().default(true),
+ // Maximum number of simultaneously active attempts in the legacy streaming hedge.
+ legacyHedgeMaxInFlight: integer('legacy_hedge_max_in_flight').notNull().default(2),
+
// Bounded streaming Discovery (disabled by default until explicitly enabled).
discoveryEnabled: boolean('discovery_enabled').notNull().default(false),
discoveryConcurrency: integer('discovery_concurrency').notNull().default(2),
@@ -1067,7 +1071,12 @@ export const systemSettings = pgTable('system_settings', {
createdAt: timestamp('created_at', { withTimezone: true }).defaultNow(),
updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow(),
-});
+}, (table) => ({
+ legacyHedgeMaxInFlightRange: check(
+ 'system_settings_legacy_hedge_max_in_flight_range',
+ sql`${table.legacyHedgeMaxInFlight} >= 1 AND ${table.legacyHedgeMaxInFlight} <= 4`
+ ),
+}));
// Notification Settings table - Webhook 通知配置
export const notificationSettings = pgTable('notification_settings', {
diff --git a/src/lib/api-client/v1/openapi-types.gen.ts b/src/lib/api-client/v1/openapi-types.gen.ts
index 01539443a..a2d7ccee0 100644
--- a/src/lib/api-client/v1/openapi-types.gen.ts
+++ b/src/lib/api-client/v1/openapi-types.gen.ts
@@ -12139,6 +12139,8 @@ export interface operations {
billNonSuccessfulRequests: boolean;
/** @description Whether streaming-hedge (provider racing) losers are kept alive, drained, and billed (their cost accumulates into the request total). */
billHedgeLosers: boolean;
+ /** @description Maximum simultaneously active provider attempts for one legacy streaming hedge request (including the primary attempt). */
+ legacyHedgeMaxInFlight: number;
/** @description Whether bounded streaming Discovery is enabled. */
discoveryEnabled: boolean;
/** @description Maximum number of normal Discovery attempts in the initial batch. */
@@ -12428,6 +12430,8 @@ export interface operations {
billNonSuccessfulRequests?: boolean;
/** @description Whether streaming-hedge (provider racing) losers are kept alive, drained, and billed (their cost accumulates into the request total). */
billHedgeLosers?: boolean;
+ /** @description Maximum simultaneously active provider attempts for one legacy streaming hedge request (including the primary attempt). */
+ legacyHedgeMaxInFlight?: number;
/** @description Whether bounded streaming Discovery is enabled. */
discoveryEnabled?: boolean;
/** @description Maximum number of normal Discovery attempts in the initial batch. */
@@ -12590,6 +12594,8 @@ export interface operations {
billNonSuccessfulRequests: boolean;
/** @description Whether streaming-hedge (provider racing) losers are kept alive, drained, and billed (their cost accumulates into the request total). */
billHedgeLosers: boolean;
+ /** @description Maximum simultaneously active provider attempts for one legacy streaming hedge request (including the primary attempt). */
+ legacyHedgeMaxInFlight: number;
/** @description Whether bounded streaming Discovery is enabled. */
discoveryEnabled: boolean;
/** @description Maximum number of normal Discovery attempts in the initial batch. */
diff --git a/src/lib/api/v1/schemas/system-config.ts b/src/lib/api/v1/schemas/system-config.ts
index 944e4cab2..3c38a215c 100644
--- a/src/lib/api/v1/schemas/system-config.ts
+++ b/src/lib/api/v1/schemas/system-config.ts
@@ -106,6 +106,14 @@ export const SystemSettingsSchema = z
.describe(
"Whether streaming-hedge (provider racing) losers are kept alive, drained, and billed (their cost accumulates into the request total)."
),
+ legacyHedgeMaxInFlight: z
+ .number()
+ .int()
+ .min(1)
+ .max(4)
+ .describe(
+ "Maximum simultaneously active provider attempts for one legacy streaming hedge request (including the primary attempt)."
+ ),
discoveryEnabled: z.boolean().describe("Whether bounded streaming Discovery is enabled."),
discoveryConcurrency: z
.number()
diff --git a/src/lib/config/system-settings-cache.ts b/src/lib/config/system-settings-cache.ts
index 770877194..15d814c03 100644
--- a/src/lib/config/system-settings-cache.ts
+++ b/src/lib/config/system-settings-cache.ts
@@ -127,6 +127,7 @@ export const DEFAULT_SETTINGS: Pick<
| "stickySlaMs"
| "racingTotalTimeoutMs"
| "stickyTimeoutCooldownMs"
+ | "legacyHedgeMaxInFlight"
> = {
enableHttp2: false,
enableOpenaiResponsesWebsocket: true,
@@ -167,6 +168,7 @@ export const DEFAULT_SETTINGS: Pick<
stickySlaMs: 20_000,
racingTotalTimeoutMs: 60_000,
stickyTimeoutCooldownMs: 300_000,
+ legacyHedgeMaxInFlight: 2,
};
/**
@@ -227,6 +229,7 @@ export async function getCachedSystemSettings(): Promise {
codexPriorityBillingSource: DEFAULT_SETTINGS.codexPriorityBillingSource,
billNonSuccessfulRequests: false,
billHedgeLosers: true,
+ legacyHedgeMaxInFlight: DEFAULT_SETTINGS.legacyHedgeMaxInFlight,
timezone: null,
verboseProviderError: false,
passThroughUpstreamErrorMessage: DEFAULT_SETTINGS.passThroughUpstreamErrorMessage,
diff --git a/src/lib/langfuse/trace-proxy-request.test.ts b/src/lib/langfuse/trace-proxy-request.test.ts
index 43a842c0b..60ce55c06 100644
--- a/src/lib/langfuse/trace-proxy-request.test.ts
+++ b/src/lib/langfuse/trace-proxy-request.test.ts
@@ -26,6 +26,7 @@ const ERROR_REASONS = new Set([
"vendor_type_all_timeout",
"endpoint_pool_exhausted",
"client_abort",
+ "client_abort_no_first_byte",
]);
function isSuccessReason(reason: string | undefined): boolean {
@@ -71,6 +72,10 @@ describe("isErrorReason", () => {
expect(isErrorReason("client_abort")).toBe(true);
});
+ test("client_abort_no_first_byte is an error reason", () => {
+ expect(isErrorReason("client_abort_no_first_byte")).toBe(true);
+ });
+
test("system_error is an error reason", () => {
expect(isErrorReason("system_error")).toBe(true);
});
diff --git a/src/lib/langfuse/trace-proxy-request.ts b/src/lib/langfuse/trace-proxy-request.ts
index 3fb1aa237..4d63c84e1 100644
--- a/src/lib/langfuse/trace-proxy-request.ts
+++ b/src/lib/langfuse/trace-proxy-request.ts
@@ -85,6 +85,7 @@ const ERROR_REASONS = new Set([
"vendor_type_all_timeout",
"endpoint_pool_exhausted",
"client_abort",
+ "client_abort_no_first_byte",
]);
function isErrorReason(reason: string | undefined): boolean {
diff --git a/src/lib/ledger-backfill/trigger.sql b/src/lib/ledger-backfill/trigger.sql
index bbe051782..876ee55e7 100644
--- a/src/lib/ledger-backfill/trigger.sql
+++ b/src/lib/ledger-backfill/trigger.sql
@@ -44,7 +44,8 @@ BEGIN
'hedge_winner',
'hedge_loser_cancelled',
'hedge_loser_billed',
- 'client_abort'
+ 'client_abort',
+ 'client_abort_no_first_byte'
)
OR last_status_code IS NOT NULL
OR COALESCE(last_error_message, '') <> '' THEN
@@ -96,7 +97,9 @@ BEGIN
RETURN 'excluded';
END IF;
- IF COALESCE(last_status_code, status_code) IN (404, 499) THEN
+ IF COALESCE(last_status_code, status_code) = 404
+ OR (COALESCE(last_status_code, status_code) = 499
+ AND last_reason IS DISTINCT FROM 'client_abort_no_first_byte') THEN
RETURN 'excluded';
END IF;
diff --git a/src/lib/redis/live-chain-store.ts b/src/lib/redis/live-chain-store.ts
index 1a3f89a75..a58fcd123 100644
--- a/src/lib/redis/live-chain-store.ts
+++ b/src/lib/redis/live-chain-store.ts
@@ -151,6 +151,7 @@ function deriveLegacyActiveProviders(chain: ProviderChainItem[]): LiveProviderSn
case "hedge_loser_cancelled":
case "hedge_loser_billed":
case "client_abort":
+ case "client_abort_no_first_byte":
activeProviders.delete(item.id);
break;
}
@@ -192,6 +193,8 @@ export function inferPhase(
return "streaming";
case "client_abort":
return "aborted";
+ case "client_abort_no_first_byte":
+ return "failed";
default:
return "forwarding";
}
diff --git a/src/lib/request-outcome.ts b/src/lib/request-outcome.ts
index ae44378d2..e23106923 100644
--- a/src/lib/request-outcome.ts
+++ b/src/lib/request-outcome.ts
@@ -135,7 +135,10 @@ export function classifyRequestOutcomeSignal(
return buildExcludedTaxonomy("matched_rule");
}
- if (signal.statusCode === 499 || signal.reason === "client_abort") {
+ if (
+ (signal.statusCode === 499 && signal.reason !== "client_abort_no_first_byte") ||
+ signal.reason === "client_abort"
+ ) {
return buildExcludedTaxonomy("client_abort");
}
diff --git a/src/lib/utils/provider-chain-formatter.ts b/src/lib/utils/provider-chain-formatter.ts
index af8240013..c41191084 100644
--- a/src/lib/utils/provider-chain-formatter.ts
+++ b/src/lib/utils/provider-chain-formatter.ts
@@ -110,7 +110,8 @@ function getProviderStatus(item: ProviderChainItem): "✓" | "✗" | "⚡" | "
item.reason === "client_error_non_retryable" ||
item.reason === "endpoint_pool_exhausted" ||
item.reason === "vendor_type_all_timeout" ||
- item.reason === "client_abort"
+ item.reason === "client_abort" ||
+ item.reason === "client_abort_no_first_byte"
) {
return "✗";
}
@@ -153,7 +154,8 @@ export function isActualRequest(item: ProviderChainItem): boolean {
item.reason === "client_error_non_retryable" ||
item.reason === "endpoint_pool_exhausted" ||
item.reason === "vendor_type_all_timeout" ||
- item.reason === "client_abort"
+ item.reason === "client_abort" ||
+ item.reason === "client_abort_no_first_byte"
) {
return true;
}
diff --git a/src/lib/validation/schemas.ts b/src/lib/validation/schemas.ts
index 80ecdb9dd..e06dde3e2 100644
--- a/src/lib/validation/schemas.ts
+++ b/src/lib/validation/schemas.ts
@@ -32,6 +32,20 @@ export {
DISCOVERY_WINDOW_INVALID_ERROR_CODE,
} from "@/lib/validation/discovery-settings";
+export const LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE = "LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID";
+
+export function getLegacyHedgeMaxInFlightValidationErrorCode(
+ issues: ReadonlyArray<{ message: string; path: readonly PropertyKey[] }>
+): string | undefined {
+ return issues.some(
+ (issue) =>
+ issue.path[0] === "legacyHedgeMaxInFlight" ||
+ issue.message === LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE
+ )
+ ? LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE
+ : undefined;
+}
+
const CACHE_TTL_PREFERENCE = z.enum(["inherit", "5m", "1h"]);
const CONTEXT_1M_PREFERENCE = z.enum(["inherit", "force_enable", "disabled"]);
@@ -1022,6 +1036,16 @@ export const UpdateSystemSettingsSchema = z
billNonSuccessfulRequests: z.boolean().optional(),
// 供应商竞速输家计费（可选；默认开启）
billHedgeLosers: z.boolean().optional(),
+ // Legacy streaming hedge concurrency cap (inclusive of the primary attempt).
+ legacyHedgeMaxInFlight: z.preprocess(
+ (value) => (typeof value === "string" && value.trim() !== "" ? Number(value.trim()) : value),
+ z
+ .number(LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE)
+ .int(LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE)
+ .min(1, LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE)
+ .max(4, LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID_ERROR_CODE)
+ .optional()
+ ),
// Bounded streaming Discovery（默认关闭；启用前需满足总窗口约束）
discoveryEnabled: z.boolean().optional(),
discoveryConcurrency: z.coerce
diff --git a/src/repository/_shared/transformers.test.ts b/src/repository/_shared/transformers.test.ts
index 460917b8e..935efc24c 100644
--- a/src/repository/_shared/transformers.test.ts
+++ b/src/repository/_shared/transformers.test.ts
@@ -300,6 +300,16 @@ describe("src/repository/_shared/transformers.ts", () => {
expect(toSystemSettings({ replayCacheTtlMinutes: 45 }).replayCacheTtlMinutes).toBe(45);
});
+ it.each([1, 2, 4])("应保留有效 legacy hedge 并发上限 %s", (value) => {
+ expect(toSystemSettings({ legacyHedgeMaxInFlight: value }).legacyHedgeMaxInFlight).toBe(
+ value
+ );
+ });
+
+ it.each([0, 5, 1.5, "3", null])("应将无效 legacy hedge 并发上限 %s 回退为 2", (value) => {
+ expect(toSystemSettings({ legacyHedgeMaxInFlight: value }).legacyHedgeMaxInFlight).toBe(2);
+ });
+
it.each([5, 120])("应保留 Replay 缓存时间的有效边界 %s", (value) => {
expect(toSystemSettings({ replayCacheTtlMinutes: value }).replayCacheTtlMinutes).toBe(value);
});
diff --git a/src/repository/_shared/transformers.ts b/src/repository/_shared/transformers.ts
index b49454fd8..5a53d2c0a 100644
--- a/src/repository/_shared/transformers.ts
+++ b/src/repository/_shared/transformers.ts
@@ -257,6 +257,13 @@ export function toSystemSettings(dbSettings: any): SystemSettings {
replayCacheTtlMinutes <= REPLAY_CACHE_TTL_MINUTES_MAX
? replayCacheTtlMinutes
: REPLAY_CACHE_TTL_MINUTES_DEFAULT;
+ const legacyHedgeMaxInFlight =
+ typeof dbSettings?.legacyHedgeMaxInFlight === "number" &&
+ Number.isInteger(dbSettings.legacyHedgeMaxInFlight) &&
+ dbSettings.legacyHedgeMaxInFlight >= 1 &&
+ dbSettings.legacyHedgeMaxInFlight <= 4
+ ? dbSettings.legacyHedgeMaxInFlight
+ : 2;
return {
id: dbSettings?.id ?? 0,
@@ -271,6 +278,7 @@ export function toSystemSettings(dbSettings: any): SystemSettings {
: "requested",
billNonSuccessfulRequests: dbSettings?.billNonSuccessfulRequests ?? false,
billHedgeLosers: dbSettings?.billHedgeLosers ?? true,
+ legacyHedgeMaxInFlight,
timezone: dbSettings?.timezone ?? null,
enableAutoCleanup: dbSettings?.enableAutoCleanup ?? false,
cleanupRetentionDays: dbSettings?.cleanupRetentionDays ?? 30,
diff --git a/src/repository/_shared/usage-log-filters.ts b/src/repository/_shared/usage-log-filters.ts
index 2ae3e9c8c..6f5089052 100644
--- a/src/repository/_shared/usage-log-filters.ts
+++ b/src/repository/_shared/usage-log-filters.ts
@@ -122,6 +122,7 @@ export const RETRY_COUNT_EXPR: SQL = sql`(
'endpoint_pool_exhausted',
'vendor_type_all_timeout',
'client_abort',
+ 'client_abort_no_first_byte',
'http2_fallback'
)
OR (
diff --git a/src/repository/system-config.ts b/src/repository/system-config.ts
index c26d8fcba..22e7a9f98 100644
--- a/src/repository/system-config.ts
+++ b/src/repository/system-config.ts
@@ -151,6 +151,7 @@ function createFallbackSettings(): SystemSettings {
codexPriorityBillingSource: "requested",
billNonSuccessfulRequests: false,
billHedgeLosers: true,
+ legacyHedgeMaxInFlight: 2,
timezone: null,
enableAutoCleanup: false,
cleanupRetentionDays: 30,
@@ -284,6 +285,12 @@ const RECENT_COLUMN_LADDER: ReadonlyArray<{
// 本层更新失败（仍有列缺失）时记录的告警
updateWarn: string;
}> = [
+ {
+ key: "legacyHedgeMaxInFlight",
+ column: systemSettings.legacyHedgeMaxInFlight,
+ selectWarn: "system_settings 缺少 legacyHedgeMaxInFlight,回退到上一代字段集。",
+ updateWarn: "system_settings 缺少 legacyHedgeMaxInFlight,继续降级更新。",
+ },
{
key: "replayCacheTtlMinutes",
column: systemSettings.replayCacheTtlMinutes,
@@ -414,6 +421,7 @@ const RECENT_COLUMN_LADDER: ReadonlyArray<{
// 历史世代字段集（冻结）：passThrough 世代之前的 schema 没有以下五列。
// 注意：世代字段集相对近代阶梯末层会重新选取更晚引入的列（与历史实现一致）。
const PASS_THROUGH_ERA_OMIT: readonly string[] = [
+ "legacyHedgeMaxInFlight",
"billHedgeLosers",
"billNonSuccessfulRequests",
"passThroughUpstreamErrorMessage",
@@ -715,6 +723,10 @@ export async function updateSystemSettings(
updates.billHedgeLosers = payload.billHedgeLosers;
}
+ if (payload.legacyHedgeMaxInFlight !== undefined) {
+ updates.legacyHedgeMaxInFlight = payload.legacyHedgeMaxInFlight;
+ }
+
if (payload.discoveryEnabled !== undefined) {
updates.discoveryEnabled = payload.discoveryEnabled;
}
diff --git a/src/types/message.ts b/src/types/message.ts
index ea932916c..4f0d11018 100644
--- a/src/types/message.ts
+++ b/src/types/message.ts
@@ -50,6 +50,7 @@ export interface ProviderChainItem {
| "hedge_loser_cancelled" // 该供应商输掉 Hedge 竞速，请求被取消（未对输家计费）
| "hedge_loser_billed" // 该供应商输掉 Hedge 竞速，但其上游响应被后台拿回并计费
| "client_abort" // 客户端在响应完成前断开连接
+ | "client_abort_no_first_byte" // 客户端在阈值后断开且该供应商未返回首字节
| "affinity_hit"; // 最长前缀亲和命中（软提名，已通过全套硬校验）
// === 选择方法（细化） ===
diff --git a/src/types/routing-trace.ts b/src/types/routing-trace.ts
index 665b020f1..9d68f6fa4 100644
--- a/src/types/routing-trace.ts
+++ b/src/types/routing-trace.ts
@@ -15,6 +15,8 @@ export type RoutingTraceEventType =
| "fallback_promoted"
| "winner_committed"
| "binding_finalized"
+ | "hedge_slot_saturated"
+ | "client_abort_no_first_byte"
| "request_finished";
export type RoutingTraceAttemptKind = "sticky" | "normal" | "fallback";
@@ -30,6 +32,7 @@ export interface RoutingTraceConfigV1 {
stickyTimeoutCooldownMs: number;
/** The binding/session TTL in seconds; optional for traces written before this field existed. */
sessionTtlSeconds?: number;
+ legacyHedgeMaxInFlight?: number;
}
export interface RoutingTraceProviderV1 {
@@ -52,6 +55,11 @@ export interface RoutingTraceEventV1 {
provider?: RoutingTraceProviderV1;
outcome?: string;
cancellationKind?: string;
+ effectiveThresholdMs?: number;
+ activeAttemptCount?: number;
+ configuredCap?: number;
+ circuitAccountingApplied?: boolean;
+ availabilityAccountingApplied?: boolean;
statusCode?: number;
reason?: string;
bindingAction?: "create" | "renew" | "clear" | "none";
@@ -106,6 +114,8 @@ const ROUTING_TRACE_EVENT_TYPES = new Set([
"fallback_promoted",
"winner_committed",
"binding_finalized",
+ "hedge_slot_saturated",
+ "client_abort_no_first_byte",
"request_finished",
]);
@@ -166,6 +176,21 @@ function normalizeRoutingTraceEvent(value: unknown): RoutingTraceEventV1 | null
...(nonEmptyString(event.cancellationKind)
? { cancellationKind: event.cancellationKind as string }
: {}),
+ ...(finiteNumber(event.effectiveThresholdMs) !== undefined
+ ? { effectiveThresholdMs: event.effectiveThresholdMs as number }
+ : {}),
+ ...(finiteNumber(event.activeAttemptCount) !== undefined
+ ? { activeAttemptCount: event.activeAttemptCount as number }
+ : {}),
+ ...(finiteNumber(event.configuredCap) !== undefined
+ ? { configuredCap: event.configuredCap as number }
+ : {}),
+ ...(typeof event.circuitAccountingApplied === "boolean"
+ ? { circuitAccountingApplied: event.circuitAccountingApplied }
+ : {}),
+ ...(typeof event.availabilityAccountingApplied === "boolean"
+ ? { availabilityAccountingApplied: event.availabilityAccountingApplied }
+ : {}),
...(finiteNumber(event.statusCode) !== undefined
? { statusCode: event.statusCode as number }
: {}),
@@ -191,6 +216,7 @@ function normalizeRoutingTraceConfig(value: unknown): RoutingTraceConfigV1 | und
const racingTotalTimeoutMs = finiteNumber(config.racingTotalTimeoutMs);
const stickyTimeoutCooldownMs = finiteNumber(config.stickyTimeoutCooldownMs);
const sessionTtlSeconds = finiteNumber(config.sessionTtlSeconds);
+ const legacyHedgeMaxInFlight = finiteNumber(config.legacyHedgeMaxInFlight);
if (
discoveryConcurrency === undefined ||
maxDiscoveryRounds === undefined ||
@@ -209,6 +235,7 @@ function normalizeRoutingTraceConfig(value: unknown): RoutingTraceConfigV1 | und
racingTotalTimeoutMs,
stickyTimeoutCooldownMs,
...(sessionTtlSeconds !== undefined ? { sessionTtlSeconds } : {}),
+ ...(legacyHedgeMaxInFlight !== undefined ? { legacyHedgeMaxInFlight } : {}),
};
}
diff --git a/src/types/system-config.ts b/src/types/system-config.ts
index a1eea2b3a..693845799 100644
--- a/src/types/system-config.ts
+++ b/src/types/system-config.ts
@@ -51,6 +51,9 @@ export interface SystemSettings {
// 其费用异步累加进该请求的总花费（与上游对多个供应商分别计费保持一致）。
billHedgeLosers: boolean;
+ // Legacy streaming hedge concurrency cap (includes the primary attempt).
+ legacyHedgeMaxInFlight: number;
+
// 系统时区配置 (IANA timezone identifier)
// 用于统一后端时间边界计算和前端日期/时间显示
// null 表示使用环境变量 TZ 或默认 UTC
@@ -201,6 +204,9 @@ export interface UpdateSystemSettingsInput {
// 供应商竞速输家计费（可选）
billHedgeLosers?: boolean;
+ // Legacy streaming hedge concurrency cap (includes the primary attempt).
+ legacyHedgeMaxInFlight?: number;
+
discoveryEnabled?: boolean;
discoveryConcurrency?: number;
maxDiscoveryRounds?: number;
diff --git a/tests/api/v1/system/system-config.test.ts b/tests/api/v1/system/system-config.test.ts
index 440c23239..487139755 100644
--- a/tests/api/v1/system/system-config.test.ts
+++ b/tests/api/v1/system/system-config.test.ts
@@ -42,6 +42,7 @@ const settings: SystemSettings = {
currencyDisplay: "USD",
billingModelSource: "original",
codexPriorityBillingSource: "requested",
+ legacyHedgeMaxInFlight: 2,
timezone: "Asia/Shanghai",
enableAutoCleanup: false,
cleanupRetentionDays: 30,
@@ -225,6 +226,23 @@ describe("v1 system config endpoints", () => {
expect(saveSystemSettingsMock).not.toHaveBeenCalled();
});
+ test("returns a stable error code for invalid legacy hedge concurrency", async () => {
+ for (const value of [0, 5, true, [2]]) {
+ const invalid = await callV1Route({
+ method: "PUT",
+ pathname: "/api/v1/system/settings",
+ headers: { Authorization: "Bearer admin-token" },
+ body: { legacyHedgeMaxInFlight: value },
+ });
+
+ expect(invalid.response.status).toBe(400);
+ expect(invalid.json).toMatchObject({
+ errorCode: "LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID",
+ });
+ }
+ expect(saveSystemSettingsMock).not.toHaveBeenCalled();
+ });
+
test("rejects malformed and non-json system settings update bodies", async () => {
const handlers = await import("@/app/api/v1/resources/system/handlers");
const malformed = await handlers.updateSystemSettings({
diff --git a/tests/integration/billing-model-source.test.ts b/tests/integration/billing-model-source.test.ts
index f8576f970..e8b1db5cd 100644
--- a/tests/integration/billing-model-source.test.ts
+++ b/tests/integration/billing-model-source.test.ts
@@ -146,6 +146,7 @@ function makeSystemSettings(
currencyDisplay: "USD",
billingModelSource,
codexPriorityBillingSource,
+ legacyHedgeMaxInFlight: 2,
timezone: null,
enableAutoCleanup: false,
cleanupRetentionDays: 30,
diff --git a/tests/unit/api/admin-system-config-route.test.ts b/tests/unit/api/admin-system-config-route.test.ts
index d566c8699..873d19dcf 100644
--- a/tests/unit/api/admin-system-config-route.test.ts
+++ b/tests/unit/api/admin-system-config-route.test.ts
@@ -103,4 +103,24 @@ describe("POST /api/admin/system-config", () => {
});
expect(mocks.updateSystemSettings).not.toHaveBeenCalled();
});
+
+ it.each([[0], [5], [true], [[2]]])(
+ "returns a stable error for invalid legacy hedge concurrency %s",
+ async (value) => {
+ const { POST } = await import("@/app/api/admin/system-config/route");
+ const response = await POST(
+ new Request("http://localhost/api/admin/system-config", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({ legacyHedgeMaxInFlight: value }),
+ })
+ );
+
+ expect(response.status).toBe(400);
+ await expect(response.json()).resolves.toEqual({
+ error: "LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID",
+ });
+ expect(mocks.updateSystemSettings).not.toHaveBeenCalled();
+ }
+ );
});
diff --git a/tests/unit/lib/config/system-settings-cache.test.ts b/tests/unit/lib/config/system-settings-cache.test.ts
index 2fc12998c..ce852000a 100644
--- a/tests/unit/lib/config/system-settings-cache.test.ts
+++ b/tests/unit/lib/config/system-settings-cache.test.ts
@@ -47,6 +47,7 @@ function createSettings(overrides: Partial = {}): SystemSettings
currencyDisplay: "USD",
billingModelSource: "original",
codexPriorityBillingSource: "requested",
+ legacyHedgeMaxInFlight: 2,
timezone: null,
enableAutoCleanup: false,
cleanupRetentionDays: 30,
diff --git a/tests/unit/lib/request-outcome.test.ts b/tests/unit/lib/request-outcome.test.ts
index a1c1afb2e..735751c08 100644
--- a/tests/unit/lib/request-outcome.test.ts
+++ b/tests/unit/lib/request-outcome.test.ts
@@ -22,6 +22,20 @@ describe("request outcome taxonomy", () => {
});
});
+ it("counts thresholded no-first-byte client aborts as provider failures", () => {
+ expect(
+ classifyRequestOutcomeSignal({
+ reason: "client_abort_no_first_byte",
+ statusCode: 499,
+ errorMessage: "Client aborted before provider first byte threshold",
+ })
+ ).toMatchObject({
+ outcome: "failure",
+ locus: "upstream",
+ countability: "countable",
+ });
+ });
+
it("excludes both cancelled and billed hedge losers from success-rate", () => {
expect(
classifyRequestOutcomeSignal({
diff --git a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts
index 5301c56ce..36f4a5733 100644
--- a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts
+++ b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts
@@ -1823,6 +1823,202 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => {
}
});
+ test("legacy serial client abort after the health threshold records one provider failure", async () => {
+ vi.useFakeTimers();
+
+ try {
+ const provider = createProvider({ id: 1, name: "p1", firstByteTimeoutStreamingMs: 0 });
+ const clientAbortController = new AbortController();
+ const session = createSession(clientAbortController.signal);
+ setProviderWithSessionRef(session, provider);
+ mocks.categorizeErrorAsync.mockResolvedValueOnce(ProxyErrorCategory.CLIENT_ABORT);
+
+ const doForward = vi.spyOn(
+ ProxyForwarder as unknown as {
+ doForward: (...args: unknown[]) => Promise;
+ },
+ "doForward"
+ );
+ doForward.mockImplementationOnce(async (...args) => {
+ const runtime = args[0] as ProxySession & AttemptRuntime;
+ runtime.clearResponseTimeout = vi.fn();
+ (args[7] as () => void)();
+ return await new Promise((_, reject) => {
+ setTimeout(() => {
+ clientAbortController.abort(new Error("client_cancelled"));
+ reject(new UpstreamProxyError("Request aborted by client", 499, undefined, true));
+ }, 30_000);
+ });
+ });
+
+ const responsePromise = ProxyForwarder.send(session);
+ const rejection = expect(responsePromise).rejects.toMatchObject({ statusCode: 499 });
+ await vi.advanceTimersByTimeAsync(30_000);
+ await rejection;
+ expect(mocks.recordFailure).toHaveBeenCalledTimes(1);
+ expect(mocks.recordFailure).toHaveBeenCalledWith(provider.id, expect.any(Error));
+ expect(
+ session
+ .getRoutingTrace()
+ ?.events.some((event) => event.type === "client_abort_no_first_byte")
+ ).toBe(true);
+ expect(
+ session.getProviderChain().some((item) => item.reason === "client_abort_no_first_byte")
+ ).toBe(true);
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+
+ test("legacy hedge cap one preserves serial timeout fallback and records saturation", async () => {
+ vi.useFakeTimers();
+
+ try {
+ const provider1 = createProvider({ id: 1, name: "p1", firstByteTimeoutStreamingMs: 100 });
+ const provider2 = createProvider({ id: 2, name: "p2", firstByteTimeoutStreamingMs: 100 });
+ const session = createSession();
+ setProviderWithSessionRef(session, provider1);
+ mocks.getCachedSystemSettings.mockResolvedValue({
+ billHedgeLosers: false,
+ legacyHedgeMaxInFlight: 1,
+ enableThinkingSignatureRectifier: true,
+ enableThinkingBudgetRectifier: true,
+ });
+ mocks.pickRandomProviderWithExclusion.mockResolvedValueOnce(provider2);
+
+ const doForward = vi.spyOn(
+ ProxyForwarder as unknown as {
+ doForward: (...args: unknown[]) => Promise;
+ },
+ "doForward"
+ );
+ const controller1 = new AbortController();
+ const controller2 = new AbortController();
+ doForward.mockImplementationOnce(async (attemptSession, providerForRequest, ...args) => {
+ const runtime = attemptSession as ProxySession & AttemptRuntime;
+ runtime.responseController = controller1;
+ runtime.clearResponseTimeout = vi.fn();
+ (args[5] as (() => void) | undefined)?.();
+ expect((providerForRequest as Provider).firstByteTimeoutStreamingMs).toBe(100);
+ return createDelayedFailure({
+ delayMs: 100,
+ error: new Error("p1 timed out"),
+ controller: controller1,
+ });
+ });
+ doForward.mockImplementationOnce(async (attemptSession) => {
+ const runtime = attemptSession as ProxySession & AttemptRuntime;
+ runtime.responseController = controller2;
+ runtime.clearResponseTimeout = vi.fn();
+ return createStreamingResponse({
+ label: "p2",
+ firstChunkDelayMs: 10,
+ controller: controller2,
+ });
+ });
+
+ const responsePromise = ProxyForwarder.send(session);
+ await vi.advanceTimersByTimeAsync(100);
+ expect(doForward).toHaveBeenCalledTimes(2);
+ await vi.advanceTimersByTimeAsync(10);
+ const response = await responsePromise;
+ expect(await response.text()).toContain('"provider":"p2"');
+ expect(session.getRoutingTrace()?.mode).toBe("legacy_hedge");
+ expect(session.getRoutingTrace()?.config?.legacyHedgeMaxInFlight).toBe(1);
+ expect(
+ session.getRoutingTrace()?.events.filter((event) => event.type === "hedge_slot_saturated")
+ ).toEqual([
+ expect.objectContaining({
+ activeAttemptCount: 1,
+ configuredCap: 1,
+ }),
+ ]);
+ expect(mocks.recordFailure).toHaveBeenCalledWith(1, expect.any(Error));
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+
+ test("legacy hedge cap three launches a third candidate on the next threshold", async () => {
+ vi.useFakeTimers();
+
+ try {
+ const provider1 = createProvider({ id: 1, name: "p1", firstByteTimeoutStreamingMs: 100 });
+ const provider2 = createProvider({ id: 2, name: "p2", firstByteTimeoutStreamingMs: 100 });
+ const provider3 = createProvider({ id: 3, name: "p3", firstByteTimeoutStreamingMs: 100 });
+ const session = createSession();
+ setProviderWithSessionRef(session, provider1);
+ mocks.getCachedSystemSettings.mockResolvedValue({
+ billHedgeLosers: false,
+ legacyHedgeMaxInFlight: 3,
+ enableThinkingSignatureRectifier: true,
+ enableThinkingBudgetRectifier: true,
+ });
+ mocks.pickRandomProviderWithExclusion
+ .mockResolvedValueOnce(provider2)
+ .mockResolvedValueOnce(provider3);
+
+ const doForward = vi.spyOn(
+ ProxyForwarder as unknown as {
+ doForward: (...args: unknown[]) => Promise;
+ },
+ "doForward"
+ );
+ const controller1 = new AbortController();
+ const controller2 = new AbortController();
+ const controller3 = new AbortController();
+ doForward
+ .mockImplementationOnce(async (attemptSession) => {
+ const runtime = attemptSession as ProxySession & AttemptRuntime;
+ runtime.responseController = controller1;
+ runtime.clearResponseTimeout = vi.fn();
+ return createStreamingResponse({
+ label: "p1",
+ firstChunkDelayMs: 1000,
+ controller: controller1,
+ });
+ })
+ .mockImplementationOnce(async (attemptSession) => {
+ const runtime = attemptSession as ProxySession & AttemptRuntime;
+ runtime.responseController = controller2;
+ runtime.clearResponseTimeout = vi.fn();
+ return createStreamingResponse({
+ label: "p2",
+ firstChunkDelayMs: 1000,
+ controller: controller2,
+ });
+ })
+ .mockImplementationOnce(async (attemptSession) => {
+ const runtime = attemptSession as ProxySession & AttemptRuntime;
+ runtime.responseController = controller3;
+ runtime.clearResponseTimeout = vi.fn();
+ return createStreamingResponse({
+ label: "p3",
+ firstChunkDelayMs: 10,
+ controller: controller3,
+ });
+ });
+
+ const responsePromise = ProxyForwarder.send(session);
+ await vi.advanceTimersByTimeAsync(100);
+ expect(doForward).toHaveBeenCalledTimes(2);
+ await vi.advanceTimersByTimeAsync(100);
+ expect(doForward).toHaveBeenCalledTimes(3);
+ await vi.advanceTimersByTimeAsync(10);
+ const response = await responsePromise;
+ expect(await response.text()).toContain('"provider":"p3"');
+ expect(controller1.signal.aborted).toBe(true);
+ expect(controller2.signal.aborted).toBe(true);
+ expect(controller3.signal.aborted).toBe(false);
+ expect(session.getRoutingTrace()?.config?.legacyHedgeMaxInFlight).toBe(3);
+ expect(
+ session.getRoutingTrace()?.events.some((event) => event.type === "hedge_slot_saturated")
+ ).toBe(false);
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+
test("client abort before any winner should abort all in-flight attempts, return 499, and clear sticky provider binding", async () => {
vi.useFakeTimers();
@@ -1870,10 +2066,11 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => {
const controller1 = new AbortController();
const controller2 = new AbortController();
- doForward.mockImplementationOnce(async (attemptSession, providerForRequest) => {
+ doForward.mockImplementationOnce(async (attemptSession, providerForRequest, ...args) => {
const runtime = attemptSession as ProxySession & AttemptRuntime;
runtime.responseController = controller1;
runtime.clearResponseTimeout = vi.fn();
+ (args[5] as (() => void) | undefined)?.();
expect(
ModelRedirector.apply(attemptSession as ProxySession, providerForRequest as Provider)
).toBe(true);
@@ -1884,10 +2081,11 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => {
});
});
- doForward.mockImplementationOnce(async (attemptSession, providerForRequest) => {
+ doForward.mockImplementationOnce(async (attemptSession, providerForRequest, ...args) => {
const runtime = attemptSession as ProxySession & AttemptRuntime;
runtime.responseController = controller2;
runtime.clearResponseTimeout = vi.fn();
+ (args[5] as (() => void) | undefined)?.();
expect(
ModelRedirector.apply(attemptSession as ProxySession, providerForRequest as Provider)
).toBe(true);
@@ -1913,13 +2111,14 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => {
expect(controller1.signal.aborted).toBe(true);
expect(controller2.signal.aborted).toBe(true);
expect(mocks.clearSessionProviders).toHaveBeenCalledWith("sess-hedge", new Set([1, 2]), null);
- expect(mocks.recordFailure).not.toHaveBeenCalled();
+ expect(mocks.recordFailure).toHaveBeenCalledTimes(1);
expect(mocks.recordSuccess).not.toHaveBeenCalled();
const chain = session.getProviderChain();
expect(
- chain.find((item) => item.id === provider1.id && item.reason === "client_abort")
- ?.modelRedirect
+ chain.find(
+ (item) => item.id === provider1.id && item.reason === "client_abort_no_first_byte"
+ )?.modelRedirect
).toMatchObject({
originalModel: requestedModel,
redirectedModel: "accounts/fireworks/routers/kimi-k2p5-turbo",
diff --git a/tests/unit/proxy/response-handler-client-abort-drain.test.ts b/tests/unit/proxy/response-handler-client-abort-drain.test.ts
index 779d18c9a..e3a6b9064 100644
--- a/tests/unit/proxy/response-handler-client-abort-drain.test.ts
+++ b/tests/unit/proxy/response-handler-client-abort-drain.test.ts
@@ -2199,6 +2199,39 @@ describe("ProxyResponseHandler stream client abort finalization", () => {
);
});
+ it("attributes an old no-first-byte client abort once", async () => {
+ vi.mocked(recordFailure).mockClear();
+ const clientController = new AbortController();
+ const session = createSession(clientController.signal);
+ setDeferredStreamingFinalization(session, {
+ providerId: 1,
+ providerName: "avemujica-responses",
+ providerPriority: 1,
+ attemptNumber: 1,
+ totalProvidersAttempted: 1,
+ isFirstAttempt: true,
+ isFailoverSuccess: false,
+ endpointId: 42,
+ endpointUrl: "https://api.test.invalid/v1",
+ upstreamStatusCode: 200,
+ healthAttemptId: "legacy-serial-1-1",
+ healthAttemptStartedAtMonotonic: performance.now() - 1_000,
+ healthAttributionThresholdMs: 1,
+ healthFirstByteSeen: false,
+ });
+ const upstream = createControllableEmptyResponsesSse();
+
+ await ProxyResponseHandler.dispatch(session, upstream.response);
+ clientController.abort(new Error("client detached"));
+ upstream.close();
+ await drainAsyncTasks();
+
+ expect(recordFailure).toHaveBeenCalledTimes(1);
+ expect(session.getProviderChain()).toEqual(
+ expect.arrayContaining([expect.objectContaining({ reason: "client_abort_no_first_byte" })])
+ );
+ });
+
it("stops a detached source as soon as compact terminal usage is captured", async () => {
const clientController = new AbortController();
const session = createSession(clientController.signal);
diff --git a/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts b/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts
index c1a0df38c..d51cb3014 100644
--- a/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts
+++ b/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts
@@ -8,6 +8,7 @@ import { ProxySession } from "@/app/v1/_lib/proxy/session";
import { setDeferredStreamingFinalization } from "@/app/v1/_lib/proxy/stream-finalization";
import { AsyncTaskManager } from "@/lib/async-task-manager";
import { SessionManager } from "@/lib/session-manager";
+import { recordFailure } from "@/lib/circuit-breaker";
import {
updateMessageRequestDetails,
updateMessageRequestDetailsDurably,
@@ -774,6 +775,66 @@ describe("ProxyResponseHandler - Gemini stream passthrough timeouts", () => {
}
});
+ test("Gemini passthrough does not attribute a client abort after the first byte", async () => {
+ asyncTasks.length = 0;
+ vi.mocked(recordFailure).mockClear();
+ const clientAbortController = new AbortController();
+ const provider = createProvider({ firstByteTimeoutStreamingMs: 1 });
+ const session = createSession({
+ clientAbortSignal: clientAbortController.signal,
+ messageId: 5,
+ userId: 1,
+ });
+ session.setProvider(provider);
+ setDeferredStreamingFinalization(session, {
+ providerId: provider.id,
+ providerName: provider.name,
+ providerPriority: provider.priority,
+ attemptNumber: 1,
+ totalProvidersAttempted: 1,
+ isFirstAttempt: true,
+ isFailoverSuccess: false,
+ endpointId: null,
+ endpointUrl: provider.url,
+ upstreamStatusCode: 200,
+ healthAttemptId: "legacy-serial-1-1",
+ healthAttemptStartedAtMonotonic: performance.now() - 1_000,
+ healthAttributionThresholdMs: 1,
+ healthFirstByteSeen: false,
+ });
+ const encoder = new TextEncoder();
+ let upstreamController: ReadableStreamDefaultController | null = null;
+ const upstream = new ReadableStream({
+ start(controller) {
+ upstreamController = controller;
+ controller.enqueue(
+ encoder.encode('{"candidates":[{"content":{"parts":[{"text":"x"}]}}]}\n')
+ );
+ },
+ });
+
+ const downstream = await (
+ ProxyResponseHandler as unknown as {
+ handleStream: (session: ProxySession, response: Response) => Promise;
+ }
+ ).handleStream(
+ session,
+ new Response(upstream, { status: 200, headers: { "content-type": "text/event-stream" } })
+ );
+ const reader = downstream.body?.getReader();
+ expect(reader).toBeTruthy();
+ if (!reader) throw new Error("Missing body reader");
+ await reader.read();
+ clientAbortController.abort(new Error("client_cancelled"));
+ upstreamController?.close();
+ await expectAllFulfilled(asyncTasks);
+
+ expect(recordFailure).not.toHaveBeenCalled();
+ expect(session.getProviderChain()).not.toEqual(
+ expect.arrayContaining([expect.objectContaining({ reason: "client_abort_no_first_byte" })])
+ );
+ });
+
test("Gemini 流式透传超大单 chunk 应保留尾部 usage 且不把截断快照作为完整正文存储", async () => {
asyncTasks.length = 0;
vi.mocked(SessionManager.storeSessionResponse).mockClear();
diff --git a/tests/unit/proxy/routing-trace.test.ts b/tests/unit/proxy/routing-trace.test.ts
index 9317fd80e..74f07506d 100644
--- a/tests/unit/proxy/routing-trace.test.ts
+++ b/tests/unit/proxy/routing-trace.test.ts
@@ -61,6 +61,7 @@ function makeTraceSession(startTime = 1_000): ProxySession {
liveObservabilityFlushPromise: null,
liveObservabilityClosePromise: null,
liveObservabilityClosed: false,
+ liveActiveProviders: new Map(),
routingTraceTerminalLogged: false,
providerChain: [],
ttftMs: null,
@@ -190,6 +191,47 @@ describe("ProxySession routing trace recorder", () => {
nowSpy.mockRestore();
});
+ it("marks a thresholded client abort as a failed routing outcome", () => {
+ const session = makeTraceSession();
+ session.initializeRoutingTrace({
+ mode: "legacy_hedge",
+ discoveryEnabled: false,
+ eligible: false,
+ startedAt: 1_000,
+ });
+ session.addProviderToChain(
+ { id: 1, name: "slow", providerType: "openai", priority: 1 } as never,
+ { reason: "client_abort_no_first_byte", attemptNumber: 1 }
+ );
+ session.setRoutingTraceSummary({
+ outcome: "client_abort",
+ statusCode: 499,
+ durationMs: 1_000,
+ ttftMs: null,
+ attemptsPerRequest: 1,
+ maxActiveAttempts: 1,
+ rounds: 0,
+ providerMs: 1_000,
+ fallbackPromotions: 0,
+ cancelFailures: 0,
+ winnerOrigin: "none",
+ winnerProviderId: null,
+ winnerRound: null,
+ });
+
+ session.finalizeRoutingTrace(499);
+
+ expect(session.getRoutingTrace()?.summary).toMatchObject({
+ outcome: "failed",
+ statusCode: 499,
+ });
+ expect(session.getRoutingTrace()?.events.at(-1)).toMatchObject({
+ type: "request_finished",
+ outcome: "failed",
+ statusCode: 499,
+ });
+ });
+
it("caps the trace at 512 events and persists the truncated snapshot independently", async () => {
const session = makeTraceSession();
session.initializeRoutingTrace({
diff --git a/tests/unit/repository/system-config-degradation-ladder.test.ts b/tests/unit/repository/system-config-degradation-ladder.test.ts
index 5469eaf3c..08b0ce198 100644
--- a/tests/unit/repository/system-config-degradation-ladder.test.ts
+++ b/tests/unit/repository/system-config-degradation-ladder.test.ts
@@ -7,6 +7,7 @@ import type { UpdateSystemSettingsInput } from "@/types/system-config";
// 近代新增列（最新在前），降级链按引入顺序逐层累计剥离。
const RECENT_COLUMNS = [
+ "legacyHedgeMaxInFlight",
"replayCacheTtlMinutes",
"cacheEffectivenessEnabled",
"replayEnabled",
@@ -44,6 +45,7 @@ const FULL_COLUMNS = [
"stickyTimeoutCooldownMs",
"enableGeminiFunctionIdRectifier",
"billHedgeLosers",
+ "legacyHedgeMaxInFlight",
"billNonSuccessfulRequests",
"passThroughUpstreamErrorMessage",
"fakeStreamingWhitelist",
@@ -91,6 +93,7 @@ const FULL_COLUMNS = [
// 历史世代字段集（冻结）：passThrough 世代之前的 schema 没有以下五列，
// 但仍包含 enableThinkingEffortConflictRectifier / allowNonConversationEndpointProviderFallback。
const PASS_THROUGH_ERA_OMIT = [
+ "legacyHedgeMaxInFlight",
"billHedgeLosers",
"billNonSuccessfulRequests",
"passThroughUpstreamErrorMessage",
diff --git a/tests/unit/repository/system-config-update-missing-columns.test.ts b/tests/unit/repository/system-config-update-missing-columns.test.ts
index b968e07e4..fb0596f82 100644
--- a/tests/unit/repository/system-config-update-missing-columns.test.ts
+++ b/tests/unit/repository/system-config-update-missing-columns.test.ts
@@ -299,12 +299,12 @@ describe("SystemSettings：数据库缺列时的保存兜底", () => {
vi.setSystemTime(now);
// 第一次 select(fullSelection) 因新列缺失而抛 42703；
- // 第二次仅去掉最新的 replayCacheTtlMinutes 后仍失败;
- // 第三次累计去掉 cacheEffectivenessEnabled 后命中.
+ // The new legacy hedge column is the newest rung, so it is stripped before replay columns.
const selectMock = vi
.fn()
.mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
.mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
+ .mockReturnValueOnce(createRejectedThenableQuery({ code: "42703" }))
.mockReturnValueOnce(
createThenableQuery([
{
@@ -337,22 +337,23 @@ describe("SystemSettings：数据库缺列时的保存兜底", () => {
const result = await getSystemSettings();
// 降级读取成功（未抛错），缺失列由 transformer 落默认值。
- expect(selectMock).toHaveBeenCalledTimes(3);
+ expect(selectMock).toHaveBeenCalledTimes(4);
expect(result.siteTitle).toBe("CC Hub");
expect(result.enableHttp2).toBe(true);
expect(result.affinityIgnoreClientSessionId).toBe(true);
expect(result.streamGateMode).toBe("enforce");
- const thirdSelection = selectMock.mock.calls[2]?.[0] as Record;
- expect(thirdSelection).not.toHaveProperty("replayCacheTtlMinutes");
- expect(thirdSelection).not.toHaveProperty("cacheEffectivenessEnabled");
- expect(thirdSelection).toHaveProperty("replayEnabled");
- expect(thirdSelection).toHaveProperty("affinityIgnoreClientSessionId");
- expect(thirdSelection).toHaveProperty("streamGateMode");
- expect(thirdSelection).toHaveProperty("stickyTimeoutCooldownMs");
- expect(thirdSelection).toHaveProperty("racingTotalTimeoutMs");
- expect(thirdSelection).toHaveProperty("enableGeminiFunctionIdRectifier");
- expect(thirdSelection).toHaveProperty("enableThinkingEffortConflictRectifier");
+ const fourthSelection = selectMock.mock.calls[3]?.[0] as Record;
+ expect(fourthSelection).not.toHaveProperty("legacyHedgeMaxInFlight");
+ expect(fourthSelection).not.toHaveProperty("replayCacheTtlMinutes");
+ expect(fourthSelection).not.toHaveProperty("cacheEffectivenessEnabled");
+ expect(fourthSelection).toHaveProperty("replayEnabled");
+ expect(fourthSelection).toHaveProperty("affinityIgnoreClientSessionId");
+ expect(fourthSelection).toHaveProperty("streamGateMode");
+ expect(fourthSelection).toHaveProperty("stickyTimeoutCooldownMs");
+ expect(fourthSelection).toHaveProperty("racingTotalTimeoutMs");
+ expect(fourthSelection).toHaveProperty("enableGeminiFunctionIdRectifier");
+ expect(fourthSelection).toHaveProperty("enableThinkingEffortConflictRectifier");
vi.useRealTimers();
});
diff --git a/tests/unit/repository/usage-logs-min-retry-count-filter.test.ts b/tests/unit/repository/usage-logs-min-retry-count-filter.test.ts
index 149447521..e5d6d4f4b 100644
--- a/tests/unit/repository/usage-logs-min-retry-count-filter.test.ts
+++ b/tests/unit/repository/usage-logs-min-retry-count-filter.test.ts
@@ -58,6 +58,7 @@ describe("Usage logs minRetryCount filter", () => {
expect(whereSql).toContain("request_success");
expect(whereSql).toContain("retry_success");
expect(whereSql).toContain("retry_failed");
+ expect(whereSql).toContain("client_abort_no_first_byte");
expect(whereSql).toContain("statuscode");
expect(whereSql).toContain("hedge_triggered");
expect(whereSql).not.toContain("jsonb_array_length");
diff --git a/tests/unit/validation/system-settings-discovery.test.ts b/tests/unit/validation/system-settings-discovery.test.ts
index 3039b1fea..2b30ebe96 100644
--- a/tests/unit/validation/system-settings-discovery.test.ts
+++ b/tests/unit/validation/system-settings-discovery.test.ts
@@ -70,6 +70,26 @@ describe("UpdateSystemSettingsSchema Discovery settings", () => {
});
});
+ it.each([1, 2, 4])("accepts legacy hedge concurrency %s", (value) => {
+ expect(UpdateSystemSettingsSchema.parse({ legacyHedgeMaxInFlight: value })).toEqual({
+ legacyHedgeMaxInFlight: value,
+ });
+ });
+
+ it.each([0, 5, 1.5, null, true, [2]])("rejects invalid legacy hedge concurrency %s", (value) => {
+ expect(UpdateSystemSettingsSchema.safeParse({ legacyHedgeMaxInFlight: value }).success).toBe(
+ false
+ );
+ });
+
+ it("rejects invalid legacy hedge concurrency with a stable error code", () => {
+ const result = UpdateSystemSettingsSchema.safeParse({ legacyHedgeMaxInFlight: true });
+ expect(result.success).toBe(false);
+ if (!result.success) {
+ expect(result.error.issues[0]?.message).toBe("LEGACY_HEDGE_MAX_IN_FLIGHT_INVALID");
+ }
+ });
+
it("rejects inherited Object prototype names as Discovery fields", () => {
expect(isDiscoverySettingField("toString")).toBe(false);
expect(isDiscoverySettingField("valueOf")).toBe(false);