From 5f5a3652fc2c315bb17d3cd828f627fe8527a0b3 Mon Sep 17 00:00:00 2001 From: "zackary.l.jackson" Date: Tue, 20 Oct 2026 20:12:57 +0000 Subject: [PATCH 1/4] fix(sessions): skip Codex replay a complete frontier covers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../src/runtime/hosts/codex.rs | 14 +++- .../src/runtime/hosts/codex/tests.rs | 70 +++++++++++++++++++ 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index e32f0c0515..763c146ad5 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -579,9 +579,17 @@ impl CodexDiscoveryHub { detail: "Codex replay index disappeared before delivery", }, )?; - let effective_position = (consumer_generation == index.generation) - .then_some(consumer_position) - .flatten(); + // A complete durable frontier at the index's epoch already + // covers every indexed path (a restarted daemon's later + // consumers land here), so its replay starts past the end. + let effective_position = + if frontier.is_complete() && frontier == index.frontier { + index.paths.last().cloned() + } else { + (consumer_generation == index.generation) + .then_some(consumer_position) + .flatten() + }; let index_generation = index.generation; let lower = effective_position .as_ref() diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index e00702e4ef..9e797aaf73 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -1713,6 +1713,76 @@ mod recent_first_discovery_tests { assert_eq!(index.files_considered, 73); } + #[tokio::test] + async fn a_late_consumer_whose_durable_frontier_covers_the_corpus_replays_nothing() { + crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let rollouts = (0..21) + .map(|index| { + write_dated_rollout(home, ("2026", "08", "23"), &format!("covered-{index:03}")) + }) + .collect::>(); + let settled = rollouts + .iter() + .all(|path| spin_until_jsonl_change_settled(path)); + let source = CodexSource::with_home(home); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(8); + + let previous_process = CodexDiscoveryHub::default(); + previous_process.register("project", Some(home)); + previous_process.register("profile", Some(home)); + let (_, project_frontier) = drain_hub_consumer( + &previous_process, + "project", + &source, + bounds, + CodexDiscoveryFrontier::initial(), + ) + .await; + let (_, profile_frontier) = drain_hub_consumer( + &previous_process, + "profile", + &source, + bounds, + CodexDiscoveryFrontier::initial(), + ) + .await; + assert_eq!(project_frontier, profile_frontier); + + // A restarted daemon registers its consumers one by one, so every + // consumer after the first joins a hub whose shared scan already began. + let restarted = CodexDiscoveryHub::default(); + restarted.register("project", Some(home)); + let (project_paths, _) = + drain_hub_consumer(&restarted, "project", &source, bounds, project_frontier).await; + restarted.register("profile", Some(home)); + let (profile_paths, profile_after_restart) = + drain_hub_consumer(&restarted, "profile", &source, bounds, profile_frontier).await; + + if settled { + assert!(project_paths.is_empty()); + assert!( + profile_paths.is_empty(), + "a consumer whose complete frontier matches the corpus epoch was re-offered {} \ + rollouts", + profile_paths.len() + ); + assert_eq!(profile_after_restart, profile_frontier); + } + + let added = write_dated_rollout(home, ("2026", "08", "24"), "after-restart"); + let (grown, _) = drain_hub_consumer( + &restarted, + "profile", + &source, + bounds, + profile_after_restart, + ) + .await; + assert!(grown.contains(&added)); + } + #[tokio::test] async fn completed_secondary_source_rebuilds_after_add_and_delete() { crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); From 5e4184e55884b798ff8c5844c94f21c896f701ef Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 01:34:14 -0700 Subject: [PATCH 2/4] fix(memory): admit bounded text builds and retain wait diagnostics --- .../src/code_index_scheduler.rs | 3 +- .../src/code_index_scheduler/serving.rs | 262 ++++++++------- .../tests/code_index_scheduler/serving.rs | 307 ++++++++++++++++-- .../src/resident_memory.rs | 66 +++- .../src/resident_memory/tests.rs | 127 ++++++++ .../advisory_runtime_acceptance.rs | 31 +- 6 files changed, 639 insertions(+), 157 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs index baca65a33f..a325e59813 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler.rs @@ -300,8 +300,7 @@ pub use serving::{ use serving::{ CodeIndexCommittedProgressSampleV1, CodeTextProjectionSlotV1, map_sealed_page_source_error, sha256_private_file_and_size, text_artifact_admitted_build_budget, - text_artifact_builder_budget, text_artifact_resident_memory_charge, - text_artifact_source_batch_limits, + text_artifact_builder_budget, text_artifact_source_batch_limits, }; #[cfg(test)] use tracedecay_query::retrieval::ports::TEXT_ARTIFACT_PAGE_CHUNKS_V1; diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs index de14bd949d..719d44539f 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs @@ -5,7 +5,6 @@ mod family_report; use std::{ collections::{BTreeSet, VecDeque}, io::Read, - num::NonZeroU64, path::{Path, PathBuf}, sync::{ Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError, RwLock, @@ -230,6 +229,29 @@ impl CodeIndexBuildProgressSlotStateV1 { let _ = self.publish(&generation_id, self.owner_epoch, snapshot); } + /// A text refusal belongs to its pending owner. A caller without the + /// projection claim must also identify the progress it observed before + /// attempting admission, so a delayed refusal cannot overwrite new work. + pub(super) fn block_owned_text( + &mut self, + generation_id: &CodeGenerationId, + owner_epoch: u64, + reason: CodeIndexBuildBlockedReasonV1, + expected_progress_epoch: Option, + ) -> bool { + let Some(current) = self.snapshot() else { + return false; + }; + if current.phase == CodeIndexBuildPhaseV1::Ready + || expected_progress_epoch.is_some_and(|epoch| current.progress_epoch != epoch) + { + return false; + } + let mut snapshot = current.as_ref().clone(); + snapshot.blocked_reason = Some(reason); + self.publish(generation_id, owner_epoch, snapshot) + } + pub fn snapshot(&self) -> Option> { self.snapshot.as_ref().map(Arc::clone) } @@ -947,29 +969,6 @@ pub struct DaemonCodeTextArtifactStoreV1 { worktree_id: WorktreeId, } -/// The bytes one artifact admission reserves: its ceiling plus the band above -/// the admission watermark, so the reserve call enforces that watermark -/// against both the ledger and the measured resident set. The band is not -/// memory the artifact owns and is released right after the reserve; -/// retaining it in every overlapping build would charge the same -/// process-wide margin repeatedly. The process's live set is not charged: -/// the reserve already compares the request with the measured resident set, -/// which contains it. -pub(super) fn text_artifact_resident_memory_charge( - requested: NonZeroU64, - watermark_headroom: u64, -) -> Result { - requested - .get() - .checked_add(watermark_headroom) - .and_then(NonZeroU64::new) - .ok_or_else(|| { - RetrievalPortError::Contract( - "text-artifact resident-memory accounting overflowed".to_owned(), - ) - }) -} - pub(super) fn text_artifact_admitted_build_budget( preferred_bytes: u64, minimum_bytes: u64, @@ -1170,10 +1169,8 @@ impl DaemonCodeTextArtifactStoreV1 { } } - /// Reserve one artifact memory ceiling. The atomic reserve also includes - /// the process-wide high-watermark headroom, then releases that - /// check-only margin before returning while the component ceiling stays - /// charged. + /// Charge the artifact's ceiling below the process admission watermark. + /// Sizing and charging share one measured headroom reading. fn reserve_resident_memory( &self, generation_id: &CodeGenerationId, @@ -1184,32 +1181,6 @@ impl DaemonCodeTextArtifactStoreV1 { .map(|(reservation, _)| reservation) } - /// Measure headroom and size the build: `(observed, watermark headroom, - /// admitted)` bytes. - fn text_artifact_admission( - &self, - preferred: NonZeroU64, - minimum: NonZeroU64, - ) -> Result<(u64, u64), RetrievalPortError> { - let admission_watermark = self.resident_memory.admission_watermark_bytes(); - let headroom = self.resident_memory.headroom_below(admission_watermark); - let watermark_headroom = self - .resident_memory - .snapshot() - .limit_bytes - .saturating_sub(admission_watermark); - let admitted_bytes = text_artifact_admitted_build_budget( - preferred.get(), - minimum.get(), - headroom.available_bytes, - ) - .inspect_err(|_| { - self.resident_memory - .wait_for_headroom(minimum.get(), admission_watermark); - })?; - Ok((watermark_headroom, admitted_bytes)) - } - fn reserve_resident_memory_up_to( &self, generation_id: &CodeGenerationId, @@ -1235,57 +1206,59 @@ impl DaemonCodeTextArtifactStoreV1 { "text-artifact minimum resident-memory reservation must be nonzero".to_owned(), ) })?; - let (watermark_headroom, admitted_bytes) = - match self.text_artifact_admission(preferred, minimum) { - Err(RetrievalPortError::ResidentMemoryRefused(detail)) => { - let released = self - .resident_owners - .shed(minimum.get(), std::time::Instant::now()); - for release in &released { - log_resident_owner_release_v1(release); - } - // Freed pages sit in the pool workers' allocator heaps - // whether or not an owner was shed. - let trim = release_process_allocator_memory_v1(); - tracing::info!( - event = "code_text_artifact_build_shed_retained_state", - released_owners = released.len(), - trimmed_bytes = trim.released_bytes(), - refused = %detail, - "text-artifact build shed retained state before re-measuring headroom" - ); - self.text_artifact_admission(preferred, minimum) + let admission_watermark = self.resident_memory.admission_watermark_bytes(); + let reserve = || { + let reservation = self + .resident_memory + .reserve_up_to( + ResidentMemoryKeyV1 { + project_id: self.project_id.clone(), + worktree_id: self.worktree_id.clone(), + generation_id: generation_id.clone(), + component, + }, + preferred, + admission_watermark, + ) + .map_err(|error| { + RetrievalPortError::ResidentMemoryRefused(format!( + "text-artifact resident-memory admission was refused: {error}" + )) + })?; + let admitted = text_artifact_admitted_build_budget( + preferred.get(), + minimum.get(), + reservation.reserved_bytes(), + )?; + Ok((reservation, admitted)) + }; + let (reservation, admitted) = match reserve() { + Err(RetrievalPortError::ResidentMemoryRefused(detail)) => { + let released = self + .resident_owners + .shed(minimum.get(), std::time::Instant::now()); + for release in &released { + log_resident_owner_release_v1(release); } - admission => admission, - }?; - let admitted = NonZeroU64::new(admitted_bytes).ok_or_else(|| { - RetrievalPortError::Contract( - "text-artifact admitted resident-memory reservation must be nonzero".to_owned(), - ) - })?; - let accounted = text_artifact_resident_memory_charge(admitted, watermark_headroom)?; - let mut reservation = self - .resident_memory - .reserve( - ResidentMemoryKeyV1 { - project_id: self.project_id.clone(), - worktree_id: self.worktree_id.clone(), - generation_id: generation_id.clone(), - component, - }, - accounted, - ) - .map_err(|error| { - RetrievalPortError::ResidentMemoryRefused(format!( - "text-artifact resident-memory admission was refused: {error}" - )) - })?; - reservation.shrink_to(admitted.get()).map_err(|error| { - RetrievalPortError::Contract(format!( - "text-artifact resident-memory headroom release failed: {error}" - )) + let trim = release_process_allocator_memory_v1(); + tracing::info!( + event = "code_text_artifact_build_shed_retained_state", + released_owners = released.len(), + trimmed_bytes = trim.released_bytes(), + refused = %detail, + "text-artifact build shed retained state before re-measuring headroom" + ); + reserve() + } + result => result, + } + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + self.resident_memory + .wait_for_headroom(minimum.get(), admission_watermark); + } })?; - let admitted = usize::try_from(admitted.get()).map_err(|error| { + let admitted = usize::try_from(admitted).map_err(|error| { RetrievalPortError::Contract(format!( "text-artifact admitted reservation exceeds the platform limit: {error}" )) @@ -2723,9 +2696,13 @@ impl LatestCodeTextGenerationV1 { // its phase before it re-attempts the work that was refused, so // clearing here erased the reason between two identical refusals // and status reported `blocked_reason: null` throughout a stall. - // A committed boundary is the honest clear -- it builds a fresh - // snapshot with no blocked reason -- and only that runs after work - // actually landed. + // A committed boundary clears the refusal. Ready is also a + // successful boundary: it is published only after the verified + // query owners are installed, even if finalization completed + // without another pending boundary in between. + if phase == CodeIndexBuildPhaseV1::Ready { + snapshot.blocked_reason = None; + } let _ = slot.publish(generation_id, self.text_progress_owner_epoch, snapshot); } }; @@ -2742,18 +2719,26 @@ impl LatestCodeTextGenerationV1 { fn publish_text_artifact_block(&self, error: &CodeLexicalArtifactErrorV1) { match error { CodeLexicalArtifactErrorV1::Unreserved(_) => { - self.publish_text_progress_blocked(CodeIndexBuildBlockedReasonV1::ResidentMemory); + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); } CodeLexicalArtifactErrorV1::Io(_) | CodeLexicalArtifactErrorV1::Missing(_) => { self.publish_text_progress_blocked( CodeIndexBuildBlockedReasonV1::ArtifactStoreUnavailable, + None, ); } _ => {} } } - fn publish_text_progress_blocked(&self, reason: CodeIndexBuildBlockedReasonV1) { + fn publish_text_progress_blocked( + &self, + reason: CodeIndexBuildBlockedReasonV1, + expected_progress_epoch: Option, + ) { let generation_id = &self.metadata.manifest().generation_id; { let _span = tracing::trace_span!("query.artifact.progress.publish").entered(); @@ -2762,12 +2747,12 @@ impl LatestCodeTextGenerationV1 { .text_progress_slot .write() .unwrap_or_else(std::sync::PoisonError::into_inner); - let Some(current) = slot.snapshot() else { - return; - }; - let mut snapshot = current.as_ref().clone(); - snapshot.blocked_reason = Some(reason); - let _ = slot.publish(generation_id, self.text_progress_owner_epoch, snapshot); + let _ = slot.block_owned_text( + generation_id, + self.text_progress_owner_epoch, + reason, + expected_progress_epoch, + ); } }; } @@ -2811,7 +2796,13 @@ impl LatestCodeTextGenerationV1 { maximum_work: usize, control: &dyn CodeIndexExecutionControlV1, ) -> Result { - let _workers = self + let progress_before_admission = self + .text_progress_slot + .read() + .unwrap_or_else(PoisonError::into_inner) + .snapshot() + .map(|snapshot| snapshot.progress_epoch); + let workers = self .text_artifact_store .worker_runtime .enter(&self.text_artifact_store.resident_memory) @@ -2824,8 +2815,23 @@ impl LatestCodeTextGenerationV1 { ), ) => RetrievalPortError::ResidentMemoryRefused(error.to_string()), other => RetrievalPortError::AuthorityUnavailable(other.to_string()), - })?; - let result = self.advance_text_serving_inner(maximum_work, control); + }); + let result = match workers { + Ok(_workers) => self.advance_text_serving_inner(maximum_work, control), + Err(error) => { + if matches!(&error, RetrievalPortError::ResidentMemoryRefused(_)) + && let Some(epoch) = progress_before_admission + { + // Worker admission owns no projection claim. Only the + // unchanged progress it observed can inherit this refusal. + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(epoch), + ); + } + Err(error) + } + }; if result.as_ref().is_err_and(|error| { matches!( error, @@ -3347,7 +3353,17 @@ impl LatestCodeTextGenerationV1 { let outcome = { let _span = tracing::trace_span!("query.artifact.head_open").entered(); self.open_published_head_or_begin_build(control) - }?; + } + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + // Keep the claim until this attempt's refusal is visible; + // no later wake can publish progress ahead of this block. + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); + } + })?; match outcome { TextHeadOpenOutcomeV1::Served => return Ok(true), TextHeadOpenOutcomeV1::Build(initialized) => { @@ -3638,7 +3654,15 @@ impl LatestCodeTextGenerationV1 { &content_key, build_reservation, control, - )?; + ) + .inspect_err(|error| { + if matches!(error, RetrievalPortError::ResidentMemoryRefused(_)) { + self.publish_text_progress_blocked( + CodeIndexBuildBlockedReasonV1::ResidentMemory, + None, + ); + } + })?; Ok(true) } diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 05bf0e812e..708739c618 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -3,7 +3,7 @@ use std::{ fmt::Write as _, num::NonZeroU64, path::{Path, PathBuf}, - sync::{Arc, mpsc}, + sync::{Arc, RwLock, mpsc}, thread, time::{Duration, Instant}, }; @@ -20,6 +20,7 @@ use tracedecay_contracts::{ PageRequest, PhraseSearchRequest, QualifiedNameRequest, ResolvedScope, ResultProjection, RetrievalOrder, RetrievalPortContext, RetrievalPortOutcome, RetrievalRequestMeta, SourceMetadataRequest, callable_code_operation, + code_index_freshness::{CodeIndexBuildBlockedReasonV1, CodeIndexBuildPhaseV1}, retrieval::{ CodeFacetDimension, CodeFacetRequest, CodeHierarchyRequest, CodeImpactRequest, CodeImplementationsRequest, CodeNavigationRequest, CodeTimelineRequest, @@ -49,8 +50,8 @@ use tracedecay_query::retrieval::{ }; use tracedecay_runtime_core::resident_memory::{ DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, ProcessResidentSampleV1, - RESIDENT_MEMORY_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ResidentMemoryPressureV1, - resident_memory_watermark_bytes_v1, + RESIDENT_MEMORY_PRESSURE_HIGH_WATERMARK_PERMILLE_V1, ResidentMemoryComponentIdV1, + ResidentMemoryPressureV1, resident_memory_watermark_bytes_v1, }; use tracedecay_session_temporal_store::SessionTemporalAccess; @@ -2086,42 +2087,192 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { } #[test] -fn overlapping_text_builds_share_one_admission_watermark_headroom() { - let limit_bytes = 1_000_u64; - let watermark_headroom = 100_u64; - let requested = NonZeroU64::new(300).expect("nonzero build request"); - let mut used_bytes = 0_u64; - - for _ in 0..3 { - let accounted = - super::super::text_artifact_resident_memory_charge(requested, watermark_headroom) - .expect("bounded admission accounting"); - assert!( - used_bytes + accounted.get() <= limit_bytes, - "each overlapping build fits beneath the same 900-byte high watermark" +fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn reserved_until_release() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("test process limit"); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let memory = Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::clone(&pressure), + )); + { + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), ); - used_bytes += requested.get(); + scheduler.bind_resident_memory(Arc::clone(&memory)); + published(scheduler.reconcile_now().expect("seal source generation")); } + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + scheduler.bind_resident_memory(Arc::clone(&memory)); + let text = scheduler + .servable_retained_text_generation() + .expect("publication store") + .expect("retained sealed text generation"); + let scanned = build_progress_snapshot(&scheduler); + assert_eq!(scanned.phase, CodeIndexBuildPhaseV1::SourceScan); + assert_eq!(scanned.blocked_reason, None); + let build_floor = u64::try_from(CODE_LEXICAL_ARTIFACT_BUILD_MEMORY_BUDGET_BYTES_V1) + .expect("build floor fits u64"); + let held_bytes = memory.admission_watermark_bytes() - build_floor + 1; + let competing = memory + .reserve_process_shared( + ResidentMemoryComponentIdV1::new("held-text-headroom").expect("component"), + NonZeroU64::new(held_bytes).expect("competing reservation"), + ) + .expect("hold actual ledger headroom"); + let mut headroom = pressure.subscribe_headroom(); - assert_eq!( - used_bytes, 900, - "the retained ledger owns the three build ceilings, not three margins" + for claimed_path in [true, false] { + let refused = if claimed_path { + text.advance_artifact_text_serving(1, &text.text_execution_control()) + } else { + text.advance_text_serving(1) + }; + assert!(matches!( + refused, + Err(RetrievalPortError::ResidentMemoryRefused(_)) + )); + let blocked = build_progress_snapshot(&scheduler); + assert_eq!(blocked.generation_id, scanned.generation_id); + assert_eq!(blocked.phase, CodeIndexBuildPhaseV1::SourceScan); + assert_eq!(blocked.committed_pages, 0); + assert_eq!( + blocked.blocked_reason, + Some(CodeIndexBuildBlockedReasonV1::ResidentMemory) + ); + assert!(text.text_projection_needs_work()); + assert_eq!(memory.snapshot().used_bytes, held_bytes); + } + + drop(headroom.borrow_and_update()); + drop(competing); + assert!( + headroom + .has_changed() + .expect("headroom authority remains live") + ); + assert_eq!(memory.snapshot().used_bytes, 0); + text.production_query_owners() + .expect("same owner builds after headroom is released"); + let ready = build_progress_snapshot(&scheduler); + assert_eq!(ready.generation_id, scanned.generation_id); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert!(ready.committed_pages > 0); + assert_eq!(ready.blocked_reason, None); + assert!(text.query_owners_are_ready()); + assert!(active_text_artifact_path(store.path()).is_file()); +} + +#[test] +fn ready_publication_clears_a_finalization_refusal_without_another_pending_boundary() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn ready_after_refusal() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("test process limit"); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), ); - for overflow in [ - super::super::text_artifact_resident_memory_charge( - NonZeroU64::new(u64::MAX).expect("maximum nonzero request"), - 1, - ), - super::super::text_artifact_resident_memory_charge(requested, u64::MAX), - ] { + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, pressure, + ))); + published(scheduler.reconcile_now().expect("seal source generation")); + let _workers = scheduler + .ensure_worker_plan() + .expect("fixture worker runtime"); + let text = scheduler.latest_complete().expect("text generation"); + let control = text.text_execution_control(); + assert!( + !text + .advance_text_serving(0) + .expect("initialize pending text progress") + ); + let finalization_deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING; + while !matches!( + build_progress_snapshot(&scheduler).phase, + CodeIndexBuildPhaseV1::IndexBuild | CodeIndexBuildPhaseV1::Verification + ) { assert!( - matches!( - overflow, - Err(tracedecay_query::retrieval::RetrievalPortError::Contract(_)) - ), - "overflow must remain a typed contract refusal: {overflow:?}" + Instant::now() < finalization_deadline, + "text build did not reach finalization" + ); + assert!( + !text.advance_text_serving(1).expect("begin finalization"), + "the bounded build must expose a pending finalization boundary" ); } + { + let mut slot = text.text_projection_build.lock_slot(); + let super::super::CodeTextProjectionSlotV1::Building(build) = &mut *slot else { + panic!("resumable text builder"); + }; + while !matches!( + build + .builder + .advance_started_finalization(1_000, &control) + .expect("finish the durable artifact before publication"), + CodeLexicalArtifactFinalizationStepV1::Ready(_) + ) { + assert!( + Instant::now() < finalization_deadline, + "durable finalization did not finish" + ); + } + } + text.publish_text_progress_phase(CodeIndexBuildPhaseV1::Verification, 0, 0); + let before = build_progress_snapshot(&scheduler); + let progress = scheduler.build_progress_slot(); + assert!( + progress + .write() + .expect("progress authority") + .block_owned_text( + &text.metadata().manifest().generation_id, + text.text_progress_owner_epoch, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(before.progress_epoch), + ) + ); + + // A resumed finalization can go directly from its prior refusal to + // installing query owners. No intervening Pending snapshot clears it. + assert!( + text.advance_artifact_text_serving(1, &control) + .expect("publish the finalized artifact") + ); + let ready = build_progress_snapshot(&scheduler); + assert!(text.query_owners_are_ready()); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert_eq!(ready.blocked_reason, None); + assert_eq!(ready.committed_pages, before.committed_pages); } #[test] @@ -2758,6 +2909,98 @@ fn progress_owner_epoch_rejects_original_a_after_a_b_a_replacement() { after_stale_original.progress_epoch, replacement_a.progress_epoch, "matching generation identity must not let the original A owner replace the newer A epoch" ); + assert!(!slot.block_owned_text( + &generation_a, + original_a_owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(after_stale_original.progress_epoch), + )); + assert_eq!( + slot.snapshot().expect("replacement owner").blocked_reason, + None + ); +} + +#[test] +fn delayed_memory_refusal_cannot_replace_newer_pending_or_ready_progress() { + for phase in [ + CodeIndexBuildPhaseV1::BulkCommit, + CodeIndexBuildPhaseV1::Ready, + ] { + let generation = + CodeGenerationId::new("generation.progress-refusal").expect("generation identity"); + let mut state = super::super::CodeIndexBuildProgressSlotStateV1::default(); + let owner = state.replace_generation(generation.clone()); + assert!(state.publish( + &generation, + owner, + progress_snapshot_for_generation(&generation, 1), + )); + let slot = Arc::new(RwLock::new(state)); + let (observed_tx, observed_rx) = mpsc::sync_channel(1); + let (resume_tx, resume_rx) = mpsc::sync_channel(1); + let delayed_slot = Arc::clone(&slot); + let delayed_generation = generation.clone(); + let delayed = thread::spawn(move || { + let observed_epoch = delayed_slot + .read() + .expect("observe pending progress") + .snapshot() + .expect("pending snapshot") + .progress_epoch; + observed_tx.send(()).expect("admission began"); + resume_rx + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("admission refused after concurrent progress"); + delayed_slot + .write() + .expect("publish refusal") + .block_owned_text( + &delayed_generation, + owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(observed_epoch), + ) + }); + observed_rx + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("worker observed its admission epoch"); + let newer = { + let mut current = slot.write().expect("publish concurrent progress"); + let mut next = progress_snapshot_for_generation(&generation, 2); + next.phase = phase; + assert!(current.publish(&generation, owner, next)); + current.snapshot().expect("newer progress") + }; + resume_tx.send(()).expect("release delayed refusal"); + assert!(!delayed.join().expect("delayed worker")); + + let mut current = slot.write().expect("read surviving progress"); + let after = current.snapshot().expect("current progress"); + assert_eq!(after.progress_epoch, newer.progress_epoch); + assert_eq!(after.phase, phase); + assert_eq!(after.committed_pages, 2); + assert_eq!(after.blocked_reason, None); + let accepts_current_refusal = current.block_owned_text( + &generation, + owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(after.progress_epoch), + ); + assert_eq!( + accepts_current_refusal, + phase != CodeIndexBuildPhaseV1::Ready + ); + assert_eq!( + current + .snapshot() + .expect("current refusal verdict") + .blocked_reason, + (phase != CodeIndexBuildPhaseV1::Ready) + .then_some(CodeIndexBuildBlockedReasonV1::ResidentMemory), + "an unchanged pending owner must still report its genuine wait" + ); + } } #[test] diff --git a/crates/tracedecay-runtime-core/src/resident_memory.rs b/crates/tracedecay-runtime-core/src/resident_memory.rs index 34a9c445be..a16e42ca7d 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory.rs @@ -1531,6 +1531,62 @@ impl ProcessResidentMemoryV1 { Err(self.refused(self.admission_failure(requested_bytes))) } + /// Size and charge new work from the same measured headroom reading. + /// The returned ceiling can be smaller than `preferred_bytes`; consumers + /// with a minimum working set must release a reservation below that floor. + /// Concurrent ledger charges are checked under the charge's own lock. + /// The consumer registers its working-set minimum with `wait_for_headroom` + /// when the returned budget cannot support its next operation. + #[tracing::instrument( + name = "runtime_core.resident.reserve_up_to", + level = "trace", + skip_all + )] + pub fn reserve_up_to( + self: &Arc, + key: ResidentMemoryKeyV1, + preferred_bytes: NonZeroU64, + ceiling: u64, + ) -> Result { + self.pressure.measure_admission_bytes(); + let ceiling = ceiling.min(self.limit_bytes.get()); + let mut state = self.lock_state(); + let headroom = self.headroom_from_state(&state, ceiling); + let admitted = preferred_bytes + .get() + .min(headroom.available_bytes) + .min(ceiling.saturating_sub(state.used_bytes)); + if admitted == 0 { + let failure = if self.pressure.state().is_over_budget() + || headroom.observed_bytes >= ceiling.min(self.pressure.limit_bytes()) + { + ResidentMemoryAdmissionFailureV1::ObservedOverBudget { + observed_bytes: headroom.observed_bytes, + limit_bytes: self.pressure.limit_bytes(), + high_watermark_bytes: self.pressure.high_watermark_bytes(), + requested_bytes: preferred_bytes.get(), + floor_bytes: RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, + } + } else { + ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: state.used_bytes, + requested_bytes: preferred_bytes.get(), + limit_bytes: ceiling, + } + }; + drop(state); + return Err(failure); + } + // `admitted` is at most ceiling - used_bytes, so both ledger sums fit. + state.used_bytes += admitted; + *state.charges.entry(key.clone()).or_default() += admitted; + Ok(ResidentMemoryReservationV1 { + authority: Arc::clone(self), + key, + reserved_bytes: admitted, + }) + } + /// Reserves one process-shared component without fabricating a project, /// worktree, or code-generation owner. These reservations use the same /// process ceiling and RAII release authority as project generations. @@ -1596,12 +1652,20 @@ impl ProcessResidentMemoryV1 { /// [`Self::reserve`] holds them. pub fn headroom_below(&self, ceiling: u64) -> ResidentMemoryHeadroomV1 { self.pressure.measure_admission_bytes(); + self.headroom_from_state(&self.lock_state(), ceiling) + } + + fn headroom_from_state( + &self, + state: &ResidentMemoryStateV1, + ceiling: u64, + ) -> ResidentMemoryHeadroomV1 { let measured_ceiling = ceiling.min(self.pressure.limit_bytes()); let observed_bytes = self .pressure .observed_bytes_for_limit(measured_ceiling) .unwrap_or(0); - let used_bytes = self.lock_state().used_bytes; + let used_bytes = state.used_bytes; let available_bytes = if self.pressure.state().is_over_budget() { 0 } else { diff --git a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs index 557293b4ec..10fe13f033 100644 --- a/crates/tracedecay-runtime-core/src/resident_memory/tests.rs +++ b/crates/tracedecay-runtime-core/src/resident_memory/tests.rs @@ -1335,6 +1335,35 @@ fn cgroup_headroom_respects_a_larger_but_fuller_ancestor() { ); } +#[test] +fn sized_reservations_charge_only_the_measured_cgroup_headroom() { + const MIB: u64 = 1024 * 1024; + let limit = bytes(800 * MIB); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| { + Some(ProcessResidentSampleV1 { + resident_bytes: 100 * MIB, + unreclaimable_bytes: 100 * MIB, + swapped_bytes: 0, + cgroup_headroom_bytes: Some(200 * MIB), + }) + }), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure(limit, pressure)); + let watermark = authority.admission_watermark_bytes(); + let headroom = authority.headroom_below(watermark); + assert_eq!(headroom.available_bytes, 200 * MIB); + let owner = key("project-a", "worktree-a", "generation-a", "text-build"); + let held = authority + .reserve_up_to(owner, bytes(headroom.available_bytes + 1), watermark) + .expect("the actual allocation fits without charging the 80 MiB watermark band"); + assert_eq!(held.reserved_bytes(), 200 * MIB); + assert_eq!(authority.snapshot().used_bytes, 200 * MIB); + drop(held); + assert_eq!(authority.snapshot().used_bytes, 0); +} + #[test] fn narrowed_budgets_and_waiters_share_cgroup_capacity_without_charging_peers() { const MIB: u64 = 1024 * 1024; @@ -1619,3 +1648,101 @@ fn allocator_reclaim_preserves_cgroup_pressure_until_sample_recovers() { .is_ok() ); } + +#[test] +fn sized_admission_remeasures_and_charges_the_supported_smaller_budget() { + const MIB: u64 = 1024 * 1024; + let measured = Arc::new(AtomicU64::new(300 * MIB)); + let sampled = Arc::clone(&measured); + let reads = Arc::new(AtomicU64::new(0)); + let sampled_reads = Arc::clone(&reads); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + bytes(1000 * MIB), + Arc::new(move || { + sampled_reads.fetch_add(1, Ordering::SeqCst); + let observed = sampled.load(Ordering::SeqCst); + Some(ProcessResidentSampleV1 { + resident_bytes: observed, + unreclaimable_bytes: observed, + swapped_bytes: 0, + cgroup_headroom_bytes: None, + }) + }), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + bytes(1000 * MIB), + Arc::clone(&pressure), + )); + let watermark = authority.admission_watermark_bytes(); + assert_eq!(watermark, 900 * MIB); + assert_eq!( + authority.headroom_below(watermark).available_bytes, + 600 * MIB + ); + measured.store(700 * MIB, Ordering::SeqCst); + let before = reads.load(Ordering::SeqCst); + let reservation = authority + .reserve_up_to( + key("project-a", "worktree-a", "generation-a", "text-build"), + bytes(600 * MIB), + watermark, + ) + .expect("the smaller build fits despite growth since planning"); + assert_eq!(reads.load(Ordering::SeqCst), before + 1); + assert_eq!(reservation.reserved_bytes(), 200 * MIB); + assert_eq!(authority.snapshot().used_bytes, 200 * MIB); + drop(reservation); + assert_eq!(authority.snapshot().used_bytes, 0); + + measured.store(950 * MIB, Ordering::SeqCst); + let refused = authority + .reserve_up_to( + key("project-a", "worktree-a", "generation-b", "text-build"), + bytes(600 * MIB), + watermark, + ) + .expect_err("sizing must still refuse measured pressure"); + assert!(refused.is_observed_over_budget()); + assert_eq!(authority.snapshot().used_bytes, 0); +} + +#[test] +fn sized_admissions_share_the_watermark_without_charging_its_margin() { + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + bytes(1000), + Arc::new(|| None), + )); + let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( + bytes(1000), + pressure, + )); + let watermark = authority.admission_watermark_bytes(); + let owner = key("project-a", "worktree-a", "generation-a", "text-build"); + let mut reservations = (0..3) + .map(|_| { + authority + .reserve_up_to(owner.clone(), bytes(300), watermark) + .unwrap() + }) + .collect::>(); + assert_eq!(authority.snapshot().used_bytes, 900); + let refused = authority + .reserve_up_to(owner.clone(), bytes(300), watermark) + .unwrap_err(); + assert_eq!( + refused, + ResidentMemoryAdmissionFailureV1::ReservationCeiling { + used_bytes: 900, + requested_bytes: 300, + limit_bytes: 900, + } + ); + drop(reservations.pop()); + let maximum = authority + .reserve_up_to(owner, bytes(u64::MAX), watermark) + .expect("bounded sizing cannot overflow while reserving available bytes"); + assert_eq!(maximum.reserved_bytes(), 300); + assert_eq!(authority.snapshot().used_bytes, 900); + drop((reservations, maximum)); + assert_eq!(authority.snapshot().used_bytes, 0); +} diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs index 9330bde261..2d588411d9 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs @@ -1,12 +1,14 @@ //! Strict advisory runtime acceptance over authentic provider response captures. use std::collections::{BTreeSet, VecDeque}; +use std::io::SeekFrom; use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use std::time::Duration; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tracedecay_application::advisory::ci_runtime::GitHubCiOfficialResponseDecoderV1; #[cfg(feature = "test-transport")] use tracedecay_application::advisory::ci_runtime::{ @@ -1216,6 +1218,29 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { // client protocol; only a non-retryable failure is a test failure. let advisory_deadline = std::time::Instant::now() + Duration::from_secs(60); let advisory = loop { + // Capture only the already-owned log before invoking the CLI. The + // same phase deadline bounds this read, so a terminal request failure + // never starts new diagnostic work after its deadline has elapsed. + let daemon_log_before_invocation = + tokio::time::timeout_at(tokio::time::Instant::from_std(advisory_deadline), async { + const MAX_LOG_BYTES: u64 = 16 * 1024; + let mut file = tokio::fs::File::open(&daemon_log).await?; + let end = file.seek(SeekFrom::End(0)).await?; + file.seek(SeekFrom::Start(end.saturating_sub(MAX_LOG_BYTES))) + .await?; + let mut bytes = Vec::new(); + file.take(MAX_LOG_BYTES).read_to_end(&mut bytes).await?; + Ok::<_, std::io::Error>(String::from_utf8_lossy(&bytes).into_owned()) + }) + .await + .map_or_else( + |_| "daemon log snapshot exceeded the registration deadline".to_owned(), + |result| result.unwrap_or_else(|error| format!("daemon log unavailable: {error}")), + ); + assert!( + std::time::Instant::now() < advisory_deadline, + "advisory registration deadline elapsed before invocation\ndaemon log captured before invocation (up to 16 KiB):\n{daemon_log_before_invocation}", + ); let advisory = common::tracedecay_command_with_home(environment.home()) .args([ "tool", @@ -1243,13 +1268,13 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { }); assert!( retryable_unavailable, - "registered advisory cycle failed\nstdout:\n{stdout}\nstderr:\n{}", + "registered advisory cycle failed\nstdout:\n{stdout}\nstderr:\n{}\ndaemon log captured before this invocation (up to 16 KiB):\n{daemon_log_before_invocation}", String::from_utf8_lossy(&advisory.stderr), ); assert!( std::time::Instant::now() < advisory_deadline, - "advisory cycle stayed unavailable past the deferred registration deadline\nstdout:\n{stdout}\ndaemon log:\n{}", - std::fs::read_to_string(&daemon_log).expect("read isolated advisory daemon log"), + "advisory cycle stayed unavailable past the deferred registration deadline\nstdout:\n{stdout}\nstderr:\n{}\ndaemon log captured before this invocation (up to 16 KiB):\n{daemon_log_before_invocation}", + String::from_utf8_lossy(&advisory.stderr), ); tokio::time::sleep(Duration::from_millis(250)).await; }; From 4e1d1e129562e7b7842182d9642c4174ffc5edcc Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 04:42:06 -0700 Subject: [PATCH 3/4] fix(tests): assert native headroom watch changes directly --- .../tests/code_index_scheduler/serving.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 708739c618..be4f194e80 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -2163,11 +2163,7 @@ fn retained_text_admission_reports_its_block_and_recovers_after_headroom_release drop(headroom.borrow_and_update()); drop(competing); - assert!( - headroom - .has_changed() - .expect("headroom authority remains live") - ); + assert!(headroom.has_changed()); assert_eq!(memory.snapshot().used_bytes, 0); text.production_query_owners() .expect("same owner builds after headroom is released"); From 1fd2d3613acd00761cc7e6473fa69ef1dba74045 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 20:40:39 -0700 Subject: [PATCH 4/4] docs(sessions): remove replay and fixture narration --- crates/tracedecay-sessions/src/runtime/hosts/codex.rs | 3 --- crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs | 2 -- .../runtime_acceptance_suite/advisory_runtime_acceptance.rs | 3 --- 3 files changed, 8 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index fa803aa40c..6d2ae4281f 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -579,9 +579,6 @@ impl CodexDiscoveryHub { detail: "Codex replay index disappeared before delivery", }, )?; - // A complete durable frontier at the index's epoch already - // covers every indexed path (a restarted daemon's later - // consumers land here), so its replay starts past the end. let effective_position = if frontier.is_complete() && frontier == index.frontier { index.paths.last().cloned() diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index 14930966e7..d4e5ba1c6d 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -1750,8 +1750,6 @@ mod recent_first_discovery_tests { .await; assert_eq!(project_frontier, profile_frontier); - // A restarted daemon registers its consumers one by one, so every - // consumer after the first joins a hub whose shared scan already began. let restarted = CodexDiscoveryHub::default(); restarted.register("project", Some(home)); let (project_paths, _) = diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs index 2d588411d9..9b492ece73 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs @@ -1218,9 +1218,6 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { // client protocol; only a non-retryable failure is a test failure. let advisory_deadline = std::time::Instant::now() + Duration::from_secs(60); let advisory = loop { - // Capture only the already-owned log before invoking the CLI. The - // same phase deadline bounds this read, so a terminal request failure - // never starts new diagnostic work after its deadline has elapsed. let daemon_log_before_invocation = tokio::time::timeout_at(tokio::time::Instant::from_std(advisory_deadline), async { const MAX_LOG_BYTES: u64 = 16 * 1024;