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