diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 2ab2ff738a..a4a652ace0 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -1038,7 +1038,7 @@ struct CodexRetainedScan { enum CodexScanPhase { Directories { queued: VecDeque<(PathBuf, u8, u8, u64)>, - current: Option>, + current: Option>, }, Files { next_directory: usize, @@ -1050,7 +1050,18 @@ enum CodexScanPhase { struct CodexDirectoryResume { queued: VecDeque<(PathBuf, u8, u8, u64)>, - current: Option>, + current: Option>, +} + +/// Retain both the native iterator and its unfinished name witness across +/// budgeted discovery passes; neither can be reconstructed from an offset. +struct CodexDirectoryListing { + path: PathBuf, + depth: u8, + root_order: u8, + entries: std::fs::ReadDir, + fingerprint: Option, + directory_index: usize, } #[derive(Clone, Copy)] @@ -2077,6 +2088,10 @@ fn retained_scan_step( let directory_work_limit = work_limit.saturating_mul(usize::from(MAX_SCAN_DEPTH).saturating_add(2)); let mut directory_work = 0usize; + // Flat archived directories can contain thousands of non-transcript + // entries. Retain their ReadDir cursor across turns, just as for child + // directories, instead of enumerating the whole archive in one call. + let mut directory_entries = 0usize; let mut file_work = 0usize; let mut paths = Vec::new(); let mut selected_sources = Vec::new(); @@ -2183,7 +2198,14 @@ fn retained_scan_step( }, )?; } - let identity = codex_directory_witness(&path)?; + // Do not enumerate a potentially huge unsettled archive + // just to obtain its witness before the bounded iterator. + let stat = directory_stat_witness(&path)?; + let identity = stat.map(|witness| CodexDirectoryWitness { + stat: witness.stat, + entries: None, + }); + let directory_index = scan.directories.len(); scan.directories.push(CodexDirectoryIdentity { path: path.clone(), root_order, @@ -2199,19 +2221,32 @@ fn retained_scan_step( source, } })?; - *current = Some(Box::new((path, depth, root_order, listed))); + *current = Some(Box::new(CodexDirectoryListing { + path, + depth, + root_order, + entries: listed, + fingerprint: stat + .filter(|witness| !witness.settled) + .map(|_| CodexDirectoryEntryFingerprint::default()), + directory_index, + })); continue; } - let (dir, depth, root_order, listed) = current.as_deref_mut().ok_or( + let directory = current.as_deref_mut().ok_or( TranscriptIngestError::InvalidCodexDiscoveryFrontier { detail: "retained Codex directory iterator disappeared", }, )?; - match listed.next() { + if directory_entries >= directory_work_limit { + break; + } + match directory.entries.next() { Some(entry) => { + directory_entries += 1; let entry = entry.map_err(|source| TranscriptIngestError::ScanIo { operation: "read Codex transcript directory entry", - path: dir.clone(), + path: directory.path.clone(), source, })?; let file_type = @@ -2222,13 +2257,16 @@ fn retained_scan_step( path: entry.path(), source, })?; + if let Some(fingerprint) = directory.fingerprint.as_mut() { + fingerprint.observe(&entry.file_name(), file_type)?; + } // Rollout files are not structural work. Charging them // here spends the pass on names that are not child // directories, so the newest sessions are not emitted // until every older day has been listed. if file_type.is_dir() && !file_type.is_symlink() { directory_work += 1; - if *depth >= MAX_SCAN_DEPTH { + if directory.depth >= MAX_SCAN_DEPTH { return Err(TranscriptIngestError::ScanIo { operation: "traverse Codex transcript directory depth", path: entry.path(), @@ -2275,7 +2313,7 @@ fn retained_scan_step( detail: "Codex transcript directory depth overflowed", }, )?, - *root_order, + directory.root_order, 0, )); let resume_directories = CodexDirectoryResume { @@ -2298,13 +2336,27 @@ fn retained_scan_step( detail: "Codex transcript directory depth overflowed", }, )?, - *root_order, + directory.root_order, charge, )); } } } - None => *current = None, + None => { + if let Some(fingerprint) = directory.fingerprint.take() + && let Some(identity) = scan + .directories + .get_mut(directory.directory_index) + .filter(|item| { + item.path == directory.path + && item.root_order == directory.root_order + }) + .and_then(|item| item.identity.as_mut()) + { + identity.entries = Some(fingerprint.finish()); + } + *current = None; + } } } CodexScanPhase::Files { @@ -2603,6 +2655,7 @@ fn codex_directory_changed( Ok(directory_entry_fingerprint(path)? != entries) } +#[cfg(test)] fn codex_directory_witness(path: &Path) -> TranscriptIngestResult> { let Some(current) = directory_stat_witness(path)? else { return Ok(None); @@ -2665,13 +2718,66 @@ fn codex_directory_stat_identity( codex_corpus_identity(path, metadata, stat_change) } +/// Order-independent bounded-memory set witness. Directory entries are +/// unique names, so their separately domain-hashed names can be combined +/// without storing and sorting an unbounded filename list. +#[derive(Default)] +struct CodexDirectoryEntryFingerprint { + names: u64, + xor: [u8; 32], +} + +impl CodexDirectoryEntryFingerprint { + fn observe( + &mut self, + name: &std::ffi::OsStr, + file_type: std::fs::FileType, + ) -> TranscriptIngestResult<()> { + if !file_type.is_dir() + && Path::new(name).extension().and_then(|ext| ext.to_str()) != Some("jsonl") + { + return Ok(()); + } + let tag = if file_type.is_symlink() { + 2_u8 + } else if file_type.is_dir() { + 1 + } else { + 0 + }; + let mut hasher = Sha256::new(); + hasher.update(b"tracedecay-codex-directory-entry-v2"); + hasher.update([tag]); + hash_directory_name(&mut hasher, name); + let digest: [u8; 32] = hasher.finalize().into(); + for (accumulator, byte) in self.xor.iter_mut().zip(digest) { + *accumulator ^= byte; + } + self.names = self.names.checked_add(1).ok_or( + TranscriptIngestError::InvalidCodexDiscoveryFrontier { + detail: "Codex directory fingerprint entry count overflowed", + }, + )?; + Ok(()) + } + + fn finish(self) -> [u8; 32] { + let mut hasher = Sha256::new(); + // These witnesses are process-local, not durable corpus identities. + hasher.update(b"tracedecay-codex-directory-entries-v2"); + hasher.update(self.names.to_le_bytes()); + hasher.update(self.xor); + hasher.finalize().into() + } +} + 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", path: path.to_path_buf(), source, })?; - let mut names = Vec::new(); + let mut fingerprint = CodexDirectoryEntryFingerprint::default(); for entry in listed { let entry = entry.map_err(|source| TranscriptIngestError::ScanIo { operation: "fingerprint Codex transcript directory entry", @@ -2685,24 +2791,9 @@ fn directory_entry_fingerprint(path: &Path) -> TranscriptIngestResult<[u8; 32]> path: entry.path(), source, })?; - let tag = if file_type.is_symlink() { - 2_u8 - } else if file_type.is_dir() { - 1 - } else { - 0 - }; - names.push((entry.file_name(), tag)); - } - names.sort(); - let mut hasher = Sha256::new(); - hasher.update(b"tracedecay-codex-directory-entries-v1"); - for (name, tag) in names { - hasher.update([tag]); - hash_directory_name(&mut hasher, &name); - hasher.update([0xff]); + fingerprint.observe(&entry.file_name(), file_type)?; } - Ok(hasher.finalize().into()) + Ok(fingerprint.finish()) } fn hash_directory_name(hasher: &mut Sha256, name: &std::ffi::OsStr) { diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index 771f91636d..8220a29aa8 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -1230,7 +1230,7 @@ mod recent_first_discovery_tests { use crate::runtime::hosts::codex::{ CodexCorpusEpoch, CodexDiscoveryDelivery, CodexDiscoveryFrontier, CodexDiscoveryHub, CodexDiscoverySourceKey, CodexDiscoveryState, CodexExactSessionPathAuthority, - CodexIndexedPath, CodexReplayIndex, EXACT_HOOK_DISCOVERY_UNITS_PER_CALL, + CodexIndexedPath, CodexReplayIndex, CodexScanPhase, EXACT_HOOK_DISCOVERY_UNITS_PER_CALL, IDLE_FULL_VALIDATION_CYCLES, MAX_EXACT_HOOK_SESSION_REQUESTS, MAX_EXACT_HOOK_SOURCE_AUTHORITIES, MAX_SCAN_DEPTH, codex_corpus_identity, codex_directory_changed, codex_directory_stat_identity, codex_directory_witness, @@ -2675,6 +2675,46 @@ mod recent_first_discovery_tests { assert!(found, "the changed hot bucket must deliver the new rollout"); } + #[test] + fn ignored_codex_archive_entries_do_not_change_candidate_fingerprint() { + let temp = TempDir::new().unwrap(); + let archive = temp.path().join("archived_sessions"); + std::fs::create_dir_all(&archive).unwrap(); + std::fs::write(archive.join("rollout-existing.jsonl"), b"{}\n").unwrap(); + let initial = directory_entry_fingerprint(&archive).unwrap(); + + // Non-JSONL names are in the directory but not in the candidate + // set. The previous fingerprint changed for every ignored addition. + for index in 0..128 { + std::fs::write( + archive.join(format!("rollout-{index:04}.jsonl.zst")), + b"zst", + ) + .unwrap(); + } + assert_eq!(directory_entry_fingerprint(&archive).unwrap(), initial); + + std::fs::rename( + archive.join("rollout-0000.jsonl.zst"), + archive.join("rollout-renamed.jsonl.zst"), + ) + .unwrap(); + assert_eq!(directory_entry_fingerprint(&archive).unwrap(), initial); + + let eligible = archive.join("rollout-new.jsonl"); + std::fs::write(&eligible, b"{}\n").unwrap(); + assert_ne!(directory_entry_fingerprint(&archive).unwrap(), initial); + std::fs::remove_file(eligible).unwrap(); + assert_eq!(directory_entry_fingerprint(&archive).unwrap(), initial); + + std::fs::create_dir(archive.join("2026")).unwrap(); + assert_ne!( + directory_entry_fingerprint(&archive).unwrap(), + initial, + "child directories must remain part of traversal authority" + ); + } + #[test] #[cfg(unix)] fn unsettled_directory_stat_does_not_change_with_each_sample() { @@ -2828,6 +2868,130 @@ mod recent_first_discovery_tests { ); } + #[test] + fn archived_non_jsonl_entries_yield_directory_enumeration_across_windows() { + let temp = TempDir::new().unwrap(); + let archive = temp.path().join(".codex/archived_sessions"); + std::fs::create_dir_all(&archive).unwrap(); + for index in 0..64 { + std::fs::write( + archive.join(format!("rollout-{index:03}.zst")), + b"compressed", + ) + .unwrap(); + } + let admitted = archive.join("rollout-admitted.jsonl"); + std::fs::write(&admitted, b"{}\n").unwrap(); + + let source = CodexSource::with_home(temp.path()); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(1); + let mut state = CodexDiscoveryState::default(); + let mut frontier = CodexDiscoveryFrontier::initial(); + let first = retained_pass(&source, &mut state, bounds, frontier); + assert!(first.report.is_truncated()); + assert!(first.report.paths.is_empty()); + assert!( + matches!( + state.scan.as_ref().map(|scan| &scan.phase), + Some(CodexScanPhase::Directories { + current: Some(_), + .. + }) + ), + "non-transcript directory entries must not exhaust one discovery turn" + ); + frontier = first.next_frontier; + + let mut covered = BTreeSet::new(); + for _ in 0..256 { + let pass = retained_pass(&source, &mut state, bounds, frontier); + covered.extend(pass.report.paths); + frontier = pass.next_frontier; + if frontier.is_complete() { + break; + } + } + assert_eq!(covered, BTreeSet::from([admitted])); + assert!(frontier.is_complete(), "bounded pages must converge"); + let idle = retained_pass(&source, &mut state, bounds, frontier); + assert!(idle.report.paths.is_empty()); + } + + #[test] + #[cfg(unix)] + fn unsettled_flat_archive_fingerprint_is_incremental_across_budgeted_passes() { + let temp = TempDir::new().unwrap(); + let archive = temp.path().join(".codex/archived_sessions"); + std::fs::create_dir_all(&archive).unwrap(); + for index in 0..128 { + std::fs::write( + archive.join(format!("rollout-{index:04}.jsonl.zst")), + b"compressed", + ) + .unwrap(); + } + let eligible = archive.join("rollout-eligible.jsonl"); + std::fs::write(&eligible, b"{}\n").unwrap(); + let expected = directory_entry_fingerprint(&archive).unwrap(); + assert!( + !directory_stat_witness(&archive).unwrap().unwrap().settled, + "the fixture must exercise the coarse-clock name-witness path" + ); + + let source = CodexSource::with_home(temp.path()); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(1); + let mut state = CodexDiscoveryState::default(); + let mut frontier = CodexDiscoveryFrontier::initial(); + let first = retained_pass(&source, &mut state, bounds, frontier); + assert!(first.report.is_truncated()); + let scan = state.scan.as_ref().expect("unfinished structural scan"); + let CodexScanPhase::Directories { + current: Some(current), + .. + } = &scan.phase + else { + panic!("the archive must retain a live directory iterator"); + }; + assert_eq!(current.path, archive); + assert!( + scan.directories[current.directory_index] + .identity + .expect("directory stat witness") + .entries + .is_none(), + "no eager whole-archive fingerprint before bounded traversal" + ); + frontier = first.next_frontier; + + let mut covered = BTreeSet::new(); + for _ in 0..512 { + let pass = retained_pass(&source, &mut state, bounds, frontier); + covered.extend(pass.report.paths); + frontier = pass.next_frontier; + if frontier.is_complete() { + break; + } + } + assert!(frontier.is_complete(), "bounded traversal must converge"); + assert_eq!(covered, BTreeSet::from([eligible])); + let idle = state.idle.as_ref().expect("completed sweep retains witnesses"); + let archive_identity = idle + .directories + .iter() + .find(|directory| directory.path == archive) + .and_then(|directory| directory.identity) + .expect("archive directory witness"); + assert_eq!( + archive_identity.entries, + Some(expected), + "the same full candidate witness must be assembled exactly once" + ); + + let stable = retained_pass(&source, &mut state, bounds, frontier); + assert!(stable.report.paths.is_empty()); + assert!(stable.next_frontier.is_complete()); + } + /// Sweep-complete is store-durable: a fresh source reading only admission /// parse offsets (the production persist path) idles instead of restarting /// truncated-from-zero. MemoryHostAdmission is the same offset table a