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<Self>,
+    ) -> impl Future<Output = Result<(), FeedbackRuntimeError>> + 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<Output = Result<(), FeedbackRuntimeError>> + Send + 'static + use<> {
+        self.observation_sink.close_and_drain()
     }
 
     pub fn owner(&self) -> Arc<ConcreteFeedbackOwner> {
@@ -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<PathBuf, ProjectRuntime>) -> Vec
 }
 
 async fn shut_down_feedback(runtimes: &BTreeMap<PathBuf, ProjectRuntime>) -> Vec<String> {
+    // 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::<std::collections::BTreeSet<_>>();
+        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..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
@@ -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,35 +53,94 @@ impl SealedSegmentFingerprintV1 {
     }
 }
 
-/// Per segments root, the segments of the generation last proven there.
+type VerifiedSegmentMap = BTreeMap<ManifestDigest, SealedSegmentFingerprintV1>;
+
+// 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<PathBuf, (Arc<VerifiedSegmentMap>, u64)>,
+    last_access: u64,
+}
+
+impl VerifiedRootCacheV1 {
+    fn next_access(&mut self) -> u64 {
+        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
+/// share that immutable snapshot without evicting or copying its entries.
 #[derive(Debug, Default)]
 pub(crate) struct VerifiedSealedSegmentsV1 {
-    roots: Mutex<HashMap<PathBuf, BTreeMap<ManifestDigest, SealedSegmentFingerprintV1>>>,
+    roots: Mutex<VerifiedRootCacheV1>,
 }
 
 /// 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<ManifestDigest, SealedSegmentFingerprintV1>,
-    proven: BTreeMap<ManifestDigest, SealedSegmentFingerprintV1>,
+    prior: Option<Arc<VerifiedSegmentMap>>,
+    proven: VerifiedSegmentMap,
+    has_new_fingerprints: bool,
 }
 
 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();
+        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(),
             prior,
             proven: BTreeMap::new(),
+            has_new_fingerprints: false,
         }
     }
 }
@@ -95,7 +154,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
@@ -117,17 +181,482 @@ impl SealedSegmentProofV1<'_> {
         fingerprint: Option<SealedSegmentFingerprintV1>,
     ) {
         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);
         }
     }
 
-    /// The proof authenticated every segment: remember exactly this
-    /// generation's segments, so the record stays one generation large.
-    pub(super) fn commit(self) {
-        self.verified
+    /// 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(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
+        // 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
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        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),
+            _ => false,
+        };
+        if unchanged {
+            let accessed = roots.next_access();
+            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)
-            .insert(self.segments_root, self.proven);
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        roots.trim_idle();
+    }
+}
+
+#[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::{
+        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();
+        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 (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);
+        first.commit();
+
+        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),
+            "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"
+        );
+        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]
+    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 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 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();
+        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)
+            );
+        }
+    }
+}
diff --git a/scripts/bench-hot-paths.py b/scripts/bench-hot-paths.py
old mode 100755
new mode 100644
index 14d4e3d1c00..d8c12232283
--- 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
 
@@ -330,6 +339,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 {
@@ -442,23 +453,60 @@ 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 "#"
-        with target.open("a", encoding="utf-8") as handle:
-            handle.write(f"\n{marker} bench-hot-paths edit {time.time_ns()}\n")
+        # 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.
         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')}")
-            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:
@@ -468,8 +516,9 @@ 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", "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
old mode 100755
new mode 100644
index 1dbf94aaad3..4c5458d554d
--- 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
@@ -58,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": []}
@@ -167,6 +173,160 @@ 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
+        # 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"
+            # 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
+            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.7"))
+                events.append("ready")
+                return {"generation_id": "generation.8"}
+
+            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())
+            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)
+
     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)