diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index c75fba704f2..2ab2ff738ac 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -1056,7 +1056,7 @@ struct CodexDirectoryResume { #[derive(Clone, Copy)] struct CodexDirectoryWitness { stat: [u8; 32], - entries: [u8; 32], + entries: Option<[u8; 32]>, } #[derive(Clone)] @@ -1735,6 +1735,12 @@ impl CodexSource { } else { index + 1 }; + // The hot prefix was already checked earlier in this tick. + // Advance the rotating cursor through it without repeating + // directory stats or unsettled entry-name fingerprint scans. + if index < hot { + continue; + } if let Some(directory) = idle.directories.get(index) { if codex_directory_changed(&directory.path, directory.identity)? { changed = true; @@ -2578,11 +2584,23 @@ fn codex_directory_changed( let Some(stored) = stored else { return Ok(true); }; - // An unsettled stat proves nothing either way; the entry names do. + // A different native identity proves the directory changed, even while + // the filesystem change clock is unsettled. Do not enumerate and sort its + // entries (or miss a metadata-only change with identical entry names). + if current.stat != stored.stat { + return Ok(true); + } + // A settled, identical stat proves no directory mutation. if current.settled { - return Ok(current.stat != stored.stat); + return Ok(false); } - Ok(directory_entry_fingerprint(path)? != stored.entries) + // An older settled directory did not need an entry-name witness. + // If its stat becomes unsettled, conservatively revalidate it rather + // than assuming that newly added rollouts were already discovered. + let Some(entries) = stored.entries else { + return Ok(true); + }; + Ok(directory_entry_fingerprint(path)? != entries) } fn codex_directory_witness(path: &Path) -> TranscriptIngestResult> { @@ -2591,7 +2609,14 @@ fn codex_directory_witness(path: &Path) -> TranscriptIngestResult TranscriptIngestResult TranscriptIngestResult<[u8; 32]> { + let stat_change = match changed { + ChangeStamp::Unvouched(_) => { + let native_change_time = RewriteWitness::NATIVE + .change_time_nanos(metadata) + .map_or(0, |time| u128::from_le_bytes(time.to_le_bytes())); + ChangeStamp::Unvouched(native_change_time) + } + settled => settled, + }; + codex_corpus_identity(path, metadata, stat_change) +} + fn directory_entry_fingerprint(path: &Path) -> TranscriptIngestResult<[u8; 32]> { let listed = std::fs::read_dir(path).map_err(|source| TranscriptIngestError::ScanIo { operation: "fingerprint Codex transcript directory", diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index d40140c2b62..771f91636db 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -1218,6 +1218,7 @@ mod recent_first_discovery_tests { use serde_json::json; use tempfile::TempDir; use tracedecay_domain::ProjectId; + use tracedecay_private_fs::ChangeStamp; use tracedecay_runtime_core::native_runtime::{ LONG_POLL, NativeTask, NativeTaskOutcome, block_on, long_native_polls, }; @@ -1231,7 +1232,9 @@ mod recent_first_discovery_tests { CodexDiscoverySourceKey, CodexDiscoveryState, CodexExactSessionPathAuthority, CodexIndexedPath, CodexReplayIndex, EXACT_HOOK_DISCOVERY_UNITS_PER_CALL, IDLE_FULL_VALIDATION_CYCLES, MAX_EXACT_HOOK_SESSION_REQUESTS, - MAX_EXACT_HOOK_SOURCE_AUTHORITIES, MAX_SCAN_DEPTH, indexed_replay_pass, + MAX_EXACT_HOOK_SOURCE_AUTHORITIES, MAX_SCAN_DEPTH, codex_corpus_identity, + codex_directory_changed, codex_directory_stat_identity, codex_directory_witness, + directory_entry_fingerprint, directory_stat_witness, indexed_replay_pass, replay_index_entries_visited_for_test, reset_replay_index_entries_visited_for_test, }; use crate::runtime::ingest::quarantine::SourceSweep; @@ -2616,6 +2619,152 @@ mod recent_first_discovery_tests { ); } + #[test] + fn hot_codex_directory_is_detected_even_when_rotation_starts_on_hot_prefix() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let newest = write_dated_rollout(home, ("2026", "08", "29"), "current"); + write_dated_rollout(home, ("2026", "08", "01"), "older"); + let source = CodexSource::with_home(home); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(8); + let mut state = CodexDiscoveryState::default(); + let mut frontier = CodexDiscoveryFrontier::initial(); + let mut admitted = BTreeSet::new(); + for _ in 0..64 { + let pass = retained_pass(&source, &mut state, bounds, frontier); + admitted.extend(pass.report.paths); + frontier = pass.next_frontier; + if frontier.is_complete() { + break; + } + } + assert!(frontier.is_complete(), "the initial corpus must converge"); + assert!(admitted.contains(&newest)); + + let idle = state.idle.as_mut().expect("completed idle authority"); + let hot_index = idle + .directories + .iter() + .position(|directory| directory.path == newest.parent().unwrap()) + .expect("newest bucket must be retained"); + assert!( + hot_index < super::super::IDLE_HOT_DIRECTORIES, + "the newly changed bucket must be in the hot prefix" + ); + idle.next = hot_index; + + let added = newest.parent().unwrap().join("rollout-after-idle.jsonl"); + std::fs::write(&added, b"{}").unwrap(); + spin_until_jsonl_change_settled(&added); + let detection = retained_pass(&source, &mut state, bounds, frontier); + assert!( + !detection.next_frontier.is_complete(), + "the hot probe must invalidate a changed bucket before rotation skips its duplicate" + ); + + frontier = detection.next_frontier; + let mut found = detection.report.paths.contains(&added); + for _ in 0..64 { + let pass = retained_pass(&source, &mut state, bounds, frontier); + found |= pass.report.paths.contains(&added); + frontier = pass.next_frontier; + if found { + break; + } + } + assert!(found, "the changed hot bucket must deliver the new rollout"); + } + + #[test] + #[cfg(unix)] + fn unsettled_directory_stat_does_not_change_with_each_sample() { + let temp = TempDir::new().unwrap(); + let bucket = temp.path().join(".codex/sessions/2026/11/08"); + std::fs::create_dir_all(&bucket).unwrap(); + std::fs::write(bucket.join("rollout-existing.jsonl"), b"{}\n").unwrap(); + let metadata = std::fs::metadata(&bucket).unwrap(); + let first_stamp = ChangeStamp::Unvouched(1); + let next_stamp = ChangeStamp::Unvouched(2); + + // Each native Unvouched stamp is deliberately different. The raw + // file-content identity must not accidentally become stable. + let first_sample = codex_corpus_identity(&bucket, &metadata, first_stamp).unwrap(); + let next_sample = codex_corpus_identity(&bucket, &metadata, next_stamp).unwrap(); + assert_ne!(first_sample, next_sample); + + // Directory change detection, unlike file-content authority, must + // not force a fresh traversal on every sample of identical metadata. + let first = codex_directory_stat_identity(&bucket, &metadata, first_stamp).unwrap(); + let second = codex_directory_stat_identity(&bucket, &metadata, next_stamp).unwrap(); + assert_eq!( + first, second, + "identical native directory metadata must survive repeated unsettled probes" + ); + + // The settled transition still invalidates an unsettled snapshot: + // a prior coarse-window name check cannot vouch for a later write. + let settled = + codex_directory_stat_identity(&bucket, &metadata, ChangeStamp::Settled(0)).unwrap(); + assert_ne!(first, settled); + } + + #[test] + #[cfg(unix)] + fn unsettled_bucket_native_change_invalidates_even_with_same_entry_names() { + let temp = TempDir::new().unwrap(); + let bucket = temp.path().join(".codex/sessions/2026/11/08"); + std::fs::create_dir_all(&bucket).unwrap(); + std::fs::write(bucket.join("rollout-existing.jsonl"), b"{}\n").unwrap(); + + // A recently changed bucket uses an entry-name witness when the + // clock is still coarse. Model a different previously observed stat + // (such as a directory metadata update) with identical entry names. + let mut stored = codex_directory_witness(&bucket).unwrap().unwrap(); + assert!( + !directory_stat_witness(&bucket).unwrap().unwrap().settled, + "this regression requires the unsettled-clock branch" + ); + assert!(stored.entries.is_some()); + assert_eq!( + stored.entries, + Some(directory_entry_fingerprint(&bucket).unwrap()) + ); + stored.stat[0] ^= 1; + + assert!( + codex_directory_changed(&bucket, Some(stored)).unwrap(), + "a changed native identity must invalidate even when the entry-name fingerprint is unchanged" + ); + } + + #[test] + #[cfg(unix)] + fn settled_codex_bucket_skips_name_scans_and_detects_new_rollouts() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let historical = write_dated_rollout(home, ("2025", "11", "01"), "settled"); + let bucket = historical.parent().unwrap(); + + // The rollout fixture waits for the native change-time witness to + // settle, so the parent directory is also outside that time quantum. + assert!( + directory_stat_witness(bucket).unwrap().unwrap().settled, + "historical directory must have a settled change-time witness" + ); + let witness = codex_directory_witness(bucket).unwrap().unwrap(); + assert!( + witness.entries.is_none(), + "a settled historical bucket must not enumerate names twice" + ); + assert!(!codex_directory_changed(bucket, Some(witness)).unwrap()); + + std::fs::write(bucket.join("rollout-new.jsonl"), b"{}\n").unwrap(); + assert!( + codex_directory_changed(bucket, Some(witness)).unwrap(), + "a new rollout must invalidate the cheap settled-directory witness" + ); + } + #[test] fn codex_validation_retains_its_cursor_without_scanning_the_corpus_in_one_call() { let temp = TempDir::new().unwrap();