From 835291a360a706f86a1b354c3b3f3de759556b88 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:13:22 -0800 Subject: [PATCH 1/4] perf(sessions): bound non-transcript Codex directory enumeration Refs #4756 A flat archived directory with many .zst/non-JSONL entries previously consumed its entire ReadDir iterator in one structural scan turn, outside the directory-work budget. Charge directory entries independently of child directory traversal while retaining the iterator across turns. Add a real archived-directory regression requiring the first scan to yield in the directory phase, then converge on exactly the eligible JSONL path. Partial performance repair only; durable restart continuation and affected host CPU/write-latency acceptance remain outstanding. --- .../src/runtime/hosts/codex.rs | 8 ++++ .../src/runtime/hosts/codex/tests.rs | 48 ++++++++++++++++++- 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 2ab2ff738a..ab2c2ee3f1 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -2077,6 +2077,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(); @@ -2207,8 +2211,12 @@ fn retained_scan_step( detail: "retained Codex directory iterator disappeared", }, )?; + if directory_entries >= directory_work_limit { + break; + } match listed.next() { Some(entry) => { + directory_entries += 1; let entry = entry.map_err(|source| TranscriptIngestError::ScanIo { operation: "read Codex transcript directory entry", path: dir.clone(), diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index 771f91636d..aadcc61323 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, @@ -2828,6 +2828,52 @@ 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()); + } + /// 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 From 0390a59ed2292c9a7c06c8628772e3826b631d6e Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:37:18 -0800 Subject: [PATCH 2/4] perf(sessions): exclude ignored archive names from Codex probes Refs #4756 Keep unsettled directory name witnesses scoped to traversable child buckets and JSONL candidate paths. Compressed rollouts and other ignored file churn must not independently invalidate the candidate witness. Cover ignored additions and renames, candidate admission, and child bucket discovery with a real-filesystem regression. --- .../src/runtime/hosts/codex.rs | 11 +++++- .../src/runtime/hosts/codex/tests.rs | 36 +++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index ab2c2ee3f1..26ad74a7e7 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -2693,6 +2693,15 @@ fn directory_entry_fingerprint(path: &Path) -> TranscriptIngestResult<[u8; 32]> path: entry.path(), source, })?; + // Only real subdirectories and potential JSONL candidates affect + // discovery. Ignored compressed archives must not change the name + // witness when the coarse native directory timestamp stays equal. + let name = entry.file_name(); + if !file_type.is_dir() + && Path::new(&name).extension().and_then(|ext| ext.to_str()) != Some("jsonl") + { + continue; + } let tag = if file_type.is_symlink() { 2_u8 } else if file_type.is_dir() { @@ -2700,7 +2709,7 @@ fn directory_entry_fingerprint(path: &Path) -> TranscriptIngestResult<[u8; 32]> } else { 0 }; - names.push((entry.file_name(), tag)); + names.push((name, tag)); } names.sort(); let mut hasher = Sha256::new(); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index aadcc61323..12b9b70e1d 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -2675,6 +2675,42 @@ 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() { From b71eb2bfee4e1aee8f7c22f0f74cc97cc76248ab Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:25:55 -0800 Subject: [PATCH 3/4] style(sessions): match pinned rustfmt on Codex regressions Refs #4756 Apply the two exact formatting hunks emitted by Bazel CI; preserve every behavioral test assertion. --- .../src/runtime/hosts/codex/tests.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index 12b9b70e1d..efaf61a00f 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -2686,7 +2686,11 @@ mod recent_first_discovery_tests { // 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(); + std::fs::write( + archive.join(format!("rollout-{index:04}.jsonl.zst")), + b"zst", + ) + .unwrap(); } assert_eq!(directory_entry_fingerprint(&archive).unwrap(), initial); @@ -2870,8 +2874,11 @@ mod recent_first_discovery_tests { 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(); + 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(); From 3912dd4255e334b706294dfc8c9461cf78318b00 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:20:37 -0800 Subject: [PATCH 4/4] perf(sessions): bound Codex name witness to discovery slices Refs #4756 --- .../src/runtime/hosts/codex.rs | 150 +++++++++++++----- .../src/runtime/hosts/codex/tests.rs | 75 +++++++++ 2 files changed, 187 insertions(+), 38 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 26ad74a7e7..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)] @@ -2187,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, @@ -2203,10 +2221,19 @@ 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", }, @@ -2214,12 +2241,12 @@ fn retained_scan_step( if directory_entries >= directory_work_limit { break; } - match listed.next() { + 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 = @@ -2230,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(), @@ -2283,7 +2313,7 @@ fn retained_scan_step( detail: "Codex transcript directory depth overflowed", }, )?, - *root_order, + directory.root_order, 0, )); let resume_directories = CodexDirectoryResume { @@ -2306,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 { @@ -2611,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); @@ -2673,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", @@ -2693,33 +2791,9 @@ fn directory_entry_fingerprint(path: &Path) -> TranscriptIngestResult<[u8; 32]> path: entry.path(), source, })?; - // Only real subdirectories and potential JSONL candidates affect - // discovery. Ignored compressed archives must not change the name - // witness when the coarse native directory timestamp stays equal. - let name = entry.file_name(); - if !file_type.is_dir() - && Path::new(&name).extension().and_then(|ext| ext.to_str()) != Some("jsonl") - { - continue; - } - let tag = if file_type.is_symlink() { - 2_u8 - } else if file_type.is_dir() { - 1 - } else { - 0 - }; - names.push((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 efaf61a00f..8220a29aa8 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -2917,6 +2917,81 @@ mod recent_first_discovery_tests { 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