ci: integration batch A (64 reviewed prove-* PRs)

crates/tracedecay-application/src/observability/delivery_spool.rs

@@ -13,7 +13,8 @@ use tracedecay_domain::{
     DeliverySettlementV1, canonical_json_bytes, canonical_sha256, sha256_hex_suffix,
 };
 use tracedecay_private_fs::framed_log::{
-    DirectorySyncPolicy, atomic_write, read_bounded, sync_directory, validate_regular_or_missing,
+    DirectorySyncPolicy, atomic_write, is_owned_temporary_name, read_bounded,
+    remove_abandoned_temporaries, sync_directory, validate_regular_or_missing,
 };
 
 use super::ObservabilityProducerIdentityV1;
@@ -148,6 +149,11 @@ impl DeliveryRecorderSpoolV1 {
             std::fs::TryLockError::WouldBlock => DeliveryRecorderSpoolError::Busy,
             std::fs::TryLockError::Error(_) => DeliveryRecorderSpoolError::Io,
         })?;
+        // The lease is exclusive now, so every staging temporary still in the
+        // root was abandoned by a killed publisher. Sweeping it here is what
+        // makes a crashed daemon's project reopenable.
+        remove_abandoned_temporaries(&root, DIRECTORY_POLICY)
+            .map_err(|_| DeliveryRecorderSpoolError::Io)?;
         let receipt_paths = scan_receipt_paths(&root)?;
         let spool = Self {
             root,
@@ -265,7 +271,7 @@ fn scan_receipt_paths(root: &Path) -> Result<Vec<PathBuf>, DeliveryRecorderSpool
             .file_name()
             .into_string()
             .map_err(|_| DeliveryRecorderSpoolError::UnsafePath)?;
-        if name == LOCK_FILE {
+        if name == LOCK_FILE || is_owned_temporary_name(&name) {
             continue;
         }
         if !valid_receipt_name(&name)
@@ -425,4 +431,28 @@ mod tests {
             Err(DeliveryRecorderSpoolError::InvalidReceipt)
         );
     }
+
+    /// A publisher killed between staging and rename leaves its `.tmp` behind.
+    /// The next open owns that residue: it must sweep it and mount, not refuse
+    /// the project's observability spool as an unsafe path forever.
+    #[test]
+    fn open_sweeps_a_staging_temporary_left_by_a_killed_publisher() {
+        let root = tempfile::tempdir().expect("spool root");
+        let spool = DeliveryRecorderSpoolV1::open(root.path().to_path_buf()).expect("first open");
+        let receipt =
+            DeliveryRecorderSourceReceiptV1::new(settlement(), identity()).expect("receipt");
+        assert!(spool.append(&receipt).expect("append receipt"));
+        drop(spool);
+
+        // Exactly what `framed_log::temporary_path` stages beside a receipt.
+        let abandoned = root
+            .path()
+            .join(".00000000000000000000000000000000.delivery.v1.json.delivery.4242.7.tmp");
+        std::fs::write(&abandoned, b"partial").expect("abandoned staging temporary");
+
+        let reopened =
+            DeliveryRecorderSpoolV1::open(root.path().to_path_buf()).expect("reopen after crash");
+        assert!(!abandoned.exists(), "the staging temporary must be swept");
+        assert_eq!(reopened.pending(8).expect("pending receipts").len(), 1);
+    }
 }

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs

@@ -4093,6 +4093,17 @@ async fn dashboard_freshness_reports_pending_rebuild_liveness() {
         .expect("mount daemon-owned scheduler");
     wait_for_initial_generation(&registry, fixture.path()).await;
     wait_for_dashboard_ready(&registry, fixture.path()).await;
+    // The mount seats exact/lexical before the clone successor is built, so
+    // `wait_for_dashboard_ready` returns with that backfill still pending, and
+    // the admission below only parks a *new* pass at its dequeue point. A
+    // backfill slice advances under the clone-successor slot lock, which
+    // `clone_index_status` takes with `try_lock`: a sample that lands inside
+    // one reports `Unavailable { "clone-index status is being updated" }`
+    // before the source-stale branch can answer `Stale` (CI run 35432037843).
+    // Drain the mount-era backfill and burn the wake permits it banks, so the
+    // held admission is the only scheduling this sample can observe.
+    drain_clone_backfill(&registry, fixture.path()).await;
+    settled_owner_with_idle_admission(&registry, fixture.path()).await;
 
     let admission = registry
         .background_reconcile_admission()
@@ -4118,10 +4129,14 @@ async fn dashboard_freshness_reports_pending_rebuild_liveness() {
         projected.rebuild_in_flight,
         "a pending scheduler wake must keep stale serving typed as rebuilding"
     );
-    assert!(matches!(
-        projected.clone_index,
-        Some(tracedecay_contracts::code_index_freshness::CodeCloneIndexStatusV1::Stale { .. })
-    ));
+    assert!(
+        matches!(
+            projected.clone_index,
+            Some(tracedecay_contracts::code_index_freshness::CodeCloneIndexStatusV1::Stale { .. })
+        ),
+        "a settled clone index under a stale source must read Stale: {:?}",
+        projected.clone_index
+    );
     drop(admission);
     registry.shutdown().await;
 }

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/serving.rs

@@ -4148,6 +4148,30 @@ async fn root_graph_ready_does_not_depend_on_the_publication_decode_cache() {
         .unwrap_or_else(std::sync::PoisonError::into_inner)
         .hold_active_decode();
 
+    // `waiter_count` counts every task parked on this store's active decode,
+    // not this call's. The owner's settle pass decodes the active generation
+    // inside graph prepare and drops its reconcile-pass guard before it gets
+    // there, so no admission or quiescence helper can fence it out of the
+    // hold, and against a bare `== 0` its park reads as a readiness park.
+    //
+    // Park the owner first and take its count as the floor instead. A park
+    // cannot end while the hold is up and the owner is blocked in the step it
+    // parked in, so every later rise is a readiness call joining the flight.
+    registry.request_complete_generation(fixture.path()).await;
+    let deadline = Instant::now() + Duration::from_secs(30);
+    let parked_owner = loop {
+        let parked = held_decode.waiter_count();
+        if parked > 0 {
+            break parked;
+        }
+        assert!(
+            Instant::now() <= deadline,
+            "the owner's settle pass never reached the held decode, so its park \
+             cannot be sequenced ahead of the readiness calls"
+        );
+        tokio::time::sleep(Duration::from_millis(2)).await;
+    };
+
     let ready = tokio::time::timeout(
         Duration::from_secs(30),
         registry.latest_complete_ready_decoded_for_root_scope(fixture.path(), &scope),
@@ -4162,7 +4186,7 @@ async fn root_graph_ready_does_not_depend_on_the_publication_decode_cache() {
     );
     assert_eq!(
         held_decode.waiter_count(),
-        0,
+        parked_owner,
         "root graph readiness must not join the publication decode flight"
     );
 
@@ -4180,7 +4204,7 @@ async fn root_graph_ready_does_not_depend_on_the_publication_decode_cache() {
     );
     assert_eq!(
         held_decode.waiter_count(),
-        0,
+        parked_owner,
         "scope query readiness must not join the publication decode flight"
     );
 

crates/tracedecay-hooks/src/delivery_spool.rs

@@ -17,7 +17,8 @@ use tracedecay_domain::{
     canonical_json_bytes, canonical_sha256, sha256_hex_suffix,
 };
 use tracedecay_private_fs::framed_log::{
-    DirectorySyncPolicy, atomic_write, read_bounded, sync_directory, validate_regular_or_missing,
+    DirectorySyncPolicy, atomic_write, is_owned_temporary_name, read_bounded,
+    remove_abandoned_temporaries, sync_directory, validate_regular_or_missing,
 };
 
 const MAX_PENDING_RECEIPTS: usize = 1_024;
@@ -187,6 +188,10 @@ impl HookDeliveryReceiptSpoolV1 {
             }
         }
         let spool = Self { root, _lock: lock };
+        // The lock is held now, so every staging temporary still in the root
+        // was abandoned by a killed publisher rather than owned by a live one.
+        remove_abandoned_temporaries(&spool.root, DIRECTORY_POLICY)
+            .map_err(|_| HookDeliverySpoolError::Io)?;
         spool.receipt_paths()?;
         Ok(spool)
     }
@@ -306,7 +311,7 @@ impl HookDeliveryReceiptSpoolV1 {
                 .file_name()
                 .into_string()
                 .map_err(|_| HookDeliverySpoolError::UnsafePath)?;
-            if name == LOCK_FILE {
+            if name == LOCK_FILE || is_owned_temporary_name(&name) {
                 continue;
             }
             if !valid_receipt_name(&name)

crates/tracedecay-private-fs/src/framed_log.rs

@@ -214,12 +214,14 @@ pub fn read_bounded(path: &Path, maximum: usize) -> io::Result<Option<Vec<u8>>>
     Ok(Some(bytes))
 }
 
+const TEMPORARY_SUFFIX: &str = ".tmp";
+
 fn temporary_path(path: &Path, kind: &str) -> PathBuf {
     static NONCE: AtomicU64 = AtomicU64::new(1);
     let nonce = NONCE.fetch_add(1, Ordering::Relaxed);
     let parent = path.parent().unwrap_or_else(|| Path::new("."));
     parent.join(format!(
-        ".{}.{}.{}.{}.tmp",
+        ".{}.{}.{}.{}{TEMPORARY_SUFFIX}",
         path.file_name()
             .and_then(|name| name.to_str())
             .unwrap_or("spool"),
@@ -229,6 +231,48 @@ fn temporary_path(path: &Path, kind: &str) -> PathBuf {
     ))
 }
 
+/// Is this directory entry one of this module's staging temporaries?
+///
+/// [`with_owned_temp_publish`] stages into `.<destination>.<kind>.<pid>.<nonce>.tmp`
+/// beside the destination and publishes by rename, so a publisher killed
+/// between the two leaves that name behind. A reader enumerating a private
+/// directory must recognize the residue as its own rather than refusing the
+/// whole directory as foreign.
+#[must_use]
+pub fn is_owned_temporary_name(name: &str) -> bool {
+    name.starts_with('.') && name.ends_with(TEMPORARY_SUFFIX)
+}
+
+/// Delete staging temporaries abandoned by a killed publisher.
+///
+/// Only the exclusive owner of `dir` may call this: a live publisher in any
+/// process still owns its staging file. A non-regular entry under a staging
+/// name is foreign and is left in place for the caller's own path validation
+/// to reject.
+pub fn remove_abandoned_temporaries(dir: &Path, policy: DirectorySyncPolicy) -> io::Result<()> {
+    let mut removed = false;
+    for entry in fs::read_dir(dir)? {
+        let entry = entry?;
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else {
+            continue;
+        };
+        if !is_owned_temporary_name(name) {
+            continue;
+        }
+        let path = entry.path();
+        if !fs::symlink_metadata(&path)?.file_type().is_file() {
+            continue;
+        }
+        fs::remove_file(&path)?;
+        removed = true;
+    }
+    if removed {
+        sync_directory(dir, policy)?;
+    }
+    Ok(())
+}
+
 fn remove_owned_temp(path: &Path) {
     let _ = fs::remove_file(path);
 }

crates/tracedecay-store-runtime/src/session_registry/maintenance.rs

@@ -664,7 +664,14 @@ mod tests {
 
     use super::*;
 
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    // The held destructor below parks one worker thread for the whole test, and
+    // `begin_shutdown` injects both aborts at once. A worker drains
+    // `inject.len() / worker_threads + 1` notifications in one batch, so with two
+    // workers the one that blocks can also capture the second shard's abort into
+    // its local queue, where the surviving worker never gets notified to steal it
+    // and the second retirement hangs instead of running slowly. Four workers keep
+    // that batch at one task each and leave spare capacity for the blocked one.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn cancelled_shutdown_retains_incomplete_future_drop_and_independent_retirement() {
         struct HeldDrop {
             started: Arc<Notify>,

crates/tracedecay/src/daemon/tests.rs

@@ -37,6 +37,7 @@ mod invocation_ownership;
 mod lifecycle;
 mod logging;
 mod multi_root_journey;
+mod multi_root_scope_set_cas_mcp;
 mod ownership;
 mod remote_project_deletion;
 mod remote_project_recovery;