fix(daemon): close socket-group and pointer-rename races

crates/tracedecay-application/src/git_intelligence.rs

@@ -1762,6 +1762,14 @@ mod tests {
                 "user.email=fixture@example.com",
                 "-c",
                 "commit.gpgsign=false",
+                // `git commit` spawns a detached `git maintenance run --auto`
+                // that holds `.git/objects/maintenance.lock` after the commit
+                // returns; the byte-identical snapshot must not see it appear
+                // or vanish between its two walks.
+                "-c",
+                "maintenance.auto=false",
+                "-c",
+                "gc.auto=0",
             ])
             .args(args)
             .current_dir(self.path())

crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs

@@ -17,6 +17,10 @@ fn shipped_binary_stops_quietly_when_a_pipeline_reader_exits() {
     let output = Command::new("sh")
         .args(["-c", r#""$TRACEDECAY_BIN" tool | head -n 4"#])
         .env("TRACEDECAY_BIN", env!("CARGO_BIN_EXE_tracedecay"))
+        // A hotpath-enabled binary binds its metrics port on start; when a
+        // sibling test's daemon already holds it, the bind failure lands on
+        // stderr and breaks the quiet-pipeline assertion below.
+        .env("HOTPATH_METRICS_SERVER_OFF", "true")
         .output()
         .expect("tracedecay tool pipeline should run");
 

crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs

@@ -997,6 +997,52 @@ fn text_artifact_retention_collects_staging_database_sidecars_with_their_owner()
     );
 }
 
+/// The inventory scans the artifact root without the generation-store lock, so
+/// the text-artifact builder can retire a `.staging` family between the
+/// directory listing and the stat. A vanished entry is already reclaimed and
+/// must leave the plan intact rather than failing it with a storage error.
+#[test]
+fn text_artifact_inventory_skips_an_entry_reclaimed_during_the_scan() {
+    let store = tempfile::TempDir::new().expect("artifact store");
+    let artifacts_root = code_text_artifacts_root(store.path());
+    std::fs::create_dir_all(&artifacts_root).expect("create artifact root");
+    let staging_family = ["a", "b", "c"]
+        .into_iter()
+        .map(|seed| {
+            let path = artifacts_root.join(format!(".text-artifact-{}.staging", seed.repeat(64)));
+            std::fs::write(&path, b"staging").expect("write staging evidence");
+            path
+        })
+        .collect::<Vec<_>>();
+
+    // The scan probes cancellation once on entry and once per directory entry,
+    // before it takes that entry. Retiring from the third probe on leaves the
+    // listing already taken and one entry already inspected, so every further
+    // name the scan holds names a file that is gone from disk.
+    let probes = std::sync::atomic::AtomicUsize::new(0);
+    let retire_during_the_scan = || {
+        if probes.fetch_add(1, std::sync::atomic::Ordering::Relaxed) >= 2 {
+            for path in &staging_family {
+                let _ = std::fs::remove_file(path);
+            }
+        }
+        false
+    };
+
+    let inventory = plan_collectable_text_artifacts_cancellable(
+        store.path(),
+        None,
+        GenerationDigestVerificationV1::Full,
+        &retire_during_the_scan,
+    )
+    .expect("an entry reclaimed mid-scan leaves the store plannable");
+    assert!(
+        inventory.candidates.len() < staging_family.len(),
+        "an entry that vanished before its stat is reclaimed, not planned: {:?}",
+        inventory.candidates
+    );
+}
+
 #[test]
 fn applied_retention_refuses_a_busy_generation_store_and_retries() {
     let (store, _) = fixture_store(2);

crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs

@@ -198,6 +198,15 @@ fn mutate_verified_text_artifact_under_lock(
             "publication pointer exceeds its durable byte bound".to_owned(),
         ));
     }
+    // Re-read immediately before the rename. A pointer that is no longer the
+    // one this mutation observed — including a truncated file — must not be
+    // replaced by the in-memory copy.
+    let current = read_active_pointer(store_root)?;
+    if &current != expected_pointer {
+        return Err(CodeGenerationRetentionErrorV1::Conflict(
+            "active generation pointer changed before text-artifact mutation".to_owned(),
+        ));
+    }
     atomic_write(
         &store_root.join(ACTIVE_POINTER_FILE),
         "code-generation-text-artifact-mutation",
@@ -394,7 +403,22 @@ pub(super) fn plan_collectable_text_artifacts_cancellable(
             )
         })?;
         let path = entry.path();
-        let metadata = std::fs::symlink_metadata(&path).map_err(storage)?;
+        // This inventory reads the artifact root without the generation-store
+        // lock, so an entry the listing just named can already be gone: the
+        // text-artifact builder retires a `.staging` family (the staging
+        // database and its `-journal`/`-wal`/`-shm` sidecars) under that lock
+        // while this scan runs. A vanished entry is reclaimed, which is what
+        // this inventory would have planned anyway, so it is not a candidate
+        // and not a failure. Failing the plan here turned every publish that
+        // raced a maintenance tick into a loud `retention_plan_failed` pass
+        // (master run 35422072661, `Storage("No such file or directory")`).
+        // A completed artifact the durable index *references* is verified
+        // above, before this scan, and stays fail-closed if it disappears.
+        let metadata = match std::fs::symlink_metadata(&path) {
+            Ok(metadata) => metadata,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(storage(error)),
+        };
         if !metadata.file_type().is_file() {
             return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
                 "code text artifact inventory path '{}' is not a regular file",

crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs

@@ -1230,38 +1230,94 @@ impl DaemonCodeIndexPublicationStoreV1 {
                 "durable code-generation index exceeds its retention bounds",
             ));
         }
-        *self
+        let mut memo = self
             .pointer_memo
             .lock()
-            .unwrap_or_else(PoisonError::into_inner) = Some(PublicationPointerMemoV1 {
-            mtime,
-            size,
-            digest,
-            pointer: pointer.clone(),
-        });
+            .unwrap_or_else(PoisonError::into_inner);
+        // Install only when the file is still the bytes just parsed. A rename
+        // that landed during validation owns the memo.
+        if std::fs::read(&self.active_path).ok().as_deref() == Some(bytes.as_slice()) {
+            *memo = Some(PublicationPointerMemoV1 {
+                mtime,
+                size,
+                digest,
+                pointer: pointer.clone(),
+            });
+        }
         Ok(Some(pointer))
     }
 
     fn remember_publication_pointer(&self, pointer: &DurablePublicationPointerV1, bytes: &[u8]) {
-        let metadata = match std::fs::metadata(&self.active_path) {
-            Ok(metadata) => metadata,
-            Err(_) => {
-                *self
-                    .pointer_memo
-                    .lock()
-                    .unwrap_or_else(PoisonError::into_inner) = None;
-                return;
-            }
-        };
-        *self
+        let mut memo = self
             .pointer_memo
             .lock()
-            .unwrap_or_else(PoisonError::into_inner) = Some(PublicationPointerMemoV1 {
-            mtime: metadata.modified().ok(),
-            size: metadata.len(),
-            digest: Self::state_digest(bytes),
-            pointer: pointer.clone(),
-        });
+            .unwrap_or_else(PoisonError::into_inner);
+        // The memo and the file it names are one critical section. A publisher
+        // that observed older bytes must not install them over a newer file.
+        match std::fs::read(&self.active_path) {
+            Ok(current) if current == bytes => {
+                let metadata = std::fs::metadata(&self.active_path).ok();
+                *memo = Some(PublicationPointerMemoV1 {
+                    mtime: metadata
+                        .as_ref()
+                        .and_then(|metadata| metadata.modified().ok()),
+                    size: metadata.map_or(0, |metadata| metadata.len()),
+                    digest: Self::state_digest(bytes),
+                    pointer: pointer.clone(),
+                });
+            }
+            Ok(_) => {}
+            Err(_) => *memo = None,
+        }
+    }
+
+    /// Replace the active pointer only when it is still the exact bytes this
+    /// publication observed under the store lock.
+    ///
+    /// `rename(2)` replaces whatever occupies the path, including a truncated
+    /// or rewritten pointer. The observation is the compare-and-swap token:
+    /// a mismatch is a refusal, not a rewrite. `lock` is the witness that
+    /// this critical section is the exclusive owner of the store.
+    pub(super) fn commit_observed_pointer(
+        &self,
+        _lock: &CodeGenerationStoreLockV1,
+        observed: Option<&[u8]>,
+        pointer: &DurablePublicationPointerV1,
+        bytes: &[u8],
+    ) -> Result<(), CodeIndexPublicationStoreErrorV1> {
+        let current = match std::fs::read(&self.active_path) {
+            Ok(current) => Some(current),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+            Err(error) => return Err(Self::unavailable(error)),
+        };
+        if current.as_deref() != observed {
+            return Err(match current {
+                Some(current)
+                    if serde_json::from_slice::<DurablePublicationPointerV1>(&current).is_err() =>
+                {
+                    Self::corruption("active code-generation pointer is corrupt")
+                }
+                _ => CodeIndexPublicationStoreErrorV1::CompareAndSwap,
+            });
+        }
+        let temporary = self
+            .active_path
+            .with_extension(format!("json.{}.tmp", std::process::id()));
+        if temporary.exists() {
+            std::fs::remove_file(&temporary).map_err(Self::unavailable)?;
+        }
+        Self::write_durable(&temporary, bytes)?;
+        if let Err(error) = std::fs::rename(&temporary, &self.active_path) {
+            let _ = std::fs::remove_file(&temporary);
+            return Err(Self::unavailable(error));
+        }
+        Self::sync_directory(
+            self.active_path
+                .parent()
+                .ok_or_else(|| Self::unavailable("active pointer has no parent directory"))?,
+        )?;
+        self.remember_publication_pointer(pointer, bytes);
+        Ok(())
     }
 
     pub(super) fn read_retained_partitioned_segment(
@@ -2174,6 +2230,15 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 {
         } else {
             self.read_publication_pointer()?
         };
+        // The bytes behind `prior_pointer`, captured under the store lock.
+        // The commit below refuses to rename unless the file is still these
+        // exact bytes, so a pointer that changed after this observation is
+        // not overwritten.
+        let prior_bytes = if prior_pointer.is_some() {
+            Some(std::fs::read(&self.active_path).map_err(Self::unavailable)?)
+        } else {
+            None
+        };
         if undecoded_expectation.is_none()
             && prior_pointer
                 .as_ref()
@@ -2551,22 +2616,8 @@ impl CodeIndexAtomicPublicationPort for DaemonCodeIndexPublicationStoreV1 {
         } else {
             None
         };
-        let temporary = self
-            .active_path
-            .with_extension(format!("json.{}.tmp", std::process::id()));
-        if temporary.exists() {
-            std::fs::remove_file(&temporary).map_err(Self::unavailable)?;
-        }
         hotpath::measure_block!("code_index.generation.publish.pointer_commit", {
-            Self::write_durable(&temporary, &bytes)?;
-            std::fs::rename(&temporary, &self.active_path).map_err(Self::unavailable)?;
-            Self::sync_directory(
-                self.active_path
-                    .parent()
-                    .ok_or_else(|| Self::unavailable("active pointer has no parent directory"))?,
-            )?;
-            self.remember_publication_pointer(&pointer, &bytes);
-            Ok::<(), CodeIndexPublicationStoreErrorV1>(())
+            self.commit_observed_pointer(&_store_lock, prior_bytes.as_deref(), &pointer, &bytes)
         })?;
         drop(source_fence);
         let mut state = self.cache.lock_state()?;

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

@@ -2648,3 +2648,58 @@ fn publication_pointer_memo_follows_bytes_when_size_and_mtime_stay_put() {
         "equal size and mtime must not reuse the previous pointer"
     );
 }
+
+#[test]
+fn stale_pointer_commit_does_not_replace_a_changed_active_pointer() {
+    let store = TempDir::new().expect("store root");
+    let project = TempDir::new().expect("project root");
+    let publication = super::super::DaemonCodeIndexPublicationStoreV1::new(
+        store.path(),
+        project.path(),
+        SanitizerRevision::new(tracedecay_privacy::CODE_SOURCE_SANITIZER_VERSION_V1)
+            .expect("sanitizer revision"),
+    )
+    .expect("open publication store");
+    let pointer_path = store.path().join("active-code-generation-v1.json");
+    let observed = same_length_publication_pointer("generation.observed", 0x31);
+    let observed_bytes = serde_json::to_vec(&observed).expect("encode observed pointer");
+    let replacement = same_length_publication_pointer("generation.replacement", 0x32);
+    let replacement_bytes = serde_json::to_vec(&replacement).expect("encode replacement pointer");
+    std::fs::write(&pointer_path, &observed_bytes).expect("write observed pointer");
+    let store_lock = acquire_code_generation_store_lock(store.path()).expect("store lock");
+
+    publication
+        .commit_observed_pointer(
+            &store_lock,
+            Some(&observed_bytes),
+            &replacement,
+            &replacement_bytes,
+        )
+        .expect("matching observation publishes");
+    assert_eq!(
+        std::fs::read(&pointer_path).expect("published pointer"),
+        replacement_bytes
+    );
+
+    std::fs::write(&pointer_path, b"{").expect("truncate active pointer");
+    let error = publication
+        .commit_observed_pointer(
+            &store_lock,
+            Some(&replacement_bytes),
+            &observed,
+            &observed_bytes,
+        )
+        .expect_err("a stale observation must not publish");
+    assert!(
+        matches!(
+            error,
+            CodeIndexPublicationStoreErrorV1::CorruptionResetRequired(_)
+        ),
+        "corrupt pointer is a closed publication failure, not a rewrite: {error:?}"
+    );
+    assert_eq!(
+        std::fs::read(&pointer_path).expect("faulted pointer remains"),
+        b"{",
+        "the truncated pointer must still be the file"
+    );
+}

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

@@ -4022,7 +4022,9 @@ async fn diagnostics_change_generation_advances_for_out_of_band_git_drift() {
 async fn elapsed_freshness_window_alone_does_not_make_dashboard_state_stale() {
     let fixture = GitFixture::new(&[("src/main.rs", "fn main() {}\n")]);
     let store = TempDir::new().expect("store root");
-    let registry = CodeIndexSchedulerRegistryV1::new(1);
+    // Single-permit admission: holding it below parks the background worker,
+    // which the host's default bound cannot do.
+    let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1);
     registry
         .mount_worktree(
             test_project_id(),
@@ -4033,18 +4035,33 @@ async fn elapsed_freshness_window_alone_does_not_make_dashboard_state_stale() {
         .expect("mount daemon-owned scheduler");
     wait_for_initial_generation(&registry, fixture.path()).await;
     wait_for_dashboard_ready(&registry, fixture.path()).await;
+    // The mount leaves clone backfill behind, and the wakes that drain it
+    // leave a banked permit whose no-op pass projects `Verifying` instead of
+    // `Fresh` (CI run 35425541839). Settle the mount-era chain, hold the
+    // admission so no pass can start under the sample, and prove the
+    // pending-wake slot stays empty, exactly as the text-progress test does.
+    drain_clone_backfill(&registry, fixture.path()).await;
+    settled_owner_with_idle_admission(&registry, fixture.path()).await;
+    let _quiet_owner = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
     let canonical = fixture.path().canonicalize().expect("canonical fixture");
-    {
+    let scope = {
         let mounted = registry.mounted.lock().await;
-        mounted
-            .get(&canonical)
-            .expect("mounted worktree")
+        let worktree = mounted.get(&canonical).expect("mounted worktree");
+        worktree
             .scheduler
             .lock()
             .expect("scheduler")
             .policy
             .staleness_threshold = Duration::ZERO;
-    }
+        tracedecay_contracts::ResolvedScope::new(
+            test_project_id(),
+            worktree.repository_id.clone(),
+            worktree.worktree_id.clone(),
+            None,
+        )
+        .expect("resolved scope")
+    };
+    clear_pending_wake_until_quiet(&registry, &scope).await;
 
     let projected = registry
         .dashboard_freshness(fixture.path())