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..d6272604004 --- /dev/null +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs @@ -0,0 +1,357 @@ +//! 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(); +}