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