diff --git a/backend/src/api/admin/dashboard_routes.rs b/backend/src/api/admin/dashboard_routes.rs index 5c2dcf14a..ead99988f 100644 --- a/backend/src/api/admin/dashboard_routes.rs +++ b/backend/src/api/admin/dashboard_routes.rs @@ -40,7 +40,7 @@ use crate::{ usage::{ insights::RequestHealthTimeBucket, query::{UsageQueryFilter, UsageRecordAccountUsage}, - types::{UsageRecord, metadata_string}, + types::{UsageRecord, metadata_i64, metadata_string}, }, }, }; @@ -105,10 +105,25 @@ struct DashboardUsageRecordData { subagent_kind: Option, token_details: UsageRecordTokenDetailsData, billing: Option, + first_token_latency_ms: Option, + latency_ms: Option, + latency_details: DashboardUsageLatencyDetailsData, first_token_latency_ms_display: String, latency_ms_display: String, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct DashboardUsageLatencyDetailsData { + first_reasoning_ms: Option, + first_text_ms: Option, + transport_decision_wait_ms: Option, + ws_connect_ms: Option, + upstream_headers_ms: Option, + first_event_ms: Option, + openai_processing_ms: Option, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct DashboardWireProfileData { @@ -551,6 +566,7 @@ fn dashboard_usage_record_data( let token_details = usage_token_details(&record); let billing = usage_billing(&record, upstream_model.as_deref(), &token_details); let first_token_latency_ms = record.first_token_ms; + let latency_details = dashboard_usage_latency_details(&record.metadata); DashboardUsageRecordData { id: record.id, @@ -571,13 +587,32 @@ fn dashboard_usage_record_data( subagent_kind, token_details, billing, + first_token_latency_ms, + latency_ms: record.latency_ms, + latency_details, first_token_latency_ms_display: format_duration_ms(first_token_latency_ms), latency_ms_display: format_duration_ms(record.latency_ms), } } +fn dashboard_usage_latency_details(metadata: &Value) -> DashboardUsageLatencyDetailsData { + DashboardUsageLatencyDetailsData { + first_reasoning_ms: metadata_duration_ms(metadata, "firstReasoningMs"), + first_text_ms: metadata_duration_ms(metadata, "firstTextMs"), + transport_decision_wait_ms: metadata_duration_ms(metadata, "transportDecisionWaitMs"), + ws_connect_ms: metadata_duration_ms(metadata, "wsConnectMs"), + upstream_headers_ms: metadata_duration_ms(metadata, "upstreamHeadersMs"), + first_event_ms: metadata_duration_ms(metadata, "firstEventMs"), + openai_processing_ms: metadata_duration_ms(metadata, "openaiProcessingMs"), + } +} + +fn metadata_duration_ms(metadata: &Value, key: &str) -> Option { + metadata_i64(metadata, &[key]).filter(|value| *value >= 0) +} + fn wire_profile_data(state: &AppState) -> DashboardWireProfileData { - let profile = &state.services.wire_profile; + let profile = state.services.wire_profile.snapshot(); let release_snapshot = state.services.desktop_release.snapshot(); let release_status = if release_snapshot.last_error.is_some() { "check_failed" diff --git a/backend/src/api/router.rs b/backend/src/api/router.rs index 34708a581..829a11a94 100644 --- a/backend/src/api/router.rs +++ b/backend/src/api/router.rs @@ -28,7 +28,7 @@ use crate::{ }, update::service::SystemUpdateService, upstream::openai::{ - desktop_release::DesktopReleaseStatus, profile::CodexWireProfile, + desktop_release::DesktopReleaseStatus, profile::CodexWireProfileState, token_client::OpenAiTokenClient, }, }; @@ -51,7 +51,7 @@ pub struct ApiServices { pub token_refresh: Arc>, pub responses: Arc, pub session_affinity: Arc, - pub wire_profile: Arc, + pub wire_profile: CodexWireProfileState, pub desktop_release: DesktopReleaseStatus, pub process_control: Arc, pub system_update: Arc, diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index 3a6617264..3237cd9f1 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -51,7 +51,7 @@ use crate::{ update::service::SystemUpdateService, upstream::openai::{ desktop_release::DesktopReleaseStatus, - profile::CodexWireProfile, + profile::{CodexWireProfile, CodexWireProfileState}, token_client::{OpenAiTokenClient, TokenClientConfig, default_openai_token_client}, transport::{ CodexBackendClient, CodexModelCatalogClient, CodexWebSocketPool, @@ -117,7 +117,7 @@ pub struct Services { pub session_affinity: Arc, pub codex: Arc, pub websocket_pool: Option>, - pub wire_profile: Arc, + pub wire_profile: CodexWireProfileState, pub desktop_release: DesktopReleaseStatus, pub account_pseudonymizer: Arc, pub system_update: Arc, @@ -146,7 +146,7 @@ impl Services { pub fn new( config: &AppConfig, stores: BackgroundTaskStores, - wire_profile: Arc, + wire_profile: CodexWireProfileState, ) -> Self { Self::try_new(config, stores, wire_profile) .expect("failed to build runtime services with configured TLS transport") @@ -155,7 +155,7 @@ impl Services { pub fn try_new( config: &AppConfig, stores: BackgroundTaskStores, - wire_profile: Arc, + wire_profile: CodexWireProfileState, ) -> Result { Self::try_with_usage_record_options( config, @@ -168,7 +168,7 @@ impl Services { pub fn try_with_usage_record_options( config: &AppConfig, stores: BackgroundTaskStores, - wire_profile: Arc, + wire_profile: CodexWireProfileState, usage_record_options: UsageRecordOptions, ) -> Result { let mut identity_secret = [0u8; 32]; @@ -185,7 +185,7 @@ impl Services { fn try_with_identity_secret_and_usage_record_options( config: &AppConfig, stores: BackgroundTaskStores, - wire_profile: Arc, + wire_profile: CodexWireProfileState, identity_secret: [u8; 32], usage_record_options: UsageRecordOptions, ) -> Result { @@ -578,7 +578,7 @@ async fn build_router( let data_dir = ensure_data_dir(&config.runtime.data_directory)?; let identity_secret = load_or_create_identity_secret(&data_dir)?; - let wire_profile = Arc::new(wire_profile_from_config(&config.wire_profile)); + let wire_profile = CodexWireProfileState::new(wire_profile_from_config(&config.wire_profile)); let runtime_config = crate::bootstrap::state::RuntimeConfig::from(&config); let services = Services::try_with_identity_secret_and_usage_record_options( &config, diff --git a/backend/src/bootstrap/tasks/coordinator.rs b/backend/src/bootstrap/tasks/coordinator.rs index 834b06709..99fbdbeca 100644 --- a/backend/src/bootstrap/tasks/coordinator.rs +++ b/backend/src/bootstrap/tasks/coordinator.rs @@ -91,6 +91,7 @@ impl TaskCoordinator { "desktop_release_update", start_desktop_release_update_task( services.desktop_release.clone(), + services.wire_profile.clone(), crate::upstream::openai::desktop_release::CODEX_DESKTOP_APPCAST_URL.to_string(), ), ); diff --git a/backend/src/bootstrap/tasks/desktop_release_update.rs b/backend/src/bootstrap/tasks/desktop_release_update.rs index 8ba82b468..5ff9b501e 100644 --- a/backend/src/bootstrap/tasks/desktop_release_update.rs +++ b/backend/src/bootstrap/tasks/desktop_release_update.rs @@ -5,6 +5,7 @@ use tracing::{info, warn}; use crate::upstream::openai::desktop_release::{ APPCAST_POLL_INTERVAL, DesktopReleaseChecker, DesktopReleaseStatus, }; +use crate::upstream::openai::profile::CodexWireProfileState; use super::{ coordinator::SchedulerHandle, @@ -14,9 +15,15 @@ use super::{ /// 启动 Desktop 官方发布检查后台任务。 pub fn start_desktop_release_update_task( status: DesktopReleaseStatus, + wire_profile: CodexWireProfileState, appcast_url: String, ) -> SchedulerHandle { - let checker = DesktopReleaseChecker::with_client(reqwest::Client::new(), appcast_url, status); + let checker = DesktopReleaseChecker::with_client( + reqwest::Client::new(), + appcast_url, + status, + wire_profile, + ); spawn_periodic_task( DesktopReleaseUpdateTask { checker, @@ -43,7 +50,7 @@ impl PeriodicTaskRunner for DesktopReleaseUpdateTask { Ok(release) => info!( version = %release.version, build = %release.build, - "Codex Desktop 最新发布信息已刷新" + "Codex Desktop 最新发布信息已刷新并已同步请求身份" ), Err(error) if first_tick => { warn!(error = %error, "Codex Desktop 首次发布检查失败"); diff --git a/backend/src/dispatch/controllers/telemetry/events.rs b/backend/src/dispatch/controllers/telemetry/events.rs index 516f4a832..6d2b24ebe 100644 --- a/backend/src/dispatch/controllers/telemetry/events.rs +++ b/backend/src/dispatch/controllers/telemetry/events.rs @@ -21,8 +21,8 @@ use crate::{ upstream::openai::{ protocol::responses::{CodexResponsesRequest, ResponsesSseFailure}, transport::{ - CodexBackendTransport, CodexClientError, CodexUpstreamDiagnostics, - WebSocketPoolDecision, + CodexBackendTransport, CodexClientError, CodexResponseMetadata, + CodexUpstreamDiagnostics, WebSocketPoolDecision, }, }, }; @@ -313,6 +313,11 @@ fn enrich_live_response_stream_metadata( "firstEventMs", context.transport_metrics.first_event_ms, ); + insert_optional_i64( + object, + "openaiProcessingMs", + openai_processing_ms(context.response_metadata), + ); if let Some(http_version) = context.transport_metrics.http_version.as_deref() { object .entry("httpVersion".to_string()) @@ -683,16 +688,40 @@ fn metadata_string_field<'a>(metadata: &'a Value, field: &str) -> Option<&'a str metadata.get(field).and_then(Value::as_str) } -pub(super) fn insert_first_token_ms(metadata: &mut Value, first_token_ms: Option) { - let Some(first_token_ms) = first_token_ms else { +pub(super) fn insert_output_timing_ms( + metadata: &mut Value, + first_token_ms: Option, + first_reasoning_ms: Option, + first_text_ms: Option, +) { + if let Some(object) = metadata.as_object_mut() { + insert_optional_i64(object, "firstTokenMs", first_token_ms); + insert_optional_i64(object, "firstReasoningMs", first_reasoning_ms); + insert_optional_i64(object, "firstTextMs", first_text_ms); + } +} + +pub(super) fn insert_openai_processing_ms( + metadata: &mut Value, + response_metadata: &CodexResponseMetadata, +) { + let Some(object) = metadata.as_object_mut() else { return; }; - if let Some(object) = metadata.as_object_mut() { - object.insert( - "firstTokenMs".to_string(), - Value::Number(first_token_ms.into()), - ); - } + insert_optional_i64( + object, + "openaiProcessingMs", + openai_processing_ms(response_metadata), + ); +} + +fn openai_processing_ms(response_metadata: &CodexResponseMetadata) -> Option { + response_metadata + .client_headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("openai-processing-ms")) + .and_then(|(_, value)| value.parse::().ok()) + .filter(|value| *value >= 0) } pub(super) fn insert_websocket_pool_decision( diff --git a/backend/src/dispatch/controllers/telemetry/mod.rs b/backend/src/dispatch/controllers/telemetry/mod.rs index d645ab659..0c98516e2 100644 --- a/backend/src/dispatch/controllers/telemetry/mod.rs +++ b/backend/src/dispatch/controllers/telemetry/mod.rs @@ -9,7 +9,7 @@ use serde_json::Value; use crate::{ dispatch::{ controllers::telemetry::events::{ - ResponseUpstreamErrorEventRecord, insert_first_token_ms, + ResponseUpstreamErrorEventRecord, insert_openai_processing_ms, insert_output_timing_ms, insert_response_status_metadata, insert_response_trace_metadata, insert_response_upstream_diagnostics, insert_websocket_pool_decision, record_live_response_stream_event, record_response_upstream_error_event, @@ -151,18 +151,27 @@ impl TelemetryController { .effective_model .as_deref() .unwrap_or(exit.display_model); - let first_token_ms = request_first_token_ms( + let first_token_ms = request_timing_ms( exit.started_at, exit.attempt.started_at(), exit.response.first_token_ms, ); + let first_reasoning_ms = request_timing_ms( + exit.started_at, + exit.attempt.started_at(), + exit.response.first_reasoning_ms, + ); + let first_text_ms = request_timing_ms( + exit.started_at, + exit.attempt.started_at(), + exit.response.first_text_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": first_token_ms, "usage": exit.response.usage, "effectiveModel": effective_model, "modelsEtag": exit.response.response_metadata.models_etag.as_deref(), @@ -174,6 +183,13 @@ impl TelemetryController { "firstEventMs": exit.response.transport_metrics.first_event_ms, "httpVersion": exit.response.transport_metrics.http_version.as_deref(), }); + insert_output_timing_ms( + &mut metadata, + first_token_ms, + first_reasoning_ms, + first_text_ms, + ); + insert_openai_processing_ms(&mut metadata, &exit.response.response_metadata); insert_response_status_metadata( &mut metadata, 200, @@ -255,7 +271,12 @@ impl TelemetryController { "firstEventMs": summary.first_event_ms, "usage": summary.usage, }); - insert_first_token_ms(&mut metadata, summary.first_token_ms); + insert_output_timing_ms( + &mut metadata, + summary.first_token_ms, + summary.first_reasoning_ms, + summary.first_text_ms, + ); record_live_response_stream_event( context, status_code, @@ -271,7 +292,7 @@ impl TelemetryController { } } -fn request_first_token_ms( +fn request_timing_ms( request_started_at: Instant, attempt_started_at: Instant, attempt_first_token_ms: Option, @@ -295,6 +316,20 @@ async fn record_success( rate_limit_headers: &[(String, String)], body: &str, ) { + let mut metadata = serde_json::json!({ + "stream": true, + "completed": completed, + "incomplete": !completed, + "responseId": response.get("id").and_then(Value::as_str), + "firstEventMs": summary.first_event_ms, + "usage": summary.usage, + }); + insert_output_timing_ms( + &mut metadata, + summary.first_token_ms, + summary.first_reasoning_ms, + summary.first_text_ms, + ); record_live_response_stream_event( context, 200, @@ -304,15 +339,7 @@ async fn record_success( } else { "v1 responses stream incomplete" }, - serde_json::json!({ - "stream": true, - "completed": completed, - "incomplete": !completed, - "responseId": response.get("id").and_then(Value::as_str), - "firstEventMs": summary.first_event_ms, - "firstTokenMs": summary.first_token_ms, - "usage": summary.usage, - }), + metadata, rate_limit_headers, body, ) @@ -367,7 +394,12 @@ async fn record_failure( .map_or(Value::Null, |response_id| { Value::String(response_id.clone()) }); - insert_first_token_ms(&mut metadata, summary.first_token_ms); + insert_output_timing_ms( + &mut metadata, + summary.first_token_ms, + summary.first_reasoning_ms, + summary.first_text_ms, + ); metadata["firstEventMs"] = Value::Number(summary.first_event_ms.into()); record_live_response_stream_event( context, diff --git a/backend/src/dispatch/lifecycle/stream.rs b/backend/src/dispatch/lifecycle/stream.rs index 2e34684b1..95054dfa3 100644 --- a/backend/src/dispatch/lifecycle/stream.rs +++ b/backend/src/dispatch/lifecycle/stream.rs @@ -28,6 +28,8 @@ pub(in crate::dispatch) struct StreamSummary { pub body: Vec, pub terminal_chunks: Vec, pub first_token_ms: Option, + pub first_reasoning_ms: Option, + pub first_text_ms: Option, pub first_event_ms: i64, pub usage: Option, pub last_response_id: Option, diff --git a/backend/src/dispatch/stream/live.rs b/backend/src/dispatch/stream/live.rs index b20b1970d..872d8eb67 100644 --- a/backend/src/dispatch/stream/live.rs +++ b/backend/src/dispatch/stream/live.rs @@ -117,6 +117,8 @@ async fn run_live_response_stream( let mut body_bytes = Vec::new(); let mut terminal_chunks = Vec::new(); let mut first_token_ms = None; + let mut first_reasoning_ms = None; + let mut first_text_ms = None; let decoded_terminal = take_stream_terminal(&mut decoder); let terminal = if decoded_terminal.is_some() { @@ -128,13 +130,19 @@ async fn run_live_response_stream( .await .err() }; - update_first_token_ms(&decoder, started_at, &mut first_token_ms); + update_output_timing_ms( + &decoder, + started_at, + &mut first_token_ms, + &mut first_reasoning_ms, + &mut first_text_ms, + ); if let Some(terminal) = terminal { return stream_summary( terminal, body_bytes, terminal_chunks, - first_token_ms, + (first_token_ms, first_reasoning_ms, first_text_ms), first_event_ms, &decoder, ); @@ -150,7 +158,13 @@ async fn run_live_response_stream( match next { Some(Ok(chunk)) => match decoder.push(chunk) { Ok(batch) => { - update_first_token_ms(&decoder, started_at, &mut first_token_ms); + update_output_timing_ms( + &decoder, + started_at, + &mut first_token_ms, + &mut first_reasoning_ms, + &mut first_text_ms, + ); if let Some(terminal) = take_stream_terminal(&mut decoder) { match stage_terminal_batch(&mut body_bytes, &mut terminal_chunks, batch) { Ok(()) => break terminal, @@ -196,12 +210,18 @@ async fn run_live_response_stream( } }; - update_first_token_ms(&decoder, started_at, &mut first_token_ms); + update_output_timing_ms( + &decoder, + started_at, + &mut first_token_ms, + &mut first_reasoning_ms, + &mut first_text_ms, + ); stream_summary( terminal, body_bytes, terminal_chunks, - first_token_ms, + (first_token_ms, first_reasoning_ms, first_text_ms), first_event_ms, &decoder, ) @@ -299,13 +319,29 @@ fn ensure_capture_capacity( Ok(()) } -fn update_first_token_ms( +fn update_output_timing_ms( decoder: &CanonicalStreamDecoder, started_at: Instant, first_token_ms: &mut Option, + first_reasoning_ms: &mut Option, + first_text_ms: &mut Option, ) { - if first_token_ms.is_none() && decoder.first_semantic_output_seen() { - *first_token_ms = Some(elapsed_millis_i64(started_at).max(1)); + let first_token_missing = first_token_ms.is_none() && decoder.first_semantic_output_seen(); + let first_reasoning_missing = + first_reasoning_ms.is_none() && decoder.first_reasoning_output_seen(); + let first_text_missing = first_text_ms.is_none() && decoder.first_text_output_seen(); + if !(first_token_missing || first_reasoning_missing || first_text_missing) { + return; + } + let elapsed_ms = elapsed_millis_i64(started_at).max(1); + if first_token_missing { + *first_token_ms = Some(elapsed_ms); + } + if first_reasoning_missing { + *first_reasoning_ms = Some(elapsed_ms); + } + if first_text_missing { + *first_text_ms = Some(elapsed_ms); } } @@ -313,15 +349,18 @@ fn stream_summary( terminal: StreamTerminal, body: Vec, terminal_chunks: Vec, - first_token_ms: Option, + output_timing: (Option, Option, Option), first_event_ms: i64, decoder: &CanonicalStreamDecoder, ) -> StreamSummary { + let (first_token_ms, first_reasoning_ms, first_text_ms) = output_timing; StreamSummary { terminal, body, terminal_chunks, first_token_ms, + first_reasoning_ms, + first_text_ms, first_event_ms, usage: decoder.usage(), last_response_id: decoder.last_response_id().map(ToString::to_string), diff --git a/backend/src/dispatch/transport/canonical.rs b/backend/src/dispatch/transport/canonical.rs index 4d5aed61a..8b69ff838 100644 --- a/backend/src/dispatch/transport/canonical.rs +++ b/backend/src/dispatch/transport/canonical.rs @@ -225,6 +225,8 @@ pub(in crate::dispatch) struct CanonicalStreamDecoder { pending: Vec, forwardable_frame_seen: bool, first_semantic_output_seen: bool, + first_reasoning_output_seen: bool, + first_text_output_seen: bool, terminal: Option, transport_done_seen: bool, completed_usage: Option, @@ -239,6 +241,8 @@ impl CanonicalStreamDecoder { pending: Vec::new(), forwardable_frame_seen: false, first_semantic_output_seen: false, + first_reasoning_output_seen: false, + first_text_output_seen: false, terminal: None, transport_done_seen: false, completed_usage: None, @@ -311,6 +315,14 @@ impl CanonicalStreamDecoder { self.first_semantic_output_seen } + pub(in crate::dispatch) fn first_reasoning_output_seen(&self) -> bool { + self.first_reasoning_output_seen + } + + pub(in crate::dispatch) fn first_text_output_seen(&self) -> bool { + self.first_text_output_seen + } + pub(in crate::dispatch) fn commit_boundary_reached(&self, policy: StreamCommitPolicy) -> bool { match policy { StreamCommitPolicy::FirstForwardableEvent => self.forwardable_frame_seen, @@ -382,6 +394,8 @@ impl CanonicalStreamDecoder { fn observe(&mut self, event: &CanonicalResponseEvent) { let signals = event.signals(); self.first_semantic_output_seen |= signals.semantic_output; + self.first_reasoning_output_seen |= signals.reasoning_output; + self.first_text_output_seen |= signals.text_output; if let Some(response_id) = event.response_id() { self.last_response_id = Some(response_id.to_string()); } diff --git a/backend/src/upstream/openai/desktop_release.rs b/backend/src/upstream/openai/desktop_release.rs index c777e23ff..f0e5a8642 100644 --- a/backend/src/upstream/openai/desktop_release.rs +++ b/backend/src/upstream/openai/desktop_release.rs @@ -8,6 +8,8 @@ use std::{ use chrono::{DateTime, Utc}; use roxmltree::{Document, Node}; +use super::profile::CodexWireProfileState; + const APPCAST_TIMEOUT: Duration = Duration::from_secs(30); /// Codex Desktop 官方 appcast 地址。 @@ -72,12 +74,13 @@ impl DesktopReleaseStatus { } } -/// 读取官方 appcast 并更新发布观测状态。 +/// 读取官方 appcast，同步运行时请求画像并更新发布观测状态。 #[derive(Clone)] pub struct DesktopReleaseChecker { client: reqwest::Client, appcast_url: String, status: DesktopReleaseStatus, + wire_profile: CodexWireProfileState, } impl DesktopReleaseChecker { @@ -85,11 +88,13 @@ impl DesktopReleaseChecker { client: reqwest::Client, appcast_url: impl Into, status: DesktopReleaseStatus, + wire_profile: CodexWireProfileState, ) -> Self { Self { client, appcast_url: appcast_url.into(), status, + wire_profile, } } @@ -98,6 +103,8 @@ impl DesktopReleaseChecker { let result = self.fetch_latest().await; match result { Ok(release) => { + self.wire_profile + .update_desktop_release(&release.version, &release.build); self.status.record_success(checked_at, release.clone()); Ok(release) } @@ -138,9 +145,15 @@ pub fn parse_latest_desktop_release(xml: &str) -> Result>, local_name: &str) -> Option bool { + let mut parts = value.split('.'); + let valid_parts = parts + .by_ref() + .filter(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())) + .count(); + valid_parts >= 2 && valid_parts == value.split('.').count() +} + #[derive(Debug, thiserror::Error)] pub enum DesktopReleaseError { #[error("Codex Desktop appcast request failed: {0}")] diff --git a/backend/src/upstream/openai/profile.rs b/backend/src/upstream/openai/profile.rs index 5601001d0..2b2004859 100644 --- a/backend/src/upstream/openai/profile.rs +++ b/backend/src/upstream/openai/profile.rs @@ -1,11 +1,13 @@ //! Codex Desktop 上游请求画像。 +use std::sync::{Arc, RwLock}; + use chrono::{DateTime, Utc}; -/// 经制品与 Codex 源码核验后固定的上游请求身份。 +/// Codex Desktop 上游请求身份。 /// -/// 该画像是启动配置，不会被在线发布检查自动改写。Desktop 发布变化后应先重新审计 -/// bundled Codex Core，再整体更新此画像。 +/// 启动配置提供经源码审计的 Core、运行环境和初始 Desktop 版本。官方 appcast +/// 检查成功后，会仅同步 Desktop 版本和构建号。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct CodexWireProfile { /// `originator` 请求头及 User-Agent 产品名。 @@ -44,3 +46,39 @@ impl CodexWireProfile { ) } } + +/// 跨上游请求与发布检查共享的运行时请求画像。 +#[derive(Debug, Clone)] +pub struct CodexWireProfileState { + profile: Arc>, +} + +impl CodexWireProfileState { + /// 从启动画像创建运行时状态。 + pub fn new(profile: CodexWireProfile) -> Self { + Self { + profile: Arc::new(RwLock::new(profile)), + } + } + + /// 返回当前画像的独立快照，避免持锁执行网络请求。 + pub fn snapshot(&self) -> CodexWireProfile { + self.profile + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + /// 使用官方 appcast 的 Desktop 版本和构建号更新当前画像。 + pub fn update_desktop_release(&self, desktop_version: &str, desktop_build: &str) { + let mut profile = self + .profile + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if profile.desktop_version == desktop_version && profile.desktop_build == desktop_build { + return; + } + profile.desktop_version = desktop_version.to_string(); + profile.desktop_build = desktop_build.to_string(); + } +} diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 060e2afd1..be7f99372 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -1,15 +1,12 @@ -use std::{collections::BTreeSet, sync::Arc, time::Instant}; +use std::{collections::BTreeSet, sync::Arc}; use serde::Serialize; use serde_json::{Map, Value}; -use crate::{ - infra::time::elapsed_millis_i64, - upstream::openai::protocol::{ - events, - schema::reconvert_tuple_values, - sse::{SseError, SseEvent, parse_sse_events}, - }, +use crate::upstream::openai::protocol::{ + events, + schema::reconvert_tuple_values, + sse::{SseError, SseEvent, parse_sse_events}, }; const MULTI_AGENT_MODE_OPEN_TAG: &str = ""; @@ -197,12 +194,14 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir /// 单个 Responses 事件对计时系统提供的稳定语义信号。 /// -/// `protocol_progress` 只说明上游仍在工作，不能替代首字；`semantic_output` -/// 表示模型已经产生客户端可消费的内容。 +/// `protocol_progress` 只说明上游仍在工作，不能替代首字；其余字段分别标记 +/// 客户端可消费的输出、reasoning 输出与正文输出。 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ResponseEventSignals { pub protocol_progress: bool, pub semantic_output: bool, + pub reasoning_output: bool, + pub text_output: bool, } /// 从已解析的 Responses 事件提取计时语义。 @@ -210,50 +209,76 @@ pub struct ResponseEventSignals { /// 生命周期帧和结构帧不会被当作输出。`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")), + let mut signals = ResponseEventSignals { + protocol_progress: !matches!(event_type, Some("response.failed" | "error")), + ..ResponseEventSignals::default() + }; + match event_type { + Some("response.output_text.delta") => { + signals.text_output = non_empty_string(value.get("delta")); + signals.semantic_output = signals.text_output; + } + Some("response.output_text.done") => { + signals.text_output = non_empty_string(value.get("text")); + signals.semantic_output = signals.text_output; + } Some("response.reasoning_summary_text.delta" | "response.reasoning_text.delta") => { - non_empty_string(value.get("delta")) + signals.reasoning_output = non_empty_string(value.get("delta")); + signals.semantic_output = signals.reasoning_output; } 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, + ) => { + signals.semantic_output = non_empty_string(value.get("delta")); + } + Some("response.function_call_arguments.done") => { + signals.semantic_output = non_empty_string(value.get("arguments")); + } + Some("response.image_generation_call.partial_image") => { + signals.semantic_output = non_empty_string( + value + .get("partial_image_b64") + .or_else(|| value.get("partial_image")), + ); + } + Some("response.output_item.done") => { + if let Some(item) = value.get("item") { + merge_output_signals(&mut signals, output_item_signals(item)); + } + } + Some("response.content_part.done") => { + if let Some(part) = value.get("part") { + merge_output_signals(&mut signals, output_item_signals(part)); + } + } + Some("response.completed" | "response.incomplete") => { + if let Some(items) = value.pointer("/response/output").and_then(Value::as_array) { + merge_output_signals(&mut signals, output_items_signals(items)); + } + } + _ => {} } + signals } /// 判断已收到的 Responses SSE 内容是否包含首个完整的语义输出事件。 pub fn response_body_has_semantic_output(body_bytes: &[u8]) -> bool { + response_body_signals(body_bytes).semantic_output +} + +fn response_body_signals(body_bytes: &[u8]) -> ResponseEventSignals { let body = String::from_utf8_lossy(body_bytes); let Some(complete_body) = complete_sse_body_prefix(&body) else { - return false; + return ResponseEventSignals::default(); }; - parse_sse_events(complete_body).is_ok_and(|events| { - events.iter().any(|event| { + let Ok(events) = parse_sse_events(complete_body) else { + return ResponseEventSignals::default(); + }; + events + .iter() + .fold(ResponseEventSignals::default(), |mut signals, event| { let Ok(value) = serde_json::from_str::(&event.data) else { - return false; + return signals; }; let event_type = event.event.as_deref().or_else(|| { value @@ -261,46 +286,60 @@ pub fn response_body_has_semantic_output(body_bytes: &[u8]) -> bool { .and_then(Value::as_str) .filter(|value| !value.trim().is_empty()) }); - response_event_signals(event_type, &value).semantic_output + merge_event_signals(&mut signals, response_event_signals(event_type, &value)); + signals }) - }) } -/// 已收到首个语义输出事件时记录耗时。 -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_semantic_output(body_bytes) { - *first_token_ms = Some(elapsed_millis_i64(started_at).max(1)); - } +fn merge_event_signals(target: &mut ResponseEventSignals, source: ResponseEventSignals) { + target.protocol_progress |= source.protocol_progress; + merge_output_signals(target, source); } -fn output_items_have_semantic_output(items: &[Value]) -> bool { - items.iter().any(output_item_has_semantic_output) +fn merge_output_signals(target: &mut ResponseEventSignals, source: ResponseEventSignals) { + target.semantic_output |= source.semantic_output; + target.reasoning_output |= source.reasoning_output; + target.text_output |= source.text_output; } -fn output_item_has_semantic_output(item: &Value) -> bool { +fn output_items_signals(items: &[Value]) -> ResponseEventSignals { + items + .iter() + .fold(ResponseEventSignals::default(), |mut signals, item| { + merge_output_signals(&mut signals, output_item_signals(item)); + signals + }) +} + +fn output_item_signals(item: &Value) -> ResponseEventSignals { + let mut signals = ResponseEventSignals::default(); match item.get("type").and_then(Value::as_str) { - Some("output_text" | "text") => non_empty_string(item.get("text")), + Some("output_text" | "text") => { + signals.text_output = non_empty_string(item.get("text")); + signals.semantic_output = signals.text_output; + } Some("reasoning") => { - non_empty_string(item.get("text")) || non_empty_string(item.get("summary")) + signals.reasoning_output = + non_empty_string(item.get("text")) || non_empty_string(item.get("summary")); + signals.semantic_output = signals.reasoning_output; + } + Some("function_call") => { + signals.semantic_output = non_empty_string(item.get("arguments")); + } + Some("custom_tool_call") => { + signals.semantic_output = non_empty_string(item.get("input")); } - 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")) + signals.semantic_output = + non_empty_string(item.get("result")) || non_empty_string(item.get("image_base64")); + } + _ => { + if let Some(items) = item.get("content").and_then(Value::as_array) { + merge_output_signals(&mut signals, output_items_signals(items)); + } } - 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)), } + signals } fn non_empty_string(value: Option<&Value>) -> bool { diff --git a/backend/src/upstream/openai/transport/client.rs b/backend/src/upstream/openai/transport/client.rs index 4c0540aeb..614dc2531 100644 --- a/backend/src/upstream/openai/transport/client.rs +++ b/backend/src/upstream/openai/transport/client.rs @@ -21,7 +21,7 @@ use serde_json::{Value, map::Map}; use thiserror::Error; use tokio_tungstenite::tungstenite::handshake::client::generate_key; -use crate::upstream::openai::profile::CodexWireProfile; +use crate::upstream::openai::profile::CodexWireProfileState; use crate::upstream::openai::protocol::events::{extract_sse_usage, retry_after_seconds_from_body}; use crate::upstream::openai::protocol::responses::{ CodexResponsesRequest, TransportRequirement, transport_requirement, @@ -277,6 +277,10 @@ pub struct CodexBackendResponse { pub rate_limit_headers: Vec<(String, String)>, /// 首个有效上游 SSE/WebSocket 事件到达代理的耗时。 pub first_token_ms: Option, + /// 首个 reasoning 输出事件到达代理的耗时。 + pub first_reasoning_ms: Option, + /// 首个正文输出事件到达代理的耗时。 + pub first_text_ms: Option, /// WebSocket 连接池决策。 pub websocket_pool_decision: Option, /// 上游诊断元数据。 @@ -382,7 +386,7 @@ pub struct CodexBackendStreamingResponse { pub struct CodexBackendClient { pub(super) client: Client, pub(super) base_url: String, - profile: Arc, + profile: CodexWireProfileState, websocket_pool: Option>, websocket_initial_event_timeout: Option, websocket_fast_path_budget: Duration, diff --git a/backend/src/upstream/openai/transport/client_sse.rs b/backend/src/upstream/openai/transport/client_sse.rs index 4e00659e8..be28a4474 100644 --- a/backend/src/upstream/openai/transport/client_sse.rs +++ b/backend/src/upstream/openai/transport/client_sse.rs @@ -1,3 +1,5 @@ +use crate::upstream::openai::{profile::CodexWireProfile, protocol::sse::SseEventDecoder}; + use super::*; impl CodexBackendClient { @@ -5,7 +7,7 @@ impl CodexBackendClient { pub fn new( client: Client, base_url: impl Into, - profile: Arc, + profile: CodexWireProfileState, ) -> Self { let base_url = base_url.into().trim_end_matches('/').to_string(); Self { @@ -90,12 +92,22 @@ impl CodexBackendClient { let mut body_bytes = Vec::new(); let mut first_token_ms = None; + let mut first_reasoning_ms = None; + let mut first_text_ms = None; let mut first_event_ms = None; + let mut output_decoder = SseEventDecoder::default(); let mut stream = http_sse_stream(response); while let Some(chunk) = stream.try_next().await? { first_event_ms.get_or_insert_with(|| elapsed_duration_millis(started_at.elapsed())); body_bytes.extend_from_slice(&chunk); - response_meta::update_first_token_ms(started_at, &body_bytes, &mut first_token_ms); + response_meta::update_response_timing_ms( + started_at, + &mut output_decoder, + &chunk, + &mut first_token_ms, + &mut first_reasoning_ms, + &mut first_text_ms, + ); } let body = String::from_utf8_lossy(&body_bytes).into_owned(); let usage = extract_sse_usage(&body).map_err(CodexClientError::InvalidSse)?; @@ -107,6 +119,8 @@ impl CodexBackendClient { set_cookie_headers, rate_limit_headers, first_token_ms, + first_reasoning_ms, + first_text_ms, websocket_pool_decision: None, diagnostics, response_metadata, @@ -391,6 +405,8 @@ impl CodexBackendClient { set_cookie_headers: exchange.set_cookie_headers, rate_limit_headers: exchange.rate_limit_headers, first_token_ms: exchange.first_token_ms, + first_reasoning_ms: exchange.first_reasoning_ms, + first_text_ms: exchange.first_text_ms, websocket_pool_decision: exchange.pool_decision, diagnostics: exchange.diagnostics, response_metadata: exchange.response_metadata, @@ -501,11 +517,12 @@ impl CodexBackendClient { context: CodexRequestContext<'_>, ) -> CodexClientResult> { let endpoint = endpoint_url(&self.base_url, "codex/models"); - let headers = self.auxiliary_request_headers(context)?; + let profile = self.profile.snapshot(); + let headers = self.auxiliary_request_headers(&profile, context)?; let response = self .client .get(endpoint) - .query(&[("client_version", self.profile.codex_version.as_str())]) + .query(&[("client_version", profile.codex_version.as_str())]) .headers(headers) .send() .await?; @@ -559,7 +576,8 @@ impl CodexBackendClient { request: &CodexResponsesRequest, context: CodexRequestContext<'_>, ) -> CodexClientResult { - let mut headers = self.request_headers(context)?; + let profile = self.profile.snapshot(); + let mut headers = self.request_headers(&profile, context)?; if let Some(subagent) = openai_subagent_from_metadata(request.client_metadata()) { headers.insert( HeaderName::from_static("x-openai-subagent"), @@ -569,10 +587,14 @@ impl CodexBackendClient { Ok(headers) } - fn request_headers(&self, context: CodexRequestContext<'_>) -> CodexClientResult { + fn request_headers( + &self, + profile: &CodexWireProfile, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { let request_id = context.client_request_id.unwrap_or(context.request_id); let mut headers = - build_codex_base_headers(&self.profile, context.access_token, context.account_id)?; + build_codex_base_headers(profile, context.access_token, context.account_id)?; headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); insert_optional_header(&mut headers, "cookie", context.cookie_header)?; headers.insert(ACCEPT, HeaderValue::from_static("text/event-stream")); @@ -617,10 +639,11 @@ impl CodexBackendClient { fn auxiliary_request_headers( &self, + profile: &CodexWireProfile, context: CodexRequestContext<'_>, ) -> CodexClientResult { let mut headers = - build_codex_base_headers(&self.profile, context.access_token, context.account_id)?; + build_codex_base_headers(profile, context.access_token, context.account_id)?; if let Some(cookie_header) = context.cookie_header { headers.insert(COOKIE, HeaderValue::from_str(cookie_header)?); } @@ -637,18 +660,16 @@ impl CodexBackendClient { &self, context: CodexRequestContext<'_>, ) -> CodexClientResult { + let profile = self.profile.snapshot(); let mut headers = HeaderMap::new(); - headers.insert( - USER_AGENT, - HeaderValue::from_str(&self.profile.user_agent())?, - ); + headers.insert(USER_AGENT, HeaderValue::from_str(&profile.user_agent())?); headers.insert( AUTHORIZATION, HeaderValue::from_str(&format!("Bearer {}", context.access_token))?, ); headers.insert( HeaderName::from_static("originator"), - HeaderValue::from_str(&self.profile.originator)?, + HeaderValue::from_str(&profile.originator)?, ); insert_optional_header(&mut headers, "chatgpt-account-id", context.account_id)?; insert_optional_header(&mut headers, "cookie", context.cookie_header)?; diff --git a/backend/src/upstream/openai/transport/response_meta.rs b/backend/src/upstream/openai/transport/response_meta.rs index b18627fcc..f7727ff7d 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_semantic_output_ms, + events::is_rate_limit_header_name, responses::response_event_signals, sse::SseEventDecoder, }; use super::{client::CodexResponseMetadata, diagnostics::CodexUpstreamDiagnostics}; @@ -106,10 +106,40 @@ fn upsert_client_header(headers: &mut Vec<(String, String)>, header: (String, St } } -pub(super) fn update_first_token_ms( +pub(super) fn update_response_timing_ms( started_at: Instant, - body_bytes: &[u8], + decoder: &mut SseEventDecoder, + chunk: &[u8], first_token_ms: &mut Option, + first_reasoning_ms: &mut Option, + first_text_ms: &mut Option, ) { - update_first_response_semantic_output_ms(started_at, body_bytes, first_token_ms); + let Ok(events) = decoder.push(chunk) else { + return; + }; + for event in events { + let Ok(value) = serde_json::from_str::(&event.data) else { + continue; + }; + let event_type = event.event.as_deref().or_else(|| { + value + .get("type") + .and_then(serde_json::Value::as_str) + .filter(|value| !value.trim().is_empty()) + }); + let signals = response_event_signals(event_type, &value); + if !signals.semantic_output { + continue; + } + let elapsed_ms = crate::infra::time::elapsed_millis_i64(started_at).max(1); + if first_token_ms.is_none() { + *first_token_ms = Some(elapsed_ms); + } + if signals.reasoning_output && first_reasoning_ms.is_none() { + *first_reasoning_ms = Some(elapsed_ms); + } + if signals.text_output && first_text_ms.is_none() { + *first_text_ms = Some(elapsed_ms); + } + } } diff --git a/backend/src/upstream/openai/transport/websocket.rs b/backend/src/upstream/openai/transport/websocket.rs index 50eff519d..7c05459f7 100644 --- a/backend/src/upstream/openai/transport/websocket.rs +++ b/backend/src/upstream/openai/transport/websocket.rs @@ -26,8 +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_semantic_output, - transport_requirement, + CodexResponsesRequest, TransportRequirement, transport_requirement, }; use crate::upstream::openai::protocol::sse::SseError; use crate::upstream::openai::protocol::websocket::{ diff --git a/backend/src/upstream/openai/transport/websocket_frames.rs b/backend/src/upstream/openai/transport/websocket_frames.rs index 371528d6d..202c6b307 100644 --- a/backend/src/upstream/openai/transport/websocket_frames.rs +++ b/backend/src/upstream/openai/transport/websocket_frames.rs @@ -1,5 +1,7 @@ //! WebSocket 帧交换、流转发与错误类型。 +use crate::upstream::openai::protocol::sse::SseEventDecoder; + use super::*; /// Responses WebSocket exchange result. @@ -17,6 +19,10 @@ pub struct CodexWebSocketExchange { pub rate_limit_headers: Vec<(String, String)>, /// 首个有效上游 WebSocket 事件到达代理的耗时。 pub first_token_ms: Option, + /// 首个 reasoning 输出事件到达代理的耗时。 + pub first_reasoning_ms: Option, + /// 首个正文输出事件到达代理的耗时。 + pub first_text_ms: Option, /// 首个上游协议事件到达代理的耗时。 pub first_event_ms: Option, /// WebSocket 连接池决策。 @@ -272,7 +278,10 @@ pub(super) async fn collect_websocket_response( let mut body = String::new(); let mut saw_upstream_activity = false; let mut first_token_ms = None; + let mut first_reasoning_ms = None; + let mut first_text_ms = None; let mut first_event_ms = None; + let mut output_decoder = SseEventDecoder::default(); loop { let receive_timeout = receive_idle_timeout(saw_upstream_activity, initial_event_timeout); @@ -340,15 +349,15 @@ pub(super) async fn collect_websocket_response( _ => None, }; let forwarded = if let Some(frame) = websocket_event_to_sse_frame(&raw) { - let has_semantic_output = response_body_has_semantic_output(frame.as_bytes()); body.push_str(&frame); - if has_semantic_output { - response_meta::update_first_token_ms( - started_at, - body.as_bytes(), - &mut first_token_ms, - ); - } + response_meta::update_response_timing_ms( + started_at, + &mut output_decoder, + frame.as_bytes(), + &mut first_token_ms, + &mut first_reasoning_ms, + &mut first_text_ms, + ); true } else { false @@ -362,6 +371,8 @@ pub(super) async fn collect_websocket_response( set_cookie_headers: metadata.set_cookie_headers.clone(), rate_limit_headers: metadata.rate_limit_headers.clone(), first_token_ms, + first_reasoning_ms, + first_text_ms, first_event_ms, pool_decision: None, connection_local_continuation: false, diff --git a/backend/tests/api/admin/dashboard_routes.rs b/backend/tests/api/admin/dashboard_routes.rs index 261e774c3..36681e22b 100644 --- a/backend/tests/api/admin/dashboard_routes.rs +++ b/backend/tests/api/admin/dashboard_routes.rs @@ -8,7 +8,7 @@ use codex_proxy_rs::{ bootstrap::services::Services, infra::time::{china_datetime, china_day_start, china_quarter_hour_start}, telemetry::{usage::store::PgUsageRecordStore, usage::types::UsageRecord}, - upstream::openai::profile::CodexWireProfile, + upstream::openai::profile::CodexWireProfileState, }; use serde_json::{Value, json}; use sqlx::PgPool; @@ -426,15 +426,23 @@ async fn dashboard_summary_should_match_versioned_model_names_for_billing() { } #[tokio::test] -async fn dashboard_summary_should_return_backend_formatted_time_fields() { +async fn dashboard_summary_should_return_backend_formatted_and_latency_time_fields() { let (app, store, _pool, _dir) = dashboard_test_app("dashboard-backend-formatted-time", test_wire_profile()).await; let record_at = Utc::now() - Duration::seconds(1); let mut record = usage_record_with_tokens(record_at, 10); record.metadata = json!({ "reasoningEffort": "max", - "reasoningPreset": "ultra" + "reasoningPreset": "ultra", + "firstReasoningMs": 5020, + "firstTextMs": 9620, + "transportDecisionWaitMs": 90, + "upstreamHeadersMs": 460, + "firstEventMs": 520, + "openaiProcessingMs": 24600 }); + record.first_token_ms = Some(5020); + record.latency_ms = Some(27_600); store.append(&record).await.unwrap(); let body = dashboard_summary(app).await; @@ -447,6 +455,20 @@ async fn dashboard_summary_should_return_backend_formatted_time_fields() { assert_eq!(body["data"]["usageRecords"][0]["reasoningPreset"], "ultra"); assert!(body["data"]["usageRecords"][0]["subagentKind"].is_null()); assert!(body["data"]["usageRecords"][0].get("metadata").is_none()); + assert_eq!(body["data"]["usageRecords"][0]["firstTokenLatencyMs"], 5020); + assert_eq!(body["data"]["usageRecords"][0]["latencyMs"], 27_600); + assert_eq!( + body["data"]["usageRecords"][0]["latencyDetails"]["firstReasoningMs"], + 5020 + ); + assert_eq!( + body["data"]["usageRecords"][0]["latencyDetails"]["firstTextMs"], + 9620 + ); + assert_eq!( + body["data"]["usageRecords"][0]["latencyDetails"]["openaiProcessingMs"], + 24600 + ); } #[tokio::test] @@ -738,7 +760,7 @@ async fn dashboard_summary_total_billing_for_usage_record( async fn dashboard_test_app( db_name: &str, - wire_profile: std::sync::Arc, + wire_profile: CodexWireProfileState, ) -> ( axum::Router, PgUsageRecordStore, diff --git a/backend/tests/bootstrap/config/mod.rs b/backend/tests/bootstrap/config/mod.rs index 743b01702..90cde1639 100644 --- a/backend/tests/bootstrap/config/mod.rs +++ b/backend/tests/bootstrap/config/mod.rs @@ -150,7 +150,7 @@ fn config_loader_should_reject_removed_tls_section() { #[test] fn config_loader_should_reject_missing_explicit_fields() { - let yaml = complete_config().replace(" enabled: true\n\n\n# Compose", "\n\n# Compose"); + let yaml = complete_config().replacen(" telemetry:\n enabled: true\n", "", 1); let directory = write_config(&yaml); let error = BootstrapConfig::load_from_path(directory.path().join("config.yaml")).unwrap_err(); diff --git a/backend/tests/bootstrap/tasks/desktop_release_update.rs b/backend/tests/bootstrap/tasks/desktop_release_update.rs index 8dd1f07e0..55dd9a1dc 100644 --- a/backend/tests/bootstrap/tasks/desktop_release_update.rs +++ b/backend/tests/bootstrap/tasks/desktop_release_update.rs @@ -6,6 +6,7 @@ async fn desktop_release_update_task_should_start_background_checker() { let handle = codex_proxy_rs::bootstrap::tasks::desktop_release_update::start_desktop_release_update_task( status, + crate::support::wire_profile::test_wire_profile(), "https://example.invalid/appcast.xml".to_string(), ); @@ -36,9 +37,11 @@ async fn desktop_release_update_task_should_record_initial_appcast_observation() .await; let status = DesktopReleaseStatus::default(); + let wire_profile = crate::support::wire_profile::test_wire_profile(); let handle = codex_proxy_rs::bootstrap::tasks::desktop_release_update::start_desktop_release_update_task( status.clone(), + wire_profile.clone(), format!("{}/appcast.xml", server.uri()), ); @@ -53,6 +56,9 @@ async fn desktop_release_update_task_should_record_initial_appcast_observation() assert!(release.signature_present); assert!(snapshot.checked_at.is_some()); assert!(snapshot.last_error.is_none()); + let profile = wire_profile.snapshot(); + assert_eq!(profile.desktop_version, "26.900.1"); + assert_eq!(profile.desktop_build, "7001"); } #[tokio::test] @@ -65,11 +71,12 @@ async fn desktop_release_update_task_should_record_fetch_failure_without_mutatin .await; let profile = crate::support::wire_profile::test_wire_profile(); - let original_user_agent = profile.user_agent(); + let original_profile = profile.snapshot(); let status = DesktopReleaseStatus::default(); let handle = codex_proxy_rs::bootstrap::tasks::desktop_release_update::start_desktop_release_update_task( status.clone(), + profile.clone(), format!("{}/appcast.xml", server.uri()), ); @@ -81,6 +88,6 @@ async fn desktop_release_update_task_should_record_fetch_failure_without_mutatin } handle.shutdown().await; - assert_eq!(profile.user_agent(), original_user_agent); + assert_eq!(profile.snapshot(), original_profile); assert!(status.snapshot().latest.is_none()); } diff --git a/backend/tests/dispatch/service/usage_logging.rs b/backend/tests/dispatch/service/usage_logging.rs index 168daff1d..e75f92bc4 100644 --- a/backend/tests/dispatch/service/usage_logging.rs +++ b/backend/tests/dispatch/service/usage_logging.rs @@ -119,6 +119,7 @@ async fn responses_should_use_imported_account_record_usage_cookie_and_usage_rec "set-cookie", "cf_clearance=new; Domain=.chatgpt.com; Path=/", ) + .insert_header("openai-processing-ms", "42") .set_body_string(RESPONSES_COMPLETED_USAGE_SSE), ) .expect(1) @@ -188,6 +189,9 @@ async fn responses_should_use_imported_account_record_usage_cookie_and_usage_rec assert_eq!(metadata["stream"], false); assert_eq!(metadata["reasoningEffort"], "max"); assert_eq!(metadata["reasoningPreset"], "ultra"); + assert_eq!(metadata["firstTextMs"], event.first_token_ms.unwrap()); + assert!(metadata.get("firstReasoningMs").is_none()); + assert_eq!(metadata["openaiProcessingMs"], 42); assert_eq!(event.input_tokens, Some(7)); } @@ -689,6 +693,8 @@ async fn responses_stream_should_record_usage_record_after_completed_stream() { event.first_token_ms.is_some_and(|value| value > 0), "stream usage metadata should include first-token latency: {metadata:?}", ); + assert_eq!(metadata["firstTextMs"], event.first_token_ms.unwrap()); + assert!(metadata.get("firstReasoningMs").is_none()); 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/support/wire_profile.rs b/backend/tests/support/wire_profile.rs index 4086040fa..ab4debb21 100644 --- a/backend/tests/support/wire_profile.rs +++ b/backend/tests/support/wire_profile.rs @@ -1,8 +1,6 @@ -use std::sync::Arc; - use codex_proxy_rs::{ bootstrap::{config::WireProfileConfig, services::wire_profile_from_config}, - upstream::openai::profile::CodexWireProfile, + upstream::openai::profile::{CodexWireProfile, CodexWireProfileState}, }; use crate::support::storage::timestamp; @@ -25,10 +23,10 @@ pub(crate) fn test_wire_profile_value() -> CodexWireProfile { wire_profile_from_config(&test_wire_profile_config()) } -pub(crate) fn test_wire_profile() -> Arc { - Arc::new(test_wire_profile_value()) +pub(crate) fn test_wire_profile() -> CodexWireProfileState { + CodexWireProfileState::new(test_wire_profile_value()) } -pub(crate) fn wire_profile(profile: CodexWireProfile) -> Arc { - Arc::new(profile) +pub(crate) fn wire_profile(profile: CodexWireProfile) -> CodexWireProfileState { + CodexWireProfileState::new(profile) } diff --git a/backend/tests/upstream/openai/desktop_release.rs b/backend/tests/upstream/openai/desktop_release.rs index ca9b0602c..d0e91ad6b 100644 --- a/backend/tests/upstream/openai/desktop_release.rs +++ b/backend/tests/upstream/openai/desktop_release.rs @@ -64,8 +64,33 @@ fn appcast_parser_should_reject_release_without_codex_desktop_version() { assert!(error.to_string().contains("shortVersionString")); } +#[test] +fn appcast_parser_should_reject_non_numeric_desktop_version() { + let error = parse_latest_desktop_release(&APPCAST.replace("26.707.72221", "26.707.beta")) + .expect_err("non-numeric version should fail"); + + assert_eq!( + error.to_string(), + "Codex Desktop appcast field `shortVersionString` is invalid" + ); +} + +#[test] +fn appcast_parser_should_reject_non_numeric_desktop_build() { + let error = parse_latest_desktop_release(&APPCAST.replace( + "5307", + "build", + )) + .expect_err("non-numeric build should fail"); + + assert_eq!( + error.to_string(), + "Codex Desktop appcast field `version` is invalid" + ); +} + #[tokio::test] -async fn desktop_release_checker_should_update_observation_only() { +async fn desktop_release_checker_should_sync_latest_release_to_wire_profile() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/appcast.xml")) @@ -73,10 +98,12 @@ async fn desktop_release_checker_should_update_observation_only() { .mount(&server) .await; let status = DesktopReleaseStatus::default(); + let wire_profile = crate::support::wire_profile::test_wire_profile(); let checker = DesktopReleaseChecker::with_client( reqwest::Client::new(), format!("{}/appcast.xml", server.uri()), status.clone(), + wire_profile.clone(), ); checker.check_and_record().await.expect("check appcast"); @@ -91,4 +118,7 @@ async fn desktop_release_checker_should_update_observation_only() { ); assert!(snapshot.checked_at.is_some()); assert!(snapshot.last_error.is_none()); + let profile = wire_profile.snapshot(); + assert_eq!(profile.desktop_version, "26.707.72221"); + assert_eq!(profile.desktop_build, "5307"); } diff --git a/backend/tests/upstream/openai/protocol/sse.rs b/backend/tests/upstream/openai/protocol/sse.rs index abbfa6d09..e35c876d1 100644 --- a/backend/tests/upstream/openai/protocol/sse.rs +++ b/backend/tests/upstream/openai/protocol/sse.rs @@ -60,6 +60,8 @@ fn response_event_signals_should_count_text_inside_completed_output_item() { ); assert!(signals.semantic_output); + assert!(signals.text_output); + assert!(!signals.reasoning_output); } #[test] @@ -70,6 +72,8 @@ fn response_event_signals_should_count_reasoning_as_semantic_output() { ); assert!(signals.semantic_output); + assert!(signals.reasoning_output); + assert!(!signals.text_output); } #[test] diff --git a/backend/tests/upstream/openai/transport/headers.rs b/backend/tests/upstream/openai/transport/headers.rs index f0b757498..eb7054ce6 100644 --- a/backend/tests/upstream/openai/transport/headers.rs +++ b/backend/tests/upstream/openai/transport/headers.rs @@ -508,7 +508,7 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m } #[tokio::test] -async fn codex_backend_client_should_send_effective_wire_profile_user_agent() { +async fn codex_backend_client_should_send_latest_wire_profile_user_agent() { let server = wiremock::MockServer::start().await; let sse_body = include_str!("../../../fixtures/responses/http_sse/completed_usage_basic.sse"); wiremock::Mock::given(wiremock::matchers::method("POST")) @@ -522,12 +522,13 @@ async fn codex_backend_client_should_send_effective_wire_profile_user_agent() { .await; let profile = crate::support::wire_profile::test_wire_profile(); - let expected_user_agent = profile.user_agent(); let client = CodexBackendClient::new( reqwest::Client::builder().no_proxy().build().unwrap(), server.uri(), - profile, + profile.clone(), ); + profile.update_desktop_release("26.900.1", "7001"); + let expected_user_agent = profile.snapshot().user_agent(); let mut request = codex_proxy_rs::upstream::openai::protocol::responses::CodexResponsesRequest::new_http_sse( "gpt-5.5", @@ -681,7 +682,7 @@ async fn codex_backend_client_models_should_use_original_auxiliary_headers() { .mount(&server) .await; let profile = crate::support::wire_profile::test_wire_profile(); - let client_version = profile.codex_version.clone(); + let client_version = profile.snapshot().codex_version; let client = CodexBackendClient::new( reqwest::Client::builder().no_proxy().build().unwrap(), server.uri(), diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 853b582f2..ac623a341 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -49,7 +49,7 @@ x-cpr: wire_profile: # 该画像来自 Desktop 制品与 bundled Codex Core 的联合审计。 - # appcast 只报告新 Desktop 制品，不会自动改写这里的请求身份。 + # appcast 成功检查后会自动同步 Desktop 版本和 Build；其余字段保持此处配置。 originator: 'Codex Desktop' codex_version: '0.144.2' desktop_version: '26.707.72221' @@ -81,7 +81,6 @@ x-cpr: telemetry: enabled: true - # Compose 凭据桥接区。通常不需要修改。 services: postgres-credentials: diff --git a/frontend/src/views/dashboard/components/UsageRecordCard.vue b/frontend/src/views/dashboard/components/UsageRecordCard.vue index dedcb56b4..d3c8df414 100644 --- a/frontend/src/views/dashboard/components/UsageRecordCard.vue +++ b/frontend/src/views/dashboard/components/UsageRecordCard.vue @@ -12,6 +12,7 @@ import { } from '@/views/usage/constants' import UsageClientIpCell from '@/views/usage/components/UsageClientIpCell.vue' import UsageBillingCell from '@/views/usage/components/UsageBillingCell.vue' +import UsageLatencyCell from '@/views/usage/components/UsageLatencyCell.vue' import UsageModelCell from '@/views/usage/components/UsageModelCell.vue' import UsageReasoningEffortCell from '@/views/usage/components/UsageReasoningEffortCell.vue' import UsageTokenCell from '@/views/usage/components/UsageTokenCell.vue' @@ -91,6 +92,10 @@ const dashboardUsageRecordColumns = usageRecordColumns.filter((column) => column + + diff --git a/frontend/src/views/usage/components/UsageLatencyCell.vue b/frontend/src/views/usage/components/UsageLatencyCell.vue new file mode 100644 index 000000000..cf62ab519 --- /dev/null +++ b/frontend/src/views/usage/components/UsageLatencyCell.vue @@ -0,0 +1,91 @@ + + + diff --git a/frontend/src/views/usage/constants.ts b/frontend/src/views/usage/constants.ts index 326ab5478..d93c90301 100644 --- a/frontend/src/views/usage/constants.ts +++ b/frontend/src/views/usage/constants.ts @@ -1,3 +1,5 @@ +import { formatDuration } from './utils/format' + export const usageRecordColumns = [ { key: 'accountEmail', @@ -47,20 +49,11 @@ export const usageRecordColumns = [ ellipsis: false, }, { - key: 'firstTokenLatencyMsDisplay', - label: '首字', - width: '104px', + key: 'latency', + label: '延迟', + width: '156px', align: 'right' as const, ellipsis: false, - cellClass: 'font-mono text-[12px] font-[650] tabular-nums text-(--cp-text-secondary)', - }, - { - key: 'latencyMsDisplay', - label: '耗时', - width: '96px', - align: 'right' as const, - ellipsis: false, - cellClass: 'font-mono text-[12px] font-[650] tabular-nums text-(--cp-text-secondary)', }, { key: 'createdAtDisplay', @@ -301,6 +294,52 @@ export function usageTokenDetails(record: any) { } } +export function usageLatencyDetails(record: any) { + const latencyDetails = record?.latencyDetails || record?.metadata || {} + const firstTokenMs = durationValue(record?.firstTokenLatencyMs ?? latencyDetails.firstTokenMs) + const totalMs = durationValue(record?.latencyMs) + const firstReasoningMs = durationValue(latencyDetails.firstReasoningMs) + const firstTextMs = durationValue(latencyDetails.firstTextMs) + const breakdownItems = [] + + if (firstTokenMs !== null && totalMs !== null && firstTokenMs <= totalMs) { + breakdownItems.push({ label: '首字等待', value: formatDuration(firstTokenMs) }) + + if (firstTextMs !== null && firstTextMs >= firstTokenMs && firstTextMs <= totalMs) { + const beforeTextMs = firstTextMs - firstTokenMs + if (beforeTextMs > 0) { + breakdownItems.push({ + label: firstReasoningMs === firstTokenMs ? '推理到正文' : '首个输出到正文', + value: formatDuration(beforeTextMs), + }) + } + breakdownItems.push({ label: '正文生成', value: formatDuration(totalMs - firstTextMs) }) + } else { + breakdownItems.push({ + label: '首个输出后完成', + value: formatDuration(totalMs - firstTokenMs), + }) + } + } + + const transportItems = [ + { label: '传输决策等待', value: durationValue(latencyDetails.transportDecisionWaitMs) }, + { label: 'WebSocket 连接', value: durationValue(latencyDetails.wsConnectMs) }, + { label: '上游响应头', value: durationValue(latencyDetails.upstreamHeadersMs) }, + { label: '首个上游事件', value: durationValue(latencyDetails.firstEventMs) }, + { label: '上游处理', value: durationValue(latencyDetails.openaiProcessingMs) }, + ] + .filter((item) => item.value !== null) + .map((item) => ({ ...item, value: formatDuration(item.value) })) + + return { + firstTokenDisplay: formatDuration(firstTokenMs), + totalDisplay: formatDuration(totalMs), + breakdownItems, + transportItems, + } +} + export function usageBilling(record: any) { return record?.billing || null } @@ -342,6 +381,10 @@ function numberValue(value: unknown) { return typeof value === 'number' && Number.isFinite(value) ? value : 0 } +function durationValue(value: unknown) { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 ? value : null +} + function extractInputText(body: any) { const input = body?.input if (typeof input === 'string') { diff --git a/frontend/src/views/usage/index.vue b/frontend/src/views/usage/index.vue index 20213ad8a..6fb342092 100644 --- a/frontend/src/views/usage/index.vue +++ b/frontend/src/views/usage/index.vue @@ -23,6 +23,7 @@ import UsageClientIpCell from './components/UsageClientIpCell.vue' import UsageBillingCell from './components/UsageBillingCell.vue' import UsageFilters from './components/UsageFilters.vue' import UsageInsightsGrid from './components/UsageInsightsGrid.vue' +import UsageLatencyCell from './components/UsageLatencyCell.vue' import UsageModelCell from './components/UsageModelCell.vue' import OpsErrorPanel from './components/OpsErrorPanel.vue' import UsageRecordDetailModal from './components/UsageRecordDetailModal.vue' @@ -201,6 +202,10 @@ watch(timeRange, () => { + +