1use std::collections::HashMap;
2
3use serde::{Deserialize, Serialize};
4
5use crate::config::ServiceConfigManager;
6use crate::logging::RetryInfo;
7use crate::pricing::CostBreakdown;
8use crate::runtime_identity::ProviderEndpointKey;
9use crate::sessions;
10use crate::usage::{CacheInputAccounting, UsageMetrics};
11
12fn bool_is_false(value: &bool) -> bool {
13 !*value
14}
15
16const OUTPUT_RATE_SANITY_CEIL: f64 = 5_000.0;
17
18fn service_tier_is_fast(value: Option<&str>) -> bool {
19 value
20 .map(str::trim)
21 .is_some_and(|tier| tier.eq_ignore_ascii_case("priority"))
22}
23
24fn generation_ms_from_duration(duration_ms: u64, ttfb_ms: Option<u64>) -> Option<u64> {
25 if duration_ms == 0 {
26 return None;
27 }
28 let ttfb_ms = ttfb_ms.unwrap_or(0);
29 if ttfb_ms > 0 && ttfb_ms < duration_ms {
30 Some(duration_ms.saturating_sub(ttfb_ms))
31 } else {
32 Some(duration_ms)
33 }
34}
35
36fn effective_ttfb_ms_for_request(
37 duration_ms: u64,
38 ttfb_ms: Option<u64>,
39 retry: Option<&RetryInfo>,
40) -> Option<u64> {
41 let raw_ttfb = ttfb_ms.filter(|value| *value > 0)?;
42 let Some(retry) = retry else {
43 return Some(raw_ttfb);
44 };
45 if duration_ms == 0 {
46 return Some(raw_ttfb);
47 }
48
49 let attempts = retry.route_attempts_or_derived();
50 let Some(final_attempt) = attempts.iter().rev().find(|attempt| {
51 !attempt.skipped
52 && (attempt.decision == "completed"
53 || attempt
54 .status_code
55 .is_some_and(|status| (200..300).contains(&status)))
56 }) else {
57 return Some(raw_ttfb);
58 };
59 let (Some(final_attempt_elapsed_ms), Some(final_attempt_headers_ms)) =
60 (final_attempt.duration_ms, final_attempt.upstream_headers_ms)
61 else {
62 return Some(raw_ttfb);
63 };
64
65 if final_attempt_elapsed_ms <= final_attempt_headers_ms
66 || final_attempt_elapsed_ms >= duration_ms
67 || raw_ttfb >= final_attempt_elapsed_ms
68 {
69 return Some(raw_ttfb);
70 }
71
72 let elapsed_before_final_attempt =
73 final_attempt_elapsed_ms.saturating_sub(final_attempt_headers_ms);
74 let corrected = elapsed_before_final_attempt.saturating_add(raw_ttfb);
75 if corrected > raw_ttfb && corrected < duration_ms {
76 Some(corrected)
77 } else {
78 Some(raw_ttfb)
79 }
80}
81
82fn output_tokens_per_second(
83 output_tokens: i64,
84 duration_ms: u64,
85 generation_ms: u64,
86) -> Option<f64> {
87 if output_tokens <= 0 || duration_ms == 0 || generation_ms == 0 {
88 return None;
89 }
90 let rate = output_tokens as f64 / (generation_ms as f64 / 1000.0);
91 let rate = if generation_ms.saturating_mul(10) < duration_ms && rate > OUTPUT_RATE_SANITY_CEIL {
92 output_tokens as f64 / (duration_ms as f64 / 1000.0)
93 } else {
94 rate
95 };
96 rate.is_finite().then_some(rate).filter(|rate| *rate > 0.0)
97}
98
99fn default_attempt_count() -> u32 {
100 1
101}
102
103#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
104pub struct RequestObservability {
105 #[serde(default, skip_serializing_if = "Option::is_none")]
106 pub trace_id: Option<String>,
107 #[serde(default, skip_serializing_if = "Option::is_none")]
108 pub duration_ms: Option<u64>,
109 #[serde(default, skip_serializing_if = "Option::is_none")]
110 pub ttfb_ms: Option<u64>,
111 #[serde(default, skip_serializing_if = "Option::is_none")]
112 pub generation_ms: Option<u64>,
113 #[serde(default, skip_serializing_if = "Option::is_none")]
114 pub output_tokens_per_second: Option<f64>,
115 #[serde(default = "default_attempt_count")]
116 pub attempt_count: u32,
117 #[serde(default)]
118 pub route_attempt_count: usize,
119 #[serde(default, skip_serializing_if = "bool_is_false")]
120 pub retried: bool,
121 #[serde(default, skip_serializing_if = "bool_is_false")]
122 pub cross_station_failover: bool,
123 #[serde(default, skip_serializing_if = "bool_is_false")]
124 pub same_station_retry: bool,
125 #[serde(default, skip_serializing_if = "bool_is_false")]
126 pub fast_mode: bool,
127 #[serde(default, skip_serializing_if = "bool_is_false")]
128 pub streaming: bool,
129}
130
131impl Default for RequestObservability {
132 fn default() -> Self {
133 Self {
134 trace_id: None,
135 duration_ms: None,
136 ttfb_ms: None,
137 generation_ms: None,
138 output_tokens_per_second: None,
139 attempt_count: 1,
140 route_attempt_count: 0,
141 retried: false,
142 cross_station_failover: false,
143 same_station_retry: false,
144 fast_mode: false,
145 streaming: false,
146 }
147 }
148}
149
150#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
151pub struct ActiveRequest {
152 pub id: u64,
153 #[serde(default, skip_serializing_if = "Option::is_none")]
154 pub trace_id: Option<String>,
155 #[serde(skip_serializing_if = "Option::is_none")]
156 pub session_id: Option<String>,
157 #[serde(default, skip_serializing_if = "Option::is_none")]
158 pub session_identity_source: Option<SessionIdentitySource>,
159 #[serde(skip_serializing_if = "Option::is_none")]
160 pub client_name: Option<String>,
161 #[serde(skip_serializing_if = "Option::is_none")]
162 pub client_addr: Option<String>,
163 #[serde(skip_serializing_if = "Option::is_none")]
164 pub cwd: Option<String>,
165 #[serde(skip_serializing_if = "Option::is_none")]
166 pub model: Option<String>,
167 #[serde(skip_serializing_if = "Option::is_none")]
168 pub reasoning_effort: Option<String>,
169 #[serde(skip_serializing_if = "Option::is_none")]
170 pub service_tier: Option<String>,
171 #[serde(skip_serializing_if = "Option::is_none")]
172 pub station_name: Option<String>,
173 #[serde(skip_serializing_if = "Option::is_none")]
174 pub provider_id: Option<String>,
175 #[serde(skip_serializing_if = "Option::is_none")]
176 pub upstream_base_url: Option<String>,
177 #[serde(skip_serializing_if = "Option::is_none")]
178 pub route_decision: Option<RouteDecisionProvenance>,
179 pub service: String,
180 pub method: String,
181 pub path: String,
182 pub started_at_ms: u64,
183}
184
185#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
186pub struct FinishedRequest {
187 pub id: u64,
188 #[serde(default, skip_serializing_if = "Option::is_none")]
189 pub trace_id: Option<String>,
190 #[serde(skip_serializing_if = "Option::is_none")]
191 pub session_id: Option<String>,
192 #[serde(default, skip_serializing_if = "Option::is_none")]
193 pub session_identity_source: Option<SessionIdentitySource>,
194 #[serde(skip_serializing_if = "Option::is_none")]
195 pub client_name: Option<String>,
196 #[serde(skip_serializing_if = "Option::is_none")]
197 pub client_addr: Option<String>,
198 #[serde(skip_serializing_if = "Option::is_none")]
199 pub cwd: Option<String>,
200 #[serde(skip_serializing_if = "Option::is_none")]
201 pub model: Option<String>,
202 #[serde(skip_serializing_if = "Option::is_none")]
203 pub reasoning_effort: Option<String>,
204 #[serde(skip_serializing_if = "Option::is_none")]
205 pub service_tier: Option<String>,
206 #[serde(skip_serializing_if = "Option::is_none")]
207 pub station_name: Option<String>,
208 #[serde(skip_serializing_if = "Option::is_none")]
209 pub provider_id: Option<String>,
210 #[serde(skip_serializing_if = "Option::is_none")]
211 pub upstream_base_url: Option<String>,
212 #[serde(skip_serializing_if = "Option::is_none")]
213 pub route_decision: Option<RouteDecisionProvenance>,
214 #[serde(skip_serializing_if = "Option::is_none")]
215 pub usage: Option<UsageMetrics>,
216 #[serde(default, skip_serializing_if = "CostBreakdown::is_unknown")]
217 pub cost: CostBreakdown,
218 #[serde(skip_serializing_if = "Option::is_none")]
219 pub retry: Option<crate::logging::RetryInfo>,
220 #[serde(default)]
221 pub observability: RequestObservability,
222 pub service: String,
223 pub method: String,
224 pub path: String,
225 pub status_code: u16,
226 pub duration_ms: u64,
227 #[serde(skip_serializing_if = "Option::is_none")]
228 pub ttfb_ms: Option<u64>,
229 #[serde(default, skip_serializing_if = "bool_is_false")]
230 pub streaming: bool,
231 pub ended_at_ms: u64,
232}
233
234impl FinishedRequest {
235 pub fn cache_input_accounting(&self) -> CacheInputAccounting {
236 CacheInputAccounting::for_service(&self.service)
237 }
238
239 pub fn cache_hit_rate(&self) -> Option<f64> {
240 self.usage
241 .as_ref()
242 .and_then(|usage| usage.cache_hit_rate_with_accounting(self.cache_input_accounting()))
243 }
244
245 pub fn observability_view(&self) -> RequestObservability {
246 RequestObservability::from_finished_request(self)
247 }
248
249 pub fn refresh_observability(&mut self) {
250 self.observability = self.observability_view();
251 }
252
253 pub fn generation_ms(&self) -> Option<u64> {
254 self.observability_view().generation_ms
255 }
256
257 pub fn output_tokens_per_second(&self) -> Option<f64> {
258 self.observability_view().output_tokens_per_second
259 }
260
261 pub fn attempt_count(&self) -> u32 {
262 self.observability_view().attempt_count
263 }
264
265 pub fn is_fast_mode(&self) -> bool {
266 self.observability_view().fast_mode
267 }
268
269 pub fn crossed_station_boundary(&self) -> bool {
270 self.observability_view().cross_station_failover
271 }
272}
273
274impl RequestObservability {
275 pub fn from_finished_request(request: &FinishedRequest) -> Self {
276 let retry = request.retry.as_ref();
277 let attempt_count = retry.map(|retry| retry.attempts.max(1)).unwrap_or(1);
278 let route_attempts = retry
279 .map(|retry| retry.route_attempts_or_derived())
280 .unwrap_or_default();
281 let route_attempt_count = route_attempts.len();
282 let final_station = request
283 .station_name
284 .as_deref()
285 .map(str::trim)
286 .filter(|station| !station.is_empty());
287 let has_station_context = final_station.is_some()
288 && route_attempts
289 .iter()
290 .any(|attempt| attempt.station_name.as_deref().is_some());
291 let cross_station_failover = final_station.is_some_and(|final_station| {
292 route_attempts
293 .iter()
294 .filter_map(|attempt| attempt.station_name.as_deref())
295 .any(|station| station != final_station)
296 });
297 let retried = attempt_count > 1;
298 let same_station_retry = retried && has_station_context && !cross_station_failover;
299 let ttfb_ms = effective_ttfb_ms_for_request(
300 request.duration_ms,
301 request.ttfb_ms,
302 request.retry.as_ref(),
303 );
304 let generation_ms = generation_ms_from_duration(request.duration_ms, ttfb_ms);
305 let output_tokens_per_second = request.usage.as_ref().and_then(|usage| {
306 if usage.output_tokens == 0 {
307 return None;
308 }
309 let generation_ms = generation_ms?;
310 if generation_ms == 0 {
311 return None;
312 }
313 output_tokens_per_second(usage.output_tokens, request.duration_ms, generation_ms)
314 });
315 let decided_fast = request
316 .route_decision
317 .as_ref()
318 .and_then(|decision| decision.effective_service_tier.as_ref())
319 .is_some_and(|value| service_tier_is_fast(Some(value.value.as_str())));
320
321 Self {
322 trace_id: request
323 .trace_id
324 .clone()
325 .or_else(|| request.observability.trace_id.clone()),
326 duration_ms: Some(request.duration_ms),
327 ttfb_ms,
328 generation_ms,
329 output_tokens_per_second,
330 attempt_count,
331 route_attempt_count,
332 retried,
333 cross_station_failover,
334 same_station_retry,
335 fast_mode: decided_fast || service_tier_is_fast(request.service_tier.as_deref()),
336 streaming: request.streaming || request.observability.streaming,
337 }
338 }
339}
340
341#[derive(Debug, Clone)]
342pub struct FinishRequestParams {
343 pub id: u64,
344 pub status_code: u16,
345 pub duration_ms: u64,
346 pub ended_at_ms: u64,
347 pub observed_service_tier: Option<String>,
348 pub usage: Option<UsageMetrics>,
349 pub retry: Option<crate::logging::RetryInfo>,
350 pub ttfb_ms: Option<u64>,
351 pub streaming: bool,
352}
353
354#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
355pub struct SessionStats {
356 pub turns_total: u64,
357 #[serde(default, skip_serializing_if = "Option::is_none")]
358 pub last_session_identity_source: Option<SessionIdentitySource>,
359 #[serde(skip_serializing_if = "Option::is_none")]
360 pub last_client_name: Option<String>,
361 #[serde(skip_serializing_if = "Option::is_none")]
362 pub last_client_addr: Option<String>,
363 #[serde(skip_serializing_if = "Option::is_none")]
364 pub last_model: Option<String>,
365 #[serde(skip_serializing_if = "Option::is_none")]
366 pub last_reasoning_effort: Option<String>,
367 #[serde(skip_serializing_if = "Option::is_none")]
368 pub last_service_tier: Option<String>,
369 #[serde(skip_serializing_if = "Option::is_none")]
370 pub last_provider_id: Option<String>,
371 #[serde(skip_serializing_if = "Option::is_none")]
372 pub last_station_name: Option<String>,
373 #[serde(skip_serializing_if = "Option::is_none")]
374 pub last_route_decision: Option<RouteDecisionProvenance>,
375 #[serde(skip_serializing_if = "Option::is_none")]
376 pub last_usage: Option<UsageMetrics>,
377 pub total_usage: UsageMetrics,
378 pub turns_with_usage: u64,
379 #[serde(skip_serializing_if = "Option::is_none")]
380 pub last_status: Option<u16>,
381 #[serde(skip_serializing_if = "Option::is_none")]
382 pub last_duration_ms: Option<u64>,
383 #[serde(skip_serializing_if = "Option::is_none")]
384 pub last_ended_at_ms: Option<u64>,
385 pub last_seen_ms: u64,
386}
387
388#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
389#[serde(rename_all = "snake_case")]
390pub enum SessionContinuityMode {
391 #[default]
392 DefaultProfile,
393 ManualProfile,
394}
395
396#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
397#[serde(rename_all = "snake_case")]
398pub enum SessionObservationScope {
399 #[default]
400 ObservedOnly,
401 HostLocalEnriched,
402}
403
404#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
405#[serde(rename_all = "snake_case")]
406pub enum SessionIdentitySource {
407 Header,
408 BodySessionId,
409 PromptCacheKey,
410 MetadataSessionId,
411 PreviousResponseId,
412}
413
414#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
415pub struct SessionBinding {
416 pub session_id: String,
417 #[serde(skip_serializing_if = "Option::is_none")]
418 pub profile_name: Option<String>,
419 #[serde(skip_serializing_if = "Option::is_none")]
420 pub station_name: Option<String>,
421 #[serde(skip_serializing_if = "Option::is_none")]
422 pub model: Option<String>,
423 #[serde(skip_serializing_if = "Option::is_none")]
424 pub reasoning_effort: Option<String>,
425 #[serde(skip_serializing_if = "Option::is_none")]
426 pub service_tier: Option<String>,
427 pub continuity_mode: SessionContinuityMode,
428 pub created_at_ms: u64,
429 pub updated_at_ms: u64,
430 pub last_seen_ms: u64,
431}
432
433#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
434#[serde(rename_all = "snake_case")]
435pub enum RouteValueSource {
436 RequestPayload,
437 SessionOverride,
438 GlobalOverride,
439 ProfileDefault,
440 StationMapping,
441 RuntimeFallback,
442}
443
444#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
445pub struct ResolvedRouteValue {
446 pub value: String,
447 pub source: RouteValueSource,
448}
449
450impl ResolvedRouteValue {
451 pub(crate) fn new(value: impl Into<String>, source: RouteValueSource) -> Self {
452 Self {
453 value: value.into(),
454 source,
455 }
456 }
457}
458
459#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
460pub struct RouteDecisionProvenance {
461 pub decided_at_ms: u64,
462 #[serde(skip_serializing_if = "Option::is_none")]
463 pub binding_profile_name: Option<String>,
464 #[serde(skip_serializing_if = "Option::is_none")]
465 pub binding_continuity_mode: Option<SessionContinuityMode>,
466 #[serde(skip_serializing_if = "Option::is_none")]
467 pub effective_model: Option<ResolvedRouteValue>,
468 #[serde(skip_serializing_if = "Option::is_none")]
469 pub effective_reasoning_effort: Option<ResolvedRouteValue>,
470 #[serde(skip_serializing_if = "Option::is_none")]
471 pub effective_service_tier: Option<ResolvedRouteValue>,
472 #[serde(skip_serializing_if = "Option::is_none")]
473 pub effective_station: Option<ResolvedRouteValue>,
474 #[serde(skip_serializing_if = "Option::is_none")]
475 pub effective_upstream_base_url: Option<ResolvedRouteValue>,
476 #[serde(skip_serializing_if = "Option::is_none")]
477 pub provider_id: Option<String>,
478 #[serde(skip_serializing_if = "Option::is_none")]
479 pub endpoint_id: Option<String>,
480 #[serde(default, skip_serializing_if = "Vec::is_empty")]
481 pub route_path: Vec<String>,
482}
483
484#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
485pub struct SessionRouteAffinity {
486 pub route_graph_key: String,
487 #[serde(default, skip_serializing_if = "Option::is_none")]
488 pub session_identity_source: Option<SessionIdentitySource>,
489 pub provider_endpoint: ProviderEndpointKey,
490 pub upstream_base_url: String,
491 #[serde(default, skip_serializing_if = "Vec::is_empty")]
492 pub route_path: Vec<String>,
493 pub last_selected_at_ms: u64,
494 pub last_changed_at_ms: u64,
495 pub change_reason: String,
496}
497
498#[derive(Debug, Clone, PartialEq, Eq)]
499pub struct SessionRouteAffinityTarget {
500 pub route_graph_key: String,
501 pub session_identity_source: Option<SessionIdentitySource>,
502 pub provider_endpoint: ProviderEndpointKey,
503 pub upstream_base_url: String,
504 pub route_path: Vec<String>,
505}
506
507impl SessionRouteAffinityTarget {
508 pub(crate) fn same_target(&self, affinity: &SessionRouteAffinity) -> bool {
509 self.route_graph_key == affinity.route_graph_key
510 && self.provider_endpoint == affinity.provider_endpoint
511 && self.upstream_base_url == affinity.upstream_base_url
512 && self.route_path == affinity.route_path
513 }
514}
515
516#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
517pub struct SessionIdentityCard {
518 pub session_id: Option<String>,
519 #[serde(default, skip_serializing_if = "Option::is_none")]
520 pub session_identity_source: Option<SessionIdentitySource>,
521 #[serde(default)]
522 pub observation_scope: SessionObservationScope,
523 #[serde(skip_serializing_if = "Option::is_none")]
524 pub host_local_transcript_path: Option<String>,
525 #[serde(skip_serializing_if = "Option::is_none")]
526 pub last_client_name: Option<String>,
527 #[serde(skip_serializing_if = "Option::is_none")]
528 pub last_client_addr: Option<String>,
529 #[serde(skip_serializing_if = "Option::is_none")]
530 pub cwd: Option<String>,
531 pub active_count: u64,
532 #[serde(skip_serializing_if = "Option::is_none")]
533 pub active_started_at_ms_min: Option<u64>,
534 #[serde(skip_serializing_if = "Option::is_none")]
535 pub last_status: Option<u16>,
536 #[serde(skip_serializing_if = "Option::is_none")]
537 pub last_duration_ms: Option<u64>,
538 #[serde(skip_serializing_if = "Option::is_none")]
539 pub last_ended_at_ms: Option<u64>,
540 #[serde(skip_serializing_if = "Option::is_none")]
541 pub last_model: Option<String>,
542 #[serde(skip_serializing_if = "Option::is_none")]
543 pub last_reasoning_effort: Option<String>,
544 #[serde(skip_serializing_if = "Option::is_none")]
545 pub last_service_tier: Option<String>,
546 #[serde(skip_serializing_if = "Option::is_none")]
547 pub last_provider_id: Option<String>,
548 #[serde(skip_serializing_if = "Option::is_none")]
549 pub last_station_name: Option<String>,
550 #[serde(skip_serializing_if = "Option::is_none")]
551 pub last_upstream_base_url: Option<String>,
552 #[serde(skip_serializing_if = "Option::is_none")]
553 pub last_usage: Option<UsageMetrics>,
554 #[serde(skip_serializing_if = "Option::is_none")]
555 pub total_usage: Option<UsageMetrics>,
556 #[serde(skip_serializing_if = "Option::is_none")]
557 pub turns_total: Option<u64>,
558 #[serde(skip_serializing_if = "Option::is_none")]
559 pub turns_with_usage: Option<u64>,
560 #[serde(skip_serializing_if = "Option::is_none")]
561 pub binding_profile_name: Option<String>,
562 #[serde(skip_serializing_if = "Option::is_none")]
563 pub binding_continuity_mode: Option<SessionContinuityMode>,
564 #[serde(skip_serializing_if = "Option::is_none")]
565 pub last_route_decision: Option<RouteDecisionProvenance>,
566 #[serde(skip_serializing_if = "Option::is_none")]
567 pub route_affinity: Option<SessionRouteAffinity>,
568 #[serde(skip_serializing_if = "Option::is_none")]
569 pub effective_model: Option<ResolvedRouteValue>,
570 #[serde(skip_serializing_if = "Option::is_none")]
571 pub effective_reasoning_effort: Option<ResolvedRouteValue>,
572 #[serde(skip_serializing_if = "Option::is_none")]
573 pub effective_service_tier: Option<ResolvedRouteValue>,
574 #[serde(skip_serializing_if = "Option::is_none")]
575 pub effective_station: Option<ResolvedRouteValue>,
576 #[serde(skip_serializing_if = "Option::is_none")]
577 pub effective_upstream_base_url: Option<ResolvedRouteValue>,
578 #[serde(skip_serializing_if = "Option::is_none")]
579 pub override_effort: Option<String>,
580 #[serde(skip_serializing_if = "Option::is_none")]
581 pub override_station_name: Option<String>,
582 #[serde(skip_serializing_if = "Option::is_none")]
583 pub override_model: Option<String>,
584 #[serde(skip_serializing_if = "Option::is_none")]
585 pub override_service_tier: Option<String>,
586}
587
588#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
589pub struct SessionManualOverrides {
590 #[serde(skip_serializing_if = "Option::is_none")]
591 pub reasoning_effort: Option<String>,
592 #[serde(skip_serializing_if = "Option::is_none")]
593 pub station_name: Option<String>,
594 #[serde(skip_serializing_if = "Option::is_none")]
595 pub route_target: Option<String>,
596 #[serde(skip_serializing_if = "Option::is_none")]
597 pub model: Option<String>,
598 #[serde(skip_serializing_if = "Option::is_none")]
599 pub service_tier: Option<String>,
600}
601
602impl SessionManualOverrides {
603 pub fn is_empty(&self) -> bool {
604 self.reasoning_effort.is_none()
605 && self.station_name.is_none()
606 && self.route_target.is_none()
607 && self.model.is_none()
608 && self.service_tier.is_none()
609 }
610}
611
612#[derive(Debug, Clone)]
613pub(super) struct SessionEffortOverride {
614 pub(super) effort: String,
615 #[allow(dead_code)]
616 pub(super) updated_at_ms: u64,
617 pub(super) last_seen_ms: u64,
618}
619
620#[derive(Debug, Clone)]
621pub(super) struct SessionStationOverride {
622 pub(super) station_name: String,
623 #[allow(dead_code)]
624 pub(super) updated_at_ms: u64,
625 pub(super) last_seen_ms: u64,
626}
627
628#[derive(Debug, Clone)]
629pub(super) struct SessionRouteTargetOverride {
630 pub(super) target: String,
631 #[allow(dead_code)]
632 pub(super) updated_at_ms: u64,
633 pub(super) last_seen_ms: u64,
634}
635
636#[derive(Debug, Clone)]
637pub(super) struct SessionModelOverride {
638 pub(super) model: String,
639 #[allow(dead_code)]
640 pub(super) updated_at_ms: u64,
641 pub(super) last_seen_ms: u64,
642}
643
644#[derive(Debug, Clone)]
645pub(super) struct SessionServiceTierOverride {
646 pub(super) service_tier: String,
647 #[allow(dead_code)]
648 pub(super) updated_at_ms: u64,
649 pub(super) last_seen_ms: u64,
650}
651
652#[derive(Debug, Clone)]
653pub(super) struct SessionBindingEntry {
654 pub(super) binding: SessionBinding,
655}
656
657#[derive(Debug, Clone)]
658pub(super) struct SessionCwdCacheEntry {
659 pub(super) cwd: Option<String>,
660 pub(super) last_seen_ms: u64,
661}
662
663fn empty_session_identity_card(session_id: Option<String>) -> SessionIdentityCard {
664 SessionIdentityCard {
665 session_id,
666 session_identity_source: None,
667 observation_scope: SessionObservationScope::ObservedOnly,
668 host_local_transcript_path: None,
669 last_client_name: None,
670 last_client_addr: None,
671 cwd: None,
672 active_count: 0,
673 active_started_at_ms_min: None,
674 last_status: None,
675 last_duration_ms: None,
676 last_ended_at_ms: None,
677 last_model: None,
678 last_reasoning_effort: None,
679 last_service_tier: None,
680 last_provider_id: None,
681 last_station_name: None,
682 last_upstream_base_url: None,
683 last_usage: None,
684 total_usage: None,
685 turns_total: None,
686 turns_with_usage: None,
687 binding_profile_name: None,
688 binding_continuity_mode: None,
689 last_route_decision: None,
690 route_affinity: None,
691 effective_model: None,
692 effective_reasoning_effort: None,
693 effective_service_tier: None,
694 effective_station: None,
695 effective_upstream_base_url: None,
696 override_effort: None,
697 override_station_name: None,
698 override_model: None,
699 override_service_tier: None,
700 }
701}
702
703fn non_empty_trimmed(value: Option<&str>) -> Option<String> {
704 value
705 .map(str::trim)
706 .filter(|value| !value.is_empty())
707 .map(ToOwned::to_owned)
708}
709
710fn resolve_effective_observed_value(
711 override_value: Option<&str>,
712 observed_value: Option<&str>,
713 binding_value: Option<&str>,
714) -> Option<ResolvedRouteValue> {
715 if let Some(value) = non_empty_trimmed(override_value) {
716 return Some(ResolvedRouteValue::new(
717 value,
718 RouteValueSource::SessionOverride,
719 ));
720 }
721 let binding_value = non_empty_trimmed(binding_value);
722 if let Some(binding) = binding_value {
723 return Some(ResolvedRouteValue::new(
724 binding,
725 RouteValueSource::ProfileDefault,
726 ));
727 }
728 non_empty_trimmed(observed_value)
729 .map(|observed| ResolvedRouteValue::new(observed, RouteValueSource::RequestPayload))
730}
731
732fn binding_request_field_value<'a>(
733 binding: Option<&'a SessionBinding>,
734 value: impl FnOnce(&'a SessionBinding) -> Option<&'a str>,
735) -> Option<&'a str> {
736 let binding = binding?;
737 if binding.continuity_mode != SessionContinuityMode::ManualProfile {
738 return None;
739 }
740 value(binding)
741}
742
743fn classify_session_observation_scope(card: &SessionIdentityCard) -> SessionObservationScope {
744 if card.cwd.is_some() {
745 SessionObservationScope::HostLocalEnriched
746 } else {
747 SessionObservationScope::ObservedOnly
748 }
749}
750
751fn resolve_effective_station_value(
752 card: &SessionIdentityCard,
753 global_station_override: Option<&str>,
754 binding_station_name: Option<&str>,
755) -> Option<ResolvedRouteValue> {
756 if let Some(value) = non_empty_trimmed(card.override_station_name.as_deref()) {
757 return Some(ResolvedRouteValue::new(
758 value,
759 RouteValueSource::SessionOverride,
760 ));
761 }
762 if let Some(value) = non_empty_trimmed(global_station_override) {
763 return Some(ResolvedRouteValue::new(
764 value,
765 RouteValueSource::GlobalOverride,
766 ));
767 }
768 let binding = non_empty_trimmed(binding_station_name);
769 if let Some(binding) = binding {
770 return Some(ResolvedRouteValue::new(
771 binding,
772 RouteValueSource::ProfileDefault,
773 ));
774 }
775 non_empty_trimmed(card.last_station_name.as_deref())
776 .map(|observed| ResolvedRouteValue::new(observed, RouteValueSource::RuntimeFallback))
777}
778
779fn apply_basic_effective_route(
780 card: &mut SessionIdentityCard,
781 global_station_override: Option<&str>,
782 binding: Option<&SessionBinding>,
783) {
784 card.effective_model = resolve_effective_observed_value(
785 card.override_model.as_deref(),
786 card.last_model.as_deref(),
787 binding_request_field_value(binding, |binding| binding.model.as_deref()),
788 );
789 card.effective_reasoning_effort = resolve_effective_observed_value(
790 card.override_effort.as_deref(),
791 card.last_reasoning_effort.as_deref(),
792 binding_request_field_value(binding, |binding| binding.reasoning_effort.as_deref()),
793 );
794 card.effective_service_tier = resolve_effective_observed_value(
795 card.override_service_tier.as_deref(),
796 card.last_service_tier.as_deref(),
797 binding_request_field_value(binding, |binding| binding.service_tier.as_deref()),
798 );
799 card.binding_profile_name = binding.and_then(|binding| binding.profile_name.clone());
800 card.binding_continuity_mode = binding.map(|binding| binding.continuity_mode);
801 card.effective_station = resolve_effective_station_value(
802 card,
803 global_station_override,
804 binding.and_then(|binding| binding.station_name.as_deref()),
805 );
806 card.effective_upstream_base_url = match (
807 card.effective_station.as_ref(),
808 non_empty_trimmed(card.last_station_name.as_deref()),
809 non_empty_trimmed(card.last_upstream_base_url.as_deref()),
810 ) {
811 (Some(station), Some(last_station), Some(upstream)) if station.value == last_station => {
812 Some(ResolvedRouteValue::new(
813 upstream,
814 RouteValueSource::RuntimeFallback,
815 ))
816 }
817 _ => None,
818 };
819}
820
821pub fn enrich_session_identity_cards_with_runtime(
822 cards: &mut [SessionIdentityCard],
823 mgr: &ServiceConfigManager,
824) {
825 for card in cards {
826 if card.effective_station.is_none()
827 && let Some(active) = mgr.active_station()
828 {
829 card.effective_station = Some(ResolvedRouteValue::new(
830 active.name.clone(),
831 RouteValueSource::RuntimeFallback,
832 ));
833 }
834
835 let effective_station_name = card
836 .effective_station
837 .as_ref()
838 .map(|value| value.value.as_str());
839 if card.effective_upstream_base_url.is_none()
840 && let Some(station_name) = effective_station_name
841 && let Some(station) = mgr.station(station_name)
842 && station.upstreams.len() == 1
843 {
844 card.effective_upstream_base_url = Some(ResolvedRouteValue::new(
845 station.upstreams[0].base_url.clone(),
846 RouteValueSource::RuntimeFallback,
847 ));
848 }
849
850 let Some(model) = card
851 .effective_model
852 .as_ref()
853 .map(|value| value.value.clone())
854 else {
855 continue;
856 };
857 let Some(station_name) = effective_station_name else {
858 continue;
859 };
860 let Some(last_station_name) = card.last_station_name.as_deref() else {
861 continue;
862 };
863 if last_station_name != station_name {
864 continue;
865 }
866 let Some(last_upstream_base_url) = card.last_upstream_base_url.as_deref() else {
867 continue;
868 };
869 let Some(station) = mgr.station(station_name) else {
870 continue;
871 };
872 let Some(upstream) = station
873 .upstreams
874 .iter()
875 .find(|upstream| upstream.base_url == last_upstream_base_url)
876 else {
877 continue;
878 };
879
880 let mapped = crate::model_routing::effective_model(&upstream.model_mapping, model.as_str());
881 if mapped != model {
882 card.effective_model = Some(ResolvedRouteValue::new(
883 mapped,
884 RouteValueSource::StationMapping,
885 ));
886 }
887 }
888}
889
890fn session_identity_sort_key(card: &SessionIdentityCard) -> u64 {
891 card.last_ended_at_ms
892 .unwrap_or(0)
893 .max(card.active_started_at_ms_min.unwrap_or(0))
894}
895
896fn route_decision_at_ms(route_decision: Option<&RouteDecisionProvenance>) -> u64 {
897 route_decision
898 .map(|decision| decision.decided_at_ms)
899 .unwrap_or(0)
900}
901
902fn update_card_route_decision(
903 card: &mut SessionIdentityCard,
904 route_decision: Option<&RouteDecisionProvenance>,
905) {
906 let Some(route_decision) = route_decision.cloned() else {
907 return;
908 };
909 let current_at = route_decision_at_ms(card.last_route_decision.as_ref());
910 if current_at <= route_decision.decided_at_ms {
911 card.last_route_decision = Some(route_decision);
912 }
913}
914
915fn merge_session_identity_source(
916 target: &mut Option<SessionIdentitySource>,
917 source: Option<SessionIdentitySource>,
918) {
919 if source.is_some() {
920 *target = source;
921 }
922}
923
924pub struct SessionIdentityCardBuildInputs<'a> {
925 pub active: &'a [ActiveRequest],
926 pub recent: &'a [FinishedRequest],
927 pub overrides: &'a HashMap<String, String>,
928 pub station_overrides: &'a HashMap<String, String>,
929 pub model_overrides: &'a HashMap<String, String>,
930 pub service_tier_overrides: &'a HashMap<String, String>,
931 pub bindings: &'a HashMap<String, SessionBinding>,
932 pub route_affinities: &'a HashMap<String, SessionRouteAffinity>,
933 pub global_station_override: Option<&'a str>,
934 pub stats: &'a HashMap<String, SessionStats>,
935}
936
937pub fn build_session_identity_cards_from_parts(
938 inputs: SessionIdentityCardBuildInputs<'_>,
939) -> Vec<SessionIdentityCard> {
940 let SessionIdentityCardBuildInputs {
941 active,
942 recent,
943 overrides,
944 station_overrides,
945 model_overrides,
946 service_tier_overrides,
947 bindings,
948 route_affinities,
949 global_station_override,
950 stats,
951 } = inputs;
952
953 use std::collections::HashMap as StdHashMap;
954
955 let mut map: StdHashMap<Option<String>, SessionIdentityCard> = StdHashMap::new();
956
957 for req in active {
958 let key = req.session_id.clone();
959 let entry = map
960 .entry(key.clone())
961 .or_insert_with(|| empty_session_identity_card(key));
962 merge_session_identity_source(
963 &mut entry.session_identity_source,
964 req.session_identity_source,
965 );
966
967 entry.active_count = entry.active_count.saturating_add(1);
968 entry.active_started_at_ms_min = Some(
969 entry
970 .active_started_at_ms_min
971 .unwrap_or(req.started_at_ms)
972 .min(req.started_at_ms),
973 );
974 if entry.cwd.is_none() {
975 entry.cwd = req.cwd.clone();
976 }
977 if entry.last_client_name.is_none() {
978 entry.last_client_name = req.client_name.clone();
979 }
980 if entry.last_client_addr.is_none() {
981 entry.last_client_addr = req.client_addr.clone();
982 }
983 if let Some(effort) = req.reasoning_effort.as_ref() {
984 entry.last_reasoning_effort = Some(effort.clone());
985 }
986 if let Some(service_tier) = req.service_tier.as_ref() {
987 entry.last_service_tier = Some(service_tier.clone());
988 }
989 if entry.last_model.is_none() {
990 entry.last_model = req.model.clone();
991 }
992 if entry.last_provider_id.is_none() {
993 entry.last_provider_id = req.provider_id.clone();
994 }
995 if entry.last_station_name.is_none() {
996 entry.last_station_name = req.station_name.clone();
997 }
998 if entry.last_upstream_base_url.is_none() {
999 entry.last_upstream_base_url = req.upstream_base_url.clone();
1000 }
1001 update_card_route_decision(entry, req.route_decision.as_ref());
1002 }
1003
1004 for r in recent {
1005 let key = r.session_id.clone();
1006 let entry = map
1007 .entry(key.clone())
1008 .or_insert_with(|| empty_session_identity_card(key));
1009 merge_session_identity_source(
1010 &mut entry.session_identity_source,
1011 r.session_identity_source,
1012 );
1013
1014 let should_update = entry
1015 .last_ended_at_ms
1016 .is_none_or(|prev| r.ended_at_ms >= prev);
1017 if should_update {
1018 entry.last_status = Some(r.status_code);
1019 entry.last_duration_ms = Some(r.duration_ms);
1020 entry.last_ended_at_ms = Some(r.ended_at_ms);
1021 entry.last_client_name = r.client_name.clone().or(entry.last_client_name.clone());
1022 entry.last_client_addr = r.client_addr.clone().or(entry.last_client_addr.clone());
1023 entry.last_model = r.model.clone().or(entry.last_model.clone());
1024 entry.last_reasoning_effort = r
1025 .reasoning_effort
1026 .clone()
1027 .or(entry.last_reasoning_effort.clone());
1028 entry.last_service_tier = r.service_tier.clone().or(entry.last_service_tier.clone());
1029 entry.last_provider_id = r.provider_id.clone().or(entry.last_provider_id.clone());
1030 entry.last_station_name = r.station_name.clone().or(entry.last_station_name.clone());
1031 entry.last_upstream_base_url = r
1032 .upstream_base_url
1033 .clone()
1034 .or(entry.last_upstream_base_url.clone());
1035 entry.last_usage = r.usage.clone().or(entry.last_usage.clone());
1036 }
1037 if entry.cwd.is_none() {
1038 entry.cwd = r.cwd.clone();
1039 }
1040 update_card_route_decision(entry, r.route_decision.as_ref());
1041 }
1042
1043 for (sid, st) in stats {
1044 let key = Some(sid.clone());
1045 let entry = map
1046 .entry(key.clone())
1047 .or_insert_with(|| empty_session_identity_card(key));
1048 merge_session_identity_source(
1049 &mut entry.session_identity_source,
1050 st.last_session_identity_source,
1051 );
1052
1053 if entry.turns_total.is_none() {
1054 entry.turns_total = Some(st.turns_total);
1055 }
1056 if entry.last_client_name.is_none() {
1057 entry.last_client_name = st.last_client_name.clone();
1058 }
1059 if entry.last_client_addr.is_none() {
1060 entry.last_client_addr = st.last_client_addr.clone();
1061 }
1062 if entry.last_status.is_none() {
1063 entry.last_status = st.last_status;
1064 }
1065 if entry.last_duration_ms.is_none() {
1066 entry.last_duration_ms = st.last_duration_ms;
1067 }
1068 if entry.last_ended_at_ms.is_none() {
1069 entry.last_ended_at_ms = st.last_ended_at_ms;
1070 }
1071 if entry.last_model.is_none() {
1072 entry.last_model = st.last_model.clone();
1073 }
1074 if entry.last_reasoning_effort.is_none() {
1075 entry.last_reasoning_effort = st.last_reasoning_effort.clone();
1076 }
1077 if entry.last_service_tier.is_none() {
1078 entry.last_service_tier = st.last_service_tier.clone();
1079 }
1080 if entry.last_provider_id.is_none() {
1081 entry.last_provider_id = st.last_provider_id.clone();
1082 }
1083 if entry.last_station_name.is_none() {
1084 entry.last_station_name = st.last_station_name.clone();
1085 }
1086 if entry.last_usage.is_none() {
1087 entry.last_usage = st.last_usage.clone();
1088 }
1089 if entry.total_usage.is_none() {
1090 entry.total_usage = Some(st.total_usage.clone());
1091 }
1092 if entry.turns_with_usage.is_none() {
1093 entry.turns_with_usage = Some(st.turns_with_usage);
1094 }
1095 update_card_route_decision(entry, st.last_route_decision.as_ref());
1096 }
1097
1098 for (sid, eff) in overrides {
1099 let key = Some(sid.clone());
1100 let entry = map
1101 .entry(key.clone())
1102 .or_insert_with(|| empty_session_identity_card(key));
1103 entry.override_effort = Some(eff.clone());
1104 }
1105
1106 for (sid, station_name) in station_overrides {
1107 let key = Some(sid.clone());
1108 let entry = map
1109 .entry(key.clone())
1110 .or_insert_with(|| empty_session_identity_card(key));
1111 entry.override_station_name = Some(station_name.clone());
1112 }
1113
1114 for (sid, model) in model_overrides {
1115 let key = Some(sid.clone());
1116 let entry = map
1117 .entry(key.clone())
1118 .or_insert_with(|| empty_session_identity_card(key));
1119 entry.override_model = Some(model.clone());
1120 }
1121
1122 for (sid, service_tier) in service_tier_overrides {
1123 let key = Some(sid.clone());
1124 let entry = map
1125 .entry(key.clone())
1126 .or_insert_with(|| empty_session_identity_card(key));
1127 entry.override_service_tier = Some(service_tier.clone());
1128 }
1129
1130 for (sid, affinity) in route_affinities {
1131 let key = Some(sid.clone());
1132 if let Some(entry) = map.get_mut(&key) {
1133 merge_session_identity_source(
1134 &mut entry.session_identity_source,
1135 affinity.session_identity_source,
1136 );
1137 entry.route_affinity = Some(affinity.clone());
1138 }
1139 }
1140
1141 let mut cards = map.into_values().collect::<Vec<_>>();
1142 for card in &mut cards {
1143 let binding = card
1144 .session_id
1145 .as_deref()
1146 .and_then(|session_id| bindings.get(session_id));
1147 apply_basic_effective_route(card, global_station_override, binding);
1148 card.observation_scope = classify_session_observation_scope(card);
1149 }
1150 cards.sort_by_key(|card| std::cmp::Reverse(session_identity_sort_key(card)));
1151 cards
1152}
1153
1154pub async fn enrich_session_identity_cards_with_host_transcripts(
1155 cards: &mut [SessionIdentityCard],
1156) {
1157 let session_ids = cards
1158 .iter()
1159 .filter_map(|card| card.session_id.as_deref())
1160 .map(str::to_owned)
1161 .collect::<Vec<_>>();
1162 if session_ids.is_empty() {
1163 return;
1164 }
1165
1166 let found = match sessions::find_codex_session_files_by_ids(&session_ids).await {
1167 Ok(found) => found,
1168 Err(_) => return,
1169 };
1170
1171 for card in cards {
1172 card.host_local_transcript_path = card
1173 .session_id
1174 .as_deref()
1175 .and_then(|sid| found.get(sid))
1176 .map(|path| path.to_string_lossy().to_string());
1177 }
1178}
1179
1180#[cfg(test)]
1181mod tests {
1182 use super::*;
1183
1184 use crate::logging::{RetryInfo, RouteAttemptLog};
1185 use crate::pricing::CostBreakdown;
1186 use crate::usage::UsageMetrics;
1187
1188 #[test]
1189 fn session_identity_cards_do_not_surface_affinity_only_sessions() {
1190 let active = Vec::<ActiveRequest>::new();
1191 let recent = Vec::<FinishedRequest>::new();
1192 let overrides = HashMap::<String, String>::new();
1193 let station_overrides = HashMap::<String, String>::new();
1194 let model_overrides = HashMap::<String, String>::new();
1195 let service_tier_overrides = HashMap::<String, String>::new();
1196 let bindings = HashMap::<String, SessionBinding>::new();
1197 let stats = HashMap::<String, SessionStats>::new();
1198 let mut route_affinities = HashMap::<String, SessionRouteAffinity>::new();
1199 route_affinities.insert(
1200 "sid-old".to_string(),
1201 SessionRouteAffinity {
1202 route_graph_key: "codex/default".to_string(),
1203 session_identity_source: Some(SessionIdentitySource::Header),
1204 provider_endpoint: ProviderEndpointKey::new("codex", "old", "default"),
1205 upstream_base_url: "https://old.example/v1".to_string(),
1206 route_path: vec!["entry".to_string()],
1207 last_selected_at_ms: 10,
1208 last_changed_at_ms: 10,
1209 change_reason: "restored".to_string(),
1210 },
1211 );
1212
1213 let cards = build_session_identity_cards_from_parts(SessionIdentityCardBuildInputs {
1214 active: &active,
1215 recent: &recent,
1216 overrides: &overrides,
1217 station_overrides: &station_overrides,
1218 model_overrides: &model_overrides,
1219 service_tier_overrides: &service_tier_overrides,
1220 bindings: &bindings,
1221 route_affinities: &route_affinities,
1222 global_station_override: None,
1223 stats: &stats,
1224 });
1225
1226 assert!(
1227 cards.is_empty(),
1228 "restored route affinity alone should not appear as an observed TUI session"
1229 );
1230 }
1231
1232 fn sample_finished_request() -> FinishedRequest {
1233 FinishedRequest {
1234 id: 7,
1235 trace_id: Some("codex-7".to_string()),
1236 session_id: Some("sid".to_string()),
1237 session_identity_source: Some(SessionIdentitySource::Header),
1238 client_name: None,
1239 client_addr: None,
1240 cwd: None,
1241 model: Some("gpt-5".to_string()),
1242 reasoning_effort: None,
1243 service_tier: Some("default".to_string()),
1244 station_name: Some("primary".to_string()),
1245 provider_id: Some("primary-provider".to_string()),
1246 upstream_base_url: Some("https://primary.example/v1".to_string()),
1247 route_decision: Some(RouteDecisionProvenance {
1248 effective_service_tier: Some(ResolvedRouteValue {
1249 value: "priority".to_string(),
1250 source: RouteValueSource::ProfileDefault,
1251 }),
1252 ..Default::default()
1253 }),
1254 usage: Some(UsageMetrics {
1255 output_tokens: 200,
1256 total_tokens: 200,
1257 ..UsageMetrics::default()
1258 }),
1259 cost: CostBreakdown::default(),
1260 retry: Some(RetryInfo {
1261 attempts: 2,
1262 upstream_chain: vec![
1263 "backup:https://backup.example/v1 (idx=0) transport_error=timeout".to_string(),
1264 "primary:https://primary.example/v1 (idx=0) status=200 class=-".to_string(),
1265 ],
1266 route_attempts: Vec::new(),
1267 }),
1268 observability: RequestObservability::default(),
1269 service: "codex".to_string(),
1270 method: "POST".to_string(),
1271 path: "/v1/responses".to_string(),
1272 status_code: 200,
1273 duration_ms: 1_500,
1274 ttfb_ms: Some(500),
1275 streaming: true,
1276 ended_at_ms: 2_000,
1277 }
1278 }
1279
1280 #[test]
1281 fn finished_request_observability_derives_canonical_request_facts() {
1282 let request = sample_finished_request();
1283
1284 let observability = request.observability_view();
1285
1286 assert_eq!(observability.trace_id.as_deref(), Some("codex-7"));
1287 assert_eq!(observability.duration_ms, Some(1_500));
1288 assert_eq!(observability.ttfb_ms, Some(500));
1289 assert_eq!(observability.generation_ms, Some(1_000));
1290 assert_eq!(observability.attempt_count, 2);
1291 assert_eq!(observability.route_attempt_count, 2);
1292 assert!(observability.retried);
1293 assert!(observability.cross_station_failover);
1294 assert!(observability.fast_mode);
1295 assert!(observability.streaming);
1296 assert_eq!(observability.output_tokens_per_second, Some(200.0));
1297 }
1298
1299 #[test]
1300 fn finished_request_serializes_materialized_observability_for_operator_api() {
1301 let mut request = sample_finished_request();
1302 request.refresh_observability();
1303
1304 let value = serde_json::to_value(&request).expect("finished request json");
1305
1306 assert_eq!(value["observability"]["trace_id"].as_str(), Some("codex-7"));
1307 assert_eq!(
1308 value["observability"]["generation_ms"].as_u64(),
1309 Some(1_000)
1310 );
1311 assert_eq!(
1312 value["observability"]["output_tokens_per_second"].as_f64(),
1313 Some(200.0)
1314 );
1315 assert_eq!(value["observability"]["attempt_count"].as_u64(), Some(2));
1316 assert_eq!(value["observability"]["streaming"].as_bool(), Some(true));
1317 }
1318
1319 #[test]
1320 fn finished_request_legacy_payload_still_derives_observability() {
1321 let request: FinishedRequest = serde_json::from_value(serde_json::json!({
1322 "id": 8,
1323 "trace_id": "codex-8",
1324 "usage": {
1325 "output_tokens": 120,
1326 "total_tokens": 120
1327 },
1328 "cost": {},
1329 "service": "codex",
1330 "method": "POST",
1331 "path": "/v1/responses",
1332 "status_code": 200,
1333 "duration_ms": 1_500,
1334 "ttfb_ms": 300,
1335 "ended_at_ms": 2_500
1336 }))
1337 .expect("legacy finished request");
1338
1339 assert_eq!(request.observability.attempt_count, 1);
1340 assert!(!request.streaming);
1341
1342 let observability = request.observability_view();
1343 assert_eq!(observability.generation_ms, Some(1_200));
1344 assert_eq!(observability.output_tokens_per_second, Some(100.0));
1345 assert_eq!(observability.attempt_count, 1);
1346 assert!(!observability.fast_mode);
1347 }
1348
1349 #[test]
1350 fn finished_request_corrects_legacy_attempt_relative_stream_ttfb() {
1351 let mut request = sample_finished_request();
1352 request.duration_ms = 10_000;
1353 request.ttfb_ms = Some(1_210);
1354 request.usage = Some(UsageMetrics {
1355 output_tokens: 100,
1356 total_tokens: 100,
1357 ..UsageMetrics::default()
1358 });
1359 request.retry = Some(RetryInfo {
1360 attempts: 2,
1361 upstream_chain: Vec::new(),
1362 route_attempts: vec![
1363 RouteAttemptLog {
1364 attempt_index: 0,
1365 decision: "failed_status".to_string(),
1366 status_code: Some(429),
1367 upstream_headers_ms: Some(50),
1368 duration_ms: Some(500),
1369 raw: "failed".to_string(),
1370 ..RouteAttemptLog::default()
1371 },
1372 RouteAttemptLog {
1373 attempt_index: 1,
1374 decision: "completed".to_string(),
1375 status_code: Some(200),
1376 upstream_headers_ms: Some(1_200),
1377 duration_ms: Some(2_200),
1378 raw: "completed".to_string(),
1379 ..RouteAttemptLog::default()
1380 },
1381 ],
1382 });
1383
1384 let observability = request.observability_view();
1385
1386 assert_eq!(observability.ttfb_ms, Some(2_210));
1387 assert_eq!(observability.generation_ms, Some(7_790));
1388 let rate = observability.output_tokens_per_second.expect("output rate");
1389 assert!((rate - (100.0 / 7.79)).abs() < f64::EPSILON);
1390 }
1391
1392 #[test]
1393 fn finished_request_does_not_double_correct_global_stream_ttfb() {
1394 let mut request = sample_finished_request();
1395 request.duration_ms = 10_000;
1396 request.ttfb_ms = Some(2_210);
1397 request.retry = Some(RetryInfo {
1398 attempts: 2,
1399 upstream_chain: Vec::new(),
1400 route_attempts: vec![RouteAttemptLog {
1401 attempt_index: 1,
1402 decision: "completed".to_string(),
1403 status_code: Some(200),
1404 upstream_headers_ms: Some(1_200),
1405 duration_ms: Some(2_200),
1406 raw: "completed".to_string(),
1407 ..RouteAttemptLog::default()
1408 }],
1409 });
1410
1411 let observability = request.observability_view();
1412
1413 assert_eq!(observability.ttfb_ms, Some(2_210));
1414 assert_eq!(observability.generation_ms, Some(7_790));
1415 }
1416}