diff --git a/rust/src/core/jsonl_scanner.rs b/rust/src/core/jsonl_scanner.rs index 883a31d89f..13e78bcfe9 100755 --- a/rust/src/core/jsonl_scanner.rs +++ b/rust/src/core/jsonl_scanner.rs @@ -373,6 +373,8 @@ pub(crate) struct CodexSessionMetadata { pub lineage: CodexSessionLineage, pub fork_timestamp: Option, pub history_base_thread_id: Option, + pub is_subagent: bool, + pub subagent_history_start_ordinal: Option, } /// Running totals for Codex token counting @@ -396,6 +398,10 @@ pub struct CodexForkAccountingState { pub inherited_totals: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub remaining_inherited_totals: Option, + /// True when the child log itself supplied enough copied-prefix history to + /// establish the inherited baseline without consulting a parent cache row. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub locally_resolved: bool, } /// Snapshot of the last validated cost report, persisted so spend surfaces keep @@ -456,6 +462,7 @@ pub struct CodexParseResult { pub fork_baseline: Option, /// Remaining inherited counters used when a fork emits last-only rows. pub remaining_inherited_totals: Option, + pub fork_baseline_locally_resolved: bool, } /// A billable Codex token-count delta. diff --git a/rust/src/core/jsonl_scanner/codex.rs b/rust/src/core/jsonl_scanner/codex.rs index 30f159c0ca..108d4284d7 100644 --- a/rust/src/core/jsonl_scanner/codex.rs +++ b/rust/src/core/jsonl_scanner/codex.rs @@ -8,13 +8,14 @@ use helpers::{ BoundedJsonlLine, CODEX_JSONL_MAX_LINE_BYTES, nonempty_json_string, parse_rfc3339_timestamp, read_bounded_jsonl_line, read_bounded_jsonl_line_until, session_meta_field, }; -use parser::CodexParserState; +use parser::{CodexParseMode, CodexParserState}; /// Persisted Codex cache schema version. Version 0 predates 64-bit totals; /// version 1 can retain a terminal pause after treating a paginated v2 /// subagent's independent counters as an inherited fork. Version 3 adds -/// persisted paginated-fork accounting state. Rebuild older artifacts. -pub(crate) const CODEX_CACHE_SCHEMA_VERSION: u32 = 3; +/// persisted paginated-fork accounting state. Version 4 reparses copied-prefix +/// subagents with locally inferred component baselines. Rebuild older artifacts. +pub(crate) const CODEX_CACHE_SCHEMA_VERSION: u32 = 4; /// Whether a persisted Codex cache artifact matches the current schema. /// A mismatched artifact (e.g. a pre-64-bit cache from an older release) is @@ -186,6 +187,12 @@ impl JsonlScanner { .pointer("/source/subagent/thread_spawn") .is_some_and(Value::is_object)) }); + let is_subagent = payload.is_some_and(|value| { + value.get("thread_source").and_then(Value::as_str) == Some("subagent") + || value + .pointer("/source/subagent/thread_spawn") + .is_some_and(Value::is_object) + }); let history_base_thread_id = payload .and_then(|value| value.get("history_base")) .filter(|value| value.is_object()) @@ -220,6 +227,10 @@ impl JsonlScanner { payload.and_then(|value| nonempty_json_string(value.get("timestamp"))) }), history_base_thread_id, + is_subagent, + subagent_history_start_ordinal: payload + .and_then(|value| value.get("subagent_history_start_ordinal")) + .and_then(Value::as_i64), }); } @@ -353,17 +364,16 @@ impl JsonlScanner { Self::parse_codex_file_with_state_bounded_internal( file_path, range, - start_offset, - initial_model, - initial_totals, - previous_token_timestamp, - token_timestamps_monotonic, cancel, - false, - false, - None, None, max_bytes_to_read, + CodexParseMode::Standard { + start_offset, + initial_model, + initial_totals, + previous_token_timestamp, + token_timestamps_monotonic, + }, ) } @@ -389,17 +399,16 @@ impl JsonlScanner { Self::parse_codex_file_with_state_bounded_internal( file_path, range, - start_offset, - initial_model, - initial_totals, - previous_token_timestamp, - token_timestamps_monotonic, cancel, - false, - false, - None, scan_target_size, max_bytes_to_read, + CodexParseMode::Standard { + start_offset, + initial_model, + initial_totals, + previous_token_timestamp, + token_timestamps_monotonic, + }, ) } @@ -420,17 +429,14 @@ impl JsonlScanner { Self::parse_codex_file_with_state_bounded_internal( file_path, range, - 0, - None, - Some(initial_totals), - None, - None, cancel, - true, - false, - None, None, max_bytes_to_read, + CodexParseMode::ParentBaseline { + baseline: initial_totals, + paginated_continuation: false, + remaining_inherited_totals: None, + }, ) } @@ -459,6 +465,26 @@ impl JsonlScanner { ) } + pub(crate) fn parse_codex_file_with_inferred_fork_baseline( + file_path: &Path, + range: &CostUsageDayRange, + subagent_history_start_ordinal: Option, + cancel: Option<&AtomicBool>, + scan_target_size: Option, + max_bytes_to_read: Option, + ) -> std::io::Result { + Self::parse_codex_file_with_state_bounded_internal( + file_path, + range, + cancel, + scan_target_size, + max_bytes_to_read, + CodexParseMode::InferSubagent { + start_ordinal: subagent_history_start_ordinal, + }, + ) + } + /// Fork equivalent with persisted paginated-continuation accounting. #[allow( clippy::too_many_arguments, @@ -477,38 +503,24 @@ impl JsonlScanner { Self::parse_codex_file_with_state_bounded_internal( file_path, range, - 0, - None, - Some(initial_totals), - None, - None, cancel, - true, - paginated_continuation, - remaining_inherited_totals, scan_target_size, max_bytes_to_read, + CodexParseMode::ParentBaseline { + baseline: initial_totals, + paginated_continuation, + remaining_inherited_totals, + }, ) } - #[allow( - clippy::too_many_arguments, - reason = "resume state mirrors the persisted parser cache" - )] fn parse_codex_file_with_state_bounded_internal( file_path: &Path, range: &CostUsageDayRange, - start_offset: i64, - initial_model: Option, - initial_totals: Option, - previous_token_timestamp: Option, - token_timestamps_monotonic: Option, cancel: Option<&AtomicBool>, - fork_baseline_mode: bool, - paginated_continuation: bool, - remaining_inherited_totals: Option, scan_target_size: Option, max_bytes_to_read: Option, + mode: CodexParseMode, ) -> std::io::Result { let file = File::open(file_path)?; // Session JSONL files are bounded by the cache budget; sizes fit i64. @@ -518,7 +530,7 @@ impl JsonlScanner { )] let file_size = file.metadata()?.len() as i64; - let safe_start_offset = start_offset.clamp(0, file_size); + let safe_start_offset = mode.start_offset().clamp(0, file_size); let requested_target_size = scan_target_size .unwrap_or(file_size) .max(safe_start_offset) @@ -529,15 +541,7 @@ impl JsonlScanner { reader.seek(SeekFrom::Start(safe_start_offset as u64))?; } - let mut parser = CodexParserState::with_timestamp_state_and_fork_options( - initial_model, - initial_totals, - previous_token_timestamp, - token_timestamps_monotonic, - fork_baseline_mode, - paginated_continuation, - remaining_inherited_totals, - ); + let mut parser = CodexParserState::from_mode(mode); let mut parsed_bytes = safe_start_offset; let mut committed_bytes = safe_start_offset; let mut cancelled = false; @@ -626,6 +630,7 @@ impl JsonlScanner { }; let is_complete = !cancelled && !budget_exhausted && parsed_bytes >= effective_target_size; let bytes_read = parsed_bytes.saturating_sub(safe_start_offset).max(0); + let fork_baseline_locally_resolved = parser.fork_baseline_locally_resolved(); Ok(CodexParseResult { records: parser.records, parsed_bytes, @@ -644,6 +649,7 @@ impl JsonlScanner { fork_baseline_ambiguous: parser.fork_baseline_ambiguous, fork_baseline: parser.fork_baseline, remaining_inherited_totals: parser.remaining_inherited_totals, + fork_baseline_locally_resolved, }) } diff --git a/rust/src/core/jsonl_scanner/codex/parser.rs b/rust/src/core/jsonl_scanner/codex/parser.rs index 2ec70f9cf1..bdce9c68d8 100644 --- a/rust/src/core/jsonl_scanner/codex/parser.rs +++ b/rust/src/core/jsonl_scanner/codex/parser.rs @@ -22,62 +22,255 @@ pub(super) struct CodexParserState { paginated_continuation: bool, paginated_baseline_checked: bool, pub(super) fork_baseline_ambiguous: bool, + fork_baseline_inference: Option, } -impl CodexParserState { - pub(super) fn new(initial_model: Option, initial_totals: Option) -> Self { - Self::with_timestamp_state(initial_model, initial_totals, None, None) - } - - fn with_timestamp_state( +pub(super) enum CodexParseMode { + Standard { + start_offset: i64, initial_model: Option, initial_totals: Option, previous_token_timestamp: Option, token_timestamps_monotonic: Option, - ) -> Self { - Self::with_timestamp_state_and_fork_mode( + }, + ParentBaseline { + baseline: CodexTotals, + paginated_continuation: bool, + remaining_inherited_totals: Option, + }, + InferSubagent { + start_ordinal: Option, + }, +} + +impl CodexParseMode { + pub(super) fn start_offset(&self) -> i64 { + match self { + Self::Standard { start_offset, .. } => *start_offset, + Self::ParentBaseline { .. } | Self::InferSubagent { .. } => 0, + } + } +} + +#[derive(Debug)] +struct ForkBaselineInference { + explicit_start_ordinal: Option, + baseline: Option, + boundary_open: bool, + inherited_opening: bool, + missing_explicit_ordinal: bool, + locally_confirmed: bool, + resolved: bool, +} + +enum ForkBaselineDecision { + SkipCopiedPrefix, + ProcessWithBaseline(CodexTotals), +} + +impl ForkBaselineInference { + fn new(explicit_start_ordinal: Option) -> Self { + Self { + explicit_start_ordinal, + baseline: explicit_start_ordinal.map(|_| CodexTotals { + input: 0, + cached: 0, + output: 0, + reasoning: None, + }), + boundary_open: false, + inherited_opening: false, + missing_explicit_ordinal: false, + locally_confirmed: false, + resolved: false, + } + } + + fn confirm_local_resolution(&mut self) { + if !self.missing_explicit_ordinal { + self.locally_confirmed = true; + } + } + + fn mark_missing_explicit_ordinal(&mut self) { + self.missing_explicit_ordinal = true; + self.locally_confirmed = false; + } + + fn observe_non_token(&mut self, obj: &Value) { + if obj.get("type").and_then(Value::as_str) == Some("turn_context") && self.inherited_opening + { + self.boundary_open = true; + } + } + + fn observe_token(&mut self, obj: &Value) -> ForkBaselineDecision { + let Some(payload) = token_count_payload(obj) else { + return ForkBaselineDecision::SkipCopiedPrefix; + }; + let Some(info) = payload.get("info") else { + return ForkBaselineDecision::SkipCopiedPrefix; + }; + let Some(total_usage) = info.get("total_token_usage") else { + return ForkBaselineDecision::SkipCopiedPrefix; + }; + let Some(last_usage) = info.get("last_token_usage") else { + return ForkBaselineDecision::SkipCopiedPrefix; + }; + let total = read_token_totals(total_usage); + let last = read_token_totals(last_usage); + let ordinal = obj.get("ordinal").and_then(Value::as_i64); + + if self.explicit_start_ordinal.is_some() && ordinal.is_none() { + self.mark_missing_explicit_ordinal(); + if !self.boundary_open { + self.baseline = Some(total); + } + return ForkBaselineDecision::SkipCopiedPrefix; + } + + if let Some(start) = self.explicit_start_ordinal + && !self.boundary_open + { + let ordinal = ordinal.expect("missing explicit ordinals return above"); + if ordinal < start { + self.baseline = Some(total); + return ForkBaselineDecision::SkipCopiedPrefix; + } + self.boundary_open = true; + } else if self.baseline.is_none() { + if totals_contain_usage(&total) && !totals_contain_usage(&last) { + self.baseline = Some(total); + self.inherited_opening = true; + self.confirm_local_resolution(); + } + return ForkBaselineDecision::SkipCopiedPrefix; + } else if !self.boundary_open { + let changed = self + .baseline + .as_ref() + .is_some_and(|baseline| baseline != &total); + if self.inherited_opening && changed && totals_contain_usage(&last) { + self.boundary_open = true; + } else { + return ForkBaselineDecision::SkipCopiedPrefix; + } + } + + let baseline = self.baseline.clone().unwrap_or(CodexTotals { + input: 0, + cached: 0, + output: 0, + reasoning: None, + }); + if total == baseline { + return ForkBaselineDecision::SkipCopiedPrefix; + } + let copied_snapshot = + totals_contain_usage(&baseline) && total == last && totals_at_least(&total, &baseline); + if copied_snapshot { + self.baseline = Some(total); + self.confirm_local_resolution(); + return ForkBaselineDecision::SkipCopiedPrefix; + } + + let owned_baseline = totals_delta(&last, &total); + self.baseline = Some(owned_baseline.clone()); + self.confirm_local_resolution(); + self.resolved = true; + ForkBaselineDecision::ProcessWithBaseline(owned_baseline) + } +} + +fn totals_contain_usage(totals: &CodexTotals) -> bool { + totals.input > 0 || totals.cached > 0 || totals.output > 0 +} + +fn totals_at_least(total: &CodexTotals, baseline: &CodexTotals) -> bool { + total.input >= baseline.input + && total.cached >= baseline.cached + && total.output >= baseline.output +} + +fn totals_delta(last: &CodexTotals, total: &CodexTotals) -> CodexTotals { + CodexTotals { + input: total.input.saturating_sub(last.input).max(0), + cached: total.cached.saturating_sub(last.cached).max(0), + output: total.output.saturating_sub(last.output).max(0), + reasoning: subtract_optional(total.reasoning, last.reasoning), + } +} + +impl CodexParserState { + pub(super) fn new(initial_model: Option, initial_totals: Option) -> Self { + Self::from_mode(CodexParseMode::Standard { + start_offset: 0, initial_model, initial_totals, - previous_token_timestamp, - token_timestamps_monotonic, - false, - ) + previous_token_timestamp: None, + token_timestamps_monotonic: None, + }) } - pub(super) fn with_timestamp_state_and_fork_mode( - initial_model: Option, - initial_totals: Option, - previous_token_timestamp: Option, - token_timestamps_monotonic: Option, - fork_baseline_mode: bool, - ) -> Self { - Self::with_timestamp_state_and_fork_options( + pub(super) fn from_mode(mode: CodexParseMode) -> Self { + let ( initial_model, initial_totals, previous_token_timestamp, token_timestamps_monotonic, - fork_baseline_mode, - false, - None, - ) - } - - pub(super) fn with_timestamp_state_and_fork_options( - initial_model: Option, - initial_totals: Option, - previous_token_timestamp: Option, - token_timestamps_monotonic: Option, - fork_baseline_mode: bool, - paginated_continuation: bool, - remaining_inherited_totals: Option, - ) -> Self { + fork_baseline, + paginated_continuation, + remaining_inherited_totals, + fork_baseline_inference, + ) = match mode { + CodexParseMode::Standard { + initial_model, + initial_totals, + previous_token_timestamp, + token_timestamps_monotonic, + .. + } => ( + initial_model, + initial_totals, + previous_token_timestamp, + token_timestamps_monotonic, + None, + false, + None, + None, + ), + CodexParseMode::ParentBaseline { + baseline, + paginated_continuation, + remaining_inherited_totals, + } => { + let remaining_inherited_totals = + remaining_inherited_totals.or_else(|| Some(baseline.clone())); + ( + None, + Some(baseline.clone()), + None, + None, + Some(baseline), + paginated_continuation, + remaining_inherited_totals, + None, + ) + } + CodexParseMode::InferSubagent { start_ordinal } => ( + None, + None, + None, + None, + None, + false, + None, + Some(ForkBaselineInference::new(start_ordinal)), + ), + }; let previous_token_timestamp_parsed = previous_token_timestamp .as_deref() .and_then(parse_rfc3339_timestamp); - let fork_baseline = fork_baseline_mode.then(|| initial_totals.clone()).flatten(); - let remaining_inherited_totals = fork_baseline - .as_ref() - .and_then(|baseline| remaining_inherited_totals.or_else(|| Some(baseline.clone()))); Self { current_model: initial_model, previous_totals: initial_totals.clone(), @@ -95,9 +288,16 @@ impl CodexParserState { paginated_continuation, paginated_baseline_checked: false, fork_baseline_ambiguous: false, + fork_baseline_inference, } } + pub(super) fn fork_baseline_locally_resolved(&self) -> bool { + self.fork_baseline_inference + .as_ref() + .is_some_and(|inference| inference.locally_confirmed) + } + pub(super) fn process_line(&mut self, line: &str, range: &CostUsageDayRange) { self.process_line_with_source_offset(line, range, 0); } @@ -108,7 +308,61 @@ impl CodexParserState { range: &CostUsageDayRange, source_end_offset: i64, ) { + if self + .fork_baseline_inference + .as_ref() + .is_some_and(|inference| !inference.resolved) + { + let Ok(obj) = serde_json::from_str::(line) else { + return; + }; + if token_count_payload(&obj).is_some() { + let decision = self + .fork_baseline_inference + .as_mut() + .expect("inference exists") + .observe_token(&obj); + let ForkBaselineDecision::ProcessWithBaseline(baseline) = decision else { + return; + }; + self.fork_baseline = Some(baseline.clone()); + self.remaining_inherited_totals = Some(baseline.clone()); + self.previous_totals = Some(baseline.clone()); + self.totals_watermark = Some(baseline); + } else { + self.fork_baseline_inference + .as_mut() + .expect("inference exists") + .observe_non_token(&obj); + if obj.get("type").and_then(Value::as_str) == Some("turn_context") { + self.update_current_model(&obj); + } + return; + } + } + let event_candidate = is_candidate_codex_line(line); + if event_candidate + && self + .fork_baseline_inference + .as_ref() + .is_some_and(|inference| { + inference.resolved && inference.explicit_start_ordinal.is_some() + }) + { + let Ok(obj) = serde_json::from_str::(line) else { + return; + }; + if token_count_payload(&obj).is_some() + && obj.get("ordinal").and_then(Value::as_i64).is_none() + { + self.fork_baseline_inference + .as_mut() + .expect("resolved inference exists") + .mark_missing_explicit_ordinal(); + return; + } + } let bare_candidate = !event_candidate && line.contains("\"usage\""); if !event_candidate && !bare_candidate { return; diff --git a/rust/src/core/jsonl_scanner/tests.rs b/rust/src/core/jsonl_scanner/tests.rs index bcb1129cb1..1aebd89e20 100644 --- a/rust/src/core/jsonl_scanner/tests.rs +++ b/rust/src/core/jsonl_scanner/tests.rs @@ -77,13 +77,11 @@ fn fork_baseline_subtracts_known_reasoning_without_affecting_core_tokens() { output: 10, reasoning: Some(4), }; - let mut state = CodexParserState::with_timestamp_state_and_fork_mode( - None, - Some(baseline), - None, - None, - true, - ); + let mut state = CodexParserState::from_mode(CodexParseMode::ParentBaseline { + baseline, + paginated_continuation: false, + remaining_inherited_totals: None, + }); assert_eq!( state.apply_totals_delta(CodexTotals { input: 20, @@ -100,13 +98,11 @@ fn fork_baseline_subtracts_known_reasoning_without_affecting_core_tokens() { output: 10, reasoning: None, }; - let mut state = CodexParserState::with_timestamp_state_and_fork_mode( - None, - Some(baseline_without_reasoning), - None, - None, - true, - ); + let mut state = CodexParserState::from_mode(CodexParseMode::ParentBaseline { + baseline: baseline_without_reasoning, + paginated_continuation: false, + remaining_inherited_totals: None, + }); assert_eq!( state.apply_totals_delta(CodexTotals { input: 20, @@ -119,6 +115,109 @@ fn fork_baseline_subtracts_known_reasoning_without_affecting_core_tokens() { assert!(!state.fork_baseline_ambiguous); } +#[test] +fn inferred_fork_waits_for_present_explicit_start_ordinal() { + let range = CostUsageDayRange::new( + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + ); + let mut state = CodexParserState::from_mode(CodexParseMode::InferSubagent { + start_ordinal: Some(10), + }); + + state.process_line( + r#"{"timestamp":"2026-09-22T10:00:00Z","type":"event_msg","payload":{"type":"token_count","info":{"model":"gpt-5.6-sol","total_token_usage":{"input_tokens":100,"cached_input_tokens":20,"output_tokens":10},"last_token_usage":{"input_tokens":0,"cached_input_tokens":0,"output_tokens":0}}}}"#, + &range, + ); + + assert!(state.records.is_empty()); + assert!(state.fork_baseline.is_none()); + assert!(!state.fork_baseline_locally_resolved()); + + state.process_line( + r#"{"ordinal":10,"timestamp":"2026-09-22T10:00:01Z","type":"event_msg","payload":{"type":"token_count","info":{"model":"gpt-5.6-sol","total_token_usage":{"input_tokens":110,"cached_input_tokens":22,"output_tokens":11},"last_token_usage":{"input_tokens":10,"cached_input_tokens":2,"output_tokens":1}}}}"#, + &range, + ); + + assert_eq!(state.records.len(), 1); + assert_eq!(state.records[0].0.input, 10); + assert_eq!(state.records[0].0.cached, 2); + assert_eq!(state.records[0].0.output, 1); + assert!(!state.fork_baseline_locally_resolved()); +} + +#[test] +fn inferred_fork_keeps_missing_ordinal_unresolved_after_boundary_opens() { + let range = CostUsageDayRange::new( + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + ); + let mut state = CodexParserState::from_mode(CodexParseMode::InferSubagent { + start_ordinal: Some(10), + }); + let token_line = |ordinal: Option, total: i64, last: i64| { + let mut value = serde_json::json!({ + "timestamp": "2026-09-22T10:00:00Z", + "type": "event_msg", + "payload": {"type": "token_count", "info": { + "model": "gpt-5.6-sol", + "total_token_usage": {"input_tokens": total, "cached_input_tokens": 0, "output_tokens": 0}, + "last_token_usage": {"input_tokens": last, "cached_input_tokens": 0, "output_tokens": 0} + }} + }); + if let Some(ordinal) = ordinal { + value["ordinal"] = serde_json::json!(ordinal); + } + value.to_string() + }; + + state.process_line(&token_line(Some(9), 100, 0), &range); + state.process_line(&token_line(Some(10), 100, 0), &range); + state.process_line(&token_line(Some(11), 110, 110), &range); + assert!(state.fork_baseline_locally_resolved()); + state.process_line(&token_line(None, 120, 10), &range); + assert!(!state.fork_baseline_locally_resolved()); + state.process_line(&token_line(Some(12), 130, 10), &range); + + assert_eq!(state.records.len(), 1); + assert_eq!(state.records[0].0.input, 10); + assert!(!state.fork_baseline_locally_resolved()); +} + +#[test] +fn inferred_fork_keeps_missing_ordinal_unresolved_after_local_resolution() { + let range = CostUsageDayRange::new( + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + NaiveDate::from_ymd_opt(2026, 9, 22).unwrap(), + ); + let mut state = CodexParserState::from_mode(CodexParseMode::InferSubagent { + start_ordinal: Some(10), + }); + let token_line = |ordinal: Option, total: i64, last: i64| { + let mut value = serde_json::json!({ + "timestamp": "2026-09-22T10:00:00Z", + "type": "event_msg", + "payload": {"type": "token_count", "info": { + "model": "gpt-5.6-sol", + "total_token_usage": {"input_tokens": total, "cached_input_tokens": 0, "output_tokens": 0}, + "last_token_usage": {"input_tokens": last, "cached_input_tokens": 0, "output_tokens": 0} + }} + }); + if let Some(ordinal) = ordinal { + value["ordinal"] = serde_json::json!(ordinal); + } + value.to_string() + }; + + state.process_line(&token_line(Some(9), 100, 0), &range); + state.process_line(&token_line(Some(10), 100, 0), &range); + state.process_line(&token_line(Some(11), 110, 10), &range); + assert!(state.fork_baseline_locally_resolved()); + state.process_line(&token_line(None, 120, 10), &range); + + assert!(!state.fork_baseline_locally_resolved()); +} + #[test] fn codex_token_pipeline_preserves_counts_above_i32_max() { let parsed = read_token_totals(&serde_json::json!({ @@ -1043,6 +1142,8 @@ fn session_meta_pre_read_accepts_snake_and_camel_fork_identity() { lineage: CodexSessionLineage::Child, fork_timestamp: Some("2026-05-31T10:00:00Z".to_string()), history_base_thread_id: Some("history-snake".to_string()), + is_subagent: false, + subagent_history_start_ordinal: None, } ); diff --git a/rust/src/cost_scanner/codex.rs b/rust/src/cost_scanner/codex.rs index f054c6eb8f..23730eb28f 100644 --- a/rust/src/cost_scanner/codex.rs +++ b/rust/src/cost_scanner/codex.rs @@ -1,5 +1,5 @@ use super::*; -use crate::core::{CodexForkAccountingState, CodexSessionLineage}; +use crate::core::{CodexForkAccountingState, CodexSessionLineage, CodexSessionMetadata}; mod cache_days; mod logical_target; @@ -13,6 +13,60 @@ use pending_range::{ }; use reconciliation::*; +#[derive(Debug)] +enum CodexAccountingMode { + Standard, + Baseline { + baseline: crate::core::CodexTotals, + paginated_continuation: bool, + remaining_inherited_totals: Option, + provenance: CodexBaselineProvenance, + }, + InferSubagent { + start_ordinal: Option, + }, + Unresolved, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CodexBaselineProvenance { + ValidatedParent { replaces_cached_state: bool }, + CachedValidatedParent, + CachedLocalInference, +} + +impl CodexAccountingMode { + fn is_unresolved(&self) -> bool { + matches!(self, Self::Unresolved) + } + + fn infers_subagent_baseline(&self) -> bool { + matches!(self, Self::InferSubagent { .. }) + } + + fn locally_resolved(&self) -> bool { + matches!( + self, + Self::Baseline { + provenance: CodexBaselineProvenance::CachedLocalInference, + .. + } + ) + } + + fn requires_cached_reparse(&self) -> bool { + matches!( + self, + Self::Baseline { + provenance: CodexBaselineProvenance::ValidatedParent { + replaces_cached_state: true + }, + .. + } + ) + } +} + fn summary_from_cached_report( report: &CachedCostReport, period_start: NaiveDate, @@ -41,65 +95,17 @@ fn summary_from_cached_report( } } -fn codex_fork_parent_is_safe(cache: &CostUsageCache, usage: &CostUsageFileUsage) -> bool { - let uses_parent_baseline = usage.codex_lineage.uses_parent_baseline() +fn codex_usage_uses_parent(usage: &CostUsageFileUsage) -> bool { + usage.codex_lineage.uses_parent_baseline() || (matches!(usage.codex_lineage, CodexSessionLineage::Root) - && usage.codex_forked_from_id.is_some()); - !uses_parent_baseline - || usage - .codex_forked_from_id - .as_deref() - .is_some_and(|parent_id| { - codex_parent_baseline(cache, parent_id, usage.codex_fork_timestamp.as_deref()) - .is_some() - }) + && usage.codex_forked_from_id.is_some()) } -/// Return a parent cumulative baseline only when exactly one cached session -/// identity is current, complete, timestamp-ordered, and safe to trust. -fn codex_parent_baseline( - cache: &CostUsageCache, - parent_session_id: &str, - child_fork_timestamp: Option<&str>, -) -> Option { - let mut baseline = None; - for (path_key, usage) in &cache.files { - if usage.codex_session_id.as_deref() != Some(parent_session_id) { - continue; - } - if usage.codex_unresolved_fork_parent - || usage.codex_token_timestamps_monotonic != Some(true) - { - return None; - } - let metadata = fs::metadata(path_key).ok()?; - if let (Some(expected), Some(actual)) = ( - usage.codex_file_identity.as_ref(), - JsonlScanner::codex_file_identity(Path::new(path_key), &metadata), - ) && expected != &actual - { - return None; - } - #[allow(clippy::cast_possible_wrap, reason = "session file sizes fit i64")] - let size = metadata.len().min(i64::MAX as u64) as i64; - if usage.mtime_unix_ms != system_time_to_unix_ms(metadata.modified().ok()) - || usage.size != size - || usage.parsed_bytes.unwrap_or(0) < size - { - return None; - } - let last_totals = usage.last_totals.clone()?; - let last_token_timestamp = usage.codex_last_token_timestamp.as_deref()?; - let child_fork_timestamp = child_fork_timestamp?; - if !JsonlScanner::codex_timestamp_at_or_before(last_token_timestamp, child_fork_timestamp) { - return None; - } - if baseline.replace(last_totals).is_some() { - // Duplicate identities make the dependency ambiguous. - return None; - } - } - baseline +fn codex_fork_uses_local_inference(usage: &CostUsageFileUsage) -> bool { + usage + .codex_fork_accounting_state + .as_ref() + .is_some_and(|state| state.locally_resolved) } fn is_codex_path_in_scan_window( @@ -129,6 +135,13 @@ struct CodexScanCandidate { mtime_unix_ms: i64, } +struct CodexPreparedCandidate { + path: PathBuf, + session_metadata: CodexSessionMetadata, + lineage_gate: CodexLineageGate, + parent_owner_expected: bool, +} + #[derive(Debug, Clone, Copy, Default)] struct CodexFileScanOutcome { bytes_read: i64, @@ -185,6 +198,7 @@ impl CostScanner { sessions_dirs: &[PathBuf], range: &CostUsageDayRange, cache: &CostUsageCache, + planner: &CodexLineagePlanner, cancel: Option<&AtomicBool>, stats: &mut CostScanStats, ) -> (Vec, bool) { @@ -246,7 +260,7 @@ impl CostScanner { }; let mtime_unix_ms = system_time_to_unix_ms(metadata.modified().ok()); let unchanged_complete = - cached_codex_file_is_complete_for_range(cache, &path_key, range); + cached_codex_file_is_complete_for_range(cache, planner, &path_key, range); if unchanged_complete { stats.files_seen = stats.files_seen.saturating_add(1); stats.files_skipped = stats.files_skipped.saturating_add(1); @@ -306,7 +320,10 @@ impl CostScanner { cancel: Option<&AtomicBool>, stats: &mut CostScanStats, ) { - let _ = self.parse_codex_file_bounded(path, range, summary, cache, cancel, stats, None); + let planner = CodexLineagePlanner::new(cache); + let _ = self.parse_codex_file_bounded( + path, range, summary, cache, cancel, stats, None, None, &planner, + ); } #[allow( @@ -322,11 +339,15 @@ impl CostScanner { cancel: Option<&AtomicBool>, stats: &mut CostScanStats, max_bytes_to_read: Option, + prepared_candidate: Option<&CodexPreparedCandidate>, + planner: &CodexLineagePlanner, ) -> CodexFileScanOutcome { if is_cancelled(cancel) { return CodexFileScanOutcome::default(); } - stats.files_seen = stats.files_seen.saturating_add(1); + if prepared_candidate.is_none() { + stats.files_seen = stats.files_seen.saturating_add(1); + } let metadata = match fs::metadata(path) { Ok(metadata) => metadata, @@ -356,20 +377,21 @@ impl CostScanner { }; } let cache_entry_is_fresh = |entry: &CostUsageFileUsage| { - cached_codex_file_is_fresh(cache, entry, cache_covers_range, mtime_ms, size) + cached_codex_file_is_fresh(cache, planner, entry, cache_covers_range, mtime_ms, size) }; - let identity_matches_cached = |entry: &CostUsageFileUsage| match ( - entry.codex_file_identity.as_ref(), - file_identity.as_ref(), - ) { - (Some(expected), Some(actual)) => expected == actual, - _ => false, + let identity_matches_cached = |entry: &CostUsageFileUsage| { + codex_file_identity_matches( + entry.codex_file_identity.as_deref(), + file_identity.as_deref(), + ) }; // The compact cache is authoritative for an unchanged file. Do this // before reading even the bounded metadata prefix; raw token history // is only needed after freshness fails or a fork needs reconciliation. if let Some(entry) = cached.as_ref() + && prepared_candidate + .is_none_or(|candidate| candidate.lineage_gate == CodexLineageGate::Eligible) && cache_entry_is_fresh(entry) && identity_matches_cached(entry) { @@ -386,13 +408,17 @@ impl CostScanner { }; } - stats.codex_metadata_read_paths.push(path_key.clone()); - stats.codex_read_receipt.metadata_reads = - stats.codex_read_receipt.metadata_reads.saturating_add(1); - let session_metadata = JsonlScanner::read_codex_session_metadata(path).unwrap_or_default(); - let cached_identity_matches = cached - .as_ref() - .is_some_and(|entry| entry.mtime_unix_ms == mtime_ms && entry.size == size); + let session_metadata = if let Some(prepared) = prepared_candidate { + prepared.session_metadata.clone() + } else { + stats.codex_metadata_read_paths.push(path_key.clone()); + stats.codex_read_receipt.metadata_reads = + stats.codex_read_receipt.metadata_reads.saturating_add(1); + JsonlScanner::read_codex_session_metadata(path).unwrap_or_default() + }; + // Cached lineage metadata belongs to a physical file, not merely a + // path/size/mtime tuple. Missing identity evidence fails closed. + let cached_identity_matches = cached.as_ref().is_some_and(identity_matches_cached); let codex_session_id = session_metadata.session_id.clone().or_else(|| { cached_identity_matches .then(|| cached.as_ref()?.codex_session_id.clone()) @@ -463,29 +489,34 @@ impl CostScanner { && state.history_base_thread_id == history_base_thread_id && state.fork_timestamp == codex_fork_timestamp }); - let fork_baseline = cached_fork_accounting_state + let matching_cached_fork_state = cached_fork_accounting_state .as_ref() - .filter(|_| cached_fork_state_matches) - .and_then(|state| state.inherited_totals.clone()) - .or_else(|| { - is_fork - .then_some(codex_forked_from_id.as_deref()) - .flatten() - .and_then(|parent_id| { - codex_parent_baseline(cache, parent_id, codex_fork_timestamp.as_deref()) - }) - }); - let remaining_inherited_totals = cached_fork_accounting_state - .as_ref() - .filter(|_| cached_fork_state_matches) - .and_then(|state| state.remaining_inherited_totals.clone()); + .filter(|_| cached_fork_state_matches); let paginated_continuation = is_fork && codex_forked_from_id.is_some() && history_base_thread_id .as_deref() .is_some_and(|history_base| Some(history_base) != codex_forked_from_id.as_deref()); + let lineage_gate = prepared_candidate + .map(|candidate| candidate.lineage_gate) + .unwrap_or_default(); + let parent_owner_expected = + prepared_candidate.is_some_and(|candidate| candidate.parent_owner_expected); + let lineage_decision = planner.decision_for_scan( + cache, + is_fork, + lineage_gate, + codex_forked_from_id.as_deref(), + codex_fork_timestamp.as_deref(), + parent_owner_expected, + ); + let accounting_mode = lineage_decision.accounting_mode( + matching_cached_fork_state, + &session_metadata, + paginated_continuation, + ); - if is_fork && fork_baseline.is_none() { + if accounting_mode.is_unresolved() { cache.files.insert( path_key, CostUsageFileUsage { @@ -515,9 +546,10 @@ impl CostScanner { } if let Some(entry) = &cached - && cached_codex_file_is_fresh(cache, entry, cache_covers_range, mtime_ms, size) - && (entry.codex_file_identity.is_none() || identity_matches_cached(entry)) + && cached_codex_file_is_fresh(cache, planner, entry, cache_covers_range, mtime_ms, size) + && identity_matches_cached(entry) && !cached_identity_changed + && !accounting_mode.requires_cached_reparse() { let (session_cost, has_tokens) = add_codex_days_map_to_summary(summary, &entry.days, range); @@ -525,11 +557,6 @@ impl CostScanner { summary.total_cost_usd += session_cost; summary.sessions_count += 1; } - if entry.codex_file_identity != file_identity { - let mut refreshed = entry.clone(); - refreshed.codex_file_identity = file_identity.clone(); - cache.files.insert(path_key.clone(), refreshed); - } stats.files_skipped = stats.files_skipped.saturating_add(1); return CodexFileScanOutcome { bytes_read: 0, @@ -623,22 +650,15 @@ impl CostScanner { } } - let parse_target_size = cached - .as_ref() - .and_then(|entry| codex_resumable_scan_target_size(size, entry)); - let parse_result = match if let Some(baseline) = fork_baseline.clone() { - JsonlScanner::parse_codex_file_with_state_bounded_fork_target_with_accounting( - path, - range, - baseline, - paginated_continuation, - remaining_inherited_totals.clone(), - cancel, - parse_target_size, - max_bytes_to_read, - ) - } else { - JsonlScanner::parse_codex_file_with_state_bounded( + let parse_target_size = (!accounting_mode.requires_cached_reparse()) + .then(|| { + cached + .as_ref() + .and_then(|entry| codex_resumable_scan_target_size(size, entry)) + }) + .flatten(); + let parse_result = match match &accounting_mode { + CodexAccountingMode::Standard => JsonlScanner::parse_codex_file_with_state_bounded( path, range, 0, @@ -648,7 +668,35 @@ impl CostScanner { None, cancel, max_bytes_to_read, - ) + ), + CodexAccountingMode::Baseline { + baseline, + paginated_continuation, + remaining_inherited_totals, + .. + } => JsonlScanner::parse_codex_file_with_state_bounded_fork_target_with_accounting( + path, + range, + baseline.clone(), + *paginated_continuation, + remaining_inherited_totals.clone(), + cancel, + parse_target_size, + max_bytes_to_read, + ), + CodexAccountingMode::InferSubagent { start_ordinal } => { + JsonlScanner::parse_codex_file_with_inferred_fork_baseline( + path, + range, + *start_ordinal, + cancel, + parse_target_size, + max_bytes_to_read, + ) + } + CodexAccountingMode::Unresolved => { + unreachable!("unresolved forks return before parsing") + } } { Ok(result) => result, Err(_) => return CodexFileScanOutcome::default(), @@ -656,7 +704,10 @@ impl CostScanner { stats.token_timestamp_comparisons = stats .token_timestamp_comparisons .saturating_add(parse_result.token_timestamp_comparisons); - if parse_result.fork_baseline_ambiguous { + if parse_result.fork_baseline_ambiguous + || (accounting_mode.infers_subagent_baseline() + && !parse_result.fork_baseline_locally_resolved) + { cache.files.insert( path_key, CostUsageFileUsage { @@ -696,18 +747,20 @@ impl CostScanner { bytes_read: parse_result.bytes_read, is_complete: parse_result.is_complete, }; - let codex_fork_accounting_state = if is_fork { - parse_result - .fork_baseline - .clone() - .map(|inherited_totals| CodexForkAccountingState { - session_id: codex_session_id.clone(), - forked_from_id: codex_forked_from_id.clone(), - history_base_thread_id: history_base_thread_id.clone(), - fork_timestamp: codex_fork_timestamp.clone(), - inherited_totals: Some(inherited_totals), - remaining_inherited_totals: parse_result.remaining_inherited_totals.clone(), - }) + let locally_resolved = + accounting_mode.locally_resolved() || parse_result.fork_baseline_locally_resolved; + let codex_fork_accounting_state = if is_fork + && (parse_result.fork_baseline.is_some() || parse_result.fork_baseline_locally_resolved) + { + Some(CodexForkAccountingState { + session_id: codex_session_id.clone(), + forked_from_id: codex_forked_from_id.clone(), + history_base_thread_id: history_base_thread_id.clone(), + fork_timestamp: codex_fork_timestamp.clone(), + inherited_totals: parse_result.fork_baseline.clone(), + remaining_inherited_totals: parse_result.remaining_inherited_totals.clone(), + locally_resolved, + }) } else { None }; diff --git a/rust/src/cost_scanner/codex/logical_target.rs b/rust/src/cost_scanner/codex/logical_target.rs index a98fb1e63e..b261f174fe 100644 --- a/rust/src/cost_scanner/codex/logical_target.rs +++ b/rust/src/cost_scanner/codex/logical_target.rs @@ -1,7 +1,560 @@ use super::*; +#[cfg(test)] +use std::cell::Cell; +use std::collections::VecDeque; + +#[cfg(test)] +thread_local! { static CODEX_LINEAGE_GRAPH_BUILDS: Cell = const { Cell::new(0) }; } + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(super) enum CodexLineageGate { + #[default] + Eligible, + Unsafe, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum CodexLineageDecision { + Root, + ParentAbsent, + ParentReady(crate::core::CodexTotals), + Unsafe, +} + +struct CodexLineageNode { + path: String, + session_id: Option, + parent_id: Option, + candidate_index: Option, + may_infer_missing_parent: bool, + may_author_parent: bool, + initially_unsafe: bool, +} + +struct CodexLineageGraph { + nodes: Vec, + session_owners: HashMap>, + parent_indices: Vec>, + gates: Vec, + candidate_node_indices: Vec, + ordered_candidate_indices: Vec, +} + +impl CodexLineageGraph { + fn new(cache: &CostUsageCache, candidates: Option<&[CodexPreparedCandidate]>) -> Self { + #[cfg(test)] + CODEX_LINEAGE_GRAPH_BUILDS.with(|builds| builds.set(builds.get() + 1)); + let candidate_paths = candidates + .into_iter() + .flatten() + .map(|candidate| candidate.path.to_string_lossy().to_string()) + .collect::>(); + let mut cached_paths = cache + .files + .keys() + .filter(|path| !candidate_paths.contains(*path)) + .cloned() + .collect::>(); + cached_paths.sort(); + + let mut nodes = Vec::with_capacity(cached_paths.len() + candidate_paths.len()); + for path in cached_paths { + let usage = &cache.files[&path]; + let uses_parent = codex_usage_uses_parent(usage); + let locally_inferred = codex_fork_uses_local_inference(usage); + nodes.push(CodexLineageNode { + path, + session_id: usage.codex_session_id.clone(), + parent_id: uses_parent + .then(|| usage.codex_forked_from_id.clone()) + .flatten(), + candidate_index: None, + may_infer_missing_parent: locally_inferred, + may_author_parent: !locally_inferred && !usage.codex_unresolved_fork_parent, + initially_unsafe: usage.codex_unresolved_fork_parent, + }); + } + + let mut candidate_node_indices = Vec::new(); + if let Some(candidates) = candidates { + candidate_node_indices.reserve(candidates.len()); + for (candidate_index, candidate) in candidates.iter().enumerate() { + let cached = cache + .files + .get(&candidate.path.to_string_lossy().to_string()); + let cached_identity_matches = cached.is_some_and(|usage| { + let Ok(metadata) = fs::metadata(&candidate.path) else { + return false; + }; + let expected = usage.codex_file_identity.as_deref(); + let actual = JsonlScanner::codex_file_identity(&candidate.path, &metadata); + codex_file_identity_matches(expected, actual.as_deref()) + }); + let metadata_owns_identity = candidate.session_metadata.session_id.is_some(); + let uses_parent = if metadata_owns_identity { + candidate.session_metadata.lineage.uses_parent_baseline() + || candidate.session_metadata.forked_from_id.is_some() + } else if cached_identity_matches { + cached.is_some_and(super::codex_usage_uses_parent) + } else { + candidate.session_metadata.lineage.uses_parent_baseline() + || candidate.session_metadata.forked_from_id.is_some() + }; + let session_id = candidate.session_metadata.session_id.clone().or_else(|| { + cached_identity_matches + .then(|| cached.and_then(|usage| usage.codex_session_id.clone())) + .flatten() + }); + let parent_id = if metadata_owns_identity { + candidate.session_metadata.forked_from_id.clone() + } else { + candidate + .session_metadata + .forked_from_id + .clone() + .or_else(|| { + (cached_identity_matches && uses_parent) + .then(|| { + cached.and_then(|usage| usage.codex_forked_from_id.clone()) + }) + .flatten() + }) + }; + // Freshly read identity-bearing metadata owns this candidate's + // current lineage state. Retain cached lineage flags only when + // the bounded metadata read could not establish an identity. + let cached_fallback = if metadata_owns_identity { + None + } else { + cached_identity_matches.then_some(cached).flatten() + }; + nodes.push(CodexLineageNode { + path: candidate.path.to_string_lossy().to_string(), + session_id, + parent_id: uses_parent.then_some(parent_id).flatten(), + candidate_index: Some(candidate_index), + may_infer_missing_parent: candidate.session_metadata.is_subagent + || cached_fallback.is_some_and(super::codex_fork_uses_local_inference), + may_author_parent: cached_fallback.is_none_or(|usage| { + !super::codex_fork_uses_local_inference(usage) + && !usage.codex_unresolved_fork_parent + }), + initially_unsafe: cached_fallback + .is_some_and(|usage| usage.codex_unresolved_fork_parent), + }); + candidate_node_indices.push(nodes.len() - 1); + } + } + + let mut session_owners = HashMap::>::new(); + for (index, node) in nodes.iter().enumerate() { + if let Some(session_id) = node.session_id.as_ref() { + session_owners + .entry(session_id.clone()) + .or_default() + .push(index); + } + } + + let mut gates = nodes + .iter() + .map(|node| { + if node.initially_unsafe { + CodexLineageGate::Unsafe + } else { + CodexLineageGate::Eligible + } + }) + .collect::>(); + for owners in session_owners.values().filter(|owners| owners.len() > 1) { + for &index in owners { + gates[index] = CodexLineageGate::Unsafe; + } + } + + let mut parent_indices = vec![None; nodes.len()]; + for (index, node) in nodes.iter().enumerate() { + let Some(parent_id) = node.parent_id.as_ref() else { + continue; + }; + match session_owners.get(parent_id) { + Some(owners) if owners.len() == 1 => parent_indices[index] = Some(owners[0]), + Some(_) => { + gates[index] = CodexLineageGate::Unsafe; + } + None if !node.may_infer_missing_parent => { + gates[index] = CodexLineageGate::Unsafe; + } + None => {} + } + } + + // This single topological pass both rejects cycles/unsafe ancestry and + // orders candidates. Cached-parent validation consumes the same gates. + let mut completed = vec![false; nodes.len()]; + let mut ordered_candidate_indices = Vec::with_capacity(candidate_node_indices.len()); + let mut children = vec![Vec::new(); nodes.len()]; + let mut ready = VecDeque::new(); + for (index, parent) in parent_indices.iter().enumerate() { + match parent { + Some(parent_index) => children[*parent_index].push(index), + None if gates[index] == CodexLineageGate::Eligible => ready.push_back(index), + None => {} + } + } + while let Some(index) = ready.pop_front() { + if completed[index] || gates[index] == CodexLineageGate::Unsafe { + continue; + } + completed[index] = true; + if let Some(candidate_index) = nodes[index].candidate_index { + ordered_candidate_indices.push(candidate_index); + } + if nodes[index].may_author_parent { + for child in &children[index] { + if gates[*child] == CodexLineageGate::Eligible { + ready.push_back(*child); + } + } + } + } + + for index in 0..nodes.len() { + if !completed[index] { + gates[index] = CodexLineageGate::Unsafe; + if let Some(candidate_index) = nodes[index].candidate_index { + ordered_candidate_indices.push(candidate_index); + } + } + } + + Self { + nodes, + session_owners, + parent_indices, + gates, + candidate_node_indices, + ordered_candidate_indices, + } + } + + fn unique_owner(&self, session_id: &str) -> Result, ()> { + match self.session_owners.get(session_id).map(Vec::as_slice) { + None | Some([]) => Ok(None), + Some([index]) => Ok(Some(*index)), + Some(_) => Err(()), + } + } + + fn apply_candidate_plan( + &self, + candidates: &mut Vec, + sessions_dirs: &[PathBuf], + range: &CostUsageDayRange, + ) -> Vec { + if !candidates.is_empty() { + for (candidate_index, candidate) in candidates.iter_mut().enumerate() { + let node_index = self.candidate_node_indices[candidate_index]; + candidate.lineage_gate = self.gates[node_index]; + candidate.parent_owner_expected = self.parent_indices[node_index].is_some(); + } + + let mut remaining = candidates.drain(..).map(Some).collect::>(); + for candidate_index in &self.ordered_candidate_indices { + candidates.push( + remaining[*candidate_index] + .take() + .expect("candidate is ordered once"), + ); + } + } + + // Keep the complete graph for parent resolution, but invalidate only + // unsafe cached nodes in the active range or in the ancestor closure + // required to resolve an active node. + let mut relevant = vec![false; self.nodes.len()]; + let mut pending = VecDeque::new(); + for (index, node) in self.nodes.iter().enumerate() { + if super::is_codex_path_in_scan_window(Path::new(&node.path), sessions_dirs, range) { + relevant[index] = true; + pending.push_back(index); + } + } + while let Some(index) = pending.pop_front() { + let Some(parent_id) = self.nodes[index].parent_id.as_deref() else { + continue; + }; + if let Some(owners) = self.session_owners.get(parent_id) { + for &owner in owners { + if !relevant[owner] { + relevant[owner] = true; + pending.push_back(owner); + } + } + } + } + + self.nodes + .iter() + .zip(&self.gates) + .enumerate() + .filter(|(index, (node, gate))| { + relevant[*index] + && node.candidate_index.is_none() + && **gate == CodexLineageGate::Unsafe + && !node.initially_unsafe + }) + .map(|(_, (node, _))| node.path.clone()) + .collect() + } +} + +pub(super) struct CodexLineagePlanner { + graph: Option, +} + +impl CodexLineagePlanner { + pub(super) fn new(cache: &CostUsageCache) -> Self { + Self { + graph: Self::needs_graph(cache, None).then(|| CodexLineageGraph::new(cache, None)), + } + } + + pub(super) fn plan_candidates_by_lineage( + cache: &CostUsageCache, + candidates: &mut Vec, + sessions_dirs: &[PathBuf], + range: &CostUsageDayRange, + ) -> (Self, Vec) { + let graph = Self::needs_graph(cache, Some(candidates)) + .then(|| CodexLineageGraph::new(cache, Some(candidates))); + let unsafe_paths = graph.as_ref().map_or_else(Vec::new, |graph| { + graph.apply_candidate_plan(candidates, sessions_dirs, range) + }); + (Self { graph }, unsafe_paths) + } + + fn needs_graph(cache: &CostUsageCache, candidates: Option<&[CodexPreparedCandidate]>) -> bool { + cache.files.values().any(super::codex_usage_uses_parent) + || candidates.is_some_and(|items| { + items.iter().any(|candidate| { + candidate.session_metadata.lineage.uses_parent_baseline() + || candidate.session_metadata.forked_from_id.is_some() + }) + }) + } + + #[cfg(test)] + pub(crate) fn reset_graph_build_count() { + CODEX_LINEAGE_GRAPH_BUILDS.with(|count| count.set(0)); + } + #[cfg(test)] + pub(crate) fn graph_build_count() -> usize { + CODEX_LINEAGE_GRAPH_BUILDS.with(Cell::get) + } + + fn graph(&self) -> Option<&CodexLineageGraph> { + self.graph.as_ref() + } + + pub(super) fn cached_usage_is_safe( + &self, + cache: &CostUsageCache, + usage: &CostUsageFileUsage, + ) -> bool { + let locally_resolved = super::codex_fork_uses_local_inference(usage); + match self.decision_for_usage(cache, usage) { + CodexLineageDecision::Root => true, + CodexLineageDecision::ParentAbsent => locally_resolved, + CodexLineageDecision::ParentReady(_) => !locally_resolved, + CodexLineageDecision::Unsafe => false, + } + } + + pub(super) fn decision_for_scan( + &self, + cache: &CostUsageCache, + uses_parent: bool, + gate: CodexLineageGate, + parent_id: Option<&str>, + fork_timestamp: Option<&str>, + parent_owner_expected: bool, + ) -> CodexLineageDecision { + if gate == CodexLineageGate::Unsafe { + return CodexLineageDecision::Unsafe; + } + if !uses_parent { + return CodexLineageDecision::Root; + } + parent_id.map_or(CodexLineageDecision::Unsafe, |parent_id| { + self.resolve_parent(cache, parent_id, fork_timestamp, parent_owner_expected) + }) + } + + pub(super) fn decision_for_usage( + &self, + cache: &CostUsageCache, + usage: &CostUsageFileUsage, + ) -> CodexLineageDecision { + if usage.codex_unresolved_fork_parent { + return CodexLineageDecision::Unsafe; + } + if !super::codex_usage_uses_parent(usage) { + return CodexLineageDecision::Root; + } + usage + .codex_forked_from_id + .as_deref() + .map_or(CodexLineageDecision::Unsafe, |parent_id| { + self.resolve_parent( + cache, + parent_id, + usage.codex_fork_timestamp.as_deref(), + false, + ) + }) + } + + /// Resolve one parent identity through the persisted graph. Absence is + /// distinct from ambiguity and transitive unsafety so local inference is + /// allowed only when no owner exists at all. + fn resolve_parent( + &self, + cache: &CostUsageCache, + parent_session_id: &str, + child_fork_timestamp: Option<&str>, + parent_owner_expected: bool, + ) -> CodexLineageDecision { + let Some(graph) = self.graph() else { + return CodexLineageDecision::Unsafe; + }; + let node_index = match graph.unique_owner(parent_session_id) { + Ok(None) if !parent_owner_expected => return CodexLineageDecision::ParentAbsent, + Ok(None) | Err(()) => return CodexLineageDecision::Unsafe, + Ok(Some(index)) => index, + }; + self.parent_owner_baseline(cache, node_index, child_fork_timestamp) + .map_or( + CodexLineageDecision::Unsafe, + CodexLineageDecision::ParentReady, + ) + } + + fn parent_owner_baseline( + &self, + cache: &CostUsageCache, + node_index: usize, + child_fork_timestamp: Option<&str>, + ) -> Option { + let graph = self.graph()?; + let node = graph.nodes.get(node_index)?; + if graph.gates[node_index] == CodexLineageGate::Unsafe || !node.may_author_parent { + return None; + } + let usage = cache.files.get(&node.path)?; + if usage.codex_unresolved_fork_parent + || usage.codex_token_timestamps_monotonic != Some(true) + || super::codex_fork_uses_local_inference(usage) + { + return None; + } + + if super::codex_usage_uses_parent(usage) { + let inherited = usage + .codex_fork_accounting_state + .as_ref()? + .inherited_totals + .as_ref()?; + let parent_index = graph.parent_indices[node_index]?; + let baseline = self.parent_owner_baseline( + cache, + parent_index, + usage.codex_fork_timestamp.as_deref(), + )?; + if &baseline != inherited { + return None; + } + } + + let metadata = fs::metadata(&node.path).ok()?; + let expected_identity = usage.codex_file_identity.as_ref()?; + let actual_identity = JsonlScanner::codex_file_identity(Path::new(&node.path), &metadata)?; + if expected_identity != &actual_identity { + return None; + } + #[allow(clippy::cast_possible_wrap, reason = "session file sizes fit i64")] + let size = metadata.len().min(i64::MAX as u64) as i64; + if usage.mtime_unix_ms != system_time_to_unix_ms(metadata.modified().ok()) + || usage.size != size + || usage.parsed_bytes.unwrap_or(0) < size + { + return None; + } + let last_totals = usage.last_totals.clone()?; + let last_token_timestamp = usage.codex_last_token_timestamp.as_deref()?; + let child_fork_timestamp = child_fork_timestamp?; + JsonlScanner::codex_timestamp_at_or_before(last_token_timestamp, child_fork_timestamp) + .then_some(last_totals) + } +} + +impl CodexLineageDecision { + pub(super) fn accounting_mode( + &self, + matching_cached_state: Option<&CodexForkAccountingState>, + metadata: &CodexSessionMetadata, + paginated_continuation: bool, + ) -> CodexAccountingMode { + match self { + Self::Root => CodexAccountingMode::Standard, + Self::Unsafe => CodexAccountingMode::Unresolved, + Self::ParentReady(baseline) => { + let replaces_cached_state = matching_cached_state.is_some_and(|state| { + state.locally_resolved || state.inherited_totals.as_ref() != Some(baseline) + }); + let cached_parent_state = matching_cached_state.filter(|state| { + !state.locally_resolved && state.inherited_totals.as_ref() == Some(baseline) + }); + CodexAccountingMode::Baseline { + baseline: baseline.clone(), + paginated_continuation, + remaining_inherited_totals: cached_parent_state + .and_then(|state| state.remaining_inherited_totals.clone()), + provenance: CodexBaselineProvenance::ValidatedParent { + replaces_cached_state, + }, + } + } + Self::ParentAbsent => { + if let Some(state) = matching_cached_state + && let Some(baseline) = state.inherited_totals.clone() + { + return CodexAccountingMode::Baseline { + baseline, + paginated_continuation, + remaining_inherited_totals: state.remaining_inherited_totals.clone(), + provenance: if state.locally_resolved { + CodexBaselineProvenance::CachedLocalInference + } else { + CodexBaselineProvenance::CachedValidatedParent + }, + }; + } + if metadata.is_subagent { + CodexAccountingMode::InferSubagent { + start_ordinal: metadata.subagent_history_start_ordinal, + } + } else { + CodexAccountingMode::Unresolved + } + } + } + } +} pub(super) fn cached_codex_file_is_fresh( cache: &CostUsageCache, + planner: &CodexLineagePlanner, entry: &CostUsageFileUsage, cache_covers_range: bool, mtime_unix_ms: i64, @@ -13,11 +566,18 @@ pub(super) fn cached_codex_file_is_fresh( && entry.size == size && codex_scan_target_size(entry) == size && entry.parsed_bytes.unwrap_or(0) >= size - && super::codex_fork_parent_is_safe(cache, entry) + && planner.cached_usage_is_safe(cache, entry) +} + +pub(super) fn codex_file_identity_matches(expected: Option<&str>, actual: Option<&str>) -> bool { + expected + .zip(actual) + .is_some_and(|(expected, actual)| expected == actual) } pub(super) fn cached_codex_file_is_complete_for_range( cache: &CostUsageCache, + planner: &CodexLineagePlanner, path_key: &str, range: &CostUsageDayRange, ) -> bool { @@ -26,14 +586,10 @@ pub(super) fn cached_codex_file_is_complete_for_range( let Ok(metadata) = fs::metadata(path_key) else { return false; }; - let identity_matches = match ( - usage.codex_file_identity.as_ref(), - JsonlScanner::codex_file_identity(Path::new(path_key), &metadata).as_ref(), - ) { - (Some(expected), Some(actual)) => expected == actual, - (Some(_), None) => false, - (None, _) => true, - }; + let identity_matches = codex_file_identity_matches( + usage.codex_file_identity.as_deref(), + JsonlScanner::codex_file_identity(Path::new(path_key), &metadata).as_deref(), + ); #[allow(clippy::cast_possible_wrap, reason = "session file sizes fit i64")] let size = metadata.len().min(i64::MAX as u64) as i64; identity_matches @@ -42,10 +598,59 @@ pub(super) fn cached_codex_file_is_complete_for_range( && codex_scan_target_size(usage) == size && usage.parsed_bytes.unwrap_or(0) >= size && !usage.codex_unresolved_fork_parent - && super::codex_fork_parent_is_safe(cache, usage) + // Reconsider locally inferred children after this pass has + // had a chance to discover and cache their parent. + && !super::codex_fork_uses_local_inference(usage) + && planner.cached_usage_is_safe(cache, usage) }) } +/// Process cached local-inference children after all other candidates. A +/// parent discovered in this pass must enter the cache before its unchanged +/// child can decide whether the inferred baseline is still authoritative. +pub(super) fn defer_codex_locally_inferred_candidates( + candidates: &mut Vec, + cache: &CostUsageCache, +) { + if candidates.len() < 2 { + return; + } + + let mut other = Vec::with_capacity(candidates.len()); + let mut locally_inferred = Vec::new(); + for candidate in candidates.drain(..) { + let path_key = candidate.path.to_string_lossy(); + if cache + .files + .get(path_key.as_ref()) + .is_some_and(super::codex_fork_uses_local_inference) + { + locally_inferred.push(candidate); + } else { + other.push(candidate); + } + } + other.extend(locally_inferred); + candidates.extend(other); +} + +pub(super) fn invalidate_codex_unsafe_lineage(cache: &mut CostUsageCache, paths: &[String]) { + for path in paths { + let Some(usage) = cache.files.get_mut(path) else { + continue; + }; + usage.days.clear(); + usage.parsed_bytes = Some(0); + usage.codex_scan_target_size = None; + usage.last_model = None; + usage.last_totals = None; + usage.codex_token_timestamps_monotonic = None; + usage.codex_last_token_timestamp = None; + usage.codex_fork_accounting_state = None; + usage.codex_unresolved_fork_parent = true; + } +} + /// Give paths already in the durable queue their saved turn before newly /// discovered dirty paths. The scanner appends unfinished paths after this /// pass, making the queue a round-robin cursor instead of a newest-first loop. diff --git a/rust/src/cost_scanner/codex/reconciliation.rs b/rust/src/cost_scanner/codex/reconciliation.rs index b16d211b3b..8451c3f72b 100644 --- a/rust/src/cost_scanner/codex/reconciliation.rs +++ b/rust/src/cost_scanner/codex/reconciliation.rs @@ -108,14 +108,10 @@ fn codex_pending_path_affects_current_window( if codex_logical_target_has_unconsumed_tail(observed_size, usage) { return true; } - let identity_matches = match ( - usage.codex_file_identity.as_ref(), - JsonlScanner::codex_file_identity(Path::new(path_key), &metadata).as_ref(), - ) { - (Some(expected), Some(actual)) => expected == actual, - (Some(_), None) => false, - (None, _) => true, - }; + let identity_matches = super::codex_file_identity_matches( + usage.codex_file_identity.as_deref(), + JsonlScanner::codex_file_identity(Path::new(path_key), &metadata).as_deref(), + ); if !identity_matches || usage.mtime_unix_ms != system_time_to_unix_ms(metadata.modified().ok()) || usage.size != observed_size @@ -270,6 +266,7 @@ mod tests { let old_usage = cache.files.get_mut(&old_key).unwrap(); old_usage.mtime_unix_ms = system_time_to_unix_ms(metadata.modified().ok()); old_usage.size = i64::try_from(metadata.len()).unwrap(); + old_usage.codex_file_identity = JsonlScanner::codex_file_identity(&old_path, &metadata); let range = active_range(); assert!(codex_current_window_is_established(&cache, &range)); @@ -279,6 +276,32 @@ mod tests { assert_eq!(report.sessions_count, 1); } + #[test] + fn historical_pending_entry_without_or_mismatched_identity_blocks_publication() { + let root = tempfile::tempdir().unwrap(); + let old_path = root.path().join("old.jsonl"); + let current_path = root.path().join("current.jsonl"); + std::fs::write(&old_path, vec![0_u8; 100]).unwrap(); + std::fs::write(¤t_path, vec![0_u8; 100]).unwrap(); + let old_key = old_path.to_string_lossy().into_owned(); + let current_key = current_path.to_string_lossy().into_owned(); + let metadata = std::fs::metadata(&old_path).unwrap(); + let range = active_range(); + + for cached_identity in [None, Some("different-file".to_string())] { + let mut cache = historical_pending_cache(&old_key, ¤t_key); + let old_usage = cache.files.get_mut(&old_key).unwrap(); + old_usage.mtime_unix_ms = system_time_to_unix_ms(metadata.modified().ok()); + old_usage.size = i64::try_from(metadata.len()).unwrap(); + old_usage.codex_file_identity = cached_identity; + + assert!(codex_pending_path_affects_current_window( + &cache, &old_key, &range + )); + assert!(!codex_current_window_is_established(&cache, &range)); + } + } + #[test] fn metadata_failure_blocks_historical_pending_publication() { let old_path = r"C:\sessions\missing-old.jsonl"; diff --git a/rust/src/cost_scanner/codex/scan.rs b/rust/src/cost_scanner/codex/scan.rs index e983adfd6a..94cb91f701 100644 --- a/rust/src/cost_scanner/codex/scan.rs +++ b/rust/src/cost_scanner/codex/scan.rs @@ -177,8 +177,15 @@ pub(super) fn scan_codex_detailed_with_cache( cache.codex_pending_scan_root_paths = pending_scan.root_paths.clone(); cache.codex_pending_scan_timezone = Some(pending_scan.timezone.clone()); - let (mut candidates, discovery_complete) = - scanner.collect_codex_candidates(&sessions_dirs, scan_range, &cache, cancel, &mut stats); + let cached_lineage = CodexLineagePlanner::new(&cache); + let (mut candidates, discovery_complete) = scanner.collect_codex_candidates( + &sessions_dirs, + scan_range, + &cache, + &cached_lineage, + cancel, + &mut stats, + ); let candidate_limit = if scanner.options.codex_candidate_limit == 0 { usize::MAX } else { @@ -197,43 +204,78 @@ pub(super) fn scan_codex_detailed_with_cache( let mut bytes_read_this_refresh = 0_i64; let mut pending_next = cache.codex_pending_paths.clone(); let pending_paths_before_pass = cache.codex_pending_paths.clone(); + let mut invalidated_unsafe_lineage = false; prioritize_codex_pending_candidates(&mut candidates, &pending_paths_before_pass); + defer_codex_locally_inferred_candidates(&mut candidates, &cache); if discovery_complete && !is_cancelled(cancel) { - pending_next - .retain(|path| !cached_codex_file_is_complete_for_range(&cache, path, scan_range)); + pending_next.retain(|path| { + !cached_codex_file_is_complete_for_range(&cache, &cached_lineage, path, scan_range) + }); } - let mut incomplete_processed = Vec::new(); - for (index, candidate) in candidates.iter().enumerate() { - if is_cancelled(cancel) - || index >= candidate_limit - || bytes_read_this_refresh >= refresh_byte_limit - { - for deferred in &candidates[index..] { - let key = deferred.path.to_string_lossy().to_string(); - if !pending_next.contains(&key) { - pending_next.push(key); - } + // Admit one bounded set, inspect each admitted candidate once, and order + // that set by lineage before reading token history. This makes cold + // child-before-parent scans parent-first without a second parse pass. + let deferred_candidates = candidates.split_off(candidate_limit.min(candidates.len())); + let deferred_paths = deferred_candidates + .into_iter() + .map(|candidate| candidate.path) + .collect::>(); + let mut work_queue = Vec::with_capacity(candidates.len()); + let mut cancelled_during_preparation = Vec::new(); + for candidate in candidates { + if is_cancelled(cancel) { + cancelled_during_preparation.push(candidate.path); + continue; + } + let key = candidate.path.to_string_lossy().to_string(); + stats.files_seen = stats.files_seen.saturating_add(1); + stats.codex_metadata_read_paths.push(key); + stats.codex_read_receipt.metadata_reads = + stats.codex_read_receipt.metadata_reads.saturating_add(1); + work_queue.push(CodexPreparedCandidate { + session_metadata: JsonlScanner::read_codex_session_metadata(&candidate.path) + .unwrap_or_default(), + path: candidate.path, + lineage_gate: CodexLineageGate::Eligible, + parent_owner_expected: false, + }); + } + let mut unprocessed = Vec::new(); + let cancelled_before_plan = !cancelled_during_preparation.is_empty() || is_cancelled(cancel); + let lineage_planner = if cancelled_before_plan { + unprocessed.extend(work_queue.drain(..).map(|candidate| candidate.path)); + unprocessed.extend(cancelled_during_preparation); + cached_lineage + } else { + let (planner, unsafe_cached_paths) = CodexLineagePlanner::plan_candidates_by_lineage( + &cache, + &mut work_queue, + &sessions_dirs, + scan_range, + ); + invalidated_unsafe_lineage = !unsafe_cached_paths.is_empty(); + if invalidated_unsafe_lineage { + cache.previous_report = None; + } + invalidate_codex_unsafe_lineage(&mut cache, &unsafe_cached_paths); + for path in unsafe_cached_paths { + if !pending_next.contains(&path) { + pending_next.push(path); } - stats.files_deferred = stats.files_deferred.saturating_add( - u32::try_from((candidates.len() - index).min(u32::MAX as usize)) - .unwrap_or(u32::MAX), - ); - break; } + planner + }; + let mut incomplete_processed = Vec::new(); + for (index, candidate) in work_queue.iter().enumerate() { let refresh_remaining = refresh_byte_limit.saturating_sub(bytes_read_this_refresh); let allowance = per_file_limit.min(refresh_remaining); - if allowance <= 0 { - for deferred in &candidates[index..] { - let key = deferred.path.to_string_lossy().to_string(); - if !pending_next.contains(&key) { - pending_next.push(key); - } - } - stats.files_deferred = stats.files_deferred.saturating_add( - u32::try_from((candidates.len() - index).min(u32::MAX as usize)) - .unwrap_or(u32::MAX), + if is_cancelled(cancel) || allowance <= 0 { + unprocessed.extend( + work_queue[index..] + .iter() + .map(|candidate| candidate.path.clone()), ); break; } @@ -246,6 +288,8 @@ pub(super) fn scan_codex_detailed_with_cache( cancel, &mut stats, Some(allowance), + Some(candidate), + &lineage_planner, ); bytes_read_this_refresh = bytes_read_this_refresh.saturating_add(outcome.bytes_read.max(0)); stats.codex_bytes_read = stats @@ -275,6 +319,16 @@ pub(super) fn scan_codex_detailed_with_cache( apply_codex_source_row_plan(&mut cache, &key, plan); } } + unprocessed.extend(deferred_paths); + stats.files_deferred = stats.files_deferred.saturating_add( + u32::try_from(unprocessed.len().min(u32::MAX as usize)).unwrap_or(u32::MAX), + ); + for path in unprocessed { + let key = path.to_string_lossy().to_string(); + if !pending_next.contains(&key) { + pending_next.push(key); + } + } pending_next.extend(incomplete_processed); let mut pruned_paths_pending = Vec::new(); @@ -344,7 +398,7 @@ pub(super) fn scan_codex_detailed_with_cache( // the range so unchanged files stay on the cache fast path. cache.scan_since_key = Some(scan_range.scan_since_key.clone()); cache.scan_until_key = Some(scan_range.scan_until_key.clone()); - } else if cache.previous_report.is_none() { + } else if cache.previous_report.is_none() && !invalidated_unsafe_lineage { cache.previous_report = established_report_before_scan; } if !is_cancelled(cancel) { diff --git a/rust/src/cost_scanner/tests.rs b/rust/src/cost_scanner/tests.rs index 3b16c92a1c..9a930a3e7b 100644 --- a/rust/src/cost_scanner/tests.rs +++ b/rust/src/cost_scanner/tests.rs @@ -1449,10 +1449,16 @@ fn cost_scan_second_pass_skips_unchanged_files_via_cache() { // Second pass with default debounce still inspects files but skips re-parse. // Use app_driven so we exercise per-file mtime skip rather than whole-scan debounce. + CodexLineagePlanner::reset_graph_build_count(); let (summary2, stats2) = scanner.scan_codex_detailed(None); assert_eq!(stats2.files_seen, 2); assert_eq!(stats2.files_skipped, 2, "cache hit skips re-parse"); assert_eq!(stats2.files_parsed, 0); + assert_eq!( + CodexLineagePlanner::graph_build_count(), + 0, + "warm root-only scan must bypass lineage graph construction" + ); assert!(stats2.codex_metadata_read_paths.is_empty()); assert!(stats2.codex_history_read_paths.is_empty()); assert_eq!(stats2.codex_read_receipt, Default::default()); @@ -3036,6 +3042,12 @@ fn incomplete_or_buffered_empty_codex_fragment_is_not_marked_complete() { assert!(buffered_cache.codex_pending_paths.contains(&buffered_key)); } +#[cfg(test)] +#[path = "tests/copied_prefix.rs"] +mod copied_prefix; +#[cfg(test)] +#[path = "tests/lineage_cache.rs"] +mod lineage_cache; #[cfg(test)] #[path = "tests/paginated.rs"] mod paginated; diff --git a/rust/src/cost_scanner/tests/copied_prefix.rs b/rust/src/cost_scanner/tests/copied_prefix.rs new file mode 100644 index 0000000000..23e1dccb22 --- /dev/null +++ b/rust/src/cost_scanner/tests/copied_prefix.rs @@ -0,0 +1,578 @@ +use super::*; + +fn write_copied_prefix_subagent_fixture( + sessions_root: &Path, + name: &str, + session_id: &str, + parent_id: &str, + base: DateTime, + owned: bool, +) -> PathBuf { + let day = base.with_timezone(&Local).date_naive(); + let day_dir = sessions_root + .join(day.format("%Y").to_string()) + .join(day.format("%m").to_string()) + .join(day.format("%d").to_string()); + std::fs::create_dir_all(&day_dir).unwrap(); + let path = day_dir.join(name); + let mut lines = vec![ + serde_json::json!({ + "type": "session_meta", "ordinal": 0, "timestamp": base.to_rfc3339(), + "payload": { + "id": session_id, "forked_from_id": parent_id, + "subagent_history_start_ordinal": 10, + "thread_source": "subagent", + "source": {"subagent": {"thread_spawn": {"parent_thread_id": parent_id}}} + } + }), + token_row(base, 2, [1_000, 900, 100], [0, 0, 0], "gpt-5.6-sol"), + serde_json::json!({ + "type": "turn_context", "ordinal": 10, "timestamp": base.to_rfc3339(), + "payload": {"model": "gpt-5.6-sol"} + }), + token_row( + base, + 12, + [1_000, 900, 100], + [1_000, 900, 100], + "gpt-5.6-sol", + ), + token_row( + base, + 13, + [5_000, 3_900, 500], + [5_000, 3_900, 500], + "gpt-5.6-sol", + ), + ]; + if owned { + lines.extend([ + token_row(base, 19, [5_050, 3_910, 505], [50, 10, 5], "gpt-5.6-sol"), + token_row( + base + Duration::seconds(1), + 20, + [5_070, 3_915, 510], + [20, 5, 5], + "gpt-5.6-sol", + ), + token_row( + base + Duration::seconds(2), + 21, + [5_070, 3_915, 510], + [20, 5, 5], + "gpt-5.6-sol", + ), + ]); + } + let body = lines + .into_iter() + .map(|line| line.to_string()) + .collect::>() + .join("\n") + + "\n"; + std::fs::write(&path, body).unwrap(); + path +} + +fn token_row( + timestamp: DateTime, + ordinal: i64, + total: [i64; 3], + last: [i64; 3], + model: &str, +) -> serde_json::Value { + serde_json::json!({ + "type": "event_msg", "ordinal": ordinal, "timestamp": timestamp.to_rfc3339(), + "payload": {"type": "token_count", "info": { + "model": model, + "total_token_usage": { + "input_tokens": total[0], "cached_input_tokens": total[1], "output_tokens": total[2] + }, + "last_token_usage": { + "input_tokens": last[0], "cached_input_tokens": last[1], "output_tokens": last[2] + } + }} + }) +} + +#[test] +fn copied_prefix_subagent_infers_advancing_baseline_without_parent() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "missing-parent", + Utc::now() - Duration::hours(1), + true, + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, _, cache) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(summary.input_tokens, 70); + assert_eq!(summary.cached_tokens, 15); + assert_eq!(summary.output_tokens, 10); + assert_eq!(summary.sessions_count, 1); + let usage = &cache.files[&child.to_string_lossy().to_string()]; + assert!(!usage.codex_unresolved_fork_parent); + assert!( + usage + .codex_fork_accounting_state + .as_ref() + .is_some_and(|state| state.locally_resolved) + ); + assert_eq!( + usage.days.values().next().unwrap()["gpt-5.6-sol"], + vec![70, 15, 10] + ); + + let (cached, stats, _) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(cached.input_tokens, 70); + assert!(stats.codex_history_read_paths.is_empty()); +} + +#[test] +fn copied_prefix_subagent_inherited_only_suffix_is_not_billed() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "missing-parent", + Utc::now() - Duration::hours(1), + false, + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, _, cache) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(summary.input_tokens, 0); + assert_eq!(summary.output_tokens, 0); + assert_eq!(summary.sessions_count, 0); + let usage = &cache.files[&child.to_string_lossy().to_string()]; + assert!(usage.days.is_empty()); + assert!(!usage.codex_unresolved_fork_parent); + let state = usage.codex_fork_accounting_state.as_ref().unwrap(); + assert!(state.locally_resolved); + assert!(state.inherited_totals.is_none()); + + let (cached, stats, _) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(cached.input_tokens, 0); + assert_eq!(cached.output_tokens, 0); + assert_eq!(cached.sessions_count, 0); + assert!(stats.codex_history_read_paths.is_empty()); +} + +#[test] +fn copied_prefix_subagent_prefers_validated_parent_baseline() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let parent = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + None, + base, + base, + &[1_000], + ); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "parent-id", + base + Duration::seconds(10), + true, + ); + let now = std::time::SystemTime::now(); + std::fs::OpenOptions::new() + .write(true) + .open(&child) + .unwrap() + .set_modified(now - std::time::Duration::from_secs(20)) + .unwrap(); + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = false; + let parent_size = std::fs::metadata(&parent).unwrap().len(); + let child_size = std::fs::metadata(&child).unwrap().len(); + options.codex_max_session_file_bytes = + i64::try_from(parent_size.max(child_size)).expect("fixture size fits i64"); + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (_, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + let state = cache.files[&child.to_string_lossy().to_string()] + .codex_fork_accounting_state + .as_ref() + .unwrap(); + + assert_eq!(state.inherited_totals.as_ref().unwrap().input, 1_000); + assert!(!state.locally_resolved); + assert_eq!(stats.files_seen, 2); + assert_eq!(stats.codex_read_receipt.metadata_reads, 2); + assert_eq!(stats.codex_read_receipt.history_reads, 2); + assert_eq!( + stats.codex_bytes_read, + parent_size.saturating_add(child_size), + "one bounded parse per candidate must enforce the per-file allowance" + ); + assert_eq!( + stats.codex_history_read_paths, + vec![ + parent.to_string_lossy().to_string(), + child.to_string_lossy().to_string(), + ] + ); +} + +#[test] +fn candidate_limit_counts_each_child_parent_candidate_once() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let parent = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + None, + base, + base, + &[1_000], + ); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "parent-id", + base + Duration::seconds(10), + true, + ); + let now = std::time::SystemTime::now(); + std::fs::OpenOptions::new() + .write(true) + .open(&child) + .unwrap() + .set_modified(now - std::time::Duration::from_secs(20)) + .unwrap(); + std::fs::OpenOptions::new() + .write(true) + .open(&parent) + .unwrap() + .set_modified(now - std::time::Duration::from_secs(10)) + .unwrap(); + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = false; + options.codex_candidate_limit = 1; + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (_, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(stats.files_seen, 1); + assert_eq!(stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(stats.codex_read_receipt.history_reads, 1); + assert_eq!( + stats.codex_metadata_read_paths, + vec![child.to_string_lossy().to_string()] + ); + assert_eq!( + cache.codex_pending_paths, + vec![parent.to_string_lossy().to_string()] + ); +} + +#[test] +fn cold_scan_orders_multi_level_parent_chain_before_children() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let ancestor = write_codex_fork_session_fixture( + &sessions, + "ancestor.jsonl", + "ancestor-id", + None, + base, + base, + &[1_000], + ); + let parent = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + Some("ancestor-id"), + base + Duration::seconds(10), + base + Duration::seconds(10), + &[1_500], + ); + let child = write_codex_fork_session_fixture( + &sessions, + "child.jsonl", + "child-id", + Some("parent-id"), + base + Duration::seconds(20), + base + Duration::seconds(20), + &[2_000], + ); + let now = std::time::SystemTime::now(); + for (path, age) in [(&child, 30), (&parent, 20), (&ancestor, 10)] { + std::fs::OpenOptions::new() + .write(true) + .open(path) + .unwrap() + .set_modified(now - std::time::Duration::from_secs(age)) + .unwrap(); + } + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = false; + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (_, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(stats.files_seen, 3); + assert_eq!(stats.codex_read_receipt.metadata_reads, 3); + assert_eq!(stats.codex_read_receipt.history_reads, 3); + assert_eq!( + stats.codex_history_read_paths, + vec![ + ancestor.to_string_lossy().to_string(), + parent.to_string_lossy().to_string(), + child.to_string_lossy().to_string(), + ] + ); + let parent_state = cache.files[&parent.to_string_lossy().to_string()] + .codex_fork_accounting_state + .as_ref() + .unwrap(); + let child_state = cache.files[&child.to_string_lossy().to_string()] + .codex_fork_accounting_state + .as_ref() + .unwrap(); + assert_eq!(parent_state.inherited_totals.as_ref().unwrap().input, 1_000); + assert_eq!(child_state.inherited_totals.as_ref().unwrap().input, 1_500); +} + +fn assert_unsafe_lineage_is_unresolved( + summary: &CostSummary, + stats: &CostScanStats, + cache: &CostUsageCache, + paths: &[&Path], +) { + assert_eq!(summary.sessions_count, 0); + assert_eq!(summary.input_tokens, 0); + assert_eq!( + stats.codex_read_receipt.metadata_reads, + u32::try_from(paths.len()).expect("fixture count fits u32") + ); + assert_eq!(stats.codex_read_receipt.history_reads, 0); + assert!(stats.codex_history_read_paths.is_empty()); + for path in paths { + let usage = &cache.files[&path.to_string_lossy().to_string()]; + assert!(usage.codex_unresolved_fork_parent); + assert!(usage.codex_fork_accounting_state.is_none()); + assert!(usage.days.is_empty()); + } +} + +#[test] +fn duplicate_parent_session_ids_fail_closed_with_their_child() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let first_parent = write_codex_fork_session_fixture( + &sessions, + "first-parent.jsonl", + "parent-id", + None, + base, + base, + &[1_000], + ); + let second_parent = write_codex_fork_session_fixture( + &sessions, + "second-parent.jsonl", + "parent-id", + None, + base + Duration::seconds(1), + base + Duration::seconds(1), + &[2_000], + ); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "parent-id", + base + Duration::seconds(2), + true, + ); + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = false; + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_unsafe_lineage_is_unresolved( + &summary, + &stats, + &cache, + &[&first_parent, &second_parent, &child], + ); +} + +#[test] +fn two_node_subagent_cycle_fails_closed() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let first = write_copied_prefix_subagent_fixture( + &sessions, + "first.jsonl", + "first-id", + "second-id", + base, + true, + ); + let second = write_copied_prefix_subagent_fixture( + &sessions, + "second.jsonl", + "second-id", + "first-id", + base + Duration::seconds(1), + true, + ); + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = false; + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_unsafe_lineage_is_unresolved(&summary, &stats, &cache, &[&first, &second]); +} + +#[test] +fn self_referential_subagent_fails_closed() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let session = write_copied_prefix_subagent_fixture( + &sessions, + "self-cycle.jsonl", + "self-id", + "self-id", + Utc::now() - Duration::hours(1), + true, + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_unsafe_lineage_is_unresolved(&summary, &stats, &cache, &[&session]); +} + +fn assert_cached_inference_is_replaced_when_parent_appears( + prefer_newest_codex_sessions_first: bool, +) { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let child = write_copied_prefix_subagent_fixture( + &sessions, + "child.jsonl", + "child-id", + "parent-id", + base + Duration::seconds(10), + true, + ); + let mut options = CostScanOptions::app_driven(); + options.prefer_newest_codex_sessions_first = prefer_newest_codex_sessions_first; + let scanner = CostScanner::new(7) + .with_options(options) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions.clone()]); + + let (_, _, inferred_cache) = scanner.scan_codex_detailed_with_cache(None); + let inferred_state = inferred_cache.files[&child.to_string_lossy().to_string()] + .codex_fork_accounting_state + .as_ref() + .unwrap(); + assert!(inferred_state.locally_resolved); + assert_eq!( + inferred_state.inherited_totals.as_ref().unwrap().input, + 5_000 + ); + + let parent = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + None, + base, + base, + &[1_000], + ); + let now = std::time::SystemTime::now(); + std::fs::OpenOptions::new() + .write(true) + .open(parent) + .unwrap() + .set_modified(now - std::time::Duration::from_secs(10)) + .unwrap(); + + let (_, stats, validated_cache) = scanner.scan_codex_detailed_with_cache(None); + let validated_state = validated_cache.files[&child.to_string_lossy().to_string()] + .codex_fork_accounting_state + .as_ref() + .unwrap(); + assert!(!validated_state.locally_resolved); + assert_eq!( + validated_state.inherited_totals.as_ref().unwrap().input, + 1_000 + ); + assert!( + stats + .codex_history_read_paths + .contains(&child.to_string_lossy().to_string()), + "the unchanged child must be reparsed when baseline provenance changes" + ); +} + +#[test] +fn copied_prefix_subagent_replaces_cached_inference_when_parent_is_visited_first() { + assert_cached_inference_is_replaced_when_parent_appears(true); +} + +#[test] +fn copied_prefix_subagent_replaces_cached_inference_when_child_would_be_visited_first() { + assert_cached_inference_is_replaced_when_parent_appears(false); +} diff --git a/rust/src/cost_scanner/tests/lineage_cache.rs b/rust/src/cost_scanner/tests/lineage_cache.rs new file mode 100644 index 0000000000..19ac317153 --- /dev/null +++ b/rust/src/cost_scanner/tests/lineage_cache.rs @@ -0,0 +1,507 @@ +use super::*; + +fn write_subagent( + sessions_root: &Path, + name: &str, + session_id: &str, + parent_id: &str, + timestamp: DateTime, +) -> PathBuf { + let day = timestamp.with_timezone(&Local).date_naive(); + let day_dir = sessions_root + .join(day.format("%Y").to_string()) + .join(day.format("%m").to_string()) + .join(day.format("%d").to_string()); + std::fs::create_dir_all(&day_dir).unwrap(); + let path = day_dir.join(name); + let rows = [ + serde_json::json!({ + "type": "session_meta", "ordinal": 0, "timestamp": timestamp.to_rfc3339(), + "payload": { + "id": session_id, + "forked_from_id": parent_id, + "subagent_history_start_ordinal": 10, + "thread_source": "subagent", + "source": {"subagent": {"thread_spawn": {"parent_thread_id": parent_id}}} + } + }), + lineage_token_row(timestamp, 2, 1_000, 0), + serde_json::json!({ + "type": "turn_context", "ordinal": 10, "timestamp": timestamp.to_rfc3339(), + "payload": {"model": "gpt-5.6-sol"} + }), + lineage_token_row(timestamp, 12, 1_000, 1_000), + lineage_token_row(timestamp + Duration::seconds(1), 20, 1_050, 50), + ]; + let body = rows + .into_iter() + .map(|row| row.to_string()) + .collect::>() + .join("\n") + + "\n"; + std::fs::write(&path, body).unwrap(); + path +} + +fn lineage_token_row( + timestamp: DateTime, + ordinal: i64, + total_input: i64, + last_input: i64, +) -> serde_json::Value { + serde_json::json!({ + "type": "event_msg", "ordinal": ordinal, "timestamp": timestamp.to_rfc3339(), + "payload": {"type": "token_count", "info": { + "model": "gpt-5.6-sol", + "total_token_usage": { + "input_tokens": total_input, "cached_input_tokens": 0, "output_tokens": 5 + }, + "last_token_usage": { + "input_tokens": last_input, "cached_input_tokens": 0, "output_tokens": 5 + } + }} + }) +} + +fn write_missing_ordinal_subagent( + sessions_root: &Path, + name: &str, + base: DateTime, + include_owned_usage: bool, +) -> PathBuf { + let day = base.with_timezone(&Local).date_naive(); + let day_dir = sessions_root + .join(day.format("%Y").to_string()) + .join(day.format("%m").to_string()) + .join(day.format("%d").to_string()); + std::fs::create_dir_all(&day_dir).unwrap(); + let path = day_dir.join(name); + let mut missing_ordinal = lineage_token_row(base, 11, 100, 0); + missing_ordinal.as_object_mut().unwrap().remove("ordinal"); + let tail = if include_owned_usage { + lineage_token_row(base, 12, 120, 10) + } else { + lineage_token_row(base, 12, 100, 0) + }; + let rows = [ + serde_json::json!({ + "type": "session_meta", "ordinal": 0, "timestamp": base.to_rfc3339(), + "payload": { + "id": "child-id", + "forked_from_id": "absent-parent-id", + "subagent_history_start_ordinal": 10, + "thread_source": "subagent", + "source": {"subagent": {"thread_spawn": {"parent_thread_id": "absent-parent-id"}}} + } + }), + lineage_token_row(base, 9, 100, 0), + lineage_token_row(base, 10, 100, 0), + missing_ordinal, + tail, + ]; + let body = rows + .into_iter() + .map(|row| row.to_string()) + .collect::>() + .join("\n") + + "\n"; + std::fs::write(&path, body).unwrap(); + path +} + +fn bounded_scanner(sessions: &Path, cache_root: &Path) -> CostScanner { + let mut options = CostScanOptions::app_driven(); + options.codex_candidate_limit = 1; + options.prefer_newest_codex_sessions_first = false; + CostScanner::new(7) + .with_options(options) + .with_cache_root(cache_root) + .with_sessions_dirs(vec![sessions.to_path_buf()]) +} + +fn assert_locally_inferred(cache: &CostUsageCache, path: &Path) { + let usage = &cache.files[&path.to_string_lossy().to_string()]; + assert!(!usage.codex_unresolved_fork_parent); + assert!( + usage + .codex_fork_accounting_state + .as_ref() + .is_some_and(|state| state.locally_resolved) + ); +} + +fn assert_unresolved(cache: &CostUsageCache, path: &Path) { + let usage = &cache.files[&path.to_string_lossy().to_string()]; + assert!(usage.codex_unresolved_fork_parent); + assert!(usage.days.is_empty()); + assert!(usage.codex_fork_accounting_state.is_none()); +} + +#[test] +fn replaced_parent_with_same_path_size_and_mtime_cannot_author_lineage() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let parent = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + None, + base, + base, + &[1_000], + ); + let child = write_subagent( + &sessions, + "child.jsonl", + "child-id", + "parent-id", + base + Duration::seconds(10), + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions.clone()]); + let (_, _, cache) = scanner.scan_codex_detailed_with_cache(None); + let parent_key = parent.to_string_lossy().to_string(); + let child_usage = &cache.files[&child.to_string_lossy().to_string()]; + assert!(matches!( + CodexLineagePlanner::new(&cache).decision_for_usage(&cache, child_usage), + CodexLineageDecision::ParentReady(_) + )); + + let old_identity = cache.files[&parent_key] + .codex_file_identity + .clone() + .expect("parent identity persisted"); + let old_metadata = std::fs::metadata(&parent).unwrap(); + let old_mtime = old_metadata.modified().unwrap(); + let old_size = old_metadata.len(); + let rotated = parent.with_extension("old"); + std::fs::rename(&parent, &rotated).unwrap(); + let replacement = write_codex_fork_session_fixture( + &sessions, + "parent.jsonl", + "parent-id", + None, + base, + base, + &[2_000], + ); + std::fs::OpenOptions::new() + .write(true) + .open(&replacement) + .unwrap() + .set_modified(old_mtime) + .unwrap(); + let replacement_metadata = std::fs::metadata(&replacement).unwrap(); + assert_eq!(replacement_metadata.len(), old_size); + let replacement_identity = + JsonlScanner::codex_file_identity(&replacement, &replacement_metadata) + .expect("replacement identity available"); + assert_ne!(replacement_identity, old_identity); + + assert_eq!( + CodexLineagePlanner::new(&cache).decision_for_usage(&cache, child_usage), + CodexLineageDecision::Unsafe + ); +} + +#[test] +fn missing_explicit_ordinal_keeps_subagent_cache_unresolved() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let child = write_missing_ordinal_subagent(&sessions, "child.jsonl", base, true); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, _, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(summary.input_tokens, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &child); +} + +#[test] +fn missing_ordinal_cannot_complete_zero_usage_subagent_cache() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let child = write_missing_ordinal_subagent(&sessions, "child.jsonl", base, false); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, _, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(summary.input_tokens, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &child); +} + +#[test] +fn missing_ordinal_after_local_resolution_keeps_subagent_cache_unresolved() { + use std::io::Write as _; + + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let child = write_subagent( + &sessions, + "child.jsonl", + "child-id", + "missing-parent-id", + base, + ); + let mut missing_ordinal = lineage_token_row(base + Duration::seconds(2), 21, 1_060, 10); + missing_ordinal.as_object_mut().unwrap().remove("ordinal"); + std::fs::OpenOptions::new() + .append(true) + .open(&child) + .unwrap() + .write_all(format!("{missing_ordinal}\n").as_bytes()) + .unwrap(); + let scanner = bounded_scanner(&sessions, &cache_root); + + let (summary, _, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(summary.input_tokens, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &child); +} + +#[test] +fn legacy_cache_without_file_identity_is_reparsed() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let session = write_codex_fork_session_fixture( + &sessions, + "session.jsonl", + "root-session-id", + None, + base, + base, + &[1_000], + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (_, _, _) = scanner.scan_codex_detailed_with_cache(None); + let session_key = session.to_string_lossy().to_string(); + let mut legacy_cache = JsonlScanner::load_cache(ProviderId::Codex, Some(&cache_root)); + legacy_cache + .files + .get_mut(&session_key) + .unwrap() + .codex_file_identity = None; + JsonlScanner::save_cache(ProviderId::Codex, &mut legacy_cache, Some(&cache_root)); + + let (_, stats, refreshed_cache) = scanner.scan_codex_detailed_with_cache(None); + + assert!(stats.codex_history_read_paths.contains(&session_key)); + assert!( + refreshed_cache.files[&session_key] + .codex_file_identity + .is_some() + ); +} + +#[test] +fn bounded_refresh_detects_duplicate_parent_owners_across_cache_and_candidate() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let child = write_subagent(&sessions, "child.jsonl", "child-id", "parent-id", base); + let scanner = bounded_scanner(&sessions, &cache_root); + + let (_, first_stats, first_cache) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(first_stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(first_stats.codex_read_receipt.history_reads, 1); + assert_locally_inferred(&first_cache, &child); + + let first_parent = write_codex_fork_session_fixture( + &sessions, + "parent-a.jsonl", + "parent-id", + None, + base - Duration::seconds(2), + base - Duration::seconds(2), + &[1_000], + ); + let (_, second_stats, second_cache) = scanner.scan_codex_detailed_with_cache(None); + assert_eq!(second_stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(second_stats.codex_read_receipt.history_reads, 1); + assert_locally_inferred(&second_cache, &child); + + let second_parent = write_codex_fork_session_fixture( + &sessions, + "parent-b.jsonl", + "parent-id", + None, + base - Duration::seconds(1), + base - Duration::seconds(1), + &[2_000], + ); + let (summary, third_stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(third_stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(third_stats.codex_read_receipt.history_reads, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &first_parent); + assert_unresolved(&cache, &second_parent); + assert_unresolved(&cache, &child); +} + +#[test] +fn bounded_refresh_detects_equal_timestamp_two_node_cycle() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let first = write_subagent(&sessions, "first.jsonl", "first-id", "second-id", base); + let scanner = bounded_scanner(&sessions, &cache_root); + let (_, _, first_cache) = scanner.scan_codex_detailed_with_cache(None); + assert_locally_inferred(&first_cache, &first); + + let second = write_subagent(&sessions, "second.jsonl", "second-id", "first-id", base); + let (summary, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(stats.codex_read_receipt.history_reads, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &first); + assert_unresolved(&cache, &second); +} + +#[test] +fn bounded_refresh_rejects_self_cycle_migration() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let session = write_subagent(&sessions, "self.jsonl", "self-id", "missing-id", base); + let scanner = bounded_scanner(&sessions, &cache_root); + let (_, _, first_cache) = scanner.scan_codex_detailed_with_cache(None); + assert_locally_inferred(&first_cache, &session); + + write_subagent( + &sessions, + "self.jsonl", + "self-id", + "self-id", + base + Duration::seconds(1), + ); + let (summary, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(stats.codex_read_receipt.history_reads, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&cache, &session); +} + +#[test] +fn bounded_refresh_rejects_dependent_of_locally_inferred_parent() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let base = Utc::now() - Duration::hours(1); + let parent = write_subagent(&sessions, "parent.jsonl", "parent-id", "missing-id", base); + let scanner = bounded_scanner(&sessions, &cache_root); + let (_, _, first_cache) = scanner.scan_codex_detailed_with_cache(None); + assert_locally_inferred(&first_cache, &parent); + + let dependent = write_subagent( + &sessions, + "dependent.jsonl", + "dependent-id", + "parent-id", + base + Duration::seconds(1), + ); + let (_, stats, cache) = scanner.scan_codex_detailed_with_cache(None); + + assert_eq!(stats.codex_read_receipt.metadata_reads, 1); + assert_eq!(stats.codex_read_receipt.history_reads, 0); + assert_locally_inferred(&cache, &parent); + assert_unresolved(&cache, &dependent); +} + +#[test] +fn current_refresh_scopes_unsafe_cache_invalidation_to_range_and_dependencies() { + let root = tempfile::tempdir().unwrap(); + let sessions = root.path().join("sessions"); + let cache_root = root.path().join("cache"); + let active_time = Utc::now() - Duration::hours(1); + let old_date = Local::now().date_naive() - Duration::days(30); + let old_day = old_date.format("%Y-%m-%d").to_string(); + let old_dir = sessions + .join(old_date.format("%Y").to_string()) + .join(old_date.format("%m").to_string()) + .join(old_date.format("%d").to_string()); + let mut cache = CostUsageCache::default(); + + { + let mut add_cached = |name: &str, session_id: &str, parent_id: Option<&str>| { + let path = old_dir.join(name).to_string_lossy().to_string(); + let mut usage = cached_usage_with_packed(&old_day, "gpt-5.6-sol", vec![100, 0, 5, 0]); + usage.codex_session_id = Some(session_id.to_string()); + usage.codex_forked_from_id = parent_id.map(str::to_string); + cache.files.insert(path, usage); + }; + add_cached("unrelated-a.jsonl", "unrelated-a", Some("unrelated-b")); + add_cached("unrelated-b.jsonl", "unrelated-b", Some("unrelated-a")); + add_cached("required-a.jsonl", "required-parent", None); + add_cached("required-b.jsonl", "required-parent", None); + } + JsonlScanner::save_cache(ProviderId::Codex, &mut cache, Some(&cache_root)); + + let active_child = write_codex_fork_session_fixture( + &sessions, + "active-child.jsonl", + "active-child", + Some("required-parent"), + active_time, + active_time + Duration::seconds(1), + &[1_000_000, 1_000_140], + ); + let scanner = CostScanner::new(7) + .with_options(CostScanOptions::app_driven()) + .with_cache_root(&cache_root) + .with_sessions_dirs(vec![sessions]); + + let (summary, _, refreshed) = scanner.scan_codex_detailed_with_cache(None); + let cached_path = |name: &str| old_dir.join(name).to_string_lossy().to_string(); + + for path in [ + cached_path("unrelated-a.jsonl"), + cached_path("unrelated-b.jsonl"), + ] { + let usage = refreshed + .files + .get(&path) + .expect("unrelated history retained"); + assert_eq!(usage.days[&old_day]["gpt-5.6-sol"], vec![100, 0, 5, 0]); + assert!(!usage.codex_unresolved_fork_parent); + } + for path in [ + cached_path("required-a.jsonl"), + cached_path("required-b.jsonl"), + ] { + assert_unresolved(&refreshed, Path::new(&path)); + } + assert_eq!(summary.input_tokens, 0); + assert_eq!(summary.sessions_count, 0); + assert_unresolved(&refreshed, &active_child); +}