From de1412c35f208f8d4d9a45ed531cc3e5289e928c Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 1 Oct 2026 12:05:44 +0000 Subject: [PATCH] perf(sessions): skip codex rescans for unchanged rollouts An unchanged resumed rollout rebuilt its prior context by re-reading the file up to its cursor, and resolved its scope matcher (a git spawn) on every pass, even though it admitted no frame. Past 512 rollouts the prior-context cache thrashed, so every pass re-read every rollout. Rebuild prior context only when the scan has a frame, and resolve the scope matcher lazily on the first frame that consults it. --- .../src/runtime/hosts/codex.rs | 2 + .../src/runtime/hosts/codex/context.rs | 37 +++++++++++++ .../src/runtime/hosts/codex/observation.rs | 13 +++-- .../jsonl_observation_admission/tests.rs | 55 +++++++++++++++++++ 4 files changed, 103 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 18b613b3a7..67c5ca7e75 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -90,6 +90,8 @@ use crate::runtime::source::{ jsonl_file_change_token, }; +#[cfg(test)] +pub(crate) use context::{evict_prior_context_for_test, prior_context_scan_count_for_test}; #[cfg(test)] pub(crate) use meta::session_meta_read_count_for_test; pub use meta::{CodexMeta, session_meta_from_record, turn_context_from_record}; diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/context.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/context.rs index 13a03b9ecf..8033f18292 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/context.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/context.rs @@ -36,6 +36,15 @@ impl CodexContextState { let Ok(mut file) = std::fs::File::open(path) else { return Self::from_meta(meta); }; + #[cfg(test)] + { + *PRIOR_CONTEXT_SCANS + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entry(path.to_path_buf()) + .or_default() += 1; + } let generation = prior_context_generation(&file); let (mut state, mut offset) = match generation .and_then(|generation| cached_prior_context(path, generation, before_offset)) @@ -141,6 +150,34 @@ struct PriorContextCache { static PRIOR_CONTEXT_CACHE: OnceLock> = OnceLock::new(); +#[cfg(test)] +static PRIOR_CONTEXT_SCANS: OnceLock>> = OnceLock::new(); + +/// Rollout opens that rebuilt `path`'s prior context. +#[cfg(test)] +pub(crate) fn prior_context_scan_count_for_test(path: &Path) -> usize { + PRIOR_CONTEXT_SCANS + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(path) + .copied() + .unwrap_or_default() +} + +/// Evicts `path`'s resumable context, as a full cache does for every rollout +/// past its capacity. +#[cfg(test)] +pub(crate) fn evict_prior_context_for_test(path: &Path) { + if let Some(cache) = PRIOR_CONTEXT_CACHE.get() { + cache + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entries + .remove(path); + } +} + fn prior_context_generation(file: &std::fs::File) -> Option { use std::hash::{Hash, Hasher}; let meta = file.metadata().ok()?; diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index 657bf4f1eb..8e1caf0a9c 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -622,7 +622,10 @@ async fn admit_codex_jsonl_page( cancellation, } = context; let scope = admission_scope.scope(); - let scope_matcher = admission_scope.scope_matcher(); + // Resolving the scope runs git for the root. An unchanged rollout admits + // no frame and must not pay it on every pass. + let resolved_scope = OnceLock::new(); + let scope_matcher = || resolved_scope.get_or_init(|| admission_scope.scope_matcher()); let source_location_path = admission_scope.projection_project_path(Some(meta.cwd.as_path())); let source_location = CodexObservationLocation { project_path: source_location_path, @@ -650,7 +653,9 @@ async fn admit_codex_jsonl_page( scan.frame_bytes().all(|frame| !frame.is_empty()), "shared JSONL admission never exposes empty native frames" ); - let context = if scan.resumed { + // The prior context is rebuilt by reading the rollout up to its + // cursor; a scan with no new frame never consults it. + let context = if scan.resumed && scan.frame_bytes().next().is_some() { CodexContextState::scan_prior(path, scan.start_offset, meta) } else { CodexContextState::from_meta(meta) @@ -669,7 +674,7 @@ async fn admit_codex_jsonl_page( // decode per frame to reach it. let in_scope = *state .scope_verdict - .get_or_insert_with(|| scope_matcher.accepts(state.context.cwd.as_deref())); + .get_or_insert_with(|| scope_matcher().accepts(state.context.cwd.as_deref())); if !in_scope && !hints.may_change_codex_context { return Ok(JsonlFrameAdmission::non_durable_before_decode( ObservationCoverageReason::OutOfScope, @@ -687,7 +692,7 @@ async fn admit_codex_jsonl_page( } if !*state .scope_verdict - .get_or_insert_with(|| scope_matcher.accepts(state.context.cwd.as_deref())) + .get_or_insert_with(|| scope_matcher().accepts(state.context.cwd.as_deref())) { non_durable_reason = Some(ObservationCoverageReason::OutOfScope); return Err(ObservationRecordParseErrorV1::NormalizationFailed); diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 0fd2389032..24c9d968af 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1334,6 +1334,61 @@ async fn content_refusals_cover_past_so_the_stream_converges() { ); } +#[tokio::test] +async fn an_unchanged_resumed_rollout_does_not_rebuild_its_prior_context() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + let first = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("first pass"); + assert_eq!(first.bytes_consumed, len); + + crate::runtime::hosts::codex::evict_prior_context_for_test(&path); + let replay = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("unchanged pass"); + assert_eq!(replay.bytes_consumed, 0); + assert_eq!(replay.frames_persisted, 0); + assert_eq!( + crate::runtime::hosts::codex::prior_context_scan_count_for_test(&path), + 0, + "an unchanged rollout admits no frame, so it must not be re-read for its context" + ); + + let appended = json!({ + "timestamp": "2026-01-01T00:00:02.000Z", + "type": "event_msg", + "payload": {"type": "user_message", "message": "appended message"} + }) + .to_string() + + "\n"; + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + std::io::Write::write_all(&mut file, appended.as_bytes()).unwrap(); + drop(file); + let resumed = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("appended pass"); + assert!( + resumed.resumed, + "the appended pass resumes at the committed cursor" + ); + assert_eq!( + resumed.bytes_consumed, + u64::try_from(appended.len()).unwrap() + ); + assert_eq!( + crate::runtime::hosts::codex::prior_context_scan_count_for_test(&path), + 1 + ); + assert_eq!(resumed.frames_persisted, 1); +} + #[tokio::test] async fn codex_session_meta_prefix_is_decoded_once_across_consumers() { // The shared metadata cache retains entries up to