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