1use std::collections::HashMap;
2use std::fs::File;
3use std::io::{BufRead, BufReader};
4use std::path::Path;
5
6use serde_json::Value as JsonValue;
7
8pub use crate::logging::request_log_path;
9use crate::pricing::{CostAdjustments, estimate_request_cost_from_operator_catalog_for_service};
10use crate::state::{
11 FinishedRequest, RequestObservability, RouteDecisionProvenance, SessionIdentitySource,
12};
13use crate::usage::{CacheInputAccounting, UsageMetrics};
14
15#[derive(Debug, Clone, PartialEq)]
16pub struct RequestLogLine {
17 raw: String,
18 value: Option<JsonValue>,
19}
20
21impl RequestLogLine {
22 pub fn from_raw(raw: impl Into<String>) -> Self {
23 let raw = raw.into();
24 let value = serde_json::from_str::<JsonValue>(&raw).ok();
25 Self { raw, value }
26 }
27
28 pub fn raw(&self) -> &str {
29 &self.raw
30 }
31
32 pub fn value(&self) -> Option<&JsonValue> {
33 self.value.as_ref()
34 }
35
36 pub fn is_valid_json(&self) -> bool {
37 self.value.is_some()
38 }
39
40 pub fn display_lines(&self) -> Vec<String> {
41 self.value
42 .as_ref()
43 .map(format_request_log_record_lines)
44 .unwrap_or_default()
45 }
46}
47
48#[derive(Debug, Default, Clone, PartialEq, Eq)]
49pub struct RequestLogFilters {
50 pub session: Option<String>,
51 pub model: Option<String>,
52 pub station: Option<String>,
53 pub provider: Option<String>,
54 pub path: Option<String>,
55 pub status_min: Option<u64>,
56 pub status_max: Option<u64>,
57 pub fast: bool,
58 pub retried: bool,
59}
60
61impl RequestLogFilters {
62 pub fn is_empty(&self) -> bool {
63 self.session.is_none()
64 && self.model.is_none()
65 && self.station.is_none()
66 && self.provider.is_none()
67 && self.path.is_none()
68 && self.status_min.is_none()
69 && self.status_max.is_none()
70 && !self.fast
71 && !self.retried
72 }
73
74 pub fn matches(&self, record: &JsonValue) -> bool {
75 if let Some(expected) = self.session.as_deref()
76 && !field_contains(str_field(record, "session_id"), expected)
77 {
78 return false;
79 }
80 if let Some(expected) = self.model.as_deref()
81 && !field_contains(request_model(record).as_deref(), expected)
82 {
83 return false;
84 }
85 if let Some(expected) = self.station.as_deref()
86 && !field_contains(Some(station_name(record)), expected)
87 {
88 return false;
89 }
90 if let Some(expected) = self.provider.as_deref()
91 && !field_contains(str_field(record, "provider_id"), expected)
92 {
93 return false;
94 }
95 if let Some(expected) = self.path.as_deref()
96 && !field_contains(str_field(record, "path"), expected)
97 {
98 return false;
99 }
100 if let Some(min) = self.status_min
101 && u64_field(record, "status_code").unwrap_or(0) < min
102 {
103 return false;
104 }
105 if let Some(max) = self.status_max
106 && u64_field(record, "status_code").unwrap_or(0) > max
107 {
108 return false;
109 }
110 if self.fast && !request_is_fast(record) {
111 return false;
112 }
113 if self.retried && !request_was_retried(record) {
114 return false;
115 }
116 true
117 }
118}
119
120#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
121pub struct RequestUsageAggregate {
122 pub requests: u64,
123 pub duration_ms_total: u64,
124 pub input_tokens: i64,
125 pub output_tokens: i64,
126 pub reasoning_tokens: i64,
127 pub cache_read_input_tokens: i64,
128 pub cache_creation_input_tokens: i64,
129 pub total_tokens: i64,
130}
131
132impl RequestUsageAggregate {
133 pub fn record(
134 &mut self,
135 duration_ms: u64,
136 usage: Option<&UsageMetrics>,
137 accounting: CacheInputAccounting,
138 ) {
139 self.requests = self.requests.saturating_add(1);
140 self.duration_ms_total = self.duration_ms_total.saturating_add(duration_ms);
141 let Some(usage) = usage else {
142 return;
143 };
144
145 self.input_tokens = self.input_tokens.saturating_add(usage.input_tokens.max(0));
146 self.output_tokens = self
147 .output_tokens
148 .saturating_add(usage.output_tokens.max(0));
149 self.reasoning_tokens = self
150 .reasoning_tokens
151 .saturating_add(reasoning_tokens(usage));
152 self.cache_read_input_tokens = self
153 .cache_read_input_tokens
154 .saturating_add(usage.cache_read_tokens_total());
155 self.cache_creation_input_tokens = self
156 .cache_creation_input_tokens
157 .saturating_add(cache_creation_tokens(usage));
158 self.total_tokens = self
159 .total_tokens
160 .saturating_add(total_tokens(usage, accounting));
161 }
162
163 pub fn average_duration_ms(&self) -> u64 {
164 self.duration_ms_total
165 .checked_div(self.requests)
166 .unwrap_or(0)
167 }
168
169 pub fn summary_line(&self, station_name: &str) -> String {
170 format!(
171 "{} | {} | {} | {} | {} | {} | {} | {} | {}",
172 station_name,
173 self.requests,
174 self.input_tokens,
175 self.output_tokens,
176 self.cache_read_input_tokens,
177 self.cache_creation_input_tokens,
178 self.reasoning_tokens,
179 self.total_tokens,
180 self.average_duration_ms()
181 )
182 }
183}
184
185#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default)]
186#[serde(rename_all = "snake_case")]
187pub enum RequestUsageSummaryGroup {
188 #[default]
189 Station,
190 Provider,
191 Model,
192 Session,
193}
194
195impl RequestUsageSummaryGroup {
196 pub fn column_name(self) -> &'static str {
197 match self {
198 Self::Station => "station_name",
199 Self::Provider => "provider_id",
200 Self::Model => "model",
201 Self::Session => "session_id",
202 }
203 }
204
205 fn key(self, record: &JsonValue) -> String {
206 match self {
207 Self::Station => station_name(record).to_string(),
208 Self::Provider => str_field(record, "provider_id").unwrap_or("-").to_string(),
209 Self::Model => request_model(record).unwrap_or_else(|| "-".to_string()),
210 Self::Session => str_field(record, "session_id").unwrap_or("-").to_string(),
211 }
212 }
213}
214
215#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
216pub struct RequestUsageSummaryRow {
217 pub group_value: String,
218 pub aggregate: RequestUsageAggregate,
219}
220
221pub fn read_request_log_lines(path: &Path) -> std::io::Result<Vec<RequestLogLine>> {
222 let file = File::open(path)?;
223 let reader = BufReader::new(file);
224 Ok(reader
225 .lines()
226 .map_while(Result::ok)
227 .map(RequestLogLine::from_raw)
228 .collect())
229}
230
231pub fn tail_request_log(path: &Path, limit: usize) -> std::io::Result<Vec<RequestLogLine>> {
232 let lines = read_request_log_lines(path)?;
233 let total = lines.len();
234 let start = total.saturating_sub(limit);
235 Ok(lines[start..].to_vec())
236}
237
238pub fn tail_finished_requests_from_log(
239 path: &Path,
240 limit: usize,
241) -> std::io::Result<Vec<FinishedRequest>> {
242 let lines = tail_request_log(path, limit)?;
243 let mut requests = lines
244 .iter()
245 .filter_map(|line| {
246 line.value()
247 .and_then(finished_request_from_request_log_record)
248 })
249 .collect::<Vec<_>>();
250 requests.reverse();
251 Ok(requests)
252}
253
254pub fn find_finished_requests_from_log(
255 path: &Path,
256 filters: &RequestLogFilters,
257 limit: usize,
258) -> std::io::Result<Vec<FinishedRequest>> {
259 let lines = find_request_log(path, filters, limit)?;
260 Ok(lines
261 .iter()
262 .filter_map(|line| {
263 line.value()
264 .and_then(finished_request_from_request_log_record)
265 })
266 .collect())
267}
268
269pub fn summarize_request_log(
270 path: &Path,
271 group: RequestUsageSummaryGroup,
272 filters: &RequestLogFilters,
273 limit: usize,
274) -> std::io::Result<Vec<RequestUsageSummaryRow>> {
275 let lines = read_request_log_lines(path)?;
276 Ok(summarize_request_log_lines(
277 lines.iter(),
278 group,
279 filters,
280 limit,
281 ))
282}
283
284pub fn find_request_log(
285 path: &Path,
286 filters: &RequestLogFilters,
287 limit: usize,
288) -> std::io::Result<Vec<RequestLogLine>> {
289 let lines = read_request_log_lines(path)?;
290 Ok(lines
291 .iter()
292 .rev()
293 .filter(|line| line.value().is_some_and(|record| filters.matches(record)))
294 .take(limit)
295 .cloned()
296 .collect())
297}
298
299fn summarize_request_log_lines<'a>(
300 lines: impl IntoIterator<Item = &'a RequestLogLine>,
301 group: RequestUsageSummaryGroup,
302 filters: &RequestLogFilters,
303 limit: usize,
304) -> Vec<RequestUsageSummaryRow> {
305 let mut aggregate: HashMap<String, RequestUsageAggregate> = HashMap::new();
306 for line in lines {
307 let Some(record) = line.value() else {
308 continue;
309 };
310 if !filters.matches(record) {
311 continue;
312 }
313 let group_value = group.key(record);
314 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
315 let service = str_field(record, "service").unwrap_or("-");
316 let usage = usage_metrics(record);
317 let entry = aggregate.entry(group_value).or_default();
318 entry.record(
319 duration_ms,
320 usage.as_ref(),
321 CacheInputAccounting::for_service(service),
322 );
323 }
324
325 let mut items: Vec<RequestUsageSummaryRow> = aggregate
326 .into_iter()
327 .map(|(group_value, aggregate)| RequestUsageSummaryRow {
328 group_value,
329 aggregate,
330 })
331 .collect();
332 items.sort_by(|a, b| {
333 b.aggregate
334 .total_tokens
335 .cmp(&a.aggregate.total_tokens)
336 .then_with(|| a.group_value.cmp(&b.group_value))
337 });
338 items.into_iter().take(limit).collect()
339}
340
341pub fn format_request_log_record_lines(record: &JsonValue) -> Vec<String> {
342 let ts = i64_field(record, "timestamp_ms").unwrap_or(0);
343 let service = str_field(record, "service").unwrap_or("-");
344 let method = str_field(record, "method").unwrap_or("-");
345 let path = str_field(record, "path").unwrap_or("-");
346 let status = u64_field(record, "status_code").unwrap_or(0);
347 let provider = str_field(record, "provider_id").unwrap_or("-");
348 let endpoint = provider_endpoint_display(record);
349 let station = station_name(record);
350 let model = request_model(record).unwrap_or_else(|| "-".to_string());
351 let tier = service_tier_display(record);
352
353 let mut lines = vec![format!(
354 "[{}] {} {} {} status={} endpoint={} provider={} station={} model={} tier={}",
355 ts, service, method, path, status, endpoint, provider, station, model, tier
356 )];
357
358 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
359 let ttfb_ms = u64_field(record, "ttfb_ms");
360 let usage = usage_metrics(record);
361 let speed = usage
362 .as_ref()
363 .and_then(|usage| output_tokens_per_second(usage, duration_ms, ttfb_ms));
364 let cost = request_cost_display(service, model.as_str(), usage.as_ref());
365
366 lines.push(format!(
367 " timing duration={} ttfb={} output_speed={} cost={}",
368 format_ms(duration_ms),
369 format_optional_ms(ttfb_ms),
370 format_optional_speed(speed),
371 cost
372 ));
373
374 if let Some(usage) = usage.as_ref() {
375 lines.push(format!(
376 " tokens input={} output={} cache_read={} cache_create={} reasoning={} total={}",
377 usage.input_tokens.max(0),
378 usage.output_tokens.max(0),
379 usage.cache_read_tokens_total(),
380 cache_creation_tokens(usage),
381 reasoning_tokens(usage),
382 total_tokens(usage, CacheInputAccounting::for_service(service))
383 ));
384 } else {
385 lines.push(" tokens -".to_string());
386 }
387
388 lines
389}
390
391pub fn request_log_record_model(record: &JsonValue) -> Option<String> {
392 request_model(record)
393}
394
395pub fn request_log_record_station(record: &JsonValue) -> &str {
396 station_name(record)
397}
398
399pub fn request_log_record_is_fast(record: &JsonValue) -> bool {
400 request_is_fast(record)
401}
402
403pub fn request_log_record_was_retried(record: &JsonValue) -> bool {
404 request_was_retried(record)
405}
406
407pub fn finished_request_from_request_log_record(record: &JsonValue) -> Option<FinishedRequest> {
408 let timestamp_ms = u64_field(record, "timestamp_ms").unwrap_or(0);
409 let request_id = u64_field(record, "request_id").unwrap_or(timestamp_ms);
410 let status_code = u64_field(record, "status_code")
411 .and_then(|status| u16::try_from(status).ok())
412 .unwrap_or(0);
413 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
414 let usage = usage_metrics(record);
415 let model = request_model(record);
416 let service = str_field(record, "service").unwrap_or("-");
417 let cost = estimate_request_cost_from_operator_catalog_for_service(
418 model.as_deref(),
419 usage.as_ref(),
420 CostAdjustments::default(),
421 service,
422 );
423 let retry = record
424 .get("retry")
425 .and_then(|retry| serde_json::from_value(retry.clone()).ok());
426 let route_decision = record.get("route_decision").and_then(|route_decision| {
427 serde_json::from_value::<RouteDecisionProvenance>(route_decision.clone()).ok()
428 });
429 let session_identity_source = record
430 .get("session_identity_source")
431 .and_then(|source| serde_json::from_value::<SessionIdentitySource>(source.clone()).ok());
432
433 let mut request = FinishedRequest {
434 id: request_id,
435 trace_id: str_field(record, "trace_id").map(ToOwned::to_owned),
436 session_id: str_field(record, "session_id").map(ToOwned::to_owned),
437 session_identity_source,
438 client_name: str_field(record, "client_name").map(ToOwned::to_owned),
439 client_addr: str_field(record, "client_addr").map(ToOwned::to_owned),
440 cwd: str_field(record, "cwd").map(ToOwned::to_owned),
441 model,
442 reasoning_effort: str_field(record, "reasoning_effort").map(ToOwned::to_owned),
443 service_tier: service_tier_value(record),
444 station_name: non_dash(station_name(record)).map(ToOwned::to_owned),
445 provider_id: str_field(record, "provider_id").map(ToOwned::to_owned),
446 upstream_base_url: str_field(record, "upstream_base_url").map(ToOwned::to_owned),
447 route_decision,
448 usage,
449 cost,
450 retry,
451 observability: RequestObservability::default(),
452 service: service.to_string(),
453 method: str_field(record, "method").unwrap_or("-").to_string(),
454 path: str_field(record, "path").unwrap_or("-").to_string(),
455 status_code,
456 duration_ms,
457 ttfb_ms: u64_field(record, "ttfb_ms"),
458 streaming: record
459 .get("streaming")
460 .and_then(|value| value.as_bool())
461 .unwrap_or(false),
462 ended_at_ms: timestamp_ms,
463 };
464 request.refresh_observability();
465 Some(request)
466}
467
468fn field_contains(value: Option<&str>, expected: &str) -> bool {
469 let expected = expected.trim().to_ascii_lowercase();
470 if expected.is_empty() {
471 return true;
472 }
473 value
474 .map(|value| value.to_ascii_lowercase().contains(&expected))
475 .unwrap_or(false)
476}
477
478fn request_is_fast(record: &JsonValue) -> bool {
479 record
480 .get("observability")
481 .and_then(|observability| observability.get("fast_mode"))
482 .and_then(|value| value.as_bool())
483 .unwrap_or(false)
484 || service_tier_display(record).eq_ignore_ascii_case("priority(fast)")
485}
486
487fn request_was_retried(record: &JsonValue) -> bool {
488 if record
489 .get("observability")
490 .and_then(|observability| observability.get("retried"))
491 .and_then(|value| value.as_bool())
492 .unwrap_or(false)
493 {
494 return true;
495 }
496 if record
497 .get("observability")
498 .and_then(|observability| observability.get("attempt_count"))
499 .and_then(|value| value.as_u64())
500 .is_some_and(|attempts| attempts > 1)
501 {
502 return true;
503 }
504 if record
505 .get("retry")
506 .and_then(|retry| retry.get("route_attempts"))
507 .and_then(|attempts| attempts.as_array())
508 .is_some_and(|attempts| attempts.len() > 1)
509 {
510 return true;
511 }
512 if record
513 .get("retry")
514 .and_then(|retry| retry.get("attempts"))
515 .and_then(|attempts| attempts.as_u64())
516 .is_some_and(|attempts| attempts > 1)
517 {
518 return true;
519 }
520 record
521 .get("retry")
522 .and_then(|retry| retry.get("upstream_chain"))
523 .and_then(|attempts| attempts.as_array())
524 .is_some_and(|attempts| attempts.len() > 1)
525}
526
527fn usage_metrics(record: &JsonValue) -> Option<UsageMetrics> {
528 record
529 .get("usage")
530 .and_then(|usage| serde_json::from_value::<UsageMetrics>(usage.clone()).ok())
531}
532
533fn request_cost_display(service: &str, model: &str, usage: Option<&UsageMetrics>) -> String {
534 let model = model.trim();
535 if model.is_empty() || model == "-" {
536 return "-".to_string();
537 }
538 let cost = estimate_request_cost_from_operator_catalog_for_service(
539 Some(model),
540 usage,
541 CostAdjustments::default(),
542 service,
543 );
544 cost.display_total_with_confidence()
545}
546
547fn output_tokens_per_second(
548 usage: &UsageMetrics,
549 duration_ms: u64,
550 ttfb_ms: Option<u64>,
551) -> Option<f64> {
552 let output_tokens = usage.output_tokens.max(0);
553 if output_tokens == 0 || duration_ms == 0 {
554 return None;
555 }
556 let generation_ms = match ttfb_ms {
557 Some(ttfb) if ttfb > 0 && ttfb < duration_ms => duration_ms.saturating_sub(ttfb),
558 _ => duration_ms,
559 };
560 if generation_ms == 0 {
561 return None;
562 }
563 Some(output_tokens as f64 / (generation_ms as f64 / 1000.0))
564}
565
566fn request_model(record: &JsonValue) -> Option<String> {
567 str_field(record, "model")
568 .map(ToOwned::to_owned)
569 .or_else(|| nested_str(record, &["route_decision", "effective_model", "value"]))
570 .or_else(|| model_from_route_attempts(record))
571 .or_else(|| model_from_legacy_retry_chain(record))
572}
573
574fn model_from_route_attempts(record: &JsonValue) -> Option<String> {
575 record
576 .get("retry")
577 .and_then(|retry| retry.get("route_attempts"))
578 .and_then(|attempts| attempts.as_array())
579 .and_then(|attempts| {
580 attempts
581 .iter()
582 .rev()
583 .filter(|attempt| {
584 !attempt
585 .get("skipped")
586 .and_then(|value| value.as_bool())
587 .unwrap_or(false)
588 })
589 .find_map(|attempt| str_field(attempt, "model").map(ToOwned::to_owned))
590 })
591}
592
593fn model_from_legacy_retry_chain(record: &JsonValue) -> Option<String> {
594 record
595 .get("retry")
596 .and_then(|retry| retry.get("upstream_chain"))
597 .and_then(|chain| chain.as_array())
598 .and_then(|chain| {
599 chain
600 .iter()
601 .rev()
602 .filter_map(|entry| entry.as_str())
603 .find_map(|entry| raw_kv_field(entry, "model"))
604 })
605}
606
607fn raw_kv_field(raw: &str, key: &str) -> Option<String> {
608 let prefix = format!("{key}=");
609 raw.split_whitespace()
610 .find_map(|part| part.strip_prefix(&prefix))
611 .map(|value| value.trim().trim_matches(',').to_string())
612 .filter(|value| !value.is_empty() && value != "-")
613}
614
615fn service_tier_display(record: &JsonValue) -> String {
616 let tier = service_tier_value(record).unwrap_or_else(|| "-".to_string());
617 if tier.eq_ignore_ascii_case("priority") {
618 "priority(fast)".to_string()
619 } else {
620 tier
621 }
622}
623
624fn service_tier_value(record: &JsonValue) -> Option<String> {
625 record.get("service_tier").and_then(|tier| {
626 tier.as_str()
627 .map(ToOwned::to_owned)
628 .or_else(|| service_tier_log_value(tier))
629 })
630}
631
632fn service_tier_log_value(tier: &JsonValue) -> Option<String> {
633 ["actual", "effective", "requested"]
634 .iter()
635 .find_map(|key| str_field(tier, key).map(ToOwned::to_owned))
636}
637
638fn station_name(record: &JsonValue) -> &str {
639 str_field(record, "station_name")
640 .or_else(|| str_field(record, "config_name"))
641 .unwrap_or("-")
642}
643
644fn provider_endpoint_display(record: &JsonValue) -> String {
645 str_field(record, "provider_endpoint_key")
646 .map(ToOwned::to_owned)
647 .or_else(|| {
648 let provider = str_field(record, "provider_id")?;
649 let endpoint = str_field(record, "endpoint_id")?;
650 Some(format!("{provider}.{endpoint}"))
651 })
652 .unwrap_or_else(|| "-".to_string())
653}
654
655fn non_dash(value: &str) -> Option<&str> {
656 (value != "-").then_some(value)
657}
658
659fn str_field<'a>(record: &'a JsonValue, key: &str) -> Option<&'a str> {
660 record
661 .get(key)
662 .and_then(|value| value.as_str())
663 .map(str::trim)
664 .filter(|value| !value.is_empty())
665}
666
667fn nested_str(record: &JsonValue, path: &[&str]) -> Option<String> {
668 let mut current = record;
669 for key in path {
670 current = current.get(*key)?;
671 }
672 current
673 .as_str()
674 .map(str::trim)
675 .filter(|value| !value.is_empty())
676 .map(ToOwned::to_owned)
677}
678
679fn i64_field(record: &JsonValue, key: &str) -> Option<i64> {
680 record.get(key).and_then(|value| value.as_i64())
681}
682
683fn u64_field(record: &JsonValue, key: &str) -> Option<u64> {
684 record.get(key).and_then(|value| value.as_u64())
685}
686
687fn reasoning_tokens(usage: &UsageMetrics) -> i64 {
688 usage.reasoning_output_tokens_total().max(0)
689}
690
691fn cache_creation_tokens(usage: &UsageMetrics) -> i64 {
692 usage.cache_creation_tokens_total().max(0)
693}
694
695fn total_tokens(usage: &UsageMetrics, accounting: CacheInputAccounting) -> i64 {
696 if usage.total_tokens > 0 {
697 usage.total_tokens
698 } else {
699 let breakdown = usage.cache_usage_breakdown(accounting);
700 breakdown
701 .effective_input_tokens
702 .saturating_add(usage.output_tokens.max(0))
703 .saturating_add(breakdown.cache_read_input_tokens)
704 .saturating_add(breakdown.cache_creation_input_tokens)
705 }
706}
707
708fn format_ms(value: u64) -> String {
709 format!("{value}ms")
710}
711
712fn format_optional_ms(value: Option<u64>) -> String {
713 value.map(format_ms).unwrap_or_else(|| "-".to_string())
714}
715
716fn format_optional_speed(value: Option<f64>) -> String {
717 value
718 .map(|speed| format!("{speed:.2} tok/s"))
719 .unwrap_or_else(|| "-".to_string())
720}
721
722#[cfg(test)]
723mod tests {
724 use super::*;
725 use serde_json::json;
726
727 #[test]
728 fn display_lines_include_route_model_fast_cache_and_speed() {
729 let record = json!({
730 "timestamp_ms": 123,
731 "service": "codex",
732 "method": "POST",
733 "path": "/v1/responses",
734 "status_code": 200,
735 "duration_ms": 2000,
736 "ttfb_ms": 500,
737 "provider_id": "relay",
738 "endpoint_id": "default",
739 "provider_endpoint_key": "codex/relay/default",
740 "service_tier": { "effective": "priority" },
741 "usage": {
742 "input_tokens": 1000,
743 "output_tokens": 30,
744 "cache_read_input_tokens": 10,
745 "cache_creation_5m_input_tokens": 5,
746 "reasoning_output_tokens": 7,
747 "total_tokens": 1045
748 },
749 "retry": {
750 "route_attempts": [
751 { "decision": "completed", "model": "gpt-5" }
752 ]
753 }
754 });
755
756 let lines = format_request_log_record_lines(&record);
757
758 assert!(lines[0].contains("endpoint=codex/relay/default"));
759 assert!(lines[0].contains("station=-"));
760 assert!(lines[0].contains("model=gpt-5"));
761 assert!(lines[0].contains("tier=priority(fast)"));
762 assert!(lines[1].contains("output_speed=20.00 tok/s"));
763 assert!(lines[1].contains("cost="));
764 assert!(lines[2].contains("cache_read=10"));
765 assert!(lines[2].contains("cache_create=5"));
766 assert!(lines[2].contains("reasoning=7"));
767 }
768
769 #[test]
770 fn request_model_reads_legacy_retry_chain_model() {
771 let record = json!({
772 "retry": {
773 "upstream_chain": [
774 "main:https://relay.example/v1 (idx=0) status=200 class=- model=gpt-5.4-mini"
775 ]
776 }
777 });
778
779 assert_eq!(request_model(&record).as_deref(), Some("gpt-5.4-mini"));
780 }
781
782 #[test]
783 fn request_model_prefers_top_level_model_from_current_request_log_schema() {
784 let record = json!({
785 "model": "gpt-5",
786 "route_decision": {
787 "effective_model": { "value": "relay-gpt-5", "source": "station_mapping" }
788 },
789 "retry": {
790 "route_attempts": [
791 { "decision": "completed", "model": "legacy-gpt-5" }
792 ]
793 }
794 });
795
796 assert_eq!(request_model(&record).as_deref(), Some("gpt-5"));
797 }
798
799 #[test]
800 fn request_model_reads_route_decision_when_top_level_model_is_missing() {
801 let record = json!({
802 "route_decision": {
803 "effective_model": { "value": "relay-gpt-5", "source": "station_mapping" }
804 }
805 });
806
807 assert_eq!(request_model(&record).as_deref(), Some("relay-gpt-5"));
808 }
809
810 #[test]
811 fn usage_aggregate_summary_includes_cache_and_average_duration() {
812 let mut aggregate = RequestUsageAggregate::default();
813 aggregate.record(
814 200,
815 Some(&UsageMetrics {
816 input_tokens: 10,
817 output_tokens: 5,
818 cache_read_input_tokens: 3,
819 cache_creation_1h_input_tokens: 2,
820 reasoning_output_tokens: 4,
821 ..UsageMetrics::default()
822 }),
823 CacheInputAccounting::default(),
824 );
825 aggregate.record(400, None, CacheInputAccounting::default());
826
827 assert_eq!(
828 aggregate.summary_line("main"),
829 "main | 2 | 10 | 5 | 3 | 2 | 4 | 20 | 300"
830 );
831 }
832
833 #[test]
834 fn filters_match_route_model_fast_retry_and_status() {
835 let record = json!({
836 "path": "/v1/responses",
837 "session_id": "sid-abc",
838 "station_name": "main-station",
839 "provider_id": "relay-one",
840 "status_code": 429,
841 "service_tier": { "actual": "priority" },
842 "observability": {
843 "attempt_count": 2,
844 "retried": true
845 },
846 "retry": {
847 "route_attempts": [
848 { "decision": "failed_status", "model": "gpt-5.4-high" },
849 { "decision": "completed", "model": "gpt-5.4-high" }
850 ]
851 }
852 });
853
854 let filters = RequestLogFilters {
855 session: Some("abc".to_string()),
856 model: Some("5.4".to_string()),
857 station: Some("main".to_string()),
858 provider: Some("relay".to_string()),
859 path: Some("responses".to_string()),
860 status_min: Some(400),
861 status_max: Some(499),
862 fast: true,
863 retried: true,
864 };
865
866 assert!(filters.matches(&record));
867 }
868
869 #[test]
870 fn filters_reject_nonmatching_model() {
871 let record = json!({
872 "model": "gpt-5.4-high",
873 "status_code": 200
874 });
875 let filters = RequestLogFilters {
876 model: Some("mini".to_string()),
877 ..RequestLogFilters::default()
878 };
879
880 assert!(!filters.matches(&record));
881 }
882
883 #[test]
884 fn filters_match_compact_request_path() {
885 let compact = json!({
886 "path": "/responses/compact",
887 "status_code": 200
888 });
889 let ordinary = json!({
890 "path": "/responses",
891 "status_code": 200
892 });
893 let filters = RequestLogFilters {
894 path: Some("responses/compact".to_string()),
895 ..RequestLogFilters::default()
896 };
897
898 assert!(filters.matches(&compact));
899 assert!(!filters.matches(&ordinary));
900 }
901
902 #[test]
903 fn tail_keeps_invalid_raw_lines_but_summary_ignores_them() {
904 let lines = [
905 RequestLogLine::from_raw(
906 r#"{"station_name":"a","duration_ms":100,"usage":{"total_tokens":7}}"#,
907 ),
908 RequestLogLine::from_raw("not-json"),
909 RequestLogLine::from_raw(
910 r#"{"station_name":"b","duration_ms":200,"usage":{"total_tokens":3}}"#,
911 ),
912 ];
913
914 assert!(!lines[1].is_valid_json());
915 let rows = summarize_request_log_lines(
916 lines.iter(),
917 RequestUsageSummaryGroup::Station,
918 &RequestLogFilters::default(),
919 10,
920 );
921
922 assert_eq!(rows.len(), 2);
923 assert_eq!(rows[0].group_value, "a");
924 assert_eq!(rows[0].aggregate.total_tokens, 7);
925 assert_eq!(rows[1].group_value, "b");
926 assert_eq!(rows[1].aggregate.total_tokens, 3);
927 }
928
929 #[test]
930 fn summary_can_group_by_provider_model_or_session_with_filters() {
931 let lines = [
932 RequestLogLine::from_raw(
933 r#"{"session_id":"sid-a","station_name":"s1","provider_id":"p1","model":"gpt-5","status_code":200,"usage":{"total_tokens":7}}"#,
934 ),
935 RequestLogLine::from_raw(
936 r#"{"session_id":"sid-b","station_name":"s1","provider_id":"p1","model":"gpt-5.4","status_code":429,"usage":{"total_tokens":11}}"#,
937 ),
938 RequestLogLine::from_raw(
939 r#"{"session_id":"sid-b","station_name":"s2","provider_id":"p2","model":"gpt-5.4","status_code":200,"usage":{"total_tokens":3}}"#,
940 ),
941 ];
942
943 let provider_rows = summarize_request_log_lines(
944 lines.iter(),
945 RequestUsageSummaryGroup::Provider,
946 &RequestLogFilters::default(),
947 10,
948 );
949 assert_eq!(provider_rows[0].group_value, "p1");
950 assert_eq!(provider_rows[0].aggregate.total_tokens, 18);
951 assert_eq!(provider_rows[1].group_value, "p2");
952
953 let model_rows = summarize_request_log_lines(
954 lines.iter(),
955 RequestUsageSummaryGroup::Model,
956 &RequestLogFilters {
957 status_min: Some(400),
958 ..RequestLogFilters::default()
959 },
960 10,
961 );
962 assert_eq!(model_rows.len(), 1);
963 assert_eq!(model_rows[0].group_value, "gpt-5.4");
964 assert_eq!(model_rows[0].aggregate.total_tokens, 11);
965
966 let session_rows = summarize_request_log_lines(
967 lines.iter(),
968 RequestUsageSummaryGroup::Session,
969 &RequestLogFilters::default(),
970 10,
971 );
972 assert_eq!(session_rows[0].group_value, "sid-b");
973 assert_eq!(session_rows[0].aggregate.requests, 2);
974 }
975
976 #[test]
977 fn request_log_record_projects_to_finished_request_for_ui_reuse() {
978 let record = json!({
979 "timestamp_ms": 1234,
980 "request_id": 42,
981 "trace_id": "codex-42",
982 "service": "codex",
983 "method": "POST",
984 "path": "/v1/responses",
985 "status_code": 200,
986 "duration_ms": 1500,
987 "ttfb_ms": 500,
988 "station_name": "primary",
989 "provider_id": "relay",
990 "upstream_base_url": "https://relay.example/v1",
991 "session_id": "sid-a",
992 "reasoning_effort": "medium",
993 "service_tier": { "actual": "priority" },
994 "usage": {
995 "input_tokens": 100,
996 "output_tokens": 50,
997 "total_tokens": 150
998 },
999 "retry": {
1000 "attempts": 2,
1001 "upstream_chain": [
1002 "primary:https://relay.example/v1 (idx=0) status=429 class=rate_limit model=gpt-5.4",
1003 "primary:https://relay.example/v1 (idx=1) status=200 class=- model=gpt-5.4"
1004 ]
1005 }
1006 });
1007
1008 let request =
1009 finished_request_from_request_log_record(&record).expect("finished request projection");
1010
1011 assert_eq!(request.id, 42);
1012 assert_eq!(request.trace_id.as_deref(), Some("codex-42"));
1013 assert_eq!(request.session_id.as_deref(), Some("sid-a"));
1014 assert_eq!(request.model.as_deref(), Some("gpt-5.4"));
1015 assert_eq!(request.service_tier.as_deref(), Some("priority"));
1016 assert!(request.is_fast_mode());
1017 assert_eq!(request.attempt_count(), 2);
1018 assert_eq!(request.output_tokens_per_second(), Some(50.0));
1019 assert_eq!(request.ended_at_ms, 1234);
1020 }
1021}