From f6a2a8510f56e393371dba3b9964be8ca36ff255 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 01:31:35 +0000 Subject: [PATCH 1/5] fix(runtime-core): commit the missing background CPU module perf(observation) landed an import of tracedecay_runtime_core::background_cpu, but neither the module file nor its declaration was added, so the pushed integration branch does not compile: every branch cut from it fails on an unresolved import. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/background_cpu.rs | 461 ++++++++++++++++++ crates/tracedecay-runtime-core/src/lib.rs | 1 + 2 files changed, 462 insertions(+) create mode 100644 crates/tracedecay-runtime-core/src/background_cpu.rs diff --git a/crates/tracedecay-runtime-core/src/background_cpu.rs b/crates/tracedecay-runtime-core/src/background_cpu.rs new file mode 100644 index 0000000000..75366049a4 --- /dev/null +++ b/crates/tracedecay-runtime-core/src/background_cpu.rs @@ -0,0 +1,461 @@ +//! Process-wide weighted admission for background CPU work. +//! +//! The authority counts active CPU units rather than owning an executor. Code +//! indexing, semantic native threads, and session preparation can therefore +//! use their existing execution substrates while sharing one hard process +//! ceiling. FIFO waiter order prevents a continuously busy class from starving +//! another class, and RAII releases capacity on success, cancellation, or +//! unwind. + +use std::cell::Cell; +use std::collections::VecDeque; +use std::fmt; +use std::num::NonZeroUsize; +use std::sync::{Arc, Condvar, Mutex, OnceLock}; + +#[derive(Debug)] +struct BackgroundCpuWaiterV1 { + units: usize, +} + +#[derive(Default)] +struct BackgroundCpuStateV1 { + active_units: usize, + waiters: VecDeque>, +} + +/// One process-wide background CPU budget shared across subsystems. +pub struct ProcessBackgroundCpuV1 { + width: NonZeroUsize, + state: Mutex, + available: Condvar, +} + +impl fmt::Debug for ProcessBackgroundCpuV1 { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProcessBackgroundCpuV1") + .field("width", &self.width) + .field("active_units", &self.active_units()) + .finish_non_exhaustive() + } +} + +thread_local! { + static BACKGROUND_CPU_DEPTH: Cell = const { Cell::new(0) }; + static BACKGROUND_CPU_UNITS: Cell = const { Cell::new(0) }; +} + +struct BackgroundCpuScopeV1; + +impl BackgroundCpuScopeV1 { + fn enter() -> Self { + BACKGROUND_CPU_DEPTH.with(|depth| depth.set(depth.get().saturating_add(1))); + Self + } +} + +struct YieldedBackgroundCpuV1<'a> { + authority: &'a Arc, + units: usize, + depth: usize, +} + +impl Drop for YieldedBackgroundCpuV1<'_> { + fn drop(&mut self) { + self.authority.admit_units(self.units); + BACKGROUND_CPU_UNITS.with(|units| units.set(self.units)); + BACKGROUND_CPU_DEPTH.with(|depth| depth.set(self.depth)); + } +} + +impl Drop for BackgroundCpuScopeV1 { + fn drop(&mut self) { + BACKGROUND_CPU_DEPTH.with(|depth| { + let remaining = depth.get().saturating_sub(1); + depth.set(remaining); + if remaining == 0 { + BACKGROUND_CPU_UNITS.with(|units| units.set(0)); + } + }); + } +} + +impl ProcessBackgroundCpuV1 { + fn new(width: NonZeroUsize) -> Self { + Self { + width, + state: Mutex::new(BackgroundCpuStateV1::default()), + available: Condvar::new(), + } + } + + #[must_use] + pub const fn width(&self) -> NonZeroUsize { + self.width + } + + #[must_use] + pub fn active_units(&self) -> usize { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .active_units + } + + #[must_use] + pub fn waiting_work_units(&self) -> usize { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + waiting_units(&state) + } + + /// Acquire one CPU unit, waiting in FIFO order when the process budget is + /// full. The returned guard must remain alive for the active work unit. + pub fn acquire(self: &Arc) -> BackgroundCpuPermitV1 { + self.acquire_units(1) + } + + /// Acquire one CPU unit only when no earlier waiter exists and capacity is + /// immediately available. + pub fn try_acquire(self: &Arc) -> Option { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !state.waiters.is_empty() || state.active_units >= self.width.get() { + return None; + } + state.active_units += 1; + record_state(&state, self.width); + Some(BackgroundCpuPermitV1 { + authority: Arc::clone(self), + units: 1, + }) + } + + /// Run one active work unit under the process budget. Nested work on the + /// same thread reuses a sufficient parent admission instead of waiting on + /// itself. + pub fn with_permit(self: &Arc, operation: impl FnOnce() -> R) -> R { + self.with_permits(1, operation) + } + + /// Run a weighted work unit, clamped to the entire process width. Semantic + /// inference uses its native intra-op thread count as the weight; ordinary + /// index/session preparation uses one. + pub fn with_permits( + self: &Arc, + requested_units: usize, + operation: impl FnOnce() -> R, + ) -> R { + let units = requested_units.max(1).min(self.width.get()); + let active_units = BACKGROUND_CPU_UNITS.with(Cell::get); + if active_units >= units { + return operation(); + } + if active_units > 0 { + let depth = BACKGROUND_CPU_DEPTH.with(Cell::get); + self.release(active_units); + BACKGROUND_CPU_UNITS.with(|active| active.set(0)); + BACKGROUND_CPU_DEPTH.with(|active| active.set(0)); + let _restore = YieldedBackgroundCpuV1 { + authority: self, + units: active_units, + depth, + }; + let _permit = self.acquire_units(units); + let _scope = BackgroundCpuScopeV1::enter(); + BACKGROUND_CPU_UNITS.with(|active| active.set(units)); + return operation(); + } + let _permit = self.acquire_units(units); + let _scope = BackgroundCpuScopeV1::enter(); + BACKGROUND_CPU_UNITS.with(|active| active.set(units)); + operation() + } + + /// Temporarily yield the caller's active units while a nested executor + /// fans out independently admitted leaf work. This prevents a parent + /// Rayon worker from holding capacity while it waits for child workers, + /// including a full-width weighted child. Capacity is reacquired before + /// the parent resumes, including during unwind. + pub fn with_yielded_permits(self: &Arc, operation: impl FnOnce() -> R) -> R { + let units = BACKGROUND_CPU_UNITS.with(Cell::get); + if units == 0 { + return operation(); + } + let depth = BACKGROUND_CPU_DEPTH.with(Cell::get); + self.release(units); + BACKGROUND_CPU_UNITS.with(|active| active.set(0)); + BACKGROUND_CPU_DEPTH.with(|active| active.set(0)); + let _restore = YieldedBackgroundCpuV1 { + authority: self, + units, + depth, + }; + operation() + } + + fn acquire_units(self: &Arc, units: usize) -> BackgroundCpuPermitV1 { + self.admit_units(units); + BackgroundCpuPermitV1 { + authority: Arc::clone(self), + units, + } + } + + fn admit_units(&self, units: usize) { + let waiter = Arc::new(BackgroundCpuWaiterV1 { units }); + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.waiters.push_back(Arc::clone(&waiter)); + record_state(&state, self.width); + loop { + let is_front = state + .waiters + .front() + .is_some_and(|front| Arc::ptr_eq(front, &waiter)); + if is_front && state.active_units.saturating_add(waiter.units) <= self.width.get() { + state.waiters.pop_front(); + state.active_units += waiter.units; + record_state(&state, self.width); + self.available.notify_all(); + return; + } + state = self + .available + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + + fn release(&self, units: usize) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + debug_assert!(state.active_units >= units); + state.active_units = state.active_units.saturating_sub(units); + record_state(&state, self.width); + self.available.notify_all(); + } +} + +fn record_state(state: &BackgroundCpuStateV1, width: NonZeroUsize) { + hotpath::gauge!("runtime_core.background_cpu.width").set(width.get()); + hotpath::gauge!("runtime_core.background_cpu.active_units").set(state.active_units); + hotpath::gauge!("runtime_core.background_cpu.waiting_work_units").set(waiting_units(state)); +} + +fn waiting_units(state: &BackgroundCpuStateV1) -> usize { + state + .waiters + .iter() + .fold(0usize, |total, waiter| total.saturating_add(waiter.units)) +} + +/// RAII ownership of active CPU capacity. Dropping it is cancellation-safe and +/// releases the exact acquired weight. +pub struct BackgroundCpuPermitV1 { + authority: Arc, + units: usize, +} + +impl fmt::Debug for BackgroundCpuPermitV1 { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("BackgroundCpuPermitV1") + .field("units", &self.units) + .finish_non_exhaustive() + } +} + +impl Drop for BackgroundCpuPermitV1 { + fn drop(&mut self) { + self.authority.release(self.units); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum BackgroundCpuInstallErrorV1 { + #[error( + "background CPU authority is already installed at width {installed_width}, not requested width {requested_width}" + )] + ConflictingWidth { + installed_width: usize, + requested_width: usize, + }, + #[error("background CPU authority installation did not settle")] + InstallationDidNotSettle, +} + +static PROCESS_BACKGROUND_CPU: OnceLock> = OnceLock::new(); + +/// Install or idempotently reuse the one process background CPU authority. +pub fn install_process_background_cpu( + width: NonZeroUsize, +) -> Result, BackgroundCpuInstallErrorV1> { + if let Some(installed) = PROCESS_BACKGROUND_CPU.get() { + return compare_installed_width(installed, width); + } + let requested = Arc::new(ProcessBackgroundCpuV1::new(width)); + match PROCESS_BACKGROUND_CPU.set(Arc::clone(&requested)) { + Ok(()) => Ok(requested), + Err(_) => PROCESS_BACKGROUND_CPU.get().map_or_else( + || Err(BackgroundCpuInstallErrorV1::InstallationDidNotSettle), + |installed| compare_installed_width(installed, width), + ), + } +} + +fn compare_installed_width( + installed: &Arc, + requested: NonZeroUsize, +) -> Result, BackgroundCpuInstallErrorV1> { + if installed.width == requested { + Ok(Arc::clone(installed)) + } else { + Err(BackgroundCpuInstallErrorV1::ConflictingWidth { + installed_width: installed.width.get(), + requested_width: requested.get(), + }) + } +} + +/// Installed process authority, or `None` before daemon worker-plan admission. +#[must_use] +pub fn process_background_cpu() -> Option> { + PROCESS_BACKGROUND_CPU.get().map(Arc::clone) +} + +#[cfg(test)] +mod tests { + use std::num::NonZeroUsize; + use std::sync::{ + Arc, Barrier, + atomic::{AtomicUsize, Ordering}, + }; + use std::time::Duration; + + use super::*; + + #[test] + fn combined_classes_never_exceed_width_and_both_progress() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + let active = Arc::new(AtomicUsize::new(0)); + let maximum = Arc::new(AtomicUsize::new(0)); + let index_completed = Arc::new(AtomicUsize::new(0)); + let session_completed = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(Barrier::new(17)); + let mut workers = Vec::new(); + for ordinal in 0..16 { + let authority = Arc::clone(&authority); + let active = Arc::clone(&active); + let maximum = Arc::clone(&maximum); + let index_completed = Arc::clone(&index_completed); + let session_completed = Arc::clone(&session_completed); + let start = Arc::clone(&start); + workers.push(std::thread::spawn(move || { + start.wait(); + authority.with_permit(|| { + let current = active.fetch_add(1, Ordering::SeqCst) + 1; + maximum.fetch_max(current, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(5)); + active.fetch_sub(1, Ordering::SeqCst); + if ordinal % 2 == 0 { + index_completed.fetch_add(1, Ordering::SeqCst); + } else { + session_completed.fetch_add(1, Ordering::SeqCst); + } + }); + })); + } + start.wait(); + for worker in workers { + worker.join().expect("background worker"); + } + + assert!(maximum.load(Ordering::SeqCst) <= 4); + assert_eq!(index_completed.load(Ordering::SeqCst), 8); + assert_eq!(session_completed.load(Ordering::SeqCst), 8); + } + + #[test] + fn waiting_demand_sums_weighted_work_units() { + let state = BackgroundCpuStateV1 { + active_units: 4, + waiters: VecDeque::from([ + Arc::new(BackgroundCpuWaiterV1 { units: 4 }), + Arc::new(BackgroundCpuWaiterV1 { units: 1 }), + ]), + }; + + assert_eq!(waiting_units(&state), 5); + } + + #[test] + fn weighted_units_and_nested_work_share_one_width() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + authority.with_permits(4, || { + assert_eq!(authority.active_units(), 4); + authority.with_permit(|| assert_eq!(authority.active_units(), 4)); + }); + authority.with_permit(|| { + assert_eq!(authority.active_units(), 1); + authority.with_permits(4, || assert_eq!(authority.active_units(), 4)); + assert_eq!(authority.active_units(), 1); + }); + assert_eq!(authority.active_units(), 0); + } + + #[test] + fn nested_executor_yields_parent_units_and_reacquires_them() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + authority.with_permit(|| { + assert_eq!(authority.active_units(), 1); + authority.with_yielded_permits(|| { + assert_eq!(authority.active_units(), 0); + authority.with_permits(4, || assert_eq!(authority.active_units(), 4)); + }); + assert_eq!(authority.active_units(), 1); + }); + assert_eq!(authority.active_units(), 0); + } + + #[test] + fn panic_and_cancellation_drop_release_every_unit() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(2).expect("nonzero width"), + )); + let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + authority.with_permits(2, || panic!("injected background panic")); + })); + assert!(panic.is_err()); + assert_eq!(authority.active_units(), 0); + + let nested_panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + authority.with_permit(|| { + authority.with_yielded_permits(|| panic!("injected nested executor panic")); + }); + })); + assert!(nested_panic.is_err()); + assert_eq!(authority.active_units(), 0); + + let cancelled = authority.acquire(); + assert_eq!(authority.active_units(), 1); + drop(cancelled); + assert_eq!(authority.active_units(), 0); + assert!(authority.try_acquire().is_some()); + } +} diff --git a/crates/tracedecay-runtime-core/src/lib.rs b/crates/tracedecay-runtime-core/src/lib.rs index 7a9d0ef453..7a176ffb26 100644 --- a/crates/tracedecay-runtime-core/src/lib.rs +++ b/crates/tracedecay-runtime-core/src/lib.rs @@ -76,6 +76,7 @@ #![allow(rustdoc::broken_intra_doc_links)] #![allow(rustdoc::private_intra_doc_links)] +pub mod background_cpu; pub mod branch; pub mod branch_meta; pub mod cancellation; From 5d4d51b635e45b41522b757d0ccc3de81f970dde Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 02:11:50 +0000 Subject: [PATCH 2/5] fix(build): land the batching commit's unpushed consumers perf(observation) added an ObservationApplicationError variant and a HostAdmissionOutcome field without updating four consumers, and code_index_scheduler references four parallelism items whose file was never git added. The pushed branch therefore does not compile, so every branch cut from it fails before its own work is even built. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 1 + crates/tracedecay-sessions/src/runtime/cline_like.rs | 1 + crates/tracedecay-sessions/src/runtime/ingest/failure.rs | 5 +++++ crates/tracedecay-sessions/src/runtime/ingest/tests.rs | 2 ++ crates/tracedecay-sessions/src/runtime/kiro.rs | 1 + 5 files changed, 10 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index fd8e61aa2a..e0220635f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6143,6 +6143,7 @@ dependencies = [ "tracedecay-code-extraction", "tracedecay-domain", "tracedecay-graph-db", + "tracedecay-runtime-core", "tracedecay-rusqlite-runtime", "tracedecay-store", "tree-sitter", diff --git a/crates/tracedecay-sessions/src/runtime/cline_like.rs b/crates/tracedecay-sessions/src/runtime/cline_like.rs index 697be26a60..225df03015 100644 --- a/crates/tracedecay-sessions/src/runtime/cline_like.rs +++ b/crates/tracedecay-sessions/src/runtime/cline_like.rs @@ -1414,6 +1414,7 @@ mod observation_tests { status: HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_unavailable"), + recovery: None, }, ); assert!(matches!( diff --git a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs index f4e78e63d2..2cede6fc6e 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs @@ -531,6 +531,11 @@ pub fn classify_claude_observation_failure( crate::observation::ObservationApplicationError::BatchContainsNonDurable => { permanent("observation_batch_non_durable") } + // The worker went away before reaching a verdict, so nothing was + // decided about the payload. Re-running the same input can succeed. + crate::observation::ObservationApplicationError::BatchWorkerStopped => { + unavailable("observation_batch_worker_stopped") + } }, Ingest::MissingParsedRecord => permanent("observation_parsed_record_missing"), Ingest::InvalidFrameState => permanent("observation_frame_state_invalid"), diff --git a/crates/tracedecay-sessions/src/runtime/ingest/tests.rs b/crates/tracedecay-sessions/src/runtime/ingest/tests.rs index 5debd4682c..67487406f2 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/tests.rs @@ -234,6 +234,7 @@ fn still_mounting_admission_failures_keep_the_admission_retryability() { status: crate::admission::HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_write_failed"), + recovery: None, }, ); @@ -254,6 +255,7 @@ fn permanent_admission_failures_still_classify_permanent() { status: crate::admission::HostAdmissionStatus::Degraded, retryable: false, reason_code: Some("invalid_observation_contract"), + recovery: None, }, ); diff --git a/crates/tracedecay-sessions/src/runtime/kiro.rs b/crates/tracedecay-sessions/src/runtime/kiro.rs index 1f37377a6e..3070457e76 100644 --- a/crates/tracedecay-sessions/src/runtime/kiro.rs +++ b/crates/tracedecay-sessions/src/runtime/kiro.rs @@ -1332,6 +1332,7 @@ mod observation_tests { status: HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_unavailable"), + recovery: None, }, ); assert!(matches!( From 3b61fc6051e06697004adb7e87a01abe9fda8347 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 02:30:48 +0000 Subject: [PATCH 3/5] fix(build): cover the root crate's base breakage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CARRIED BASE FIX — not part of the daemon/MCP hygiene work, and safe to drop once the peer's unpushed commits reach origin. PR #681 unbreaks the workspace crates but leaves three root-crate sites that still do not compile on this base: - `HostAdmissionOutcome` gained a `recovery` field; two initializers in the root crate were not updated. Set to `None`, matching every other site the field was added to. - The hooks-crate extraction left `HookCompletedReadinessDistributions` counters unreachable from the root observation benchmark. These are the accessor hunks of origin/claude/fix-extraction-visibility (#675), verbatim; the rest of that commit depends on unpushed peer work and is deliberately not carried. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/hooks/analytics/readiness.rs | 25 +++++++++++++++++++ src/host_admission.rs | 1 + .../integration_test_support.rs | 1 + .../claude_observation_benchmark/baseline.rs | 12 +++++---- 4 files changed, 34 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay-agent-hosts/src/hooks/analytics/readiness.rs b/crates/tracedecay-agent-hosts/src/hooks/analytics/readiness.rs index 20b485a129..c3e06e86c4 100644 --- a/crates/tracedecay-agent-hosts/src/hooks/analytics/readiness.rs +++ b/crates/tracedecay-agent-hosts/src/hooks/analytics/readiness.rs @@ -179,6 +179,31 @@ pub struct HookCompletedReadinessDistributions { pub(crate) disposition_values_folded_to_unknown: u64, } +impl HookCompletedReadinessDistributions { + /// Readiness counters read by the root-crate observation benchmark. The + /// struct moved into this crate, so its `pub(crate)` fields are no longer + /// reachable from there; these expose exactly the five it folds. + pub fn source_event(&self) -> &str { + &self.source_event + } + + pub fn input_rows_received(&self) -> u64 { + self.input_rows_received + } + + pub fn input_rows_processed(&self) -> u64 { + self.input_rows_processed + } + + pub fn input_rows_dropped_at_cap(&self) -> u64 { + self.input_rows_dropped_at_cap + } + + pub fn events_considered(&self) -> u64 { + self.events_considered + } +} + #[derive(Default)] struct MutableNumericSummary { present_count: u64, diff --git a/src/host_admission.rs b/src/host_admission.rs index 592ab6fc23..1f7cbe2d09 100644 --- a/src/host_admission.rs +++ b/src/host_admission.rs @@ -1117,6 +1117,7 @@ const fn registered_authority_unavailable_outcome() -> HostAdmissionOutcome { status: HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("registered_authority_unavailable"), + recovery: None, } } diff --git a/src/host_admission/integration_test_support.rs b/src/host_admission/integration_test_support.rs index 6e5d8e3d06..ad18f681fa 100644 --- a/src/host_admission/integration_test_support.rs +++ b/src/host_admission/integration_test_support.rs @@ -290,6 +290,7 @@ impl HostAdmissionTestRuntimeV1 { status: HostAdmissionStatus::Unavailable, retryable: false, reason_code: Some("project_authority_unbound"), + recovery: None, })?; self.facade() .get_source_cursor( diff --git a/src/sessions/claude_observation_benchmark/baseline.rs b/src/sessions/claude_observation_benchmark/baseline.rs index 82cee2e402..03ae10e23a 100644 --- a/src/sessions/claude_observation_benchmark/baseline.rs +++ b/src/sessions/claude_observation_benchmark/baseline.rs @@ -266,7 +266,7 @@ pub(super) fn validate_hook_telemetry_readiness() { ); assert_eq!(readiness.unavailable_measurements.len(), 1); assert_eq!( - readiness.readiness_distributions.source_event, + readiness.readiness_distributions.source_event(), "hook_completed" ); assert_eq!( @@ -274,13 +274,15 @@ pub(super) fn validate_hook_telemetry_readiness() { .expect("serialize empty readiness distributions")["collection_status"], "no_samples" ); - assert_eq!(readiness.readiness_distributions.input_rows_received, 0); - assert_eq!(readiness.readiness_distributions.input_rows_processed, 0); + assert_eq!(readiness.readiness_distributions.input_rows_received(), 0); + assert_eq!(readiness.readiness_distributions.input_rows_processed(), 0); assert_eq!( - readiness.readiness_distributions.input_rows_dropped_at_cap, + readiness + .readiness_distributions + .input_rows_dropped_at_cap(), 0 ); - assert_eq!(readiness.readiness_distributions.events_considered, 0); + assert_eq!(readiness.readiness_distributions.events_considered(), 0); assert_eq!( readiness.canonical_contract["latency_semantics"]["host_ipc_rtt"]["event_field"], "daemon_rtt_us" From c645d67bf48e72dd26e0569c743aa21014075325 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 02:49:43 +0000 Subject: [PATCH 4/5] perf(mcp): cache the static tools/list catalog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `get_maximal_tool_definitions` assembled every ToolDefinition and its JSON Schema on each call, and `tools/list` is on that path (`mcp.tools_list.compose`). The registry is a pure function of static data: the application catalog is already a LazyLock snapshot and `ast_grep_available` is already a OnceLock host probe. Assemble it once per process and clone per caller. Nothing session-scoped is frozen into the cache. The per-session passes — the context call budget, the warming budget, and the profile, capability, and scope filtering — already ran after the registry was built and still do, on the clone. A test pins that a later session's budget cannot rewrite an earlier one's definitions. The dispatch error is not Clone, and a failure here is a deterministic catalog defect rather than a transient condition, so the cache retains the rendered message and replays it. Co-Authored-By: Claude Opus 5 (1M context) --- src/mcp/tools/definitions.rs | 35 +++++++++++++ src/mcp/tools/definitions/tests.rs | 79 ++++++++++++++++++++++++++++++ 2 files changed, 114 insertions(+) diff --git a/src/mcp/tools/definitions.rs b/src/mcp/tools/definitions.rs index 5806e74951..94bafa17a8 100644 --- a/src/mcp/tools/definitions.rs +++ b/src/mcp/tools/definitions.rs @@ -11,6 +11,7 @@ use serde_json::{Value, json}; use std::collections::BTreeSet; +use std::sync::LazyLock; use tracedecay_tool_catalog::{CapabilityId, FeatureId, ProfileId, ScopeDimension}; use super::ToolDefinition; @@ -427,8 +428,42 @@ pub fn get_tool_definitions() Ok(definitions) } +/// Counts how many times the maximal registry was actually assembled. +/// +/// The registry is deterministic, so a correct cache builds it exactly once +/// per process no matter how many `tools/list` requests arrive. +#[cfg(test)] +pub(super) static MAXIMAL_DEFINITION_BUILDS: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + +/// The maximal registry, assembled once per process and cloned per caller. +/// +/// Every input is static for the life of the process: the application catalog +/// is a `LazyLock` snapshot and `ast_grep_available()` is a `OnceLock` host +/// probe. Nothing session-scoped is frozen here — the per-session passes +/// (`apply_context_budget`, `apply_context_warming_budget`, and the +/// profile/capability filtering in +/// `get_catalog_filtered_tool_definitions_with_budget`) all run on the *clone* +/// this returns, after the cache. pub(super) fn get_maximal_tool_definitions() -> Result, super::dispatch::McpDispatchMetadataError> { + // The error type is not `Clone`, and a failure here is a deterministic + // catalog/schema defect rather than a transient condition, so the cache + // retains the rendered message and replays it. + static MAXIMAL_DEFINITIONS: LazyLock, String>> = + LazyLock::new(|| build_maximal_tool_definitions().map_err(|error| error.to_string())); + match &*MAXIMAL_DEFINITIONS { + Ok(definitions) => Ok(definitions.clone()), + Err(message) => Err(super::dispatch::McpDispatchMetadataError::Initialization( + message.clone(), + )), + } +} + +fn build_maximal_tool_definitions() +-> Result, super::dispatch::McpDispatchMetadataError> { + #[cfg(test)] + MAXIMAL_DEFINITION_BUILDS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); let application_registry = tracedecay_application::mcp_executable_binding_registry().map_err(|error| { super::dispatch::McpDispatchMetadataError::Initialization(error.to_string()) diff --git a/src/mcp/tools/definitions/tests.rs b/src/mcp/tools/definitions/tests.rs index 3ebfe923d1..69fb0756a4 100644 --- a/src/mcp/tools/definitions/tests.rs +++ b/src/mcp/tools/definitions/tests.rs @@ -309,3 +309,82 @@ fn lcm_compatibility_definitions_expose_only_opaque_continuation_cursors() { "relevance" ); } + +/// The MCP tool catalog is static per build, so `tools/list` must not +/// re-assemble every JSON schema on each request. +/// +/// Falsifiable on a count, never a duration: the assembly counter may advance +/// at most once for the whole process, however many callers ask for it. +#[test] +fn maximal_tool_definitions_are_assembled_once_per_process() { + use std::sync::atomic::Ordering; + + let first = get_maximal_tool_definitions().expect("tool definitions"); + // Read the baseline *after* the first call so the one legitimate build is + // already counted; a cached registry can never advance it again. + let baseline = MAXIMAL_DEFINITION_BUILDS.load(Ordering::SeqCst); + + for _ in 0..8 { + let again = get_maximal_tool_definitions().expect("tool definitions"); + assert_eq!( + again.len(), + first.len(), + "the cached registry must serve the same tool set" + ); + } + + assert_eq!( + MAXIMAL_DEFINITION_BUILDS.load(Ordering::SeqCst), + baseline, + "the maximal tool registry was re-assembled after it had already been \ + built; tools/list rebuilds the whole catalog per request" + ); +} + +/// Caching the registry must not freeze anything session-scoped into it. +/// +/// The per-session passes mutate the vector they are handed, so every caller +/// has to receive an independent clone. If the cache handed out shared state, +/// one session's context budget would be visible to the next. +#[test] +fn per_session_budget_does_not_leak_through_the_cached_registry() { + fn context_description(definitions: &[ToolDefinition]) -> String { + definitions + .iter() + .find(|definition| definition.name == "tracedecay_context") + .map(|definition| definition.description.clone()) + .expect("tracedecay_context is advertised") + } + + let small = get_tool_definitions_with_budget(11, 2).expect("tool definitions"); + let small_description = context_description(&small); + assert!( + small_description.contains("2 calls maximum"), + "budget must reach the context description: {small_description}" + ); + + let large = get_tool_definitions_with_budget(999_999, 9).expect("tool definitions"); + let large_description = context_description(&large); + assert!( + large_description.contains("9 calls maximum"), + "budget must reach the context description: {large_description}" + ); + + assert_ne!( + small_description, large_description, + "two sessions with different budgets must not share one description" + ); + assert_eq!( + context_description(&small), + small_description, + "the earlier session's definitions must not be rewritten by a later one" + ); + + // A third, unbudgeted read must still see the neutral registry. + let neutral = get_tool_definitions().expect("tool definitions"); + let neutral_description = context_description(&neutral); + assert!( + !neutral_description.contains("9 calls maximum"), + "an unbudgeted caller inherited another session's budget: {neutral_description}" + ); +} From 2ef09960f6a5176eab48fc801c127d9ad6c148c0 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 03:07:24 +0000 Subject: [PATCH 5/5] test(daemon): pin connection panic isolation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both accept loops serve clients from a JoinSet and keep draining join_next, and the workspace does not build with panic=abort, so a panicking connection already unwinds into a contained JoinError rather than taking the daemon down. Nothing was broken; what was missing is a guard on the part that fails silently. DaemonClientPermit has no Drop of its own — the semaphore permit is released when the last Arc lease drops. With 64 concurrent client slots, one permit stranded per panic degrades the daemon into refusing every client, with no error that names the cause. Covers both unwind paths, because they leave admission in different states: a connection that panics still holding its slot, and one that panics while parked, after park_admission surrendered the slot and before it could re-acquire. Asserted on task outcomes and permit counts, never on elapsed time. Verified falsifiable: leaking the permit in with_connection_admission fails it with "stranded its admission slot: 3 of 4 free". Co-Authored-By: Claude Opus 5 (1M context) --- src/daemon/tests/lifecycle.rs | 123 ++++++++++++++++++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/src/daemon/tests/lifecycle.rs b/src/daemon/tests/lifecycle.rs index aa935aeeca..b5baa20fe6 100644 --- a/src/daemon/tests/lifecycle.rs +++ b/src/daemon/tests/lifecycle.rs @@ -1217,3 +1217,126 @@ async fn draining_waits_for_one_bounded_in_flight_request() { client.await.expect("client task should finish"); assert!(lifecycle.try_enter().is_none()); } + +/// A panicking connection must not take down the daemon, wedge its peers, or +/// strand the admission slot it was holding. +/// +/// Both accept loops serve every client from a `JoinSet` and keep draining +/// `join_next`, so one connection unwinding has to be contained at that task +/// boundary. Two failure shapes are covered, because they unwind through +/// different admission states: +/// +/// * a connection that panics while it still holds its slot, and +/// * a connection that panics *while parked* — `park_admission` has already +/// surrendered the slot and will never reach its re-acquire, so the unwind +/// must not double-release it or leave it unaccounted. +/// +/// Asserted on task outcomes and permit counts, never on elapsed time. +#[tokio::test] +async fn a_panicking_connection_isolates_from_its_peers_and_returns_its_permit() { + use super::super::{ + DaemonClientAdmission, DaemonClientAdmissionClass, DaemonClientAdmissionOutcome, + }; + + const GENERAL_CAPACITY: usize = 4; + const PANICKING_CLIENTS: usize = 2; + + let admission = DaemonClientAdmission::with_reserved_capacity(GENERAL_CAPACITY + 1, 1); + let baseline = admission.available_general_permits(); + assert_eq!(baseline, GENERAL_CAPACITY); + + let (barrier, _) = tokio::sync::watch::channel(false); + let mut clients: tokio::task::JoinSet<&'static str> = tokio::task::JoinSet::new(); + + for index in 0..GENERAL_CAPACITY { + let permit = match admission.try_admit() { + DaemonClientAdmissionOutcome::Admitted(permit) => permit, + DaemonClientAdmissionOutcome::Saturated(response) => { + panic!("client rejected below capacity: {response:?}") + } + }; + assert_eq!(permit.class(), DaemonClientAdmissionClass::General); + let mut lifted = barrier.subscribe(); + clients.spawn(super::super::with_connection_admission( + permit, + async move { + // Client 0 unwinds immediately, still holding its slot. + assert!(index != 0, "connection task panicked holding its slot"); + super::super::park_admission(async move { + while !*lifted.borrow_and_update() { + lifted.changed().await.expect("barrier sender retained"); + } + // Client 1 unwinds from inside the park, after the slot was + // surrendered and before it could be re-acquired. + assert!(index != 1, "connection task panicked while parked"); + }) + .await; + "connection completed" + }, + )); + } + + // Every slot comes back while the peers are still parked: the panicking + // client released its permit on the way out rather than stranding it. + let returned = tokio::time::timeout(std::time::Duration::from_secs(10), async { + while admission.available_general_permits() < GENERAL_CAPACITY { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await; + assert!( + returned.is_ok(), + "a panicking connection stranded its admission slot: {} of {GENERAL_CAPACITY} free", + admission.available_general_permits() + ); + + barrier.send(true).expect("lift barrier"); + + let mut completed = 0usize; + let mut panicked = 0usize; + while let Some(joined) = clients.join_next().await { + match joined { + Ok(outcome) => { + assert_eq!(outcome, "connection completed"); + completed += 1; + } + Err(error) => { + assert!( + error.is_panic(), + "a connection task ended for a reason other than its own panic: {error}" + ); + panicked += 1; + } + } + } + + // The accept loop observed each panic as a contained task outcome, and the + // peers of a panicking connection still ran to completion. + assert_eq!( + panicked, PANICKING_CLIENTS, + "the JoinSet must surface every panic as a contained task result" + ); + assert_eq!( + completed, + GENERAL_CAPACITY - PANICKING_CLIENTS, + "a panicking connection wedged its peers" + ); + + // No permit leaks across either unwind path, so the daemon still admits. + assert_eq!( + admission.available_general_permits(), + baseline, + "admission permits leaked across a connection panic" + ); + let next = match admission.try_admit() { + DaemonClientAdmissionOutcome::Admitted(permit) => permit, + DaemonClientAdmissionOutcome::Saturated(response) => { + panic!("the daemon stopped admitting clients after a panic: {response:?}") + } + }; + assert_eq!( + next.class(), + DaemonClientAdmissionClass::General, + "recovered capacity must be general, not the reserved control lane" + ); +}