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 BodySessionId,
339 PromptCacheKey,
340 MetadataSessionId,
341 PreviousResponseId,
342}
343
344#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
345pub struct SessionBinding {
346 pub session_id: String,
347 #[serde(skip_serializing_if = "Option::is_none")]
348 pub profile_name: Option<String>,
349 #[serde(skip_serializing_if = "Option::is_none")]
350 pub station_name: Option<String>,
351 #[serde(skip_serializing_if = "Option::is_none")]
352 pub model: Option<String>,
353 #[serde(skip_serializing_if = "Option::is_none")]
354 pub reasoning_effort: Option<String>,
355 #[serde(skip_serializing_if = "Option::is_none")]
356 pub service_tier: Option<String>,
357 pub continuity_mode: SessionContinuityMode,
358 pub created_at_ms: u64,
359 pub updated_at_ms: u64,
360 pub last_seen_ms: u64,
361}
362
363#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
364#[serde(rename_all = "snake_case")]
365pub enum RouteValueSource {
366 RequestPayload,
367 SessionOverride,
368 GlobalOverride,
369 ProfileDefault,
370 StationMapping,
371 RuntimeFallback,
372}
373
374#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
375pub struct ResolvedRouteValue {
376 pub value: String,
377 pub source: RouteValueSource,
378}
379
380impl ResolvedRouteValue {
381 pub(crate) fn new(value: impl Into<String>, source: RouteValueSource) -> Self {
382 Self {
383 value: value.into(),
384 source,
385 }
386 }
387}
388
389#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
390pub struct RouteDecisionProvenance {
391 pub decided_at_ms: u64,
392 #[serde(skip_serializing_if = "Option::is_none")]
393 pub binding_profile_name: Option<String>,
394 #[serde(skip_serializing_if = "Option::is_none")]
395 pub binding_continuity_mode: Option<SessionContinuityMode>,
396 #[serde(skip_serializing_if = "Option::is_none")]
397 pub effective_model: Option<ResolvedRouteValue>,
398 #[serde(skip_serializing_if = "Option::is_none")]
399 pub effective_reasoning_effort: Option<ResolvedRouteValue>,
400 #[serde(skip_serializing_if = "Option::is_none")]
401 pub effective_service_tier: Option<ResolvedRouteValue>,
402 #[serde(skip_serializing_if = "Option::is_none")]
403 pub effective_station: Option<ResolvedRouteValue>,
404 #[serde(skip_serializing_if = "Option::is_none")]
405 pub effective_upstream_base_url: Option<ResolvedRouteValue>,
406 #[serde(skip_serializing_if = "Option::is_none")]
407 pub provider_id: Option<String>,
408 #[serde(skip_serializing_if = "Option::is_none")]
409 pub endpoint_id: Option<String>,
410 #[serde(default, skip_serializing_if = "Vec::is_empty")]
411 pub route_path: Vec<String>,
412}
413
414#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
415pub struct SessionRouteAffinity {
416 pub route_graph_key: String,
417 #[serde(default, skip_serializing_if = "Option::is_none")]
418 pub session_identity_source: Option<SessionIdentitySource>,
419 pub provider_endpoint: ProviderEndpointKey,
420 pub upstream_base_url: String,
421 #[serde(default, skip_serializing_if = "Vec::is_empty")]
422 pub route_path: Vec<String>,
423 pub last_selected_at_ms: u64,
424 pub last_changed_at_ms: u64,
425 pub change_reason: String,
426}
427
428#[derive(Debug, Clone, PartialEq, Eq)]
429pub struct SessionRouteAffinityTarget {
430 pub route_graph_key: String,
431 pub session_identity_source: Option<SessionIdentitySource>,
432 pub provider_endpoint: ProviderEndpointKey,
433 pub upstream_base_url: String,
434 pub route_path: Vec<String>,
435}
436
437impl SessionRouteAffinityTarget {
438 pub(crate) fn same_target(&self, affinity: &SessionRouteAffinity) -> bool {
439 self.route_graph_key == affinity.route_graph_key
440 && self.provider_endpoint == affinity.provider_endpoint
441 && self.upstream_base_url == affinity.upstream_base_url
442 && self.route_path == affinity.route_path
443 }
444}
445
446#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
447pub struct SessionIdentityCard {
448 pub session_id: Option<String>,
449 #[serde(default, skip_serializing_if = "Option::is_none")]
450 pub session_identity_source: Option<SessionIdentitySource>,
451 #[serde(default)]
452 pub observation_scope: SessionObservationScope,
453 #[serde(skip_serializing_if = "Option::is_none")]
454 pub host_local_transcript_path: Option<String>,
455 #[serde(skip_serializing_if = "Option::is_none")]
456 pub last_client_name: Option<String>,
457 #[serde(skip_serializing_if = "Option::is_none")]
458 pub last_client_addr: Option<String>,
459 #[serde(skip_serializing_if = "Option::is_none")]
460 pub cwd: Option<String>,
461 pub active_count: u64,
462 #[serde(skip_serializing_if = "Option::is_none")]
463 pub active_started_at_ms_min: Option<u64>,
464 #[serde(skip_serializing_if = "Option::is_none")]
465 pub last_status: Option<u16>,
466 #[serde(skip_serializing_if = "Option::is_none")]
467 pub last_duration_ms: Option<u64>,
468 #[serde(skip_serializing_if = "Option::is_none")]
469 pub last_ended_at_ms: Option<u64>,
470 #[serde(skip_serializing_if = "Option::is_none")]
471 pub last_model: Option<String>,
472 #[serde(skip_serializing_if = "Option::is_none")]
473 pub last_reasoning_effort: Option<String>,
474 #[serde(skip_serializing_if = "Option::is_none")]
475 pub last_service_tier: Option<String>,
476 #[serde(skip_serializing_if = "Option::is_none")]
477 pub last_provider_id: Option<String>,
478 #[serde(skip_serializing_if = "Option::is_none")]
479 pub last_station_name: Option<String>,
480 #[serde(skip_serializing_if = "Option::is_none")]
481 pub last_upstream_base_url: Option<String>,
482 #[serde(skip_serializing_if = "Option::is_none")]
483 pub last_usage: Option<UsageMetrics>,
484 #[serde(skip_serializing_if = "Option::is_none")]
485 pub total_usage: Option<UsageMetrics>,
486 #[serde(skip_serializing_if = "Option::is_none")]
487 pub turns_total: Option<u64>,
488 #[serde(skip_serializing_if = "Option::is_none")]
489 pub turns_with_usage: Option<u64>,
490 #[serde(skip_serializing_if = "Option::is_none")]
491 pub binding_profile_name: Option<String>,
492 #[serde(skip_serializing_if = "Option::is_none")]
493 pub binding_continuity_mode: Option<SessionContinuityMode>,
494 #[serde(skip_serializing_if = "Option::is_none")]
495 pub last_route_decision: Option<RouteDecisionProvenance>,
496 #[serde(skip_serializing_if = "Option::is_none")]
497 pub route_affinity: Option<SessionRouteAffinity>,
498 #[serde(skip_serializing_if = "Option::is_none")]
499 pub effective_model: Option<ResolvedRouteValue>,
500 #[serde(skip_serializing_if = "Option::is_none")]
501 pub effective_reasoning_effort: Option<ResolvedRouteValue>,
502 #[serde(skip_serializing_if = "Option::is_none")]
503 pub effective_service_tier: Option<ResolvedRouteValue>,
504 #[serde(skip_serializing_if = "Option::is_none")]
505 pub effective_station: Option<ResolvedRouteValue>,
506 #[serde(skip_serializing_if = "Option::is_none")]
507 pub effective_upstream_base_url: Option<ResolvedRouteValue>,
508 #[serde(skip_serializing_if = "Option::is_none")]
509 pub override_effort: Option<String>,
510 #[serde(skip_serializing_if = "Option::is_none")]
511 pub override_station_name: Option<String>,
512 #[serde(skip_serializing_if = "Option::is_none")]
513 pub override_model: Option<String>,
514 #[serde(skip_serializing_if = "Option::is_none")]
515 pub override_service_tier: Option<String>,
516}
517
518#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
519pub struct SessionManualOverrides {
520 #[serde(skip_serializing_if = "Option::is_none")]
521 pub reasoning_effort: Option<String>,
522 #[serde(skip_serializing_if = "Option::is_none")]
523 pub station_name: Option<String>,
524 #[serde(skip_serializing_if = "Option::is_none")]
525 pub route_target: Option<String>,
526 #[serde(skip_serializing_if = "Option::is_none")]
527 pub model: Option<String>,
528 #[serde(skip_serializing_if = "Option::is_none")]
529 pub service_tier: Option<String>,
530}
531
532impl SessionManualOverrides {
533 pub fn is_empty(&self) -> bool {
534 self.reasoning_effort.is_none()
535 && self.station_name.is_none()
536 && self.route_target.is_none()
537 && self.model.is_none()
538 && self.service_tier.is_none()
539 }
540}
541
542#[derive(Debug, Clone)]
543pub(super) struct SessionEffortOverride {
544 pub(super) effort: String,
545 #[allow(dead_code)]
546 pub(super) updated_at_ms: u64,
547 pub(super) last_seen_ms: u64,
548}
549
550#[derive(Debug, Clone)]
551pub(super) struct SessionStationOverride {
552 pub(super) station_name: String,
553 #[allow(dead_code)]
554 pub(super) updated_at_ms: u64,
555 pub(super) last_seen_ms: u64,
556}
557
558#[derive(Debug, Clone)]
559pub(super) struct SessionRouteTargetOverride {
560 pub(super) target: String,
561 #[allow(dead_code)]
562 pub(super) updated_at_ms: u64,
563 pub(super) last_seen_ms: u64,
564}
565
566#[derive(Debug, Clone)]
567pub(super) struct SessionModelOverride {
568 pub(super) model: String,
569 #[allow(dead_code)]
570 pub(super) updated_at_ms: u64,
571 pub(super) last_seen_ms: u64,
572}
573
574#[derive(Debug, Clone)]
575pub(super) struct SessionServiceTierOverride {
576 pub(super) service_tier: String,
577 #[allow(dead_code)]
578 pub(super) updated_at_ms: u64,
579 pub(super) last_seen_ms: u64,
580}
581
582#[derive(Debug, Clone)]
583pub(super) struct SessionBindingEntry {
584 pub(super) binding: SessionBinding,
585}
586
587#[derive(Debug, Clone)]
588pub(super) struct SessionCwdCacheEntry {
589 pub(super) cwd: Option<String>,
590 pub(super) last_seen_ms: u64,
591}
592
593fn empty_session_identity_card(session_id: Option<String>) -> SessionIdentityCard {
594 SessionIdentityCard {
595 session_id,
596 session_identity_source: None,
597 observation_scope: SessionObservationScope::ObservedOnly,
598 host_local_transcript_path: None,
599 last_client_name: None,
600 last_client_addr: None,
601 cwd: None,
602 active_count: 0,
603 active_started_at_ms_min: None,
604 last_status: None,
605 last_duration_ms: None,
606 last_ended_at_ms: None,
607 last_model: None,
608 last_reasoning_effort: None,
609 last_service_tier: None,
610 last_provider_id: None,
611 last_station_name: None,
612 last_upstream_base_url: None,
613 last_usage: None,
614 total_usage: None,
615 turns_total: None,
616 turns_with_usage: None,
617 binding_profile_name: None,
618 binding_continuity_mode: None,
619 last_route_decision: None,
620 route_affinity: None,
621 effective_model: None,
622 effective_reasoning_effort: None,
623 effective_service_tier: None,
624 effective_station: None,
625 effective_upstream_base_url: None,
626 override_effort: None,
627 override_station_name: None,
628 override_model: None,
629 override_service_tier: None,
630 }
631}
632
633fn non_empty_trimmed(value: Option<&str>) -> Option<String> {
634 value
635 .map(str::trim)
636 .filter(|value| !value.is_empty())
637 .map(ToOwned::to_owned)
638}
639
640fn resolve_effective_observed_value(
641 override_value: Option<&str>,
642 observed_value: Option<&str>,
643 binding_value: Option<&str>,
644) -> Option<ResolvedRouteValue> {
645 if let Some(value) = non_empty_trimmed(override_value) {
646 return Some(ResolvedRouteValue::new(
647 value,
648 RouteValueSource::SessionOverride,
649 ));
650 }
651 let binding_value = non_empty_trimmed(binding_value);
652 if let Some(binding) = binding_value {
653 return Some(ResolvedRouteValue::new(
654 binding,
655 RouteValueSource::ProfileDefault,
656 ));
657 }
658 non_empty_trimmed(observed_value)
659 .map(|observed| ResolvedRouteValue::new(observed, RouteValueSource::RequestPayload))
660}
661
662fn binding_request_field_value<'a>(
663 binding: Option<&'a SessionBinding>,
664 value: impl FnOnce(&'a SessionBinding) -> Option<&'a str>,
665) -> Option<&'a str> {
666 let binding = binding?;
667 if binding.continuity_mode != SessionContinuityMode::ManualProfile {
668 return None;
669 }
670 value(binding)
671}
672
673fn classify_session_observation_scope(card: &SessionIdentityCard) -> SessionObservationScope {
674 if card.cwd.is_some() {
675 SessionObservationScope::HostLocalEnriched
676 } else {
677 SessionObservationScope::ObservedOnly
678 }
679}
680
681fn resolve_effective_station_value(
682 card: &SessionIdentityCard,
683 global_station_override: Option<&str>,
684 binding_station_name: Option<&str>,
685) -> Option<ResolvedRouteValue> {
686 if let Some(value) = non_empty_trimmed(card.override_station_name.as_deref()) {
687 return Some(ResolvedRouteValue::new(
688 value,
689 RouteValueSource::SessionOverride,
690 ));
691 }
692 if let Some(value) = non_empty_trimmed(global_station_override) {
693 return Some(ResolvedRouteValue::new(
694 value,
695 RouteValueSource::GlobalOverride,
696 ));
697 }
698 let binding = non_empty_trimmed(binding_station_name);
699 if let Some(binding) = binding {
700 return Some(ResolvedRouteValue::new(
701 binding,
702 RouteValueSource::ProfileDefault,
703 ));
704 }
705 non_empty_trimmed(card.last_station_name.as_deref())
706 .map(|observed| ResolvedRouteValue::new(observed, RouteValueSource::RuntimeFallback))
707}
708
709fn apply_basic_effective_route(
710 card: &mut SessionIdentityCard,
711 global_station_override: Option<&str>,
712 binding: Option<&SessionBinding>,
713) {
714 card.effective_model = resolve_effective_observed_value(
715 card.override_model.as_deref(),
716 card.last_model.as_deref(),
717 binding_request_field_value(binding, |binding| binding.model.as_deref()),
718 );
719 card.effective_reasoning_effort = resolve_effective_observed_value(
720 card.override_effort.as_deref(),
721 card.last_reasoning_effort.as_deref(),
722 binding_request_field_value(binding, |binding| binding.reasoning_effort.as_deref()),
723 );
724 card.effective_service_tier = resolve_effective_observed_value(
725 card.override_service_tier.as_deref(),
726 card.last_service_tier.as_deref(),
727 binding_request_field_value(binding, |binding| binding.service_tier.as_deref()),
728 );
729 card.binding_profile_name = binding.and_then(|binding| binding.profile_name.clone());
730 card.binding_continuity_mode = binding.map(|binding| binding.continuity_mode);
731 card.effective_station = resolve_effective_station_value(
732 card,
733 global_station_override,
734 binding.and_then(|binding| binding.station_name.as_deref()),
735 );
736 card.effective_upstream_base_url = match (
737 card.effective_station.as_ref(),
738 non_empty_trimmed(card.last_station_name.as_deref()),
739 non_empty_trimmed(card.last_upstream_base_url.as_deref()),
740 ) {
741 (Some(station), Some(last_station), Some(upstream)) if station.value == last_station => {
742 Some(ResolvedRouteValue::new(
743 upstream,
744 RouteValueSource::RuntimeFallback,
745 ))
746 }
747 _ => None,
748 };
749}
750
751pub fn enrich_session_identity_cards_with_runtime(
752 cards: &mut [SessionIdentityCard],
753 mgr: &ServiceConfigManager,
754) {
755 for card in cards {
756 if card.effective_station.is_none()
757 && let Some(active) = mgr.active_station()
758 {
759 card.effective_station = Some(ResolvedRouteValue::new(
760 active.name.clone(),
761 RouteValueSource::RuntimeFallback,
762 ));
763 }
764
765 let effective_station_name = card
766 .effective_station
767 .as_ref()
768 .map(|value| value.value.as_str());
769 if card.effective_upstream_base_url.is_none()
770 && let Some(station_name) = effective_station_name
771 && let Some(station) = mgr.station(station_name)
772 && station.upstreams.len() == 1
773 {
774 card.effective_upstream_base_url = Some(ResolvedRouteValue::new(
775 station.upstreams[0].base_url.clone(),
776 RouteValueSource::RuntimeFallback,
777 ));
778 }
779
780 let Some(model) = card
781 .effective_model
782 .as_ref()
783 .map(|value| value.value.clone())
784 else {
785 continue;
786 };
787 let Some(station_name) = effective_station_name else {
788 continue;
789 };
790 let Some(last_station_name) = card.last_station_name.as_deref() else {
791 continue;
792 };
793 if last_station_name != station_name {
794 continue;
795 }
796 let Some(last_upstream_base_url) = card.last_upstream_base_url.as_deref() else {
797 continue;
798 };
799 let Some(station) = mgr.station(station_name) else {
800 continue;
801 };
802 let Some(upstream) = station
803 .upstreams
804 .iter()
805 .find(|upstream| upstream.base_url == last_upstream_base_url)
806 else {
807 continue;
808 };
809
810 let mapped = crate::model_routing::effective_model(&upstream.model_mapping, model.as_str());
811 if mapped != model {
812 card.effective_model = Some(ResolvedRouteValue::new(
813 mapped,
814 RouteValueSource::StationMapping,
815 ));
816 }
817 }
818}
819
820fn session_identity_sort_key(card: &SessionIdentityCard) -> u64 {
821 card.last_ended_at_ms
822 .unwrap_or(0)
823 .max(card.active_started_at_ms_min.unwrap_or(0))
824}
825
826fn route_decision_at_ms(route_decision: Option<&RouteDecisionProvenance>) -> u64 {
827 route_decision
828 .map(|decision| decision.decided_at_ms)
829 .unwrap_or(0)
830}
831
832fn update_card_route_decision(
833 card: &mut SessionIdentityCard,
834 route_decision: Option<&RouteDecisionProvenance>,
835) {
836 let Some(route_decision) = route_decision.cloned() else {
837 return;
838 };
839 let current_at = route_decision_at_ms(card.last_route_decision.as_ref());
840 if current_at <= route_decision.decided_at_ms {
841 card.last_route_decision = Some(route_decision);
842 }
843}
844
845fn merge_session_identity_source(
846 target: &mut Option<SessionIdentitySource>,
847 source: Option<SessionIdentitySource>,
848) {
849 if source.is_some() {
850 *target = source;
851 }
852}
853
854pub struct SessionIdentityCardBuildInputs<'a> {
855 pub active: &'a [ActiveRequest],
856 pub recent: &'a [FinishedRequest],
857 pub overrides: &'a HashMap<String, String>,
858 pub station_overrides: &'a HashMap<String, String>,
859 pub model_overrides: &'a HashMap<String, String>,
860 pub service_tier_overrides: &'a HashMap<String, String>,
861 pub bindings: &'a HashMap<String, SessionBinding>,
862 pub route_affinities: &'a HashMap<String, SessionRouteAffinity>,
863 pub global_station_override: Option<&'a str>,
864 pub stats: &'a HashMap<String, SessionStats>,
865}
866
867pub fn build_session_identity_cards_from_parts(
868 inputs: SessionIdentityCardBuildInputs<'_>,
869) -> Vec<SessionIdentityCard> {
870 let SessionIdentityCardBuildInputs {
871 active,
872 recent,
873 overrides,
874 station_overrides,
875 model_overrides,
876 service_tier_overrides,
877 bindings,
878 route_affinities,
879 global_station_override,
880 stats,
881 } = inputs;
882
883 use std::collections::HashMap as StdHashMap;
884
885 let mut map: StdHashMap<Option<String>, SessionIdentityCard> = StdHashMap::new();
886
887 for req in active {
888 let key = req.session_id.clone();
889 let entry = map
890 .entry(key.clone())
891 .or_insert_with(|| empty_session_identity_card(key));
892 merge_session_identity_source(
893 &mut entry.session_identity_source,
894 req.session_identity_source,
895 );
896
897 entry.active_count = entry.active_count.saturating_add(1);
898 entry.active_started_at_ms_min = Some(
899 entry
900 .active_started_at_ms_min
901 .unwrap_or(req.started_at_ms)
902 .min(req.started_at_ms),
903 );
904 if entry.cwd.is_none() {
905 entry.cwd = req.cwd.clone();
906 }
907 if entry.last_client_name.is_none() {
908 entry.last_client_name = req.client_name.clone();
909 }
910 if entry.last_client_addr.is_none() {
911 entry.last_client_addr = req.client_addr.clone();
912 }
913 if let Some(effort) = req.reasoning_effort.as_ref() {
914 entry.last_reasoning_effort = Some(effort.clone());
915 }
916 if let Some(service_tier) = req.service_tier.as_ref() {
917 entry.last_service_tier = Some(service_tier.clone());
918 }
919 if entry.last_model.is_none() {
920 entry.last_model = req.model.clone();
921 }
922 if entry.last_provider_id.is_none() {
923 entry.last_provider_id = req.provider_id.clone();
924 }
925 if entry.last_station_name.is_none() {
926 entry.last_station_name = req.station_name.clone();
927 }
928 if entry.last_upstream_base_url.is_none() {
929 entry.last_upstream_base_url = req.upstream_base_url.clone();
930 }
931 update_card_route_decision(entry, req.route_decision.as_ref());
932 }
933
934 for r in recent {
935 let key = r.session_id.clone();
936 let entry = map
937 .entry(key.clone())
938 .or_insert_with(|| empty_session_identity_card(key));
939 merge_session_identity_source(
940 &mut entry.session_identity_source,
941 r.session_identity_source,
942 );
943
944 let should_update = entry
945 .last_ended_at_ms
946 .is_none_or(|prev| r.ended_at_ms >= prev);
947 if should_update {
948 entry.last_status = Some(r.status_code);
949 entry.last_duration_ms = Some(r.duration_ms);
950 entry.last_ended_at_ms = Some(r.ended_at_ms);
951 entry.last_client_name = r.client_name.clone().or(entry.last_client_name.clone());
952 entry.last_client_addr = r.client_addr.clone().or(entry.last_client_addr.clone());
953 entry.last_model = r.model.clone().or(entry.last_model.clone());
954 entry.last_reasoning_effort = r
955 .reasoning_effort
956 .clone()
957 .or(entry.last_reasoning_effort.clone());
958 entry.last_service_tier = r.service_tier.clone().or(entry.last_service_tier.clone());
959 entry.last_provider_id = r.provider_id.clone().or(entry.last_provider_id.clone());
960 entry.last_station_name = r.station_name.clone().or(entry.last_station_name.clone());
961 entry.last_upstream_base_url = r
962 .upstream_base_url
963 .clone()
964 .or(entry.last_upstream_base_url.clone());
965 entry.last_usage = r.usage.clone().or(entry.last_usage.clone());
966 }
967 if entry.cwd.is_none() {
968 entry.cwd = r.cwd.clone();
969 }
970 update_card_route_decision(entry, r.route_decision.as_ref());
971 }
972
973 for (sid, st) in stats {
974 let key = Some(sid.clone());
975 let entry = map
976 .entry(key.clone())
977 .or_insert_with(|| empty_session_identity_card(key));
978 merge_session_identity_source(
979 &mut entry.session_identity_source,
980 st.last_session_identity_source,
981 );
982
983 if entry.turns_total.is_none() {
984 entry.turns_total = Some(st.turns_total);
985 }
986 if entry.last_client_name.is_none() {
987 entry.last_client_name = st.last_client_name.clone();
988 }
989 if entry.last_client_addr.is_none() {
990 entry.last_client_addr = st.last_client_addr.clone();
991 }
992 if entry.last_status.is_none() {
993 entry.last_status = st.last_status;
994 }
995 if entry.last_duration_ms.is_none() {
996 entry.last_duration_ms = st.last_duration_ms;
997 }
998 if entry.last_ended_at_ms.is_none() {
999 entry.last_ended_at_ms = st.last_ended_at_ms;
1000 }
1001 if entry.last_model.is_none() {
1002 entry.last_model = st.last_model.clone();
1003 }
1004 if entry.last_reasoning_effort.is_none() {
1005 entry.last_reasoning_effort = st.last_reasoning_effort.clone();
1006 }
1007 if entry.last_service_tier.is_none() {
1008 entry.last_service_tier = st.last_service_tier.clone();
1009 }
1010 if entry.last_provider_id.is_none() {
1011 entry.last_provider_id = st.last_provider_id.clone();
1012 }
1013 if entry.last_station_name.is_none() {
1014 entry.last_station_name = st.last_station_name.clone();
1015 }
1016 if entry.last_usage.is_none() {
1017 entry.last_usage = st.last_usage.clone();
1018 }
1019 if entry.total_usage.is_none() {
1020 entry.total_usage = Some(st.total_usage.clone());
1021 }
1022 if entry.turns_with_usage.is_none() {
1023 entry.turns_with_usage = Some(st.turns_with_usage);
1024 }
1025 update_card_route_decision(entry, st.last_route_decision.as_ref());
1026 }
1027
1028 for (sid, eff) in overrides {
1029 let key = Some(sid.clone());
1030 let entry = map
1031 .entry(key.clone())
1032 .or_insert_with(|| empty_session_identity_card(key));
1033 entry.override_effort = Some(eff.clone());
1034 }
1035
1036 for (sid, station_name) in station_overrides {
1037 let key = Some(sid.clone());
1038 let entry = map
1039 .entry(key.clone())
1040 .or_insert_with(|| empty_session_identity_card(key));
1041 entry.override_station_name = Some(station_name.clone());
1042 }
1043
1044 for (sid, model) in model_overrides {
1045 let key = Some(sid.clone());
1046 let entry = map
1047 .entry(key.clone())
1048 .or_insert_with(|| empty_session_identity_card(key));
1049 entry.override_model = Some(model.clone());
1050 }
1051
1052 for (sid, service_tier) in service_tier_overrides {
1053 let key = Some(sid.clone());
1054 let entry = map
1055 .entry(key.clone())
1056 .or_insert_with(|| empty_session_identity_card(key));
1057 entry.override_service_tier = Some(service_tier.clone());
1058 }
1059
1060 for (sid, affinity) in route_affinities {
1061 let key = Some(sid.clone());
1062 let entry = map
1063 .entry(key.clone())
1064 .or_insert_with(|| empty_session_identity_card(key));
1065 merge_session_identity_source(
1066 &mut entry.session_identity_source,
1067 affinity.session_identity_source,
1068 );
1069 entry.route_affinity = Some(affinity.clone());
1070 }
1071
1072 let mut cards = map.into_values().collect::<Vec<_>>();
1073 for card in &mut cards {
1074 let binding = card
1075 .session_id
1076 .as_deref()
1077 .and_then(|session_id| bindings.get(session_id));
1078 apply_basic_effective_route(card, global_station_override, binding);
1079 card.observation_scope = classify_session_observation_scope(card);
1080 }
1081 cards.sort_by_key(|card| std::cmp::Reverse(session_identity_sort_key(card)));
1082 cards
1083}
1084
1085pub async fn enrich_session_identity_cards_with_host_transcripts(
1086 cards: &mut [SessionIdentityCard],
1087) {
1088 let session_ids = cards
1089 .iter()
1090 .filter_map(|card| card.session_id.as_deref())
1091 .map(str::to_owned)
1092 .collect::<Vec<_>>();
1093 if session_ids.is_empty() {
1094 return;
1095 }
1096
1097 let found = match sessions::find_codex_session_files_by_ids(&session_ids).await {
1098 Ok(found) => found,
1099 Err(_) => return,
1100 };
1101
1102 for card in cards {
1103 card.host_local_transcript_path = card
1104 .session_id
1105 .as_deref()
1106 .and_then(|sid| found.get(sid))
1107 .map(|path| path.to_string_lossy().to_string());
1108 }
1109}
1110
1111#[cfg(test)]
1112mod tests {
1113 use super::*;
1114
1115 use crate::logging::RetryInfo;
1116 use crate::pricing::CostBreakdown;
1117 use crate::usage::UsageMetrics;
1118
1119 fn sample_finished_request() -> FinishedRequest {
1120 FinishedRequest {
1121 id: 7,
1122 trace_id: Some("codex-7".to_string()),
1123 session_id: Some("sid".to_string()),
1124 session_identity_source: Some(SessionIdentitySource::Header),
1125 client_name: None,
1126 client_addr: None,
1127 cwd: None,
1128 model: Some("gpt-5".to_string()),
1129 reasoning_effort: None,
1130 service_tier: Some("default".to_string()),
1131 station_name: Some("primary".to_string()),
1132 provider_id: Some("primary-provider".to_string()),
1133 upstream_base_url: Some("https://primary.example/v1".to_string()),
1134 route_decision: Some(RouteDecisionProvenance {
1135 effective_service_tier: Some(ResolvedRouteValue {
1136 value: "priority".to_string(),
1137 source: RouteValueSource::ProfileDefault,
1138 }),
1139 ..Default::default()
1140 }),
1141 usage: Some(UsageMetrics {
1142 output_tokens: 200,
1143 total_tokens: 200,
1144 ..UsageMetrics::default()
1145 }),
1146 cost: CostBreakdown::default(),
1147 retry: Some(RetryInfo {
1148 attempts: 2,
1149 upstream_chain: vec![
1150 "backup:https://backup.example/v1 (idx=0) transport_error=timeout".to_string(),
1151 "primary:https://primary.example/v1 (idx=0) status=200 class=-".to_string(),
1152 ],
1153 route_attempts: Vec::new(),
1154 }),
1155 observability: RequestObservability::default(),
1156 service: "codex".to_string(),
1157 method: "POST".to_string(),
1158 path: "/v1/responses".to_string(),
1159 status_code: 200,
1160 duration_ms: 1_500,
1161 ttfb_ms: Some(500),
1162 streaming: true,
1163 ended_at_ms: 2_000,
1164 }
1165 }
1166
1167 #[test]
1168 fn finished_request_observability_derives_canonical_request_facts() {
1169 let request = sample_finished_request();
1170
1171 let observability = request.observability_view();
1172
1173 assert_eq!(observability.trace_id.as_deref(), Some("codex-7"));
1174 assert_eq!(observability.duration_ms, Some(1_500));
1175 assert_eq!(observability.ttfb_ms, Some(500));
1176 assert_eq!(observability.generation_ms, Some(1_000));
1177 assert_eq!(observability.attempt_count, 2);
1178 assert_eq!(observability.route_attempt_count, 2);
1179 assert!(observability.retried);
1180 assert!(observability.cross_station_failover);
1181 assert!(observability.fast_mode);
1182 assert!(observability.streaming);
1183 assert_eq!(observability.output_tokens_per_second, Some(200.0));
1184 }
1185
1186 #[test]
1187 fn finished_request_serializes_materialized_observability_for_operator_api() {
1188 let mut request = sample_finished_request();
1189 request.refresh_observability();
1190
1191 let value = serde_json::to_value(&request).expect("finished request json");
1192
1193 assert_eq!(value["observability"]["trace_id"].as_str(), Some("codex-7"));
1194 assert_eq!(
1195 value["observability"]["generation_ms"].as_u64(),
1196 Some(1_000)
1197 );
1198 assert_eq!(
1199 value["observability"]["output_tokens_per_second"].as_f64(),
1200 Some(200.0)
1201 );
1202 assert_eq!(value["observability"]["attempt_count"].as_u64(), Some(2));
1203 assert_eq!(value["observability"]["streaming"].as_bool(), Some(true));
1204 }
1205
1206 #[test]
1207 fn finished_request_legacy_payload_still_derives_observability() {
1208 let request: FinishedRequest = serde_json::from_value(serde_json::json!({
1209 "id": 8,
1210 "trace_id": "codex-8",
1211 "usage": {
1212 "output_tokens": 120,
1213 "total_tokens": 120
1214 },
1215 "cost": {},
1216 "service": "codex",
1217 "method": "POST",
1218 "path": "/v1/responses",
1219 "status_code": 200,
1220 "duration_ms": 1_500,
1221 "ttfb_ms": 300,
1222 "ended_at_ms": 2_500
1223 }))
1224 .expect("legacy finished request");
1225
1226 assert_eq!(request.observability.attempt_count, 1);
1227 assert!(!request.streaming);
1228
1229 let observability = request.observability_view();
1230 assert_eq!(observability.generation_ms, Some(1_200));
1231 assert_eq!(observability.output_tokens_per_second, Some(100.0));
1232 assert_eq!(observability.attempt_count, 1);
1233 assert!(!observability.fast_mode);
1234 }
1235}