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