From f1be639fd42920db37c6ac6261ac06d929d62758 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 20 Oct 2026 14:23:46 -0700 Subject: [PATCH 1/6] docs(release): remove unfinished Windows signing rollout --- .signpath/artifact-configuration.xml | 21 --------------------- SECURITY.md | 2 +- docs/USER-GUIDE.md | 6 ++++-- 3 files changed, 5 insertions(+), 24 deletions(-) delete mode 100644 .signpath/artifact-configuration.xml diff --git a/.signpath/artifact-configuration.xml b/.signpath/artifact-configuration.xml deleted file mode 100644 index a564642a2e..0000000000 --- a/.signpath/artifact-configuration.xml +++ /dev/null @@ -1,21 +0,0 @@ - - - - - - - - - diff --git a/SECURITY.md b/SECURITY.md index 80992d8d4b..185587893a 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -142,7 +142,7 @@ consistency, not independent publisher authenticity. Other distribution/update paths have their own verification; do not assume installer checks apply to all paths. -**Windows:** Authenticode code signing via the [SignPath.io Foundation](https://signpath.io/foundation) program is being rolled out so Windows binaries are signed as part of the release workflow (addresses the Smart App Control block reported in #79). Until that lands in a published release, Windows binaries remain unsigned. +**Windows:** Release executables are unsigned. Archive checksums and build-provenance attestations do not provide an executable code signature. ### Opt-in background daemon diff --git a/docs/USER-GUIDE.md b/docs/USER-GUIDE.md index 64dc8bf90c..eba12a1652 100644 --- a/docs/USER-GUIDE.md +++ b/docs/USER-GUIDE.md @@ -61,14 +61,16 @@ gh attestation verify tracedecay-beta--.tar.gz --repo ScriptedAlc --signer-workflow ScriptedAlchemy/tracedecay/.github/workflows/release-beta.yml ``` -Stable archives (`tracedecay--`) are signed by -`.github/workflows/release.yml`. +Stable archives (`tracedecay--`) have build-provenance +attestations from `.github/workflows/release.yml`. **Windows:** Download the x86_64 Windows archive from the [latest release](https://github.com/ScriptedAlchemy/tracedecay/releases/latest), extract `tracedecay.exe`, and place it on `PATH`. +The Windows executable is unsigned; archive checksums and build-provenance +attestations do not provide an executable code signature. **Prebuilt binaries:** From dcf78fedef54e08ce9dab33c499359b371bce11a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:34:18 -0700 Subject: [PATCH 2/6] test(hooks): await delivery across retained stop retries --- .../intent_pass_journey_test.rs | 43 +++++++++++++++++-- 1 file changed, 39 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs b/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs index 1d126368ec..6fc34da9e4 100644 --- a/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs +++ b/crates/tracedecay/tests/hooks_lsp_suite/intent_pass_journey_test.rs @@ -321,8 +321,11 @@ fn intent_outcomes(log: &Path, session: &str) -> Vec { .collect() } -fn wait_for(what: &str, log: &Path, mut ready: impl FnMut() -> bool) { - let deadline = Instant::now() + WAIT; +fn wait_for(what: &str, log: &Path, ready: impl FnMut() -> bool) { + wait_for_until(what, log, Instant::now() + WAIT, ready); +} + +fn wait_for_until(what: &str, log: &Path, deadline: Instant, mut ready: impl FnMut() -> bool) { while !ready() { assert!( Instant::now() < deadline, @@ -559,13 +562,41 @@ fn claude_stop_delivers_the_intent_suggestion_the_daemon_judged_at_the_tool_boun let stop_event = stop(&project, ENABLED_SESSION, &enabled_transcript); let delivers = |context: &String| context.contains(&suggestion_text(YES_INTENT)); + let delivery_deadline = Instant::now() + WAIT; let first_stop = run_hook(home, &project, "hook-stop", &stop_event); let mut stop_contexts = vec![hook_context(&first_stop)]; let mut stop_outputs = vec![first_stop]; - if !stop_contexts.last().is_some_and(delivers) { - wait_for( + while !stop_contexts.last().is_some_and(delivers) { + let evidence = delivery_evidence(home, &project, &stop_contexts); + let completions = evidence["enrolled_hook_receipts"] + .as_str() + .expect("enrolled hook receipts") + .lines() + .map(|line| serde_json::from_str::(line).expect("valid hook receipt")) + .filter(|receipt| { + receipt["event"] == "hook_completed" + && receipt["hook_name"] == "Stop" + && receipt["session_id"] == ENABLED_SESSION + }) + .collect::>(); + assert_eq!( + completions.len(), + stop_outputs.len(), + "each stop attempt must have a completed receipt: {evidence}" + ); + let completion = completions.last().expect("completed stop attempt"); + assert!( + completion["disposition"]["status"] == "accepted_for_replay" + && completion["disposition"]["retryable"] == true, + "only a retained, retryable stop may be retried without guidance: {evidence}" + ); + // A host retry can also miss the unchanged 25ms admission window. + // Observe replay completion before resubmitting the same event, and + // spend one overall wait budget across all attempts. + wait_for_until( "the timed-out stop callback's replay to drain", &log, + delivery_deadline, || { let status = run_tool(home, &project, "tracedecay_status", &json!({})); let replay: StatusHookReplayV1 = @@ -575,6 +606,10 @@ fn claude_stop_delivers_the_intent_suggestion_the_daemon_judged_at_the_tool_boun && intent_outcomes(&log, ENABLED_SESSION).len() > before + 1 }, ); + assert!( + Instant::now() < delivery_deadline, + "the retained stop must deliver within the journey budget: {evidence}" + ); let next_stop = run_hook(home, &project, "hook-stop", &stop_event); stop_contexts.push(hook_context(&next_stop)); stop_outputs.push(next_stop); From 1d51c2b3dfc12075e95099dcfabdb0374f19ded9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 04:46:46 -0700 Subject: [PATCH 3/6] test(hosts): identify native cache refresh divergence --- .../src/agents/claude/tests.rs | 44 ++++++++++++++++++- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs index 707bc9da32..46c56d1880 100644 --- a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs +++ b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs @@ -1,4 +1,7 @@ -use super::super::{load_json_file_strict, safe_write_json_file, safe_write_text_file}; +use super::super::{ + load_json_file_strict, safe_write_json_file, safe_write_text_file, + unexpected_bundle_entrypoints, +}; use super::*; use serde_json::json; #[cfg(unix)] @@ -148,9 +151,46 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() { .unwrap() ); copy_rendered_bundle_to_native_cache(home.path(), new_bin); + let source = plugin_deploy_dir(home.path()); + let cache = claude_current_cached_plugin_root(home.path()); + let rendered = rendered_plugin_files(new_bin, &ServeArgs::default()).unwrap(); + let relatives = rendered + .iter() + .map(|(relative, _)| (*relative).to_owned()) + .collect::>(); + for (relative, expected) in rendered { + assert_eq!( + std::fs::read_to_string(source.join(relative)).unwrap(), + expected, + "relocated marketplace bytes differ at {relative}" + ); + assert_eq!( + std::fs::read_to_string(cache.join(relative)).unwrap(), + expected, + "refreshed native cache bytes differ at {relative}" + ); + } + for root in [&source, &cache] { + assert_eq!( + unexpected_bundle_entrypoints( + root, + &relatives, + &[".claude-plugin", "agents", "commands", "hooks", "skills"], + ) + .unwrap(), + [], + "refresh left auto-discovered entrypoints outside the catalog at {}", + root.display() + ); + } + assert!( + claude_plugin_registration_is_active(home.path()).unwrap(), + "refresh must preserve the exact native plugin registration" + ); assert!( claude_plugin_is_natively_active(home.path(), Some((new_bin, &ServeArgs::default()))) - .unwrap() + .unwrap(), + "matching refreshed bytes must retain the exact discovery inventory" ); } From a7ef027b9a44185bbd240f83cb7691a5da831baf Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 04:58:50 -0700 Subject: [PATCH 4/6] fix(memory): size admission atomically and preserve blocked progress --- .../src/code_index_scheduler.rs | 3 +- .../src/code_index_scheduler/serving.rs | 262 ++++++------ .../tests/code_index_scheduler/serving.rs | 379 ++++++++++++++++-- .../src/resident_memory.rs | 66 ++- .../src/resident_memory/tests.rs | 127 ++++++ 5 files changed, 683 insertions(+), 154 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..e5c882effc 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,11 @@ use std::{ fmt::Write as _, num::NonZeroU64, path::{Path, PathBuf}, - sync::{Arc, mpsc}, + sync::{ + Arc, Mutex, RwLock, + atomic::{AtomicBool, Ordering}, + mpsc, + }, thread, time::{Duration, Instant}, }; @@ -20,6 +24,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 +54,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 +2091,260 @@ 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()); + 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] +fn concurrent_text_admission_refusal_cannot_block_a_ready_owner() { + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn concurrent_text_owner() {}\n")]); + let store = TempDir::new().expect("store root"); + let limit = NonZeroU64::new(4 * 1024 * 1024 * 1024).expect("process limit"); + let (sampling, sampled) = mpsc::channel(); + let (resume, resumed) = mpsc::channel(); + let resumed = Mutex::new(resumed); + let paused = AtomicBool::new(false); + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(move || { + let refused = thread::current().name() == Some("refused-text-advance"); + if refused && !paused.swap(true, Ordering::AcqRel) { + sampling.send(()).expect("announce refused sample"); + resumed + .lock() + .expect("sample resume receiver") + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("resume refused sample"); + } + Some(ProcessResidentSampleV1 { + resident_bytes: 0, + unreclaimable_bytes: 0, + swapped_bytes: 0, + cgroup_headroom_bytes: refused.then_some(0), + }) + }), + )); + let memory = Arc::new(ProcessResidentMemoryV1::with_pressure(limit, pressure)); + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + scheduler.bind_resident_memory(memory); + published(scheduler.reconcile_now().expect("seal source generation")); + let text = scheduler.latest_complete().expect("text generation"); + thread::scope(|scope| { + let refused_text = text.clone(); + let refused = thread::Builder::new() + .name("refused-text-advance".to_owned()) + .spawn_scoped(scope, move || refused_text.advance_text_serving(1)) + .expect("spawn refused advance"); + sampled + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("first advance reaches memory admission"); + let ready_text = text.clone(); + let (starting, started) = mpsc::channel(); + let ready = scope.spawn(move || { + starting.send(()).expect("announce concurrent advance"); + ready_text.production_query_owners() + }); + started + .recv_timeout(SERVING_SEAT_FAILURE_CEILING) + .expect("second advance starts while the first is admitted"); + resume.send(()).expect("release refused sample"); + assert!(matches!( + refused.join().expect("refused advance joins"), + Err(RetrievalPortError::ResidentMemoryRefused(_)) + )); + ready + .join() + .expect("ready advance joins") + .expect("concurrent owner serves after the refusal"); + }); + let ready = build_progress_snapshot(&scheduler); + assert!(text.query_owners_are_ready()); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert_eq!(ready.blocked_reason, None); } #[test] @@ -2758,6 +2981,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); +} From 80dc86ff26f26c8d71e3efcae0ac11c52d83a919 Mon Sep 17 00:00:00 2001 From: Zack Jackson Date: Wed, 21 Oct 2026 07:53:08 -0700 Subject: [PATCH 5/6] test(hosts): make native cache inventory assertion explicit --- .../src/agents/claude/tests.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs index 46c56d1880..db426a6db3 100644 --- a/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs +++ b/crates/tracedecay-agent-hosts/src/agents/claude/tests.rs @@ -171,15 +171,15 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() { ); } for root in [&source, &cache] { - assert_eq!( - unexpected_bundle_entrypoints( - root, - &relatives, - &[".claude-plugin", "agents", "commands", "hooks", "skills"], - ) - .unwrap(), - [], - "refresh left auto-discovered entrypoints outside the catalog at {}", + let unexpected = unexpected_bundle_entrypoints( + root, + &relatives, + &[".claude-plugin", "agents", "commands", "hooks", "skills"], + ) + .unwrap(); + assert!( + unexpected.is_empty(), + "refresh left auto-discovered entrypoints outside the catalog at {}: {unexpected:?}", root.display() ); } From 58ad3b353617ed349f7103b678378ea20c2627db Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:36:18 -0700 Subject: [PATCH 6/6] test(index): isolate fixture pressure from concurrent workers --- .../src/code_index_scheduler/memory_tests.rs | 38 +++++++++---------- .../src/code_index_scheduler/reconcile.rs | 19 ++++++++-- .../registry/resident_memory.rs | 16 ++++++-- .../code_index_scheduler/tests/residency.rs | 12 +----- .../tests/code_index_scheduler/reconcile.rs | 16 +++++--- .../tests/code_index_scheduler/serving.rs | 13 +++---- 6 files changed, 64 insertions(+), 50 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs index 0ebe34ada8..62139a246d 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/memory_tests.rs @@ -73,6 +73,17 @@ fn expected_worker_reservation_on(runtime: &CodeIndexWorkerRuntimeV1, remaining_ ) } +/// Ledger scenarios own their measured input independently of sibling tests. +pub(super) fn ledger_memory(limit: NonZeroU64) -> Arc { + Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )), + )) +} + #[test] fn captured_source_bytes_are_charged_until_the_snapshot_drops() { let project = fixture(); @@ -84,9 +95,7 @@ fn captured_source_bytes_are_charged_until_the_snapshot_drops() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new( - NonZeroU64::new(1024 * 1024).expect("source memory limit"), - )); + let authority = ledger_memory(NonZeroU64::new(1024 * 1024).expect("source memory limit")); scheduler.bind_resident_memory(Arc::clone(&authority)); let captured = scheduler @@ -119,9 +128,7 @@ fn reconcile_leaves_no_captured_source_charged_after_the_build() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let captured = scheduler .capture_authoritative_snapshot(None) @@ -171,7 +178,7 @@ fn source_capture_refuses_before_build_when_retention_cannot_be_charged() { Arc::new(SharedCodeIndexBytePoolV1::default()), ) .expect("open scheduler"); - let authority = Arc::new(ProcessResidentMemoryV1::new(NonZeroU64::MIN)); + let authority = ledger_memory(NonZeroU64::MIN); scheduler.bind_resident_memory(Arc::clone(&authority)); assert!(matches!( @@ -291,9 +298,7 @@ fn worker_memory_reservation_is_charged_and_released_by_raii() { 2, DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1.get(), ); - let authority = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let reservation = scheduler .reserve_worker_memory() @@ -325,14 +330,7 @@ fn default_authority_worker_reserve_leaves_typed_snapshot_headroom() { let runtime = bind_automatic_worker_runtime(&scheduler, 128, 128 * 1024 * 1024 * 1024); // This case isolates the ledger's worker/snapshot split. Live process // pressure can narrow the slab and is exercised with explicit samples below. - let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - Arc::new(|| None), - )); - let authority = Arc::new(ProcessResidentMemoryV1::with_pressure( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - pressure, - )); + let authority = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&authority)); let _worker = scheduler @@ -376,12 +374,12 @@ fn worker_memory_reservation_refusal_is_typed() { 2, DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1.get(), ); - let authority = Arc::new(ProcessResidentMemoryV1::new( + let authority = ledger_memory( NonZeroU64::new( tracedecay_code_index::parallelism::INDEX_WORKER_RESIDENT_BUDGET_BYTES_V1 - 1, ) .expect("positive test limit"), - )); + ); scheduler.bind_resident_memory(authority); assert!(matches!( scheduler.reserve_worker_memory(), diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs index 14e34cf58d..d8bbfe3f5e 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/reconcile.rs @@ -37,6 +37,8 @@ use tracedecay_domain::{ }; use tracedecay_graph_db::GraphConflictContextV1; use tracedecay_privacy::CODE_SOURCE_SANITIZER_VERSION_V1; +#[cfg(any(test, feature = "test-helpers"))] +use tracedecay_runtime_core::resident_memory::ResidentMemoryPressureV1; use tracedecay_runtime_core::resident_memory::{ ProcessResidentMemoryV1, RESIDENT_MEMORY_PRESSURE_ADMISSION_FLOOR_BYTES_V1, RESIDENT_OWNER_IDLE_WINDOW_V1, ResidentMemoryAdmissionFailureV1, ResidentMemoryComponentIdV1, @@ -1250,7 +1252,9 @@ impl HistoricalCodeIndexGenerationOwnerV1 { } impl CodeIndexWorktreeSchedulerV1 { - /// A standalone owner under the path policy a fresh profile resolves. + /// A standalone fixture owner under the path policy a fresh profile resolves. + /// Concurrent cases do not contribute measured pressure to this owner; + /// pressure scenarios bind their own explicit readings after opening it. #[cfg(any(test, feature = "test-helpers"))] pub fn open( project_id: ProjectId, @@ -1258,14 +1262,23 @@ impl CodeIndexWorktreeSchedulerV1 { store_root: PathBuf, byte_pool: Arc, ) -> Result { - Self::open_with_policy( + let mut scheduler = Self::open_with_policy( project_id, project_root, store_root, byte_pool, CodeIndexHintPolicyV1::default(), crate::config::registry_default_index_path_policy(), - ) + )?; + let limit = detected_process_resident_memory_limit_v1(); + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::with_pressure( + limit, + Arc::new(ResidentMemoryPressureV1::with_sampler( + limit, + Arc::new(|| None), + )), + ))); + Ok(scheduler) } pub fn open_with_policy( diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs index b971b0dc1a..486d42d8fb 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/resident_memory.rs @@ -2,11 +2,12 @@ use std::sync::Arc; -#[cfg(any(test, feature = "test-helpers"))] -use tracedecay_runtime_core::resident_memory::DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1; pub use tracedecay_runtime_core::resident_memory::ProcessResidentMemoryV1; #[cfg(any(test, feature = "test-helpers"))] -use tracedecay_runtime_core::resident_memory::RESIDENT_OWNER_IDLE_WINDOW_V1; +use tracedecay_runtime_core::resident_memory::{ + DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, RESIDENT_OWNER_IDLE_WINDOW_V1, + ResidentMemoryPressureV1, +}; use tracedecay_runtime_core::resident_memory::{ResidentOwnersV1, process_resident_owners_v1}; use super::CodeIndexSchedulerRegistryV1; @@ -15,10 +16,17 @@ impl CodeIndexSchedulerRegistryV1 { #[cfg(any(test, feature = "test-helpers"))] #[cfg_attr(not(test), allow(dead_code))] pub fn new(max_worktrees: usize) -> Self { + // A fixture owns its ledger and pressure readings. Concurrent tests' + // RSS is not an input; pressure journeys pass an explicit authority. + let pressure = Arc::new(ResidentMemoryPressureV1::with_sampler( + DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, + Arc::new(|| None), + )); Self::with_resident_memory_and_progress_producer_incarnation( max_worktrees, - Arc::new(ProcessResidentMemoryV1::new( + Arc::new(ProcessResidentMemoryV1::with_pressure( DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, + pressure, )), 1, ) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs index 4cd725b7cb..875d53625d 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/residency.rs @@ -1476,14 +1476,6 @@ fn ledger_leaving_one_mib( .expect("the held memory fits the ledger") } -fn sixteen_gib_authority() -> Arc { - let limit = NonZeroU64::new(16 * 1024 * 1024 * 1024).unwrap(); - Arc::new(ProcessResidentMemoryV1::with_pressure( - limit, - Arc::new(ResidentMemoryPressureV1::new(limit)), - )) -} - /// A 16 GiB authority reading a fixed process size, so a held ledger refuses /// a refresh however much memory the tests beside it hold. fn fixed_reading_authority() -> Arc { @@ -1510,7 +1502,7 @@ async fn a_refresh_refused_for_memory_publishes_when_an_idle_owner_releases_it() let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]); let store = TempDir::new().expect("store root"); let owners = Arc::new(ResidentOwnersV1::new(IDLE_WINDOW)); - let resident_memory = sixteen_gib_authority(); + let resident_memory = fixed_reading_authority(); let (registry, _) = mounted_core_query_worktree_in( CodeIndexSchedulerRegistryV1::with_resident_memory(1, Arc::clone(&resident_memory)) .with_resident_owners(Arc::clone(&owners)), @@ -1576,7 +1568,7 @@ async fn a_refresh_refused_by_the_ledger_publishes_when_the_blocking_reservation let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]); let store = TempDir::new().expect("store root"); let owners = Arc::new(ResidentOwnersV1::new(IDLE_WINDOW)); - let resident_memory = sixteen_gib_authority(); + let resident_memory = fixed_reading_authority(); let (registry, _) = mounted_core_query_worktree_in( CodeIndexSchedulerRegistryV1::with_resident_memory(1, Arc::clone(&resident_memory)) .with_resident_owners(Arc::clone(&owners)), diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs index 30f6f177a3..182c11a2ba 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/reconcile.rs @@ -30,9 +30,11 @@ use tracedecay_domain::{ use tracedecay_runtime_core::resident_memory::{ DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, ProcessResidentMemoryV1, ResidentHoldingV1, ResidentOwnerBytesV1, ResidentOwnerKindV1, ResidentOwnerReleaseV1, ResidentOwnerSampleV1, - ResidentOwnerScopeV1, ResidentOwnerV1, ResidentOwnersV1, sampled_process_resident_bytes_v1, + ResidentOwnerScopeV1, ResidentOwnerV1, ResidentOwnersV1, + detected_process_resident_memory_limit_v1, sampled_process_resident_bytes_v1, }; +use super::super::memory_tests::ledger_memory; #[cfg(unix)] use super::stage_raw_git_path; use super::{ @@ -1437,12 +1439,17 @@ fn retained_decode_incremental_seal_and_text_publish_stay_below_the_high_waterma let fixture = GitFixture::new(&sources); let store = TempDir::new().expect("store root"); + // This scale journey measures real process pressure through admission + // as well as its RSS assertions, so retain the production pressure cell. + let memory_limit = detected_process_resident_memory_limit_v1(); + { let mut seed = scheduler( &fixture, store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); + seed.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new(memory_limit))); published(seed.reconcile_now().expect("seed retained generation")); } @@ -1455,6 +1462,7 @@ fn retained_decode_incremental_seal_and_text_publish_stay_below_the_high_waterma store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); + scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new(memory_limit))); assert_eq!(scheduler.publication.sealed_decode_count(), 0); assert!(matches!( scheduler @@ -11410,7 +11418,7 @@ fn graph_off_changed_source_worker_memory_denial_retries_without_decode() { ) > tight_limit.get(), "the fixture limit must deny the installed worker plan" ); - let tight = Arc::new(ProcessResidentMemoryV1::new(tight_limit)); + let tight = ledger_memory(tight_limit); scheduler.bind_resident_memory(Arc::clone(&tight)); let denied = scheduler.reconcile_retained_text_generation_with(&metadata_a, true); assert!( @@ -11435,9 +11443,7 @@ fn graph_off_changed_source_worker_memory_denial_retries_without_decode() { "a denied worker reservation must not leak a charge" ); - let adequate = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let adequate = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); scheduler.bind_resident_memory(Arc::clone(&adequate)); let outcome = scheduler .reconcile_retained_text_generation_with(&metadata_a, 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 e5c882effc..8e3588c588 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 @@ -60,6 +60,7 @@ use tracedecay_runtime_core::resident_memory::{ use tracedecay_session_temporal_store::SessionTemporalAccess; +use super::super::memory_tests::ledger_memory; use super::{ ALPHA_LIB_V1, CALLER_PAGE, CALLER_STAR, GitFixture, OwnerSignals, ReadyRetrievalControlV1, SERVING_SEAT_FAILURE_CEILING, active_text_artifact_path, application_context, @@ -2072,9 +2073,9 @@ fn reader_reservation_refusal_precedes_missing_artifact_access() { store.path().to_path_buf(), Arc::new(SharedCodeIndexBytePoolV1::default()), ); - scheduler.bind_resident_memory(Arc::new(ProcessResidentMemoryV1::new( + scheduler.bind_resident_memory(ledger_memory( std::num::NonZeroU64::new(1024 * 1024).expect("tight memory limit"), - ))); + )); let latest = scheduler.latest_complete().expect("restored generation"); assert!( matches!( @@ -2391,9 +2392,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { Arc::new(SharedCodeIndexBytePoolV1::default()), ); published(scheduler.reconcile_now().expect("publish")); - let tight = Arc::new(ProcessResidentMemoryV1::new( - std::num::NonZeroU64::new(1024 * 1024).expect("tight limit"), - )); + let tight = ledger_memory(std::num::NonZeroU64::new(1024 * 1024).expect("tight limit")); scheduler.bind_resident_memory(Arc::clone(&tight)); let latest = scheduler.latest_complete().expect("latest generation"); let denied = latest.advance_text_serving(usize::MAX); @@ -2471,9 +2470,7 @@ fn text_artifact_ceilings_reserve_through_process_resident_memory() { // An adequate authority admits the build and holds the reader charge // for as long as the artifact owners serve. - let adequate = Arc::new(ProcessResidentMemoryV1::new( - DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1, - )); + let adequate = ledger_memory(DEFAULT_PROCESS_RESIDENT_MEMORY_LIMIT_V1); { // A fresh scheduler restores the durable generation; the unchanged // fixture needs no republish.