From 20fa6400b19dbf8118a85019e020982e1a7b2df1 Mon Sep 17 00:00:00 2001 From: Brisbanehuang Date: Wed, 22 Jul 2026 10:07:35 -0400 Subject: [PATCH 1/2] fix(discovery): preserve final rescue and bill ready losers --- messages/en/dashboard.json | 8 +- messages/en/settings/config.json | 6 +- messages/ja/dashboard.json | 8 +- messages/ja/settings/config.json | 6 +- messages/ru/dashboard.json | 8 +- messages/ru/settings/config.json | 6 +- messages/zh-CN/dashboard.json | 8 +- messages/zh-CN/settings/config.json | 6 +- messages/zh-TW/dashboard.json | 8 +- messages/zh-TW/settings/config.json | 6 +- .../_components/error-details-dialog.test.tsx | 298 ++++++++++++++++++ .../components/DiscoveryTraceView.tsx | 174 +++++++++- .../components/LogicTraceTab.tsx | 13 +- .../components/SummaryTab.tsx | 25 +- .../v1/_lib/proxy/discovery-coordinator.ts | 83 ++++- src/app/v1/_lib/proxy/forwarder.ts | 249 ++++++++++++++- src/app/v1/_lib/proxy/response-handler.ts | 16 +- .../integration/proxy-hedge-lifecycle.test.ts | 140 ++++++++ .../unit/proxy/discovery-coordinator.test.ts | 98 +++++- .../proxy-forwarder-hedge-first-byte.test.ts | 176 +++++++++++ ...sponse-handler-exported-finalizers.test.ts | 38 ++- 21 files changed, 1327 insertions(+), 53 deletions(-) diff --git a/messages/en/dashboard.json b/messages/en/dashboard.json index cad025e41..63193a051 100644 --- a/messages/en/dashboard.json +++ b/messages/en/dashboard.json @@ -299,6 +299,7 @@ "title": "Billing Details", "input": "Input", "output": "Output", + "cacheWrite": "Cache Write", "cacheWrite5m": "Cache Write (5m)", "cacheWrite1h": "Cache Write (1h)", "cacheRead": "Cache Read", @@ -496,7 +497,12 @@ "endpoint": "Endpoint", "error": "Upstream error", "cancellation": "Cancellation reason", - "timeline": "Attempt timeline" + "timeline": "Attempt timeline", + "billing": "Cost and usage", + "billed": "Included in total cost", + "billedCost": "Billed cost", + "winnerCost": "Winner cost", + "notObtained": "Cancelled; usage unavailable" }, "cancellationKinds": { "discovery_loser": "Another Discovery attempt won", diff --git a/messages/en/settings/config.json b/messages/en/settings/config.json index 50a311026..e5654556d 100644 --- a/messages/en/settings/config.json +++ b/messages/en/settings/config.json @@ -116,8 +116,8 @@ "billNonSuccessfulRequestsDesc": "When enabled, requests with non-success status (e.g., 499 client cancellation) are billed by token usage if upstream returned positive usage data. Default off.", "billNonSuccessfulRequestsTooltip": "Useful when an upstream provider counts tokens regardless of the final status (e.g., aborted streaming responses). Fake-200 upstream errors remain unbilled.", "billHedgeLosers": "Bill Provider-Racing Losers by Token Usage", - "billHedgeLosersDesc": "When provider racing (streaming hedge) is on, losing providers are kept connected in the background, drained for their token usage, and billed - their cost is added into this request's total. Default on.", - "billHedgeLosersTooltip": "Upstreams usually bill a request even after we cancel it. Keeping racing losers alive lets us reclaim their token counts so CCH's cost matches what every upstream actually charged. Each loser's cost is accumulated asynchronously into the request total.", + "billHedgeLosersDesc": "When provider racing (streaming hedge or bounded Discovery) is on, losers that have already produced a valid response prefix may be kept connected in the background, drained for usage, and billed. Their cost is added to this request's total. Other losers are cancelled. Default on.", + "billHedgeLosersTooltip": "Only losers with a readable valid response are drained. SLA timeouts, explicit failures, and attempts without a valid prefix are cancelled; their unknown upstream cost is not added.", "discoveryEnabled": "Enable bounded provider Discovery", "discoveryEnabledDesc": "When enabled, cold-start streaming requests probe multiple providers within a bounded window and keep at most one fallback. It is disabled by default.", "discoveryConcurrency": "Discovery initial concurrency", @@ -128,7 +128,7 @@ "stickyTimeoutCooldownMs": "Sticky timeout cooldown (milliseconds)", "stickyBindingTtl": "Sticky binding validity", "stickyBindingTtlDesc": "Read-only. This follows the SESSION_TTL environment variable and is shared with other session snapshots.", - "discoveryWindowDesc": "The total timeout must be at least Sticky SLA + maximum rounds × Discovery SLA. Discovery losers are cancelled and are not drained or billed by the legacy Hedge path.", + "discoveryWindowDesc": "The total timeout must be at least Sticky SLA + maximum rounds × Discovery SLA. With loser billing enabled, only ready Discovery losers may be drained; SLA timeouts and failed or no-prefix attempts remain cancelled.", "discoveryWindowInvalid": "Discovery total timeout is shorter than the configured Sticky and Discovery windows.", "discoverySettingsInvalid": "One or more Discovery values are outside the allowed range.", "verboseProviderError": "Verbose Provider Error", diff --git a/messages/ja/dashboard.json b/messages/ja/dashboard.json index 1c63760a8..2d06c01cf 100644 --- a/messages/ja/dashboard.json +++ b/messages/ja/dashboard.json @@ -299,6 +299,7 @@ "title": "課金詳細", "input": "入力", "output": "出力", + "cacheWrite": "キャッシュ書き込み", "cacheWrite5m": "キャッシュ書き込み (5m)", "cacheWrite1h": "キャッシュ書き込み (1h)", "cacheRead": "キャッシュ読み取り", @@ -496,7 +497,12 @@ "endpoint": "エンドポイント", "error": "上流エラー", "cancellation": "キャンセル理由", - "timeline": "試行タイムライン" + "timeline": "試行タイムライン", + "billing": "費用と使用量", + "billed": "合計費用に計上済み", + "billedCost": "計上費用", + "winnerCost": "勝者の費用", + "notObtained": "キャンセル済み、使用量未取得" }, "cancellationKinds": { "discovery_loser": "別の Discovery 試行が勝利しました", diff --git a/messages/ja/settings/config.json b/messages/ja/settings/config.json index 85a3b9e5b..f453b8519 100644 --- a/messages/ja/settings/config.json +++ b/messages/ja/settings/config.json @@ -118,8 +118,8 @@ "billNonSuccessfulRequestsDesc": "有効にすると、非 2xx ステータス（例: クライアント中断による 499）のリクエストでも、上流が正の token 使用量を返した場合は使用量に応じて課金されます。既定はオフ。", "billNonSuccessfulRequestsTooltip": "上流プロバイダーがレスポンス失敗時にも token をカウントするケース（ストリーム中断でも token を計上する等）に有用です。fake-200 の偽成功エラー応答は引き続き課金されません。", "billHedgeLosers": "プロバイダー競争（hedge）の敗者を Token 使用量で課金", - "billHedgeLosersDesc": "プロバイダー競争（ストリーミング hedge）が有効な場合、競争に敗れたプロバイダーを即座に切断せず、バックグラウンドで接続を維持して token 使用量を取得し課金します。その費用はこのリクエストの合計に加算されます。既定はオン。", - "billHedgeLosersTooltip": "上流はこちらが能動的にキャンセルしたリクエストも通常は課金します。競争の敗者を生かしておくことで token 数を回収し、CCH の課金を各上流の実際の課金と一致させます。各敗者の費用は非同期にリクエストの費用へ加算されます。", + "billHedgeLosersDesc": "プロバイダー競争（ストリーミング hedge または制限付き Discovery）が有効な場合、有効な応答プレフィックスを受信済みの敗者だけをバックグラウンドで読み切って課金します。費用はリクエスト合計に加算され、それ以外の敗者はキャンセルされます。既定はオン。", + "billHedgeLosersTooltip": "読み取り可能な有効応答がある敗者だけを読み切ります。SLA タイムアウト、明示的な失敗、有効なプレフィックスがない試行はキャンセルされ、確認できない上流費用は加算されません。", "discoveryEnabled": "制限付き Provider Discovery を有効化", "discoveryEnabledDesc": "有効にすると、コールドスタートのストリーミングリクエストで複数 Provider を制限時間内に探索し、フォールバックを最大 1 つ保持します。既定はオフです。", "discoveryConcurrency": "Discovery 初期並列数", @@ -130,7 +130,7 @@ "stickyTimeoutCooldownMs": "Sticky タイムアウト後のクールダウン（ミリ秒）", "stickyBindingTtl": "Sticky バインドの有効期間", "stickyBindingTtlDesc": "読み取り専用です。この値は環境変数 SESSION_TTL に従い、他の Session スナップショットと共有されます。", - "discoveryWindowDesc": "合計タイムアウトは Sticky SLA + 最大ラウンド数 × Discovery SLA 以上にしてください。", + "discoveryWindowDesc": "合計タイムアウトは Sticky SLA + 最大ラウンド数 × Discovery SLA 以上にしてください。敗者課金を有効にした場合も、有効な応答を受信済みの Discovery 敗者だけを読み切り、SLA タイムアウトや失敗はキャンセルします。", "discoveryWindowInvalid": "Discovery 合計タイムアウトが設定された Sticky/Discovery ウィンドウより短くなっています。", "discoverySettingsInvalid": "1 つ以上の Discovery 設定値が許容範囲外です。", "verboseProviderError": "詳細なプロバイダーエラー", diff --git a/messages/ru/dashboard.json b/messages/ru/dashboard.json index d22020ede..861729fae 100644 --- a/messages/ru/dashboard.json +++ b/messages/ru/dashboard.json @@ -299,6 +299,7 @@ "title": "Детали биллинга", "input": "Входные", "output": "Выходные", + "cacheWrite": "Запись в кэш", "cacheWrite5m": "Запись кэша (5m)", "cacheWrite1h": "Запись кэша (1h)", "cacheRead": "Чтение кэша", @@ -496,7 +497,12 @@ "endpoint": "Upstream", "error": "Ошибка upstream", "cancellation": "Причина отмены", - "timeline": "Хронология попытки" + "timeline": "Хронология попытки", + "billing": "Стоимость и использование", + "billed": "Включено в общую стоимость", + "billedCost": "Начисленная стоимость", + "winnerCost": "Стоимость победителя", + "notObtained": "Отменено, usage не получен" }, "cancellationKinds": { "discovery_loser": "Победила другая попытка Discovery", diff --git a/messages/ru/settings/config.json b/messages/ru/settings/config.json index 31aed5319..490614560 100644 --- a/messages/ru/settings/config.json +++ b/messages/ru/settings/config.json @@ -118,8 +118,8 @@ "billNonSuccessfulRequestsDesc": "Если включено, запросы с не-2xx статусом (например, 499 при отмене клиентом) будут тарифицироваться по фактическому количеству токенов, если апстрим вернул положительные данные об использовании. По умолчанию выключено.", "billNonSuccessfulRequestsTooltip": "Полезно, когда апстрим считает токены даже при неудачном статусе (например, прерванные стриминговые ответы). Поддельные ответы 200 (fake-200) по-прежнему не тарифицируются.", "billHedgeLosers": "Тарифицировать проигравших в гонке провайдеров по токенам", - "billHedgeLosersDesc": "Когда включена гонка провайдеров (streaming hedge), проигравшие провайдеры не отключаются сразу, а остаются на связи в фоне, дочитываются для получения использования токенов и тарифицируются - их стоимость добавляется в общую стоимость этого запроса. По умолчанию включено.", - "billHedgeLosersTooltip": "Апстримы обычно тарифицируют запрос, даже если мы его отменили. Сохраняя проигравших в гонке, мы получаем их счётчики токенов, чтобы расходы CCH совпадали с тем, что фактически списал каждый апстрим. Стоимость каждого проигравшего асинхронно добавляется к стоимости запроса.", + "billHedgeLosersDesc": "Если включена гонка провайдеров (streaming hedge или ограниченный Discovery), в фоне дочитываются и тарифицируются только проигравшие, уже отдавшие корректный префикс ответа. Их стоимость добавляется к общей стоимости запроса. Остальные проигравшие отменяются. По умолчанию включено.", + "billHedgeLosersTooltip": "Дочитываются только проигравшие с доступным корректным ответом. Тайм-ауты SLA, явные ошибки и попытки без корректного префикса отменяются; неизвестная стоимость апстрима не добавляется.", "discoveryEnabled": "Включить ограниченное обнаружение провайдеров", "discoveryEnabledDesc": "При включении потоковые запросы холодного старта проверяют несколько провайдеров в ограниченном окне и сохраняют не более одного резервного. По умолчанию выключено.", "discoveryConcurrency": "Начальная параллельность Discovery", @@ -130,7 +130,7 @@ "stickyTimeoutCooldownMs": "Пауза после тайм-аута Sticky (миллисекунды)", "stickyBindingTtl": "Срок действия привязки Sticky", "stickyBindingTtlDesc": "Только для чтения. Значение следует переменной окружения SESSION_TTL и используется другими снимками Session.", - "discoveryWindowDesc": "Общий тайм-аут должен быть не меньше SLA Sticky + максимальное число раундов × SLA Discovery.", + "discoveryWindowDesc": "Общий тайм-аут должен быть не меньше SLA Sticky + максимальное число раундов × SLA Discovery. При включённой тарификации проигравших дочитываются только готовые проигравшие Discovery; тайм-ауты SLA и ошибки остаются отменёнными.", "discoveryWindowInvalid": "Общий тайм-аут Discovery меньше настроенного окна Sticky и Discovery.", "discoverySettingsInvalid": "Одно или несколько значений Discovery находятся вне допустимого диапазона.", "verboseProviderError": "Подробные ошибки провайдеров", diff --git a/messages/zh-CN/dashboard.json b/messages/zh-CN/dashboard.json index 143047609..66746c250 100644 --- a/messages/zh-CN/dashboard.json +++ b/messages/zh-CN/dashboard.json @@ -299,6 +299,7 @@ "title": "计费详情", "input": "输入", "output": "输出", + "cacheWrite": "缓存写入", "cacheWrite5m": "缓存写入 (5m)", "cacheWrite1h": "缓存写入 (1h)", "cacheRead": "缓存读取", @@ -496,7 +497,12 @@ "endpoint": "端点", "error": "上游错误", "cancellation": "取消原因", - "timeline": "尝试时间线" + "timeline": "尝试时间线", + "billing": "费用与用量", + "billed": "已计入总费用", + "billedCost": "已计入费用", + "winnerCost": "赢家费用", + "notObtained": "已取消，未取得用量" }, "cancellationKinds": { "discovery_loser": "其他 Discovery 尝试已胜出", diff --git a/messages/zh-CN/settings/config.json b/messages/zh-CN/settings/config.json index acc8fc05d..f6ef4054a 100644 --- a/messages/zh-CN/settings/config.json +++ b/messages/zh-CN/settings/config.json @@ -45,8 +45,8 @@ "billNonSuccessfulRequestsDesc": "开启后，对于响应非 2xx 状态码（例如客户端中断的 499）的请求，只要上游返回了正向 token 用量，仍按用量计费。默认关闭。", "billNonSuccessfulRequestsTooltip": "适用于上游供应商即使响应失败也按 token 计费的场景（如流式中断时仍计入 token）。fake-200 假成功错误响应仍不会计费。", "billHedgeLosers": "对供应商竞速输家计费", - "billHedgeLosersDesc": "开启供应商竞速后，竞速落败的供应商不再被直接掐断，而是在后台保持连接、拿回其 token 用量并计费，其费用会累加进本条请求的总花费。默认开启。", - "billHedgeLosersTooltip": "上游通常即使请求被我们主动取消也照样计费。保活竞速输家可以拿回它们的 token 计数，使 CCH 的扣费与每个上游实际扣费保持一致。每个输家的费用都会异步累加到该请求的花费中。", + "billHedgeLosersDesc": "开启供应商竞速（流式 Hedge 或有界 Discovery）后，只有已经收到有效响应前缀的输家会在后台继续读取并按用量计费，费用会累加到本条请求总花费；其他输家会取消。默认开启。", + "billHedgeLosersTooltip": "只保活已经收到可读取有效响应的输家；SLA 超时、明确失败或未取得有效首字的请求会取消，无法确认的上游费用不会计入。", "discoveryEnabled": "启用有界供应商 Discovery", "discoveryEnabledDesc": "启用后，冷启动流式请求会在限定窗口内探测多个供应商，并且最多保留一个保底请求。默认关闭。", "discoveryConcurrency": "Discovery 首轮并发数", @@ -57,7 +57,7 @@ "stickyTimeoutCooldownMs": "Sticky 超时冷却（毫秒）", "stickyBindingTtl": "Sticky 绑定有效期", "stickyBindingTtlDesc": "只读；该值跟随环境变量 SESSION_TTL，并与其他 Session 快照共用。", - "discoveryWindowDesc": "总超时必须不小于 Sticky SLA + 最大轮数 × Discovery SLA。Discovery 输家会取消，不走旧 Hedge 的 drain 或输家计费。", + "discoveryWindowDesc": "总超时必须不小于 Sticky SLA + 最大轮数 × Discovery SLA。开启输家计费后，只有已取得有效首字的 Discovery 输家会后台读取；SLA 超时、失败或未取得首字的请求仍会取消。", "discoveryWindowInvalid": "Discovery 总超时短于已配置的 Sticky 与 Discovery 窗口。", "discoverySettingsInvalid": "一个或多个 Discovery 配置值超出允许范围。", "verboseProviderError": "详细供应商错误信息", diff --git a/messages/zh-TW/dashboard.json b/messages/zh-TW/dashboard.json index 39285b38c..56a19eec9 100644 --- a/messages/zh-TW/dashboard.json +++ b/messages/zh-TW/dashboard.json @@ -299,6 +299,7 @@ "title": "計費詳情", "input": "輸入", "output": "輸出", + "cacheWrite": "快取寫入", "cacheWrite5m": "快取寫入（5m）", "cacheWrite1h": "快取寫入（1h）", "cacheRead": "快取讀取", @@ -496,7 +497,12 @@ "endpoint": "端點", "error": "上游錯誤", "cancellation": "取消緣由", - "timeline": "嘗試時間線" + "timeline": "嘗試時間線", + "billing": "費用與用量", + "billed": "已計入總費用", + "billedCost": "已計入費用", + "winnerCost": "贏家費用", + "notObtained": "已中止，未取得用量" }, "cancellationKinds": { "discovery_loser": "其他 Discovery 嘗試已勝出", diff --git a/messages/zh-TW/settings/config.json b/messages/zh-TW/settings/config.json index 7a166f96c..308df4e58 100644 --- a/messages/zh-TW/settings/config.json +++ b/messages/zh-TW/settings/config.json @@ -118,8 +118,8 @@ "billNonSuccessfulRequestsDesc": "開啟後，對於回應非 2xx 狀態碼（例如客戶端中斷的 499）的請求，只要上游回報了正向 token 用量，仍會按用量計費。預設關閉。", "billNonSuccessfulRequestsTooltip": "適用於上游供應商即使回應失敗也按 token 計費的情境（例如串流中斷時仍記入 token）。fake-200 偽成功錯誤響應仍不會計費。", "billHedgeLosers": "對供應商競速輸家計費", - "billHedgeLosersDesc": "開啟供應商競速後，競速落敗的供應商不再被直接掐斷，而是在後台保持連線、取回其 token 用量並計費，其費用會累加進本條請求的總花費。預設開啟。", - "billHedgeLosersTooltip": "上游通常即使請求被我們主動取消也照常計費。保活競速輸家可以取回它們的 token 計數，使 CCH 的扣費與每個上游實際扣費保持一致。每個輸家的費用都會非同步累加到該請求的花費中。", + "billHedgeLosersDesc": "開啟供應商競速（串流 Hedge 或有界 Discovery）後，只有已收到有效回應前綴的輸家會在後台繼續讀取並按用量計費，費用會累加到本條請求總花費；其他輸家會取消。預設開啟。", + "billHedgeLosersTooltip": "只保活已收到可讀取有效回應的輸家；SLA 逾時、明確失敗或未取得有效首字的請求會取消，無法確認的上游費用不會計入。", "discoveryEnabled": "啟用有界供應商 Discovery", "discoveryEnabledDesc": "啟用後，冷啟動串流請求會在限定視窗內探測多個供應商，並且最多保留一個保底請求。預設關閉。", "discoveryConcurrency": "Discovery 首輪並發數", @@ -130,7 +130,7 @@ "stickyTimeoutCooldownMs": "Sticky 逾時冷卻（毫秒）", "stickyBindingTtl": "Sticky 綁定有效期", "stickyBindingTtlDesc": "唯讀；此值跟隨環境變數 SESSION_TTL，並與其他 Session 快照共用。", - "discoveryWindowDesc": "總逾時必須不小於 Sticky SLA + 最大輪數 × Discovery SLA。Discovery 輸家會取消，不走舊 Hedge 的 drain 或輸家計費。", + "discoveryWindowDesc": "總逾時必須不小於 Sticky SLA + 最大輪數 × Discovery SLA。開啟輸家計費後，只有已取得有效首字的 Discovery 輸家會在後台讀取；SLA 逾時、失敗或未取得首字的請求仍會取消。", "discoveryWindowInvalid": "Discovery 總逾時短於已設定的 Sticky 與 Discovery 視窗。", "discoverySettingsInvalid": "一個或多個 Discovery 設定值超出允許範圍。", "verboseProviderError": "詳細供應商錯誤資訊", diff --git a/src/app/[locale]/dashboard/logs/_components/error-details-dialog.test.tsx b/src/app/[locale]/dashboard/logs/_components/error-details-dialog.test.tsx index b43cc1b45..5ec3fc94e 100644 --- a/src/app/[locale]/dashboard/logs/_components/error-details-dialog.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/error-details-dialog.test.tsx @@ -346,6 +346,7 @@ const messages = { circuitOpen: "Circuit open", }, billingDetails: { + ...dashboardMessages.logs.details.billingDetails, title: "Billing details", input: "Input", output: "Output", @@ -1651,6 +1652,303 @@ describe("error-details-dialog routing trace", () => { unmount(); }); + test("maps billed Discovery losers to attempt cards without treating cancelled usage as zero", () => { + const billingTrace: RoutingTraceV1 = { + version: 1, + mode: "discovery", + startedAt: 1_000, + updatedAt: 8_000, + discoveryEnabled: true, + eligible: true, + events: [ + { + type: "attempt_started", + at: 1_000, + elapsedMs: 0, + round: 1, + attemptId: "101:1", + attemptKind: "normal", + provider: { id: 101, name: "billed-loser", priority: 1 }, + }, + { + type: "attempt_ready", + at: 2_000, + elapsedMs: 1_000, + round: 1, + attemptId: "101:1", + attemptKind: "normal", + provider: { id: 101, name: "billed-loser", priority: 1 }, + }, + { + type: "attempt_finished", + at: 3_000, + elapsedMs: 2_000, + round: 1, + attemptId: "101:1", + attemptKind: "normal", + provider: { id: 101, name: "billed-loser", priority: 1 }, + outcome: "cancelled", + cancellationKind: "winner_committed", + }, + { + type: "attempt_started", + at: 1_100, + elapsedMs: 100, + round: 1, + attemptId: "102:2", + attemptKind: "normal", + provider: { id: 102, name: "winner-provider", priority: 1 }, + }, + { + type: "winner_committed", + at: 2_500, + elapsedMs: 1_500, + round: 1, + attemptId: "102:2", + attemptKind: "normal", + provider: { id: 102, name: "winner-provider", priority: 1 }, + statusCode: 200, + }, + { + type: "attempt_started", + at: 1_200, + elapsedMs: 200, + round: 1, + attemptId: "103:3", + attemptKind: "normal", + provider: { id: 103, name: "cancelled-without-usage", priority: 2 }, + }, + { + type: "attempt_finished", + at: 3_100, + elapsedMs: 2_100, + round: 1, + attemptId: "103:3", + attemptKind: "normal", + provider: { id: 103, name: "cancelled-without-usage", priority: 2 }, + outcome: "cancelled", + cancellationKind: "discovery_sla_timeout", + }, + { + type: "attempt_started", + at: 1_300, + elapsedMs: 300, + round: 1, + attemptId: "101:4", + attemptKind: "normal", + provider: { id: 101, name: "billed-loser", priority: 1 }, + }, + { + type: "attempt_finished", + at: 3_200, + elapsedMs: 2_200, + round: 1, + attemptId: "101:4", + attemptKind: "normal", + provider: { id: 101, name: "billed-loser", priority: 1 }, + outcome: "cancelled", + cancellationKind: "discovery_sla_timeout", + }, + ], + summary: { + outcome: "success", + statusCode: 200, + durationMs: 7_000, + ttfbMs: 1_500, + attemptsPerRequest: 4, + maxActiveAttempts: 4, + rounds: 1, + providerMs: 5_100, + fallbackPromotions: 0, + cancelFailures: 0, + winnerOrigin: "normal", + winnerProviderId: 102, + winnerRound: 1, + }, + }; + const { container, unmount } = renderClientWithIntl( + + ); + + const cards = Array.from( + container.querySelectorAll("[data-testid='discovery-attempt']") + ); + const repeatedProviderCards = cards.filter((card) => + card.textContent?.includes("billed-loser") + ); + const billedLoser = repeatedProviderCards[0]; + const sameProviderWithoutUsage = repeatedProviderCards[1]; + const winner = cards.find((card) => card.textContent?.includes("winner-provider")); + const cancelled = cards.find((card) => card.textContent?.includes("cancelled-without-usage")); + click(billedLoser?.querySelector("[data-testid='discovery-attempt-toggle']") ?? null); + click( + sameProviderWithoutUsage?.querySelector("[data-testid='discovery-attempt-toggle']") ?? null + ); + click(winner?.querySelector("[data-testid='discovery-attempt-toggle']") ?? null); + click(cancelled?.querySelector("[data-testid='discovery-attempt-toggle']") ?? null); + + expect(billedLoser?.textContent).toContain("Included in total cost"); + expect(billedLoser?.textContent).toContain("$0.006000"); + expect(billedLoser?.textContent).toContain("Input 50"); + expect(billedLoser?.textContent).toContain("Output 12"); + expect(billedLoser?.textContent).toContain("Cache Write 10"); + expect(sameProviderWithoutUsage?.textContent).toContain("Cancelled; usage unavailable"); + expect(sameProviderWithoutUsage?.textContent).not.toContain("$0.006000"); + expect(winner?.textContent).toContain("Winner cost"); + expect(winner?.textContent).toContain("$0.009000"); + expect(winner?.textContent).toContain("Input 100"); + expect(winner?.textContent).toContain("Output 30"); + expect(winner?.textContent).toContain("Cache Write 20"); + expect(cancelled?.textContent).toContain("Cancelled; usage unavailable"); + expect(cancelled?.textContent).not.toContain("$0.000000"); + + const summaryTable = container.querySelector( + "[data-testid='provider-racing-billing-table']" + ); + expect(summaryTable?.textContent).toContain("winner-provider"); + expect(summaryTable?.textContent).toContain("#2"); + expect(summaryTable?.textContent).toContain("$0.009000"); + expect(summaryTable?.textContent).toContain("billed-loser"); + expect(summaryTable?.textContent).toContain("#1"); + expect(summaryTable?.textContent).toContain("$0.006000"); + expect(summaryTable?.textContent).toContain("$0.015000"); + unmount(); + }); + + test("does not attach an older loser cost to a later winner from the same provider", () => { + const trace: RoutingTraceV1 = { + version: 1, + mode: "discovery", + startedAt: 1_000, + updatedAt: 3_000, + discoveryEnabled: true, + eligible: true, + events: [ + { + type: "attempt_started", + at: 1_000, + elapsedMs: 0, + round: 1, + attemptId: "101:4", + attemptKind: "normal", + provider: { id: 101, name: "same-provider-winner", priority: 1 }, + }, + { + type: "attempt_ready", + at: 2_000, + elapsedMs: 1_000, + round: 1, + attemptId: "101:4", + attemptKind: "normal", + provider: { id: 101, name: "same-provider-winner", priority: 1 }, + }, + { + type: "winner_committed", + at: 2_100, + elapsedMs: 1_100, + round: 1, + attemptId: "101:4", + attemptKind: "normal", + provider: { id: 101, name: "same-provider-winner", priority: 1 }, + outcome: "winner", + statusCode: 200, + }, + ], + summary: { + outcome: "success", + statusCode: 200, + durationMs: 2_000, + ttfbMs: 1_100, + attemptsPerRequest: 2, + maxActiveAttempts: 2, + rounds: 1, + providerMs: 3_000, + fallbackPromotions: 0, + cancelFailures: 0, + winnerOrigin: "normal", + winnerProviderId: 101, + winnerRound: 1, + }, + }; + const { container, unmount } = renderClientWithIntl( + + ); + + const winner = container.querySelector("[data-testid='discovery-attempt']"); + click(winner?.querySelector("[data-testid='discovery-attempt-toggle']") ?? null); + expect(winner?.textContent).toContain("Winner cost"); + expect(winner?.textContent).toContain("$0.009000"); + expect(winner?.textContent).not.toContain("$0.006000"); + unmount(); + }); + test("labels a lease conflict as single-route protection while retaining provider selection", () => { const protectedTrace: RoutingTraceV1 = { version: 1, diff --git a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/DiscoveryTraceView.tsx b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/DiscoveryTraceView.tsx index 0905540b1..48fd063bf 100644 --- a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/DiscoveryTraceView.tsx +++ b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/DiscoveryTraceView.tsx @@ -16,9 +16,12 @@ import { import { useTranslations } from "next-intl"; import { useState } from "react"; import { Badge } from "@/components/ui/badge"; -import { cn } from "@/lib/utils"; +import { cn, formatTokenAmount } from "@/lib/utils"; +import { formatCurrency } from "@/lib/utils/currency"; +import { summarizeHedgeBilling } from "@/lib/utils/hedge-billing"; import { redactJsonString } from "@/lib/utils/message-redaction"; import { sanitizeErrorTextForDetail } from "@/lib/utils/upstream-error-detection"; +import type { HedgeLoserBilling } from "@/types/cost-breakdown"; import type { ProviderChainItem } from "@/types/message"; import type { RoutingTraceV1 } from "@/types/routing-trace"; @@ -80,6 +83,9 @@ type AttemptView = { fallbackPromoted: boolean; winnerCommitted: boolean; chainItem: ProviderChainItem | null; + billingEntry: HedgeLoserBilling | null; + billingStatus: "none" | "billed" | "not_obtained"; + winnerCostUsd: string | null; history: Array<{ type: string; elapsedMs: number | null; @@ -89,6 +95,38 @@ type AttemptView = { }>; }; +function findDiscoveryBillingEntry( + hedgeLosers: HedgeLoserBilling[] | null | undefined, + attempt: Pick +): HedgeLoserBilling | null { + if (!hedgeLosers || hedgeLosers.length === 0 || attempt.providerId == null) return null; + const candidates = hedgeLosers.filter((entry) => entry.providerId === attempt.providerId); + if (candidates.length === 0) return null; + + const attemptNumbers = [attempt.sequence ?? attempt.chainItem?.attemptNumber].filter( + (number): number is number => number != null && Number.isFinite(number) + ); + for (const attemptNumber of attemptNumbers) { + const exact = candidates.find((entry) => entry.attemptNumber === attemptNumber); + if (exact) return exact; + } + return null; +} + +function inferMissingBillingStatus(attempt: AttemptView): AttemptView["billingStatus"] { + if (attempt.winnerCommitted || attempt.outcome === "winner") return "none"; + if ( + attempt.outcome === "cancelled" || + attempt.outcome === "timeout" || + attempt.outcome === "client_abort" || + attempt.outcome === "deadline" || + attempt.outcome === "failed" + ) { + return "not_obtained"; + } + return "none"; +} + function asRecord(value: unknown): TraceRecord { return value && typeof value === "object" && !Array.isArray(value) ? (value as TraceRecord) : {}; } @@ -101,6 +139,12 @@ function asNumber(value: unknown): number | null { return typeof value === "number" && Number.isFinite(value) ? value : null; } +function isDisplayableCost(value: unknown): value is string { + if (typeof value !== "string" || value.trim() === "") return false; + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0; +} + function parseAttemptSequence(attemptId: string): number | null { const match = /:(\d+)$/.exec(attemptId); return match ? Number(match[1]) : null; @@ -278,7 +322,12 @@ function applyEventOutcome( } } -function buildAttempts(trace: RoutingTraceV1, providerChain: ProviderChainItem[]): AttemptView[] { +function buildAttempts( + trace: RoutingTraceV1, + providerChain: ProviderChainItem[], + hedgeLosers: HedgeLoserBilling[] | null | undefined, + costUsd: string | null | undefined +): AttemptView[] { const attempts = new Map(); const lookup = buildProviderChainLookup(providerChain); @@ -314,6 +363,9 @@ function buildAttempts(trace: RoutingTraceV1, providerChain: ProviderChainItem[] fallbackPromoted: false, winnerCommitted: false, chainItem: findChainItem(attemptId, provider.id, lookup), + billingEntry: null, + billingStatus: "none", + winnerCostUsd: null, history: [], }; @@ -370,10 +422,30 @@ function buildAttempts(trace: RoutingTraceV1, providerChain: ProviderChainItem[] } } - return [...attempts.values()].sort((a, b) => { + const sortedAttempts = [...attempts.values()].sort((a, b) => { if (a.round !== b.round) return a.round - b.round; return (a.startedAt ?? Number.MAX_SAFE_INTEGER) - (b.startedAt ?? Number.MAX_SAFE_INTEGER); }); + + const hedgeSummary = summarizeHedgeBilling(costUsd, hedgeLosers); + for (const attempt of sortedAttempts) { + const billingEntry = findDiscoveryBillingEntry(hedgeLosers, attempt); + if (billingEntry) { + attempt.billingEntry = billingEntry; + attempt.billingStatus = "billed"; + continue; + } + + attempt.billingStatus = inferMissingBillingStatus(attempt); + if (attempt.winnerCommitted || attempt.outcome === "winner") { + // When billed losers exist, subtract them from the persisted request total + // so the winner card agrees with the existing hedge billing table. With no + // loser entry, costUsd is the only safe winner amount available. + attempt.winnerCostUsd = hedgeSummary?.winnerCost ?? costUsd ?? null; + } + } + + return sortedAttempts; } function numberFrom(record: TraceRecord, ...keys: string[]): number | null { @@ -508,12 +580,23 @@ export function RoutingModeBanner({ trace }: { trace: RoutingTraceV1 }) { export function DiscoveryTraceView({ trace, providerChain = [], + hedgeLosers, + costUsd, + winnerUsage, }: { trace: RoutingTraceV1; providerChain?: ProviderChainItem[]; + hedgeLosers?: HedgeLoserBilling[] | null; + costUsd?: string | null; + winnerUsage?: { + inputTokens?: number | null; + outputTokens?: number | null; + cacheCreationInputTokens?: number | null; + cacheReadInputTokens?: number | null; + }; }) { const t = useTranslations("dashboard.logs.details.routingTrace"); - const attempts = buildAttempts(trace, providerChain); + const attempts = buildAttempts(trace, providerChain, hedgeLosers, costUsd); const grouped = new Map(); for (const attempt of attempts) { const group = grouped.get(attempt.round) ?? []; @@ -682,7 +765,7 @@ export function DiscoveryTraceView({
{roundAttempts.map((attempt) => ( - + ))}
@@ -702,8 +785,20 @@ function TraceValue({ label, value }: { label: string; value: string | number | ); } -function AttemptCard({ attempt }: { attempt: AttemptView }) { +function AttemptCard({ + attempt, + winnerUsage, +}: { + attempt: AttemptView; + winnerUsage?: { + inputTokens?: number | null; + outputTokens?: number | null; + cacheCreationInputTokens?: number | null; + cacheReadInputTokens?: number | null; + }; +}) { const t = useTranslations("dashboard.logs.details.routingTrace"); + const tDetails = useTranslations("dashboard.logs.details"); const [expanded, setExpanded] = useState(false); const style = outcomeStyle(attempt.outcome); const Icon = style.icon; @@ -725,6 +820,72 @@ function AttemptCard({ attempt }: { attempt: AttemptView }) { : attempt.cancellationKind : null; + const billingEntry = attempt.billingEntry; + const billedCost = attempt.winnerCostUsd ?? billingEntry?.costUsd; + const tokenRows = billingEntry + ? [ + ["input", billingEntry.inputTokens], + ["output", billingEntry.outputTokens], + ["cacheWrite", billingEntry.cacheCreationInputTokens], + ["cacheRead", billingEntry.cacheReadInputTokens], + ].filter(([, value]) => typeof value === "number" && Number.isFinite(value)) + : []; + const winnerTokenRows = attempt.winnerCommitted + ? [ + ["input", winnerUsage?.inputTokens], + ["output", winnerUsage?.outputTokens], + ["cacheWrite", winnerUsage?.cacheCreationInputTokens], + ["cacheRead", winnerUsage?.cacheReadInputTokens], + ].filter(([, value]) => typeof value === "number" && Number.isFinite(value)) + : []; + const renderBillingDetails = () => { + const isWinner = attempt.winnerCommitted || attempt.outcome === "winner"; + const effectiveStatus = + isWinner && attempt.winnerCostUsd != null ? "billed" : attempt.billingStatus; + if (effectiveStatus === "none") return null; + const statusKey = effectiveStatus === "billed" ? "billed" : "notObtained"; + const statusClass = + effectiveStatus === "billed" + ? "border-emerald-200 bg-emerald-50 text-emerald-700 dark:border-emerald-800 dark:bg-emerald-950/20 dark:text-emerald-300" + : "border-slate-200 bg-slate-50 text-slate-600 dark:border-slate-700 dark:bg-slate-900/40 dark:text-slate-300"; + const rows = isWinner ? winnerTokenRows : tokenRows; + + return ( +
+
+ {t("attemptDetails.billing")} + + {t(`attemptDetails.${statusKey}`)} + +
+ {effectiveStatus === "billed" && isDisplayableCost(billedCost) && ( +
+ + {isWinner ? t("attemptDetails.winnerCost") : t("attemptDetails.billedCost")} + + {formatCurrency(billedCost, "USD", 6)} +
+ )} + {effectiveStatus === "billed" && rows.length > 0 && ( +
+ {rows.map(([kind, value]) => ( + + {tDetails(`billingDetails.${kind}`)} {formatTokenAmount(value as number)} + + ))} +
+ )} +
+ ); + }; + return (
{cancellationLabel} )} + {renderBillingDetails()} {attempt.history.length > 0 && (
{t("attemptDetails.timeline")}
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 22bc1263e..c03f50fe5 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 @@ -200,7 +200,18 @@ export function LogicTraceTab({ return (
- +
); } diff --git a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/SummaryTab.tsx b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/SummaryTab.tsx index f53948a10..e26ba88cc 100644 --- a/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/SummaryTab.tsx +++ b/src/app/[locale]/dashboard/logs/_components/error-details-dialog/components/SummaryTab.tsx @@ -66,6 +66,7 @@ export function SummaryTab({ costBreakdown, hedgeLosers, providerChain, + routingTrace, context1mApplied, durationMs, ttfbMs, @@ -627,10 +628,23 @@ export function SummaryTab({ {(() => { const hedgeWinnerStep = providerChain?.find((item) => item.reason === "hedge_winner") ?? null; + const discoveryWinnerEvent = + routingTrace?.mode === "discovery" + ? routingTrace.events.findLast( + (event) => + event.type === "winner_committed" && + (routingTrace.summary?.winnerProviderId == null || + event.provider?.id === routingTrace.summary.winnerProviderId) + ) + : null; + const discoveryWinner = discoveryWinnerEvent?.provider; + const discoveryWinnerAttempt = discoveryWinnerEvent?.attemptId?.match(/:(\d+)$/)?.[1]; const hedgeTable = buildHedgeBillingTable(costUsd, hedgeLosers, { - providerId: hedgeWinnerStep?.id ?? null, - providerName: hedgeWinnerStep?.name ?? null, - attemptNumber: hedgeWinnerStep?.attemptNumber ?? null, + providerId: hedgeWinnerStep?.id ?? discoveryWinner?.id ?? null, + providerName: hedgeWinnerStep?.name ?? discoveryWinner?.name ?? null, + attemptNumber: + hedgeWinnerStep?.attemptNumber ?? + (discoveryWinnerAttempt ? Number(discoveryWinnerAttempt) : null), inputTokens, outputTokens, cacheCreationInputTokens, @@ -649,7 +663,10 @@ export function SummaryTab({ {t("billingDetails.hedgeMergedCount", { count: hedgeTable.count })}
-
+
diff --git a/src/app/v1/_lib/proxy/discovery-coordinator.ts b/src/app/v1/_lib/proxy/discovery-coordinator.ts index 922bbf496..6f3a3bbbe 100644 --- a/src/app/v1/_lib/proxy/discovery-coordinator.ts +++ b/src/app/v1/_lib/proxy/discovery-coordinator.ts @@ -20,6 +20,11 @@ export type DiscoveryAttempt = { providerId: number; priority: number; kind: DiscoveryAttemptKind; + /** + * A final-round rescue is kept alive beside the primary fallback. It can + * rescue the request, but it must never establish or renew Sticky. + */ + finalRescue?: boolean; /** Selector/endpoint setup occupies a slot but cannot become a winner or fallback. */ setupOnly?: boolean; ready: boolean; @@ -31,7 +36,12 @@ export type DiscoveryAttempt = { export type DiscoveryAction = | { type: "commit_normal"; attemptId: string } | { type: "promote_fallback"; attemptId: string } - | { type: "cancel"; attemptIds: string[]; promoteAttemptId?: string } + | { + type: "cancel"; + attemptIds: string[]; + promoteAttemptId?: string; + rescueAttemptId?: string; + } | { type: "launch"; slots: number; @@ -59,6 +69,7 @@ export class DiscoveryCoordinator { private requestEpoch = 0; private roundEpoch = 0; private roundOpen = true; + private finalRescueAttemptId: string | null = null; constructor(options: DiscoveryCoordinatorOptions) { this.concurrency = Math.max(1, Math.floor(options.concurrency)); @@ -89,6 +100,7 @@ export class DiscoveryCoordinator { } this.round += 1; this.roundEpoch += 1; + this.finalRescueAttemptId = null; if (!this.isTerminal) { this.state = "DISCOVERY_RACING"; this.roundOpen = true; @@ -171,7 +183,9 @@ export class DiscoveryCoordinator { } get snapshot(): DiscoveryAttempt[] { - return Array.from(this.attempts.values()).map((attempt) => ({ ...attempt })); + return Array.from(this.attempts.values()).map((attempt) => ({ + ...attempt, + })); } /** Ignore events from a cancelled request or an old round. */ @@ -189,6 +203,15 @@ export class DiscoveryCoordinator { if (!attempt?.pending || attempt.setupOnly === true) return { type: "none" }; attempt.ready = true; if (attempt.kind === "fallback") { + // Once the final rescue lane is open, the primary fallback and the + // standby are peers. Whichever produces a valid prefix first may rescue + // the request; do not keep a ready fallback waiting for a pending + // standby that may never respond. + if (this.finalRescueAttemptId) { + attempt.pending = false; + this.state = "FALLBACK_ACTIVE"; + return { type: "promote_fallback", attemptId: attempt.id }; + } const pendingNormal = Array.from(this.attempts.values()).some( (candidate) => candidate.pending && candidate.kind === "normal" ); @@ -281,8 +304,9 @@ export class DiscoveryCoordinator { /** * Close the current SLA window. At a boundary a ready normal always wins; - * otherwise the best still-pending normal becomes the sole fallback. A - * fallback that is merely ready is held until no normal can still win. + * otherwise the best still-pending normal becomes the primary fallback. A + * non-final fallback is held until no normal can still win; the final window + * may retain one additional pending normal as a bounded rescue lane. */ onRoundBoundary(requestEpoch = this.requestEpoch, roundEpoch = this.roundEpoch): DiscoveryAction { if (!this.acceptsEpoch(requestEpoch, roundEpoch) || this.isTerminal) return { type: "none" }; @@ -311,6 +335,26 @@ export class DiscoveryCoordinator { ) .sort(compareAttempts); if (currentFallback && (pendingNormal.length > 0 || setupAttemptIds.length > 0)) { + // In the final round preserve one best normal alongside the primary + // fallback. This is a bounded rescue lane: at most two upstreams stay + // alive regardless of the configured discovery concurrency. + if (this.round >= this.maxRounds && this.concurrency >= 2 && pendingNormal.length > 0) { + const rescue = pendingNormal[0]; + rescue.finalRescue = true; + this.finalRescueAttemptId = rescue.id; + const cancelAttemptIds = [ + ...pendingNormal.slice(1).map((attempt) => attempt.id), + ...setupAttemptIds, + ]; + for (const id of cancelAttemptIds) this.attempts.get(id)!.pending = false; + this.state = "FALLBACK_READY_HELD"; + return { + type: "cancel", + attemptIds: cancelAttemptIds, + rescueAttemptId: rescue.id, + }; + } + const cancelAttemptIds = [...pendingNormal.map((attempt) => attempt.id), ...setupAttemptIds]; for (const attempt of pendingNormal) attempt.pending = false; if (this.round < this.maxRounds) { @@ -332,7 +376,10 @@ export class DiscoveryCoordinator { if (setupAttemptIds.length > 0) { if (this.round >= this.maxRounds) { this.state = "TERMINAL_FAILED"; - return { type: "terminal_failure", cancelAttemptIds: setupAttemptIds }; + return { + type: "terminal_failure", + cancelAttemptIds: setupAttemptIds, + }; } this.beginRound(); return { @@ -347,7 +394,19 @@ export class DiscoveryCoordinator { const fallback = pendingNormal[0]; fallback.kind = "fallback"; this.state = "FALLBACK_READY_HELD"; - const losers = [...pendingNormal.slice(1).map((attempt) => attempt.id), ...setupAttemptIds]; + // In the final round, retain one additional normal as a standby rescue. + // It remains a normal candidate for race purposes but is explicitly + // ineligible for Sticky binding. + const finalRescue = + this.round >= this.maxRounds && this.concurrency >= 2 ? pendingNormal[1] : undefined; + if (finalRescue) { + finalRescue.finalRescue = true; + this.finalRescueAttemptId = finalRescue.id; + } + const losers = [ + ...pendingNormal.slice(finalRescue ? 2 : 1).map((attempt) => attempt.id), + ...setupAttemptIds, + ]; for (const id of losers) this.attempts.get(id)!.pending = false; if (this.round < this.maxRounds) { @@ -360,7 +419,12 @@ export class DiscoveryCoordinator { promoteAttemptId: fallback.id, }; } - return { type: "cancel", attemptIds: losers, promoteAttemptId: fallback.id }; + return { + type: "cancel", + attemptIds: losers, + promoteAttemptId: fallback.id, + ...(finalRescue ? { rescueAttemptId: finalRescue.id } : {}), + }; } onDeadline(): DiscoveryAction { @@ -414,7 +478,10 @@ export class DiscoveryCoordinator { if (readyNormal.type === "commit_normal") return readyNormal; const fallback = pending.find((attempt) => attempt.kind === "fallback"); - if (fallback?.ready && pending.every((attempt) => attempt.kind === "fallback")) { + if ( + fallback?.ready && + (this.finalRescueAttemptId != null || pending.every((attempt) => attempt.kind === "fallback")) + ) { return this.commitWinner(fallback.id); } return { type: "none" }; diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 130774e3b..9be7963c8 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -104,7 +104,7 @@ import { validateOpenAIImageRequest, } from "./openai-image-compat"; import { ProxyProviderResolver } from "./provider-selector"; -import { finalizeHedgeLoserBilling } from "./response-handler"; +import { finalizeHedgeLoserBilling, hasStreamCompletionMarker } from "./response-handler"; import type { ProxySession } from "./session"; import { type DeferredStreamingHedgeBindingAuthority, @@ -5140,6 +5140,10 @@ export class ProxyForwarder { const racingDeadlineAt = requestStartedAt + totalTimeoutMs; const protocol = ProxyForwarder.discoveryProtocol(session); const rawCrossProviderFallbackEnabled = session.isRawCrossProviderFallbackEnabled(); + // 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 coordinator = new DiscoveryCoordinator({ concurrency, maxRounds }); const discoveryMetrics = new DiscoveryRequestMetrics( { @@ -5157,6 +5161,8 @@ export class ProxyForwarder { StreamingHedgeAttempt & { id: string; kind: "normal" | "fallback"; + /** Retained beside the primary fallback in the final round. */ + finalRescue: boolean; controller: AbortController; parser: DiscoveryValidityParser; chunks: Uint8Array[]; @@ -5386,6 +5392,148 @@ export class ProxyForwarder { return attempt.modelRedirect; }; + /** + * Drain a Discovery loser only after a winner has committed and only when + * the loser already produced a protocol-valid prefix. Discovery attempts + * that were still waiting for headers/first byte are cancelled normally; + * retaining those would turn an SLA race into an unbounded cost fan-out. + * + * The helper deliberately requires a natural drain completion before + * invoking finalizeHedgeLoserBilling. A cancellation, transport error, or + * drain cap therefore never creates a cost entry for Discovery. + */ + const startDiscoveryLoserBilling = (attempt: (typeof winner & { id: string }) | null) => { + if (!attempt || attempt.loserBillingStarted) return; + attempt.loserBillingStarted = true; + + const reader = attempt.reader; + const response = attempt.response; + const messageRequestId = session.messageContext?.id; + const messageRequestCreatedAtMs = session.messageContext?.createdAt.getTime(); + if (!reader || !response || messageRequestId == null) { + const cancelPromise = reader?.cancel("discovery_loser_no_billing"); + cancelPromise?.catch(() => undefined); + releaseProviderRef(attempt); + if (attempt.releaseAgent && !attempt.agentReleased) { + attempt.agentReleased = true; + try { + attempt.releaseAgent(); + } catch { + /* release is best effort */ + } + } + return; + } + + attempt.clearResponseTimeout?.(); + const controller = attempt.responseController; + const drainTimeoutMs = getEnvConfig().HEDGE_LOSER_DRAIN_TIMEOUT_MS; + const drainTimer = setTimeout(() => { + try { + controller?.abort(new Error("discovery_loser_drain_timeout")); + } catch { + /* abort is best effort */ + } + }, drainTimeoutMs); + + void (async () => { + const decoder = new TextDecoder(); + const chunks: string[] = []; + let totalBytes = 0; + let drainComplete = false; + const MAX_DRAIN_BYTES = 32 * 1024 * 1024; + + // The validity parser may have consumed one or more chunks before the + // loser was held. Replay those bytes so usage markers in the prefix + // (for example Anthropic message_start) are available to billing. + const bufferedChunks = attempt.chunks.splice(0); + for (const chunk of bufferedChunks) { + chunks.push(decoder.decode(chunk, { stream: true })); + totalBytes += chunk.byteLength; + } + + try { + while (true) { + const { value, done } = await reader.read(); + if (done) { + drainComplete = true; + break; + } + if (!value) continue; + chunks.push(decoder.decode(value, { stream: true })); + totalBytes += value.byteLength; + if (totalBytes > MAX_DRAIN_BYTES) { + logger.warn("[Discovery] Loser drain exceeded cap; skipping billing", { + sessionId: attempt.session.sessionId ?? null, + providerId: attempt.provider.id, + providerName: attempt.provider.name, + totalBytes, + }); + try { + controller?.abort(new Error("discovery_loser_drain_cap")); + } catch { + /* abort is best effort */ + } + break; + } + } + } catch (drainError) { + logger.debug("[Discovery] Loser drain ended before natural completion", { + sessionId: attempt.session.sessionId ?? null, + providerId: attempt.provider.id, + providerName: attempt.provider.name, + error: drainError instanceof Error ? drainError.message : String(drainError), + }); + } + + const flushed = decoder.decode(); + if (flushed) chunks.push(flushed); + const allContent = chunks.join(""); + + // Discovery billing is intentionally stricter than the legacy helper's + // partial-usage safety net: a cancelled/failed drain is not a billable + // loser. This avoids charging a provider whose response was cut off by + // winner handoff or the drain cap. + if ( + drainComplete && + hasStreamCompletionMarker(allContent, attempt.session.originalFormat) + ) { + await finalizeHedgeLoserBilling({ + messageRequestId, + messageRequestCreatedAtMs: messageRequestCreatedAtMs ?? Date.now(), + loserSession: attempt.session, + provider: attempt.provider, + attemptNumber: attempt.sequence, + upstreamStatusCode: response.status, + allContent, + drainComplete: true, + requireUsage: true, + billingContext: attempt.billingSnapshot ?? undefined, + }); + } + })() + .catch((billingError) => { + logger.debug("[Discovery] Loser billing task failed", { + sessionId: attempt.session.sessionId ?? null, + providerId: attempt.provider.id, + providerName: attempt.provider.name, + error: billingError instanceof Error ? billingError.message : String(billingError), + }); + }) + .finally(() => { + clearTimeout(drainTimer); + releaseProviderRef(attempt); + if (attempt.releaseAgent && !attempt.agentReleased) { + attempt.agentReleased = true; + try { + attempt.releaseAgent(); + } catch { + /* release is best effort */ + } + } + }); + }; + const cleanupAttempt = ( attempt: (typeof winner & { id: string }) | null, cancellationKind: DiscoveryCancellationKind | null, @@ -5393,11 +5541,21 @@ export class ProxyForwarder { ) => { if (attempt?.readerTransferred) return; if (!attempt) return; + // This flag is set only by cancelLosers when a winner has committed. It + // prevents terminal cleanup, round SLA cancellation and client aborts + // from accidentally entering the billing drain path. + const preserveForLoserBilling = + cancellationKind === "discovery_loser" && + attempt.billAsLoser && + attempt.ready && + attempt.response != null && + attempt.reader != null && + !attempt.readerCancelled; attempt.pending = false; if (cancellationKind && !attempt.cancellationKind) { attempt.cancellationKind = cancellationKind; } - if (!attempt.controller.signal.aborted) { + if (!preserveForLoserBilling && !attempt.controller.signal.aborted) { try { attempt.controller.abort( cancellationKind @@ -5408,7 +5566,7 @@ export class ProxyForwarder { /* abort is best effort */ } } - if (attempt.reader && !attempt.readerCancelled) { + if (!preserveForLoserBilling && attempt.reader && !attempt.readerCancelled) { attempt.readerCancelled = true; try { const cancelPromise = attempt.reader.cancel( @@ -5425,7 +5583,7 @@ export class ProxyForwarder { }); } } - if (attempt.releaseAgent && !attempt.agentReleased) { + if (!preserveForLoserBilling && attempt.releaseAgent && !attempt.agentReleased) { attempt.agentReleased = true; try { attempt.releaseAgent(); @@ -5433,8 +5591,10 @@ export class ProxyForwarder { /* release is idempotent */ } } - releaseProviderRef(attempt); - attempt.chunks.length = 0; + if (!preserveForLoserBilling) { + releaseProviderRef(attempt); + attempt.chunks.length = 0; + } discoveryMetrics.attemptFinished(attempt.id, { providerId: attempt.provider.id, outcome: cancellationKind ? "cancelled" : "failed", @@ -5462,6 +5622,9 @@ export class ProxyForwarder { ...(failure?.reason ? { reason: failure.reason } : {}), }); } + if (preserveForLoserBilling) { + startDiscoveryLoserBilling(attempt); + } }; const cancelAttempt = ( @@ -5474,7 +5637,23 @@ export class ProxyForwarder { cancellationKind: DiscoveryCancellationKind = "discovery_loser" ) => { for (const attempt of attempts.values()) { - if (attempt !== keep) cancelAttempt(attempt, cancellationKind); + if (attempt === keep) continue; + // Only a committed winner may authorize loser billing. In particular, + // terminal failure and deadline cleanup must not retain upstreams just + // because the global setting is enabled. + if ( + keep && + cancellationKind === "discovery_loser" && + billHedgeLosers && + attempt.pending && + attempt.ready && + attempt.response != null && + attempt.reader != null && + !attempt.readerCancelled + ) { + attempt.billAsLoser = true; + } + cancelAttempt(attempt, cancellationKind); } }; @@ -5584,6 +5763,35 @@ export class ProxyForwarder { committed = true; winner = attempt; attempt.pending = false; + // Snapshot the original-session billing context before an alternative + // winner is synced onto the parent session. Otherwise a loser that uses + // the parent session would be priced with the winner's redirected model + // or group multiplier. + if (billHedgeLosers && attempt.session !== session) { + for (const other of attempts.values()) { + if ( + other !== attempt && + other.pending && + other.ready && + other.response != null && + other.reader != null && + !other.readerCancelled + ) { + other.billAsLoser = true; + if (other.session === session && !other.billingSnapshot) { + const loserRequest = session.request.message as Record; + other.billingSnapshot = { + originalModel: session.getOriginalModel(), + redirectedModel: session.getCurrentModel(), + requestedServiceTier: + typeof loserRequest.service_tier === "string" ? loserRequest.service_tier : null, + context1mApplied: session.getContext1mApplied(), + groupCostMultiplier: session.getGroupCostMultiplier(), + }; + } + } + } + } // From this point ResponseHandler owns the reader and agent release. // No coordinator/timer path may cancel or release this attempt again. attempt.readerTransferred = true; @@ -5597,7 +5805,11 @@ export class ProxyForwarder { outcome: "winner", }); const winnerOrigin = - attempt.traceRound === 0 && attempt.kind === "normal" ? "sticky" : attempt.kind; + attempt.traceRound === 0 && attempt.kind === "normal" && !attempt.finalRescue + ? "sticky" + : attempt.kind === "fallback" || attempt.finalRescue + ? "fallback" + : "normal"; if (!attempt.traceFinished) { attempt.traceFinished = true; session.appendRoutingTraceEvent({ @@ -5647,7 +5859,10 @@ export class ProxyForwarder { ProxyForwarder.syncWinningAttemptSession(session, attempt.session); const bindingIntent = - attempt.kind === "fallback" || !bindingWriteAllowed || !session.isSessionBindingAllowed() + attempt.kind === "fallback" || + attempt.finalRescue || + !bindingWriteAllowed || + !session.isSessionBindingAllowed() ? "none" : bindingSnapshot?.providerId == null ? "create" @@ -5664,7 +5879,7 @@ export class ProxyForwarder { endpointUrl: attempt.endpointAudit.endpointUrl, upstreamStatusCode: attempt.response.status, isHedgeWinner: false, - billHedgeLosers: false, + billHedgeLosers, bindingIntent, bindingSnapshot, requiresCompletionMarkerForBinding: bindingIntent === "create" || bindingIntent === "renew", @@ -5920,6 +6135,7 @@ export class ProxyForwarder { return false; } let effectiveKind = kind; + let effectiveFinalRescue = false; if (retrySetupReservation) { const reservedAttempt = coordinator.snapshot.find( (candidate) => candidate.id === retrySetupReservation.placeholderAttemptId @@ -5929,6 +6145,7 @@ export class ProxyForwarder { return false; } effectiveKind = reservedAttempt.kind; + effectiveFinalRescue = reservedAttempt.finalRescue === true; coordinator.removeAttempt(retrySetupReservation.placeholderAttemptId); if (providerSessionRefOwnedByReservation) { retrySetupReservation.providerSessionRefOwned = false; @@ -5958,6 +6175,7 @@ export class ProxyForwarder { const attempt = { id, kind: effectiveKind, + finalRescue: effectiveFinalRescue, controller, parser: new DiscoveryValidityParser(protocol), chunks: [], @@ -6005,6 +6223,7 @@ export class ProxyForwarder { } as typeof winner & { id: string; kind: "normal" | "fallback"; + finalRescue: boolean; controller: AbortController; parser: DiscoveryValidityParser; chunks: Uint8Array[]; @@ -6025,6 +6244,7 @@ export class ProxyForwarder { providerId: provider.id, priority: effectivePriority, kind: effectiveKind, + finalRescue: effectiveFinalRescue, ready: false, pending: true, round: currentRound, @@ -6934,6 +7154,15 @@ export class ProxyForwarder { ); } } + if ("rescueAttemptId" in action && action.rescueAttemptId) { + const rescue = attempts.get(action.rescueAttemptId); + if (rescue) { + // The coordinator retains this attempt only as a final rescue + // lane. It may win the request, but its response must never create + // or renew Sticky. + rescue.finalRescue = true; + } + } // A final-round fallback may still be waiting for a protocol-valid // prefix. Keep it alive; markReady will commit it when it becomes safe. if (action.type === "cancel") return; diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index 855bd7ef4..9bf1bc9c9 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -1368,7 +1368,10 @@ function inspectStreamCompletion( } } -function hasStreamCompletionMarker(text: string, format: ProxySession["originalFormat"]): boolean { +export function hasStreamCompletionMarker( + text: string, + format: ProxySession["originalFormat"] +): boolean { return inspectStreamCompletion(text, format).hasMarker; } @@ -5672,6 +5675,12 @@ export async function finalizeHedgeLoserBilling(params: { * per-request fee for a truncated stream. */ drainComplete: boolean; + /** + * Discovery only accounts responses that report usage explicitly. Legacy + * Hedge leaves this false so complete per-request-priced responses retain + * their established billing semantics. + */ + requireUsage?: boolean; /** * Billing context captured BEFORE the shared session could be polluted by * syncWinningAttemptSession (only set for the INITIAL provider's losing attempt, whose @@ -5695,6 +5704,7 @@ export async function finalizeHedgeLoserBilling(params: { upstreamStatusCode, allContent, drainComplete, + requireUsage = false, billingContext, } = params; @@ -5713,6 +5723,10 @@ export async function finalizeHedgeLoserBilling(params: { ); } + if (requireUsage && !usageForCost) { + return null; + } + // Truncated drain (timeout / cap / abort) with no parsed usage: do NOT fall through to // the per-request-fee sentinel — that would over-bill a phantom fee for an incomplete stream. if (!drainComplete && !usageForCost) { diff --git a/tests/integration/proxy-hedge-lifecycle.test.ts b/tests/integration/proxy-hedge-lifecycle.test.ts index 80e3b2841..7cf7df4b7 100644 --- a/tests/integration/proxy-hedge-lifecycle.test.ts +++ b/tests/integration/proxy-hedge-lifecycle.test.ts @@ -504,6 +504,7 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p const client = new AbortController(); try { state.discoveryEnabled = true; + state.billHedgeLosers = true; const initialProvider = createProvider(1, loser.baseUrl, 0); const winningProvider = createProvider(2, winner.baseUrl, 0); winningProvider.priority = initialProvider.priority; @@ -527,6 +528,7 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p expect(loser.abortCount()).toBe(1); expect(winner.abortCount()).toBe(0); + expect(state.addLoserCost).not.toHaveBeenCalled(); expect(state.acquireDiscoveryLease).toHaveBeenCalledTimes(1); expect(state.compareAndSetBinding).toHaveBeenCalledTimes(1); expect(state.releaseDiscoveryLease).toHaveBeenCalledTimes(1); @@ -537,6 +539,144 @@ describe("proxy hedge transport/lifecycle integration (persistence and control-p } }); + it("drains and bills only a ready Discovery loser after a normal winner commits", async () => { + const [loser, winner] = await Promise.all([startUpstream(), startUpstream()]); + const client = new AbortController(); + try { + state.discoveryEnabled = true; + state.billHedgeLosers = true; + const initialProvider = createProvider(1, loser.baseUrl, 0); + const winningProvider = createProvider(2, winner.baseUrl, 0); + // The lower-priority initial attempt may become ready, but the priority + // gate keeps it held until the preferred candidate resolves. + initialProvider.priority = 2; + winningProvider.priority = 1; + state.providers.push(winningProvider); + const session = await createSession(initialProvider, "/v1/messages", client.signal); + session.sessionId = "integration-discovery"; + const agents = watchAgentReleases(2); + + const forwarded = ProxyForwarder.send(session); + const [loserResponse] = await Promise.all([loser.response, winner.response]); + loserResponse.write( + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"loser"}}\n\n' + ); + await vi.waitFor(() => { + expect(session.getRoutingTrace()?.events).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + attemptId: `${initialProvider.id}:1`, + outcome: "held", + type: "attempt_held", + }), + ]) + ); + }); + + await winner.send( + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"winner"}}\n\n' + + 'event: message_delta\ndata: {"type":"message_delta","usage":{"input_tokens":8,"output_tokens":2}}\n\n' + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n' + ); + const forwardedResponse = await forwarded; + await loser.send( + 'event: message_delta\ndata: {"type":"message_delta","usage":{"input_tokens":7,"output_tokens":2}}\n\n' + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n' + ); + const downstream = await ProxyResponseHandler.dispatch(session, forwardedResponse); + await expect(downstream.text()).resolves.toContain("winner"); + await state.loserBilled.promise; + await settleTasks(); + await agents.released; + + expect(state.addLoserCost).toHaveBeenCalledTimes(1); + expect(state.addLoserCost).toHaveBeenCalledWith( + MESSAGE.id, + expect.objectContaining({ toString: expect.any(Function) }), + expect.objectContaining({ + attemptNumber: 1, + costUsd: "0.011", + providerId: initialProvider.id, + }) + ); + expect(state.updateWinnerCost).toHaveBeenCalledTimes(1); + expect(state.updateMessageRequestCostWithBreakdown).not.toHaveBeenCalled(); + expect(session.getRoutingTrace()?.events).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + attemptId: `${initialProvider.id}:1`, + cancellationKind: "discovery_loser", + outcome: "cancelled", + type: "attempt_finished", + }), + ]) + ); + expect(loser.abortCount()).toBe(0); + expect(winner.abortCount()).toBe(0); + expect(agents.release).toHaveBeenCalledTimes(2); + expect(agents.pool.getPoolStats().activeRequests).toBe(0); + } finally { + client.abort(new Error("fixture cleanup")); + await Promise.all([loser.close(), winner.close()]); + } + }); + + it("does not bill a drained Discovery loser without a protocol completion marker", async () => { + const [loser, winner] = await Promise.all([startUpstream(), startUpstream()]); + const client = new AbortController(); + try { + state.discoveryEnabled = true; + state.billHedgeLosers = true; + const initialProvider = createProvider(1, loser.baseUrl, 0); + const winningProvider = createProvider(2, winner.baseUrl, 0); + initialProvider.priority = 2; + winningProvider.priority = 1; + state.providers.push(winningProvider); + const session = await createSession(initialProvider, "/v1/messages", client.signal); + session.sessionId = "integration-discovery"; + const agents = watchAgentReleases(2); + + const forwarded = ProxyForwarder.send(session); + const [loserResponse] = await Promise.all([loser.response, winner.response]); + loserResponse.write( + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"partial loser"}}\n\n' + ); + await vi.waitFor(() => { + expect(session.getRoutingTrace()?.events).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + attemptId: `${initialProvider.id}:1`, + outcome: "held", + type: "attempt_held", + }), + ]) + ); + }); + + await winner.send( + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"winner"}}\n\n' + + 'event: message_stop\ndata: {"type":"message_stop"}\n\n' + ); + const forwardedResponse = await forwarded; + // Positive usage alone is insufficient: without message_stop this is a + // truncated Anthropic stream and must not be added to the request cost. + await loser.send( + 'event: message_delta\ndata: {"type":"message_delta","usage":{"input_tokens":7,"output_tokens":2}}\n\n' + ); + const downstream = await ProxyResponseHandler.dispatch(session, forwardedResponse); + await expect(downstream.text()).resolves.toContain("winner"); + await agents.released; + + expect(state.addLoserCost).not.toHaveBeenCalled(); + expect(loser.abortCount()).toBe(0); + expect(winner.abortCount()).toBe(0); + expect(agents.pool.getPoolStats().activeRequests).toBe(0); + } finally { + client.abort(new Error("fixture cleanup")); + await Promise.all([loser.close(), winner.close()]); + } + }); + it("fences loser timers after winner settlement and releases each launched transport once", async () => { const [slow, winner, fenced] = await Promise.all([ startUpstream(), diff --git a/tests/unit/proxy/discovery-coordinator.test.ts b/tests/unit/proxy/discovery-coordinator.test.ts index 311a17024..a52e15aff 100644 --- a/tests/unit/proxy/discovery-coordinator.test.ts +++ b/tests/unit/proxy/discovery-coordinator.test.ts @@ -60,12 +60,108 @@ describe("DiscoveryCoordinator", () => { expect(coordinator.onRoundBoundary()).toEqual({ type: "cancel", - attemptIds: ["b"], + attemptIds: [], promoteAttemptId: "a", + rescueAttemptId: "b", + }); + expect(coordinator.snapshot.find((item) => item.id === "a")).toMatchObject({ + kind: "fallback", + pending: true, + }); + expect(coordinator.snapshot.find((item) => item.id === "b")).toMatchObject({ + kind: "normal", + finalRescue: true, + pending: true, }); expect(coordinator.canRefillCurrentRound).toBe(false); }); + it("retains one final rescue beside an existing fallback", () => { + const coordinator = new DiscoveryCoordinator({ + concurrency: 3, + maxRounds: 1, + }); + coordinator.addAttempt(attempt("fallback", 1, "fallback")); + coordinator.addAttempt(attempt("normal-a", 2)); + coordinator.addAttempt(attempt("normal-b", 3)); + + expect(coordinator.onRoundBoundary()).toEqual({ + type: "cancel", + attemptIds: ["normal-b"], + rescueAttemptId: "normal-a", + }); + expect(coordinator.snapshot.find((item) => item.id === "fallback")).toMatchObject({ + kind: "fallback", + pending: true, + }); + expect(coordinator.snapshot.find((item) => item.id === "normal-a")).toMatchObject({ + kind: "normal", + finalRescue: true, + pending: true, + }); + expect(coordinator.snapshot.find((item) => item.id === "normal-b")?.pending).toBe(false); + }); + + it("lets either final rescue lane win, without opening Sticky eligibility", () => { + const fallbackFirst = new DiscoveryCoordinator({ + concurrency: 2, + maxRounds: 1, + }); + fallbackFirst.addAttempt(attempt("fallback", 1, "fallback")); + fallbackFirst.addAttempt(attempt("standby", 2)); + const boundary = fallbackFirst.onRoundBoundary(); + expect(boundary).toMatchObject({ rescueAttemptId: "standby" }); + expect(fallbackFirst.markReady("fallback")).toEqual({ + type: "promote_fallback", + attemptId: "fallback", + }); + + const standbyFirst = new DiscoveryCoordinator({ + concurrency: 2, + maxRounds: 1, + }); + standbyFirst.addAttempt(attempt("fallback", 1, "fallback")); + standbyFirst.addAttempt(attempt("standby", 2)); + standbyFirst.onRoundBoundary(); + expect(standbyFirst.markReady("standby")).toEqual({ + type: "commit_normal", + attemptId: "standby", + }); + }); + + it("does not create a final rescue lane when concurrency is one", () => { + const coordinator = new DiscoveryCoordinator({ + concurrency: 1, + maxRounds: 1, + }); + coordinator.addAttempt(attempt("fallback", 1, "fallback")); + coordinator.addAttempt(attempt("normal", 2)); + + expect(coordinator.onRoundBoundary()).toEqual({ + type: "cancel", + attemptIds: ["normal"], + }); + expect(coordinator.snapshot.find((item) => item.id === "fallback")?.pending).toBe(true); + expect(coordinator.snapshot.find((item) => item.id === "normal")?.pending).toBe(false); + }); + + it("waits for a final rescue after the primary fallback fails", () => { + const coordinator = new DiscoveryCoordinator({ + concurrency: 2, + maxRounds: 1, + }); + coordinator.addAttempt(attempt("fallback", 1, "fallback")); + coordinator.addAttempt(attempt("standby", 2)); + coordinator.onRoundBoundary(); + + expect(coordinator.markFailed("fallback")).toEqual({ type: "none" }); + expect(coordinator.snapshot.find((item) => item.id === "standby")?.pending).toBe(true); + expect(coordinator.markReady("standby")).toEqual({ + type: "commit_normal", + attemptId: "standby", + }); + }); + it("counts setup reservations as occupied slots without promoting them to fallback", () => { const coordinator = new DiscoveryCoordinator({ concurrency: 2, maxRounds: 2 }); coordinator.addAttempt(attempt("transport", 1)); 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 ce0a09001..534c05732 100644 --- a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts +++ b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts @@ -5133,6 +5133,182 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => { } }); + test("a final-round standby can rescue after the boundary without writing Sticky", async () => { + vi.useFakeTimers(); + try { + const primary = createProvider({ id: 1, name: "primary", priority: 1 }); + const standby = createProvider({ id: 2, name: "standby", priority: 2 }); + const session = createSession(); + session.authState = { + success: true, + user: null, + key: { id: 36 }, + apiKey: null, + } as typeof session.authState; + session.setProvider(primary); + mocks.getCachedSystemSettings.mockResolvedValue({ + discoveryEnabled: true, + discoveryConcurrency: 2, + maxDiscoveryRounds: 1, + discoverySlaMs: 10, + stickySlaMs: 10, + racingTotalTimeoutMs: 100, + stickyTimeoutCooldownMs: 300_000, + }); + mocks.pickDiscoveryProviders.mockResolvedValueOnce([standby]); + + const streamControllers = new Map>(); + const abortedProviders = new Set(); + const doForward = vi.spyOn( + ProxyForwarder as unknown as { + doForward: (...args: unknown[]) => Promise; + }, + "doForward" + ); + doForward.mockImplementation( + async (attemptSession, _provider, _baseUrl, _audit, _count, _stream, signal) => { + const providerId = (attemptSession as ProxySession).provider!.id; + return new Response( + new ReadableStream({ + start(controller) { + streamControllers.set(providerId, controller); + signal?.addEventListener( + "abort", + () => { + abortedProviders.add(providerId); + try { + controller.close(); + } catch { + // The winner can close naturally before loser cleanup. + } + }, + { once: true } + ); + }, + }), + { headers: { "content-type": "text/event-stream" } } + ); + } + ); + + const responsePromise = ProxyForwarder.send(session); + await vi.advanceTimersByTimeAsync(0); + expect(doForward).toHaveBeenCalledTimes(2); + + // Both attempts are still pending when the only Discovery SLA window + // closes. The primary becomes the fallback and standby becomes the one + // retained final rescue lane. + await vi.advanceTimersByTimeAsync(10); + streamControllers + .get(standby.id)! + .enqueue( + new TextEncoder().encode( + 'data: {"type":"content_block_delta","delta":{"text":"standby"}}\n\n' + ) + ); + streamControllers.get(standby.id)!.close(); + await vi.advanceTimersByTimeAsync(0); + + const response = await responsePromise; + expect(await response.text()).toContain('"standby"'); + expect(session.provider?.id).toBe(standby.id); + expect(abortedProviders).toContain(primary.id); + expect(peekDeferredStreamingFinalization(session)).toEqual( + expect.objectContaining({ + providerId: standby.id, + bindingIntent: "none", + requiresCompletionMarkerForBinding: false, + }) + ); + } finally { + vi.useRealTimers(); + } + }); + + test("a rectified final-round standby remains ineligible for Sticky", async () => { + vi.useFakeTimers(); + try { + const primary = createProvider({ id: 1, name: "primary", priority: 1 }); + const standby = createProvider({ id: 2, name: "standby", priority: 2 }); + const session = createSession(); + session.authState = { + success: true, + user: null, + key: { id: 37 }, + apiKey: null, + } as typeof session.authState; + session.setProvider(primary); + withThinkingBlocks(session); + mocks.getCachedSystemSettings.mockResolvedValue({ + discoveryEnabled: true, + discoveryConcurrency: 2, + maxDiscoveryRounds: 1, + discoverySlaMs: 10, + stickySlaMs: 10, + racingTotalTimeoutMs: 100, + stickyTimeoutCooldownMs: 300_000, + enableThinkingSignatureRectifier: true, + }); + mocks.pickDiscoveryProviders.mockResolvedValueOnce([standby]); + + const standbyFirst = Promise.withResolvers(); + const signatureError = new UpstreamProxyError( + "Invalid `signature` in `thinking` block", + 400, + { + body: '{"error":"invalid_signature"}', + providerId: standby.id, + providerName: standby.name, + } + ); + let standbyAttempts = 0; + const doForward = vi.spyOn( + ProxyForwarder as unknown as { + doForward: (...args: unknown[]) => Promise; + }, + "doForward" + ); + doForward.mockImplementation(async (attemptSession) => { + const providerId = (attemptSession as ProxySession).provider!.id; + if (providerId === primary.id) { + return new Response(new ReadableStream(), { + headers: { "content-type": "text/event-stream" }, + }); + } + standbyAttempts += 1; + if (standbyAttempts === 1) return standbyFirst.promise; + return new Response( + 'data: {"type":"content_block_delta","delta":{"text":"rectified standby"}}\n\n', + { headers: { "content-type": "text/event-stream" } } + ); + }); + + const responsePromise = ProxyForwarder.send(session); + await vi.advanceTimersByTimeAsync(0); + expect(doForward).toHaveBeenCalledTimes(2); + + // The boundary first marks this still-pending normal as the final rescue. + // Its provider-local rectifier retry must inherit that no-Sticky role. + await vi.advanceTimersByTimeAsync(10); + standbyFirst.reject(signatureError); + await vi.advanceTimersByTimeAsync(0); + + const response = await responsePromise; + expect(await response.text()).toContain('"rectified standby"'); + expect(standbyAttempts).toBe(2); + expect(session.provider?.id).toBe(standby.id); + expect(peekDeferredStreamingFinalization(session)).toEqual( + expect.objectContaining({ + providerId: standby.id, + bindingIntent: "none", + requiresCompletionMarkerForBinding: false, + }) + ); + } finally { + vi.useRealTimers(); + } + }); + test("the Discovery deadline releases a stalled rectifier retry reservation exactly once", async () => { vi.useFakeTimers(); try { diff --git a/tests/unit/proxy/response-handler-exported-finalizers.test.ts b/tests/unit/proxy/response-handler-exported-finalizers.test.ts index 2fc324b1f..6ce969d2a 100644 --- a/tests/unit/proxy/response-handler-exported-finalizers.test.ts +++ b/tests/unit/proxy/response-handler-exported-finalizers.test.ts @@ -102,7 +102,17 @@ function createProvider(): Provider { } satisfies Provider; } -async function createSession(provider: Provider | null): Promise { +async function createSession( + provider: Provider | null, + priceData: { + input_cost_per_token: number; + output_cost_per_token: number; + input_cost_per_request?: number; + } = { + input_cost_per_token: 1, + output_cost_per_token: 10, + } +): Promise { const request = new Request("https://hub.test/v1/messages", { body: JSON.stringify({ messages: [], model: "claude-test", stream: false }), headers: { "content-type": "application/json" }, @@ -118,7 +128,7 @@ async function createSession(provider: Provider | null): Promise { }); Object.defineProperty(session, "getResolvedPricingByBillingSource", { value: vi.fn(async () => ({ - priceData: { input_cost_per_token: 1, output_cost_per_token: 10 }, + priceData, resolvedModelName: "claude-test", resolvedPricingProviderKey: "anthropic", source: "official_fallback" as const, @@ -177,6 +187,30 @@ describe("exported response finalizers", () => { expect(mocks.addLoserCost).not.toHaveBeenCalled(); }); + it("skips Discovery loser billing without explicit usage even for per-request pricing", async () => { + const provider = createProvider(); + const session = await createSession(provider, { + input_cost_per_token: 1, + output_cost_per_token: 10, + input_cost_per_request: 5, + }); + + const billed = await finalizeHedgeLoserBilling({ + allContent: 'event: message_stop\ndata: {"type":"message_stop"}\n\n', + attemptNumber: 3, + drainComplete: true, + loserSession: session, + messageRequestCreatedAtMs: 0, + messageRequestId: 71, + provider, + requireUsage: true, + upstreamStatusCode: 200, + }); + + expect(billed).toBeNull(); + expect(mocks.addLoserCost).not.toHaveBeenCalled(); + }); + it("adds a complete hedge loser's calculated cost to the original request", async () => { const provider = createProvider(); const session = await createSession(provider); From de15e9c3209e46c1fa429667a7908c15db0d6d7b Mon Sep 17 00:00:00 2001 From: Brisbanehuang Date: Wed, 22 Jul 2026 10:31:23 -0400 Subject: [PATCH 2/2] fix(discovery): address review feedback --- docs/streaming-discovery.md | 8 ++++--- src/app/api/admin/system-config/route.ts | 9 +++++++- src/app/v1/_lib/proxy/discovery-validity.ts | 5 +++- .../api/admin-system-config-route.test.ts | 23 +++++++++++++++++++ 4 files changed, 40 insertions(+), 5 deletions(-) diff --git a/docs/streaming-discovery.md b/docs/streaming-discovery.md index 39d261241..1c6d221e5 100644 --- a/docs/streaming-discovery.md +++ b/docs/streaming-discovery.md @@ -34,9 +34,11 @@ this relationship. - A fallback that has produced a valid prefix is held until the current normal window closes, all normal candidates fail, or no candidates remain. A normal winner always has precedence during the window. -- Discovery losers are cancelled and their readers/agents/provider-session - references are released. They do not enter legacy `bill_hedge_losers` - draining. +- When `bill_hedge_losers` is enabled, a Discovery loser that already produced + a protocol-valid prefix and reached ready state may reuse the legacy + background drain and billing path. It is billed only after natural stream + completion with a completion marker and explicit usage. All other losers are + cancelled and their readers/agents/provider-session references are released. - Sticky binding is written only after a natural, successful stream completion with the protocol completion marker and a generation-aware CAS. Fake-200, incomplete, and client-aborted streams do not create or renew Sticky. diff --git a/src/app/api/admin/system-config/route.ts b/src/app/api/admin/system-config/route.ts index 59733bf5c..c0bf96b9a 100644 --- a/src/app/api/admin/system-config/route.ts +++ b/src/app/api/admin/system-config/route.ts @@ -7,7 +7,10 @@ import { invalidateAllOverviewCaches, invalidateAllStatisticsCaches, } from "@/lib/redis"; -import { DISCOVERY_WINDOW_INVALID_ERROR_CODE } from "@/lib/validation/discovery-settings"; +import { + DISCOVERY_WINDOW_INVALID_ERROR_CODE, + getDiscoveryValidationErrorCode, +} from "@/lib/validation/discovery-settings"; import { UpdateSystemSettingsSchema } from "@/lib/validation/schemas"; import { getSystemSettings, updateSystemSettings } from "@/repository/system-config"; @@ -139,6 +142,10 @@ export async function POST(req: Request) { return Response.json(updated); } catch (error) { if (error instanceof z.ZodError) { + const errorCode = getDiscoveryValidationErrorCode(error.issues); + if (errorCode === DISCOVERY_WINDOW_INVALID_ERROR_CODE) { + return Response.json({ error: "discoveryWindowInvalid", errorCode }, { status: 400 }); + } const firstError = error.issues[0]; return Response.json({ error: firstError.message || "数据验证失败" }, { status: 400 }); } diff --git a/src/app/v1/_lib/proxy/discovery-validity.ts b/src/app/v1/_lib/proxy/discovery-validity.ts index d3da1a605..146866141 100644 --- a/src/app/v1/_lib/proxy/discovery-validity.ts +++ b/src/app/v1/_lib/proxy/discovery-validity.ts @@ -14,6 +14,7 @@ export type DiscoveryValidity = { export const DISCOVERY_PREFIX_MAX_BYTES = 1024 * 1024; export const DISCOVERY_EVENT_MAX_COUNT = 1024; +const DISCOVERY_TEXT_ENCODER = new TextEncoder(); function hasContent(value: unknown): boolean { if (typeof value === "string") return value.trim().length > 0; @@ -201,7 +202,9 @@ export class DiscoveryValidityParser { push(chunk: Uint8Array | string): DiscoveryValidity { if (this._error) return this.result; this.bytesSeen += - typeof chunk === "string" ? new TextEncoder().encode(chunk).byteLength : chunk.byteLength; + typeof chunk === "string" + ? DISCOVERY_TEXT_ENCODER.encode(chunk).byteLength + : chunk.byteLength; if (!this._ready && this.bytesSeen > DISCOVERY_PREFIX_MAX_BYTES) { this._error = true; this._limitExceeded = true; diff --git a/tests/unit/api/admin-system-config-route.test.ts b/tests/unit/api/admin-system-config-route.test.ts index d6eb23e2f..15527976e 100644 --- a/tests/unit/api/admin-system-config-route.test.ts +++ b/tests/unit/api/admin-system-config-route.test.ts @@ -62,4 +62,27 @@ describe("POST /api/admin/system-config", () => { }); expect(mocks.updateSystemSettings).not.toHaveBeenCalled(); }); + + it("returns the same stable Discovery window code for a complete update", async () => { + 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({ + discoverySlaMs: 10_000, + stickySlaMs: 20_000, + maxDiscoveryRounds: 2, + racingTotalTimeoutMs: 39_999, + }), + }) + ); + + expect(response.status).toBe(400); + await expect(response.json()).resolves.toEqual({ + error: "discoveryWindowInvalid", + errorCode: "DISCOVERY_WINDOW_INVALID", + }); + expect(mocks.updateSystemSettings).not.toHaveBeenCalled(); + }); });