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. diff --git a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs index 06cc07fb88..deb3f268e7 100644 --- a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs +++ b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs @@ -19,7 +19,7 @@ async fn ready_reading( .await .unwrap_or_else(|error| panic!("{phase} readiness read failed: {error:?}")); let CodeIndexReadinessWaitReadV1::Reached { reading } = reading else { - panic!("exact worktree did not become ready: {reading:?}"); + panic!("exact worktree did not become ready during {phase}: {reading:?}"); }; *reading } diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs index 51fe23fcd5..1269026bf3 100644 --- a/crates/tracedecay/tests/common/mod.rs +++ b/crates/tracedecay/tests/common/mod.rs @@ -1278,9 +1278,10 @@ fn spawn_tracedecay_daemon_process( #[cfg(unix)] daemon.release_socket_on_stop(socket_path.clone()); - // Startup has no product deadline. The daemon logs `daemon_listening` - // once its endpoint is bound and published, so that line is readiness; - // stderr closing first means it exited. + // Startup has no product deadline. The daemon logs `daemon_ready` + // after bootstrap, immediately before polling its accept loop. + // `daemon_listening` only proves bind: setup can still hold connections + // in the backlog. Stderr closing before readiness means it exited. let Some(stderr) = daemon.child.stderr.take() else { // A caller redirected stderr, so only the endpoint can show readiness. loop { @@ -1321,7 +1322,7 @@ fn spawn_tracedecay_daemon_process( startup.trim() ); } - if startup[line_start..].starts_with("[tracedecay] event=daemon_listening ") { + if startup[line_start..].starts_with("[tracedecay] event=daemon_ready ") { break; } } diff --git a/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs b/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs index 08de819f7a..1ac4f4fbea 100644 --- a/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs +++ b/crates/tracedecay/tests/storage_suite/multi_connection_test/ownership.rs @@ -207,19 +207,40 @@ fn twelve_mcp_cli_and_hook_clients_share_one_daemon_profile_store_owner() { let project_id = tracedecay_runtime_core::storage::default_profile_project_id(&project_path); let data_root = tracedecay_runtime_core::storage::profile_sharded_data_root(&profile_root, &project_id); - let host = tracedecay_domain::NativeHostIdentityV1::CursorDesktop; - let (_, capture_report) = tracedecay_hooks::HookSpoolV1::open( - data_root.join("hook-v2-spool").join(host.hook_key()), - tracedecay_hooks::HookSpoolConfigV1::stock(host), - tracedecay_contracts::now_micros(), - ) - .expect("open native capture receipt"); - // Sequence survives daemon replay/acknowledgement, so this proves each - // callback was captured even if the owner already drained its envelope. + // A commit-lock timeout can stage an envelope whose first spool frame + // the daemon already acknowledged. Importing it appends a duplicate + // frame, so the sequence counts attempts rather than distinct callbacks. + // Each callback's completed receipt instead proves that its own capture + // landed, including one safely staged for replay, rather than returning + // the same transport response for an unbound hook. + let analytics = std::fs::read_to_string(data_root.join("hook_analytics.jsonl")) + .expect("read native capture outcomes"); + let mut captured_sessions = analytics + .lines() + .map(|line| serde_json::from_str::(line).expect("hook analytics row")) + .filter(|row| { + row["event"] == "hook_completed" + && row["agent"] == "cursor" + && row["hook_name"] == "afterFileEdit" + }) + .map(|row| { + assert_eq!( + row["disposition"]["status"], "accepted_for_replay", + "native callback must be durably captured: {row}" + ); + row["session_id"] + .as_str() + .expect("capture names the callback session") + .to_owned() + }) + .collect::>(); + captured_sessions.sort(); assert_eq!( - capture_report.next_sequence, - 1 + CONCURRENT_CLIENTS_PER_PATH as u64, - "every native callback must be captured, not merely accepted as unbound" + captured_sessions, + (0..CONCURRENT_CLIENTS_PER_PATH) + .map(|ordinal| format!("session-{ordinal}")) + .collect::>(), + "every native callback must have exactly one successful capture outcome" ); let daemon_stderr = std::fs::read_to_string(&daemon_stderr_path).expect("read daemon stderr"); assert!(