From 52bac3265caf4a61147611d91084858d4f3dcda7 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:54:29 -0800 Subject: [PATCH 1/9] fix(code-graph): isolate sealed proof repair on master Reapply the six exact issue-owned source and regression blobs from #4791 at 22c534b, without that branch's draft CLI-fixture ancestors. Keep authenticated snapshots across interrupted and overlapping proofs, real one-file successor and corruption regressions, and the validated shutdown and graph-tool corrections from the prior review lane. The original history and draft prerequisite remain intact on their own branches; this single-parent commit starts at master 66963c8. Exact-head Bazel and human review remain required. Refs #4507 --- .../src/feedback/concrete.rs | 73 ++++--- .../src/project_runtime/shutdown.rs | 25 ++- .../session_registry/code_graph_manifest.rs | 150 +++++++++++++ .../verified_sealed_segments.rs | 205 ++++++++++++++++-- .../tracedecay/src/daemon/engine/shutdown.rs | 84 ++++++- .../src/mcp/server/graph_tool_owner.rs | 141 +++++++++++- 6 files changed, 614 insertions(+), 64 deletions(-) diff --git a/crates/tracedecay-application/src/feedback/concrete.rs b/crates/tracedecay-application/src/feedback/concrete.rs index ad155e74195..1bbf8cd0313 100644 --- a/crates/tracedecay-application/src/feedback/concrete.rs +++ b/crates/tracedecay-application/src/feedback/concrete.rs @@ -268,29 +268,34 @@ impl ProjectFeedbackObservationSinkV1 { }) } - #[tracing::instrument(name = "usecases.feedback.close_drain", level = "trace", skip_all)] - async fn close_and_drain(&self) -> Result<(), FeedbackRuntimeError> { - let terminal = { - let _admission = self - .admission - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if self.closed.load(Ordering::Acquire) { - None - } else { - let terminal = self.prepare_terminal_envelope()?; - self.closed.store(true, Ordering::Release); - Some(terminal) + fn close_and_drain( + self: &Arc, + ) -> impl Future> + Send + 'static + use<> { + let terminal = self.close_admission(); + let sink = Arc::clone(self); + async move { + terminal?; + sink.join_worker().await + } + } + + fn close_admission(&self) -> Result<(), FeedbackRuntimeError> { + let _admission = self + .admission + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !self.closed.load(Ordering::Acquire) { + let terminal = self.prepare_terminal_envelope()?; + self.closed.store(true, Ordering::Release); + if self.control_sender.try_send(terminal).is_err() { + self.record_shutdown_failure(&FeedbackRuntimeError::ObservationWorker); } - }; - let terminal_outcome = if let Some(terminal) = terminal { - self.control_sender - .try_send(terminal) - .map_err(|_| FeedbackRuntimeError::ObservationWorker) - } else { - Ok(()) - }; + } + Ok(()) + } + #[tracing::instrument(name = "usecases.feedback.close_drain", level = "trace", skip_all)] + async fn join_worker(&self) -> Result<(), FeedbackRuntimeError> { let mut worker = self.worker.lock().await; let outcome = match worker.as_ref() { Some(task) => match task.wait().await { @@ -303,7 +308,6 @@ impl ProjectFeedbackObservationSinkV1 { None => Ok(()), }; worker.take(); - let outcome = terminal_outcome.and(outcome); if let Err(error) = &outcome { self.record_shutdown_failure(error); } @@ -525,10 +529,12 @@ impl FeedbackRuntime { }) } - /// Stop observation admission, persist the producer's terminal record, and - /// join the worker that owns the project database before store retirement. - pub async fn close_and_drain_observations(&self) -> Result<(), FeedbackRuntimeError> { - self.observation_sink.close_and_drain().await + /// Close admission immediately and return a join for the terminal write + /// and worker that owns the project database. + pub fn close_and_drain_observations( + &self, + ) -> impl Future> + Send + 'static + use<> { + self.observation_sink.close_and_drain() } pub fn owner(&self) -> Arc { @@ -2471,11 +2477,12 @@ mod tests { let boot_id = sink.boot_id.clone(); let worker_guard = sink.worker.lock().await; - let closing_sink = Arc::clone(&sink); - let close = tokio::spawn(async move { closing_sink.close_and_drain().await }); - while !sink.closed.load(Ordering::Acquire) { - tokio::task::yield_now().await; - } + let drain = sink.close_and_drain(); + assert!( + sink.closed.load(Ordering::Acquire), + "admission closes before joining the worker" + ); + let close = tokio::spawn(drain); close.abort(); assert!(close.await.unwrap_err().is_cancelled()); drop(worker_guard); @@ -2515,7 +2522,7 @@ mod tests { Err(()) }) .unwrap(); - let sink = ProjectFeedbackObservationSinkV1 { + let sink = Arc::new(ProjectFeedbackObservationSinkV1 { sender, control_sender, boot_id: canonical_sha256(&"feedback-terminal-failure").unwrap(), @@ -2525,7 +2532,7 @@ mod tests { closed: AtomicBool::new(false), shutdown_failure: AtomicU8::new(OBSERVATION_SHUTDOWN_CLEAN), worker: AsyncMutex::new(Some(worker)), - }; + }); assert!(matches!( sink.close_and_drain().await, diff --git a/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs b/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs index e44d9512207..bc79b87b698 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs @@ -185,8 +185,9 @@ impl ProjectRuntimeRegistryV1 { /// Shut every project runtime down and leave the registry empty. /// /// Routers become unavailable before feedback owners drop and Work - /// providers are joined. The drain is single-flight, so the first - /// caller's `deadline` bounds the producer drains for every caller. + /// providers are joined. The first caller supplies the observability flush + /// deadline for the shared task. The daemon bounds its own wait separately, + /// and the registry retains the task until its database workers finish. #[tracing::instrument( name = "daemon.service.project_runtime.shutdown", level = "trace", @@ -368,12 +369,22 @@ async fn shut_down_advisory(runtimes: &BTreeMap) -> Vec } async fn shut_down_feedback(runtimes: &BTreeMap) -> Vec { + // Start every project's terminal write before joining its independent + // database worker. The retained drain keeps ownership of each store + // through its terminal write, even if the daemon stops waiting. + let drains: Vec<_> = runtimes + .iter() + .filter_map(|(project_root, runtime)| { + let feedback = runtime.feedback.as_ref()?; + Some(( + project_root, + feedback.runtime().close_and_drain_observations(), + )) + }) + .collect(); let mut lost_work = Vec::new(); - for (project_root, runtime) in runtimes { - let Some(feedback) = runtime.feedback.as_ref() else { - continue; - }; - if let Err(error) = feedback.runtime().close_and_drain_observations().await { + for (project_root, drain) in drains { + if let Err(error) = drain.await { lost_work.push(format!( "{}: feedback observations: {error}", project_root.display() diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs index 1739c89b546..d83f14556e0 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs @@ -2674,9 +2674,159 @@ mod tests { &std::fs::read(fixture.scope_root.join("active-code-generation-v1.json")).unwrap(), ) .unwrap(); + let successor_digest = + SealedGraphStateDigest::try_from(pointer.state_digest.clone()).unwrap(); let successor_path = generations_root.join(pointer.generation_file); let mut successor: serde_json::Value = serde_json::from_slice(&std::fs::read(&successor_path).unwrap()).unwrap(); + + // A real one-file successor inherits unchanged source segments. + // Count production proof read callbacks rather than elapsed time: + // re-verifying the whole generation must fail even with warm OS caches. + let parent_file_digests = first_manifest["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .map(|segment| segment["segment_digest"].as_str().unwrap()) + .collect::>(); + let inherited_files = successor["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .filter(|segment| { + parent_file_digests.contains(segment["segment_digest"].as_str().unwrap()) + }) + .count(); + assert!( + inherited_files > 0, + "a one-file edit must retain sealed files" + ); + let count_proof_reads = |record: &VerifiedSealedSegmentsV1| { + let reads = AtomicUsize::new(0); + assert_eq!( + verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + record, + &|| { + reads.fetch_add(1, Ordering::SeqCst); + Ok(()) + }, + ), + Ok(()) + ); + reads.load(Ordering::SeqCst) + }; + let cold_reads = count_proof_reads(&VerifiedSealedSegmentsV1::default()); + let warm_reads = count_proof_reads(&verified); + if cfg!(unix) { + assert!( + cold_reads >= warm_reads + inherited_files, + "a one-file successor still re-verified inherited files: \ + cold={cold_reads}, warm={warm_reads}, inherited={inherited_files}" + ); + } else { + assert_eq!( + cold_reads, warm_reads, + "without reliable file fingerprints both passes must read all segments" + ); + } + + // A new segment must still be authenticated after the successor is + // warm. A proof that skips *all* segments could satisfy the read-count + // delta but would silently accept a corrupted edited file. + let changed_segment = successor["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .find(|segment| { + !parent_file_digests.contains(segment["segment_digest"].as_str().unwrap()) + }) + .expect("a one-file successor must seal a changed file"); + let changed_digest = tracedecay_domain::ManifestDigest::new( + changed_segment["segment_digest"] + .as_str() + .unwrap() + .to_owned(), + ) + .unwrap(); + let changed_path = super::segment_path(&fixture.segments_root, &changed_digest).unwrap(); + let original_bytes = std::fs::read(&changed_path).unwrap(); + assert!(!original_bytes.is_empty()); + let successor_record = VerifiedSealedSegmentsV1::default(); + assert_eq!( + verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + &successor_record, + &|| Ok(()), + ), + Ok(()) + ); + // The check callback also runs for manifest and lifecycle checks, + // so a positive count alone does not prove a segment was rehashed. + let warmed_reads = count_proof_reads(&successor_record); + let original_permissions = std::fs::metadata(&changed_path).unwrap().permissions(); + let mut writable_permissions = original_permissions.clone(); + #[allow(clippy::permissions_set_readonly_false)] + writable_permissions.set_readonly(false); + std::fs::set_permissions(&changed_path, writable_permissions).unwrap(); + let mut tampered_bytes = original_bytes.clone(); + tampered_bytes[0] ^= 1; + std::fs::write(&changed_path, tampered_bytes).unwrap(); + let corrupt_changed = verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + &successor_record, + &|| Ok(()), + ); + assert!( + matches!(corrupt_changed, Err(GraphDbError::Corrupt { .. })), + "a warmed successor accepted a corrupt changed segment: {corrupt_changed:?}" + ); + std::fs::write(&changed_path, original_bytes).unwrap(); + std::fs::set_permissions(&changed_path, original_permissions).unwrap(); + + // A failed proof must not evict the last complete source proof. + // Rechecking the repaired changed segment is necessary, but rehashing + // every untouched inherited segment would make recovery cold again. + let recovered_reads = count_proof_reads(&successor_record); + if cfg!(unix) { + assert!( + recovered_reads > warmed_reads, + "recovery did not re-authenticate repaired source: warmed={warmed_reads}, recovered={recovered_reads}" + ); + } else { + // Without fingerprint reuse, a warm proof still rehashes every + // segment. The corruption refusal above checks edited source. + assert_eq!( + recovered_reads, warmed_reads, + "a platform without fingerprints must rehash every segment" + ); + } + let rewarmed_reads = count_proof_reads(&successor_record); + assert_eq!( + rewarmed_reads, warmed_reads, + "successful recovery must restore the stable warm proof cost" + ); + let recovery_cold_reads = count_proof_reads(&VerifiedSealedSegmentsV1::default()); + if cfg!(unix) { + assert!( + recovery_cold_reads >= recovered_reads + inherited_files, + "a corrupt one-file successor evicted authenticated parent segments: \ + cold={recovery_cold_reads}, recovered={recovered_reads}, \ + inherited={inherited_files}" + ); + } else { + assert_eq!( + recovery_cold_reads, recovered_reads, + "platforms without reliable fingerprints must rehash every segment" + ); + } + let inherited = successor["generation"]["file_segments"] .as_array_mut() .unwrap() diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index a33657895de..7a44db51a02 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -12,7 +12,7 @@ use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use tracedecay_domain::ManifestDigest; @@ -53,30 +53,34 @@ impl SealedSegmentFingerprintV1 { } } -/// Per segments root, the segments of the generation last proven there. +type VerifiedSegmentMap = BTreeMap; + +/// Per segments root, the last successfully committed proof. In-flight proofs +/// share that immutable snapshot without evicting or copying its entries. #[derive(Debug, Default)] pub(crate) struct VerifiedSealedSegmentsV1 { - roots: Mutex>>, + roots: Mutex>>, } /// One proof's view: what it may skip, and what it hashed. pub(super) struct SealedSegmentProofV1<'v> { verified: &'v VerifiedSealedSegmentsV1, segments_root: PathBuf, - prior: BTreeMap, - proven: BTreeMap, + prior: Option>, + proven: VerifiedSegmentMap, } impl VerifiedSealedSegmentsV1 { - /// Start a proof under `segments_root`. The record leaves with it, so a - /// proof that fails leaves the next one to read every segment. + /// Start from the last successful proof. Cancellation, corruption or an + /// overlapping proof must not evict already authenticated identities. + /// New observations remain private until the whole proof succeeds. pub(super) fn begin(&self, segments_root: &Path) -> SealedSegmentProofV1<'_> { let prior = self .roots .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(segments_root) - .unwrap_or_default(); + .get(segments_root) + .cloned(); SealedSegmentProofV1 { verified: self, segments_root: segments_root.to_path_buf(), @@ -95,7 +99,12 @@ impl SealedSegmentProofV1<'_> { expected_size: u64, path: &Path, ) -> bool { - let Some(prior) = self.prior.get(digest).copied() else { + let Some(prior) = self + .prior + .as_ref() + .and_then(|prior| prior.get(digest)) + .copied() + else { return false; }; let current = path @@ -121,13 +130,179 @@ impl SealedSegmentProofV1<'_> { } } - /// The proof authenticated every segment: remember exactly this - /// generation's segments, so the record stays one generation large. + /// Remember exactly this generation, but do not let a late proof replace + /// a snapshot another successful proof has committed since we began. pub(super) fn commit(self) { - self.verified + let mut roots = self + .verified .roots .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert(self.segments_root, self.proven); + .unwrap_or_else(std::sync::PoisonError::into_inner); + let unchanged = match (roots.get(&self.segments_root), self.prior.as_ref()) { + (None, None) => true, + (Some(current), Some(prior)) => Arc::ptr_eq(current, prior), + _ => false, + }; + if unchanged { + roots.insert(self.segments_root, Arc::new(self.proven)); + } + } +} + +#[cfg(all(test, unix))] +mod tests { + use std::fs::{File, FileTimes}; + use std::io::Read; + use std::path::{Path, PathBuf}; + + use sha2::{Digest, Sha256}; + use tempfile::TempDir; + use tracedecay_domain::{ManifestDigest, sha256_hex_suffix}; + + use super::{SealedSegmentFingerprintV1, SealedSegmentProofV1, VerifiedSealedSegmentsV1}; + + fn write_segment(root: &Path, bytes: &[u8]) -> (ManifestDigest, PathBuf) { + let digest = ManifestDigest::from_sha256_bytes(&Sha256::digest(bytes)).unwrap(); + let path = root.join(format!( + "segment-{}.json", + sha256_hex_suffix(digest.as_str()).unwrap() + )); + std::fs::write(&path, bytes).unwrap(); + (digest, path) + } + + fn authenticate(proof: &mut SealedSegmentProofV1<'_>, digest: &ManifestDigest, path: &Path) { + let mut file = File::open(path).unwrap(); + let fingerprint = SealedSegmentFingerprintV1::of(&file.metadata().unwrap()); + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).unwrap(); + assert!(!bytes.is_empty()); + assert_eq!( + ManifestDigest::from_sha256_bytes(&Sha256::digest(&bytes)).unwrap(), + *digest + ); + proof.hashed(digest, fingerprint); + } + + #[test] + fn aborted_proof_keeps_parent_but_never_publishes_tentative_hashes() { + let root = TempDir::new().unwrap(); + let (parent, parent_path) = write_segment(root.path(), b"[1]"); + let (candidate, candidate_path) = write_segment(root.path(), b"[2]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &parent, &parent_path); + first.commit(); + + let mut interrupted = verified.begin(root.path()); + assert!(interrupted.already_verified(&parent, 3, &parent_path)); + // The reader records an identity before the caller checks its digest. + // A failed or cancelled proof must never publish that tentative fact. + std::fs::write(&candidate_path, b"[3]").unwrap(); + interrupted.hashed( + &candidate, + SealedSegmentFingerprintV1::of(&std::fs::metadata(&candidate_path).unwrap()), + ); + drop(interrupted); + + let mut retry = verified.begin(root.path()); + assert!( + retry.already_verified(&parent, 3, &parent_path), + "an interrupted successor must not make its authenticated parent cold" + ); + assert!( + !retry.already_verified(&candidate, 3, &candidate_path), + "a tentative digest from a failed proof must not become reusable" + ); + } + + #[test] + fn overlapping_proofs_share_parent_and_stale_commit_cannot_replace_successor() { + let root = TempDir::new().unwrap(); + let (parent, parent_path) = write_segment(root.path(), b"[1]"); + let (successor, successor_path) = write_segment(root.path(), b"[2]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &parent, &parent_path); + first.commit(); + + let mut stale = verified.begin(root.path()); + assert!(stale.already_verified(&parent, 3, &parent_path)); + let mut sibling = verified.begin(root.path()); + assert!( + sibling.already_verified(&parent, 3, &parent_path), + "an in-flight proof must not evict the parent's authenticated identities" + ); + drop(sibling); + let mut winner = verified.begin(root.path()); + authenticate(&mut winner, &successor, &successor_path); + winner.commit(); + stale.commit(); + + let mut latest = verified.begin(root.path()); + assert!(latest.already_verified(&successor, 3, &successor_path)); + assert!( + !latest.already_verified(&parent, 3, &parent_path), + "a stale proof must not resurrect segments pruned by a committed successor" + ); + } + + #[test] + fn retained_parent_rejects_same_size_replacement_with_restored_mtime() { + let root = TempDir::new().unwrap(); + let (digest, path) = write_segment(root.path(), b"[1]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &digest, &path); + first.commit(); + drop(verified.begin(root.path())); + + // Hold the original inode open so replacement cannot reuse it, even + // when the filesystem happens to assign identical timestamps. + let original = File::open(&path).unwrap(); + let metadata = original.metadata().unwrap(); + let before = SealedSegmentFingerprintV1::of(&metadata).unwrap(); + let (_, replacement_path) = write_segment(root.path(), b"[2]"); + File::options() + .write(true) + .open(&replacement_path) + .unwrap() + .set_times(FileTimes::new().set_modified(metadata.modified().unwrap())) + .unwrap(); + std::fs::rename(&replacement_path, &path).unwrap(); + let after = SealedSegmentFingerprintV1::of(&std::fs::metadata(&path).unwrap()).unwrap(); + assert_eq!(before.len, after.len); + assert_eq!(before.modified, after.modified); + assert_ne!(before.inode, after.inode); + assert_ne!( + ManifestDigest::from_sha256_bytes(&Sha256::digest(std::fs::read(&path).unwrap())) + .unwrap(), + digest + ); + let mut retry = verified.begin(root.path()); + assert!( + !retry.already_verified(&digest, 3, &path), + "retaining an old proof must not trust a replacement's same length and mtime" + ); + drop(original); + } + + #[test] + fn retained_proof_still_checks_root_extent_and_missing_source() { + let root = TempDir::new().unwrap(); + let other_root = TempDir::new().unwrap(); + let (digest, path) = write_segment(root.path(), b"[1]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &digest, &path); + first.commit(); + + let mut foreign = verified.begin(other_root.path()); + assert!(!foreign.already_verified(&digest, 3, &path)); + let mut retry = verified.begin(root.path()); + assert!(!retry.already_verified(&digest, 4, &path)); + assert!(retry.already_verified(&digest, 3, &path)); + std::fs::remove_file(&path).unwrap(); + assert!(!retry.already_verified(&digest, 3, &path)); } } diff --git a/crates/tracedecay/src/daemon/engine/shutdown.rs b/crates/tracedecay/src/daemon/engine/shutdown.rs index 33ddcf77eac..f8e0c4d1ce5 100644 --- a/crates/tracedecay/src/daemon/engine/shutdown.rs +++ b/crates/tracedecay/src/daemon/engine/shutdown.rs @@ -128,12 +128,19 @@ impl DaemonEngine { automation_join.shutdown_automation_schedulers().await; }, ), - ShutdownOwner::new("session_temporal_refresh", || {}, async move { - session_refresh.shutdown().await; - refresh_services - .release_profile_session_refresh_services() - .await; - }), + ShutdownOwner::new( + "session_temporal_refresh", + { + let refresh_cancel = Arc::clone(&session_refresh); + move || refresh_cancel.cancel_supervisors() + }, + async move { + session_refresh.shutdown().await; + refresh_services + .release_profile_session_refresh_services() + .await; + }, + ), ShutdownOwner::new( "host_admission_replay", { @@ -315,3 +322,68 @@ impl DaemonEngine { .await } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use tracedecay_daemon_service::shutdown::prepare_shutdown_owner_phases; + use tracedecay_domain::ProjectId; + use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; + use tracedecay_session_runtime::StoreOwnerKey; + + use super::DaemonEngine; + + #[tokio::test] + async fn session_refresh_is_cancelled_even_if_shutdown_joins_are_dropped() { + let directory = tempfile::tempdir().unwrap(); + let profile_root = directory.path().join("profile"); + let project_root = directory.path().join("project"); + let project_id = ProjectId::new("project.refresh-shutdown").unwrap(); + let fixture = RegisteredGlobalDbTestRuntime::project( + &profile_root, + &project_root, + project_id.clone(), + ) + .await + .unwrap(); + let database = fixture.project_database_arc().unwrap(); + let store_root = database.db_path().parent().unwrap().to_path_buf(); + let owner = StoreOwnerKey { + global_db_path: profile_root.join("global.db"), + profile_root, + project_id: Some(project_id.as_str().to_owned()), + graph_db_path: store_root.join("graph.db"), + store_root, + }; + let engine = DaemonEngine::default(); + let registry = Arc::clone( + engine + .store_administration + .session_temporal_refresh_schedulers(), + ); + let wake = registry.ensure_project(owner.clone(), database).await; + let state = registry.project_state(&owner).await.unwrap(); + assert_eq!(registry.project_worker_count().await, 1); + assert!(wake.status().is_available()); + + let owner_phases = engine.shutdown_owner_phases().await; + assert!(!state.is_cancelled()); + let prepared = prepare_shutdown_owner_phases(owner_phases); + let cancelled_before_join = state.is_cancelled(); + + // An earlier phase can exhaust the shared deadline before this + // owner's join is polled. Cancellation must already have reached + // the real supervisor even if every prepared join is discarded. + drop(prepared); + tokio::time::timeout(Duration::from_secs(5), registry.shutdown()) + .await + .expect("the cancelled refresh supervisor must join"); + assert_eq!(registry.project_worker_count().await, 0); + assert!( + cancelled_before_join, + "refresh must be cancelled during preparation, not first inside its join" + ); + } +} diff --git a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs index db5f5392b9b..562b8d2bf7a 100644 --- a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs +++ b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs @@ -5,10 +5,9 @@ use std::path::Path; use std::sync::{Arc, Weak}; -use tracedecay_contracts::ApplicationProblem; -use tracedecay_contracts::ResolvedScope; use tracedecay_contracts::graph_tool::GraphToolResultV1; use tracedecay_contracts::retrieval::HookRuntimeResultV1; +use tracedecay_contracts::{ApplicationProblem, CancellationStage, ResolvedScope, RetryDirective}; use tracedecay_daemon_service::{ DaemonInvocationService, GraphToolFuture, GraphToolInvocationV1, ProjectGraphToolPortV1, RegisteredGraphToolOwnerV1, @@ -51,14 +50,48 @@ impl ProjectGraphToolPortV1 for McpGraphToolPort { { return Err(ApplicationProblem::runtime_mounting()); } + let operation = invocation.operation; server .compute_graph_tool(invocation) .await - .map_err(|error| graph_tool_error_problem(&error)) + .map_err(|error| graph_tool_execution_problem(operation, &error)) }) } } +/// Dispatch and worker timers can both settle an admitted read. A raw +/// interruption returned by either must retain that read's terminal kind, +/// not become an authority-unavailable refusal from before admission. +/// Owner-supplied records/details and side-effect settlement stay authoritative. +fn graph_tool_execution_problem( + operation: ApplicationSurfaceOperation, + error: &TraceDecayError, +) -> ApplicationProblem { + if operation.owner_side_effect().is_none() + && error.project_route_problem_record().is_none() + && error.project_route_typed_detail().is_none() + { + match error.project_route_context().map(|(reason, _, _)| reason) { + Some("tool_dispatch_deadline_exceeded") => { + return ApplicationProblem::TimedOut { + stage: CancellationStage::DuringRead, + retry: RetryDirective::Never, + legal_actions: Vec::new(), + }; + } + Some("tool_dispatch_cancelled") => { + return ApplicationProblem::Cancelled { + stage: CancellationStage::DuringRead, + retry: RetryDirective::Never, + legal_actions: Vec::new(), + }; + } + _ => {} + } + } + graph_tool_error_problem(error) +} + impl McpServer { /// Registry options carrying every owner this server composes; each /// dispatch overrides only its per-call fields. @@ -249,3 +282,105 @@ impl McpServer { }) } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use serde_json::json; + use tracedecay_contracts::{ApplicationProblem, CancellationStage, RequestId}; + use tracedecay_domain::errors::TraceDecayError; + use tracedecay_mcp::server::{DispatchSettlement, dispatch_cancelled_error}; + use tracedecay_mcp::tools::dispatch_ceiling::tool_dispatch_deadline_error; + use tracedecay_tool_catalog::ApplicationSurfaceOperation; + + use super::graph_tool_execution_problem; + use crate::mcp::tools::graph_tool_error_problem; + + #[test] + fn admitted_graph_read_interruptions_preserve_terminal_kind_and_stage() { + let operation = ApplicationSurfaceOperation::RenamePreview; + assert!(operation.owner_side_effect().is_none()); + for (error, expected, kind) in [ + ( + tool_dispatch_deadline_error(operation.mcp_tool_name(), Duration::ZERO), + ApplicationProblem::timed_out(CancellationStage::DuringRead).unwrap(), + "timed_out", + ), + ( + dispatch_cancelled_error( + operation.mcp_tool_name(), + DispatchSettlement::Settling, + false, + ), + ApplicationProblem::cancelled(CancellationStage::DuringRead).unwrap(), + "cancelled", + ), + ] { + let problem = graph_tool_execution_problem(operation, &error); + assert_eq!(problem, expected); + let envelope = tracedecay_api::adapter_problem( + RequestId::new("request.graph-read.interrupted").unwrap(), + problem, + ) + .unwrap(); + let wire = serde_json::to_value(envelope).unwrap(); + assert_eq!(wire["problem"]["kind"], kind); + assert_eq!(wire["problem"]["code"], kind); + assert_eq!(wire["problem"]["cancellation_stage"], "during_read"); + assert_eq!(wire["problem"]["terminality"], "admitted_terminal"); + assert_eq!(wire["problem"]["retry"], "never"); + assert_eq!(wire["problem"]["legal_actions"], json!([])); + } + } + + #[test] + fn graph_read_execution_preserves_owner_problem_records_and_details() { + let operation = ApplicationSurfaceOperation::RenamePreview; + let expected = ApplicationProblem::timed_out_before_admission(); + let envelope = tracedecay_api::adapter_problem( + RequestId::new("request.graph-read.owner-refusal").unwrap(), + expected.clone(), + ) + .unwrap(); + let error = tool_dispatch_deadline_error(operation.mcp_tool_name(), Duration::ZERO) + .with_problem_record(serde_json::to_value(&envelope.problem).unwrap()); + assert_eq!(graph_tool_execution_problem(operation, &error), expected); + + let error = tracedecay_graph_query::map_code_graph_read_runtime_error( + tracedecay_graph_query::CodeGraphReadError::Rewarming { + retry_after_millis: 7_940, + }, + ); + let detail = error.project_route_typed_detail().unwrap().clone(); + let expected = graph_tool_error_problem(&error); + let error = TraceDecayError::project_route_with_detail( + "tool_dispatch_deadline_exceeded", + true, + detail, + ); + assert_eq!(graph_tool_execution_problem(operation, &error), expected); + } + + #[test] + fn graph_read_interruption_mapping_does_not_reclassify_effects_or_other_failures() { + let effect = ApplicationSurfaceOperation::RunAffectedTests; + assert!(effect.owner_side_effect().is_some()); + let timeout = tool_dispatch_deadline_error(effect.mcp_tool_name(), Duration::ZERO); + assert_eq!( + graph_tool_execution_problem(effect, &timeout), + graph_tool_error_problem(&timeout) + ); + for reason in [ + "tool_dispatch_effect_unknown", + "tool_dispatch_shutdown", + "verified-code-graph-read-unavailable", + ] { + let error = TraceDecayError::project_route(reason, true, "owner refusal"); + assert_eq!( + graph_tool_execution_problem(ApplicationSurfaceOperation::RenamePreview, &error), + graph_tool_error_problem(&error) + ); + } + } +} From 43e845887c57e8c36302cd47c1446c82cb06d537 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:16:16 -0800 Subject: [PATCH 2/9] fix(code-graph): preserve unchanged sealed-proof snapshots An unchanged overlapping proof replaced the shared Arc with an equivalent map, so a concurrently authenticated one-file successor lost its snapshot CAS and discarded its new fingerprints. Retain snapshot identity when no fingerprint differs and the proven key count is unchanged, without adding a full map comparison. Changed and deletion-only proofs still use the existing exact-snapshot CAS. Add real-file regressions for the unchanged-proof/successor interleaving and deletion-only pruning, including an empty result. Strengthen stale commit coverage with a genuinely new authenticated segment. Existing corruption, source-identity, extent and cancellation checks are unchanged. Refs #4507 --- .../verified_sealed_segments.rs | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index 7a44db51a02..e8acb6a3e87 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -68,6 +68,7 @@ pub(super) struct SealedSegmentProofV1<'v> { segments_root: PathBuf, prior: Option>, proven: VerifiedSegmentMap, + has_new_fingerprints: bool, } impl VerifiedSealedSegmentsV1 { @@ -86,6 +87,7 @@ impl VerifiedSealedSegmentsV1 { segments_root: segments_root.to_path_buf(), prior, proven: BTreeMap::new(), + has_new_fingerprints: false, } } } @@ -126,6 +128,8 @@ impl SealedSegmentProofV1<'_> { fingerprint: Option, ) { if let Some(fingerprint) = fingerprint { + let prior = self.prior.as_ref().and_then(|prior| prior.get(digest)); + self.has_new_fingerprints |= prior != Some(&fingerprint); self.proven.insert(digest.clone(), fingerprint); } } @@ -133,6 +137,18 @@ impl SealedSegmentProofV1<'_> { /// Remember exactly this generation, but do not let a late proof replace /// a snapshot another successful proof has committed since we began. pub(super) fn commit(self) { + // Every reused entry equals its prior value. With no new fingerprints, + // equal cardinality proves the same map without another full scan. + // Keep that snapshot's identity: an unchanged overlapping proof must + // not make a genuinely changed successor lose its commit below. + if !self.has_new_fingerprints + && self + .prior + .as_ref() + .is_some_and(|prior| prior.len() == self.proven.len()) + { + return; + } let mut roots = self .verified .roots @@ -221,6 +237,7 @@ mod tests { let root = TempDir::new().unwrap(); let (parent, parent_path) = write_segment(root.path(), b"[1]"); let (successor, successor_path) = write_segment(root.path(), b"[2]"); + let (abandoned, abandoned_path) = write_segment(root.path(), b"[3]"); let verified = VerifiedSealedSegmentsV1::default(); let mut first = verified.begin(root.path()); authenticate(&mut first, &parent, &parent_path); @@ -228,6 +245,7 @@ mod tests { let mut stale = verified.begin(root.path()); assert!(stale.already_verified(&parent, 3, &parent_path)); + authenticate(&mut stale, &abandoned, &abandoned_path); let mut sibling = verified.begin(root.path()); assert!( sibling.already_verified(&parent, 3, &parent_path), @@ -245,6 +263,69 @@ mod tests { !latest.already_verified(&parent, 3, &parent_path), "a stale proof must not resurrect segments pruned by a committed successor" ); + assert!(!latest.already_verified(&abandoned, 3, &abandoned_path)); + } + + #[test] + fn unchanged_overlapping_proof_preserves_authenticated_successor() { + let root = TempDir::new().unwrap(); + let (retained, retained_path) = write_segment(root.path(), b"[1]"); + let (replaced, replaced_path) = write_segment(root.path(), b"[2]"); + let (changed, changed_path) = write_segment(root.path(), b"[3]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut parent = verified.begin(root.path()); + authenticate(&mut parent, &retained, &retained_path); + authenticate(&mut parent, &replaced, &replaced_path); + parent.commit(); + + let mut successor = verified.begin(root.path()); + assert!(successor.already_verified(&retained, 3, &retained_path)); + authenticate(&mut successor, &changed, &changed_path); + + // A completed parent recheck must not invalidate the in-flight edit. + let mut unchanged = verified.begin(root.path()); + assert!(unchanged.already_verified(&retained, 3, &retained_path)); + assert!(unchanged.already_verified(&replaced, 3, &replaced_path)); + unchanged.commit(); + successor.commit(); + + let mut latest = verified.begin(root.path()); + assert!(latest.already_verified(&retained, 3, &retained_path)); + assert!( + latest.already_verified(&changed, 3, &changed_path), + "an unchanged overlapping proof must not discard authenticated edited source" + ); + assert!( + !latest.already_verified(&replaced, 3, &replaced_path), + "equal segment counts must not hide a genuinely changed proof" + ); + } + + #[test] + fn pruning_without_new_fingerprints_still_replaces_the_snapshot() { + let root = TempDir::new().unwrap(); + let (retained, retained_path) = write_segment(root.path(), b"[1]"); + let (removed, removed_path) = write_segment(root.path(), b"[2]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut parent = verified.begin(root.path()); + authenticate(&mut parent, &retained, &retained_path); + authenticate(&mut parent, &removed, &removed_path); + parent.commit(); + + let mut successor = verified.begin(root.path()); + assert!(successor.already_verified(&retained, 3, &retained_path)); + successor.commit(); + + let mut latest = verified.begin(root.path()); + assert!(latest.already_verified(&retained, 3, &retained_path)); + assert!( + !latest.already_verified(&removed, 3, &removed_path), + "a deletion-only successor must still prune obsolete authentication entries" + ); + drop(latest); + verified.begin(root.path()).commit(); + let mut empty = verified.begin(root.path()); + assert!(!empty.already_verified(&retained, 3, &retained_path)); } #[test] From e7c420e6b64f1058b8ea6ef39a3a00ec5ceccc0c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:32:50 -0800 Subject: [PATCH 3/9] fix(code-graph): bound retained sealed proofs across roots The daemon retained a full authenticated generation map for every root ever visited, even after that root stopped being used. Keep only the most recently used 32 root snapshots; an evicted root safely rehashes when revisited. Recent successful reads refresh recency without changing the snapshot Arc or the overlapping-proof compare-and-swap. Commits remain conditional on the exact parent snapshot, so eviction cannot promote in-flight stale proof. Add a real-file multi-root regression proving the cap, LRU eviction, retained recent-source reuse and new-root source proof. No bypass of file identity, manifest extent, digest or corruption checks. Refs #4507 --- .../verified_sealed_segments.rs | 116 ++++++++++++++++-- 1 file changed, 105 insertions(+), 11 deletions(-) diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index e8acb6a3e87..441cb4a9d3f 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -55,11 +55,28 @@ impl SealedSegmentFingerprintV1 { type VerifiedSegmentMap = BTreeMap; +// A root cache is a performance hint, not durable proof. Cap the number of +// generations held by one daemon; an evicted root is verified from disk again. +const MAX_VERIFIED_SEGMENT_ROOTS: usize = 32; + +#[derive(Debug, Default)] +struct VerifiedRootCacheV1 { + roots: HashMap, u64)>, + last_access: u64, +} + +impl VerifiedRootCacheV1 { + fn next_access(&mut self) -> u64 { + self.last_access = self.last_access.saturating_add(1); + self.last_access + } +} + /// Per segments root, the last successfully committed proof. In-flight proofs /// share that immutable snapshot without evicting or copying its entries. #[derive(Debug, Default)] pub(crate) struct VerifiedSealedSegmentsV1 { - roots: Mutex>>, + roots: Mutex, } /// One proof's view: what it may skip, and what it hashed. @@ -76,12 +93,17 @@ impl VerifiedSealedSegmentsV1 { /// overlapping proof must not evict already authenticated identities. /// New observations remain private until the whole proof succeeds. pub(super) fn begin(&self, segments_root: &Path) -> SealedSegmentProofV1<'_> { - let prior = self - .roots - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(segments_root) - .cloned(); + let prior = { + let mut cache = self + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let accessed = cache.next_access(); + cache.roots.get_mut(segments_root).map(|(snapshot, last_used)| { + *last_used = accessed; + Arc::clone(snapshot) + }) + }; SealedSegmentProofV1 { verified: self, segments_root: segments_root.to_path_buf(), @@ -154,13 +176,26 @@ impl SealedSegmentProofV1<'_> { .roots .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let unchanged = match (roots.get(&self.segments_root), self.prior.as_ref()) { + let unchanged = match (roots.roots.get(&self.segments_root), self.prior.as_ref()) { (None, None) => true, - (Some(current), Some(prior)) => Arc::ptr_eq(current, prior), + (Some((current, _)), Some(prior)) => Arc::ptr_eq(current, prior), _ => false, }; if unchanged { - roots.insert(self.segments_root, Arc::new(self.proven)); + let accessed = roots.next_access(); + roots + .roots + .insert(self.segments_root, (Arc::new(self.proven), accessed)); + if roots.roots.len() > MAX_VERIFIED_SEGMENT_ROOTS { + let oldest = roots + .roots + .iter() + .min_by_key(|(_, (_, accessed))| *accessed) + .map(|(root, _)| root.clone()); + if let Some(oldest) = oldest { + roots.roots.remove(&oldest); + } + } } } } @@ -175,7 +210,10 @@ mod tests { use tempfile::TempDir; use tracedecay_domain::{ManifestDigest, sha256_hex_suffix}; - use super::{SealedSegmentFingerprintV1, SealedSegmentProofV1, VerifiedSealedSegmentsV1}; + use super::{ + MAX_VERIFIED_SEGMENT_ROOTS, SealedSegmentFingerprintV1, SealedSegmentProofV1, + VerifiedSealedSegmentsV1, + }; fn write_segment(root: &Path, bytes: &[u8]) -> (ManifestDigest, PathBuf) { let digest = ManifestDigest::from_sha256_bytes(&Sha256::digest(bytes)).unwrap(); @@ -328,6 +366,62 @@ mod tests { assert!(!empty.already_verified(&retained, 3, &retained_path)); } + #[test] + fn many_distinct_roots_evict_cold_proofs_but_retain_recent_source() { + let temp = TempDir::new().unwrap(); + let verified = VerifiedSealedSegmentsV1::default(); + let mut roots = Vec::new(); + for index in 0..MAX_VERIFIED_SEGMENT_ROOTS { + let root = temp.path().join(format!("root-{index}")); + std::fs::create_dir(&root).unwrap(); + let (digest, path) = write_segment(&root, b"[1]"); + let mut proof = verified.begin(&root); + authenticate(&mut proof, &digest, &path); + proof.commit(); + roots.push((root, digest, path)); + } + + // A completed read without a commit still makes this root recently used. + let (recent_root, recent_digest, recent_path) = &roots[0]; + let mut recent = verified.begin(recent_root); + assert!(recent.already_verified(recent_digest, 3, recent_path)); + drop(recent); + + let incoming_root = temp.path().join("incoming-root"); + std::fs::create_dir(&incoming_root).unwrap(); + let (incoming_digest, incoming_path) = write_segment(&incoming_root, b"[2]"); + let mut incoming = verified.begin(&incoming_root); + authenticate(&mut incoming, &incoming_digest, &incoming_path); + incoming.commit(); + + assert_eq!( + verified + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .roots + .len(), + MAX_VERIFIED_SEGMENT_ROOTS, + "a daemon must not retain one full authenticated map per historical root" + ); + + let (cold_root, cold_digest, cold_path) = &roots[1]; + let mut cold = verified.begin(cold_root); + assert!( + !cold.already_verified(cold_digest, 3, cold_path), + "the least-recently-used source must be hashed again after eviction" + ); + drop(cold); + + let mut retained = verified.begin(recent_root); + assert!( + retained.already_verified(recent_digest, 3, recent_path), + "the recently accessed root must remain warm" + ); + let mut newest = verified.begin(&incoming_root); + assert!(newest.already_verified(&incoming_digest, 3, &incoming_path)); + } + #[test] fn retained_parent_rejects_same_size_replacement_with_restored_mtime() { let root = TempDir::new().unwrap(); From bbe4fe264de324a7e6ecc7903908c92b1cc4006f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:41:14 -0800 Subject: [PATCH 4/9] fix(code-graph): pin active sealed proofs during root eviction The bounded root cache evicted an authenticated parent while its one-file successor was still proving, so the successful successor lost the snapshot CAS and the next edit had to rehash the whole generation. Skip entries whose authenticated snapshot is held by an in-flight proof; evict only idle roots and preserve the 32-root bound. Keep the stale-proof guard, corruption checks and cold-root reauthentication unchanged. A real-file 33-root regression makes an edited root least-recently-used while its successor is in flight, and verifies inherited and changed source stay warm, replaced source stays cold, and an idle root is evicted. Also apply the exact Bazel rustfmt layout required by current-head gates. Refs #4507 --- .../verified_sealed_segments.rs | 87 ++++++++++++++++++- 1 file changed, 83 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index 441cb4a9d3f..131a60062cb 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -99,10 +99,13 @@ impl VerifiedSealedSegmentsV1 { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let accessed = cache.next_access(); - cache.roots.get_mut(segments_root).map(|(snapshot, last_used)| { - *last_used = accessed; - Arc::clone(snapshot) - }) + cache + .roots + .get_mut(segments_root) + .map(|(snapshot, last_used)| { + *last_used = accessed; + Arc::clone(snapshot) + }) }; SealedSegmentProofV1 { verified: self, @@ -187,9 +190,15 @@ impl SealedSegmentProofV1<'_> { .roots .insert(self.segments_root, (Arc::new(self.proven), accessed)); if roots.roots.len() > MAX_VERIFIED_SEGMENT_ROOTS { + // An in-flight proof holds an Arc to its parent snapshot. + // Evicting that snapshot would reject its valid successor + // at commit and make the next edit rehash the generation. + // The freshly inserted snapshot has only the cache owner, + // so an over-capacity insertion always has a candidate. let oldest = roots .roots .iter() + .filter(|(_, (snapshot, _))| Arc::strong_count(snapshot) == 1) .min_by_key(|(_, (_, accessed))| *accessed) .map(|(root, _)| root.clone()); if let Some(oldest) = oldest { @@ -422,6 +431,76 @@ mod tests { assert!(newest.already_verified(&incoming_digest, 3, &incoming_path)); } + #[test] + fn in_flight_successor_retains_parent_across_other_root_evictions() { + let temp = TempDir::new().unwrap(); + let verified = VerifiedSealedSegmentsV1::default(); + let mut roots = Vec::new(); + let mut replaced = None; + for index in 0..MAX_VERIFIED_SEGMENT_ROOTS { + let root = temp.path().join(format!("root-{index}")); + std::fs::create_dir(&root).unwrap(); + let (digest, path) = write_segment(&root, b"[1]"); + let mut parent = verified.begin(&root); + authenticate(&mut parent, &digest, &path); + if index == 0 { + let (old_digest, old_path) = write_segment(&root, b"[3]"); + authenticate(&mut parent, &old_digest, &old_path); + replaced = Some((old_digest, old_path)); + } + parent.commit(); + roots.push((root, digest, path)); + } + + let (active_root, inherited_digest, inherited_path) = &roots[0]; + let (replaced_digest, replaced_path) = replaced.unwrap(); + let mut successor = verified.begin(active_root); + assert!(successor.already_verified(inherited_digest, 3, inherited_path)); + let (changed_digest, changed_path) = write_segment(active_root, b"[2]"); + authenticate(&mut successor, &changed_digest, &changed_path); + + // Make the still-running successor's root the least recently used. + // An unpinned LRU cache would evict this root on the next insertion. + for (root, digest, path) in roots.iter().skip(1) { + let mut proof = verified.begin(root); + assert!(proof.already_verified(digest, 3, path)); + } + + let incoming_root = temp.path().join("incoming-root"); + std::fs::create_dir(&incoming_root).unwrap(); + let (incoming_digest, incoming_path) = write_segment(&incoming_root, b"[4]"); + let mut incoming = verified.begin(&incoming_root); + authenticate(&mut incoming, &incoming_digest, &incoming_path); + incoming.commit(); + successor.commit(); + + let mut latest = verified.begin(active_root); + assert!( + latest.already_verified(inherited_digest, 3, inherited_path), + "an in-flight edit must keep its inherited authenticated source" + ); + assert!( + latest.already_verified(&changed_digest, 3, &changed_path), + "evicting other roots must not discard the edited generation's proof" + ); + assert!( + !latest.already_verified(&replaced_digest, 3, &replaced_path), + "the successor must not resurrect a replaced parent segment" + ); + + let (evicted_root, evicted_digest, evicted_path) = &roots[1]; + let mut evicted = verified.begin(evicted_root); + assert!( + !evicted.already_verified(evicted_digest, 3, evicted_path), + "the oldest idle root, rather than an active proof, must be evicted" + ); + let cache = verified + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert_eq!(cache.roots.len(), MAX_VERIFIED_SEGMENT_ROOTS); + } + #[test] fn retained_parent_rejects_same_size_replacement_with_restored_mtime() { let root = TempDir::new().unwrap(); From 85c8cd081109457a4abda46aeb10a4490d1208e6 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 20:08:20 -0800 Subject: [PATCH 5/9] fix(code-graph): preserve new proofs while old roots are pinned When all 32 cached roots have in-flight proofs, admitting a newly verified root evicted the only idle candidate: the new proof itself. That made the next one-file edit rehash the entire generation. Retain the newest authenticated root temporarily while older parents are pinned; reclaim excess idle roots as the final proof references drop. Keep stale-generation fencing and source identity checks intact. A 33-root real-file regression demonstrates warm source survival, transient over-cap retention and eventual return to the 32-root bound. Refs #4507 --- .../verified_sealed_segments.rs | 140 +++++++++++++++--- 1 file changed, 120 insertions(+), 20 deletions(-) diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index 131a60062cb..66757ed0878 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -70,6 +70,34 @@ impl VerifiedRootCacheV1 { self.last_access = self.last_access.saturating_add(1); self.last_access } + + /// Keep the newest authenticated root warm when all older roots have + /// active proofs. Excess entries are bounded by those outstanding pins + /// and trimmed as soon as their proofs finish. + fn trim_idle(&mut self) { + if self.roots.len() <= MAX_VERIFIED_SEGMENT_ROOTS { + return; + } + let newest = self + .roots + .iter() + .max_by_key(|(_, (_, accessed))| *accessed) + .map(|(root, _)| root.clone()); + while self.roots.len() > MAX_VERIFIED_SEGMENT_ROOTS { + let oldest = self + .roots + .iter() + .filter(|(root, (snapshot, _))| { + newest.as_deref() != Some(root.as_path()) && Arc::strong_count(snapshot) == 1 + }) + .min_by_key(|(_, (_, accessed))| *accessed) + .map(|(root, _)| root.clone()); + let Some(oldest) = oldest else { + break; + }; + self.roots.remove(&oldest); + } + } } /// Per segments root, the last successfully committed proof. In-flight proofs @@ -161,7 +189,7 @@ impl SealedSegmentProofV1<'_> { /// Remember exactly this generation, but do not let a late proof replace /// a snapshot another successful proof has committed since we began. - pub(super) fn commit(self) { + pub(super) fn commit(mut self) { // Every reused entry equals its prior value. With no new fingerprints, // equal cardinality proves the same map without another full scan. // Keep that snapshot's identity: an unchanged overlapping proof must @@ -186,26 +214,29 @@ impl SealedSegmentProofV1<'_> { }; if unchanged { let accessed = roots.next_access(); - roots - .roots - .insert(self.segments_root, (Arc::new(self.proven), accessed)); - if roots.roots.len() > MAX_VERIFIED_SEGMENT_ROOTS { - // An in-flight proof holds an Arc to its parent snapshot. - // Evicting that snapshot would reject its valid successor - // at commit and make the next edit rehash the generation. - // The freshly inserted snapshot has only the cache owner, - // so an over-capacity insertion always has a candidate. - let oldest = roots - .roots - .iter() - .filter(|(_, (snapshot, _))| Arc::strong_count(snapshot) == 1) - .min_by_key(|(_, (_, accessed))| *accessed) - .map(|(root, _)| root.clone()); - if let Some(oldest) = oldest { - roots.roots.remove(&oldest); - } - } + roots.roots.insert( + self.segments_root.clone(), + (Arc::new(std::mem::take(&mut self.proven)), accessed), + ); + roots.trim_idle(); + } + } +} + +impl Drop for SealedSegmentProofV1<'_> { + fn drop(&mut self) { + // Release the parent pin before trimming. An all-pinned burst may + // have temporarily retained a successfully authenticated new root + // above the idle cap; leaving the last active proof reaps it. + if self.prior.take().is_none() { + return; } + let mut roots = self + .verified + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + roots.trim_idle(); } } @@ -501,6 +532,75 @@ mod tests { assert_eq!(cache.roots.len(), MAX_VERIFIED_SEGMENT_ROOTS); } + #[test] + fn all_pinned_roots_preserve_newly_authenticated_source_until_idle_trim() { + let temp = TempDir::new().unwrap(); + let verified = VerifiedSealedSegmentsV1::default(); + let mut existing = Vec::new(); + for index in 0..MAX_VERIFIED_SEGMENT_ROOTS { + let root = temp.path().join(format!("root-{index}")); + std::fs::create_dir(&root).unwrap(); + let (digest, path) = write_segment(&root, b"[1]"); + let mut initial = verified.begin(&root); + authenticate(&mut initial, &digest, &path); + initial.commit(); + existing.push((root, digest, path)); + } + + // The entire parent cache is pinned by real in-flight proofs. The + // previously sole evictable candidate was the successful newcomer. + let mut pinned = Vec::new(); + for (root, digest, path) in &existing { + let mut proof = verified.begin(root); + assert!(proof.already_verified(digest, 3, path)); + pinned.push(proof); + } + let new_root = temp.path().join("new-root"); + std::fs::create_dir(&new_root).unwrap(); + let (new_digest, new_path) = write_segment(&new_root, b"[2]"); + let mut newcomer = verified.begin(&new_root); + authenticate(&mut newcomer, &new_digest, &new_path); + newcomer.commit(); + + let mut warm = verified.begin(&new_root); + assert!( + warm.already_verified(&new_digest, 3, &new_path), + "a complete new-root proof must not be its own eviction victim" + ); + drop(warm); + assert_eq!( + verified + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .roots + .len(), + MAX_VERIFIED_SEGMENT_ROOTS + 1, + "only the one temporarily pinned-over-cap root may remain" + ); + + // Releasing old proof references restores the idle bound without + // discarding new source or reintroducing a stale parent entry. + drop(pinned); + let mut retained = verified.begin(&new_root); + assert!(retained.already_verified(&new_digest, 3, &new_path)); + assert_eq!( + verified + .roots + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .roots + .len(), + MAX_VERIFIED_SEGMENT_ROOTS + ); + let (old_root, old_digest, old_path) = &existing[0]; + let mut evicted = verified.begin(old_root); + assert!( + !evicted.already_verified(old_digest, 3, old_path), + "the oldest retired root, not the new generation, must be evicted" + ); + } + #[test] fn retained_parent_rejects_same_size_replacement_with_restored_mtime() { let root = TempDir::new().unwrap(); From a6bbbbbb6d4740ea7949ff80b8888918ca973ccf Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:19:45 -0800 Subject: [PATCH 6/9] fix(bench): include source mutation in one-file edit measurements The daemon can begin indexing on a filesystem edit before the explicit sync request. Starting wall time and peak-RSS observation after the source write misses work triggered by the edit. Arm both before writing. Exercise the real harness edit path with a deliberately slow write: require RSS observation before mutation, full wall-time inclusion, and correct readiness and peak-memory attribution. Preserve existing full-run fixture integration coverage. Refs #4507 --- scripts/bench-hot-paths.py | 7 +++- scripts/test-bench-hot-paths.py | 73 +++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 2 deletions(-) diff --git a/scripts/bench-hot-paths.py b/scripts/bench-hot-paths.py index 14d4e3d1c00..fc16f420b5e 100755 --- a/scripts/bench-hot-paths.py +++ b/scripts/bench-hot-paths.py @@ -449,12 +449,15 @@ def lane_edit_reconcile(self) -> None: return target = self.repo / tracked[0] marker = "//" if target.suffix == ".rs" else "#" - with target.open("a", encoding="utf-8") as handle: - handle.write(f"\n{marker} bench-hot-paths edit {time.time_ns()}\n") + # The file watcher may begin its pass as soon as the edit lands. + # Arm the timer and RSS sampler *before* modifying the source; a + # sync-only measurement misses work that starts during the write. start = time.perf_counter() rss_sampler = DaemonRssSampler(self.daemon_pid()) rss_sampler.start() try: + with target.open("a", encoding="utf-8") as handle: + handle.write(f"\n{marker} bench-hot-paths edit {time.time_ns()}\n") sync = self.client(["sync"]) if not sync.ok: raise HarnessError(f"`tracedecay sync` failed: {sync.stderr[-2000:].decode(errors='replace')}") diff --git a/scripts/test-bench-hot-paths.py b/scripts/test-bench-hot-paths.py index 1dbf94aaad3..1f2a9f266aa 100755 --- a/scripts/test-bench-hot-paths.py +++ b/scripts/test-bench-hot-paths.py @@ -3,13 +3,16 @@ from __future__ import annotations +import importlib.util import json import os import subprocess import sys import tempfile import textwrap +import time import unittest +from unittest import mock from pathlib import Path ROOT = Path(__file__).resolve().parent.parent @@ -167,6 +170,76 @@ def test_a_full_run_records_every_lane_and_leaves_no_run_directory(self) -> None self.assertEqual(list(self.tmp.iterdir()), []) self.assertEqual((self.repo / "src" / "lib.rs").read_text(), "pub fn answer() -> u32 { 42 }\n") + def test_edit_timer_and_rss_observer_cover_the_file_mutation(self) -> None: + # The live daemon can start indexing on the filesystem notification + # before `sync` returns. Reproduce an observable slow edit and + # prove both clocks are armed before the source is touched. + spec = importlib.util.spec_from_file_location("bench_hot_paths_timing", BENCH) + self.assertIsNotNone(spec) + self.assertIsNotNone(spec.loader) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + run = module.Run.__new__(module.Run) + run.repo = self.repo + run.generation = "generation.1" + events = [] + rows = [] + run.daemon_pid = lambda: 1234 + run.record = lambda lane, op, **data: rows.append((lane, op, data)) + + def sync(argv): + self.assertEqual(argv, ["sync"]) + events.append("sync") + return module.Call(True, b"", b"", 0.5, 0.0, 0) + + def ready(lane, previous_generation): + self.assertEqual((lane, previous_generation), ("edit_reconcile", "generation.1")) + events.append("ready") + return {"generation_id": "generation.2"} + + run.client = sync + run.wait_for_ready = ready + + class RecordingSampler: + def __init__(self, pid): + assert pid == 1234 + + def start(self): + events.append("sampler_started") + + def stop(self): + events.append("sampler_stopped") + return 2048 + + target = self.repo / "src" / "lib.rs" + original_open = Path.open + + def slow_edit_open(path, mode="r", *args, **kwargs): + if path == target and mode == "a": + self.assertIn("sampler_started", events) + events.append("edit_started") + time.sleep(0.1) + return original_open(path, mode, *args, **kwargs) + + with mock.patch.object(module, "DaemonRssSampler", RecordingSampler), mock.patch.object( + Path, "open", slow_edit_open + ): + run.lane_edit_reconcile() + self.assertEqual( + events, ["sampler_started", "edit_started", "sync", "ready", "sampler_stopped"] + ) + measured = next(data for lane, op, data in rows if (lane, op) == ("edit_reconcile", "edit_to_ready")) + self.assertGreaterEqual( + measured["wall_ms"], 90.0, + "the incremental-pass wall time must include file mutation before sync", + ) + self.assertEqual(measured["peak_daemon_rss_kb"], 2048) + self.assertIn("bench-hot-paths edit", target.read_text()) + finally: + sys.modules.pop(spec.name, None) + def test_an_unknown_seed_records_callers_unavailable_without_failing_the_run(self) -> None: result = self.bench("--seed-symbol", "missing_symbol") self.assertEqual(result.returncode, 0, result.stderr) From e4064bbcb350cb5a5cd40539b5897909e66af1c9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:21:05 -0800 Subject: [PATCH 7/9] fix(bench): fence one-file measurements to the pre-edit generation An asynchronous daemon refresh can advance the current generation after initial indexing but before the benchmark edits source. Using the old initial-generation ID could incorrectly accept that prior refresh as the result of the edit, reporting a tiny wall time and missing peak memory. Read and require the latest current, ready generation before mutating the file; only a subsequent nonempty generation ID ends the timed pass. Preserve the sampler-before-write timing and update the observed generation. Exercise a real tracked edit against a deliberately stale index-lane ID, reject a missing pre-edit ID without modifying source, and refuse a false ready result with an absent generation until a real successor appears. Refs #4507 --- scripts/bench-hot-paths.py | 20 ++++++++++- scripts/test-bench-hot-paths.py | 59 +++++++++++++++++++++++++++++++-- 2 files changed, 76 insertions(+), 3 deletions(-) mode change 100755 => 100644 scripts/bench-hot-paths.py mode change 100755 => 100644 scripts/test-bench-hot-paths.py diff --git a/scripts/bench-hot-paths.py b/scripts/bench-hot-paths.py old mode 100755 new mode 100644 index fc16f420b5e..cd4af1a5f94 --- a/scripts/bench-hot-paths.py +++ b/scripts/bench-hot-paths.py @@ -330,6 +330,8 @@ def wait_for_ready(self, lane: str, previous_generation: str | None) -> dict[str if ( freshness.get("status") == "current" and serving.get("state") == "ready" + and isinstance(generation, str) + and bool(generation) and generation != previous_generation ): return { @@ -449,6 +451,21 @@ def lane_edit_reconcile(self) -> None: return target = self.repo / tracked[0] marker = "//" if target.suffix == ".rs" else "#" + # Another daemon refresh may have advanced the generation since + # indexing. Fence the edit against the latest ready generation, not + # the possibly stale generation recorded at initial enrollment. + payload = self.status_payload() or {} + freshness = payload.get("code_index_freshness") or {} + worktree = freshness.get("worktree") or {} + serving = worktree.get("code_graph_serving") or {} + previous_generation = worktree.get("latest_generation_id") + if ( + freshness.get("status") != "current" + or serving.get("state") != "ready" + or not isinstance(previous_generation, str) + or not previous_generation + ): + raise HarnessError("edit_reconcile: no current ready generation before editing") # The file watcher may begin its pass as soon as the edit lands. # Arm the timer and RSS sampler *before* modifying the source; a # sync-only measurement misses work that starts during the write. @@ -461,7 +478,7 @@ def lane_edit_reconcile(self) -> None: sync = self.client(["sync"]) if not sync.ok: raise HarnessError(f"`tracedecay sync` failed: {sync.stderr[-2000:].decode(errors='replace')}") - ready = self.wait_for_ready("edit_reconcile", self.generation) + ready = self.wait_for_ready("edit_reconcile", previous_generation) finally: peak_rss_kb = rss_sampler.stop() if peak_rss_kb is None: @@ -471,6 +488,7 @@ def lane_edit_reconcile(self) -> None: ) else: ready["peak_daemon_rss_kb"] = peak_rss_kb + self.generation = ready["generation_id"] self.record("edit_reconcile", "sync_request", wall_ms=sync.wall_ms) self.record("edit_reconcile", "edited_file", path=tracked[0]) self.record("edit_reconcile", "edit_to_ready", diff --git a/scripts/test-bench-hot-paths.py b/scripts/test-bench-hot-paths.py old mode 100755 new mode 100644 index 1f2a9f266aa..87cda9d9020 --- a/scripts/test-bench-hot-paths.py +++ b/scripts/test-bench-hot-paths.py @@ -184,6 +184,17 @@ def test_edit_timer_and_rss_observer_cover_the_file_mutation(self) -> None: run = module.Run.__new__(module.Run) run.repo = self.repo run.generation = "generation.1" + # Another refresh advanced the daemon after indexing, before + # the measured edit. Its current generation is the true fence. + run.status_payload = lambda: { + "code_index_freshness": { + "status": "current", + "worktree": { + "latest_generation_id": "generation.7", + "code_graph_serving": {"state": "ready"}, + }, + } + } events = [] rows = [] run.daemon_pid = lambda: 1234 @@ -195,9 +206,9 @@ def sync(argv): return module.Call(True, b"", b"", 0.5, 0.0, 0) def ready(lane, previous_generation): - self.assertEqual((lane, previous_generation), ("edit_reconcile", "generation.1")) + self.assertEqual((lane, previous_generation), ("edit_reconcile", "generation.7")) events.append("ready") - return {"generation_id": "generation.2"} + return {"generation_id": "generation.8"} run.client = sync run.wait_for_ready = ready @@ -237,6 +248,50 @@ def slow_edit_open(path, mode="r", *args, **kwargs): ) self.assertEqual(measured["peak_daemon_rss_kb"], 2048) self.assertIn("bench-hot-paths edit", target.read_text()) + self.assertEqual(run.generation, "generation.8") + + # A missing generation must be refused before writing a source + # file or starting the sampler; it is not evidence of readiness. + before = target.read_bytes() + events_before = list(events) + run.status_payload = lambda: { + "code_index_freshness": { + "status": "current", + "worktree": { + "latest_generation_id": None, + "code_graph_serving": {"state": "ready"}, + }, + } + } + with self.assertRaisesRegex( + module.HarnessError, "no current ready generation before editing" + ): + run.lane_edit_reconcile() + self.assertEqual(target.read_bytes(), before) + self.assertEqual(events, events_before) + + # A ready response with no generation ID cannot finish a later + # pass. A subsequent actual generation must be observed instead. + missing = run.status_payload() + valid = { + "code_index_freshness": { + "status": "current", + "worktree": { + "latest_generation_id": "generation.9", + "code_graph_serving": {"state": "ready"}, + }, + } + } + responses = iter((missing, valid)) + run.status_payload = lambda: next(responses) + run.index_timeout = 10 + del run.wait_for_ready + with mock.patch.object(module, "proc_status", return_value=None), mock.patch.object( + module.time, "sleep", return_value=None + ): + result = run.wait_for_ready("edit_reconcile", "generation.8") + self.assertEqual(result["generation_id"], "generation.9") + self.assertEqual(result["status_polls"], 2) finally: sys.modules.pop(spec.name, None) From 232d53af11afea1a30ed68b0e79dbeb11fdfb34d Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:36:33 -0800 Subject: [PATCH 8/9] fix(benchmark): fence edits to regular clone-owned files A tracked source symlink can target a file outside the disposable checkout; appending to the first git ls-files result could mutate operator source. Git's quoted filenames also break whitespace splitting and abort real runs. Select only regular non-symlink files resolved within the private clone, and read NUL-delimited Git paths to retain spaces and newlines. Exercise the full fake-daemon benchmark with a committed external symlink and a spaced Rust source path. Require that only the safe path is selected and the external original remains unchanged. Preserve existing timing and RSS sampling behavior. Refs #4507 --- scripts/bench-hot-paths.py | 31 +++++++++++++++++++++++++------ scripts/test-bench-hot-paths.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/scripts/bench-hot-paths.py b/scripts/bench-hot-paths.py index cd4af1a5f94..e6e0102f835 100644 --- a/scripts/bench-hot-paths.py +++ b/scripts/bench-hot-paths.py @@ -444,12 +444,31 @@ def lane_memory(self, phase: str) -> None: def lane_edit_reconcile(self) -> None: log("edit_reconcile (one appended line until a new generation is ready)") - tracked = subprocess.run(["git", "-C", str(self.repo), "ls-files", "*.rs", "*.py"], - capture_output=True, text=True, check=True).stdout.split() - if not tracked: - self.record("edit_reconcile", "edit_to_ready", unavailable="no tracked .rs or .py file") + # Git's -z mode preserves whitespace and newlines in tracked paths. + # A tracked symlink may point outside the disposable clone: never + # append to one, even if its target is a regular source file. + listing = subprocess.run( + ["git", "-C", str(self.repo), "ls-files", "-z", "--", "*.rs", "*.py"], + capture_output=True, check=True, + ).stdout + repo_root = self.repo.resolve() + selected = None + for raw_name in listing.split(b"\0"): + if not raw_name: + continue + name = os.fsdecode(raw_name) + candidate = self.repo / name + if ( + not candidate.is_symlink() + and candidate.is_file() + and candidate.resolve().is_relative_to(repo_root) + ): + selected = (name, candidate) + break + if selected is None: + self.record("edit_reconcile", "edit_to_ready", unavailable="no tracked regular .rs or .py file") return - target = self.repo / tracked[0] + tracked_file, target = selected marker = "//" if target.suffix == ".rs" else "#" # Another daemon refresh may have advanced the generation since # indexing. Fence the edit against the latest ready generation, not @@ -490,7 +509,7 @@ def lane_edit_reconcile(self) -> None: ready["peak_daemon_rss_kb"] = peak_rss_kb self.generation = ready["generation_id"] self.record("edit_reconcile", "sync_request", wall_ms=sync.wall_ms) - self.record("edit_reconcile", "edited_file", path=tracked[0]) + self.record("edit_reconcile", "edited_file", path=tracked_file) self.record("edit_reconcile", "edit_to_ready", **{**ready, "wall_ms": (time.perf_counter() - start) * 1000}) diff --git a/scripts/test-bench-hot-paths.py b/scripts/test-bench-hot-paths.py index 87cda9d9020..74a720c9d40 100644 --- a/scripts/test-bench-hot-paths.py +++ b/scripts/test-bench-hot-paths.py @@ -170,6 +170,35 @@ def test_a_full_run_records_every_lane_and_leaves_no_run_directory(self) -> None self.assertEqual(list(self.tmp.iterdir()), []) self.assertEqual((self.repo / "src" / "lib.rs").read_text(), "pub fn answer() -> u32 { 42 }\n") + def test_edit_never_follows_tracked_symlinks_and_preserves_spaced_names(self) -> None: + # A Git checkout can contain a tracked symlink to a path outside the + # benchmark's private clone. The old first-file append would mutate + # that external path. A quoted, spaced Git path also broke .split(). + outside = self.root / "operator-source.py" + outside.write_text("print('untouched')\n") + if os.name != "nt": + (self.repo / "a0-external.py").symlink_to(outside) + safe = self.repo / "a1 spaced.rs" + safe.write_text("pub fn safe() {}\n") + subprocess.run(["git", "-C", str(self.repo), "add", "-A"], check=True) + subprocess.run( + ["git", "-C", str(self.repo), "-c", "user.name=t", "-c", "user.email=t@t", + "commit", "-qm", "track escaped source candidates"], + check=True, + ) + + result = self.bench() + self.assertEqual(result.returncode, 0, result.stderr) + selected = next( + row for row in self.rows() + if (row["lane"], row["op"]) == ("edit_reconcile", "edited_file") + ) + self.assertEqual(selected["path"], "a1 spaced.rs") + self.assertEqual(outside.read_text(), "print('untouched')\n") + self.assertEqual(safe.read_text(), "pub fn safe() {}\n") + self.assertEqual(list(self.tmp.iterdir()), []) + + def test_edit_timer_and_rss_observer_cover_the_file_mutation(self) -> None: # The live daemon can start indexing on the filesystem notification # before `sync` returns. Reproduce an observable slow edit and From 7540e8c5f717291a7faebee1801a41b26e25093b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:59:11 -0800 Subject: [PATCH 9/9] fix(benchmark): parse canonical structured status for readiness Use the complete typed status result before the CLI's possibly truncated text preview. Make the existing end-to-end fake-daemon regression emit preview-only text with real readiness in structuredContent, so the former parser fails before any generation becomes ready. Refs #4507 --- scripts/bench-hot-paths.py | 11 ++++++++++- scripts/test-bench-hot-paths.py | 7 +++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/scripts/bench-hot-paths.py b/scripts/bench-hot-paths.py index e6e0102f835..d8c12232283 100644 --- a/scripts/bench-hot-paths.py +++ b/scripts/bench-hot-paths.py @@ -252,7 +252,16 @@ def status_payload(self) -> dict | None: if not call.ok: return None try: - return json.loads(json.loads(call.stdout)["content"][0]["text"]) + envelope = json.loads(call.stdout) + if not isinstance(envelope, dict): + return None + # structuredContent is the complete typed answer; text can be a + # truncated preview even when the status call succeeded. + structured = envelope.get("structuredContent") + if isinstance(structured, dict): + return structured + text_payload = json.loads(envelope["content"][0]["text"]) + return text_payload if isinstance(text_payload, dict) else None except (ValueError, KeyError, IndexError, TypeError): return None diff --git a/scripts/test-bench-hot-paths.py b/scripts/test-bench-hot-paths.py index 74a720c9d40..4c5458d554d 100644 --- a/scripts/test-bench-hot-paths.py +++ b/scripts/test-bench-hot-paths.py @@ -61,13 +61,16 @@ def reply(payload, ok=True, content_payload=None, structured=None): reply({"problem": {"code": "project_not_enrolled"}}, ok=False) tool = argv[1] if tool == "tracedecay_status": - reply({ + status = { "code_index_freshness": {"status": "current", "worktree": { "latest_generation_id": f"generation.{generation}", "code_graph_serving": {"state": "ready"}}}, "memory": {"retained_bytes": 4096, "owners": [{"kind": "graph_engine", "bytes": 4096}]}, - }) + } + # The real CLI keeps the full response in structuredContent. + # A preview-only text field must not hide readiness or generation. + reply(status, content_payload={"handle": "rh_status", "preview": "{..."}, structured=status) if tool == "tracedecay_search": args = json.loads(argv[argv.index("--args") + 1]) payload = {"results": []}