From 69f164feb5bfb987346df1134eb7f28d6760eb8c Mon Sep 17 00:00:00 2001 From: ding113 Date: Sun, 2 Aug 2026 03:08:21 +0800 Subject: [PATCH] feat(logs): show live upstream providers --- .../thinking-effort-display.test.tsx | 1 + .../_components/thinking-effort-display.tsx | 2 +- .../_components/usage-logs-table.test.tsx | 2 +- .../logs/_components/usage-logs-table.tsx | 2 +- .../virtualized-logs-table.test.tsx | 60 ++++++++++++- .../_components/virtualized-logs-table.tsx | 63 +++++++++++-- src/app/v1/_lib/proxy/forwarder.ts | 5 ++ src/app/v1/_lib/proxy/session.ts | 52 ++++++++++- .../redis/live-chain-store.storage.test.ts | 65 ++++++++++++++ src/lib/redis/live-chain-store.ts | 88 ++++++++++++++++++- src/repository/usage-logs.ts | 1 + 11 files changed, 329 insertions(+), 12 deletions(-) diff --git a/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.test.tsx b/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.test.tsx index d9f75b192..5a1367fc6 100644 --- a/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.test.tsx @@ -70,6 +70,7 @@ describe("ThinkingEffortDisplay", () => { expect(html).toContain("max"); expect(html).toContain("reasoningEffort.overridden"); expect(html).toContain("lucide-arrow-right"); + expect(html).toContain("relative z-20"); }); test("显示 Anthropic 请求中的思考强度", () => { diff --git a/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.tsx b/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.tsx index 48ed68ba3..624b8cae2 100644 --- a/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.tsx +++ b/src/app/[locale]/dashboard/logs/_components/thinking-effort-display.tsx @@ -34,7 +34,7 @@ export function ThinkingEffortDisplay({ specialSettings }: ThinkingEffortDisplay {effortInfo.requestedEffort && ( diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx index d1a7117a8..26b75cf1b 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx @@ -187,7 +187,7 @@ describe("usage-logs-table thinking effort", () => { expect(cells[6]?.textContent).toContain("gpt-5.4"); expect(cells[7]?.textContent).toContain("low"); expect(cells[7]?.textContent).toContain("max"); - expect(cells[7]?.className).toContain("overflow-hidden"); + expect(cells[7]?.className).toContain("overflow-visible"); }); test("显示 Anthropic 请求的思考强度", () => { diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx index 535530b1e..2bcca1737 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx @@ -317,7 +317,7 @@ export function UsageLogsTable({ {hideReasoningEffortColumn ? null : ( - + )} diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx index b2459e1b6..d638a6eb8 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx @@ -279,7 +279,7 @@ describe("virtualized-logs-table thinking effort", () => { const effortDisplay = container.querySelector('[data-slot="thinking-effort"]'); expect(effortDisplay?.textContent).toContain("low"); expect(effortDisplay?.textContent).toContain("max"); - expect(effortDisplay?.closest(".overflow-hidden")).not.toBeNull(); + expect(effortDisplay?.closest(".overflow-visible")).not.toBeNull(); }); test("显示 Anthropic 请求的思考强度", () => { @@ -943,6 +943,64 @@ describe("virtualized-logs-table live chain display", () => { expect(html).toContain("text-indigo-500"); }); + test("stacks every currently connected racing provider and exposes the full list", () => { + setupLiveChainDefaults(); + mockLogs = [ + makeLog({ + id: 1, + statusCode: null, + providerChain: null, + _liveChain: { + chain: [], + activeProviders: [ + { id: 1, name: "openai-east-with-a-long-name" }, + { id: 2, name: "anthropic-west-with-a-long-name" }, + { id: 3, name: "gemini-central-with-a-long-name" }, + ], + phase: "hedge_racing", + updatedAt: Date.now(), + }, + }), + ]; + + const html = renderToStaticMarkup( + + ); + + expect(html).toContain('data-slot="live-provider-stack"'); + expect(html).toContain("openai-east-with-a-long-name"); + expect(html).toContain("anthropic-west-with-a-long-name"); + expect(html).toContain("gemini-central-with-a-long-name"); + expect(html).toContain('data-slot="live-provider-tooltip"'); + }); + + test("shows only the newly active provider after fallback switches", () => { + setupLiveChainDefaults(); + mockLogs = [ + makeLog({ + id: 1, + statusCode: null, + providerChain: null, + _liveChain: { + chain: [ + { id: 1, name: "primary-provider", reason: "retry_failed" }, + { id: 2, name: "fallback-provider", reason: "initial_selection" }, + ], + activeProviders: [{ id: 2, name: "fallback-provider" }], + phase: "provider_selected", + updatedAt: Date.now(), + }, + }), + ]; + + const html = renderToStaticMarkup( + + ); + + expect(html).toContain("fallback-provider"); + expect(html).not.toContain("primary-provider"); + }); + test("renders generic in-progress when live chain is empty", () => { setupLiveChainDefaults(); mockLogs = [ diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx index 6565ad4e7..223d399cb 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx @@ -97,6 +97,47 @@ function StatusBadgeOnly({ statusCode }: { statusCode: number | null }) { ); } +function LiveProviderStack({ providers }: { providers: Array<{ id: number; name: string }> }) { + const visibleProviders = providers.slice(0, 3); + const hiddenProviderCount = providers.length - visibleProviders.length; + + return ( + + + + + {visibleProviders.map((provider) => ( + + {provider.name} + + ))} + {hiddenProviderCount > 0 && ( + + +{hiddenProviderCount} + + )} + + + +
+
    + {providers.map((provider) => ( +
  • {provider.name}
    • + ))} +
+
+ + + + ); +} + interface VirtualizedLogsTableProps { filters: VirtualizedLogsTableFilters; currencyCode?: CurrencyCode; @@ -887,11 +928,21 @@ export function VirtualizedLogsTable({ log._liveChain ? (
- - {log._liveChain.chain.length > 0 - ? log._liveChain.chain[log._liveChain.chain.length - 1].name - : t("logs.details.inProgress")} - + {log._liveChain.activeProviders ? ( + log._liveChain.activeProviders.length > 0 ? ( + + ) : ( + + {t("logs.details.inProgress")} + + ) + ) : ( + + {log._liveChain.chain.length > 0 + ? log._liveChain.chain[log._liveChain.chain.length - 1].name + : t("logs.details.inProgress")} + + )} {log._liveChain.phase === "retrying" && ( +
)} diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 9e7dd1670..3f60a1c9b 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -4558,6 +4558,7 @@ export class ProxyForwarder { attempt.thresholdTimer = null; } attempts.delete(attempt); + session.removeLiveActiveProvider(attempt.provider.id); // 竞速输家计费开启：仅标记 + 记录决策链，不取消连接、不释放 agent。 // 实际的后台 drain 由 runAttempt 的 .then 流程发起（它独占 reader，避免并发读）。 @@ -4855,6 +4856,9 @@ export class ProxyForwarder { }; const handleAttemptFailure = async (attempt: StreamingHedgeAttempt, error: Error) => { + if (attempt !== winnerAttempt) { + session.removeLiveActiveProvider(attempt.provider.id); + } // 已被标记为计费输家、billing 尚未启动、却在此失败（如首块读取出错 / 赢家已提交）： // 此时 abortAttempt 已早退（未取消连接/未释放 agent），由这里兜底清理，避免 reader/agent 泄漏。 if ( @@ -5344,6 +5348,7 @@ export class ProxyForwarder { }; attempts.add(attempt); + session.addLiveActiveProvider(provider); // Record hedge participant launch in decision chain // (first provider is already recorded via initial_selection or session_reuse) diff --git a/src/app/v1/_lib/proxy/session.ts b/src/app/v1/_lib/proxy/session.ts index 5c681b44b..a926be8ce 100644 --- a/src/app/v1/_lib/proxy/session.ts +++ b/src/app/v1/_lib/proxy/session.ts @@ -2,6 +2,7 @@ import type { Context } from "hono"; import { logger } from "@/lib/logger"; import { deleteLiveChain, + type LiveProviderSnapshot, writeLiveChain, writeLiveRoutingTrace, } from "@/lib/redis/live-chain-store"; @@ -206,6 +207,8 @@ export class ProxySession { // 上游决策链（记录尝试的供应商列表） private providerChain: ProviderChainItem[]; + private liveActiveProviders = new Map(); + private liveActiveProviderCounts = new Map(); // Request-level routing observability. Discovery attempts live here rather // than providerChain because providerChain is also a billing/retry contract. @@ -423,6 +426,45 @@ export class ProxySession { if (provider) { this.providerType = provider.providerType as ProviderType; } + if (!this.liveActiveProviders) { + this.liveActiveProviders = new Map(); + } + if (!this.liveActiveProviderCounts) { + this.liveActiveProviderCounts = new Map(); + } + this.liveActiveProviders.clear(); + this.liveActiveProviderCounts.clear(); + if (provider) { + this.liveActiveProviders.set(provider.id, { id: provider.id, name: provider.name }); + this.liveActiveProviderCounts.set(provider.id, 1); + } + this.persistLiveChain(); + } + + addLiveActiveProvider(provider: Pick): void { + if (!this.liveActiveProviders) { + this.liveActiveProviders = new Map(); + } + if (!this.liveActiveProviderCounts) { + this.liveActiveProviderCounts = new Map(); + } + this.liveActiveProviders.set(provider.id, { id: provider.id, name: provider.name }); + this.liveActiveProviderCounts.set( + provider.id, + (this.liveActiveProviderCounts.get(provider.id) ?? 0) + 1 + ); + this.persistLiveChain(); + } + + removeLiveActiveProvider(providerId: number): void { + const count = this.liveActiveProviderCounts?.get(providerId) ?? 0; + if (count <= 1) { + this.liveActiveProviderCounts?.delete(providerId); + if (this.liveActiveProviders?.delete(providerId)) this.persistLiveChain(); + return; + } + this.liveActiveProviderCounts.set(providerId, count - 1); + this.persistLiveChain(); } setSessionBindingSnapshot(snapshot: SessionBindingSnapshot | null): void { @@ -862,11 +904,19 @@ export class ProxySession { this.liveRoutingTraceDirty = false; const chain = writeChain ? structuredClone(this.providerChain) : null; + const activeProviders = writeChain + ? structuredClone([...this.liveActiveProviders.values()]) + : null; const routingTrace = writeRoutingTrace ? structuredClone(this.routingTrace) : null; const writes: Promise[] = []; if (chain) { writes.push( - writeLiveChain(this.sessionId as string, this.requestSequence as number, chain) + writeLiveChain( + this.sessionId as string, + this.requestSequence as number, + chain, + activeProviders ?? [] + ) ); } if (routingTrace) { diff --git a/src/lib/redis/live-chain-store.storage.test.ts b/src/lib/redis/live-chain-store.storage.test.ts index d2a9492ef..03831859b 100644 --- a/src/lib/redis/live-chain-store.storage.test.ts +++ b/src/lib/redis/live-chain-store.storage.test.ts @@ -102,6 +102,71 @@ describe("live-chain routing trace storage", () => { }); }); + it("derives all currently connected Discovery providers from attempt lifecycle events", async () => { + const trace = makeTrace({ + updatedAt: 240, + events: [ + { + type: "attempt_started", + at: 200, + elapsedMs: 100, + round: 1, + attemptId: "11:1", + attemptKind: "normal", + provider: { id: 11, name: "provider-a" }, + }, + { + type: "attempt_started", + at: 210, + elapsedMs: 110, + round: 1, + attemptId: "12:1", + attemptKind: "normal", + provider: { id: 12, name: "provider-b" }, + }, + { + type: "attempt_finished", + at: 220, + elapsedMs: 120, + round: 1, + attemptId: "11:1", + attemptKind: "normal", + provider: { id: 11, name: "provider-a" }, + outcome: "failed", + }, + { + type: "attempt_started", + at: 230, + elapsedMs: 130, + round: 1, + attemptId: "13:1", + attemptKind: "fallback", + provider: { id: 13, name: "provider-c" }, + }, + ], + }); + await writeLiveRoutingTrace("racing", 1, trace); + + await expect(readLiveChain("racing", 1)).resolves.toMatchObject({ + activeProviders: [ + { id: 12, name: "provider-b" }, + { id: 13, name: "provider-c" }, + ], + }); + }); + + it("switches the active legacy provider after a serial fallback", async () => { + await writeLiveChain("fallback", 1, [ + { id: 21, name: "primary", reason: "initial_selection", timestamp: 100 }, + { id: 21, name: "primary", reason: "retry_failed", timestamp: 200 }, + { id: 22, name: "fallback", reason: "initial_selection", timestamp: 210 }, + ]); + + await expect(readLiveChain("fallback", 1)).resolves.toMatchObject({ + activeProviders: [{ id: 22, name: "fallback" }], + }); + }); + it("returns an early trace before the provider chain snapshot exists", async () => { const trace = makeTrace({ updatedAt: 150, diff --git a/src/lib/redis/live-chain-store.ts b/src/lib/redis/live-chain-store.ts index 7429bff9f..3322a6e7f 100644 --- a/src/lib/redis/live-chain-store.ts +++ b/src/lib/redis/live-chain-store.ts @@ -4,10 +4,16 @@ import type { ProviderChainItem } from "@/types/message"; import { normalizeRoutingTrace, type RoutingTraceV1 } from "@/types/routing-trace"; import { RedisKVStore } from "./redis-kv-store"; +export interface LiveProviderSnapshot { + id: number; + name: string; +} + export interface LiveChainSnapshot { chain: ProviderChainItem[]; phase: string; updatedAt: number; + activeProviders?: LiveProviderSnapshot[]; routingTrace?: RoutingTraceV1 | null; } @@ -78,6 +84,80 @@ function inferDiscoveryPhase(trace: RoutingTraceV1): string { return "discovery_racing"; } +function deriveDiscoveryActiveProviders(trace: RoutingTraceV1): LiveProviderSnapshot[] | undefined { + const activeAttempts = new Map(); + let sawAttemptLifecycle = false; + + for (const event of trace.events) { + if (event.type === "attempt_started" && event.attemptId && event.provider) { + sawAttemptLifecycle = true; + activeAttempts.set(event.attemptId, { + id: event.provider.id, + name: event.provider.name ?? String(event.provider.id), + }); + continue; + } + + if (event.type === "attempt_finished" && event.attemptId) { + sawAttemptLifecycle = true; + activeAttempts.delete(event.attemptId); + continue; + } + + if (event.type === "winner_committed" && event.attemptId && event.provider) { + sawAttemptLifecycle = true; + activeAttempts.clear(); + activeAttempts.set(event.attemptId, { + id: event.provider.id, + name: event.provider.name ?? String(event.provider.id), + }); + continue; + } + + if (event.type === "request_finished") { + sawAttemptLifecycle = true; + activeAttempts.clear(); + } + } + + if (!sawAttemptLifecycle) return undefined; + return [ + ...new Map([...activeAttempts.values()].map((provider) => [provider.id, provider])).values(), + ]; +} + +function deriveLegacyActiveProviders(chain: ProviderChainItem[]): LiveProviderSnapshot[] { + const activeProviders = new Map(); + + for (const item of chain) { + const provider = { id: item.id, name: item.name }; + switch (item.reason) { + case "initial_selection": + case "session_reuse": + case "affinity_hit": + case "hedge_launched": + activeProviders.set(item.id, provider); + break; + case "hedge_winner": + case "request_success": + case "retry_success": + activeProviders.clear(); + activeProviders.set(item.id, provider); + break; + case "retry_failed": + case "system_error": + case "resource_not_found": + case "hedge_loser_cancelled": + case "hedge_loser_billed": + case "client_abort": + activeProviders.delete(item.id); + break; + } + } + + return [...activeProviders.values()]; +} + export function inferPhase( chain: ProviderChainItem[], routingTrace?: RoutingTraceV1 | null @@ -122,6 +202,8 @@ function mergeSnapshot( // snapshot shape. Prefer the independently updated trace when both exist. const routingTrace = normalizeRoutingTrace(storedRoutingTrace) ?? normalizeRoutingTrace(snapshot?.routingTrace); + const activeProviders = + routingTrace?.mode === "discovery" ? deriveDiscoveryActiveProviders(routingTrace) : undefined; // Trace recording starts before provider selection can append to the legacy // chain. Keep that earliest Discovery state visible instead of waiting for a @@ -132,6 +214,7 @@ function mergeSnapshot( chain: [], phase: inferPhase([], routingTrace), updatedAt: routingTrace.updatedAt, + ...(activeProviders ? { activeProviders } : {}), routingTrace, }; } @@ -139,6 +222,7 @@ function mergeSnapshot( return { ...snapshot, phase: inferPhase(snapshot.chain, routingTrace), + ...(activeProviders ? { activeProviders } : {}), routingTrace, }; } @@ -146,12 +230,14 @@ function mergeSnapshot( export async function writeLiveChain( sessionId: string, requestSequence: number, - chain: ProviderChainItem[] + chain: ProviderChainItem[], + activeProviders: LiveProviderSnapshot[] = deriveLegacyActiveProviders(chain) ): Promise { const snapshot: LiveChainSnapshot = { chain, phase: inferPhase(chain), updatedAt: Date.now(), + activeProviders, }; await store.set(buildKey(sessionId, requestSequence), snapshot); } diff --git a/src/repository/usage-logs.ts b/src/repository/usage-logs.ts index 9a5a6121b..56c912c50 100644 --- a/src/repository/usage-logs.ts +++ b/src/repository/usage-logs.ts @@ -113,6 +113,7 @@ export interface UsageLogRow { chain: ProviderChainItem[]; phase: string; updatedAt: number; + activeProviders?: Array<{ id: number; name: string }>; routingTrace?: RoutingTraceV1 | null; } | null; anthropicEffort?: string | null;