1use std::collections::{HashMap, VecDeque};
2use std::fs::File;
3use std::io::{BufRead, BufReader};
4use std::path::{Path, PathBuf};
5
6use serde_json::Value as JsonValue;
7
8use crate::local_log_store::{LogRetention, repair_log};
9pub use crate::logging::request_log_path;
10use crate::logging::request_log_retention;
11use crate::pricing::{CostAdjustments, estimate_request_cost_from_operator_catalog_for_service};
12use crate::state::{
13 FinishedRequest, RequestObservability, RouteDecisionProvenance, SessionIdentitySource,
14};
15use crate::usage::{CacheInputAccounting, UsageMetrics};
16
17#[derive(Debug, Clone, PartialEq)]
18pub struct RequestLogLine {
19 raw: String,
20 value: Option<JsonValue>,
21}
22
23impl RequestLogLine {
24 pub fn from_raw(raw: impl Into<String>) -> Self {
25 let raw = raw.into();
26 let value = serde_json::from_str::<JsonValue>(&raw).ok();
27 Self { raw, value }
28 }
29
30 pub fn raw(&self) -> &str {
31 &self.raw
32 }
33
34 pub fn value(&self) -> Option<&JsonValue> {
35 self.value.as_ref()
36 }
37
38 pub fn is_valid_json(&self) -> bool {
39 self.value.is_some()
40 }
41
42 pub fn display_lines(&self) -> Vec<String> {
43 self.value
44 .as_ref()
45 .map(format_request_log_record_lines)
46 .unwrap_or_default()
47 }
48}
49
50#[derive(Debug, Default, Clone, PartialEq, Eq)]
51pub struct RequestLogFilters {
52 pub session: Option<String>,
53 pub model: Option<String>,
54 pub station: Option<String>,
55 pub provider: Option<String>,
56 pub path: Option<String>,
57 pub status_min: Option<u64>,
58 pub status_max: Option<u64>,
59 pub fast: bool,
60 pub retried: bool,
61}
62
63impl RequestLogFilters {
64 pub fn is_empty(&self) -> bool {
65 self.session.is_none()
66 && self.model.is_none()
67 && self.station.is_none()
68 && self.provider.is_none()
69 && self.path.is_none()
70 && self.status_min.is_none()
71 && self.status_max.is_none()
72 && !self.fast
73 && !self.retried
74 }
75
76 pub fn matches(&self, record: &JsonValue) -> bool {
77 if let Some(expected) = self.session.as_deref()
78 && !field_contains(str_field(record, "session_id"), expected)
79 {
80 return false;
81 }
82 if let Some(expected) = self.model.as_deref()
83 && !field_contains(request_model(record).as_deref(), expected)
84 {
85 return false;
86 }
87 if let Some(expected) = self.station.as_deref()
88 && !field_contains(Some(station_name(record)), expected)
89 {
90 return false;
91 }
92 if let Some(expected) = self.provider.as_deref()
93 && !field_contains(str_field(record, "provider_id"), expected)
94 {
95 return false;
96 }
97 if let Some(expected) = self.path.as_deref()
98 && !field_contains(str_field(record, "path"), expected)
99 {
100 return false;
101 }
102 if let Some(min) = self.status_min
103 && u64_field(record, "status_code").unwrap_or(0) < min
104 {
105 return false;
106 }
107 if let Some(max) = self.status_max
108 && u64_field(record, "status_code").unwrap_or(0) > max
109 {
110 return false;
111 }
112 if self.fast && !request_is_fast(record) {
113 return false;
114 }
115 if self.retried && !request_was_retried(record) {
116 return false;
117 }
118 true
119 }
120}
121
122#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
123pub struct RequestUsageAggregate {
124 pub requests: u64,
125 pub duration_ms_total: u64,
126 pub input_tokens: i64,
127 pub output_tokens: i64,
128 pub reasoning_tokens: i64,
129 pub cache_read_input_tokens: i64,
130 pub cache_creation_input_tokens: i64,
131 pub total_tokens: i64,
132}
133
134impl RequestUsageAggregate {
135 pub fn record(
136 &mut self,
137 duration_ms: u64,
138 usage: Option<&UsageMetrics>,
139 accounting: CacheInputAccounting,
140 ) {
141 self.requests = self.requests.saturating_add(1);
142 self.duration_ms_total = self.duration_ms_total.saturating_add(duration_ms);
143 let Some(usage) = usage else {
144 return;
145 };
146
147 self.input_tokens = self.input_tokens.saturating_add(usage.input_tokens.max(0));
148 self.output_tokens = self
149 .output_tokens
150 .saturating_add(usage.output_tokens.max(0));
151 self.reasoning_tokens = self
152 .reasoning_tokens
153 .saturating_add(reasoning_tokens(usage));
154 self.cache_read_input_tokens = self
155 .cache_read_input_tokens
156 .saturating_add(usage.cache_read_tokens_total());
157 self.cache_creation_input_tokens = self
158 .cache_creation_input_tokens
159 .saturating_add(cache_creation_tokens(usage));
160 self.total_tokens = self
161 .total_tokens
162 .saturating_add(total_tokens(usage, accounting));
163 }
164
165 pub fn average_duration_ms(&self) -> u64 {
166 self.duration_ms_total
167 .checked_div(self.requests)
168 .unwrap_or(0)
169 }
170
171 pub fn summary_line(&self, station_name: &str) -> String {
172 format!(
173 "{} | {} | {} | {} | {} | {} | {} | {} | {}",
174 station_name,
175 self.requests,
176 self.input_tokens,
177 self.output_tokens,
178 self.cache_read_input_tokens,
179 self.cache_creation_input_tokens,
180 self.reasoning_tokens,
181 self.total_tokens,
182 self.average_duration_ms()
183 )
184 }
185}
186
187#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default)]
188#[serde(rename_all = "snake_case")]
189pub enum RequestUsageSummaryGroup {
190 #[default]
191 Station,
192 Provider,
193 Model,
194 Session,
195}
196
197impl RequestUsageSummaryGroup {
198 pub fn column_name(self) -> &'static str {
199 match self {
200 Self::Station => "station_name",
201 Self::Provider => "provider_id",
202 Self::Model => "model",
203 Self::Session => "session_id",
204 }
205 }
206
207 fn key(self, record: &JsonValue) -> String {
208 match self {
209 Self::Station => station_name(record).to_string(),
210 Self::Provider => str_field(record, "provider_id").unwrap_or("-").to_string(),
211 Self::Model => request_model(record).unwrap_or_else(|| "-".to_string()),
212 Self::Session => str_field(record, "session_id").unwrap_or("-").to_string(),
213 }
214 }
215}
216
217#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
218pub struct RequestUsageSummaryRow {
219 pub group_value: String,
220 pub aggregate: RequestUsageAggregate,
221}
222
223#[derive(Debug, Clone, PartialEq, Eq)]
224pub struct RequestLedgerStore {
225 path: PathBuf,
226 retention: LogRetention,
227}
228
229impl Default for RequestLedgerStore {
230 fn default() -> Self {
231 Self::new(request_log_path())
232 }
233}
234
235impl RequestLedgerStore {
236 pub fn new(path: impl Into<PathBuf>) -> Self {
237 Self {
238 path: path.into(),
239 retention: request_log_retention(),
240 }
241 }
242
243 pub fn from_path(path: impl AsRef<Path>) -> Self {
244 Self::new(path.as_ref().to_path_buf())
245 }
246
247 #[cfg(test)]
248 fn with_retention(path: impl Into<PathBuf>, retention: LogRetention) -> Self {
249 Self {
250 path: path.into(),
251 retention,
252 }
253 }
254
255 pub fn path(&self) -> &Path {
256 &self.path
257 }
258
259 pub fn exists(&self) -> bool {
260 self.path.exists()
261 }
262
263 fn repair_before_read(&self) {
264 repair_log(&self.path, self.retention);
265 }
266
267 fn open_after_repair(&self) -> std::io::Result<Option<File>> {
268 self.repair_before_read();
269 match File::open(&self.path) {
270 Ok(file) => Ok(Some(file)),
271 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
272 Err(error) => Err(error),
273 }
274 }
275
276 pub fn read_lines(&self) -> std::io::Result<Vec<RequestLogLine>> {
277 let Some(file) = self.open_after_repair()? else {
278 return Ok(Vec::new());
279 };
280 let reader = BufReader::new(file);
281 Ok(reader
282 .lines()
283 .map_while(Result::ok)
284 .map(RequestLogLine::from_raw)
285 .collect())
286 }
287
288 pub fn tail_lines(&self, limit: usize) -> std::io::Result<Vec<RequestLogLine>> {
289 if limit == 0 {
290 return Ok(Vec::new());
291 }
292 let Some(file) = self.open_after_repair()? else {
293 return Ok(Vec::new());
294 };
295 let reader = BufReader::new(file);
296 let mut ring = VecDeque::with_capacity(limit);
297 for line in reader.lines().map_while(Result::ok) {
298 if ring.len() == limit {
299 ring.pop_front();
300 }
301 ring.push_back(RequestLogLine::from_raw(line));
302 }
303 Ok(ring.into_iter().collect())
304 }
305
306 pub fn tail_finished_requests(&self, limit: usize) -> std::io::Result<Vec<FinishedRequest>> {
307 let lines = self.tail_lines(limit)?;
308 let mut requests = lines
309 .iter()
310 .filter_map(|line| {
311 line.value()
312 .and_then(finished_request_from_request_log_record)
313 })
314 .collect::<Vec<_>>();
315 requests.reverse();
316 Ok(requests)
317 }
318
319 pub fn find_lines(
320 &self,
321 filters: &RequestLogFilters,
322 limit: usize,
323 ) -> std::io::Result<Vec<RequestLogLine>> {
324 if limit == 0 {
325 return Ok(Vec::new());
326 }
327 let Some(file) = self.open_after_repair()? else {
328 return Ok(Vec::new());
329 };
330 let reader = BufReader::new(file);
331 let mut ring = VecDeque::with_capacity(limit);
332 for line in reader.lines().map_while(Result::ok) {
333 let line = RequestLogLine::from_raw(line);
334 if !line.value().is_some_and(|record| filters.matches(record)) {
335 continue;
336 }
337 if ring.len() == limit {
338 ring.pop_front();
339 }
340 ring.push_back(line);
341 }
342 Ok(ring.into_iter().rev().collect())
343 }
344
345 pub fn find_finished_requests(
346 &self,
347 filters: &RequestLogFilters,
348 limit: usize,
349 ) -> std::io::Result<Vec<FinishedRequest>> {
350 let lines = self.find_lines(filters, limit)?;
351 Ok(lines
352 .iter()
353 .filter_map(|line| {
354 line.value()
355 .and_then(finished_request_from_request_log_record)
356 })
357 .collect())
358 }
359
360 pub fn summarize(
361 &self,
362 group: RequestUsageSummaryGroup,
363 filters: &RequestLogFilters,
364 limit: usize,
365 ) -> std::io::Result<Vec<RequestUsageSummaryRow>> {
366 let lines = self.read_lines()?;
367 Ok(summarize_request_log_lines(
368 lines.iter(),
369 group,
370 filters,
371 limit,
372 ))
373 }
374}
375
376pub fn read_request_log_lines(path: &Path) -> std::io::Result<Vec<RequestLogLine>> {
377 RequestLedgerStore::from_path(path).read_lines()
378}
379
380pub fn tail_request_log(path: &Path, limit: usize) -> std::io::Result<Vec<RequestLogLine>> {
381 RequestLedgerStore::from_path(path).tail_lines(limit)
382}
383
384pub fn tail_finished_requests_from_log(
385 path: &Path,
386 limit: usize,
387) -> std::io::Result<Vec<FinishedRequest>> {
388 RequestLedgerStore::from_path(path).tail_finished_requests(limit)
389}
390
391pub fn find_finished_requests_from_log(
392 path: &Path,
393 filters: &RequestLogFilters,
394 limit: usize,
395) -> std::io::Result<Vec<FinishedRequest>> {
396 RequestLedgerStore::from_path(path).find_finished_requests(filters, limit)
397}
398
399pub fn summarize_request_log(
400 path: &Path,
401 group: RequestUsageSummaryGroup,
402 filters: &RequestLogFilters,
403 limit: usize,
404) -> std::io::Result<Vec<RequestUsageSummaryRow>> {
405 RequestLedgerStore::from_path(path).summarize(group, filters, limit)
406}
407
408pub fn find_request_log(
409 path: &Path,
410 filters: &RequestLogFilters,
411 limit: usize,
412) -> std::io::Result<Vec<RequestLogLine>> {
413 RequestLedgerStore::from_path(path).find_lines(filters, limit)
414}
415
416fn summarize_request_log_lines<'a>(
417 lines: impl IntoIterator<Item = &'a RequestLogLine>,
418 group: RequestUsageSummaryGroup,
419 filters: &RequestLogFilters,
420 limit: usize,
421) -> Vec<RequestUsageSummaryRow> {
422 let mut aggregate: HashMap<String, RequestUsageAggregate> = HashMap::new();
423 for line in lines {
424 let Some(record) = line.value() else {
425 continue;
426 };
427 if !filters.matches(record) {
428 continue;
429 }
430 let group_value = group.key(record);
431 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
432 let service = str_field(record, "service").unwrap_or("-");
433 let usage = usage_metrics(record);
434 let entry = aggregate.entry(group_value).or_default();
435 entry.record(
436 duration_ms,
437 usage.as_ref(),
438 CacheInputAccounting::for_service(service),
439 );
440 }
441
442 let mut items: Vec<RequestUsageSummaryRow> = aggregate
443 .into_iter()
444 .map(|(group_value, aggregate)| RequestUsageSummaryRow {
445 group_value,
446 aggregate,
447 })
448 .collect();
449 items.sort_by(|a, b| {
450 b.aggregate
451 .total_tokens
452 .cmp(&a.aggregate.total_tokens)
453 .then_with(|| a.group_value.cmp(&b.group_value))
454 });
455 items.into_iter().take(limit).collect()
456}
457
458pub fn format_request_log_record_lines(record: &JsonValue) -> Vec<String> {
459 let ts = i64_field(record, "timestamp_ms").unwrap_or(0);
460 let service = str_field(record, "service").unwrap_or("-");
461 let method = str_field(record, "method").unwrap_or("-");
462 let path = str_field(record, "path").unwrap_or("-");
463 let status = u64_field(record, "status_code").unwrap_or(0);
464 let provider = str_field(record, "provider_id").unwrap_or("-");
465 let endpoint = provider_endpoint_display(record);
466 let station = station_name(record);
467 let model = request_model(record).unwrap_or_else(|| "-".to_string());
468 let tier = service_tier_display(record);
469
470 let mut lines = vec![format!(
471 "[{}] {} {} {} status={} endpoint={} provider={} station={} model={} tier={}",
472 ts, service, method, path, status, endpoint, provider, station, model, tier
473 )];
474
475 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
476 let ttfb_ms = u64_field(record, "ttfb_ms");
477 let usage = usage_metrics(record);
478 let finished_request = finished_request_from_request_log_record(record);
479 let observability = finished_request
480 .as_ref()
481 .map(|request| request.observability_view());
482 let effective_ttfb_ms = observability
483 .as_ref()
484 .and_then(|observability| observability.ttfb_ms)
485 .or(ttfb_ms);
486 let speed = observability
487 .as_ref()
488 .and_then(|observability| observability.output_tokens_per_second)
489 .or_else(|| {
490 usage
491 .as_ref()
492 .and_then(|usage| output_tokens_per_second(usage, duration_ms, effective_ttfb_ms))
493 });
494 let cost = request_cost_display(service, model.as_str(), usage.as_ref());
495
496 lines.push(format!(
497 " timing duration={} ttfb={} output_speed={} cost={}",
498 format_ms(duration_ms),
499 format_optional_ms(effective_ttfb_ms),
500 format_optional_speed(speed),
501 cost
502 ));
503
504 if let Some(usage) = usage.as_ref() {
505 lines.push(format!(
506 " tokens input={} output={} cache_read={} cache_create={} reasoning={} total={}",
507 usage.input_tokens.max(0),
508 usage.output_tokens.max(0),
509 usage.cache_read_tokens_total(),
510 cache_creation_tokens(usage),
511 reasoning_tokens(usage),
512 total_tokens(usage, CacheInputAccounting::for_service(service))
513 ));
514 } else {
515 lines.push(" tokens -".to_string());
516 }
517
518 lines
519}
520
521pub fn request_log_record_model(record: &JsonValue) -> Option<String> {
522 request_model(record)
523}
524
525pub fn request_log_record_station(record: &JsonValue) -> &str {
526 station_name(record)
527}
528
529pub fn request_log_record_is_fast(record: &JsonValue) -> bool {
530 request_is_fast(record)
531}
532
533pub fn request_log_record_was_retried(record: &JsonValue) -> bool {
534 request_was_retried(record)
535}
536
537pub fn finished_request_from_request_log_record(record: &JsonValue) -> Option<FinishedRequest> {
538 let timestamp_ms = u64_field(record, "timestamp_ms").unwrap_or(0);
539 let request_id = u64_field(record, "request_id").unwrap_or(timestamp_ms);
540 let status_code = u64_field(record, "status_code")
541 .and_then(|status| u16::try_from(status).ok())
542 .unwrap_or(0);
543 let duration_ms = u64_field(record, "duration_ms").unwrap_or(0);
544 let usage = usage_metrics(record);
545 let model = request_model(record);
546 let service = str_field(record, "service").unwrap_or("-");
547 let cost = estimate_request_cost_from_operator_catalog_for_service(
548 model.as_deref(),
549 usage.as_ref(),
550 CostAdjustments::default(),
551 service,
552 );
553 let retry = record
554 .get("retry")
555 .and_then(|retry| serde_json::from_value(retry.clone()).ok());
556 let route_decision = record.get("route_decision").and_then(|route_decision| {
557 serde_json::from_value::<RouteDecisionProvenance>(route_decision.clone()).ok()
558 });
559 let session_identity_source = record
560 .get("session_identity_source")
561 .and_then(|source| serde_json::from_value::<SessionIdentitySource>(source.clone()).ok());
562
563 let mut request = FinishedRequest {
564 id: request_id,
565 trace_id: str_field(record, "trace_id").map(ToOwned::to_owned),
566 session_id: str_field(record, "session_id").map(ToOwned::to_owned),
567 session_identity_source,
568 client_name: str_field(record, "client_name").map(ToOwned::to_owned),
569 client_addr: str_field(record, "client_addr").map(ToOwned::to_owned),
570 cwd: str_field(record, "cwd").map(ToOwned::to_owned),
571 model,
572 reasoning_effort: str_field(record, "reasoning_effort").map(ToOwned::to_owned),
573 service_tier: service_tier_value(record),
574 station_name: non_dash(station_name(record)).map(ToOwned::to_owned),
575 provider_id: str_field(record, "provider_id").map(ToOwned::to_owned),
576 upstream_base_url: str_field(record, "upstream_base_url").map(ToOwned::to_owned),
577 route_decision,
578 usage,
579 cost,
580 retry,
581 observability: RequestObservability::default(),
582 service: service.to_string(),
583 method: str_field(record, "method").unwrap_or("-").to_string(),
584 path: str_field(record, "path").unwrap_or("-").to_string(),
585 status_code,
586 duration_ms,
587 ttfb_ms: u64_field(record, "ttfb_ms"),
588 streaming: record
589 .get("streaming")
590 .and_then(|value| value.as_bool())
591 .unwrap_or(false),
592 ended_at_ms: timestamp_ms,
593 };
594 request.refresh_observability();
595 Some(request)
596}
597
598fn field_contains(value: Option<&str>, expected: &str) -> bool {
599 let expected = expected.trim().to_ascii_lowercase();
600 if expected.is_empty() {
601 return true;
602 }
603 value
604 .map(|value| value.to_ascii_lowercase().contains(&expected))
605 .unwrap_or(false)
606}
607
608fn request_is_fast(record: &JsonValue) -> bool {
609 record
610 .get("observability")
611 .and_then(|observability| observability.get("fast_mode"))
612 .and_then(|value| value.as_bool())
613 .unwrap_or(false)
614 || service_tier_display(record).eq_ignore_ascii_case("priority(fast)")
615}
616
617fn request_was_retried(record: &JsonValue) -> bool {
618 if record
619 .get("observability")
620 .and_then(|observability| observability.get("retried"))
621 .and_then(|value| value.as_bool())
622 .unwrap_or(false)
623 {
624 return true;
625 }
626 if record
627 .get("observability")
628 .and_then(|observability| observability.get("attempt_count"))
629 .and_then(|value| value.as_u64())
630 .is_some_and(|attempts| attempts > 1)
631 {
632 return true;
633 }
634 if record
635 .get("retry")
636 .and_then(|retry| retry.get("route_attempts"))
637 .and_then(|attempts| attempts.as_array())
638 .is_some_and(|attempts| attempts.len() > 1)
639 {
640 return true;
641 }
642 if record
643 .get("retry")
644 .and_then(|retry| retry.get("attempts"))
645 .and_then(|attempts| attempts.as_u64())
646 .is_some_and(|attempts| attempts > 1)
647 {
648 return true;
649 }
650 record
651 .get("retry")
652 .and_then(|retry| retry.get("upstream_chain"))
653 .and_then(|attempts| attempts.as_array())
654 .is_some_and(|attempts| attempts.len() > 1)
655}
656
657fn usage_metrics(record: &JsonValue) -> Option<UsageMetrics> {
658 record
659 .get("usage")
660 .and_then(|usage| serde_json::from_value::<UsageMetrics>(usage.clone()).ok())
661}
662
663fn request_cost_display(service: &str, model: &str, usage: Option<&UsageMetrics>) -> String {
664 let model = model.trim();
665 if model.is_empty() || model == "-" {
666 return "-".to_string();
667 }
668 let cost = estimate_request_cost_from_operator_catalog_for_service(
669 Some(model),
670 usage,
671 CostAdjustments::default(),
672 service,
673 );
674 cost.display_total_with_confidence()
675}
676
677fn output_tokens_per_second(
678 usage: &UsageMetrics,
679 duration_ms: u64,
680 ttfb_ms: Option<u64>,
681) -> Option<f64> {
682 const OUTPUT_RATE_SANITY_CEIL: f64 = 5_000.0;
683 let output_tokens = usage.output_tokens.max(0);
684 if output_tokens == 0 || duration_ms == 0 {
685 return None;
686 }
687 let generation_ms = match ttfb_ms {
688 Some(ttfb) if ttfb > 0 && ttfb < duration_ms => duration_ms.saturating_sub(ttfb),
689 _ => duration_ms,
690 };
691 if generation_ms == 0 {
692 return None;
693 }
694 let rate = output_tokens as f64 / (generation_ms as f64 / 1000.0);
695 if generation_ms.saturating_mul(10) < duration_ms && rate > OUTPUT_RATE_SANITY_CEIL {
696 return Some(output_tokens as f64 / (duration_ms as f64 / 1000.0));
697 }
698 Some(rate)
699}
700
701fn request_model(record: &JsonValue) -> Option<String> {
702 str_field(record, "model")
703 .map(ToOwned::to_owned)
704 .or_else(|| nested_str(record, &["route_decision", "effective_model", "value"]))
705 .or_else(|| model_from_route_attempts(record))
706 .or_else(|| model_from_legacy_retry_chain(record))
707}
708
709fn model_from_route_attempts(record: &JsonValue) -> Option<String> {
710 record
711 .get("retry")
712 .and_then(|retry| retry.get("route_attempts"))
713 .and_then(|attempts| attempts.as_array())
714 .and_then(|attempts| {
715 attempts
716 .iter()
717 .rev()
718 .filter(|attempt| {
719 !attempt
720 .get("skipped")
721 .and_then(|value| value.as_bool())
722 .unwrap_or(false)
723 })
724 .find_map(|attempt| str_field(attempt, "model").map(ToOwned::to_owned))
725 })
726}
727
728fn model_from_legacy_retry_chain(record: &JsonValue) -> Option<String> {
729 record
730 .get("retry")
731 .and_then(|retry| retry.get("upstream_chain"))
732 .and_then(|chain| chain.as_array())
733 .and_then(|chain| {
734 chain
735 .iter()
736 .rev()
737 .filter_map(|entry| entry.as_str())
738 .find_map(|entry| raw_kv_field(entry, "model"))
739 })
740}
741
742fn raw_kv_field(raw: &str, key: &str) -> Option<String> {
743 let prefix = format!("{key}=");
744 raw.split_whitespace()
745 .find_map(|part| part.strip_prefix(&prefix))
746 .map(|value| value.trim().trim_matches(',').to_string())
747 .filter(|value| !value.is_empty() && value != "-")
748}
749
750fn service_tier_display(record: &JsonValue) -> String {
751 let tier = service_tier_value(record).unwrap_or_else(|| "-".to_string());
752 if tier.eq_ignore_ascii_case("priority") {
753 "priority(fast)".to_string()
754 } else {
755 tier
756 }
757}
758
759fn service_tier_value(record: &JsonValue) -> Option<String> {
760 record.get("service_tier").and_then(|tier| {
761 tier.as_str()
762 .map(ToOwned::to_owned)
763 .or_else(|| service_tier_log_value(tier))
764 })
765}
766
767fn service_tier_log_value(tier: &JsonValue) -> Option<String> {
768 ["actual", "effective", "requested"]
769 .iter()
770 .find_map(|key| str_field(tier, key).map(ToOwned::to_owned))
771}
772
773fn station_name(record: &JsonValue) -> &str {
774 str_field(record, "station_name")
775 .or_else(|| str_field(record, "config_name"))
776 .unwrap_or("-")
777}
778
779fn provider_endpoint_display(record: &JsonValue) -> String {
780 str_field(record, "provider_endpoint_key")
781 .map(ToOwned::to_owned)
782 .or_else(|| {
783 let provider = str_field(record, "provider_id")?;
784 let endpoint = str_field(record, "endpoint_id")?;
785 Some(format!("{provider}.{endpoint}"))
786 })
787 .unwrap_or_else(|| "-".to_string())
788}
789
790fn non_dash(value: &str) -> Option<&str> {
791 (value != "-").then_some(value)
792}
793
794fn str_field<'a>(record: &'a JsonValue, key: &str) -> Option<&'a str> {
795 record
796 .get(key)
797 .and_then(|value| value.as_str())
798 .map(str::trim)
799 .filter(|value| !value.is_empty())
800}
801
802fn nested_str(record: &JsonValue, path: &[&str]) -> Option<String> {
803 let mut current = record;
804 for key in path {
805 current = current.get(*key)?;
806 }
807 current
808 .as_str()
809 .map(str::trim)
810 .filter(|value| !value.is_empty())
811 .map(ToOwned::to_owned)
812}
813
814fn i64_field(record: &JsonValue, key: &str) -> Option<i64> {
815 record.get(key).and_then(|value| value.as_i64())
816}
817
818fn u64_field(record: &JsonValue, key: &str) -> Option<u64> {
819 record.get(key).and_then(|value| value.as_u64())
820}
821
822fn reasoning_tokens(usage: &UsageMetrics) -> i64 {
823 usage.reasoning_output_tokens_total().max(0)
824}
825
826fn cache_creation_tokens(usage: &UsageMetrics) -> i64 {
827 usage.cache_creation_tokens_total().max(0)
828}
829
830fn total_tokens(usage: &UsageMetrics, accounting: CacheInputAccounting) -> i64 {
831 if usage.total_tokens > 0 {
832 usage.total_tokens
833 } else {
834 let breakdown = usage.cache_usage_breakdown(accounting);
835 breakdown
836 .effective_input_tokens
837 .saturating_add(usage.output_tokens.max(0))
838 .saturating_add(breakdown.cache_read_input_tokens)
839 .saturating_add(breakdown.cache_creation_input_tokens)
840 }
841}
842
843fn format_ms(value: u64) -> String {
844 format!("{value}ms")
845}
846
847fn format_optional_ms(value: Option<u64>) -> String {
848 value.map(format_ms).unwrap_or_else(|| "-".to_string())
849}
850
851fn format_optional_speed(value: Option<f64>) -> String {
852 value
853 .map(|speed| format!("{speed:.2} tok/s"))
854 .unwrap_or_else(|| "-".to_string())
855}
856
857#[cfg(test)]
858mod tests {
859 use super::*;
860 use serde_json::json;
861
862 #[test]
863 fn display_lines_include_route_model_fast_cache_and_speed() {
864 let record = json!({
865 "timestamp_ms": 123,
866 "service": "codex",
867 "method": "POST",
868 "path": "/v1/responses",
869 "status_code": 200,
870 "duration_ms": 2000,
871 "ttfb_ms": 500,
872 "provider_id": "relay",
873 "endpoint_id": "default",
874 "provider_endpoint_key": "codex/relay/default",
875 "service_tier": { "effective": "priority" },
876 "usage": {
877 "input_tokens": 1000,
878 "output_tokens": 30,
879 "cache_read_input_tokens": 10,
880 "cache_creation_5m_input_tokens": 5,
881 "reasoning_output_tokens": 7,
882 "total_tokens": 1045
883 },
884 "retry": {
885 "route_attempts": [
886 { "decision": "completed", "model": "gpt-5" }
887 ]
888 }
889 });
890
891 let lines = format_request_log_record_lines(&record);
892
893 assert!(lines[0].contains("endpoint=codex/relay/default"));
894 assert!(lines[0].contains("station=-"));
895 assert!(lines[0].contains("model=gpt-5"));
896 assert!(lines[0].contains("tier=priority(fast)"));
897 assert!(lines[1].contains("output_speed=20.00 tok/s"));
898 assert!(lines[1].contains("cost="));
899 assert!(lines[2].contains("cache_read=10"));
900 assert!(lines[2].contains("cache_create=5"));
901 assert!(lines[2].contains("reasoning=7"));
902 }
903
904 #[test]
905 fn display_lines_use_corrected_ttfb_for_attempt_relative_stream_logs() {
906 let record = json!({
907 "timestamp_ms": 123,
908 "service": "codex",
909 "method": "POST",
910 "path": "/v1/responses",
911 "status_code": 200,
912 "duration_ms": 10000,
913 "ttfb_ms": 1210,
914 "usage": {
915 "output_tokens": 100,
916 "total_tokens": 100
917 },
918 "retry": {
919 "attempts": 2,
920 "upstream_chain": [],
921 "route_attempts": [
922 {
923 "attempt_index": 0,
924 "decision": "failed_status",
925 "status_code": 429,
926 "upstream_headers_ms": 50,
927 "duration_ms": 500,
928 "raw": "failed"
929 },
930 {
931 "attempt_index": 1,
932 "decision": "completed",
933 "status_code": 200,
934 "upstream_headers_ms": 1200,
935 "duration_ms": 2200,
936 "raw": "completed"
937 }
938 ]
939 }
940 });
941
942 let lines = format_request_log_record_lines(&record);
943
944 assert!(lines[1].contains("ttfb=2210ms"));
945 assert!(lines[1].contains("output_speed=12.84 tok/s"));
946 }
947
948 #[test]
949 fn request_model_reads_legacy_retry_chain_model() {
950 let record = json!({
951 "retry": {
952 "upstream_chain": [
953 "main:https://relay.example/v1 (idx=0) status=200 class=- model=gpt-5.4-mini"
954 ]
955 }
956 });
957
958 assert_eq!(request_model(&record).as_deref(), Some("gpt-5.4-mini"));
959 }
960
961 #[test]
962 fn request_model_prefers_top_level_model_from_current_request_log_schema() {
963 let record = json!({
964 "model": "gpt-5",
965 "route_decision": {
966 "effective_model": { "value": "relay-gpt-5", "source": "station_mapping" }
967 },
968 "retry": {
969 "route_attempts": [
970 { "decision": "completed", "model": "legacy-gpt-5" }
971 ]
972 }
973 });
974
975 assert_eq!(request_model(&record).as_deref(), Some("gpt-5"));
976 }
977
978 #[test]
979 fn request_model_reads_route_decision_when_top_level_model_is_missing() {
980 let record = json!({
981 "route_decision": {
982 "effective_model": { "value": "relay-gpt-5", "source": "station_mapping" }
983 }
984 });
985
986 assert_eq!(request_model(&record).as_deref(), Some("relay-gpt-5"));
987 }
988
989 #[test]
990 fn usage_aggregate_summary_includes_cache_and_average_duration() {
991 let mut aggregate = RequestUsageAggregate::default();
992 aggregate.record(
993 200,
994 Some(&UsageMetrics {
995 input_tokens: 10,
996 output_tokens: 5,
997 cache_read_input_tokens: 3,
998 cache_creation_1h_input_tokens: 2,
999 reasoning_output_tokens: 4,
1000 ..UsageMetrics::default()
1001 }),
1002 CacheInputAccounting::default(),
1003 );
1004 aggregate.record(400, None, CacheInputAccounting::default());
1005
1006 assert_eq!(
1007 aggregate.summary_line("main"),
1008 "main | 2 | 10 | 5 | 3 | 2 | 4 | 20 | 300"
1009 );
1010 }
1011
1012 #[test]
1013 fn filters_match_route_model_fast_retry_and_status() {
1014 let record = json!({
1015 "path": "/v1/responses",
1016 "session_id": "sid-abc",
1017 "station_name": "main-station",
1018 "provider_id": "relay-one",
1019 "status_code": 429,
1020 "service_tier": { "actual": "priority" },
1021 "observability": {
1022 "attempt_count": 2,
1023 "retried": true
1024 },
1025 "retry": {
1026 "route_attempts": [
1027 { "decision": "failed_status", "model": "gpt-5.4-high" },
1028 { "decision": "completed", "model": "gpt-5.4-high" }
1029 ]
1030 }
1031 });
1032
1033 let filters = RequestLogFilters {
1034 session: Some("abc".to_string()),
1035 model: Some("5.4".to_string()),
1036 station: Some("main".to_string()),
1037 provider: Some("relay".to_string()),
1038 path: Some("responses".to_string()),
1039 status_min: Some(400),
1040 status_max: Some(499),
1041 fast: true,
1042 retried: true,
1043 };
1044
1045 assert!(filters.matches(&record));
1046 }
1047
1048 #[test]
1049 fn filters_reject_nonmatching_model() {
1050 let record = json!({
1051 "model": "gpt-5.4-high",
1052 "status_code": 200
1053 });
1054 let filters = RequestLogFilters {
1055 model: Some("mini".to_string()),
1056 ..RequestLogFilters::default()
1057 };
1058
1059 assert!(!filters.matches(&record));
1060 }
1061
1062 #[test]
1063 fn filters_match_compact_request_path() {
1064 let compact = json!({
1065 "path": "/responses/compact",
1066 "status_code": 200
1067 });
1068 let ordinary = json!({
1069 "path": "/responses",
1070 "status_code": 200
1071 });
1072 let filters = RequestLogFilters {
1073 path: Some("responses/compact".to_string()),
1074 ..RequestLogFilters::default()
1075 };
1076
1077 assert!(filters.matches(&compact));
1078 assert!(!filters.matches(&ordinary));
1079 }
1080
1081 #[test]
1082 fn tail_keeps_invalid_raw_lines_but_summary_ignores_them() {
1083 let lines = [
1084 RequestLogLine::from_raw(
1085 r#"{"station_name":"a","duration_ms":100,"usage":{"total_tokens":7}}"#,
1086 ),
1087 RequestLogLine::from_raw("not-json"),
1088 RequestLogLine::from_raw(
1089 r#"{"station_name":"b","duration_ms":200,"usage":{"total_tokens":3}}"#,
1090 ),
1091 ];
1092
1093 assert!(!lines[1].is_valid_json());
1094 let rows = summarize_request_log_lines(
1095 lines.iter(),
1096 RequestUsageSummaryGroup::Station,
1097 &RequestLogFilters::default(),
1098 10,
1099 );
1100
1101 assert_eq!(rows.len(), 2);
1102 assert_eq!(rows[0].group_value, "a");
1103 assert_eq!(rows[0].aggregate.total_tokens, 7);
1104 assert_eq!(rows[1].group_value, "b");
1105 assert_eq!(rows[1].aggregate.total_tokens, 3);
1106 }
1107
1108 fn temp_request_log_path(test_name: &str) -> PathBuf {
1109 let path = std::env::temp_dir().join(format!(
1110 "codex-helper-request-ledger-{test_name}-{}-{}.jsonl",
1111 std::process::id(),
1112 crate::logging::now_ms()
1113 ));
1114 let _ = std::fs::remove_file(&path);
1115 path
1116 }
1117
1118 fn write_request_log_lines(path: &Path, lines: &[&str]) {
1119 std::fs::write(path, lines.join("\n")).expect("write request log lines");
1120 }
1121
1122 #[test]
1123 fn request_ledger_store_tails_recent_lines_without_losing_raw_invalid_entries() {
1124 let path = temp_request_log_path("tail-lines");
1125 write_request_log_lines(
1126 &path,
1127 &[
1128 r#"{"path":"/one","status_code":200}"#,
1129 "not-json",
1130 r#"{"path":"/three","status_code":201}"#,
1131 ],
1132 );
1133
1134 let store = RequestLedgerStore::from_path(&path);
1135 let lines = store.tail_lines(2).expect("tail request log");
1136
1137 assert_eq!(lines.len(), 2);
1138 assert_eq!(lines[0].raw(), "not-json");
1139 assert!(!lines[0].is_valid_json());
1140 assert_eq!(
1141 lines[1]
1142 .value()
1143 .and_then(|record| str_field(record, "path")),
1144 Some("/three")
1145 );
1146 let _ = std::fs::remove_file(path);
1147 }
1148
1149 #[test]
1150 fn request_ledger_store_repairs_oversized_active_log_before_reading() {
1151 let path = temp_request_log_path("read-repair");
1152 std::fs::write(&path, vec![b'x'; 32]).expect("seed oversized request log");
1153 let store = RequestLedgerStore::with_retention(&path, LogRetention::new(16, 1));
1154
1155 let lines = store.tail_lines(10).expect("tail repaired request log");
1156
1157 assert!(lines.is_empty());
1158 assert!(
1159 !path.exists(),
1160 "oversized active request log should be rotated away before reading"
1161 );
1162 assert!(
1163 crate::local_log_store::collect_rotated_logs(&path).is_empty(),
1164 "oversized rotated request log should be pruned by retention budget"
1165 );
1166 let _ = std::fs::remove_file(path);
1167 }
1168
1169 #[test]
1170 fn request_ledger_store_find_lines_returns_newest_matching_records() {
1171 let path = temp_request_log_path("find-lines");
1172 write_request_log_lines(
1173 &path,
1174 &[
1175 r#"{"path":"/v1/responses","provider_id":"a","status_code":200}"#,
1176 r#"{"path":"/v1/responses","provider_id":"b","status_code":429}"#,
1177 r#"{"path":"/v1/models","provider_id":"b","status_code":200}"#,
1178 r#"{"path":"/v1/responses","provider_id":"c","status_code":503}"#,
1179 ],
1180 );
1181
1182 let store = RequestLedgerStore::from_path(&path);
1183 let lines = store
1184 .find_lines(
1185 &RequestLogFilters {
1186 path: Some("responses".to_string()),
1187 status_min: Some(400),
1188 ..RequestLogFilters::default()
1189 },
1190 1,
1191 )
1192 .expect("find request log");
1193
1194 assert_eq!(lines.len(), 1);
1195 assert_eq!(
1196 lines[0]
1197 .value()
1198 .and_then(|record| str_field(record, "provider_id")),
1199 Some("c")
1200 );
1201 let _ = std::fs::remove_file(path);
1202 }
1203
1204 #[test]
1205 fn summary_can_group_by_provider_model_or_session_with_filters() {
1206 let lines = [
1207 RequestLogLine::from_raw(
1208 r#"{"session_id":"sid-a","station_name":"s1","provider_id":"p1","model":"gpt-5","status_code":200,"usage":{"total_tokens":7}}"#,
1209 ),
1210 RequestLogLine::from_raw(
1211 r#"{"session_id":"sid-b","station_name":"s1","provider_id":"p1","model":"gpt-5.4","status_code":429,"usage":{"total_tokens":11}}"#,
1212 ),
1213 RequestLogLine::from_raw(
1214 r#"{"session_id":"sid-b","station_name":"s2","provider_id":"p2","model":"gpt-5.4","status_code":200,"usage":{"total_tokens":3}}"#,
1215 ),
1216 ];
1217
1218 let provider_rows = summarize_request_log_lines(
1219 lines.iter(),
1220 RequestUsageSummaryGroup::Provider,
1221 &RequestLogFilters::default(),
1222 10,
1223 );
1224 assert_eq!(provider_rows[0].group_value, "p1");
1225 assert_eq!(provider_rows[0].aggregate.total_tokens, 18);
1226 assert_eq!(provider_rows[1].group_value, "p2");
1227
1228 let model_rows = summarize_request_log_lines(
1229 lines.iter(),
1230 RequestUsageSummaryGroup::Model,
1231 &RequestLogFilters {
1232 status_min: Some(400),
1233 ..RequestLogFilters::default()
1234 },
1235 10,
1236 );
1237 assert_eq!(model_rows.len(), 1);
1238 assert_eq!(model_rows[0].group_value, "gpt-5.4");
1239 assert_eq!(model_rows[0].aggregate.total_tokens, 11);
1240
1241 let session_rows = summarize_request_log_lines(
1242 lines.iter(),
1243 RequestUsageSummaryGroup::Session,
1244 &RequestLogFilters::default(),
1245 10,
1246 );
1247 assert_eq!(session_rows[0].group_value, "sid-b");
1248 assert_eq!(session_rows[0].aggregate.requests, 2);
1249 }
1250
1251 #[test]
1252 fn request_log_record_projects_to_finished_request_for_ui_reuse() {
1253 let record = json!({
1254 "timestamp_ms": 1234,
1255 "request_id": 42,
1256 "trace_id": "codex-42",
1257 "service": "codex",
1258 "method": "POST",
1259 "path": "/v1/responses",
1260 "status_code": 200,
1261 "duration_ms": 1500,
1262 "ttfb_ms": 500,
1263 "station_name": "primary",
1264 "provider_id": "relay",
1265 "upstream_base_url": "https://relay.example/v1",
1266 "session_id": "sid-a",
1267 "reasoning_effort": "medium",
1268 "service_tier": { "actual": "priority" },
1269 "usage": {
1270 "input_tokens": 100,
1271 "output_tokens": 50,
1272 "total_tokens": 150
1273 },
1274 "retry": {
1275 "attempts": 2,
1276 "upstream_chain": [
1277 "primary:https://relay.example/v1 (idx=0) status=429 class=rate_limit model=gpt-5.4",
1278 "primary:https://relay.example/v1 (idx=1) status=200 class=- model=gpt-5.4"
1279 ]
1280 }
1281 });
1282
1283 let request =
1284 finished_request_from_request_log_record(&record).expect("finished request projection");
1285
1286 assert_eq!(request.id, 42);
1287 assert_eq!(request.trace_id.as_deref(), Some("codex-42"));
1288 assert_eq!(request.session_id.as_deref(), Some("sid-a"));
1289 assert_eq!(request.model.as_deref(), Some("gpt-5.4"));
1290 assert_eq!(request.service_tier.as_deref(), Some("priority"));
1291 assert!(request.is_fast_mode());
1292 assert_eq!(request.attempt_count(), 2);
1293 assert_eq!(request.output_tokens_per_second(), Some(50.0));
1294 assert_eq!(request.ended_at_ms, 1234);
1295 }
1296}