From 73c0ecdc4cf816dfd75ca6a5f0423bd07615a3b4 Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 16 Jul 2026 15:57:19 +0800 Subject: [PATCH] feat(telemetry): add first-token latency observability Measure TTFT from the first semantic response event across streaming, retry, scheduler, usage telemetry, and dashboard paths. Expose first-token percentiles and concise UI labels, and isolate concurrent Redis-backed test runs. Co-authored-by: Codex --- backend/src/dispatch/controllers/mod.rs | 92 +++++++++++- .../src/dispatch/controllers/telemetry/mod.rs | 25 +++- backend/src/dispatch/lifecycle/attempt.rs | 18 +-- backend/src/dispatch/lifecycle/trace.rs | 8 + backend/src/dispatch/service.rs | 5 +- backend/src/dispatch/stream/live.rs | 2 +- backend/src/dispatch/transport/canonical.rs | 49 ++---- backend/src/fleet/pool/mod.rs | 7 +- backend/src/fleet/pool/state.rs | 11 +- backend/src/fleet/scheduler/feedback.rs | 91 ++++++++--- backend/src/fleet/scheduler/mod.rs | 10 +- backend/src/fleet/scheduler/strategy/smart.rs | 2 +- backend/src/telemetry/usage/insights.rs | 42 +++--- backend/src/upstream/openai/failure.rs | 2 +- .../src/upstream/openai/protocol/responses.rs | 141 ++++++++++++------ .../openai/transport/response_meta.rs | 4 +- .../upstream/openai/transport/websocket.rs | 2 +- .../openai/transport/websocket_frames.rs | 18 +-- .../tests/dispatch/service/usage_logging.rs | 2 +- backend/tests/fleet/scheduler/feedback.rs | 60 ++++---- backend/tests/fleet/scheduler/score.rs | 6 +- backend/tests/support/storage.rs | 91 +++++++++-- backend/tests/upstream/openai/protocol/mod.rs | 3 +- backend/tests/upstream/openai/protocol/sse.rs | 76 +++++++--- .../dashboard/composables/useDashboard.ts | 2 +- .../usage/components/UsagePerformanceCard.vue | 14 +- .../components/UsageRecordDetailModal.vue | 2 +- .../usage/composables/useUsageRecordsTable.ts | 8 +- frontend/src/views/usage/constants.ts | 2 +- 29 files changed, 552 insertions(+), 243 deletions(-) diff --git a/backend/src/dispatch/controllers/mod.rs b/backend/src/dispatch/controllers/mod.rs index 6a5cc1ede..b0588cf8e 100644 --- a/backend/src/dispatch/controllers/mod.rs +++ b/backend/src/dispatch/controllers/mod.rs @@ -51,6 +51,7 @@ use crate::{ }, }, fleet::pool::AccountPoolService, + fleet::scheduler::AttemptFeedback, upstream::openai::{ failure::UpstreamFailureFacts, protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, @@ -453,6 +454,18 @@ impl ControllerSet { ) .await; } + let feedback = if exit.completed { + AttemptFeedback::Completed { + first_token_ms: nonnegative_u64(exit.response.first_token_ms), + } + } else { + AttemptFeedback::Incomplete { + first_token_ms: nonnegative_u64(exit.response.first_token_ms), + } + }; + self.account_pool + .report_attempt(exit.account_id, feedback) + .await; if successful { let _ = best_effort_controller_io( @@ -657,6 +670,12 @@ impl ControllerSet { }), ) .await; + self.account_pool + .report_attempt( + &context.account_id, + stream_feedback(summary, context.started_at, context.attempt.started_at()), + ) + .await; let fallback = usage::StreamUsageExit { rate_limit_headers: context.rate_limit_headers.clone(), @@ -726,7 +745,6 @@ impl ControllerSet { }), ) .await; - if !matches!(outcome, FinalOutcome::Cancelled | FinalOutcome::Shutdown) { if let Some(classification) = stream_failure.as_ref() { self.apply_stream_policy_failure(&scope, &context.account_id, classification) @@ -750,6 +768,13 @@ impl ControllerSet { usage::UsageController::observe_attempt(&self.account_pool, &scope.usage, observation), ) .await; + if let (Some(account), Some(feedback)) = + (observation.account.as_ref(), attempt_feedback(observation)) + { + self.account_pool + .report_attempt(&account.id, feedback) + .await; + } if matches!( observation.kind, @@ -953,6 +978,71 @@ impl ControllerSet { } } +fn attempt_feedback(observation: &AttemptObservation) -> Option { + observation.attempt.as_ref()?; + if matches!( + observation.kind, + AttemptObservationKind::CompleteResponse( + crate::dispatch::lifecycle::contract::CompleteResponseFacts::Completed + | crate::dispatch::lifecycle::contract::CompleteResponseFacts::Incomplete, + ) + ) { + return None; + } + Some(AttemptFeedback::FailedBeforeFirstToken) +} + +fn stream_feedback( + summary: &crate::dispatch::lifecycle::stream::StreamSummary, + request_started_at: Instant, + attempt_started_at: Instant, +) -> AttemptFeedback { + let first_token_ms = attempt_first_token_ms( + summary.first_token_ms, + request_started_at, + attempt_started_at, + ); + match summary.terminal { + StreamTerminal::Completed { .. } => AttemptFeedback::Completed { first_token_ms }, + StreamTerminal::Incomplete { .. } => AttemptFeedback::Incomplete { first_token_ms }, + StreamTerminal::Cancelled | StreamTerminal::DownstreamClosed => AttemptFeedback::Cancelled, + StreamTerminal::Shutdown => AttemptFeedback::Shutdown, + StreamTerminal::Failed { .. } + | StreamTerminal::UpstreamClosed + | StreamTerminal::UpstreamError { .. } + | StreamTerminal::ProtocolError { .. } + | StreamTerminal::CaptureLimitExceeded => { + summary + .first_token_ms + .map_or(AttemptFeedback::FailedBeforeFirstToken, |_| { + AttemptFeedback::FailedAfterFirstToken { + first_token_ms: first_token_ms.unwrap_or(0), + } + }) + } + } +} + +fn attempt_first_token_ms( + request_first_token_ms: Option, + request_started_at: Instant, + attempt_started_at: Instant, +) -> Option { + let offset_ms = i64::try_from( + attempt_started_at + .saturating_duration_since(request_started_at) + .as_millis(), + ) + .ok()?; + request_first_token_ms + .and_then(|value| value.checked_sub(offset_ms)) + .and_then(|value| u64::try_from(value).ok()) +} + +fn nonnegative_u64(value: Option) -> Option { + value.and_then(|value| u64::try_from(value).ok()) +} + fn response_failure(observation: &AttemptObservation) -> Option<&ResponsesSseFailure> { match &observation.kind { AttemptObservationKind::CompleteResponse( diff --git a/backend/src/dispatch/controllers/telemetry/mod.rs b/backend/src/dispatch/controllers/telemetry/mod.rs index bf8638f31..d645ab659 100644 --- a/backend/src/dispatch/controllers/telemetry/mod.rs +++ b/backend/src/dispatch/controllers/telemetry/mod.rs @@ -2,6 +2,8 @@ mod events; +use std::time::{Duration, Instant}; + use serde_json::Value; use crate::{ @@ -149,13 +151,18 @@ impl TelemetryController { .effective_model .as_deref() .unwrap_or(exit.display_model); + let first_token_ms = request_first_token_ms( + exit.started_at, + exit.attempt.started_at(), + exit.response.first_token_ms, + ); let mut metadata = serde_json::json!({ "responseId": exit.response_id, "stream": false, "completed": exit.completed, "incomplete": !exit.completed, "transport": backend_transport_name(exit.response.transport), - "firstTokenMs": exit.response.first_token_ms, + "firstTokenMs": first_token_ms, "usage": exit.response.usage, "effectiveModel": effective_model, "modelsEtag": exit.response.response_metadata.models_etag.as_deref(), @@ -264,6 +271,22 @@ impl TelemetryController { } } +fn request_first_token_ms( + request_started_at: Instant, + attempt_started_at: Instant, + attempt_first_token_ms: Option, +) -> Option { + let attempt_first_token_ms = u64::try_from(attempt_first_token_ms?).ok()?; + let output_at = + attempt_started_at.checked_add(Duration::from_millis(attempt_first_token_ms))?; + i64::try_from( + output_at + .saturating_duration_since(request_started_at) + .as_millis(), + ) + .ok() +} + async fn record_success( context: &StreamContext<'_>, summary: &StreamSummary, diff --git a/backend/src/dispatch/lifecycle/attempt.rs b/backend/src/dispatch/lifecycle/attempt.rs index a733db143..eeff75de9 100644 --- a/backend/src/dispatch/lifecycle/attempt.rs +++ b/backend/src/dispatch/lifecycle/attempt.rs @@ -50,13 +50,8 @@ use super::{ }; pub(in crate::dispatch) enum AttemptMode { - Complete { - started_at: Instant, - tuple_schema: Option, - }, - Stream { - tuple_schema: Option, - }, + Complete { tuple_schema: Option }, + Stream { tuple_schema: Option }, } pub(in crate::dispatch) struct AttemptRunnerDependencies<'a> { @@ -288,15 +283,12 @@ impl<'a> AttemptRunner<'a> { }; match &self.mode { - AttemptMode::Complete { - started_at, - tuple_schema, - } => { + AttemptMode::Complete { tuple_schema } => { let result = create_response_with_account( upstream_context, &account_scoped_request, prepared_transport, - *started_at, + attempt.started_at(), ) .await; let attempt_request = account_scoped_request.into_request(); @@ -420,7 +412,7 @@ impl<'a> AttemptRunner<'a> { body, &mut decoder, attempt_request.stream_commit_policy, - self.dependencies.started_at, + attempt.started_at(), ) .await { diff --git a/backend/src/dispatch/lifecycle/trace.rs b/backend/src/dispatch/lifecycle/trace.rs index 7e61f0ec6..72abf86cf 100644 --- a/backend/src/dispatch/lifecycle/trace.rs +++ b/backend/src/dispatch/lifecycle/trace.rs @@ -1,5 +1,7 @@ //! Responses 请求级账号尝试轨迹。 +use std::time::Instant; + #[derive(Debug, Default)] pub(in crate::dispatch) struct ResponseDispatchTrace { next_attempt_index: i64, @@ -10,6 +12,7 @@ pub(in crate::dispatch) struct ResponseDispatchTrace { pub(in crate::dispatch) struct ResponseDispatchAttempt { index: i64, account_id: String, + started_at: Instant, } impl ResponseDispatchTrace { @@ -20,6 +23,7 @@ impl ResponseDispatchTrace { let attempt = ResponseDispatchAttempt { index: self.next_attempt_index, account_id: account_id.to_string(), + started_at: Instant::now(), }; self.next_attempt_index += 1; self.attempts.push(attempt.clone()); @@ -39,4 +43,8 @@ impl ResponseDispatchAttempt { pub(in crate::dispatch) fn account_id(&self) -> &str { &self.account_id } + + pub(in crate::dispatch) fn started_at(&self) -> Instant { + self.started_at + } } diff --git a/backend/src/dispatch/service.rs b/backend/src/dispatch/service.rs index 58f277359..4f2c79dcc 100644 --- a/backend/src/dispatch/service.rs +++ b/backend/src/dispatch/service.rs @@ -220,10 +220,7 @@ impl ResponseDispatchService { requested_model, started_at, }, - AttemptMode::Complete { - started_at, - tuple_schema, - }, + AttemptMode::Complete { tuple_schema }, request, controller_scope, candidates, diff --git a/backend/src/dispatch/stream/live.rs b/backend/src/dispatch/stream/live.rs index c897b2ff7..b20b1970d 100644 --- a/backend/src/dispatch/stream/live.rs +++ b/backend/src/dispatch/stream/live.rs @@ -304,7 +304,7 @@ fn update_first_token_ms( started_at: Instant, first_token_ms: &mut Option, ) { - if first_token_ms.is_none() && decoder.first_output_seen() { + if first_token_ms.is_none() && decoder.first_semantic_output_seen() { *first_token_ms = Some(elapsed_millis_i64(started_at).max(1)); } } diff --git a/backend/src/dispatch/transport/canonical.rs b/backend/src/dispatch/transport/canonical.rs index 8ca11b4ec..4d5aed61a 100644 --- a/backend/src/dispatch/transport/canonical.rs +++ b/backend/src/dispatch/transport/canonical.rs @@ -6,8 +6,8 @@ use serde_json::Value; use crate::upstream::openai::protocol::{ events::{TokenUsage, extract_usage}, responses::{ - CollectedResponse, ResponsesSseFailure, StreamCommitPolicy, - reconvert_responses_sse_event_tuple_values, + CollectedResponse, ResponseEventSignals, ResponsesSseFailure, StreamCommitPolicy, + reconvert_responses_sse_event_tuple_values, response_event_signals, }, sse::{ MAX_SSE_EVENT_BUFFER_BYTES, SseError, encode_sse_event, parse_sse_events, sse_frame_end, @@ -123,34 +123,8 @@ impl CanonicalResponseEvent { .or_else(|| extract_usage(&self.data)) } - pub(in crate::dispatch) fn has_first_output(&self) -> bool { - match self.event_type() { - Some( - "response.output_text.delta" - | "response.reasoning_summary_text.delta" - | "response.reasoning_text.delta" - | "response.function_call_arguments.delta" - | "response.custom_tool_call_input.delta", - ) => self - .data - .get("delta") - .and_then(Value::as_str) - .is_some_and(|value| !value.is_empty()), - Some("response.output_text.done") => self - .data - .get("text") - .and_then(Value::as_str) - .is_some_and(|value| !value.is_empty()), - Some("response.function_call_arguments.done") => self - .data - .get("arguments") - .and_then(Value::as_str) - .is_some_and(|value| !value.is_empty()), - Some("response.output_item.done") => { - self.data.get("item").is_some_and(Value::is_object) - } - _ => false, - } + pub(in crate::dispatch) fn signals(&self) -> ResponseEventSignals { + response_event_signals(self.event_type(), &self.data) } pub(in crate::dispatch) fn terminal(&self) -> Option { @@ -230,7 +204,7 @@ impl CanonicalStreamBatch { /// 同一上游 chunk 可能同时包含输出与终态，事件顺序不能被网络分块抹掉。 pub(in crate::dispatch) fn first_failure(&self) -> Option { for event in self.chunks.iter().flat_map(|chunk| chunk.events()) { - if event.has_first_output() { + if event.signals().semantic_output { return None; } if let Some(terminal) = event.terminal() { @@ -250,7 +224,7 @@ pub(in crate::dispatch) struct CanonicalStreamDecoder { tuple_schema: Option, pending: Vec, forwardable_frame_seen: bool, - first_output_seen: bool, + first_semantic_output_seen: bool, terminal: Option, transport_done_seen: bool, completed_usage: Option, @@ -264,7 +238,7 @@ impl CanonicalStreamDecoder { tuple_schema, pending: Vec::new(), forwardable_frame_seen: false, - first_output_seen: false, + first_semantic_output_seen: false, terminal: None, transport_done_seen: false, completed_usage: None, @@ -333,15 +307,15 @@ impl CanonicalStreamDecoder { self.transport_done_seen } - pub(in crate::dispatch) fn first_output_seen(&self) -> bool { - self.first_output_seen + pub(in crate::dispatch) fn first_semantic_output_seen(&self) -> bool { + self.first_semantic_output_seen } pub(in crate::dispatch) fn commit_boundary_reached(&self, policy: StreamCommitPolicy) -> bool { match policy { StreamCommitPolicy::FirstForwardableEvent => self.forwardable_frame_seen, StreamCommitPolicy::UntilOutputOrTerminal => { - self.first_output_seen || self.terminal.is_some() + self.first_semantic_output_seen || self.terminal.is_some() } } } @@ -406,7 +380,8 @@ impl CanonicalStreamDecoder { } fn observe(&mut self, event: &CanonicalResponseEvent) { - self.first_output_seen |= event.has_first_output(); + let signals = event.signals(); + self.first_semantic_output_seen |= signals.semantic_output; if let Some(response_id) = event.response_id() { self.last_response_id = Some(response_id.to_string()); } diff --git a/backend/src/fleet/pool/mod.rs b/backend/src/fleet/pool/mod.rs index 70292391b..fe4a97116 100644 --- a/backend/src/fleet/pool/mod.rs +++ b/backend/src/fleet/pool/mod.rs @@ -15,7 +15,7 @@ use thiserror::Error; use crate::fleet::quota::{quota_snapshot_limit_reached, quota_snapshot_reset_at}; use crate::fleet::refresh::{JwtExpiry, jwt_expiry}; use crate::fleet::scheduler::{ - AccountScheduler, ScoreWeights, + AccountScheduler, AttemptFeedback, ScoreWeights, candidates::{self, CandidateFilter, CandidateRequest}, }; use crate::fleet::store::AccountStore; @@ -223,6 +223,11 @@ impl AccountPoolService { StdDuration::from_millis(request_interval_ms); } + /// 回灌一个已经启动的上游 attempt，供 Smart 调度器更新运行时健康反馈。 + pub async fn report_attempt(&self, account_id: &str, feedback: AttemptFeedback) { + self.pool.lock().await.report_attempt(account_id, feedback); + } + /// 持续接收运行时设置并更新账号池参数。 pub async fn subscribe_settings( self: Arc, diff --git a/backend/src/fleet/pool/state.rs b/backend/src/fleet/pool/state.rs index 5169baf83..f99ab5d42 100644 --- a/backend/src/fleet/pool/state.rs +++ b/backend/src/fleet/pool/state.rs @@ -206,10 +206,13 @@ impl AccountPool { self.options = options; } - /// 回灌一次请求结果到调度器的运行时反馈（供 Smart 打分）。 - pub fn report_feedback(&self, account_id: &str, success: bool, first_token_ms: Option) { - self.scheduler - .report_feedback(account_id, success, first_token_ms); + /// 回灌一个已启动上游 attempt 的运行时反馈（供 Smart 打分）。 + pub fn report_attempt( + &self, + account_id: &str, + feedback: crate::fleet::scheduler::AttemptFeedback, + ) { + self.scheduler.report_attempt(account_id, feedback); } /// 更新账号状态。 diff --git a/backend/src/fleet/scheduler/feedback.rs b/backend/src/fleet/scheduler/feedback.rs index 8c002d022..0b572278a 100644 --- a/backend/src/fleet/scheduler/feedback.rs +++ b/backend/src/fleet/scheduler/feedback.rs @@ -1,7 +1,7 @@ -//! 账号运行时反馈:错误率与 TTFT 的 EWMA 统计。 +//! 账号运行时 feedback：首字前失败率、首字后中断率与 TTFT 的 EWMA 统计。 //! -//! 每个账号维护两个指数加权移动平均(EWMA):错误率与首个有效输出事件延迟(TTFT， -//! 毫秒)。请求结束后经 [`FeedbackStats::report`] 回灌，直接进 [`super::score`] +//! 每个账号维护三个指数加权移动平均(EWMA)。每个已启动 attempt 结束后经 +//! [`FeedbackStats::report_attempt`] 回灌，直接进 [`super::score`] //! 打分并可驱动 sticky escape。统计以 `AtomicU64`(存 `f64` 位模式)+ CAS 无锁更新, //! 对齐 sub2api 的设计,避免选择热路径上的锁与 DB 往返。 @@ -12,13 +12,32 @@ use std::sync::atomic::{AtomicU64, Ordering}; /// EWMA 平滑系数(新样本权重)。与 sub2api 一致取 0.2。 const EWMA_ALPHA: f64 = 0.2; +/// 单个上游 attempt 对账号调度器的反馈。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AttemptFeedback { + Completed { first_token_ms: Option }, + Incomplete { first_token_ms: Option }, + FailedBeforeFirstToken, + FailedAfterFirstToken { first_token_ms: u64 }, + Cancelled, + Shutdown, +} + +impl AttemptFeedback { + fn affects_account_health(self) -> bool { + !matches!(self, Self::Cancelled | Self::Shutdown) + } +} + /// 单账号的运行时 EWMA 统计。 /// /// `*_bits` 存 `f64` 的位模式;`NaN`(全 1 的哨兵)表示"尚无样本"。 #[derive(Debug)] struct AccountFeedback { - /// 错误率 EWMA，范围 `[0, 1]`。 - error_rate_bits: AtomicU64, + /// 首字前失败率 EWMA，范围 `[0, 1]`。 + pre_token_failure_rate_bits: AtomicU64, + /// 首字后中断率 EWMA，范围 `[0, 1]`。 + post_token_abort_rate_bits: AtomicU64, /// TTFT EWMA，单位毫秒。 ttft_ms_bits: AtomicU64, } @@ -29,16 +48,22 @@ const EMPTY_BITS: u64 = f64::NAN.to_bits(); impl Default for AccountFeedback { fn default() -> Self { Self { - error_rate_bits: AtomicU64::new(EMPTY_BITS), + pre_token_failure_rate_bits: AtomicU64::new(EMPTY_BITS), + post_token_abort_rate_bits: AtomicU64::new(EMPTY_BITS), ttft_ms_bits: AtomicU64::new(EMPTY_BITS), } } } impl AccountFeedback { - /// 读取当前错误率 EWMA;无样本时返回 `None`。 - fn error_rate(&self) -> Option { - load_sample(&self.error_rate_bits) + /// 读取当前首字前失败率 EWMA;无样本时返回 `None`。 + fn pre_token_failure_rate(&self) -> Option { + load_sample(&self.pre_token_failure_rate_bits) + } + + /// 读取当前首字后中断率 EWMA;无样本时返回 `None`。 + fn post_token_abort_rate(&self) -> Option { + load_sample(&self.post_token_abort_rate_bits) } /// 读取当前 TTFT EWMA(毫秒);无样本时返回 `None`。 @@ -46,12 +71,26 @@ impl AccountFeedback { load_sample(&self.ttft_ms_bits) } - /// 回灌一次请求结果。 - fn report(&self, success: bool, first_token_ms: Option) { - let error_sample = if success { 0.0 } else { 1.0 }; - update_ewma(&self.error_rate_bits, error_sample); - if let Some(ms) = first_token_ms { - update_ewma(&self.ttft_ms_bits, ms as f64); + /// 回灌一次已经开始的 attempt。 + fn report_attempt(&self, feedback: AttemptFeedback) { + match feedback { + AttemptFeedback::Completed { first_token_ms } + | AttemptFeedback::Incomplete { first_token_ms } => { + update_ewma(&self.pre_token_failure_rate_bits, 0.0); + update_ewma(&self.post_token_abort_rate_bits, 0.0); + if let Some(ms) = first_token_ms { + update_ewma(&self.ttft_ms_bits, ms as f64); + } + } + AttemptFeedback::FailedBeforeFirstToken => { + update_ewma(&self.pre_token_failure_rate_bits, 1.0); + } + AttemptFeedback::FailedAfterFirstToken { first_token_ms } => { + update_ewma(&self.pre_token_failure_rate_bits, 0.0); + update_ewma(&self.post_token_abort_rate_bits, 1.0); + update_ewma(&self.ttft_ms_bits, first_token_ms as f64); + } + AttemptFeedback::Cancelled | AttemptFeedback::Shutdown => {} } } } @@ -87,8 +126,10 @@ fn update_ewma(bits: &AtomicU64, sample: f64) { /// 账号 EWMA 反馈的读数快照,供打分使用。 #[derive(Debug, Clone, Copy, Default, PartialEq)] pub struct FeedbackSample { - /// 错误率 EWMA，范围 `[0, 1]`;无样本时为 `None`。 - pub error_rate: Option, + /// 首字前失败率 EWMA，范围 `[0, 1]`;无样本时为 `None`。 + pub pre_token_failure_rate: Option, + /// 首字后中断率 EWMA，范围 `[0, 1]`;无样本时为 `None`。 + pub post_token_abort_rate: Option, /// TTFT EWMA(毫秒);无样本时为 `None`。 pub ttft_ms: Option, } @@ -116,23 +157,27 @@ impl FeedbackStats { .unwrap_or_else(std::sync::PoisonError::into_inner); match accounts.get(account_id) { Some(feedback) => FeedbackSample { - error_rate: feedback.error_rate(), + pre_token_failure_rate: feedback.pre_token_failure_rate(), + post_token_abort_rate: feedback.post_token_abort_rate(), ttft_ms: feedback.ttft_ms(), }, None => FeedbackSample::default(), } } - /// 回灌一次请求结果:成功/失败 + 首个有效输出事件延迟(毫秒)。 - pub fn report(&self, account_id: &str, success: bool, first_token_ms: Option) { + /// 回灌一个已启动上游 attempt 的结果。 + pub fn report_attempt(&self, account_id: &str, sample: AttemptFeedback) { + if !sample.affects_account_health() { + return; + } // 快路径:账号已存在时只需读锁 + 原子更新。 { let accounts = self .accounts .read() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(feedback) = accounts.get(account_id) { - feedback.report(success, first_token_ms); + if let Some(account_feedback) = accounts.get(account_id) { + account_feedback.report_attempt(sample); return; } } @@ -144,7 +189,7 @@ impl FeedbackStats { accounts .entry(account_id.to_string()) .or_default() - .report(success, first_token_ms); + .report_attempt(sample); } /// 移除指定账号的反馈记录(账号被删除时调用)。 diff --git a/backend/src/fleet/scheduler/mod.rs b/backend/src/fleet/scheduler/mod.rs index ae064cc90..fcf67511d 100644 --- a/backend/src/fleet/scheduler/mod.rs +++ b/backend/src/fleet/scheduler/mod.rs @@ -23,7 +23,7 @@ use crate::fleet::account::Account; use crate::infra::time::parse_rfc3339_utc; pub use candidates::{CandidateFilter, CandidateRequest}; -pub use feedback::{FeedbackSample, FeedbackStats}; +pub use feedback::{AttemptFeedback, FeedbackSample, FeedbackStats}; pub use strategy::smart::rank_candidates; pub use strategy::{ScoreBreakdown, ScoreWeights}; @@ -69,7 +69,7 @@ pub struct SelectionInput<'a> { /// 账号调度器：账号选择的对外门面。 /// /// 持有轮转游标（并列破并列用）、打分权重（Smart 用）与 EWMA 反馈存储。选择热路径 -/// 只读反馈与权重，请求结束后经 [`AccountScheduler::report_feedback`] 回灌。游标用 +/// 只读反馈与权重，请求结束后经 [`AccountScheduler::report_attempt`] 回灌。游标用 /// [`Mutex`] 保护（选择需可变，但 `AccountScheduler` 以共享引用参与并发请求）。 #[derive(Debug)] pub struct AccountScheduler { @@ -150,9 +150,9 @@ impl AccountScheduler { } } - /// 回灌一次请求结果到运行时 EWMA 反馈（错误率 / TTFT），供 Smart 打分。 - pub fn report_feedback(&self, account_id: &str, success: bool, first_token_ms: Option) { - self.feedback.report(account_id, success, first_token_ms); + /// 回灌一个已启动上游 attempt 的运行时反馈，供 Smart 打分。 + pub fn report_attempt(&self, account_id: &str, feedback: AttemptFeedback) { + self.feedback.report_attempt(account_id, feedback); } /// 移除指定账号的运行时反馈（账号被删除时调用）。 diff --git a/backend/src/fleet/scheduler/strategy/smart.rs b/backend/src/fleet/scheduler/strategy/smart.rs index 455ef1abc..658b1514c 100644 --- a/backend/src/fleet/scheduler/strategy/smart.rs +++ b/backend/src/fleet/scheduler/strategy/smart.rs @@ -284,7 +284,7 @@ fn score_account( ); // 错误率 EWMA：本身即 [0,1]，越低越健康。无样本时视为 0（满分健康）。 - let error_rate = 1.0 - sample.error_rate.unwrap_or(0.0).clamp(0.0, 1.0); + let error_rate = 1.0 - sample.pre_token_failure_rate.unwrap_or(0.0).clamp(0.0, 1.0); // TTFT：延迟越低越好。无样本的账号视为中性满分，避免冷启动被压。 let ttft = match sample.ttft_ms { diff --git a/backend/src/telemetry/usage/insights.rs b/backend/src/telemetry/usage/insights.rs index 3309f1526..b7f2330a9 100644 --- a/backend/src/telemetry/usage/insights.rs +++ b/backend/src/telemetry/usage/insights.rs @@ -264,11 +264,11 @@ struct UsagePerformanceInsights { latency_p50_ms: Option, latency_p95_ms: Option, latency_p99_ms: Option, - ttft_p50_ms: Option, - ttft_p95_ms: Option, - ttft_p99_ms: Option, + first_token_p50_ms: Option, + first_token_p95_ms: Option, + first_token_p99_ms: Option, latency_coverage: f64, - ttft_coverage: f64, + first_token_coverage: f64, points: Vec, } @@ -280,9 +280,9 @@ struct UsagePerformancePoint { latency_p50_ms: Option, latency_p95_ms: Option, latency_p99_ms: Option, - ttft_p50_ms: Option, - ttft_p95_ms: Option, - ttft_p99_ms: Option, + first_token_p50_ms: Option, + first_token_p95_ms: Option, + first_token_p99_ms: Option, } #[derive(Debug, Serialize)] @@ -565,13 +565,13 @@ async fn load_performance( select count(*)::bigint as request_count, count(latency_ms)::bigint as latency_count, - count(first_token_ms)::bigint as ttft_count, + count(first_token_ms)::bigint as first_token_count, percentile_cont(0.50) within group (order by latency_ms) as latency_p50_ms, percentile_cont(0.95) within group (order by latency_ms) as latency_p95_ms, percentile_cont(0.99) within group (order by latency_ms) as latency_p99_ms, - percentile_cont(0.50) within group (order by first_token_ms) as ttft_p50_ms, - percentile_cont(0.95) within group (order by first_token_ms) as ttft_p95_ms, - percentile_cont(0.99) within group (order by first_token_ms) as ttft_p99_ms + percentile_cont(0.50) within group (order by first_token_ms) as first_token_p50_ms, + percentile_cont(0.95) within group (order by first_token_ms) as first_token_p95_ms, + percentile_cont(0.99) within group (order by first_token_ms) as first_token_p99_ms from usage_records where created_at >= $1 and created_at < $2 "#, @@ -586,9 +586,9 @@ where created_at >= $1 and created_at < $2 percentile_cont(0.50) within group (order by latency_ms) as latency_p50_ms,\n\ percentile_cont(0.95) within group (order by latency_ms) as latency_p95_ms,\n\ percentile_cont(0.99) within group (order by latency_ms) as latency_p99_ms,\n\ - percentile_cont(0.50) within group (order by first_token_ms) as ttft_p50_ms,\n\ - percentile_cont(0.95) within group (order by first_token_ms) as ttft_p95_ms,\n\ - percentile_cont(0.99) within group (order by first_token_ms) as ttft_p99_ms\n\ + percentile_cont(0.50) within group (order by first_token_ms) as first_token_p50_ms,\n\ + percentile_cont(0.95) within group (order by first_token_ms) as first_token_p95_ms,\n\ + percentile_cont(0.99) within group (order by first_token_ms) as first_token_p99_ms\n\ from usage_records\n\ where created_at >= $1 and created_at < $2\n\ group by bucket_start order by bucket_start" @@ -610,9 +610,9 @@ where created_at >= $1 and created_at < $2 latency_p50_ms: row.get("latency_p50_ms"), latency_p95_ms: row.get("latency_p95_ms"), latency_p99_ms: row.get("latency_p99_ms"), - ttft_p50_ms: row.get("ttft_p50_ms"), - ttft_p95_ms: row.get("ttft_p95_ms"), - ttft_p99_ms: row.get("ttft_p99_ms"), + first_token_p50_ms: row.get("first_token_p50_ms"), + first_token_p95_ms: row.get("first_token_p95_ms"), + first_token_p99_ms: row.get("first_token_p99_ms"), }, ); } @@ -633,11 +633,11 @@ where created_at >= $1 and created_at < $2 latency_p50_ms: summary.get("latency_p50_ms"), latency_p95_ms: summary.get("latency_p95_ms"), latency_p99_ms: summary.get("latency_p99_ms"), - ttft_p50_ms: summary.get("ttft_p50_ms"), - ttft_p95_ms: summary.get("ttft_p95_ms"), - ttft_p99_ms: summary.get("ttft_p99_ms"), + first_token_p50_ms: summary.get("first_token_p50_ms"), + first_token_p95_ms: summary.get("first_token_p95_ms"), + first_token_p99_ms: summary.get("first_token_p99_ms"), latency_coverage: rate(nonnegative(summary.get("latency_count")), request_count), - ttft_coverage: rate(nonnegative(summary.get("ttft_count")), request_count), + first_token_coverage: rate(nonnegative(summary.get("first_token_count")), request_count), points, }) } diff --git a/backend/src/upstream/openai/failure.rs b/backend/src/upstream/openai/failure.rs index b87f9b39b..9462f3ed6 100644 --- a/backend/src/upstream/openai/failure.rs +++ b/backend/src/upstream/openai/failure.rs @@ -113,7 +113,7 @@ pub fn upstream_failure_facts(error: &CodexClientError) -> UpstreamFailureFacts | CodexWebSocketExchangeError::OriginHalfOpenBusy | CodexWebSocketExchangeError::SharedConnectFailed | CodexWebSocketExchangeError::ClosedBeforeTerminal - | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { .. }, + | CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstToken { .. }, ) => simple_facts(UpstreamFailureKind::WebSocketTransport, error), CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous { .. }) => { simple_facts(UpstreamFailureKind::PostSendAmbiguous, error) diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 3904247f5..060e2afd1 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -195,69 +195,120 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir } } -/// 判断已收到的 Responses SSE 内容是否包含首个完整的真实输出事件。 -pub fn response_body_has_first_output(body_bytes: &[u8]) -> bool { +/// 单个 Responses 事件对计时系统提供的稳定语义信号。 +/// +/// `protocol_progress` 只说明上游仍在工作，不能替代首字；`semantic_output` +/// 表示模型已经产生客户端可消费的内容。 +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ResponseEventSignals { + pub protocol_progress: bool, + pub semantic_output: bool, +} + +/// 从已解析的 Responses 事件提取计时语义。 +/// +/// 生命周期帧和结构帧不会被当作输出。`response.output_item.done` 与终态帧只有 +/// 实际携带文本、工具参数、推理或图片结果时才算语义输出。 +pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> ResponseEventSignals { + let protocol_progress = !matches!(event_type, Some("response.failed" | "error")); + let semantic_output = match event_type { + Some("response.output_text.delta") => non_empty_string(value.get("delta")), + Some("response.output_text.done") => non_empty_string(value.get("text")), + Some("response.reasoning_summary_text.delta" | "response.reasoning_text.delta") => { + non_empty_string(value.get("delta")) + } + Some( + "response.function_call_arguments.delta" | "response.custom_tool_call_input.delta", + ) => non_empty_string(value.get("delta")), + Some("response.function_call_arguments.done") => non_empty_string(value.get("arguments")), + Some("response.image_generation_call.partial_image") => non_empty_string( + value + .get("partial_image_b64") + .or_else(|| value.get("partial_image")), + ), + Some("response.output_item.done") => value + .get("item") + .is_some_and(output_item_has_semantic_output), + Some("response.content_part.done") => value + .get("part") + .is_some_and(output_item_has_semantic_output), + Some("response.completed" | "response.incomplete") => value + .pointer("/response/output") + .and_then(Value::as_array) + .is_some_and(|items| output_items_have_semantic_output(items)), + _ => false, + }; + ResponseEventSignals { + protocol_progress, + semantic_output, + } +} + +/// 判断已收到的 Responses SSE 内容是否包含首个完整的语义输出事件。 +pub fn response_body_has_semantic_output(body_bytes: &[u8]) -> bool { let body = String::from_utf8_lossy(body_bytes); let Some(complete_body) = complete_sse_body_prefix(&body) else { return false; }; - parse_sse_events(complete_body) - .is_ok_and(|events| events.iter().any(response_sse_event_has_first_output)) + parse_sse_events(complete_body).is_ok_and(|events| { + events.iter().any(|event| { + let Ok(value) = serde_json::from_str::(&event.data) else { + return false; + }; + let event_type = event.event.as_deref().or_else(|| { + value + .get("type") + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) + }); + response_event_signals(event_type, &value).semantic_output + }) + }) } -/// 已收到首个有效 Responses SSE 输出事件时记录耗时。 -pub fn update_first_response_output_ms( +/// 已收到首个语义输出事件时记录耗时。 +pub fn update_first_response_semantic_output_ms( started_at: Instant, body_bytes: &[u8], first_token_ms: &mut Option, ) { - if first_token_ms.is_none() && response_body_has_first_output(body_bytes) { + if first_token_ms.is_none() && response_body_has_semantic_output(body_bytes) { *first_token_ms = Some(elapsed_millis_i64(started_at).max(1)); } } -fn response_sse_event_has_first_output(event: &SseEvent) -> bool { - let data = event.data.trim(); - if data.is_empty() || data == "[DONE]" { - return false; - } +fn output_items_have_semantic_output(items: &[Value]) -> bool { + items.iter().any(output_item_has_semantic_output) +} - let value = serde_json::from_str::(data).ok(); - let event_type = event.event.as_deref().or_else(|| { - value - .as_ref() - .and_then(|value| value.get("type")) - .and_then(Value::as_str) - }); - - match event_type { - Some("response.output_text.delta") - | Some("response.reasoning_summary_text.delta") - | Some("response.reasoning_text.delta") - | Some("response.function_call_arguments.delta") - | Some("response.custom_tool_call_input.delta") => value - .as_ref() - .and_then(|value| value.get("delta")) - .and_then(Value::as_str) - .is_some_and(|delta| !delta.is_empty()), - Some("response.output_text.done") => value - .as_ref() - .and_then(|value| value.get("text")) - .and_then(Value::as_str) - .is_some_and(|text| !text.is_empty()), - Some("response.function_call_arguments.done") => value - .as_ref() - .and_then(|value| value.get("arguments")) - .and_then(Value::as_str) - .is_some_and(|arguments| !arguments.is_empty()), - Some("response.output_item.added" | "response.output_item.done") => value - .as_ref() - .and_then(|value| value.get("item")) - .is_some_and(Value::is_object), - _ => false, +fn output_item_has_semantic_output(item: &Value) -> bool { + match item.get("type").and_then(Value::as_str) { + Some("output_text" | "text") => non_empty_string(item.get("text")), + Some("reasoning") => { + non_empty_string(item.get("text")) || non_empty_string(item.get("summary")) + } + Some("function_call") => non_empty_string(item.get("arguments")), + Some("custom_tool_call") => non_empty_string(item.get("input")), + Some("image_generation_call") => { + non_empty_string(item.get("result")) || non_empty_string(item.get("image_base64")) + } + Some("message") => item + .get("content") + .and_then(Value::as_array) + .is_some_and(|items| output_items_have_semantic_output(items)), + _ => item + .get("content") + .and_then(Value::as_array) + .is_some_and(|items| output_items_have_semantic_output(items)), } } +fn non_empty_string(value: Option<&Value>) -> bool { + value + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()) +} + fn complete_sse_body_prefix(body: &str) -> Option<&str> { let lf_end = body.rfind("\n\n").map(|index| index + 2); let crlf_end = body.rfind("\r\n\r\n").map(|index| index + 4); diff --git a/backend/src/upstream/openai/transport/response_meta.rs b/backend/src/upstream/openai/transport/response_meta.rs index 765ea7e34..b18627fcc 100644 --- a/backend/src/upstream/openai/transport/response_meta.rs +++ b/backend/src/upstream/openai/transport/response_meta.rs @@ -5,7 +5,7 @@ use std::time::Instant; use reqwest::header::{HeaderMap, SET_COOKIE}; use crate::upstream::openai::protocol::{ - events::is_rate_limit_header_name, responses::update_first_response_output_ms, + events::is_rate_limit_header_name, responses::update_first_response_semantic_output_ms, }; use super::{client::CodexResponseMetadata, diagnostics::CodexUpstreamDiagnostics}; @@ -111,5 +111,5 @@ pub(super) fn update_first_token_ms( body_bytes: &[u8], first_token_ms: &mut Option, ) { - update_first_response_output_ms(started_at, body_bytes, first_token_ms); + update_first_response_semantic_output_ms(started_at, body_bytes, first_token_ms); } diff --git a/backend/src/upstream/openai/transport/websocket.rs b/backend/src/upstream/openai/transport/websocket.rs index eb519158f..50eff519d 100644 --- a/backend/src/upstream/openai/transport/websocket.rs +++ b/backend/src/upstream/openai/transport/websocket.rs @@ -26,7 +26,7 @@ use tungstenite::{ use crate::infra::time::china_filename_timestamp_millis; use crate::upstream::openai::protocol::events::{self, TokenUsage}; use crate::upstream::openai::protocol::responses::{ - CodexResponsesRequest, TransportRequirement, response_body_has_first_output, + CodexResponsesRequest, TransportRequirement, response_body_has_semantic_output, transport_requirement, }; use crate::upstream::openai::protocol::sse::SseError; diff --git a/backend/src/upstream/openai/transport/websocket_frames.rs b/backend/src/upstream/openai/transport/websocket_frames.rs index 5ce3a3800..371528d6d 100644 --- a/backend/src/upstream/openai/transport/websocket_frames.rs +++ b/backend/src/upstream/openai/transport/websocket_frames.rs @@ -146,7 +146,7 @@ pub enum CodexWebSocketExchangeError { UnexpectedBinaryEvent, /// 复用的池连接在收到首个上游事件前失效。 #[error("reused websocket connection died before first upstream event: {message}")] - ReusedConnectionDiedBeforeFirstOutput { + ReusedConnectionDiedBeforeFirstToken { /// 底层失效原因。 message: String, }, @@ -282,7 +282,7 @@ pub(super) async fn collect_websocket_response( if !saw_upstream_activity => { if reused_connection { - return Err(reused_connection_died_before_first_output( + return Err(reused_connection_died_before_first_token( &CodexWebSocketExchangeError::InitialEventTimeout { timeout }, )); } @@ -304,7 +304,7 @@ pub(super) async fn collect_websocket_response( Message::Binary(_) => return Err(CodexWebSocketExchangeError::UnexpectedBinaryEvent), Message::Close(_) if reused_connection && !saw_upstream_activity => { return Err( - CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { + CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstToken { message: "websocket closed".to_string(), }, ); @@ -340,9 +340,9 @@ pub(super) async fn collect_websocket_response( _ => None, }; let forwarded = if let Some(frame) = websocket_event_to_sse_frame(&raw) { - let has_first_output = response_body_has_first_output(frame.as_bytes()); + let has_semantic_output = response_body_has_semantic_output(frame.as_bytes()); body.push_str(&frame); - if has_first_output { + if has_semantic_output { response_meta::update_first_token_ms( started_at, body.as_bytes(), @@ -374,7 +374,7 @@ pub(super) async fn collect_websocket_response( if reused_connection && !saw_upstream_activity { return Err( - CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { + CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstToken { message: "websocket closed before terminal event".to_string(), }, ); @@ -388,16 +388,16 @@ fn reused_stream_receive_error(error: CodexWebSocketExchangeError) -> CodexWebSo CodexWebSocketExchangeError::ClosedBeforeTerminal | CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } | CodexWebSocketExchangeError::Transport(_) => { - reused_connection_died_before_first_output(&error) + reused_connection_died_before_first_token(&error) } error => error, } } -fn reused_connection_died_before_first_output( +fn reused_connection_died_before_first_token( error: &CodexWebSocketExchangeError, ) -> CodexWebSocketExchangeError { - CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { + CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstToken { message: error.to_string(), } } diff --git a/backend/tests/dispatch/service/usage_logging.rs b/backend/tests/dispatch/service/usage_logging.rs index 3fe3b5c46..168daff1d 100644 --- a/backend/tests/dispatch/service/usage_logging.rs +++ b/backend/tests/dispatch/service/usage_logging.rs @@ -687,7 +687,7 @@ async fn responses_stream_should_record_usage_record_after_completed_stream() { assert_eq!(event.output_tokens, Some(5)); assert!( event.first_token_ms.is_some_and(|value| value > 0), - "stream usage metadata should include first output event latency: {metadata:?}", + "stream usage metadata should include first-token latency: {metadata:?}", ); let first_token_bucket: (i64, i64) = sqlx::query_as( "select first_token_latency_sum, first_token_latency_count from request_time_buckets limit 1", diff --git a/backend/tests/fleet/scheduler/feedback.rs b/backend/tests/fleet/scheduler/feedback.rs index 166a36b62..f31078671 100644 --- a/backend/tests/fleet/scheduler/feedback.rs +++ b/backend/tests/fleet/scheduler/feedback.rs @@ -1,4 +1,4 @@ -use codex_proxy_rs::fleet::scheduler::{FeedbackSample, FeedbackStats}; +use codex_proxy_rs::fleet::scheduler::{AttemptFeedback, FeedbackSample, FeedbackStats}; #[test] fn sample_of_unknown_account_is_empty() { @@ -7,38 +7,48 @@ fn sample_of_unknown_account_is_empty() { } #[test] -fn first_sample_lands_directly() { +fn failed_before_first_token_is_recorded_as_a_full_pre_token_failure() { let stats = FeedbackStats::new(); - stats.report("acct", false, Some(400)); + stats.report_attempt("acct", AttemptFeedback::FailedBeforeFirstToken); let sample = stats.sample("acct"); - assert_eq!(sample.error_rate, Some(1.0)); - assert_eq!(sample.ttft_ms, Some(400.0)); -} - -#[test] -fn subsequent_samples_blend_by_alpha() { - let stats = FeedbackStats::new(); - stats.report("acct", false, Some(1000)); // error_rate=1.0, ttft=1000 - stats.report("acct", true, Some(0)); // error_rate=0.8, ttft=800 - let sample = stats.sample("acct"); - assert!((sample.error_rate.unwrap() - 0.8).abs() < 1e-9); - assert!((sample.ttft_ms.unwrap() - 800.0).abs() < 1e-9); -} - -#[test] -fn success_only_reports_decay_error_rate() { - let stats = FeedbackStats::new(); - stats.report("acct", false, None); // error_rate=1.0, ttft untouched - stats.report("acct", true, None); // error_rate=0.8 - let sample = stats.sample("acct"); - assert!((sample.error_rate.unwrap() - 0.8).abs() < 1e-9); + assert_eq!(sample.pre_token_failure_rate, Some(1.0)); assert_eq!(sample.ttft_ms, None); } +#[test] +fn successful_first_token_decays_pre_token_failure_and_updates_ttft() { + let stats = FeedbackStats::new(); + stats.report_attempt("acct", AttemptFeedback::FailedBeforeFirstToken); + stats.report_attempt( + "acct", + AttemptFeedback::Completed { + first_token_ms: Some(1000), + }, + ); + let sample = stats.sample("acct"); + assert!((sample.pre_token_failure_rate.unwrap() - 0.8).abs() < 1e-9); + assert_eq!(sample.ttft_ms, Some(1000.0)); +} + +#[test] +fn failed_after_first_token_tracks_abort_without_penalizing_pre_token_failure() { + let stats = FeedbackStats::new(); + stats.report_attempt( + "acct", + AttemptFeedback::FailedAfterFirstToken { + first_token_ms: 200, + }, + ); + let sample = stats.sample("acct"); + assert_eq!(sample.pre_token_failure_rate, Some(0.0)); + assert_eq!(sample.post_token_abort_rate, Some(1.0)); + assert_eq!(sample.ttft_ms, Some(200.0)); +} + #[test] fn remove_clears_account() { let stats = FeedbackStats::new(); - stats.report("acct", false, Some(100)); + stats.report_attempt("acct", AttemptFeedback::FailedBeforeFirstToken); stats.remove("acct"); assert_eq!(stats.sample("acct"), FeedbackSample::default()); } diff --git a/backend/tests/fleet/scheduler/score.rs b/backend/tests/fleet/scheduler/score.rs index a6a06fc3e..03f32796f 100644 --- a/backend/tests/fleet/scheduler/score.rs +++ b/backend/tests/fleet/scheduler/score.rs @@ -1,6 +1,8 @@ use chrono::Utc; use codex_proxy_rs::fleet::account::AccountStatus; -use codex_proxy_rs::fleet::scheduler::{FeedbackStats, ScoreWeights, rank_candidates}; +use codex_proxy_rs::fleet::scheduler::{ + AttemptFeedback, FeedbackStats, ScoreWeights, rank_candidates, +}; use crate::support::accounts::test_account; @@ -32,7 +34,7 @@ fn high_error_rate_account_is_penalized() { let feedback = FeedbackStats::new(); // flaky 连续失败，错误率 EWMA 拉高。 for _ in 0..5 { - feedback.report("flaky", false, None); + feedback.report_attempt("flaky", AttemptFeedback::FailedBeforeFirstToken); } let slot_count = |_: &str| 0; diff --git a/backend/tests/support/storage.rs b/backend/tests/support/storage.rs index acff03f01..bd0d520cb 100644 --- a/backend/tests/support/storage.rs +++ b/backend/tests/support/storage.rs @@ -1,4 +1,4 @@ -use std::{env, sync::OnceLock, thread}; +use std::{collections::BTreeMap, env, sync::OnceLock, thread}; use chrono::{DateTime, Utc}; use codex_proxy_rs::{ @@ -20,9 +20,13 @@ use uuid::Uuid; const DEFAULT_TEST_DATABASE_URL: &str = "postgres://codex_proxy:codex_proxy@127.0.0.1:5432/codex_proxy"; const DEFAULT_TEST_REDIS_URL: &str = "redis://127.0.0.1:6379"; +const REDIS_TEST_PREFIX: &str = "cpr:test:"; +const REDIS_TEST_RUN_MARKER_PREFIX: &str = "cpr:test:run:"; +const REDIS_TEST_RUN_TTL_SECONDS: u64 = 60 * 60; static DATABASE_INIT_LIMIT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(2); static REDIS_TEST_CLEANUP: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new(); static REDIS_TEST_RUN_ID: OnceLock = OnceLock::new(); +static REDIS_TEST_RUN_REGISTRATION: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new(); static LOCAL_BOOTSTRAP_CONFIG: OnceLock> = OnceLock::new(); pub(crate) struct TestDatabaseGuard { @@ -125,6 +129,11 @@ async fn drop_test_database(base_url: &str, database_name: &str) -> Result<(), s pub(crate) async fn create_test_redis(label: &str) -> RedisConnection { let redis_url = test_redis_url(); + let run_id = REDIS_TEST_RUN_ID.get_or_init(|| Uuid::new_v4().simple().to_string()); + REDIS_TEST_RUN_REGISTRATION + .get_or_try_init(|| register_test_redis_run(&redis_url, run_id)) + .await + .unwrap_or_else(|error| panic!("register CPR test Redis run: {error}")); REDIS_TEST_CLEANUP .get_or_init(|| async { cleanup_stale_test_redis_keys(&redis_url) @@ -133,37 +142,89 @@ pub(crate) async fn create_test_redis(label: &str) -> RedisConnection { }) .await; let label = sanitize_label(label); - let run_id = REDIS_TEST_RUN_ID.get_or_init(|| Uuid::new_v4().simple().to_string()); - let prefix = format!("cpr:test:{run_id}:{label}:{}", Uuid::new_v4().simple()); + let prefix = format!( + "{REDIS_TEST_PREFIX}{run_id}:{label}:{}", + Uuid::new_v4().simple() + ); RedisConnection::connect(&redis_url, prefix) .await .unwrap_or_else(|error| panic!("connect CPR_TEST_REDIS_URL: {error}")) } +async fn register_test_redis_run(redis_url: &str, run_id: &str) -> redis::RedisResult<()> { + let client = redis::Client::open(redis_url)?; + let mut connection = client.get_connection_manager().await?; + let _: () = redis::cmd("SET") + .arg(test_redis_run_marker_key(run_id)) + .arg("1") + .arg("EX") + .arg(REDIS_TEST_RUN_TTL_SECONDS) + .query_async(&mut connection) + .await?; + Ok(()) +} + async fn cleanup_stale_test_redis_keys(redis_url: &str) -> redis::RedisResult<()> { let client = redis::Client::open(redis_url)?; let mut connection = client.get_connection_manager().await?; + let mut keys_by_run = BTreeMap::>::new(); let mut cursor = 0u64; loop { let (next_cursor, keys): (u64, Vec) = redis::cmd("SCAN") .arg(cursor) .arg("MATCH") - .arg("cpr:test:*") + .arg(format!("{REDIS_TEST_PREFIX}*")) .arg("COUNT") .arg(500) .query_async(&mut connection) .await?; - if !keys.is_empty() { - let _: usize = redis::cmd("UNLINK") - .arg(keys) - .query_async(&mut connection) - .await?; + for key in keys { + if let Some(run_id) = test_redis_run_id_from_key(&key) { + keys_by_run.entry(run_id.to_string()).or_default().push(key); + } } if next_cursor == 0 { - return Ok(()); + break; } cursor = next_cursor; } + + for (run_id, keys) in keys_by_run { + let active: i64 = redis::cmd("EXISTS") + .arg(test_redis_run_marker_key(&run_id)) + .query_async(&mut connection) + .await?; + if active > 0 { + continue; + } + unlink_test_redis_keys(&mut connection, &keys).await?; + } + + Ok(()) +} + +fn test_redis_run_marker_key(run_id: &str) -> String { + format!("{REDIS_TEST_RUN_MARKER_PREFIX}{run_id}") +} + +fn test_redis_run_id_from_key(key: &str) -> Option<&str> { + let suffix = key.strip_prefix(REDIS_TEST_PREFIX)?; + let (run_id, _) = suffix.split_once(':')?; + (run_id != "run" && !run_id.is_empty()).then_some(run_id) +} + +async fn unlink_test_redis_keys( + connection: &mut redis::aio::ConnectionManager, + keys: &[String], +) -> redis::RedisResult<()> { + for keys in keys.chunks(500) { + let mut command = redis::cmd("UNLINK"); + for key in keys { + command.arg(key); + } + let _: usize = command.query_async(connection).await?; + } + Ok(()) } pub(crate) fn background_task_stores(pool: PgPool, redis: RedisConnection) -> BackgroundTaskStores { @@ -231,6 +292,16 @@ fn test_database_name(label: &str) -> String { ) } +#[test] +fn redis_test_run_id_from_key_ignores_markers_and_invalid_keys() { + assert_eq!( + test_redis_run_id_from_key("cpr:test:active-run:affinity:key"), + Some("active-run") + ); + assert_eq!(test_redis_run_id_from_key("cpr:test:run:active-run"), None); + assert_eq!(test_redis_run_id_from_key("cpr:other:active-run:key"), None); +} + fn sanitize_label(label: &str) -> String { let label = label .chars() diff --git a/backend/tests/upstream/openai/protocol/mod.rs b/backend/tests/upstream/openai/protocol/mod.rs index f5b070ee7..d87db098e 100644 --- a/backend/tests/upstream/openai/protocol/mod.rs +++ b/backend/tests/upstream/openai/protocol/mod.rs @@ -6,7 +6,8 @@ use codex_proxy_rs::upstream::openai::protocol::events::{ }; use codex_proxy_rs::upstream::openai::protocol::responses::{ CodexResponsesRequest, PreviousResponseScope, TransportRequirement, - response_body_has_first_output, response_sse_event_is_terminal, transport_requirement, + response_body_has_semantic_output, response_event_signals, response_sse_event_is_terminal, + transport_requirement, }; use codex_proxy_rs::upstream::openai::protocol::sse::{ SseEventDecoder, parse_sse_events, response_failed_sse_event, diff --git a/backend/tests/upstream/openai/protocol/sse.rs b/backend/tests/upstream/openai/protocol/sse.rs index b747d04ff..abbfa6d09 100644 --- a/backend/tests/upstream/openai/protocol/sse.rs +++ b/backend/tests/upstream/openai/protocol/sse.rs @@ -1,76 +1,112 @@ use super::*; #[test] -fn first_output_detection_should_match_text_delta() { +fn first_token_detection_should_match_text_delta() { let body = include_bytes!("../../../fixtures/responses/http_sse/text_delta_hello.sse"); - assert!(response_body_has_first_output(body)); + assert!(response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_match_reasoning_delta() { +fn first_token_detection_should_match_reasoning_delta() { let body = include_bytes!("../../../fixtures/responses/http_sse/reasoning_delta.sse"); - assert!(response_body_has_first_output(body)); + assert!(response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_match_function_call_arguments_delta() { +fn first_token_detection_should_match_function_call_arguments_delta() { let body = include_bytes!("../../../fixtures/responses/http_sse/function_call_arguments_delta.sse"); - assert!(response_body_has_first_output(body)); + assert!(response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_match_type_when_event_name_is_missing() { +fn first_token_detection_should_match_type_when_event_name_is_missing() { let body = include_bytes!("../../../fixtures/responses/http_sse/data_type_text_delta.sse"); - assert!(response_body_has_first_output(body)); + assert!(response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_ignore_response_created() { +fn first_token_detection_should_ignore_response_created() { let body = include_bytes!("../../../fixtures/responses/http_sse/response_created.sse"); - assert!(!response_body_has_first_output(body)); + assert!(!response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_ignore_metadata_and_rate_limit_events() { +fn response_event_signals_should_not_count_structural_output_item_added() { + let signals = response_event_signals( + Some("response.output_item.added"), + &json!({"item": {"type": "message", "content": []}}), + ); + + assert!(signals.protocol_progress); + assert!(!signals.semantic_output); +} + +#[test] +fn response_event_signals_should_count_text_inside_completed_output_item() { + let signals = response_event_signals( + Some("response.output_item.done"), + &json!({ + "item": { + "type": "message", + "content": [{"type": "output_text", "text": "hello"}] + } + }), + ); + + assert!(signals.semantic_output); +} + +#[test] +fn response_event_signals_should_count_reasoning_as_semantic_output() { + let signals = response_event_signals( + Some("response.reasoning_text.delta"), + &json!({"delta": "thinking"}), + ); + + assert!(signals.semantic_output); +} + +#[test] +fn first_token_detection_should_ignore_metadata_and_rate_limit_events() { let body = include_bytes!( "../../../fixtures/responses/http_sse/metadata_rate_limits_no_first_output.sse" ); - assert!(!response_body_has_first_output(body)); + assert!(!response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_ignore_empty_delta() { +fn first_token_detection_should_ignore_empty_delta() { let body = include_bytes!("../../../fixtures/responses/http_sse/empty_text_delta.sse"); - assert!(!response_body_has_first_output(body)); + assert!(!response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_ignore_done_frame() { +fn first_token_detection_should_ignore_done_frame() { let body = include_bytes!("../../../fixtures/responses/http_sse/done_only.sse"); - assert!(!response_body_has_first_output(body)); + assert!(!response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_ignore_incomplete_frames() { +fn first_token_detection_should_ignore_incomplete_frames() { let body = include_bytes!("../../../fixtures/responses/http_sse/incomplete_text_delta.sse"); - assert!(!response_body_has_first_output(body)); + assert!(!response_body_has_semantic_output(body)); } #[test] -fn first_output_detection_should_match_crlf_frames() { +fn first_token_detection_should_match_crlf_frames() { let body = include_bytes!("../../../fixtures/responses/http_sse/text_delta_crlf.sse"); - assert!(response_body_has_first_output(body)); + assert!(response_body_has_semantic_output(body)); } #[test] diff --git a/frontend/src/views/dashboard/composables/useDashboard.ts b/frontend/src/views/dashboard/composables/useDashboard.ts index 38ac4fe7a..287faf2c6 100644 --- a/frontend/src/views/dashboard/composables/useDashboard.ts +++ b/frontend/src/views/dashboard/composables/useDashboard.ts @@ -220,7 +220,7 @@ export function useDashboard(): any { details: [ { label: '总请求', value: traffic.totalRequests, tone: 'info' }, { - label: '今日首字', + label: '首字均值', value: cache.averageFirstTokenLatencyMs, tone: 'info', }, diff --git a/frontend/src/views/usage/components/UsagePerformanceCard.vue b/frontend/src/views/usage/components/UsagePerformanceCard.vue index 715ac9550..d21e1d551 100644 --- a/frontend/src/views/usage/components/UsagePerformanceCard.vue +++ b/frontend/src/views/usage/components/UsagePerformanceCard.vue @@ -33,7 +33,7 @@ const performancePoints = computed(() => props.performance.points ?? []) const viewOptions = [ { label: '总耗时', value: 'total' }, - { label: '首 Token', value: 'ttft' }, + { label: '首字', value: 'firstToken' }, ] const percentileLabels = { @@ -108,10 +108,10 @@ const chartOption = computed(() => { }) function percentileValue(point: any, percentile: 'p50' | 'p95' | 'p99') { - if (activeView.value === 'ttft') { - if (percentile === 'p50') return point.ttftP50Ms - if (percentile === 'p95') return point.ttftP95Ms - return point.ttftP99Ms + if (activeView.value === 'firstToken') { + if (percentile === 'p50') return point.firstTokenP50Ms + if (percentile === 'p95') return point.firstTokenP95Ms + return point.firstTokenP99Ms } if (percentile === 'p50') return point.latencyP50Ms if (percentile === 'p95') return point.latencyP95Ms @@ -172,7 +172,7 @@ function formatTooltip(params: unknown) { as="article" :padded="false" title="响应速度" - description="常规、较慢和极慢请求的完成耗时，越低越快" + description="总耗时或首字的分位耗时，越低越快" header-collapse-at="none" header-class="px-5 pt-4" body-class="px-5 pt-3 pb-4" @@ -191,7 +191,7 @@ function formatTooltip(params: unknown) { plain :title="loading ? '正在加载性能数据' : '暂无性能数据'" :description=" - activeView === 'ttft' ? '当前范围没有首 Token 延迟样本' : '当前范围没有总耗时样本' + activeView === 'firstToken' ? '当前范围没有首字耗时样本' : '当前范围没有总耗时样本' " class="h-66 place-content-center" /> diff --git a/frontend/src/views/usage/components/UsageRecordDetailModal.vue b/frontend/src/views/usage/components/UsageRecordDetailModal.vue index b9c925e8e..f95adc6c5 100644 --- a/frontend/src/views/usage/components/UsageRecordDetailModal.vue +++ b/frontend/src/views/usage/components/UsageRecordDetailModal.vue @@ -51,7 +51,7 @@ const overviewItems = computed(() => [ { label: '时间', value: props.record?.createdAtDisplay, mono: true }, { label: '类型', value: usageRecordType(props.record) }, { label: '耗时', value: props.record?.latencyMsDisplay, mono: true }, - { label: '首 Token', value: props.record?.firstTokenLatencyMsDisplay, mono: true }, + { label: '首字', value: props.record?.firstTokenLatencyMsDisplay, mono: true }, { label: '总 Token', value: tokenDetails.value.totalTokensDisplay, mono: true }, { label: '请求 ID', value: props.record?.requestId, mono: true, wide: true }, { label: '消息', value: props.record?.message, wide: true }, diff --git a/frontend/src/views/usage/composables/useUsageRecordsTable.ts b/frontend/src/views/usage/composables/useUsageRecordsTable.ts index 83b17fb71..836c76d4e 100644 --- a/frontend/src/views/usage/composables/useUsageRecordsTable.ts +++ b/frontend/src/views/usage/composables/useUsageRecordsTable.ts @@ -210,11 +210,11 @@ function emptyOverview() { latencyP50Ms: null, latencyP95Ms: null, latencyP99Ms: null, - ttftP50Ms: null, - ttftP95Ms: null, - ttftP99Ms: null, + firstTokenP50Ms: null, + firstTokenP95Ms: null, + firstTokenP99Ms: null, latencyCoverage: 0, - ttftCoverage: 0, + firstTokenCoverage: 0, points: [], }, cost: { diff --git a/frontend/src/views/usage/constants.ts b/frontend/src/views/usage/constants.ts index afbf2f007..326ab5478 100644 --- a/frontend/src/views/usage/constants.ts +++ b/frontend/src/views/usage/constants.ts @@ -48,7 +48,7 @@ export const usageRecordColumns = [ }, { key: 'firstTokenLatencyMsDisplay', - label: '首 Token', + label: '首字', width: '104px', align: 'right' as const, ellipsis: false,