diff --git a/crates/tracedecay-application/src/git_intelligence.rs b/crates/tracedecay-application/src/git_intelligence.rs
index 5046bbf93d..ed8ff8af72 100644
--- a/crates/tracedecay-application/src/git_intelligence.rs
+++ b/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())
diff --git a/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs b/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs
index 4df4e34dd5..827cdaece2 100644
--- a/crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs
+++ b/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");
 
diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs
index 2dbe91a0c7..15ae71fb0e 100644
--- a/crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs
+++ b/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);
diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs
index c62e73dc92..d56bcbb6ac 100644
--- a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs
+++ b/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",
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs
index 3422489f5e..9be2ec5f7f 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs
+++ b/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()?;
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs
index 95f3f7a4d7..f5285d04d0 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/publication_store.rs
+++ b/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"
+    );
+}
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
index 138157469d..0ac4e3a71f 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
+++ b/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())
diff --git a/crates/tracedecay-daemon-control/src/service/probe.rs b/crates/tracedecay-daemon-control/src/service/probe.rs
index 3cec2582bd..c974b6a920 100644
--- a/crates/tracedecay-daemon-control/src/service/probe.rs
+++ b/crates/tracedecay-daemon-control/src/service/probe.rs
@@ -215,7 +215,9 @@ fn classify_daemon_protocol_identity(
     match identity {
         Ok((name, version))
             if name.as_deref() == Some("tracedecay")
-                && version.as_deref() == Some(expected_version) =>
+                && version.as_deref().is_some_and(|version| {
+                    tracedecay_daemon_protocol::versions_name_same_build(version, expected_version)
+                }) =>
         {
             DaemonProtocolState::Ready
         }
@@ -638,6 +640,56 @@ fn missing_loopback_authority() -> TraceDecayError {
     }
 }
 
+#[cfg(test)]
+mod identity_classification_tests {
+    use super::{DaemonProtocolState, classify_daemon_protocol_identity};
+
+    const SHA: &str = "84598a0b9c841b914565f46b20bb6c765706e8e5";
+
+    /// The identity a `tracedecay` daemon reporting `version` answers with.
+    fn identity(version: &str) -> (Option<String>, Option<String>) {
+        (Some("tracedecay".to_owned()), Some(version.to_owned()))
+    }
+
+    /// `tracedecay update` installs a release and then waits for the daemon
+    /// that binary starts. The release path knows the version it installed as
+    /// the bare release, while the daemon names the commit it was built from,
+    /// so readiness used to refuse the very binary it had just installed.
+    #[test]
+    fn a_daemon_naming_its_commit_is_ready_against_its_bare_release() {
+        assert_eq!(
+            classify_daemon_protocol_identity(
+                Ok(identity(&format!("0.1.0-beta.47+{SHA}"))),
+                "0.1.0-beta.47",
+            ),
+            DaemonProtocolState::Ready
+        );
+    }
+
+    /// A genuinely stale daemon is still refused, whichever side names a
+    /// commit.
+    #[test]
+    fn a_different_build_is_still_an_identity_mismatch() {
+        let stale = format!("0.1.0-beta.46+{SHA}");
+        assert_eq!(
+            classify_daemon_protocol_identity(Ok(identity(&stale)), "0.1.0-beta.47"),
+            DaemonProtocolState::IdentityMismatch {
+                name: Some("tracedecay".to_owned()),
+                version: Some(stale),
+                expected_version: "0.1.0-beta.47".to_owned(),
+            }
+        );
+        let other_commit = format!("0.1.0-beta.47+{}", "b".repeat(40));
+        assert!(matches!(
+            classify_daemon_protocol_identity(
+                Ok(identity(&other_commit)),
+                &format!("0.1.0-beta.47+{SHA}"),
+            ),
+            DaemonProtocolState::IdentityMismatch { .. }
+        ));
+    }
+}
+
 #[cfg(test)]
 mod timeout_classification_tests {
     use std::io::{self, Cursor, Read, Write};
diff --git a/crates/tracedecay-daemon-protocol/src/handshake.rs b/crates/tracedecay-daemon-protocol/src/handshake.rs
index 231c3ffa85..9bc27c73c6 100644
--- a/crates/tracedecay-daemon-protocol/src/handshake.rs
+++ b/crates/tracedecay-daemon-protocol/src/handshake.rs
@@ -146,12 +146,41 @@ impl DaemonHandshakeRefusal {
 /// Old clients send no version (empty string); that is indistinguishable from
 /// "same version before this field existed", so it never counts as skew.
 pub fn client_version_skew(client_version: &str, daemon_version: &str) -> Option<String> {
-    if client_version.is_empty() || client_version == daemon_version {
+    if client_version.is_empty() || versions_name_same_build(client_version, daemon_version) {
         return None;
     }
     Some(client_version.to_string())
 }
 
+/// Whether two reported versions name the same binary, the one comparison
+/// every version identity check in the product runs.
+///
+/// A version is `"{release}"` or `"{release}+{full sha}[.dirty]"`, and `SemVer`
+/// requires build metadata to be ignored for precedence. A side that reports
+/// only the release is therefore **less specific**, not different: the release
+/// tag `v0.1.0-beta.47` and the binary that names itself
+/// `0.1.0-beta.47+<sha>` are one identity, and treating them as a mismatch is
+/// what failed `tracedecay update`'s own readiness wait against the daemon it
+/// had just installed.
+///
+/// When both sides do name a commit they must name the same one, which keeps
+/// the skew this comparison was added to catch (367a44ad00): two checkout
+/// builds of one release differ only by commit, and a daemon left running from
+/// the previous build is exactly that case.
+#[must_use]
+pub fn versions_name_same_build(left: &str, right: &str) -> bool {
+    let (Some(left_release), Some(right_release)) = (release_version(left), release_version(right))
+    else {
+        // Neither side is a version this comparison understands, so refuse to
+        // guess and fall back to the literal texts.
+        return left == right;
+    };
+    left_release.cmp_precedence(&right_release) == std::cmp::Ordering::Equal
+        && (left_release.build == right_release.build
+            || left_release.build.is_empty()
+            || right_release.build.is_empty())
+}
+
 fn release_version(version: &str) -> Option<semver::Version> {
     semver::Version::parse(version.strip_prefix('v').unwrap_or(version)).ok()
 }
@@ -227,6 +256,52 @@ mod handshake_refusal_tests {
         );
     }
 
+    /// The observed `tracedecay update` failure: the release path reports the
+    /// bare release it installed while the daemon that binary starts names its
+    /// own commit, so readiness compared `0.1.0-beta.47` against
+    /// `0.1.0-beta.47+<sha>` and refused the daemon it had just installed.
+    #[test]
+    fn a_bare_release_and_its_own_build_are_one_identity() {
+        let build = "0.1.0-beta.47+84598a0b9c841b914565f46b20bb6c765706e8e5";
+        assert!(versions_name_same_build("0.1.0-beta.47", build));
+        assert!(versions_name_same_build(build, "0.1.0-beta.47"));
+        assert!(
+            versions_name_same_build("v0.1.0-beta.47", build),
+            "the GitHub release tag names the same identity as the binary it ships"
+        );
+        assert_eq!(client_version_skew("0.1.0-beta.47", build), None);
+        assert_eq!(client_version_skew(build, "0.1.0-beta.47"), None);
+    }
+
+    /// Build metadata still separates two builds of one release, the skew this
+    /// comparison exists to catch.
+    #[test]
+    fn two_commits_of_one_release_stay_distinguishable() {
+        let older = "0.1.0-beta.47+aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        let newer = "0.1.0-beta.47+bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+        assert!(!versions_name_same_build(older, newer));
+        assert_eq!(client_version_skew(older, newer), Some(older.to_owned()));
+        assert!(!versions_name_same_build(
+            "0.1.0-beta.46",
+            "0.1.0-beta.47+aaaa"
+        ));
+        assert!(
+            !versions_name_same_build(
+                older,
+                "0.1.0-beta.47+aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.dirty"
+            ),
+            "a dirty worktree is not the commit it was built from"
+        );
+    }
+
+    /// Nothing that fails to parse may be declared a match by accident.
+    #[test]
+    fn unparseable_versions_compare_literally() {
+        assert!(versions_name_same_build("not-a-version", "not-a-version"));
+        assert!(!versions_name_same_build("not-a-version", "0.1.0-beta.47"));
+        assert!(!versions_name_same_build("", "0.1.0-beta.47"));
+    }
+
     #[test]
     fn foreign_lines_never_parse_as_refusal_frames() {
         assert_eq!(DaemonHandshakeRefusal::from_line("{}"), None);
diff --git a/crates/tracedecay-daemon-protocol/src/lib.rs b/crates/tracedecay-daemon-protocol/src/lib.rs
index bc9112b66a..10b4fe309b 100644
--- a/crates/tracedecay-daemon-protocol/src/lib.rs
+++ b/crates/tracedecay-daemon-protocol/src/lib.rs
@@ -99,6 +99,7 @@ pub use contract::{
 pub use handshake::{
     DAEMON_HANDSHAKE_REFUSAL_PROTOCOL, DaemonHandshake, DaemonHandshakeRefusal,
     DaemonHandshakeRefusalReason, MovedStoreAdoption, client_version_skew, version_skew_action,
+    versions_name_same_build,
 };
 pub use lsp_wire::{
     ConnectionLocalRequestSequence, FramePoll, FrameSend, LspFrame, LspSessionAccess,
diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs
index 39200507c3..5072da10d1 100644
--- a/crates/tracedecay-global-db/src/tests.rs
+++ b/crates/tracedecay-global-db/src/tests.rs
@@ -739,12 +739,19 @@ async fn single_analytics_append_commits_in_one_writer_dispatch() {
         metadata_json: None,
     };
 
-    let append = harness.registered.append_analytics_event(&event);
-    tokio::pin!(append);
-    assert!(matches!(
-        futures_util::poll!(&mut append),
-        std::task::Poll::Pending
-    ));
+    // One poll hands the INSERT to the writer thread, which autocommits it and
+    // only afterwards answers the reply channel. Whether that answer has already
+    // arrived when the poll returns is a race with that thread, so the poll's own
+    // result is not the contract; abandoning the future before anyone reads the
+    // reply is. A single-dispatch autocommit survives that; a transaction-framed
+    // append would roll back and never reach the inspection connection below.
+    {
+        let append = harness.registered.append_analytics_event(&event);
+        tokio::pin!(append);
+        if let std::task::Poll::Ready(result) = futures_util::poll!(&mut append) {
+            result.expect("single analytics append");
+        }
+    }
 
     let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
     loop {
@@ -758,7 +765,7 @@ async fn single_analytics_append_commits_in_one_writer_dispatch() {
         }
         assert!(
             std::time::Instant::now() < deadline,
-            "single append did not autocommit while its future remained unpolled"
+            "single append did not autocommit after its future was abandoned"
         );
         std::thread::sleep(std::time::Duration::from_millis(5));
     }
diff --git a/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs b/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs
index 99d43cfcc3..fa15cd71fc 100644
--- a/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs
+++ b/crates/tracedecay-mcp/src/handlers/analysis/field_sites.rs
@@ -91,13 +91,12 @@ pub async fn handle_field_sites(
 
                 for site in sites {
                     let line_text = line_at(&source, site.byte).unwrap_or("");
-                    let enclosing = nodes
-                        .iter()
-                        .filter(|n| {
-                            let line = site.line.saturating_sub(1);
-                            n.metadata.start_line <= line && line <= n.end_line()
-                        })
-                        .min_by_key(|n| n.metadata.line_span);
+                    // Attribute by byte containment: a read and a write of the
+                    // same field can share one line, and two declarations can
+                    // too, so a line number cannot say which declaration a
+                    // site is inside. Masking preserves byte layout, so the
+                    // offset the scan reports indexes `source` unchanged.
+                    let enclosing = enclosing_declaration(nodes, site.byte as u64);
                     if let Some(scope) = &qualified_scope {
                         if !scope.target_exists {
                             continue;
@@ -648,3 +647,79 @@ fn line_is_comment(source: &str, byte: usize) -> bool {
     let trimmed = line.trim_start();
     trimmed.starts_with("//")
 }
+
+#[cfg(test)]
+mod field_site_attribution_tests {
+    use super::*;
+    use tracedecay_domain::{ComplexityAnalysisV1, SourceSpan};
+
+    fn digest<T>(byte: char) -> T
+    where
+        T: TryFrom<String>,
+        <T as TryFrom<String>>::Error: std::fmt::Debug,
+    {
+        T::try_from(format!("sha256:{}", byte.to_string().repeat(64))).expect("digest")
+    }
+
+    fn declaration(name: &str, span: std::ops::Range<usize>) -> VerifiedAnalysisSymbol {
+        VerifiedAnalysisSymbol {
+            occurrence: SymbolOccurrenceId::new(format!("occurrence.{name}")).expect("occurrence"),
+            path: "src/lib.rs".to_owned(),
+            source_span: Some(SourceSpan {
+                start_byte: span.start as u64,
+                end_byte: span.end as u64,
+            }),
+            metadata: LineageSymbolRecordV1 {
+                occurrence: SymbolOccurrenceId::new(format!("occurrence.{name}"))
+                    .expect("occurrence"),
+                identity: digest('1'),
+                qualified_name: name.to_owned(),
+                simple_name: name.to_owned(),
+                kind: "function".to_owned(),
+                visibility: "private".to_owned(),
+                branches: 0,
+                loops: 0,
+                max_nesting: 0,
+                complexity_analysis: ComplexityAnalysisV1::Complete,
+                // Both declarations live on line 1: the shape that made
+                // line-based attribution a coin flip.
+                line_span: 1,
+                start_line: 0,
+                signature: None,
+                docstring: None,
+                is_async: false,
+                derives: Vec::new(),
+                skip_test_coverage: false,
+                file_identity: digest('2'),
+                content_digest: digest('3'),
+            },
+        }
+    }
+
+    /// A read and a write of one field, inside two functions that share a
+    /// line, each belong to the function whose bytes contain them. Every
+    /// candidate has `line_span == 1` here, so the old smallest-line-span
+    /// selection had nothing to break the tie with and returned whichever
+    /// symbol the graph page happened to yield first.
+    #[test]
+    fn attributes_a_read_and_a_write_sharing_one_line() {
+        let source = "fn r(s: &S) -> u32 { s.count } fn w(s: &mut S) { s.count = 1; }";
+        let write_start = source.find("fn w").expect("second function");
+        let nodes = vec![
+            declaration("r", 0..write_start),
+            declaration("w", write_start..source.len()),
+        ];
+
+        let sites = find_field_references(source, "count");
+        assert_eq!(sites.len(), 2, "one read and one write: {sites:?}");
+        assert!(matches!(sites[0].kind, FieldRefKind::Read));
+        assert!(matches!(sites[1].kind, FieldRefKind::Write));
+        assert_eq!(sites[0].line, sites[1].line, "both sites share one line");
+
+        for (site, expected) in sites.iter().zip(["r", "w"]) {
+            let enclosing = enclosing_declaration(&nodes, site.byte as u64)
+                .map(|node| node.metadata.qualified_name.as_str());
+            assert_eq!(enclosing, Some(expected), "site {site:?}");
+        }
+    }
+}
diff --git a/crates/tracedecay-mcp/src/handlers/analysis/mod.rs b/crates/tracedecay-mcp/src/handlers/analysis/mod.rs
index e3f3ffe05b..dcd9a84d82 100644
--- a/crates/tracedecay-mcp/src/handlers/analysis/mod.rs
+++ b/crates/tracedecay-mcp/src/handlers/analysis/mod.rs
@@ -43,7 +43,7 @@ use tracedecay_code_index::graph_projection::CodeGraphSemanticEdgeV1;
 use tracedecay_code_index::lineage::LineageSymbolRecordV1;
 use tracedecay_domain::code_intelligence::NodeKind;
 use tracedecay_domain::errors::{Result, TraceDecayError};
-use tracedecay_domain::{RelationEdgeKindV1, SymbolOccurrenceId};
+use tracedecay_domain::{RelationEdgeKindV1, SourceSpan, SymbolOccurrenceId};
 use tracedecay_graph_query::VerifiedGraphQuery;
 
 fn path_is_rust(path: &str) -> bool {
@@ -63,6 +63,9 @@ const ANALYSIS_RELATION_BUDGET: usize = 2_000_000;
 struct VerifiedAnalysisSymbol {
     occurrence: SymbolOccurrenceId,
     path: String,
+    /// Byte range the declaration occupies in its file. Line numbers cannot
+    /// separate two declarations that share one line; this can.
+    source_span: Option<SourceSpan>,
     metadata: LineageSymbolRecordV1,
 }
 
@@ -74,6 +77,25 @@ impl VerifiedAnalysisSymbol {
     }
 }
 
+/// Innermost declaration whose source range covers `match_byte`.
+///
+/// Byte containment, not line containment: an attribute such as `#[test]` and
+/// the two functions on `#[test] fn a() {…} fn b() {…}` all sit on one line,
+/// and only the byte range says which of them the site is inside. Selecting by
+/// line also had no stable order to break ties with, since symbols arrive in
+/// occurrence order and occurrence ids are per-project digests.
+fn enclosing_declaration(
+    nodes: &[VerifiedAnalysisSymbol],
+    match_byte: u64,
+) -> Option<&VerifiedAnalysisSymbol> {
+    nodes
+        .iter()
+        .filter_map(|node| node.source_span.map(|span| (node, span)))
+        .filter(|(_, span)| span.start_byte <= match_byte && match_byte < span.end_byte)
+        .min_by_key(|(_, span)| span.end_byte.saturating_sub(span.start_byte))
+        .map(|(node, _)| node)
+}
+
 fn verified_analysis_symbols(
     graph: &VerifiedGraphQuery,
     scope_prefix: Option<&str>,
@@ -89,6 +111,10 @@ fn verified_analysis_symbols(
     page.symbols
         .into_iter()
         .map(|symbol| {
+            let source_span = symbol
+                .binding
+                .as_ref()
+                .and_then(|binding| binding.source_span);
             let path = symbol
                 .binding
                 .and_then(|binding| binding.logical_path)
@@ -109,6 +135,7 @@ fn verified_analysis_symbols(
             Ok(VerifiedAnalysisSymbol {
                 occurrence: symbol.occurrence,
                 path,
+                source_span,
                 metadata,
             })
         })
diff --git a/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs b/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs
index 920b28a21c..c0d587c1e8 100644
--- a/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs
+++ b/crates/tracedecay-mcp/src/handlers/analysis/unsafe_patterns.rs
@@ -8,7 +8,8 @@ use tracedecay_domain::errors::{Result, TraceDecayError};
 use tracedecay_graph_query::VerifiedGraphQuery;
 
 use super::{
-    VerifiedAnalysisSymbol, path_is_rust, verified_analysis_symbols, verified_analysis_unavailable,
+    VerifiedAnalysisSymbol, enclosing_declaration, path_is_rust, verified_analysis_symbols,
+    verified_analysis_unavailable,
 };
 use crate::ToolResult;
 use crate::handlers::support::{effective_path, rendered_tool_result};
@@ -41,23 +42,28 @@ fn source_may_contain_unsafe_kind(source: &str, kind: &str) -> bool {
     }
 }
 
-fn line_matches_unsafe_kind(line: &str, kind: &str) -> bool {
+/// Byte offset of the risky construct within `line`, when the line has one.
+///
+/// The offset is what lets a match be attributed to the declaration that
+/// actually contains it: two declarations can share a line, so a line number
+/// alone cannot say which one a site belongs to.
+fn line_matches_unsafe_kind(line: &str, kind: &str) -> Option<usize> {
     let trimmed = line.trim_start();
     if trimmed.starts_with("//") || trimmed.starts_with("///") {
-        return false;
+        return None;
     }
     match kind {
         "unwrap" => contains_method_call(line, "unwrap", true),
         "expect" => contains_method_call(line, "expect", false),
-        "panic" => line.contains("panic!("),
-        "todo" => line.contains("todo!("),
-        "unimplemented" => line.contains("unimplemented!("),
+        "panic" => line.find("panic!("),
+        "todo" => line.find("todo!("),
+        "unimplemented" => line.find("unimplemented!("),
         "unsafe_block" => contains_unsafe_block_start(line),
-        _ => false,
+        _ => None,
     }
 }
 
-fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> bool {
+fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> Option<usize> {
     let needle = format!(".{method}");
     let bytes = line.as_bytes();
     let mut start = 0usize;
@@ -69,18 +75,18 @@ fn contains_method_call(line: &str, method: &str, empty_parens: bool) -> bool {
         if is_word_boundary && next == Some(b'(') {
             if empty_parens {
                 if line[after + 1..].trim_start().starts_with(')') {
-                    return true;
+                    return Some(abs);
                 }
             } else {
-                return true;
+                return Some(abs);
             }
         }
         start = abs + needle.len();
     }
-    false
+    None
 }
 
-fn contains_unsafe_block_start(line: &str) -> bool {
+fn contains_unsafe_block_start(line: &str) -> Option<usize> {
     let bytes = line.as_bytes();
     let mut start = 0usize;
     while let Some(pos) = line[start..].find("unsafe") {
@@ -97,12 +103,12 @@ fn contains_unsafe_block_start(line: &str) -> bool {
                 || rest.starts_with("impl ")
                 || rest.starts_with("trait ")
             {
-                return true;
+                return Some(abs);
             }
         }
         start = abs + "unsafe".len();
     }
-    false
+    None
 }
 
 fn path_looks_like_test(path: &str) -> bool {
@@ -216,7 +222,16 @@ pub async fn handle_unsafe_patterns(
                 // Masking can erase every raw hit (all of them in comments or
                 // string literals), so the file's nodes are fetched only once a
                 // real match survives.
-                for (idx, (line, masked_line)) in source.lines().zip(masked.lines()).enumerate() {
+                // Split inclusively so each line keeps its own byte offset;
+                // masking preserves byte layout, so the two sides stay aligned.
+                let mut line_start = 0usize;
+                for (idx, (line, masked_line)) in source
+                    .split_inclusive('\n')
+                    .zip(masked.split_inclusive('\n'))
+                    .enumerate()
+                {
+                    let line_offset = line_start;
+                    line_start += line.len();
                     let line_no = (idx as u32) + 1;
                     // A mixed test/production line is not wholly test scope,
                     // so keep its production risk visible.
@@ -225,16 +240,11 @@ pub async fn handle_unsafe_patterns(
                         continue;
                     }
                     for kind in &kinds {
-                        if line_matches_unsafe_kind(masked_line, kind) {
+                        if let Some(column) = line_matches_unsafe_kind(masked_line, kind) {
                             let nodes = symbols_by_file.get(file).map_or(&[][..], Vec::as_slice);
-                            let enclosing = nodes
-                                .iter()
-                                .filter(|n| {
-                                    n.metadata.start_line.saturating_add(1) <= line_no
-                                        && line_no <= n.end_line().saturating_add(1)
-                                })
-                                .min_by_key(|n| n.metadata.line_span)
-                                .map(|n| n.metadata.qualified_name.clone());
+                            let enclosing =
+                                enclosing_declaration(nodes, (line_offset + column) as u64)
+                                    .map(|node| node.metadata.qualified_name.clone());
                             *by_kind.entry(kind.clone()).or_insert(0) += 1;
                             matches.push(json!({
                                 "kind": kind,
@@ -314,7 +324,7 @@ mod unsafe_pattern_detection_tests {
 
         for line in lines {
             for kind in kinds {
-                if line_matches_unsafe_kind(line, kind) {
+                if line_matches_unsafe_kind(line, kind).is_some() {
                     assert!(
                         source_may_contain_unsafe_kind(line, kind),
                         "prefilter would drop a real {kind} site: {line:?}"
@@ -337,47 +347,46 @@ mod unsafe_pattern_detection_tests {
     fn detects_unsafe_block_inside_safe_fn() {
         // An `unsafe { }` block living inside an otherwise-safe function, the
         // exact shape the audit fixture plants.
-        assert!(line_matches_unsafe_kind(
-            "    unsafe { *ptr as usize }",
-            "unsafe_block"
-        ));
-        assert!(contains_unsafe_block_start("    unsafe { *ptr as usize }"));
+        assert!(line_matches_unsafe_kind("    unsafe { *ptr as usize }", "unsafe_block").is_some());
+        assert!(contains_unsafe_block_start("    unsafe { *ptr as usize }").is_some());
     }
 
     #[test]
     fn detects_unsafe_fn_impl_and_trait() {
-        assert!(line_matches_unsafe_kind(
-            "pub unsafe fn raw(&self) {",
-            "unsafe_block"
-        ));
-        assert!(line_matches_unsafe_kind(
-            "unsafe impl Send for Foo {}",
-            "unsafe_block"
-        ));
-        assert!(line_matches_unsafe_kind(
-            "unsafe trait Zeroable {}",
-            "unsafe_block"
-        ));
+        assert!(line_matches_unsafe_kind("pub unsafe fn raw(&self) {", "unsafe_block").is_some());
+        assert!(line_matches_unsafe_kind("unsafe impl Send for Foo {}", "unsafe_block").is_some());
+        assert!(line_matches_unsafe_kind("unsafe trait Zeroable {}", "unsafe_block").is_some());
     }
 
     #[test]
     fn ignores_safe_code_and_comments() {
         // Plain safe code has no unsafe markers.
-        assert!(!line_matches_unsafe_kind(
-            "let x = total as usize;",
-            "unsafe_block"
-        ));
+        assert!(line_matches_unsafe_kind("let x = total as usize;", "unsafe_block").is_none());
         // The word appears only in a comment/doc line: not a real unsafe site.
-        assert!(!line_matches_unsafe_kind(
-            "// this is not unsafe { } really",
-            "unsafe_block"
-        ));
-        assert!(!line_matches_unsafe_kind(
-            "/// drop the needless unsafe block",
-            "unsafe_block"
-        ));
+        assert!(
+            line_matches_unsafe_kind("// this is not unsafe { } really", "unsafe_block").is_none()
+        );
+        assert!(
+            line_matches_unsafe_kind("/// drop the needless unsafe block", "unsafe_block")
+                .is_none()
+        );
         // A substring of a longer identifier must not trip the word-boundary check.
-        assert!(!contains_unsafe_block_start("let unsafely = 1;"));
-        assert!(!contains_unsafe_block_start("let make_unsafe_thing = 2;"));
+        assert!(contains_unsafe_block_start("let unsafely = 1;").is_none());
+        assert!(contains_unsafe_block_start("let make_unsafe_thing = 2;").is_none());
+    }
+
+    /// The reported offset is what attributes a site to a declaration, so it
+    /// has to point at the construct itself, not at the start of the line.
+    #[test]
+    fn reports_where_on_the_line_the_site_is() {
+        let line = "#[test] fn a() { Some(5).unwrap(); } pub fn b() { panic!(); }";
+        assert_eq!(
+            line_matches_unsafe_kind(line, "unwrap"),
+            Some(line.find(".unwrap()").expect("unwrap call"))
+        );
+        assert_eq!(
+            line_matches_unsafe_kind(line, "panic"),
+            Some(line.find("panic!(").expect("panic call"))
+        );
     }
 }
diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs
index 922ea3552b..bb09cccdfa 100644
--- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs
+++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/clone_successor.rs
@@ -1,3 +1,4 @@
+use std::collections::{HashMap, HashSet};
 use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
@@ -60,6 +61,7 @@ impl CodeLexicalCloneSuccessorV1 {
         memory_budget_bytes: usize,
     ) -> Result<Self, CodeLexicalArtifactErrorV1> {
         let connection = open_builder_connection(staging_path, memory_budget_bytes)?;
+        ensure_clone_occurrence_indexes(&connection)?;
         let mutation_gate = register_builder_mutation_gate(&connection)?;
         let (prior_digest, format_revision): (String, i64) = connection
             .query_row(
@@ -430,6 +432,24 @@ fn reset_clone_tables(connection: &Connection) -> Result<(), CodeLexicalArtifact
         .map_err(sqlite_error)
 }
 
+/// Lookup indexes for resume verification, which reads postings by
+/// occurrence. Both postings tables are keyed from `class`/`language`, so
+/// without these every per-occurrence read is a full table scan; replaying N
+/// committed pages after a restart then costs N scans of every posting the
+/// repository has, and a daemon sat inside that replay for hours. A prior
+/// artifact copied from a build that predates the indexes gains them here,
+/// and nothing digests or enumerates the index schema.
+fn ensure_clone_occurrence_indexes(
+    connection: &Connection,
+) -> Result<(), CodeLexicalArtifactErrorV1> {
+    connection
+        .execute_batch(
+            "CREATE INDEX IF NOT EXISTS clone_exact_postings_by_occurrence ON clone_exact_postings(symbol_occurrence_id);
+             CREATE INDEX IF NOT EXISTS clone_fingerprint_postings_by_occurrence ON clone_fingerprint_postings(symbol_occurrence_id);",
+        )
+        .map_err(sqlite_error)
+}
+
 fn append_clone_rows(
     transaction: &rusqlite::Transaction<'_>,
     page: &VerifiedSealedLexicalPageV1,
@@ -559,11 +579,106 @@ fn verify_copied_source_page(
     Ok(())
 }
 
+type CloneExactRowV1 = (i64, i64, String, String);
+
+/// Postings for one page's occurrences, read through the occurrence indexes
+/// `ensure_clone_occurrence_indexes` installs and bucketed by occurrence.
+///
+/// The postings tables are keyed from `class`/`language`; before those
+/// indexes existed a per-occurrence read scanned the whole table, and a
+/// daemon spent eleven hours replaying committed pages after a restart.
+struct ClonePagePostingsV1 {
+    exact: HashMap<String, Vec<CloneExactRowV1>>,
+    fingerprints: HashMap<String, Vec<CloneFingerprintRowV1>>,
+}
+
+impl ClonePagePostingsV1 {
+    fn read(
+        connection: &Connection,
+        occurrences: &HashSet<&str>,
+        control: &dyn CodeIndexExecutionControlV1,
+    ) -> Result<Self, CodeLexicalArtifactErrorV1> {
+        let mut exact: HashMap<String, Vec<CloneExactRowV1>> = HashMap::new();
+        let mut statement = connection
+            .prepare(
+                "SELECT class, normalization_revision, digest, payload_digest FROM clone_exact_postings WHERE symbol_occurrence_id = ?1",
+            )
+            .map_err(sqlite_error)?;
+        for occurrence in occurrences {
+            checkpoint(control)?;
+            let mut rows = statement.query([occurrence]).map_err(sqlite_error)?;
+            while let Some(row) = rows.next().map_err(sqlite_error)? {
+                exact.entry((*occurrence).to_owned()).or_default().push((
+                    row.get(0).map_err(sqlite_error)?,
+                    row.get(1).map_err(sqlite_error)?,
+                    row.get(2).map_err(sqlite_error)?,
+                    row.get(3).map_err(sqlite_error)?,
+                ));
+            }
+        }
+        drop(statement);
+
+        let mut fingerprints: HashMap<String, Vec<CloneFingerprintRowV1>> = HashMap::new();
+        let mut statement = connection
+            .prepare(
+                "SELECT language, class, normalization_revision, fingerprint, token_position, payload_digest, body_digest FROM clone_fingerprint_postings WHERE symbol_occurrence_id = ?1",
+            )
+            .map_err(sqlite_error)?;
+        for occurrence in occurrences {
+            checkpoint(control)?;
+            let mut rows = statement.query([occurrence]).map_err(sqlite_error)?;
+            while let Some(row) = rows.next().map_err(sqlite_error)? {
+                fingerprints
+                    .entry((*occurrence).to_owned())
+                    .or_default()
+                    .push((
+                        row.get(0).map_err(sqlite_error)?,
+                        row.get(1).map_err(sqlite_error)?,
+                        row.get(2).map_err(sqlite_error)?,
+                        row.get(3).map_err(sqlite_error)?,
+                        row.get(4).map_err(sqlite_error)?,
+                        row.get(5).map_err(sqlite_error)?,
+                        row.get(6).map_err(sqlite_error)?,
+                    ));
+            }
+        }
+        drop(statement);
+        checkpoint(control)?;
+
+        // Each table's primary key is unique within one occurrence, so sorting
+        // a bucket reproduces the `ORDER BY` the per-body queries used.
+        for rows in exact.values_mut() {
+            rows.sort();
+        }
+        for rows in fingerprints.values_mut() {
+            rows.sort();
+        }
+        Ok(Self {
+            exact,
+            fingerprints,
+        })
+    }
+
+    fn exact_for(&self, occurrence: &str) -> &[CloneExactRowV1] {
+        self.exact.get(occurrence).map_or(&[], Vec::as_slice)
+    }
+
+    fn fingerprints_for(&self, occurrence: &str) -> &[CloneFingerprintRowV1] {
+        self.fingerprints.get(occurrence).map_or(&[], Vec::as_slice)
+    }
+}
+
 fn verify_clone_page_rows(
     connection: &Connection,
     page: &VerifiedSealedLexicalPageV1,
     control: &dyn CodeIndexExecutionControlV1,
 ) -> Result<(), CodeLexicalArtifactErrorV1> {
+    let occurrences = page
+        .clone_bodies()
+        .iter()
+        .map(|body| body.occurrence.symbol_occurrence_id.as_str())
+        .collect::<HashSet<_>>();
+    let postings = ClonePagePostingsV1::read(connection, &occurrences, control)?;
     for body in page.clone_bodies() {
         checkpoint(control)?;
         let expected_payload = serde_json::to_vec(&body.payload)
@@ -631,24 +746,12 @@ fn verify_clone_page_rows(
                 )
             })
             .collect::<Vec<_>>();
-        let mut statement = connection
-            .prepare(
-                "SELECT class, normalization_revision, digest, payload_digest FROM clone_exact_postings WHERE symbol_occurrence_id = ?1 ORDER BY class, normalization_revision, digest",
-            )
-            .map_err(sqlite_error)?;
-        let stored_postings = statement
-            .query_map([body.occurrence.symbol_occurrence_id.as_str()], |row| {
-                Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
-            })
-            .map_err(sqlite_error)?
-            .collect::<Result<Vec<(i64, i64, String, String)>, _>>()
-            .map_err(sqlite_error)?;
-        if stored_postings != expected_postings {
+        if postings.exact_for(body.occurrence.symbol_occurrence_id.as_str()) != expected_postings {
             return Err(CodeLexicalArtifactErrorV1::Corrupt(
                 "resumed clone postings differ from their sealed source page".to_owned(),
             ));
         }
-        verify_clone_fingerprint_page_rows(connection, body)?;
+        verify_clone_fingerprint_page_rows(&postings, body)?;
     }
     Ok(())
 }
@@ -656,7 +759,7 @@ fn verify_clone_page_rows(
 type CloneFingerprintRowV1 = (String, i64, i64, i64, i64, String, String);
 
 fn verify_clone_fingerprint_page_rows(
-    connection: &Connection,
+    postings: &ClonePagePostingsV1,
     body: &CodeIndexCloneBodyV1,
 ) -> Result<(), CodeLexicalArtifactErrorV1> {
     let mut expected = Vec::new();
@@ -679,26 +782,7 @@ fn verify_clone_fingerprint_page_rows(
         }
     }
     expected.sort();
-    let mut statement = connection
-        .prepare(
-            "SELECT language, class, normalization_revision, fingerprint, token_position, payload_digest, body_digest FROM clone_fingerprint_postings WHERE symbol_occurrence_id = ?1 ORDER BY language, class, normalization_revision, fingerprint, token_position",
-        )
-        .map_err(sqlite_error)?;
-    let stored = statement
-        .query_map([body.occurrence.symbol_occurrence_id.as_str()], |row| {
-            Ok((
-                row.get(0)?,
-                row.get(1)?,
-                row.get(2)?,
-                row.get(3)?,
-                row.get(4)?,
-                row.get(5)?,
-                row.get(6)?,
-            ))
-        })
-        .map_err(sqlite_error)?
-        .collect::<Result<Vec<CloneFingerprintRowV1>, _>>()
-        .map_err(sqlite_error)?;
+    let stored = postings.fingerprints_for(body.occurrence.symbol_occurrence_id.as_str());
     if stored != expected {
         return Err(CodeLexicalArtifactErrorV1::Corrupt(
             "resumed clone fingerprints differ from their sealed source page".to_owned(),
diff --git a/crates/tracedecay-runtime-core/src/db/engine/error.rs b/crates/tracedecay-runtime-core/src/db/engine/error.rs
index 2955a7610e..d2570b249a 100644
--- a/crates/tracedecay-runtime-core/src/db/engine/error.rs
+++ b/crates/tracedecay-runtime-core/src/db/engine/error.rs
@@ -95,6 +95,12 @@ impl From<tracedecay_rusqlite_runtime::exact_sql::ExactSqlError> for Error {
             ExactSqlError::RequestLimitExceeded => {
                 Self::InvalidOperation("SQL request exceeds migration limits".to_owned())
             }
+            // A materialization ceiling is a property of the submitted
+            // statement, not of the engine: the untyped `Runtime` fallback
+            // made callers read it as a transient storage fault and replay it.
+            ExactSqlError::QueryLimitExceeded => Self::InvalidOperation(
+                "exact SQL query materialization exceeded its limit".to_owned(),
+            ),
             ExactSqlError::AuthorityDenied(message) => Self::InvalidOperation(message),
             ExactSqlError::Sqlite {
                 operation,
@@ -142,4 +148,23 @@ impl Error {
             _ => None,
         }
     }
+
+    /// True when replaying this exact statement can never succeed.
+    ///
+    /// A `SQLITE_CONSTRAINT` abort is a schema-contract trigger or constraint
+    /// refusing this exact row, and `InvalidOperation` is an admission or
+    /// materialization ceiling refusing this exact statement. Neither is a
+    /// transient engine condition, so a caller that retries one spins until
+    /// something else changes the durable state.
+    #[hotpath::skip]
+    pub const fn is_deterministic_refusal(&self) -> bool {
+        match self {
+            Self::InvalidOperation(_) => true,
+            Self::StatementBatch { source, .. } => source.is_deterministic_refusal(),
+            _ => matches!(self.sqlite_code(), Some(SQLITE_CONSTRAINT)),
+        }
+    }
 }
+
+/// `SQLITE_CONSTRAINT`: a constraint or `RAISE(ABORT)` trigger refused the row.
+const SQLITE_CONSTRAINT: i32 = 19;
diff --git a/crates/tracedecay-runtime-core/src/db/engine/tests.rs b/crates/tracedecay-runtime-core/src/db/engine/tests.rs
index a3c02d8f47..5e51bd0a70 100644
--- a/crates/tracedecay-runtime-core/src/db/engine/tests.rs
+++ b/crates/tracedecay-runtime-core/src/db/engine/tests.rs
@@ -386,3 +386,43 @@ async fn transaction_statement_batch_reports_the_exact_failed_statement() {
 }
 
 mod async_writer;
+
+#[test]
+fn deterministic_refusals_are_distinguished_from_transient_engine_faults() {
+    use tracedecay_rusqlite_runtime::exact_sql::ExactSqlError;
+
+    // A schema-contract trigger abort refuses this exact row for good.
+    let constraint = Error::Sqlite {
+        operation: "execute",
+        code: Some(19),
+        extended_code: Some(1811),
+        message: "invalid session refresh progress".to_owned(),
+    };
+    assert!(constraint.is_deterministic_refusal());
+    assert!(
+        Error::StatementBatch {
+            index: 0,
+            source: Box::new(constraint),
+        }
+        .is_deterministic_refusal()
+    );
+
+    // A materialization ceiling refuses this exact statement for good, and
+    // must not arrive as an untyped `Runtime` message.
+    let limit = Error::from(ExactSqlError::QueryLimitExceeded);
+    assert!(matches!(limit, Error::InvalidOperation(_)));
+    assert!(limit.is_deterministic_refusal());
+
+    // Contention and I/O faults stay retryable.
+    assert!(!Error::Busy.is_deterministic_refusal());
+    assert!(!Error::Runtime("writer restarted".to_owned()).is_deterministic_refusal());
+    assert!(
+        !Error::Sqlite {
+            operation: "execute",
+            code: Some(5),
+            extended_code: Some(5),
+            message: "database is locked".to_owned(),
+        }
+        .is_deterministic_refusal()
+    );
+}
diff --git a/crates/tracedecay-session-memory/src/session/retrieval.rs b/crates/tracedecay-session-memory/src/session/retrieval.rs
index 2d35c62ea6..da4653155a 100644
--- a/crates/tracedecay-session-memory/src/session/retrieval.rs
+++ b/crates/tracedecay-session-memory/src/session/retrieval.rs
@@ -565,6 +565,19 @@ fn map_execution_error(
         SessionTemporalExecutionError::Denied => SessionRetrievalOutcome::Denied,
         SessionTemporalExecutionError::Unavailable => SessionRetrievalOutcome::Unavailable,
         SessionTemporalExecutionError::ResetRequired => SessionRetrievalOutcome::ResetRequired,
+        // A partial generation is a projection still converging, not an
+        // authoritative empty root: answering `CompleteZero` there publishes
+        // "nothing exists, and that is final" for rows the store has already
+        // committed but not yet published. `map_report` refuses that for an
+        // empty ranked page; the execution-error path owes the same refusal,
+        // so the caller re-reads instead of believing the zero.
+        SessionTemporalExecutionError::Empty {
+            freshness: freshness @ SessionDataFreshness::Partial { generation_lag },
+        } => SessionRetrievalOutcome::Partial {
+            items: Vec::new(),
+            freshness,
+            omitted: generation_lag.max(1),
+        },
         SessionTemporalExecutionError::Empty { freshness } => {
             SessionRetrievalOutcome::CompleteZero { freshness }
         }
@@ -1039,6 +1052,19 @@ mod tests {
         );
     }
 
+    #[test]
+    fn partial_generation_empty_execution_is_never_an_authoritative_zero() {
+        let freshness = SessionDataFreshness::Partial { generation_lag: 1 };
+        assert_eq!(
+            map_execution_error(SessionTemporalExecutionError::Empty { freshness }),
+            SessionRetrievalOutcome::Partial {
+                items: Vec::new(),
+                freshness,
+                omitted: 1,
+            }
+        );
+    }
+
     #[test]
     fn persisted_reset_and_unavailable_remain_distinct() {
         assert_eq!(
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
index ed7fcde8e0..365f4d76f1 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs
@@ -25,6 +25,7 @@ use super::wake::{
     SessionTemporalRefreshWakeState, TerminalAttemptGuard,
 };
 use tracedecay_global_db::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1};
+use tracedecay_runtime_core::db::engine::Error as EngineError;
 use tracedecay_session_temporal_store::{
     SessionRefreshRecoveryV1, SessionRefreshRestartStateV1, SessionTemporalStore,
 };
@@ -634,11 +635,34 @@ async fn session_projection_refresh(
     run_session_temporal_refresh_pass(database, state, projector, policy).await
 }
 
-fn classify_store_error(error: &SessionStoreError) -> SessionTemporalRefreshRetryClass {
-    if error.is_storage() {
-        SessionTemporalRefreshRetryClass::Storage
-    } else {
-        SessionTemporalRefreshRetryClass::Projector
+/// True when replaying this store failure unchanged could still succeed.
+///
+/// `is_storage` only says the failure came from the storage adapter; it does
+/// not say the failure is transient. A schema-contract trigger refusing the
+/// submitted row, or an exact-SQL ceiling refusing the submitted statement, is
+/// deterministic: the worker resubmits the identical request every pass, so
+/// treating it as retryable is an unbounded spin at the backoff cap rather
+/// than a recovery. Those are terminal, and the caller durably fails the
+/// refresh instead of retrying it.
+fn is_retryable_storage(error: &SessionStoreError) -> bool {
+    matches!(error, SessionStoreError::Storage { .. }) && !is_deterministic_refusal(error)
+}
+
+/// True when the durable contract refused the exact submitted row or
+/// statement: a typed store refusal, or an engine failure that replays
+/// identically. Only such a refusal retires a running refresh. An
+/// interrupted pass (cancelled control, deadline, budget) and transient
+/// storage leave the operation for the next pass, which may hold a
+/// different control.
+fn is_deterministic_refusal(error: &SessionStoreError) -> bool {
+    match error {
+        SessionStoreError::Cancelled
+        | SessionStoreError::DeadlineExceeded
+        | SessionStoreError::BudgetExceeded { .. } => false,
+        SessionStoreError::Storage { source, .. } => source
+            .downcast_ref::<EngineError>()
+            .is_some_and(EngineError::is_deterministic_refusal),
+        _ => true,
     }
 }
 
@@ -667,7 +691,7 @@ pub async fn process_refresh_begin_requests(
                     tracedecay_store::SessionRefreshDispositionV1::Joined => report.joined += 1,
                 }
             }
-            Err(error) if error.is_storage() => {
+            Err(error) if is_retryable_storage(&error) => {
                 report.last_error = Some(format!("{error:?}"));
                 report.retryable_errors += 1;
                 report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
@@ -709,7 +733,7 @@ pub async fn begin_admitted_session_refreshes(
     {
         Ok(page) => page,
         Err(error) => {
-            if classify_store_error(&error) == SessionTemporalRefreshRetryClass::Storage {
+            if is_retryable_storage(&error) {
                 report.last_error = Some(format!("{error:?}"));
                 report.retryable_errors += 1;
                 report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
@@ -767,7 +791,7 @@ async fn complete_ready_refresh(
         Ok(_) => {
             report.completed += 1;
         }
-        Err(error) if error.is_storage() => {
+        Err(error) if is_retryable_storage(&error) => {
             report.last_error = Some(format!("{error:?}"));
             report.retryable_errors += 1;
             report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
@@ -796,6 +820,48 @@ fn record_projector_error(
     }
 }
 
+/// Typed failure recorded when the durable contract refuses the projected
+/// progress row. It is not a projector fault: the row was well formed for the
+/// state the projector read, and the durable state disagrees.
+const REFRESH_PROGRESS_REFUSED: &str = "refresh_progress_refused";
+
+/// Builds the durable failure request that retires one running refresh.
+fn durable_failure_request(
+    recovery: &SessionRefreshRecoveryV1,
+    failure_code: String,
+) -> Option<SessionRefreshFailureRequestV1> {
+    let (frontier, coverage) = match recovery.progress() {
+        Some(progress) => (progress.frontier(), *progress.coverage()),
+        None => (
+            SessionRefreshFrontierV1::new(
+                recovery.target_frontier().observed_through(),
+                recovery.source_frontier(),
+            )
+            .ok()?,
+            zero_refresh_coverage(),
+        ),
+    };
+    let request = SessionRefreshFailureRequestV1::new(
+        recovery.operation_id().clone(),
+        recovery.session_id().clone(),
+        frontier,
+        coverage,
+        failure_code,
+    )
+    .ok()?;
+    Some(
+        match recovery
+            .progress()
+            .and_then(SessionRefreshProgressV1::source_coverage)
+            .cloned()
+            .or_else(|| recovery.source_coverage(frontier.committed_through()).ok())
+        {
+            Some(source_coverage) => request.with_source_coverage(source_coverage),
+            None => request,
+        },
+    )
+}
+
 pub async fn apply_refresh_effect(
     store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
     state: &SessionTemporalRefreshWakeState,
@@ -814,43 +880,73 @@ pub async fn apply_refresh_effect(
                 .await
             {
                 Ok(_) => report.projected_batches += 1,
-                Err(error) if error.is_storage() => {
+                Err(error) if is_retryable_storage(&error) => {
                     report.last_error = Some(format!("{error:?}"));
                     report.retryable_errors += 1;
                     report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
                 }
-                Err(error) => {
-                    report.last_error = Some(format!("{error:?}"));
-                    report.terminal_errors += 1;
-                }
-            }
-        }
-        SessionTemporalRefreshEffect::Fail(request) => {
-            if !state.claim_terminal_attempt(recovery) {
-                return;
-            }
-            let mut attempt = TerminalAttemptGuard::new(state, recovery);
-            match store.fail_session_refresh(request).await {
-                Ok(_) => {
-                    report.failed += 1;
-                    state.record_terminal_discovery_failure(recovery);
-                }
-                Err(error) if error.is_storage() => {
+                Err(error) if is_deterministic_refusal(&error) => {
+                    // A refused progress row is not work the next pass can
+                    // finish: rediscovery hands the projector the same durable
+                    // state and the same row comes back refused. Retire the
+                    // operation so it leaves `running` and a fresh refresh can
+                    // be admitted, instead of resubmitting it forever.
                     report.last_error = Some(format!("{error:?}"));
-                    report.retryable_errors += 1;
-                    report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
+                    match durable_failure_request(
+                        recovery,
+                        durable_projector_failure_code(REFRESH_PROGRESS_REFUSED),
+                    ) {
+                        Some(request) => {
+                            apply_fail_effect(store, state, recovery, request, report).await;
+                        }
+                        None => report.terminal_errors += 1,
+                    }
                 }
                 Err(error) => {
-                    attempt.retain();
+                    // Cancelled control, budget ceiling: this pass could not
+                    // persist, but the row itself was not refused, so the
+                    // operation stays `running` for the next pass.
                     report.last_error = Some(format!("{error:?}"));
                     report.terminal_errors += 1;
                 }
             }
         }
+        SessionTemporalRefreshEffect::Fail(request) => {
+            apply_fail_effect(store, state, recovery, request, report).await;
+        }
         SessionTemporalRefreshEffect::Deferred => report.deferred += 1,
     }
 }
 
+async fn apply_fail_effect(
+    store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
+    state: &SessionTemporalRefreshWakeState,
+    recovery: &SessionRefreshRecoveryV1,
+    request: SessionRefreshFailureRequestV1,
+    report: &mut SessionTemporalRefreshPassReport,
+) {
+    if !state.claim_terminal_attempt(recovery) {
+        return;
+    }
+    let mut attempt = TerminalAttemptGuard::new(state, recovery);
+    match store.fail_session_refresh(request).await {
+        Ok(_) => {
+            report.failed += 1;
+            state.record_terminal_discovery_failure(recovery);
+        }
+        Err(error) if is_retryable_storage(&error) => {
+            report.last_error = Some(format!("{error:?}"));
+            report.retryable_errors += 1;
+            report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
+        }
+        Err(error) => {
+            attempt.retain();
+            report.last_error = Some(format!("{error:?}"));
+            report.terminal_errors += 1;
+        }
+    }
+}
+
 async fn project_running_refresh(
     database: &RegisteredGlobalDbLeaseV1,
     store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
@@ -894,35 +990,7 @@ async fn project_running_refresh(
         Err(error) => {
             let failure_code = durable_projector_failure_code(&error.code);
             report.last_error = Some(failure_code.clone());
-            let (frontier, coverage) = if let Some(progress) = recovery.progress() {
-                (progress.frontier(), *progress.coverage())
-            } else {
-                let Ok(frontier) = SessionRefreshFrontierV1::new(
-                    recovery.target_frontier().observed_through(),
-                    recovery.source_frontier(),
-                ) else {
-                    report.terminal_errors += 1;
-                    return;
-                };
-                (frontier, zero_refresh_coverage())
-            };
-            let request = if let Ok(request) = SessionRefreshFailureRequestV1::new(
-                recovery.operation_id().clone(),
-                recovery.session_id().clone(),
-                frontier,
-                coverage,
-                failure_code,
-            ) {
-                match recovery
-                    .progress()
-                    .and_then(SessionRefreshProgressV1::source_coverage)
-                    .cloned()
-                    .or_else(|| recovery.source_coverage(frontier.committed_through()).ok())
-                {
-                    Some(source_coverage) => request.with_source_coverage(source_coverage),
-                    None => request,
-                }
-            } else {
+            let Some(request) = durable_failure_request(recovery, failure_code) else {
                 report.terminal_errors += 1;
                 return;
             };
@@ -960,7 +1028,7 @@ async fn running_refreshes(
         Ok(recoveries) => Some(recoveries),
         Err(error) => {
             report.last_error = Some(format!("{error:?}"));
-            if classify_store_error(&error) == SessionTemporalRefreshRetryClass::Storage {
+            if is_retryable_storage(&error) {
                 report.retryable_errors += 1;
                 report.observe_retry(SessionTemporalRefreshRetryClass::Storage);
             } else {
@@ -1121,6 +1189,36 @@ mod tests {
     use tracedecay_sessions::runtime::{SessionMessageRecord, SessionRecord};
     use tracedecay_store::ParseOffset;
 
+    #[test]
+    fn deterministic_storage_refusals_are_not_retryable() {
+        // The schema-contract trigger that refused eleven hours of identical
+        // progress rows in #1794: transport-level `Storage`, but replaying it
+        // can never succeed.
+        let refused = SessionStoreError::storage(
+            "persist session refresh progress",
+            EngineError::Sqlite {
+                operation: "execute",
+                code: Some(19),
+                extended_code: Some(1811),
+                message: "invalid session refresh progress".to_owned(),
+            },
+        );
+        assert!(!is_retryable_storage(&refused));
+
+        // Contention is the transient case the retry loop exists for.
+        assert!(is_retryable_storage(&SessionStoreError::storage(
+            "persist session refresh progress",
+            EngineError::Busy,
+        )));
+
+        // Typed contract failures were already terminal and stay terminal.
+        assert!(!is_retryable_storage(
+            &SessionStoreError::InvalidStateTransition {
+                context: "refresh progress successor",
+            }
+        ));
+    }
+
     #[test]
     fn dropping_worker_instrumentation_clears_pending_state_once() {
         let state = SessionTemporalRefreshWakeState::default();
diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
index cd3c8ed6c0..f119795e41 100644
--- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
+++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs
@@ -1,16 +1,21 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use tracedecay_domain::SessionId;
+use tracedecay_domain::{SessionId, UtcMicros};
 use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness;
 use tracedecay_session_temporal_store::SessionTemporalStore;
 use tracedecay_store::{
-    SessionRefreshBeginOrJoinRequestV1, SessionRefreshFrontierV1, SessionRefreshStore,
+    SessionRefreshBeginOrJoinRequestV1, SessionRefreshFrontierV1, SessionRefreshProgressV1,
+    SessionRefreshStore, SessionTemporalProjectionBatchV1,
 };
 
-use super::projector::{CanonicalSessionTemporalProjector, SessionTemporalRefreshPolicy};
+use super::projector::{
+    CanonicalSessionTemporalProjector, SessionTemporalRefreshEffect, SessionTemporalRefreshPolicy,
+    zero_refresh_coverage,
+};
+use super::registry::SessionTemporalRefreshPassReport;
 use super::wake::{SessionTemporalRefreshRetryClass, SessionTemporalRefreshWakeState};
-use super::worker::run_session_temporal_refresh_pass;
+use super::worker::{apply_refresh_effect, run_session_temporal_refresh_pass};
 
 async fn begin_empty_refreshes(
     store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>,
@@ -80,3 +85,66 @@ async fn retryable_recovery_stops_the_pass_and_preserves_unattempted_work() {
     assert_eq!(report.backlog, Some(2));
     assert_eq!(state.pending_recovery_operations().len(), 1);
 }
+
+#[tokio::test]
+async fn refused_projection_progress_retires_the_refresh_instead_of_retrying() {
+    let harness = RegisteredGlobalDbHarness::open("refresh-refused-progress-retires").await;
+    let store = SessionTemporalStore::new(harness.registered.as_ref());
+    begin_empty_refreshes(&store, ["refused-progress"]).await;
+    let recovery = store
+        .running_session_refreshes()
+        .await
+        .expect("recoveries")
+        .pop()
+        .expect("one running recovery");
+    let state = SessionTemporalRefreshWakeState::default();
+    let mut report = SessionTemporalRefreshPassReport::default();
+
+    // Progress that claims a second committed batch while submitting the
+    // first one. The durable contract refuses it, and every later pass would
+    // hand the projector the same state and rebuild the same refused row.
+    let progress = SessionRefreshProgressV1::new(
+        recovery.operation_id().clone(),
+        recovery.session_id().clone(),
+        SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"),
+        zero_refresh_coverage(),
+        2,
+        0,
+        UtcMicros(1),
+    );
+    let batch = SessionTemporalProjectionBatchV1::new(
+        recovery.session_id().clone(),
+        recovery.candidate_generation(),
+        recovery.frozen_watermarks().clone(),
+        vec![],
+        vec![],
+        vec![],
+    )
+    .expect("batch")
+    .with_checkpoint(0, 0, 0)
+    .expect("checkpoint");
+
+    apply_refresh_effect(
+        &store,
+        &state,
+        &recovery,
+        SessionTemporalRefreshEffect::Projection { progress, batch },
+        &mut report,
+    )
+    .await;
+
+    assert_eq!(
+        report.failed, 1,
+        "a refused progress row must retire the refresh, not stay running"
+    );
+    assert_eq!(report.retryable_errors, 0);
+    assert_eq!(report.terminal_errors, 0);
+    assert!(
+        store
+            .running_session_refreshes()
+            .await
+            .expect("recoveries")
+            .is_empty(),
+        "the retired refresh must not be rediscovered"
+    );
+}
diff --git a/crates/tracedecay-store/src/session/refresh.rs b/crates/tracedecay-store/src/session/refresh.rs
index 33b52a7e8d..c6c8696dc9 100644
--- a/crates/tracedecay-store/src/session/refresh.rs
+++ b/crates/tracedecay-store/src/session/refresh.rs
@@ -282,8 +282,13 @@ impl SessionRefreshProgressV1 {
         }
         let current = self.coverage;
         let candidate = next.coverage;
+        // The durable guard admits a successor only when it strictly advances
+        // the committed frontier. Accepting an equal frontier here let a
+        // producer submit a row the trigger then refused as a SQLite
+        // constraint abort, which the worker read as transient storage and
+        // resubmitted forever. Refuse it as typed state instead.
         if self.frontier.observed_through != next.frontier.observed_through
-            || next.frontier.committed_through < self.frontier.committed_through
+            || next.frontier.committed_through <= self.frontier.committed_through
             || next.committed_batches < self.committed_batches
             || next.committed_records < self.committed_records
             || candidate.visible < current.visible
diff --git a/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs b/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs
index f4ffc7aeaa..0de934cb97 100644
--- a/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs
+++ b/crates/tracedecay-store/tests/store_suite/session_contract/refresh.rs
@@ -225,6 +225,25 @@ fn refresh_frontiers_and_progress_are_monotonic_and_terminal() {
         })
     ));
 
+    // The durable guard admits a successor only when it strictly advances the
+    // committed frontier, so a stalled successor must be refused here rather
+    // than deferred to a SQLite constraint abort the caller reads as storage.
+    let stalled = SessionRefreshProgressV1::new(
+        operation_id(),
+        session_id.clone(),
+        SessionRefreshFrontierV1::new(10, 8).unwrap(),
+        coverage(),
+        2,
+        8,
+        UtcMicros(101),
+    );
+    assert!(matches!(
+        initial.validate_successor(&stalled),
+        Err(SessionStoreError::InvalidStateTransition {
+            context: "refresh progress successor"
+        })
+    ));
+
     let terminal = SessionRefreshReceiptV1::completed(
         SessionRefreshCompletionRequestV1::new(
             operation_id(),
diff --git a/crates/tracedecay/src/daemon/branch_add.rs b/crates/tracedecay/src/daemon/branch_add.rs
index 4e2d3793f7..0b7b5f17e2 100644
--- a/crates/tracedecay/src/daemon/branch_add.rs
+++ b/crates/tracedecay/src/daemon/branch_add.rs
@@ -1,4 +1,4 @@
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use tracedecay_application::pr_tracking::{
@@ -153,6 +153,9 @@ async fn activate_and_track_manual_branch(
     let graph = Arc::clone(graph);
     let schedulers = schedulers.clone();
     let branch = branch.to_owned();
+    let published_data_root = data_root.clone();
+    let published_schedulers = schedulers.clone();
+    let published_registries = administration.session_runtime_registries();
 
     administration
         .admit_manual_branch_publication(|cancellation, admitted| async move {
@@ -215,6 +218,15 @@ async fn activate_and_track_manual_branch(
                 tracked
             }
             .await;
+            if matches!(&result, Ok(outcome) if *outcome != BranchAddOutcome::Deferred) {
+                mount_published_branch_query_authority(
+                    published_registries.as_ref(),
+                    &published_schedulers,
+                    &published_data_root,
+                    &branch,
+                )
+                .await;
+            }
             match &result {
                 Ok(outcome) => log_daemon_event(
                     "manual_branch_publication",
@@ -237,6 +249,101 @@ async fn activate_and_track_manual_branch(
         .await
 }
 
+/// Mounts the checked-in core query authority on the branch worktree this
+/// publication sealed, from the project's own durable cursor-key authority.
+///
+/// An explicitly published branch worktree is never a project-open route, so
+/// nothing else mounts its query authority: an exact branch read could only
+/// borrow one already mounted on a peer checkout of the same repository
+/// (`mount_query_authority_from_project_peer`), and that peer's own mount is
+/// deferred until it seats a text generation. A read taken right after this
+/// publication sealed its provenance therefore failed closed with a
+/// non-retryable `authority_unavailable` even though the branch generation was
+/// published and servable. Mounting here makes the generation this journey
+/// publishes queryable without depending on an unrelated worktree's
+/// activation order.
+///
+/// Runs inside the daemon-owned publication task, after the generation it
+/// serves is committed: the admitting `branch add` caller returns at admission
+/// and never waits for this, and the profile's session-registry lock is taken
+/// here rather than on that caller's path.
+///
+/// Best effort by design: the branch generation is already committed, so a
+/// missing session mount or cursor key must not retract it. The exact branch
+/// read falls back to borrowing a peer authority when this could not run.
+#[cfg(unix)]
+#[hotpath::measure(label = "daemon.branch_add.query_authority", future = true)]
+async fn mount_published_branch_query_authority(
+    registries: Option<&(super::branch_admin::SharedSessionRuntimeRegistries, PathBuf)>,
+    schedulers: &CodeIndexSchedulerRegistryV1,
+    data_root: &Path,
+    branch: &str,
+) {
+    let Some((registries, profile_root)) = registries else {
+        return;
+    };
+    let Some(source) =
+        tracedecay_runtime_core::branch_meta::load_branch_meta(data_root).and_then(|meta| {
+            meta.branches
+                .get(branch)
+                .and_then(|entry| entry.graph_source.clone())
+        })
+    else {
+        return;
+    };
+    let worktree_root = PathBuf::from(&source.worktree_root);
+    let Ok(project_id) = tracedecay_domain::ProjectId::new(source.project_id.clone()) else {
+        return;
+    };
+    let Ok(scope) =
+        tracedecay_code_index_runtime::resolved_scope_for_project(&worktree_root, &project_id)
+    else {
+        return;
+    };
+    let sessions = {
+        let registries = registries.lock().await;
+        registries
+            .get(profile_root)
+            .map(|entry| Arc::clone(&entry.registry))
+    };
+    let Some(sessions) = sessions.and_then(|registry| registry.get().cloned()) else {
+        return;
+    };
+    let Some(session_db) = sessions.mounted_project_sessions(&project_id).await else {
+        return;
+    };
+    let cursor_keys = match session_db.load_session_cursor_key_provider_result().await {
+        Ok(cursor_keys) => cursor_keys,
+        Err(error) => {
+            tracing::debug!(
+                event = "branch_query_authority_mount",
+                outcome = "unavailable",
+                branch = %branch,
+                reason = %error,
+                "durable query cursor key is unavailable for the published branch"
+            );
+            return;
+        }
+    };
+    if let Err(error) =
+        tracedecay_code_index_runtime::code_index_scheduler::query_runtime::mount_core_query_authority_on_project_open(
+            schedulers,
+            &worktree_root,
+            &scope,
+            &cursor_keys,
+        )
+        .await
+    {
+        tracing::debug!(
+            event = "branch_query_authority_mount",
+            outcome = "unavailable",
+            branch = %branch,
+            reason = %error,
+            "published branch query authority is unavailable; exact reads fall back to a peer"
+        );
+    }
+}
+
 #[cfg(unix)]
 #[hotpath::measure(label = "daemon.branch_add.owner", future = true)]
 pub(super) async fn activate_and_track_manual_branch_owned(
diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs
index f65a2c3924..899dc6fcfc 100644
--- a/crates/tracedecay/src/daemon/branch_admin.rs
+++ b/crates/tracedecay/src/daemon/branch_admin.rs
@@ -944,6 +944,24 @@ impl StoreAdministration {
         registry.mounted_session_databases().await
     }
 
+    /// The profile's session-runtime registry map and its canonical root,
+    /// taken without locking either.
+    ///
+    /// Branch publication resolves the project's durable cursor-key authority
+    /// through this inside its own background task, so an explicitly published
+    /// branch can mount its own query authority without the admitting caller
+    /// paying for a registry lock it never reads.
+    #[cfg(unix)]
+    pub(super) fn session_runtime_registries(
+        &self,
+    ) -> Option<(SharedSessionRuntimeRegistries, std::path::PathBuf)> {
+        let profile_root = self
+            .profile_identity()
+            .and_then(|identity| authority::canonical_identity_path(identity.profile_root()))
+            .ok()?;
+        Some((Arc::clone(&self.session_runtime_registries), profile_root))
+    }
+
     #[hotpath::measure(label = "daemon.branch_admin.mounted_project_servers", future = true)]
     pub(super) async fn mounted_project_servers(&self) -> Vec<Arc<crate::mcp::McpServer>> {
         let Ok(profile_root) = self
diff --git a/crates/tracedecay/src/daemon/core_proxy.rs b/crates/tracedecay/src/daemon/core_proxy.rs
index d9b3a4a712..025ef26848 100644
--- a/crates/tracedecay/src/daemon/core_proxy.rs
+++ b/crates/tracedecay/src/daemon/core_proxy.rs
@@ -909,7 +909,7 @@ fn daemon_version_skew_warning_for_request(
     client_version: &str,
 ) -> Option<String> {
     let daemon_version = proxy_initialize_metadata_for_request(request, responses).daemon_version?;
-    if daemon_version == client_version {
+    if tracedecay_daemon_protocol::versions_name_same_build(&daemon_version, client_version) {
         return None;
     }
     let action = version_skew_action(&daemon_version, client_version);
diff --git a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs b/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs
index b0d224dfec..6415ef3209 100644
--- a/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs
+++ b/crates/tracedecay/src/daemon/production_harness/generation_retention_test.rs
@@ -11,7 +11,8 @@ use super::journey_test_support::git;
 use super::*;
 use crate::daemon::maintenance::project_store_maintenance_lease;
 use tracedecay_code_index_retention::code_index_generations::{
-    MAX_CODE_GENERATION_RETENTION_BATCH_V1, prepare_next_code_generation_retention_cancellable,
+    CodeGenerationRetentionErrorV1, MAX_CODE_GENERATION_RETENTION_BATCH_V1,
+    prepare_next_code_generation_retention_cancellable,
 };
 use tracedecay_maintenance::tick::{MaintenanceContinuation, MaintenanceTickOutcome};
 
@@ -114,13 +115,29 @@ async fn mounted_code_generation_retention_continues_capped_segment_reclamation(
             &canonical_root,
         );
     let graph_replay_pool_root = graph.db().database_path().with_extension("graph-replay");
-    let plan = prepare_next_code_generation_retention_cancellable(
-        &code_store_root,
-        &BTreeSet::new(),
-        &|| false,
-        Some(&graph_replay_pool_root),
-    )
-    .expect("code generation retention plan");
+    // The planner probes the generation-store lock and answers
+    // `GenerationStoreBusy` whenever a writer owns the store; production
+    // maintenance defers that tick and comes back. This route stays mounted,
+    // so the pass tail that publishes the edits above can still own the store
+    // here. Consume the same typed answer instead of reading it as a failure.
+    let plan = tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            match prepare_next_code_generation_retention_cancellable(
+                &code_store_root,
+                &BTreeSet::new(),
+                &|| false,
+                Some(&graph_replay_pool_root),
+            ) {
+                Ok(plan) => return plan,
+                Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy) => {
+                    tokio::time::sleep(Duration::from_millis(25)).await;
+                }
+                Err(error) => panic!("code generation retention plan: {error:?}"),
+            }
+        }
+    })
+    .await
+    .expect("code generation retention plan converges");
     let first_candidate = plan
         .collectable_generations
         .iter()
diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs
index 28bbc596c4..20b9e5dca0 100644
--- a/crates/tracedecay/tests/common/mod.rs
+++ b/crates/tracedecay/tests/common/mod.rs
@@ -17,6 +17,8 @@ use std::os::unix::fs::PermissionsExt;
 use std::os::unix::process::CommandExt;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, ExitStatus, Output, Stdio};
+#[cfg(unix)]
+use std::sync::{Mutex, PoisonError};
 use std::time::{Duration, Instant};
 
 use serde_json::Value;
@@ -657,6 +659,14 @@ pub fn http_agent_with_timeout(timeout: Duration) -> ureq::Agent {
 /// panic while the child is still running, `Drop` force-stops and reaps it.
 pub struct TestChildProcess {
     child: Child,
+    /// Whether this child has been waited on. A reaped pid belongs to the
+    /// kernel again, so it must never be used to address a process group.
+    reaped: bool,
+    /// Path of a Unix socket this child published. Released after the process
+    /// group is reaped so a descendant that still holds the listen descriptor
+    /// cannot keep the path accepting.
+    #[cfg(unix)]
+    release_socket: Option<PathBuf>,
 }
 
 /// Daemon-specific name retained for test fixtures that keep a daemon alive.
@@ -664,7 +674,51 @@ pub type DaemonProcess = TestChildProcess;
 
 impl TestChildProcess {
     pub fn new(child: Child) -> Self {
-        Self { child }
+        Self {
+            child,
+            reaped: false,
+            #[cfg(unix)]
+            release_socket: None,
+        }
+    }
+
+    /// Unlink the socket this child published once it has been reaped.
+    ///
+    /// `process_group(0)` makes the child a group leader. Stopping only that
+    /// pid leaves descendants that still hold the listen socket. Group-kill
+    /// closes those descriptors; unlinking the path is what makes a later
+    /// `connect` fail even if the kernel has not finished the last close.
+    ///
+    /// Recording claims the path: a restart journey reassigns its handle
+    /// (`daemon = spawn(..)`), so the successor is already publishing when
+    /// the predecessor is dropped, and only the current publisher may unlink.
+    /// File identity is not enough for that - the successor's socket routinely
+    /// lands on the inode the predecessor's shutdown just freed.
+    #[cfg(unix)]
+    pub fn release_socket_on_stop(&mut self, path: PathBuf) {
+        claim_published_socket(&path, self.child.id());
+        self.release_socket = Some(path);
+    }
+
+    #[cfg(unix)]
+    fn release_recorded_socket(&mut self) {
+        let Some(path) = self.release_socket.take() else {
+            return;
+        };
+        if !release_published_socket_claim(&path, self.child.id()) {
+            // A successor publishes here now; its socket is not ours to unlink.
+            return;
+        }
+        match std::fs::remove_file(&path) {
+            Ok(()) => {}
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+            Err(error) => {
+                panic!(
+                    "failed to release daemon socket '{}': {error}",
+                    path.display()
+                )
+            }
+        }
     }
 
     pub fn id(&self) -> u32 {
@@ -676,7 +730,9 @@ impl TestChildProcess {
     }
 
     pub fn try_wait(&mut self) -> std::io::Result<Option<ExitStatus>> {
-        self.child.try_wait()
+        let status = self.child.try_wait()?;
+        self.reaped |= status.is_some();
+        Ok(status)
     }
 
     pub fn wait_for_exit(&mut self, timeout: Duration) -> std::io::Result<Option<ExitStatus>> {
@@ -752,9 +808,15 @@ impl TestChildProcess {
     /// Force-stops the daemon and reaps its process before returning.
     ///
     /// `Child::kill` maps to `SIGKILL` on Unix and the platform termination
-    /// primitive elsewhere, keeping fault-injection tests portable.
+    /// primitive elsewhere, keeping fault-injection tests portable. On Unix
+    /// the child's process group is signaled first, then the published socket
+    /// path is unlinked.
     pub fn kill_and_wait(&mut self) -> std::io::Result<ExitStatus> {
-        terminate_and_reap(&mut self.child)
+        let status = terminate_and_reap(&mut self.child, !self.reaped);
+        self.reaped = true;
+        #[cfg(unix)]
+        self.release_recorded_socket();
+        status
     }
 
     fn drain_stderr(&mut self) {
@@ -778,12 +840,35 @@ impl TestChildProcess {
 
 impl Drop for TestChildProcess {
     fn drop(&mut self) {
-        let _ = terminate_and_reap(&mut self.child);
+        let _ = terminate_and_reap(&mut self.child, !self.reaped);
+        self.reaped = true;
+        #[cfg(unix)]
+        self.release_recorded_socket();
     }
 }
 
-/// PID-directed stop: survives `process_group(0)` / `setsid` detachment.
-fn terminate_and_reap(child: &mut Child) -> std::io::Result<ExitStatus> {
+/// Stop a child that was detached with `process_group(0)`.
+///
+/// The child is the leader of its own group. `SIGKILL` of that pid alone
+/// leaves descendants in the group. Those descendants keep any descriptor they
+/// inherited, including a listen socket, so the path stays connectable after
+/// `wait` returns. Signaling the group first closes those descriptors; the
+/// leader kill still covers a child whose `setpgid` has not run yet.
+///
+/// `signal_group` must be false once this child has been waited on: a reaped
+/// pid is the kernel's to reissue, so negating it could address a process
+/// group this harness never created.
+fn terminate_and_reap(child: &mut Child, signal_group: bool) -> std::io::Result<ExitStatus> {
+    // Signal the group before reaping. A leader that has already exited still
+    // names the group while it is an unreaped zombie; returning on `try_wait`
+    // first would leave descendants holding the listen socket.
+    #[cfg(unix)]
+    if signal_group {
+        signal_child_process_group(child.id());
+    }
+    #[cfg(not(unix))]
+    let _ = signal_group;
+
     if let Ok(Some(status)) = child.try_wait() {
         return Ok(status);
     }
@@ -798,6 +883,50 @@ fn terminate_and_reap(child: &mut Child) -> std::io::Result<ExitStatus> {
     child.wait()
 }
 
+/// The child pid currently publishing each recorded socket path.
+#[cfg(unix)]
+static PUBLISHED_SOCKETS: Mutex<Vec<(PathBuf, u32)>> = Mutex::new(Vec::new());
+
+#[cfg(unix)]
+fn claim_published_socket(path: &Path, pid: u32) {
+    let mut claims = PUBLISHED_SOCKETS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner);
+    claims.retain(|(claimed, _)| claimed != path);
+    claims.push((path.to_path_buf(), pid));
+}
+
+/// True when `pid` is still the publisher of `path`, dropping the claim.
+#[cfg(unix)]
+fn release_published_socket_claim(path: &Path, pid: u32) -> bool {
+    let mut claims = PUBLISHED_SOCKETS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner);
+    let Some(index) = claims
+        .iter()
+        .position(|(claimed, owner)| claimed == path && *owner == pid)
+    else {
+        return false;
+    };
+    claims.swap_remove(index);
+    true
+}
+
+#[cfg(unix)]
+fn signal_child_process_group(pid: u32) {
+    let Ok(pid) = i32::try_from(pid) else {
+        return;
+    };
+    if pid == 0 {
+        return;
+    }
+    // SAFETY: `pid` is the spawned child's id. Negating it addresses the
+    // process group `process_group(0)` created with that pid as leader.
+    // `ESRCH` is ignored: the child may not be a group leader, and the pid
+    // kill in `terminate_and_reap` still stops it.
+    let _ = unsafe { libc::kill(-pid, libc::SIGKILL) };
+}
+
 /// Detach a test child from the test process group.
 ///
 /// Nextest (and other harness timeouts) signal the test's process group.
@@ -1089,13 +1218,6 @@ pub fn spawn_tracedecay_daemon_with(
     spawn_tracedecay_daemon_process(&home, &binary, configure)
 }
 
-/// How long a replacement daemon waits for a stopped predecessor's endpoint to
-/// stop accepting before reporting it as still live.
-///
-/// Generous on purpose: the wait only costs time when a predecessor is
-/// genuinely still reachable, and a real leak still fails rather than hangs.
-const PREDECESSOR_DAEMON_VACATE_TIMEOUT: Duration = Duration::from_secs(10);
-
 fn spawn_tracedecay_daemon_process(
     home: &Path,
     binary: &Path,
@@ -1118,42 +1240,20 @@ fn spawn_tracedecay_daemon_process(
             })
             .is_some_and(|address| TcpStream::connect(address).is_ok())
     };
-    // Stopping a predecessor daemon is asynchronous with respect to its
-    // endpoint: `kill` plus `wait` reaps the PID the harness spawned, but the
-    // kernel keeps the listening socket alive while *any* duplicate of that
-    // descriptor survives, including one a subprocess inherited across `fork`
-    // and still holds because it has not reached its own `exec` yet. Asserting
-    // instantaneously therefore reports an ordinary teardown tail as a live
-    // daemon, which is what `init_project_fixture` journeys (spawn, init, drop,
-    // spawn again) hit on a loaded runner. Wait a bounded time for the endpoint
-    // to stop accepting; a daemon that keeps accepting still fails with the
-    // same refusal.
-    poll_until(
-        Instant::now() + PREDECESSOR_DAEMON_VACATE_TIMEOUT,
-        Duration::from_millis(25),
-        || {
-            #[cfg(unix)]
-            let live = std::os::unix::net::UnixStream::connect(&socket_path).is_ok();
-            #[cfg(not(unix))]
-            let live = portable_daemon_connectable();
-            (!live).then_some(())
-        },
-        || {
-            #[cfg(unix)]
-            {
-                format!(
-                    "refusing to replace a live test daemon at {}",
-                    socket_path.display()
-                )
-            }
-            #[cfg(not(unix))]
-            {
-                format!(
-                    "refusing to replace a live test daemon recorded at {}",
-                    authority_path.display()
-                )
-            }
-        },
+    // A predecessor stopped through this harness has already had its process
+    // group reaped and its socket path unlinked. A path that still accepts is
+    // a daemon this spawn does not own.
+    #[cfg(unix)]
+    assert!(
+        std::os::unix::net::UnixStream::connect(&socket_path).is_err(),
+        "refusing to replace a live test daemon at {}",
+        socket_path.display()
+    );
+    #[cfg(not(unix))]
+    assert!(
+        !portable_daemon_connectable(),
+        "refusing to replace a live test daemon recorded at {}",
+        authority_path.display()
     );
 
     let mut command = Command::new(binary);
@@ -1169,6 +1269,8 @@ fn spawn_tracedecay_daemon_process(
     detach_from_test_process_group(&mut command);
     let child = command.spawn().expect("tracedecay daemon should start");
     let mut daemon = DaemonProcess::new(child);
+    #[cfg(unix)]
+    daemon.release_socket_on_stop(socket_path.clone());
 
     let deadline = Instant::now() + Duration::from_secs(10);
     poll_until(
diff --git a/crates/tracedecay/tests/daemon_suite/advanced_workflow_journey/daemon_fixture.rs b/crates/tracedecay/tests/daemon_suite/advanced_workflow_journey/daemon_fixture.rs
index d0bbc79f28..069897ebb7 100644
--- a/crates/tracedecay/tests/daemon_suite/advanced_workflow_journey/daemon_fixture.rs
+++ b/crates/tracedecay/tests/daemon_suite/advanced_workflow_journey/daemon_fixture.rs
@@ -105,6 +105,8 @@ pub(super) fn spawn_project_daemon(home: &Path, project: &Path) -> common::Daemo
         .spawn()
         .expect("advanced workflow daemon should start");
     let mut daemon = common::DaemonProcess::new(child);
+    #[cfg(unix)]
+    daemon.release_socket_on_stop(common::daemon_socket_path(home));
     let daemon_pid = u64::from(daemon.id());
     let deadline = Instant::now() + Duration::from_secs(120);
     loop {
diff --git a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
index bc1388df7f..e679f5855a 100644
--- a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
+++ b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
@@ -331,6 +331,11 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() {
         .expect_err("follower publication fails closed");
     assert_publication_error(owner_error);
     assert_publication_error(follower_error);
+    assert_eq!(
+        std::fs::read(&pointer_path).expect("faulted pointer remains"),
+        b"{",
+        "publication must not replace a pointer it did not observe"
+    );
 
     std::fs::write(pointer_path, pointer_bytes).expect("restore active pointer");
     registry.shutdown().await;
diff --git a/crates/tracedecay/tests/daemon_suite/main.rs b/crates/tracedecay/tests/daemon_suite/main.rs
index 6339fab5c6..8e86e9cd8d 100644
--- a/crates/tracedecay/tests/daemon_suite/main.rs
+++ b/crates/tracedecay/tests/daemon_suite/main.rs
@@ -31,6 +31,7 @@ mod indexing_lifecycle_test;
 mod invocation_observability;
 mod invocation_primitives;
 #[cfg(unix)]
+mod socket_lifecycle_test;
 #[cfg(unix)]
 mod stale_client_resilience_test;
 mod workflow_handoff_test;
diff --git a/crates/tracedecay/tests/daemon_suite/socket_lifecycle_test.rs b/crates/tracedecay/tests/daemon_suite/socket_lifecycle_test.rs
new file mode 100644
index 0000000000..bd2f0f1dbc
--- /dev/null
+++ b/crates/tracedecay/tests/daemon_suite/socket_lifecycle_test.rs
@@ -0,0 +1,104 @@
+//! Process-group stop must release a listen socket held by a descendant.
+//!
+//! `process_group(0)` makes the spawned child its own group leader. Killing
+//! only that pid leaves the descendant that inherited the listen descriptor,
+//! and the path stays connectable after `wait` returns.
+
+use std::os::unix::net::UnixStream;
+use std::os::unix::process::CommandExt;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+use crate::common::{TestChildProcess, poll_until};
+
+/// How long the released descendant has to finish dying.
+///
+/// The group signal is delivered to a process the harness cannot `wait` on -
+/// the descendant is reparented, not a child - so its descriptors close when
+/// the kernel finishes tearing it down, not when the leader's `wait` returns.
+/// A descendant that was never signaled keeps accepting past this deadline,
+/// which is the regression this proof exists to catch.
+const DESCENDANT_RELEASE_TIMEOUT: Duration = Duration::from_secs(10);
+
+const HOLDER: &str = r#"
+import os, socket, time
+path = os.environ["TRACEDECAY_TEST_SOCKET_PATH"]
+listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+listener.bind(path)
+listener.listen(1)
+os.fork()
+while True:
+    time.sleep(60)
+"#;
+
+#[test]
+fn group_stop_releases_an_inherited_listen_socket() {
+    let scratch = tempfile::tempdir().expect("socket scratch");
+    let socket = scratch.path().join("daemon.sock");
+    let mut command = Command::new("python3");
+    command
+        .arg("-c")
+        .arg(HOLDER)
+        .env("TRACEDECAY_TEST_SOCKET_PATH", &socket)
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
+    command.process_group(0);
+    let child = command.spawn().expect("spawn socket holder");
+    // Do not record a socket path: connect must fail because the group is
+    // dead, not because the path was unlinked.
+    let mut holder = TestChildProcess::new(child);
+
+    let ready_deadline = Instant::now() + Duration::from_secs(5);
+    while UnixStream::connect(&socket).is_err() {
+        assert!(
+            Instant::now() < ready_deadline,
+            "holder did not bind {}",
+            socket.display()
+        );
+        if holder.try_wait().expect("holder status").is_some() {
+            panic!("socket holder exited before binding");
+        }
+        std::thread::sleep(Duration::from_millis(20));
+    }
+
+    holder.kill_and_wait().expect("reap socket holder group");
+    assert!(
+        socket.exists(),
+        "this proof must not delete the socket path"
+    );
+    poll_until(
+        Instant::now() + DESCENDANT_RELEASE_TIMEOUT,
+        Duration::from_millis(20),
+        || UnixStream::connect(&socket).is_err().then_some(()),
+        || {
+            format!(
+                "process-group stop must release the inherited listen socket at {}",
+                socket.display()
+            )
+        },
+    );
+}
+
+#[test]
+fn stop_unlinks_the_socket_path_the_child_published() {
+    let scratch = tempfile::tempdir().expect("socket scratch");
+    let socket = scratch.path().join("daemon.sock");
+    std::os::unix::net::UnixListener::bind(&socket).expect("bind socket");
+    let mut command = Command::new("python3");
+    command
+        .arg("-c")
+        .arg("import time; time.sleep(60)")
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
+    command.process_group(0);
+    let child = command.spawn().expect("spawn sleeper");
+    let mut sleeper = TestChildProcess::new(child);
+    sleeper.release_socket_on_stop(socket.clone());
+    drop(sleeper);
+    assert!(
+        !socket.exists(),
+        "stopping the child must unlink the socket path it published"
+    );
+}
diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
index 2d409392bd..28f292af5f 100644
--- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
+++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
@@ -82,6 +82,51 @@ fn write_production_codex_rollouts(home: &Path, project: &Path, count: usize) {
 async fn production_codex_message_search(
     harness: &ProductionProjectCompositionHarnessV1,
     project: &Path,
+) -> Value {
+    // A `partial` generation is the store saying "still converging", the same
+    // not-ready contract as `stale`: re-read it. Every other outcome answers
+    // now, so an empty `complete_zero` still fails the assertions below.
+    let payload = tokio::time::timeout(std::time::Duration::from_secs(30), async {
+        loop {
+            let payload = production_codex_message_search_once(harness, project).await;
+            if payload["outcome"] != "partial"
+                || payload["results"]
+                    .as_array()
+                    .is_some_and(|results| !results.is_empty())
+            {
+                break payload;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("production Codex message search convergence deadline");
+    assert!(
+        payload["results"].as_array().is_some_and(|results| {
+            results.iter().any(|result| {
+                result["message"]["text"]
+                    .as_str()
+                    .is_some_and(|text| text.contains("cobalt orchard scheduler migration"))
+            })
+        }),
+        "production Codex message search was empty after completed ingest: {payload}"
+    );
+    assert!(
+        payload["results"].as_array().is_some_and(|results| {
+            results.iter().any(|result| {
+                result["message"]["text"].as_str()
+                    == Some("The cobalt orchard scheduler migration is ready for review")
+            })
+        }),
+        "production Codex message search did not hydrate the exact assistant message: {payload}"
+    );
+    payload
+}
+
+#[cfg(feature = "test-transport")]
+async fn production_codex_message_search_once(
+    harness: &ProductionProjectCompositionHarnessV1,
+    project: &Path,
 ) -> Value {
     let response = harness
         .call_tool(
@@ -108,30 +153,10 @@ async fn production_codex_message_search(
     .expect("production message search JSON");
     // Retained tools respond with the full evidence envelope; the search
     // payload the assertions consume lives under `outcome.value.payload`.
-    let payload = envelope
+    envelope
         .pointer("/outcome/value/payload")
         .cloned()
-        .unwrap_or(envelope);
-    assert!(
-        payload["results"].as_array().is_some_and(|results| {
-            results.iter().any(|result| {
-                result["message"]["text"]
-                    .as_str()
-                    .is_some_and(|text| text.contains("cobalt orchard scheduler migration"))
-            })
-        }),
-        "production Codex message search was empty after completed ingest: {payload}"
-    );
-    assert!(
-        payload["results"].as_array().is_some_and(|results| {
-            results.iter().any(|result| {
-                result["message"]["text"].as_str()
-                    == Some("The cobalt orchard scheduler migration is ready for review")
-            })
-        }),
-        "production Codex message search did not hydrate the exact assistant message: {payload}"
-    );
-    payload
+        .unwrap_or(envelope)
 }
 
 #[cfg(feature = "test-transport")]