fix(code-graph): retain authenticated proofs across interrupted successors

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,

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()

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()