Preserve WebSocket TBT metric precision

This commit is contained in:
Tom Wiltzius
2026-07-09 10:12:37 -07:00
parent d72d669ca7
commit 709bc18391
13 changed files with 135 additions and 42 deletions
@@ -156,6 +156,8 @@ const X_REASONING_INCLUDED_HEADER: &str = "x-reasoning-included";
const OPENAI_MODEL_HEADER: &str = "openai-model";
const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE: &str = "websocket_connection_limit_reached";
const WEBSOCKET_CONNECTION_LIMIT_REACHED_MESSAGE: &str = "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue.";
const RESPONSES_WEBSOCKET_TIMING_KIND: &str = "responsesapi.websocket_timing";
const RESPONSES_WEBSOCKET_TIMING_EVENT_TARGET: &str = "codex_api::responses_websocket_timing";
pub struct ResponsesWebsocketConnection {
stream: Arc<Mutex<Option<WsStream>>>,
@@ -678,6 +680,7 @@ async fn run_websocket_response_stream(
continue;
}
};
emit_responses_websocket_timing_event(event.kind(), text.as_str());
if let Some(response_turn_state) = event.turn_state()
&& let Some(turn_state) = turn_state
{
@@ -761,6 +764,22 @@ async fn run_websocket_response_stream(
Ok(())
}
fn emit_responses_websocket_timing_event(kind: &str, payload: &str) {
if kind != RESPONSES_WEBSOCKET_TIMING_KIND {
return;
}
// This full payload is excluded from always-on sinks. Opt in to the request span's event with
// `RUST_LOG='codex_api::responses_websocket_timing=trace'`.
tracing::event!(
name: RESPONSES_WEBSOCKET_TIMING_KIND,
target: RESPONSES_WEBSOCKET_TIMING_EVENT_TARGET,
tracing::Level::TRACE,
payload,
"responses websocket timing"
);
}
fn safety_buffering_for_event(
event: &ResponsesStreamEvent,
treatment: &mut SafetyBufferingTreatment,
@@ -1237,8 +1237,8 @@ async fn responses_websocket_includes_timing_metrics_header_when_runtime_metrics
assert_eq!(summary.responses_api_inference_time_ms, 450);
assert_eq!(summary.responses_api_engine_iapi_ttft_ms, 310);
assert_eq!(summary.responses_api_engine_service_ttft_ms, 340);
assert_eq!(summary.responses_api_engine_iapi_tbt_ms, 220);
assert_eq!(summary.responses_api_engine_service_tbt_ms, 260);
assert_eq!(summary.responses_api_engine_iapi_tbt_ms, 220.0);
assert_eq!(summary.responses_api_engine_service_tbt_ms, 260.0);
server.shutdown().await;
}
+31 -7
View File
@@ -205,6 +205,21 @@ impl SessionTelemetry {
}
}
fn record_duration_ms_f64(&self, name: &str, duration_ms: f64, tags: &[(&str, &str)]) {
let res: MetricsResult<()> = (|| {
let Some(metrics) = &self.metrics else {
return Ok(());
};
let tags = self.tags_with_metadata(tags)?;
metrics.record_duration_ms_f64(name, duration_ms, &tags)
})();
if let Err(e) = res {
tracing::warn!("metrics duration [{name}] failed: {e}");
}
}
/// Records a coarse startup phase for production latency breakdowns.
pub fn record_startup_phase(
&self,
@@ -1170,16 +1185,20 @@ impl SessionTelemetry {
let engine_iapi_tbt_value =
timing_metrics.and_then(|value| value.get(RESPONSES_API_ENGINE_IAPI_TBT_FIELD));
if let Some(duration) = duration_from_ms_value(engine_iapi_tbt_value) {
self.record_duration(RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC, duration, &[]);
if let Some(duration_ms) = f64_ms_value(engine_iapi_tbt_value) {
self.record_duration_ms_f64(
RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC,
duration_ms,
&[],
);
}
let engine_service_tbt_value =
timing_metrics.and_then(|value| value.get(RESPONSES_API_ENGINE_SERVICE_TBT_FIELD));
if let Some(duration) = duration_from_ms_value(engine_service_tbt_value) {
self.record_duration(
if let Some(duration_ms) = f64_ms_value(engine_service_tbt_value) {
self.record_duration_ms_f64(
RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC,
duration,
duration_ms,
&[],
);
}
@@ -1234,6 +1253,12 @@ impl SessionTelemetry {
}
fn duration_from_ms_value(value: Option<&serde_json::Value>) -> Option<Duration> {
let ms = f64_ms_value(value)?;
let clamped = ms.min(u64::MAX as f64);
Some(Duration::from_millis(clamped.round() as u64))
}
fn f64_ms_value(value: Option<&serde_json::Value>) -> Option<f64> {
let value = value?;
let ms = value
.as_f64()
@@ -1242,6 +1267,5 @@ fn duration_from_ms_value(value: Option<&serde_json::Value>) -> Option<Duration>
if !ms.is_finite() || ms < 0.0 {
return None;
}
let clamped = ms.min(u64::MAX as f64);
Some(Duration::from_millis(clamped.round() as u64))
Some(ms)
}
+17
View File
@@ -401,6 +401,23 @@ impl MetricsClient {
)
}
/// Record a duration supplied as fractional milliseconds using a histogram.
pub(crate) fn record_duration_ms_f64(
&self,
name: &str,
duration_ms: f64,
tags: &[(&str, &str)],
) -> Result<()> {
self.0.duration_histogram(
name,
duration_ms,
MILLISECOND_DURATION_UNIT,
MILLISECOND_DURATION_DESCRIPTION,
MILLISECOND_DURATION_BOUNDARIES,
tags,
)
}
/// Record a duration in seconds using a histogram with an instrument description.
pub fn record_duration_seconds_with_description(
&self,
+17 -13
View File
@@ -38,7 +38,7 @@ impl RuntimeMetricTotals {
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, Default, PartialEq)]
pub struct RuntimeMetricsSummary {
pub tool_calls: RuntimeMetricTotals,
pub api_calls: RuntimeMetricTotals,
@@ -49,8 +49,8 @@ pub struct RuntimeMetricsSummary {
pub responses_api_inference_time_ms: u64,
pub responses_api_engine_iapi_ttft_ms: u64,
pub responses_api_engine_service_ttft_ms: u64,
pub responses_api_engine_iapi_tbt_ms: u64,
pub responses_api_engine_service_tbt_ms: u64,
pub responses_api_engine_iapi_tbt_ms: f64,
pub responses_api_engine_service_tbt_ms: f64,
pub turn_ttft_ms: u64,
pub turn_ttfm_ms: u64,
}
@@ -66,8 +66,8 @@ impl RuntimeMetricsSummary {
&& self.responses_api_inference_time_ms == 0
&& self.responses_api_engine_iapi_ttft_ms == 0
&& self.responses_api_engine_service_ttft_ms == 0
&& self.responses_api_engine_iapi_tbt_ms == 0
&& self.responses_api_engine_service_tbt_ms == 0
&& self.responses_api_engine_iapi_tbt_ms == 0.0
&& self.responses_api_engine_service_tbt_ms == 0.0
&& self.turn_ttft_ms == 0
&& self.turn_ttfm_ms == 0
}
@@ -90,10 +90,10 @@ impl RuntimeMetricsSummary {
if other.responses_api_engine_service_ttft_ms > 0 {
self.responses_api_engine_service_ttft_ms = other.responses_api_engine_service_ttft_ms;
}
if other.responses_api_engine_iapi_tbt_ms > 0 {
if other.responses_api_engine_iapi_tbt_ms > 0.0 {
self.responses_api_engine_iapi_tbt_ms = other.responses_api_engine_iapi_tbt_ms;
}
if other.responses_api_engine_service_tbt_ms > 0 {
if other.responses_api_engine_service_tbt_ms > 0.0 {
self.responses_api_engine_service_tbt_ms = other.responses_api_engine_service_tbt_ms;
}
if other.turn_ttft_ms > 0 {
@@ -146,9 +146,9 @@ impl RuntimeMetricsSummary {
let responses_api_engine_service_ttft_ms =
sum_histogram_ms(snapshot, RESPONSES_API_ENGINE_SERVICE_TTFT_DURATION_METRIC);
let responses_api_engine_iapi_tbt_ms =
sum_histogram_ms(snapshot, RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC);
sum_histogram_f64(snapshot, RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC);
let responses_api_engine_service_tbt_ms =
sum_histogram_ms(snapshot, RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC);
sum_histogram_f64(snapshot, RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC);
let turn_ttft_ms = sum_histogram_ms(snapshot, TURN_TTFT_DURATION_METRIC);
let turn_ttfm_ms = sum_histogram_ms(snapshot, TURN_TTFM_DURATION_METRIC);
Self {
@@ -189,21 +189,25 @@ fn sum_counter_metric(metric: &Metric) -> u64 {
}
fn sum_histogram_ms(snapshot: &ResourceMetrics, name: &str) -> u64 {
f64_to_u64(sum_histogram_f64(snapshot, name))
}
fn sum_histogram_f64(snapshot: &ResourceMetrics, name: &str) -> f64 {
snapshot
.scope_metrics()
.flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
.filter(|metric| metric.name() == name)
.map(sum_histogram_metric_ms)
.map(sum_histogram_metric)
.sum()
}
fn sum_histogram_metric_ms(metric: &Metric) -> u64 {
fn sum_histogram_metric(metric: &Metric) -> f64 {
match metric.data() {
AggregatedMetrics::F64(MetricData::Histogram(histogram)) => histogram
.data_points()
.map(|point| f64_to_u64(point.sum()))
.map(opentelemetry_sdk::metrics::data::HistogramDataPoint::sum)
.sum(),
_ => 0,
_ => 0.0,
}
}
+3 -3
View File
@@ -90,7 +90,7 @@ fn runtime_metrics_summary_collects_tool_api_and_streaming_metrics() -> Result<(
Option<std::result::Result<Message, tokio_tungstenite::tungstenite::Error>>,
codex_api::ApiError,
> = Ok(Some(Ok(Message::Text(
r#"{"type":"responsesapi.websocket_timing","timing_metrics":{"responses_duration_excl_engine_and_client_tool_time_ms":124,"engine_service_total_ms":457,"engine_iapi_ttft_total_ms":211,"engine_service_ttft_total_ms":233,"engine_iapi_tbt_across_engine_calls_ms":377,"engine_service_tbt_across_engine_calls_ms":399}}"#
r#"{"type":"responsesapi.websocket_timing","timing_metrics":{"responses_duration_excl_engine_and_client_tool_time_ms":124.25,"engine_service_total_ms":457,"engine_iapi_ttft_total_ms":211,"engine_service_ttft_total_ms":233,"engine_iapi_tbt_across_engine_calls_ms":2.450638,"engine_service_tbt_across_engine_calls_ms":5.267279}}"#
.into(),
))));
manager.record_websocket_event(&ws_timing_response, Duration::from_millis(20));
@@ -133,8 +133,8 @@ fn runtime_metrics_summary_collects_tool_api_and_streaming_metrics() -> Result<(
responses_api_inference_time_ms: 457,
responses_api_engine_iapi_ttft_ms: 211,
responses_api_engine_service_ttft_ms: 233,
responses_api_engine_iapi_tbt_ms: 377,
responses_api_engine_service_tbt_ms: 399,
responses_api_engine_iapi_tbt_ms: 2.450638,
responses_api_engine_service_tbt_ms: 5.267279,
turn_ttft_ms: 95,
turn_ttfm_ms: 180,
};
+15
View File
@@ -33,6 +33,21 @@ fn record_duration_records_histogram() -> Result<()> {
Ok(())
}
#[test]
fn record_duration_keeps_whole_millisecond_behavior() -> Result<()> {
let (metrics, exporter) = build_metrics_with_defaults(&[])?;
metrics.record_duration("codex.request_latency", Duration::from_micros(15_999), &[])?;
metrics.shutdown()?;
let resource_metrics = latest_metrics(&exporter);
let (_, _, sum, count) = histogram_data(&resource_metrics, "codex.request_latency");
assert_eq!(sum, 15.0);
assert_eq!(count, 1);
Ok(())
}
#[test]
fn record_duration_seconds_uses_fractional_seconds_and_scaled_buckets() -> Result<()> {
let (metrics, exporter) = build_metrics_with_defaults(&[])?;
+1
View File
@@ -58,6 +58,7 @@ pub fn default_filter() -> Targets {
.with_target("codex_otel.log_only", LevelFilter::OFF)
.with_target("codex_otel.trace_safe", LevelFilter::OFF)
.with_target("rmcp::service", LevelFilter::INFO)
.with_target("codex_api::responses_websocket_timing", LevelFilter::OFF)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+6 -6
View File
@@ -1,5 +1,4 @@
use pretty_assertions::assert_eq;
use tracing_subscriber::filter::Targets;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use uuid::Uuid;
@@ -16,17 +15,18 @@ async fn sqlite_sink_drops_low_level_opentelemetry_sdk_logs() {
let layer = start(runtime.clone());
let guard = tracing_subscriber::registry()
.with(
layer
.clone()
.with_filter(Targets::new().with_default(tracing::Level::TRACE)),
)
.with(layer.clone().with_filter(default_filter()))
.set_default();
tracing::trace!(target: "opentelemetry_sdk", "dropped-trace");
tracing::debug!(target: "opentelemetry_sdk", "dropped-debug");
tracing::info!(target: "opentelemetry_sdk", "retained-info");
tracing::trace!(target: "codex_state", "retained-trace");
tracing::trace!(
target: "codex_api::responses_websocket_timing",
payload = "complete timing payload",
"dropped-websocket-timing"
);
layer.flush().await;
drop(guard);
+2 -2
View File
@@ -2013,8 +2013,8 @@ fn has_websocket_timing_metrics(summary: RuntimeMetricsSummary) -> bool {
|| summary.responses_api_inference_time_ms > 0
|| summary.responses_api_engine_iapi_ttft_ms > 0
|| summary.responses_api_engine_service_ttft_ms > 0
|| summary.responses_api_engine_iapi_tbt_ms > 0
|| summary.responses_api_engine_service_tbt_ms > 0
|| summary.responses_api_engine_iapi_tbt_ms > 0.0
|| summary.responses_api_engine_service_tbt_ms > 0.0
}
impl Drop for ChatWidget {
@@ -3156,7 +3156,7 @@ async fn runtime_metrics_websocket_timing_logs_and_final_separator_sums_totals()
chat.on_task_started();
chat.apply_runtime_metrics_delta(RuntimeMetricsSummary {
responses_api_engine_iapi_ttft_ms: 120,
responses_api_engine_service_tbt_ms: 50,
responses_api_engine_service_tbt_ms: 50.0,
..RuntimeMetricsSummary::default()
});
+8 -6
View File
@@ -136,16 +136,18 @@ pub(crate) fn runtime_metrics_label(summary: RuntimeMetricsSummary) -> Option<St
}
parts.push(format!("TTFT: {}", ttft_parts.join(" ")));
}
if summary.responses_api_engine_iapi_tbt_ms > 0
|| summary.responses_api_engine_service_tbt_ms > 0
if summary.responses_api_engine_iapi_tbt_ms > 0.0
|| summary.responses_api_engine_service_tbt_ms > 0.0
{
let mut tbt_parts = Vec::new();
if summary.responses_api_engine_iapi_tbt_ms > 0 {
let duration = format_duration_ms(summary.responses_api_engine_iapi_tbt_ms);
if summary.responses_api_engine_iapi_tbt_ms > 0.0 {
let duration =
format_duration_ms(summary.responses_api_engine_iapi_tbt_ms.round() as u64);
tbt_parts.push(format!("{duration} (iapi)"));
}
if summary.responses_api_engine_service_tbt_ms > 0 {
let duration = format_duration_ms(summary.responses_api_engine_service_tbt_ms);
if summary.responses_api_engine_service_tbt_ms > 0.0 {
let duration =
format_duration_ms(summary.responses_api_engine_service_tbt_ms.round() as u64);
tbt_parts.push(format!("{duration} (service)"));
}
parts.push(format!("TBT: {}", tbt_parts.join(" ")));
+13 -2
View File
@@ -564,8 +564,8 @@ fn final_message_separator_hides_short_worked_label_and_includes_runtime_metrics
responses_api_inference_time_ms: 1_940,
responses_api_engine_iapi_ttft_ms: 410,
responses_api_engine_service_ttft_ms: 460,
responses_api_engine_iapi_tbt_ms: 1_180,
responses_api_engine_service_tbt_ms: 1_240,
responses_api_engine_iapi_tbt_ms: 1_180.0,
responses_api_engine_service_tbt_ms: 1_240.0,
turn_ttft_ms: 0,
turn_ttfm_ms: 0,
};
@@ -585,6 +585,17 @@ fn final_message_separator_hides_short_worked_label_and_includes_runtime_metrics
assert!(rendered[0].contains("TBT: 1.2s (iapi) 1.2s (service)"));
}
#[test]
fn runtime_metrics_label_rounds_fractional_tbt_milliseconds() {
let summary = RuntimeMetricsSummary {
responses_api_engine_iapi_tbt_ms: 2.450638,
responses_api_engine_service_tbt_ms: 5.267279,
..RuntimeMetricsSummary::default()
};
insta::assert_snapshot!(runtime_metrics_label(summary).expect("TBT label"), @"TBT: 2ms (iapi) 5ms (service)");
}
#[test]
fn final_message_separator_includes_worked_label_after_one_minute() {
let cell = FinalMessageSeparator::new(Some(61), /*runtime_metrics*/ None);