From 408fdb7ec143e31e6ced01b75a2bcff1eca847ed Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:19:57 -0800 Subject: [PATCH 1/2] perf(sessions): skip Codex scope resolution for empty pages Refs #4374 Keep canonical cursor validation, source rewrite completion and cancellation on the admission path, but do not dispatch Git scope resolution when no native frame can consume its result. Do not retain an empty-page verdict. Add real JSONL/project/profile regressions for cold and warm re-offers, resolution work counts, unchanged accepted identities, later cwd changes, appended records and cancellation. Add a dedicated scope-resolution span. Bazel execution is pending hosted CI; no local Rust pass is claimed. --- .../src/runtime/hosts/codex/observation.rs | 20 + .../hosts/codex/observation/reoffer_tests.rs | 354 ++++++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index cf3816de8df..3efd30be4df 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -45,6 +45,8 @@ use tracedecay_runtime_core::resident_memory::ProcessSharedMemoryReservationV1; #[cfg(test)] mod meta_cache_tests; #[cfg(test)] +mod reoffer_tests; +#[cfg(test)] mod retired_source_tests; #[cfg(test)] mod rollout_source_tests; @@ -403,6 +405,11 @@ impl CodexObservationAdmission<'_> { /// Prepare this bounded page's context and scope decisions on the blocking /// pool. The page owns its reservation throughout preparation; only these /// decisions, not filesystem work, are used by frame callbacks and retries. + #[tracing::instrument( + name = "sessions.observation.codex.resolve_scope", + level = "trace", + skip_all + )] async fn resolve_scope( &self, matchers: &ProjectRootMatcherCache, @@ -410,6 +417,19 @@ impl CodexObservationAdmission<'_> { meta: &super::CodexMeta, scan: JsonlObservationScan, ) -> TranscriptIngestResult<(CodexAdmissionState, u64)> { + // Only native frames consume scope verdicts. An exhausted source must + // still finish canonical cursor/rewrite processing, but resolving Git + // identities (and dispatching blocking work) cannot affect that result. + // Do not cache a verdict here: a later append must resolve live scope. + if scan.frame_bytes().next().is_none() { + return Ok(( + CodexAdmissionState { + context: CodexContextState::from_meta(meta), + scope_verdicts: Arc::new(HashMap::new()), + }, + 0, + )); + } let roots = self.scope_roots(); let profile = crate::runtime::transcript_source_profile(); let matchers = matchers.clone(); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs new file mode 100644 index 00000000000..4d8966d8b09 --- /dev/null +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs @@ -0,0 +1,354 @@ +//! Empty Codex re-offers must not pay repository resolution costs or cache a +//! scope decision that can leak into the next nonempty admission page. + +use std::io::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::Instant; + +use serde_json::{Value, json}; +use tempfile::TempDir; +use tracedecay_domain::{ObservationScopeV1, ProjectId}; +use tracedecay_runtime_core::config::ProfileRoot; +use tracedecay_runtime_core::git_discovery::{ + GitRepositoryIdentityOutcome, discover_repository_identity_cli_first, +}; +use tracedecay_runtime_core::path_safety::plain_git_args; + +use super::{ + CodexJsonlAdmissionProgress, CodexObservationAdmission, codex_observation_source_v2, + try_admit_codex_jsonl_observations, +}; +use crate::admission::test_support::MemoryHostAdmission; +use crate::observation::ObservationCancellation; +use crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority; +use crate::runtime::shared::ProjectRootMatcherCache; +use crate::runtime::with_transcript_source_profile; + +static IDENTITY_RESOLUTIONS: Mutex> = Mutex::new(Vec::new()); + +fn recording_identity(path: &Path) -> GitRepositoryIdentityOutcome { + IDENTITY_RESOLUTIONS + .lock() + .unwrap() + .push(path.to_path_buf()); + discover_repository_identity_cli_first(path) +} + +fn resolution_count(root: &Path) -> usize { + IDENTITY_RESOLUTIONS + .lock() + .unwrap() + .iter() + .filter(|path| path.starts_with(root)) + .count() +} + +fn message(session: &str, item: &str) -> Value { + json!({ + "timestamp": "2026-09-04T12:00:01.004Z", + "type": "event_msg", + "payload": { + "type": "item_completed", + "thread_id": session, + "turn_id": item, + "item": { + "type": "UserMessage", + "id": item, + "content": [{"type": "text", "text": item}] + } + } + }) +} + +fn write_rollout(path: &Path, cwd: &Path, session: &str) -> u64 { + let mut file = std::fs::File::create(path).unwrap(); + writeln!( + file, + "{}", + json!({ + "timestamp": "2026-09-04T12:00:00.000Z", + "type": "session_meta", + "payload": {"id": session, "cwd": cwd} + }) + ) + .unwrap(); + writeln!(file, "{}", message(session, "initial-item")).unwrap(); + file.metadata().unwrap().len() +} + +fn append_in_scope(path: &Path, cwd: &Path, session: &str, item: &str) -> u64 { + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(path) + .unwrap(); + writeln!( + file, + "{}", + json!({"type": "turn_context", "payload": {"cwd": cwd, "model": "scope-model"}}) + ) + .unwrap(); + writeln!(file, "{}", message(session, item)).unwrap(); + file.metadata().unwrap().len() +} + +struct Fixture { + _temp: TempDir, + root: PathBuf, + project: PathBuf, + outside: PathBuf, + profile: ProfileRoot, + project_id: ProjectId, + project_admission: MemoryHostAdmission, + profile_admission: MemoryHostAdmission, +} + +impl Fixture { + fn new() -> Self { + install_test_shared_jsonl_preparation_authority(); + let temp = TempDir::new().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let project = root.join("project"); + let outside = root.join("outside"); + for repository in [&project, &outside] { + std::fs::create_dir_all(repository.join("src")).unwrap(); + let output = std::process::Command::new( + tracedecay_runtime_core::git::try_git_program().unwrap(), + ) + .args(plain_git_args(&["init", "--quiet", "--initial-branch=main"])) + .current_dir(repository) + .output() + .unwrap(); + assert!( + output.status.success(), + "git init: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + Self { + profile: ProfileRoot::under_home(root.join("owner")), + _temp: temp, + root, + project, + outside, + project_id: ProjectId::new("project.codex-idle-reoffer").unwrap(), + project_admission: MemoryHostAdmission::default(), + profile_admission: MemoryHostAdmission::default(), + } + } + + async fn admit( + &self, + path: &Path, + cache: &ProjectRootMatcherCache, + cancellation: &ObservationCancellation, + ) -> [CodexJsonlAdmissionProgress; 2] { + let project = try_admit_codex_jsonl_observations( + path, + CodexObservationAdmission::Project { + root: &self.project, + project_id: self.project_id.clone(), + }, + &self.project_admission, + None, + None, + cancellation, + cache, + ) + .await + .unwrap(); + let profile = try_admit_codex_jsonl_observations( + path, + CodexObservationAdmission::Profile { + session_id: None, + registered_roots: std::slice::from_ref(&self.project), + }, + &self.profile_admission, + None, + None, + cancellation, + cache, + ) + .await + .unwrap(); + [project, profile] + } +} + +fn assert_unchanged(progress: CodexJsonlAdmissionProgress, len: u64) { + assert!(progress.resumed); + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, len); + assert_eq!(progress.bytes_consumed, 0); + assert_eq!(progress.frames_decoded, 0); + assert_eq!(progress.frames_persisted, 0); + assert_eq!(progress.prior_context_bytes, 0); +} + +#[test] +fn unchanged_corpus_reoffers_skip_cold_scope_resolution() { + let fixture = Fixture::new(); + let mut rollouts = Vec::new(); + for index in 0..16 { + let repository = if index % 2 == 0 { + &fixture.project + } else { + &fixture.outside + }; + let cwd = repository.join(format!("cwd-{index}")); + std::fs::create_dir_all(&cwd).unwrap(); + let path = fixture.root.join(format!("rollout-{index}.jsonl")); + let len = write_rollout(&path, &cwd, &format!("reoffer-session-{index}")); + rollouts.push((path, len)); + } + tracedecay_runtime_core::native_runtime::block_on(with_transcript_source_profile( + fixture.profile.clone(), + async { + let cancellation = ObservationCancellation::default(); + let initial = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + for (path, len) in &rollouts { + for progress in fixture.admit(path, &initial, &cancellation).await { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, *len); + assert_eq!(progress.bytes_consumed, *len); + } + } + let stores = [&fixture.project_admission, &fixture.profile_admission]; + assert_eq!(stores.map(|store| store.observations().len()), [16, 16]); + let captures = stores.map(MemoryHostAdmission::capture_call_counts); + let retained = stores.map(|store| { + store + .observations() + .iter() + .map(|record| { + ( + record.observation().observation_id().clone(), + record.committed_cursor().clone(), + ) + }) + .collect::>() + }); + let before = resolution_count(&fixture.root); + assert!(before >= rollouts.len(), "the resolver probe must be live"); + drop(initial); + + // A cold process cache cannot hide redundant resolution on the + // first empty re-offer. Later rounds also exercise the warm path. + let cache = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + let started = Instant::now(); + let mut reoffers = 0; + for _ in 0..3 { + for (path, len) in &rollouts { + for progress in fixture.admit(path, &cache, &cancellation).await { + assert_unchanged(progress, *len); + reoffers += 1; + } + } + } + let resolutions = resolution_count(&fixture.root) - before; + tracing::info!( + reoffers, + git_identity_resolutions = resolutions, + wall_seconds = started.elapsed().as_secs_f64(), + "unchanged Codex corpus re-offer measurement" + ); + assert_eq!(reoffers, 96); + assert_eq!(resolutions, 0, "empty pages must not resolve Git scope"); + assert_eq!(stores.map(MemoryHostAdmission::capture_call_counts), captures); + for (store, retained) in stores.into_iter().zip(retained) { + let current = store + .observations() + .iter() + .map(|record| { + ( + record.observation().observation_id().clone(), + record.committed_cursor().clone(), + ) + }) + .collect::>(); + assert_eq!(current, retained, "accepted source identity must not change"); + } + }, + )) + .unwrap(); +} + +#[test] +fn append_after_empty_reoffer_rechecks_scope_and_source_identity() { + let fixture = Fixture::new(); + let path = fixture.root.join("rollout-live.jsonl"); + let session = "reoffer-live-session"; + let len = write_rollout(&path, &fixture.project.join("src"), session); + tracedecay_runtime_core::native_runtime::block_on(with_transcript_source_profile( + fixture.profile.clone(), + async { + let cancellation = ObservationCancellation::default(); + let initial = ProjectRootMatcherCache::default(); + let admitted = fixture.admit(&path, &initial, &cancellation).await; + assert_eq!(admitted[0].frames_persisted, 2); + assert_eq!(admitted[1].frames_persisted, 0); + drop(initial); + + let cache = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert_unchanged(progress, len); + } + assert_eq!(resolution_count(&fixture.root), 0); + + let new_len = append_in_scope( + &path, + &fixture.outside.join("src"), + session, + "outside-item", + ); + let appended = fixture.admit(&path, &cache, &cancellation).await; + assert_eq!(appended[0].frames_persisted, 0); + assert!(appended[1].frames_persisted > 0); + for progress in appended { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, new_len); + assert_eq!(progress.bytes_consumed, new_len - len); + } + assert!(resolution_count(&fixture.root) > 0); + let expected_source = codex_observation_source_v2(session).unwrap(); + let profile_records = fixture.profile_admission.observations(); + assert!(!profile_records.is_empty()); + for record in &profile_records { + assert_eq!(record.observation().source(), &expected_source); + assert_eq!(record.observation().scope(), &ObservationScopeV1::Profile); + } + assert_eq!(fixture.project_admission.observations().len(), 2); + + let final_len = append_in_scope( + &path, + &fixture.project.join("src"), + session, + "returned-item", + ); + let returned = fixture.admit(&path, &cache, &cancellation).await; + assert!(returned[0].frames_persisted > 0); + assert_eq!(returned[1].frames_persisted, 0); + for progress in returned { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, final_len); + assert_eq!(progress.bytes_consumed, final_len - new_len); + } + assert_eq!( + fixture.profile_admission.observations().len(), + profile_records.len() + ); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert_unchanged(progress, final_len); + } + + cancellation.cancel(); + let captures = fixture.project_admission.capture_call_counts(); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert!(progress.source_deferred, "cancellation is not convergence"); + assert_eq!(progress.frames_persisted, 0); + } + assert_eq!(fixture.project_admission.capture_call_counts(), captures); + }, + )) + .unwrap(); +} From 95d7bcf69c6f1e1f3e20078ae4f8406cb3179f1e Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:16:39 -0800 Subject: [PATCH 2/2] style(sessions): match pinned Codex reoffer test formatting Apply the five exact Bazel rustfmt diagnostics to the focused Codex reoffer regressions without changing tested behavior. Refs #4374 --- .../hosts/codex/observation/reoffer_tests.rs | 29 ++++++++++--------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs index 4d8966d8b09..d6272604004 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs @@ -78,10 +78,7 @@ fn write_rollout(path: &Path, cwd: &Path, session: &str) -> u64 { } fn append_in_scope(path: &Path, cwd: &Path, session: &str, item: &str) -> u64 { - let mut file = std::fs::OpenOptions::new() - .append(true) - .open(path) - .unwrap(); + let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap(); writeln!( file, "{}", @@ -115,7 +112,11 @@ impl Fixture { let output = std::process::Command::new( tracedecay_runtime_core::git::try_git_program().unwrap(), ) - .args(plain_git_args(&["init", "--quiet", "--initial-branch=main"])) + .args(plain_git_args(&[ + "init", + "--quiet", + "--initial-branch=main", + ])) .current_dir(repository) .output() .unwrap(); @@ -254,7 +255,10 @@ fn unchanged_corpus_reoffers_skip_cold_scope_resolution() { ); assert_eq!(reoffers, 96); assert_eq!(resolutions, 0, "empty pages must not resolve Git scope"); - assert_eq!(stores.map(MemoryHostAdmission::capture_call_counts), captures); + assert_eq!( + stores.map(MemoryHostAdmission::capture_call_counts), + captures + ); for (store, retained) in stores.into_iter().zip(retained) { let current = store .observations() @@ -266,7 +270,10 @@ fn unchanged_corpus_reoffers_skip_cold_scope_resolution() { ) }) .collect::>(); - assert_eq!(current, retained, "accepted source identity must not change"); + assert_eq!( + current, retained, + "accepted source identity must not change" + ); } }, )) @@ -295,12 +302,8 @@ fn append_after_empty_reoffer_rechecks_scope_and_source_identity() { } assert_eq!(resolution_count(&fixture.root), 0); - let new_len = append_in_scope( - &path, - &fixture.outside.join("src"), - session, - "outside-item", - ); + let new_len = + append_in_scope(&path, &fixture.outside.join("src"), session, "outside-item"); let appended = fixture.admit(&path, &cache, &cancellation).await; assert_eq!(appended[0].frames_persisted, 0); assert!(appended[1].frames_persisted > 0);