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): expose latency breakdowns

Browse Source 
Capture first semantic output and transport timing, then surface concise latency values with safe phase details in dashboard and usage records.

Keep the Codex Desktop request profile synchronized with successful appcast checks at runtime, without rewriting configuration or evicting existing WebSocket connections.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-16 18:07:08 +08:00
co-authored by Codex
parent 9e57f78bab
commit 549bb9f62b
31 changed files with 715 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/admin/dashboard_routes.rs
+37 -2
View File
@@ -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<String>,
token_details: UsageRecordTokenDetailsData,
billing: Option<UsageRecordBillingData>,
first_token_latency_ms: Option<i64>,
latency_ms: Option<i64>,
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<i64>,
first_text_ms: Option<i64>,
transport_decision_wait_ms: Option<i64>,
ws_connect_ms: Option<i64>,
upstream_headers_ms: Option<i64>,
first_event_ms: Option<i64>,
openai_processing_ms: Option<i64>,
}
#[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<i64> {
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"
backend/src/api/router.rs
+2 -2
View File
@@ -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<TokenRefreshService<OpenAiTokenClient>>,
pub responses: Arc<ResponseDispatchService>,
pub session_affinity: Arc<SessionAffinityService>,
pub wire_profile: Arc<CodexWireProfile>,
pub wire_profile: CodexWireProfileState,
pub desktop_release: DesktopReleaseStatus,
pub process_control: Arc<dyn ProcessControl>,
pub system_update: Arc<SystemUpdateService>,
backend/src/bootstrap/services.rs
+7 -7
View File
@@ -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<SessionAffinityService>,
pub codex: Arc<CodexBackendClient>,
pub websocket_pool: Option<Arc<CodexWebSocketPool>>,
pub wire_profile: Arc<CodexWireProfile>,
pub wire_profile: CodexWireProfileState,
pub desktop_release: DesktopReleaseStatus,
pub account_pseudonymizer: Arc<AccountPseudonymizer>,
pub system_update: Arc<SystemUpdateService>,
@@ -146,7 +146,7 @@ impl Services {
pub fn new(
config: &AppConfig,
stores: BackgroundTaskStores,
wire_profile: Arc<CodexWireProfile>,
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<CodexWireProfile>,
wire_profile: CodexWireProfileState,
) -> Result<Self, CustomCaError> {
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<CodexWireProfile>,
wire_profile: CodexWireProfileState,
usage_record_options: UsageRecordOptions,
) -> Result<Self, CustomCaError> {
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<CodexWireProfile>,
wire_profile: CodexWireProfileState,
identity_secret: [u8; 32],
usage_record_options: UsageRecordOptions,
) -> Result<Self, CustomCaError> {
@@ -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,
backend/src/bootstrap/tasks/coordinator.rs
+1
View File
@@ -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(),
),
);
backend/src/bootstrap/tasks/desktop_release_update.rs
+9 -2
View File
@@ -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 首次发布检查失败");
backend/src/dispatch/controllers/telemetry/events.rs
+39 -10
View File
@@ -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<i64>) {
let Some(first_token_ms) = first_token_ms else {
pub(super) fn insert_output_timing_ms(
metadata: &mut Value,
first_token_ms: Option<i64>,
first_reasoning_ms: Option<i64>,
first_text_ms: Option<i64>,
) {
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<i64> {
response_metadata
.client_headers
.iter()
.find(|(name, _)| name.eq_ignore_ascii_case("openai-processing-ms"))
.and_then(|(_, value)| value.parse::<i64>().ok())
.filter(|value| *value >= 0)
}
pub(super) fn insert_websocket_pool_decision(
backend/src/dispatch/controllers/telemetry/mod.rs
+47 -15
View File
@@ -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<i64>,
@@ -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,
backend/src/dispatch/lifecycle/stream.rs
+2
View File
@@ -28,6 +28,8 @@ pub(in crate::dispatch) struct StreamSummary {
pub body: Vec<u8>,
pub terminal_chunks: Vec<CanonicalResponseChunk>,
pub first_token_ms: Option<i64>,
pub first_reasoning_ms: Option<i64>,
pub first_text_ms: Option<i64>,
pub first_event_ms: i64,
pub usage: Option<TokenUsage>,
pub last_response_id: Option<String>,
backend/src/dispatch/stream/live.rs
+48 -9
View File
@@ -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<i64>,
first_reasoning_ms: &mut Option<i64>,
first_text_ms: &mut Option<i64>,
) {
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<u8>,
terminal_chunks: Vec<CanonicalResponseChunk>,
first_token_ms: Option<i64>,
output_timing: (Option<i64>, Option<i64>, Option<i64>),
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),
backend/src/dispatch/transport/canonical.rs
+14
View File
@@ -225,6 +225,8 @@ pub(in crate::dispatch) struct CanonicalStreamDecoder {
pending: Vec<u8>,
forwardable_frame_seen: bool,
first_semantic_output_seen: bool,
first_reasoning_output_seen: bool,
first_text_output_seen: bool,
terminal: Option<CanonicalStreamTerminal>,
transport_done_seen: bool,
completed_usage: Option<TokenUsage>,
@@ -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());
}
backend/src/upstream/openai/desktop_release.rs
+23 -1
View File
@@ -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<String>,
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<DesktopRelease, Desktop
let version = child_text(item, "shortVersionString")
.or_else(|| local_attribute(enclosure, "shortVersionString"))
.ok_or(DesktopReleaseError::MissingField("shortVersionString"))?;
if !is_numeric_dotted_version(&version) {
return Err(DesktopReleaseError::InvalidField("shortVersionString"));
}
let build = child_text(item, "version")
.or_else(|| local_attribute(enclosure, "version"))
.ok_or(DesktopReleaseError::MissingField("version"))?;
if !build.bytes().all(|byte| byte.is_ascii_digit()) {
return Err(DesktopReleaseError::InvalidField("version"));
}
let published_at = child_text(item, "pubDate")
.map(|value| {
DateTime::parse_from_rfc2822(&value)
@@ -187,6 +200,15 @@ fn local_attribute(node: Option<Node<'_, '_>>, local_name: &str) -> Option<Strin
.map(ToString::to_string)
}
fn is_numeric_dotted_version(value: &str) -> 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}")]
backend/src/upstream/openai/profile.rs
+41 -3
View File
@@ -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<RwLock<CodexWireProfile>>,
}
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();
}
}
backend/src/upstream/openai/protocol/responses.rs
+107 -68
View File
@@ -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 = "<multi_agent_mode>";
@@ -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::<Value>(&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<i64>,
) {
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 {
backend/src/upstream/openai/transport/client.rs
+6 -2
View File
@@ -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<i64>,
/// 首个 reasoning 输出事件到达代理的耗时。
pub first_reasoning_ms: Option<i64>,
/// 首个正文输出事件到达代理的耗时。
pub first_text_ms: Option<i64>,
/// WebSocket 连接池决策。
pub websocket_pool_decision: Option<WebSocketPoolDecision>,
/// 上游诊断元数据。
@@ -382,7 +386,7 @@ pub struct CodexBackendStreamingResponse {
pub struct CodexBackendClient {
pub(super) client: Client,
pub(super) base_url: String,
profile: Arc<CodexWireProfile>,
profile: CodexWireProfileState,
websocket_pool: Option<Arc<CodexWebSocketPool>>,
websocket_initial_event_timeout: Option<Duration>,
websocket_fast_path_budget: Duration,
backend/src/upstream/openai/transport/client_sse.rs
+34 -13
View File
@@ -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<String>,
profile: Arc<CodexWireProfile>,
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<Vec<Value>> {
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<HeaderMap> {
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<HeaderMap> {
fn request_headers(
&self,
profile: &CodexWireProfile,
context: CodexRequestContext<'_>,
) -> CodexClientResult<HeaderMap> {
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<HeaderMap> {
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<HeaderMap> {
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)?;
backend/src/upstream/openai/transport/response_meta.rs
+34 -4
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_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<i64>,
first_reasoning_ms: &mut Option<i64>,
first_text_ms: &mut Option<i64>,
) {
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::<serde_json::Value>(&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);
}
}
}
backend/src/upstream/openai/transport/websocket.rs
+1 -2
View File
@@ -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::{
backend/src/upstream/openai/transport/websocket_frames.rs
+19 -8
View File
@@ -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<i64>,
/// 首个 reasoning 输出事件到达代理的耗时。
pub first_reasoning_ms: Option<i64>,
/// 首个正文输出事件到达代理的耗时。
pub first_text_ms: Option<i64>,
/// 首个上游协议事件到达代理的耗时。
pub first_event_ms: Option<i64>,
/// 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,
backend/tests/api/admin/dashboard_routes.rs
+26 -4
View File
@@ -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<CodexWireProfile>,
wire_profile: CodexWireProfileState,
) -> (
axum::Router,
PgUsageRecordStore,
backend/tests/bootstrap/config/mod.rs
+1 -1
View File
@@ -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();
backend/tests/bootstrap/tasks/desktop_release_update.rs
+9 -2
View File
@@ -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());
}
backend/tests/dispatch/service/usage_logging.rs
+6
View File
@@ -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",
)
backend/tests/support/wire_profile.rs
+5 -7
View File
@@ -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<CodexWireProfile> {
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<CodexWireProfile> {
Arc::new(profile)
pub(crate) fn wire_profile(profile: CodexWireProfile) -> CodexWireProfileState {
CodexWireProfileState::new(profile)
}
backend/tests/upstream/openai/desktop_release.rs
+31 -1
View File
@@ -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(
"<sparkle:version>5307</sparkle:version>",
"<sparkle:version>build</sparkle:version>",
))
.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");
}
backend/tests/upstream/openai/protocol/sse.rs
+4
View File
@@ -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]
backend/tests/upstream/openai/transport/headers.rs
+5 -4
View File
@@ -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(),
deploy/config.example.yaml
+1 -2
View File
@@ -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:
frontend/src/views/dashboard/components/UsageRecordCard.vue
+5
View File
@@ -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
<template #billing="{ row }">
<UsageBillingCell :record="row" />
</template>
<template #latency="{ row }">
<UsageLatencyCell :record="row" />
</template>
</BaseTable>
</div>
</template>
frontend/src/views/usage/components/UsageLatencyCell.vue
+91
View File
@@ -0,0 +1,91 @@
<script setup lang="ts">
import { Info } from '@lucide/vue'
import { computed } from 'vue'
import BasePopover from '@/components/base/BasePopover.vue'
import { usageLatencyDetails } from '../constants'
const props = defineProps<{
record: any
}>()
const latencyDetails = computed(() => usageLatencyDetails(props.record))
</script>
<template>
<div class="flex items-center justify-end gap-1.5">
<div
class="grid grid-cols-[auto_auto] items-center justify-end gap-x-2 gap-y-1.5 font-mono text-[12px] leading-none font-[760] tabular-nums"
>
<span class="text-[11px] text-(--cp-text-muted)">首字</span>
<span class="text-(--cp-text-secondary)">{{ latencyDetails.firstTokenDisplay }}</span>
<span class="text-[11px] text-(--cp-text-muted)">总耗时</span>
<span class="text-(--cp-text-primary)">{{ latencyDetails.totalDisplay }}</span>
</div>
<BasePopover
trigger="hover"
placement="right"
width="252px"
panel-class="!p-3 text-(--cp-text-primary)"
>
<template #trigger>
<button
type="button"
class="inline-flex size-4 items-center justify-center rounded-full bg-(--cp-info-bg) text-(--cp-info) outline-none hover:bg-(--cp-default-bg-hover)"
aria-label="查看延迟明细"
>
<Info class="size-3" />
</button>
</template>
<div class="grid gap-2 text-[12px] leading-none">
<p class="m-0 font-[760] text-(--cp-text-primary)">延迟明细</p>
<div
v-if="latencyDetails.breakdownItems.length"
class="grid gap-1.5 text-(--cp-text-secondary)"
>
<div
v-for="item in latencyDetails.breakdownItems"
:key="item.label"
class="flex justify-between gap-4"
>
<span class="whitespace-nowrap">{{ item.label }}</span>
<span class="whitespace-nowrap font-mono font-[760] text-(--cp-text-primary)">
{{ item.value }}
</span>
</div>
</div>
<p v-else class="m-0 text-(--cp-text-muted)">此记录未采集完整的阶段耗时。</p>
<div class="flex justify-between border-t border-(--cp-divider-subtle) pt-2">
<span class="whitespace-nowrap text-(--cp-text-secondary)">总耗时</span>
<span class="whitespace-nowrap font-mono font-[760] text-(--cp-info-text)">
{{ latencyDetails.totalDisplay }}
</span>
</div>
<div
v-if="latencyDetails.transportItems.length"
class="grid gap-1.5 border-t border-(--cp-divider-subtle) pt-2 text-(--cp-text-secondary)"
>
<p class="m-0 font-[760] text-(--cp-text-primary)">传输观测</p>
<div
v-for="item in latencyDetails.transportItems"
:key="item.label"
class="flex justify-between gap-4"
>
<span class="whitespace-nowrap">{{ item.label }}</span>
<span class="whitespace-nowrap font-mono font-[760] text-(--cp-text-primary)">
{{ item.value }}
</span>
</div>
<p class="m-0 text-[11px] leading-snug text-(--cp-text-muted)">
与阶段耗时可能重叠，不参与总耗时相加。
</p>
</div>
</div>
</BasePopover>
</div>
</template>
frontend/src/views/usage/constants.ts
+55 -12
View File
@@ -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') {
frontend/src/views/usage/index.vue
+5
View File
@@ -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, () => {
<UsageBillingCell :record="row" />
</template>
<template #latency="{ row }">
<UsageLatencyCell :record="row" />
</template>
<template #actions="{ row }">
<div class="flex items-center justify-start">
<BaseButton