starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 16:06:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(telemetry): add first-token latency observability

Browse Source 
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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-16 15:57:19 +08:00
co-authored by Codex
parent bfd05722d5
commit 73c0ecdc4c
29 changed files with 552 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/dispatch/controllers/mod.rs
+91 -1
View File
@@ -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<AttemptFeedback> {
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<i64>,
request_started_at: Instant,
attempt_started_at: Instant,
) -> Option<u64> {
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<i64>) -> Option<u64> {
value.and_then(|value| u64::try_from(value).ok())
}
fn response_failure(observation: &AttemptObservation) -> Option<&ResponsesSseFailure> {
match &observation.kind {
AttemptObservationKind::CompleteResponse(
backend/src/dispatch/controllers/telemetry/mod.rs
+24 -1
View File
@@ -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<i64>,
) -> Option<i64> {
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,
backend/src/dispatch/lifecycle/attempt.rs
+5 -13
View File
@@ -50,13 +50,8 @@ use super::{
};
pub(in crate::dispatch) enum AttemptMode {
Complete {
started_at: Instant,
tuple_schema: Option<Value>,
},
Stream {
tuple_schema: Option<Value>,
},
Complete { tuple_schema: Option<Value> },
Stream { tuple_schema: Option<Value> },
}
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
{
backend/src/dispatch/lifecycle/trace.rs
+8
View File
@@ -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
}
}
backend/src/dispatch/service.rs
+1 -4
View File
@@ -220,10 +220,7 @@ impl ResponseDispatchService {
requested_model,
started_at,
},
AttemptMode::Complete {
started_at,
tuple_schema,
},
AttemptMode::Complete { tuple_schema },
request,
controller_scope,
candidates,
backend/src/dispatch/stream/live.rs
+1 -1
View File
@@ -304,7 +304,7 @@ fn update_first_token_ms(
started_at: Instant,
first_token_ms: &mut Option<i64>,
) {
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));
}
}
backend/src/dispatch/transport/canonical.rs
+12 -37
View File
@@ -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<CanonicalStreamTerminal> {
@@ -230,7 +204,7 @@ impl CanonicalStreamBatch {
/// 同一上游 chunk 可能同时包含输出与终态，事件顺序不能被网络分块抹掉。
pub(in crate::dispatch) fn first_failure(&self) -> Option<ResponsesSseFailure> {
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<Value>,
pending: Vec<u8>,
forwardable_frame_seen: bool,
first_output_seen: bool,
first_semantic_output_seen: bool,
terminal: Option<CanonicalStreamTerminal>,
transport_done_seen: bool,
completed_usage: Option<TokenUsage>,
@@ -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());
}
backend/src/fleet/pool/mod.rs
+6 -1
View File
@@ -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<Self>,
backend/src/fleet/pool/state.rs
+7 -4
View File
@@ -206,10 +206,13 @@ impl AccountPool {
self.options = options;
}
/// 回灌一次请求结果到调度器的运行时反馈（供 Smart 打分）。
pub fn report_feedback(&self, account_id: &str, success: bool, first_token_ms: Option<u64>) {
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);
}
/// 更新账号状态。
backend/src/fleet/scheduler/feedback.rs
+68 -23
View File
@@ -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<u64> },
Incomplete { first_token_ms: Option<u64> },
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<f64> {
load_sample(&self.error_rate_bits)
/// 读取当前首字前失败率 EWMA;无样本时返回 `None`。
fn pre_token_failure_rate(&self) -> Option<f64> {
load_sample(&self.pre_token_failure_rate_bits)
}
/// 读取当前首字后中断率 EWMA;无样本时返回 `None`。
fn post_token_abort_rate(&self) -> Option<f64> {
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<u64>) {
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<f64>,
/// 首字前失败率 EWMA，范围 `[0, 1]`;无样本时为 `None`。
pub pre_token_failure_rate: Option<f64>,
/// 首字后中断率 EWMA，范围 `[0, 1]`;无样本时为 `None`。
pub post_token_abort_rate: Option<f64>,
/// TTFT EWMA(毫秒);无样本时为 `None`。
pub ttft_ms: Option<f64>,
}
@@ -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<u64>) {
/// 回灌一个已启动上游 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);
}
/// 移除指定账号的反馈记录(账号被删除时调用)。
backend/src/fleet/scheduler/mod.rs
+5 -5
View File
@@ -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<u64>) {
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);
}
/// 移除指定账号的运行时反馈（账号被删除时调用）。
backend/src/fleet/scheduler/strategy/smart.rs
+1 -1
View File
@@ -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 {
backend/src/telemetry/usage/insights.rs
+21 -21
View File
@@ -264,11 +264,11 @@ struct UsagePerformanceInsights {
latency_p50_ms: Option<f64>,
latency_p95_ms: Option<f64>,
latency_p99_ms: Option<f64>,
ttft_p50_ms: Option<f64>,
ttft_p95_ms: Option<f64>,
ttft_p99_ms: Option<f64>,
first_token_p50_ms: Option<f64>,
first_token_p95_ms: Option<f64>,
first_token_p99_ms: Option<f64>,
latency_coverage: f64,
ttft_coverage: f64,
first_token_coverage: f64,
points: Vec<UsagePerformancePoint>,
}
@@ -280,9 +280,9 @@ struct UsagePerformancePoint {
latency_p50_ms: Option<f64>,
latency_p95_ms: Option<f64>,
latency_p99_ms: Option<f64>,
ttft_p50_ms: Option<f64>,
ttft_p95_ms: Option<f64>,
ttft_p99_ms: Option<f64>,
first_token_p50_ms: Option<f64>,
first_token_p95_ms: Option<f64>,
first_token_p99_ms: Option<f64>,
}
#[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,
})
}
backend/src/upstream/openai/failure.rs
+1 -1
View File
@@ -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)
backend/src/upstream/openai/protocol/responses.rs
+96 -45
View File
@@ -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::<Value>(&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<i64>,
) {
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::<Value>(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);
backend/src/upstream/openai/transport/response_meta.rs
+2 -2
View File
@@ -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<i64>,
) {
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);
}
backend/src/upstream/openai/transport/websocket.rs
+1 -1
View File
@@ -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;
backend/src/upstream/openai/transport/websocket_frames.rs
+9 -9
View File
@@ -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(),
}
}
backend/tests/dispatch/service/usage_logging.rs
+1 -1
View File
@@ -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",
backend/tests/fleet/scheduler/feedback.rs
+35 -25
View File
@@ -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());
}
backend/tests/fleet/scheduler/score.rs
+4 -2
View File
@@ -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;
backend/tests/support/storage.rs
+81 -10
View File
@@ -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<String> = OnceLock::new();
static REDIS_TEST_RUN_REGISTRATION: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();
static LOCAL_BOOTSTRAP_CONFIG: OnceLock<Option<BootstrapConfig>> = 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::<String, Vec<String>>::new();
let mut cursor = 0u64;
loop {
let (next_cursor, keys): (u64, Vec<String>) = 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()
backend/tests/upstream/openai/protocol/mod.rs
+2 -1
View File
@@ -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,
backend/tests/upstream/openai/protocol/sse.rs
+56 -20
View File
@@ -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]
frontend/src/views/dashboard/composables/useDashboard.ts
+1 -1
View File
@@ -220,7 +220,7 @@ export function useDashboard(): any {
details: [
{ label: '总请求', value: traffic.totalRequests, tone: 'info' },
{
label: '今日首字',
label: '首字均值',
value: cache.averageFirstTokenLatencyMs,
tone: 'info',
},
frontend/src/views/usage/components/UsagePerformanceCard.vue
+7 -7
View File
@@ -33,7 +33,7 @@ const performancePoints = computed<any[]>(() => props.performance.points ?? [])
const viewOptions = [
{ label: '总耗时', value: 'total' },
{ label: '首 Token', value: 'ttft' },
{ label: '首字', value: 'firstToken' },
]
const percentileLabels = {
@@ -108,10 +108,10 @@ const chartOption = computed<EChartsOption>(() => {
})
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"
/>
frontend/src/views/usage/components/UsageRecordDetailModal.vue
+1 -1
View File
@@ -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 },
frontend/src/views/usage/composables/useUsageRecordsTable.ts
+4 -4
View File
@@ -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: {
frontend/src/views/usage/constants.ts
+1 -1
View File
@@ -48,7 +48,7 @@ export const usageRecordColumns = [
},
{
key: 'firstTokenLatencyMsDisplay',
label: '首 Token',
label: '首字',
width: '104px',
align: 'right' as const,
ellipsis: false,