merge: integrate release recovery and evidence updates

.config/scanner-heal-required-tests.json

@@ -43,6 +43,8 @@
       "min_objects": 5,
       "max_objects": 5,
       "topology": {"nodes": 3, "drives_per_node": 4},
+      "erasure": {"data_blocks": 8, "parity_blocks": 4},
+      "erasure_set_drive_count": 12,
       "scope": "3-node x 4-drive single-set EC8+4, graceful target restart, preformatted replacement drive, exact unversioned S3 bodies and physical target shards; not mixed-version, multi-pool or long-window ABBA."
     },
     "background-target-restart-ec8-4": {

crates/ecstore/src/bucket/replication/replication_storage_boundary.rs

@@ -22,7 +22,7 @@ pub(crate) use crate::object_api::{
     GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
 };
 #[cfg(test)]
-pub(crate) use crate::object_api::{NamespaceLockFence, NamespaceLockSignalTestFence, ReadPlan};
+pub(crate) use crate::object_api::{NamespaceLockFence, NamespaceLockSignalTestFence};
 pub(crate) use crate::storage_api_contracts::list::{
     ListOperations, StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageObjectInfoOrErr, StorageWalkOptions,
 };

crates/ecstore/src/disk/local.rs

@@ -22463,6 +22463,86 @@ mod test {
         );
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn conditional_mrf_manifest_dir_fsync_failure_keeps_recovery_anchors() {
+        use tempfile::tempdir;
+
+        const MRF_COMMIT_MANIFEST_SLOT_0: &str = ".heal-mrf-commit.0.bin";
+        const MRF_COMMIT_MANIFEST_SLOT_1: &str = ".heal-mrf-commit.1.bin";
+        const MRF_SCOPED_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal-scoped.bin";
+
+        let _mode = durability_mode_override::set(DurabilityMode::Relaxed);
+        let dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let previous_manifest = Bytes::from_static(b"mrf-committed-manifest-v1");
+        let successor_manifest = Bytes::from_static(b"mrf-committed-manifest-v2");
+        let legacy_journal = Bytes::from_static(b"legacy-mrf-journal-records");
+
+        assert_eq!(
+            disk.compare_and_update_file(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0, None, Some(previous_manifest.clone()),)
+                .await
+                .expect("previous MRF manifest should commit"),
+            ConditionalFileUpdate::Updated
+        );
+        disk.write_all(RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH, legacy_journal.clone())
+            .await
+            .expect("legacy MRF journal should be retained");
+
+        let manifest_path = disk
+            .get_object_path(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0)
+            .expect("MRF manifest path should resolve");
+        let parent = manifest_path.parent().expect("MRF manifest path should have a parent");
+        assert!(
+            os::fsync_dir_recorder::was_fsynced(parent),
+            "system metadata MRF manifest publication must fsync the metadata directory even under relaxed durability"
+        );
+        os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
+
+        let err = disk
+            .compare_and_update_file(
+                RUSTFS_META_BUCKET,
+                MRF_COMMIT_MANIFEST_SLOT_0,
+                Some(previous_manifest.clone()),
+                Some(successor_manifest),
+            )
+            .await
+            .expect_err("directory fsync failure must fail the MRF manifest successor commit");
+        assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_0)
+                .await
+                .expect("previous committed MRF manifest should remain readable after rollback"),
+            previous_manifest
+        );
+
+        os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
+        let err = disk
+            .compare_and_update_file(
+                RUSTFS_META_BUCKET,
+                MRF_COMMIT_MANIFEST_SLOT_1,
+                None,
+                Some(Bytes::from_static(b"first-successor-manifest")),
+            )
+            .await
+            .expect_err("directory fsync failure must fail first MRF manifest commit");
+        assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
+        assert!(
+            matches!(
+                disk.read_all(RUSTFS_META_BUCKET, MRF_COMMIT_MANIFEST_SLOT_1).await,
+                Err(DiskError::FileNotFound)
+            ),
+            "uncommitted first MRF manifest must be removed when no committed anchor exists"
+        );
+        assert_eq!(
+            disk.read_all(RUSTFS_META_BUCKET, MRF_SCOPED_JOURNAL_PATH)
+                .await
+                .expect("legacy MRF journal should remain readable after failed manifest publication"),
+            legacy_journal
+        );
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn conditional_file_update_dir_fsync_failure_removes_new_file_without_anchor() {

crates/ecstore/src/store/heal.rs

@@ -400,6 +400,7 @@ impl ECStore {
             // never to the cluster's authoritative metadata transaction.
             let metadata_opts = HealOpts {
                 dry_run: opts.dry_run,
+                recreate: opts.recreate,
                 scan_mode: opts.scan_mode,
                 pool: Some(pool_index),
                 set: Some(set_index),

crates/heal/src/heal/erasure_healer.rs

@@ -113,6 +113,7 @@ pub struct ErasureSetHealer {
     heal_opts: HealOpts,
     source: HealRequestSource,
     target_endpoints: Arc<[String]>,
+    pool_metadata_target_endpoints: Arc<[String]>,
     replacement_task_id: Option<String>,
     replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
     mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
@@ -362,6 +363,7 @@ impl ErasureSetHealer {
             heal_opts,
             source,
             target_endpoints: Vec::new().into(),
+            pool_metadata_target_endpoints: Vec::new().into(),
             replacement_task_id: None,
             replacement_target_identities: None,
             mainline_pacer: None,
@@ -385,6 +387,13 @@ impl ErasureSetHealer {
         self
     }
 
+    pub(crate) fn with_pool_metadata_targets(mut self, mut target_endpoints: Vec<String>) -> Self {
+        target_endpoints.sort_unstable();
+        target_endpoints.dedup();
+        self.pool_metadata_target_endpoints = target_endpoints.into();
+        self
+    }
+
     pub(crate) fn with_replacement_identity_fence(
         mut self,
         replacement_target_identities: Option<Vec<ReplacementTargetIdentity>>,
@@ -948,33 +957,37 @@ impl ErasureSetHealer {
         resume_manager: &ResumeManager,
         checkpoint_manager: &CheckpointManager,
     ) -> Result<()> {
-        let ordinary_opts = if self.replacement_task_id.is_none() {
+        let mut metadata_opts = self.heal_opts;
+        metadata_opts.remove = false;
+        metadata_opts.no_lock = false;
+        if self.replacement_task_id.is_none() {
             let (pool_index, set_index) = crate::heal::utils::parse_set_disk_id(set_disk_id)?;
-            if self.heal_opts.pool.is_some_and(|pool| pool != pool_index)
-                || self.heal_opts.set.is_some_and(|set| set != set_index)
+            if metadata_opts.pool.is_some_and(|pool| pool != pool_index) || metadata_opts.set.is_some_and(|set| set != set_index)
             {
                 return Err(Error::TaskExecutionFailed {
                     message: format!("Pool metadata scope does not match resumed set {set_disk_id}"),
                 });
             }
-            Some(HealOpts {
-                dry_run: self.heal_opts.dry_run,
-                scan_mode: self.heal_opts.scan_mode,
-                pool: Some(pool_index),
-                set: Some(set_index),
-                ..Default::default()
-            })
+            metadata_opts.pool = Some(pool_index);
+            metadata_opts.set = Some(set_index);
+        }
+        let target_endpoints = if self.replacement_task_id.is_some() || self.pool_metadata_target_endpoints.is_empty() {
+            self.target_endpoints.as_ref()
         } else {
-            if self.target_endpoints.is_empty() {
+            self.pool_metadata_target_endpoints.as_ref()
+        };
+        let target_scoped_recreate = !metadata_opts.dry_run && metadata_opts.recreate && !target_endpoints.is_empty();
+        let ordinary_heal = self.replacement_task_id.is_none() && !target_scoped_recreate;
+        if !ordinary_heal {
+            if target_endpoints.is_empty() {
                 return Err(Error::TaskExecutionFailed {
                     message: "Replacement pool metadata heal requires target endpoints".to_string(),
                 });
             }
-            if !self.storage.replacement_pool_metadata_applies(&self.heal_opts).await? {
+            if !self.storage.replacement_pool_metadata_applies(&metadata_opts).await? {
                 return Ok(());
             }
-            None
-        };
+        }
 
         let object_key = format!("{RUSTFS_META_BUCKET}/{POOL_META_NAME}");
         let checkpoint_key = compose_key(&object_key, None);
@@ -992,8 +1005,8 @@ impl ErasureSetHealer {
             .set_current_item(Some(RUSTFS_META_BUCKET.to_string()), Some(POOL_META_NAME.to_string()))
             .await?;
 
-        let result = if let Some(opts) = ordinary_opts {
-            match self.storage.heal_pool_metadata(&opts).await {
+        let result = if ordinary_heal {
+            match self.storage.heal_pool_metadata(&metadata_opts).await {
                 Ok(results) if results.is_empty() => return Ok(()),
                 Ok(results) => {
                     let [result] = results.as_slice() else {
@@ -1014,10 +1027,10 @@ impl ErasureSetHealer {
         } else {
             match self
                 .storage
-                .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts)
+                .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &metadata_opts)
                 .await
             {
-                Ok((result, None)) if target_outcomes_complete(&result, &self.target_endpoints) => {
+                Ok((result, None)) if target_outcomes_complete(&result, target_endpoints) => {
                     let object_size = result_object_size_u64(&result);
                     match self
                         .storage
@@ -1025,8 +1038,8 @@ impl ErasureSetHealer {
                             RUSTFS_META_BUCKET,
                             POOL_META_NAME,
                             None,
-                            &self.heal_opts,
-                            &self.target_endpoints,
+                            &metadata_opts,
+                            target_endpoints,
                         )
                         .await
                     {
@@ -2099,7 +2112,7 @@ mod resume_loop_tests {
         /// fake models a healthy backend unless a test explicitly revokes it.
         replacement_commit_evidence: Mutex<HashMap<String, ReplacementCommitEvidence>>,
         ordinary_pool_metadata_required: AtomicBool,
-        ordinary_pool_metadata_opts: Mutex<Vec<HealOpts>>,
+        pool_metadata_opts: Mutex<Vec<HealOpts>>,
         pool_metadata_not_applicable: AtomicBool,
         fail_pool_metadata_scope: AtomicBool,
         lifecycle_expired: Mutex<HashSet<String>>,
@@ -2190,11 +2203,14 @@ mod resume_loop_tests {
         }
         async fn heal_object(
             &self,
-            _bucket: &str,
+            bucket: &str,
             object: &str,
             version_id: Option<&str>,
-            _opts: &HealOpts,
+            opts: &HealOpts,
         ) -> Result<(HealResultItem, Option<Error>)> {
+            if bucket == RUSTFS_META_BUCKET && object == POOL_META_NAME {
+                self.pool_metadata_opts.lock().expect("metadata options").push(*opts);
+            }
             self.heal_calls
                 .lock()
                 .unwrap()
@@ -2221,7 +2237,6 @@ mod resume_loop_tests {
             if !self.ordinary_pool_metadata_required.load(Ordering::SeqCst) {
                 return Ok(Vec::new());
             }
-            self.ordinary_pool_metadata_opts.lock().expect("metadata options").push(*opts);
             if !self.replacement_pool_metadata_applies(opts).await? {
                 return Ok(Vec::new());
             }
@@ -2749,7 +2764,7 @@ mod resume_loop_tests {
 
         assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
         {
-            let opts = env.storage.ordinary_pool_metadata_opts.lock().expect("metadata options");
+            let opts = env.storage.pool_metadata_opts.lock().expect("metadata options");
             assert_eq!(opts.len(), 1);
             assert_eq!((opts[0].pool, opts[0].set), (Some(0), Some(0)));
         }
@@ -2783,7 +2798,7 @@ mod resume_loop_tests {
             .await
             .expect("ordinary dry-run metadata work must not require a replacement commit");
         assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
-        let opts = env.storage.ordinary_pool_metadata_opts.lock().expect("metadata options");
+        let opts = env.storage.pool_metadata_opts.lock().expect("metadata options");
         assert_eq!(opts.len(), 1);
         assert!(opts[0].dry_run);
         assert!(!opts[0].remove);
@@ -3050,6 +3065,105 @@ mod resume_loop_tests {
         assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
     }
 
+    #[tokio::test]
+    async fn admin_recreate_target_heals_pool_metadata_before_completion() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts {
+                recreate: true,
+                remove: true,
+                no_lock: true,
+                ..Default::default()
+            },
+            HealRequestSource::Admin,
+        )
+        .with_pool_metadata_targets(vec!["replacement-a".to_string()]);
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
+
+        healer
+            .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
+            .await
+            .expect("admin recreate should heal and verify pool metadata");
+
+        assert!(env.resume.get_state().await.completed);
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+        let opts = env.storage.pool_metadata_opts.lock().expect("metadata options");
+        assert_eq!(opts.len(), 1);
+        assert_eq!((opts[0].pool, opts[0].set), (Some(0), Some(0)));
+        assert!(opts[0].recreate);
+        assert!(!opts[0].remove);
+        assert!(!opts[0].no_lock);
+    }
+
+    #[tokio::test]
+    async fn admin_recreate_pool_metadata_validates_owner_scope_before_io() {
+        for (non_owner, unknown_scope, pool) in [(true, false, None), (false, true, None), (false, false, Some(1))] {
+            let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+            env.storage.pool_metadata_not_applicable.store(non_owner, Ordering::SeqCst);
+            env.storage.fail_pool_metadata_scope.store(unknown_scope, Ordering::SeqCst);
+            let healer = ErasureSetHealer::new(
+                env.storage.clone(),
+                Arc::new(RwLock::new(HealProgress::new())),
+                CancellationToken::new(),
+                env.healer.disk.clone(),
+                HealOpts {
+                    recreate: true,
+                    pool,
+                    ..Default::default()
+                },
+                HealRequestSource::Admin,
+            )
+            .with_pool_metadata_targets(vec!["replacement-a".to_string()]);
+
+            let set_disk_id = if non_owner { "pool_0_set_1" } else { "pool_0_set_0" };
+            let result = healer
+                .execute_heal_with_resume(&[], set_disk_id, &env.resume, &env.checkpoint)
+                .await;
+
+            assert_eq!(result.is_ok(), non_owner, "only a known non-owner may skip metadata: {result:?}");
+            assert!(env.storage.calls().is_empty(), "scope validation must precede metadata I/O");
+            assert_eq!(env.resume.get_state().await.completed, non_owner);
+        }
+    }
+
+    #[tokio::test]
+    async fn admin_recreate_pool_metadata_readback_failure_keeps_resume_state() {
+        let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
+        let healer = ErasureSetHealer::new(
+            env.storage.clone(),
+            Arc::new(RwLock::new(HealProgress::new())),
+            CancellationToken::new(),
+            env.healer.disk.clone(),
+            HealOpts {
+                recreate: true,
+                pool: Some(0),
+                set: Some(0),
+                ..Default::default()
+            },
+            HealRequestSource::Admin,
+        )
+        .with_pool_metadata_targets(vec!["replacement-a".to_string()]);
+        env.storage
+            .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
+        env.storage.set_replacement_commit_evidence(POOL_META_NAME, None, false);
+
+        let error = healer
+            .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
+            .await
+            .expect_err("unconfirmed admin recreate pool metadata must keep the set incomplete");
+
+        assert!(error.to_string().contains("Erasure set heal incomplete"));
+        let state = env.resume.get_state().await;
+        assert!(!state.completed);
+        assert_eq!(state.retry_count, 1);
+        assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
+    }
+
     #[tokio::test]
     async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() {
         let env = make_env().await;

crates/heal/src/heal/manager.rs

@@ -1527,7 +1527,7 @@ impl HealManager {
         request: HealRequest,
         preserve_alias: bool,
     ) -> Result<HealAdmissionReceipt> {
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, None)
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, None)
             .await
     }
 
@@ -1563,7 +1563,7 @@ impl HealManager {
         request: HealRequest,
         mrf_notice_target: MrfRepairNoticeTarget,
     ) -> Result<HealAdmissionReceipt> {
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, Some(mrf_notice_target))
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, true, Some(mrf_notice_target))
             .await
     }
 
@@ -1583,6 +1583,7 @@ impl HealManager {
         &self,
         request: HealRequest,
         preserve_alias: bool,
+        accept_same_request_id_replay: bool,
         mrf_notice_target: Option<MrfRepairNoticeTarget>,
     ) -> Result<HealAdmissionReceipt> {
         let admission_start = Instant::now();
@@ -1661,7 +1662,11 @@ impl HealManager {
             });
         if let Some((matches_existing, duplicate_state)) = request_id_admission {
             let admission = if matches_existing {
-                HealAdmissionResult::Accepted
+                if accept_same_request_id_replay {
+                    HealAdmissionResult::Accepted
+                } else {
+                    Self::duplicate_admission_for_request(&request, &config)
+                }
             } else {
                 HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
             };
@@ -1908,7 +1913,10 @@ impl HealManager {
 
     /// Submit heal request.
     pub async fn submit_heal_request(&self, request: HealRequest) -> Result<HealAdmissionResult> {
-        Ok(self.submit_heal_request_with_receipt_and_alias(request, true).await?.result)
+        Ok(self
+            .submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, false, None)
+            .await?
+            .result)
     }
 
     /// Get task status

crates/heal/src/heal/mrf_queue.rs

@@ -297,7 +297,11 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     };
     let attempts = data[3];
     let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?);
-    let has_version = data[12] != 0;
+    let has_version = match data[12] {
+        0 => false,
+        1 => true,
+        _ => return None,
+    };
     let mut cursor = MRF_RECORD_FIXED_HEAD;
     let version_id = if has_version {
         if data.len() < cursor + 16 {
@@ -520,6 +524,8 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
 struct MrfRuntime {
     queue: MrfQueue,
     config: MrfConsumerConfig,
+    checkpoint_owner: Uuid,
+    next_checkpoint_sequence: u64,
     new_since_flush: usize,
     /// True while the in-memory pending set has changed since the last
     /// journal flush (push, pop, or an attempts bump that alters the encoded
@@ -540,6 +546,12 @@ struct MrfRuntime {
     /// exact verified repair discharges them. Live admissions never grow this
     /// set, so its size is bounded by the decoded startup journal.
     retained_replay_intents: HashMap<MrfQueueKey, MrfIntent>,
+    /// Startup replay source to remove after the retained replay
+    /// responsibilities are discharged. `None` means the runtime only needs
+    /// the legacy journal cleanup path for snapshots it wrote itself.
+    replay_cleanup: Option<ReplayCleanup>,
+    /// Last committed checkpoint published by this runtime flush path.
+    runtime_checkpoint: Option<(Uuid, u64)>,
     /// Earliest instant a full-admission retry may proceed.
     backoff_until: Option<tokio::time::Instant>,
 }
@@ -631,6 +643,34 @@ impl MrfRuntime {
             self.dirty = true;
             return;
         };
+        let (committed_persisted, committed_on_disk) = if authoritative.is_empty() {
+            (true, false)
+        } else {
+            match snapshot::publish_committed_snapshot(
+                &journal_disks().await,
+                self.checkpoint_owner,
+                self.next_checkpoint_sequence,
+                &authoritative,
+                self.config.journal_max_bytes,
+            )
+            .await
+            {
+                Ok(publication) => {
+                    self.runtime_checkpoint = Some((publication.owner, publication.sequence));
+                    self.next_checkpoint_sequence = publication.sequence.saturating_add(1);
+                    (true, true)
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        target: "rustfs::heal::mrf",
+                        error = %err,
+                        sequence = self.next_checkpoint_sequence,
+                        "MRF committed checkpoint publish failed; retaining previous replay anchor"
+                    );
+                    (false, false)
+                }
+            }
+        };
         let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -641,10 +681,10 @@ impl MrfRuntime {
         // old reader from observing a newer epoch that a new reader cannot
         // see when the canonical write is unavailable.
         let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, &legacy).await;
-        // Keep dirty until both the authoritative snapshot and its
-        // compatibility mirror have been accepted; otherwise a one-sided
-        // failure would never retry the missing file.
-        let persisted = authoritative_persisted && legacy_persisted;
+        // Keep dirty until the committed checkpoint, authoritative snapshot,
+        // and compatibility mirror have all been accepted; otherwise a
+        // one-sided failure would never retry the missing recovery anchor.
+        let persisted = committed_persisted && authoritative_persisted && legacy_persisted;
         self.new_since_flush = 0;
         // Keep the dirty flag when every disk write failed: a clean backlog
         // would otherwise never rewrite, losing the periodic persist retry a
@@ -652,7 +692,7 @@ impl MrfRuntime {
         if persisted {
             self.dirty = false;
         }
-        self.journal_on_disk |= authoritative_persisted || legacy_persisted;
+        self.journal_on_disk |= committed_on_disk || authoritative_persisted || legacy_persisted;
     }
 
     /// Drain pending intents into the heal manager until it is full, the
@@ -730,6 +770,45 @@ impl MrfRuntime {
         self.retain_replay_journal || !self.retained_replay_intents.is_empty()
     }
 
+    fn replay_cleanup_to_delete(&self) -> Option<ReplayCleanup> {
+        if self.journal_on_disk && !self.retained_replay_journal() {
+            Some(self.replay_cleanup.unwrap_or(ReplayCleanup::Legacy))
+        } else {
+            None
+        }
+    }
+
+    async fn delete_idle_recovery_anchors(&mut self) -> bool {
+        let runtime_deleted = match self.runtime_checkpoint {
+            Some((owner, sequence)) => {
+                match snapshot::delete_committed_snapshots_through(owner, sequence, self.config.journal_max_bytes).await {
+                    Ok(deleted) => deleted,
+                    Err(err) => {
+                        tracing::warn!(
+                            target: "rustfs::heal::mrf",
+                            error = %err,
+                            sequence,
+                            "MRF runtime checkpoint cleanup failed"
+                        );
+                        false
+                    }
+                }
+            }
+            None => true,
+        };
+        let replay_deleted = match self.replay_cleanup_to_delete() {
+            Some(cleanup) => delete_replay_source(cleanup, self.config.journal_max_bytes).await,
+            None => true,
+        };
+        if runtime_deleted && replay_deleted {
+            self.runtime_checkpoint = None;
+            self.replay_cleanup = None;
+            true
+        } else {
+            false
+        }
+    }
+
     fn discharge_durable_replay_anchors(&mut self) {
         if self.durable_replay_anchors.is_empty() {
             return;
@@ -808,6 +887,8 @@ struct ReplayOutcome {
     retain_journal_for_replay: bool,
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
     retained_replay_intents: HashMap<MrfQueueKey, MrfIntent>,
+    cleanup: Option<ReplayCleanup>,
+    next_checkpoint_sequence: u64,
 }
 
 fn replay_must_retain_journal(
@@ -819,10 +900,10 @@ fn replay_must_retain_journal(
     rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
 }
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum ReplayCleanup {
     Legacy,
-    Committed { sequence: u64 },
+    Committed { owner: Uuid, sequence: u64 },
 }
 
 struct ReplaySource {
@@ -835,6 +916,7 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
         return Ok(Some(ReplaySource {
             data: committed.payload().to_vec(),
             cleanup: ReplayCleanup::Committed {
+                owner: committed.owner(),
                 sequence: committed.sequence(),
             },
         }));
@@ -858,18 +940,20 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
 async fn delete_replay_source(cleanup: ReplayCleanup, max_bytes: usize) -> bool {
     let committed_deleted = match cleanup {
         ReplayCleanup::Legacy => true,
-        ReplayCleanup::Committed { sequence } => match snapshot::delete_committed_snapshots_through(sequence, max_bytes).await {
-            Ok(deleted) => deleted,
-            Err(err) => {
-                tracing::warn!(
-                    target: "rustfs::heal::mrf",
-                    error = %err,
-                    sequence,
-                    "MRF committed replay checkpoint cleanup failed"
-                );
-                false
+        ReplayCleanup::Committed { owner, sequence } => {
+            match snapshot::delete_committed_snapshots_through(owner, sequence, max_bytes).await {
+                Ok(deleted) => deleted,
+                Err(err) => {
+                    tracing::warn!(
+                        target: "rustfs::heal::mrf",
+                        error = %err,
+                        sequence,
+                        "MRF committed replay checkpoint cleanup failed"
+                    );
+                    false
+                }
             }
-        },
+        }
     };
     committed_deleted && delete_journals().await
 }
@@ -891,6 +975,8 @@ async fn replay_into(
                 retain_journal_for_replay: false,
                 durable_replay_anchors: Vec::new(),
                 retained_replay_intents: HashMap::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
         Err(err) => {
@@ -905,10 +991,16 @@ async fn replay_into(
                 retain_journal_for_replay: true,
                 durable_replay_anchors: Vec::new(),
                 retained_replay_intents: HashMap::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
     };
     let cleanup = source.cleanup;
+    let next_checkpoint_sequence = match cleanup {
+        ReplayCleanup::Legacy => 1,
+        ReplayCleanup::Committed { sequence, .. } => sequence.saturating_add(1),
+    };
     let data = source.data;
     let (decoded, truncated) = decode_journal(&data);
     let replayed = decoded.len();
@@ -1016,6 +1108,8 @@ async fn replay_into(
         retain_journal_for_replay,
         durable_replay_anchors,
         retained_replay_intents,
+        cleanup: journal_on_disk.then_some(cleanup),
+        next_checkpoint_sequence,
     }
 }
 
@@ -1026,12 +1120,16 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let mut runtime = MrfRuntime {
         queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
         config: config.clone(),
+        checkpoint_owner: Uuid::new_v4(),
+        next_checkpoint_sequence: 1,
         new_since_flush: 0,
         dirty: false,
         journal_on_disk: false,
         retain_replay_journal: false,
         durable_replay_anchors: Vec::new(),
         retained_replay_intents: HashMap::new(),
+        replay_cleanup: None,
+        runtime_checkpoint: None,
         backoff_until: None,
     };
 
@@ -1042,6 +1140,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     runtime.retain_replay_journal = replay.retain_journal_for_replay;
     runtime.durable_replay_anchors = replay.durable_replay_anchors;
     runtime.retained_replay_intents = replay.retained_replay_intents;
+    runtime.replay_cleanup = replay.cleanup;
+    runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
     // Anything still pending (e.g. the manager was full and backoff armed)
     // must be re-persisted by the next flush before replay can delete the
     // startup anchor.
@@ -1096,7 +1196,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     TickAction::DeleteJournal => {
                         // All replayed intents have either been accepted,
                         // merged, or replaced by a pending successor snapshot.
-                        if delete_journals().await {
+                        if runtime.delete_idle_recovery_anchors().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
                         }
@@ -1212,15 +1312,24 @@ mod tests {
         let bucket_incarnation_id = uuid::Uuid::new_v4();
         let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, bucket_incarnation_id)
             .expect("fresh replay lease and bucket incarnation build a durable anchor");
+        let cleanup_owner = uuid::Uuid::new_v4();
+        let cleanup = ReplayCleanup::Committed {
+            owner: cleanup_owner,
+            sequence: 17,
+        };
         let mut runtime = MrfRuntime {
             queue: MrfQueue::new(2, usize::MAX),
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: false,
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: vec![anchor],
             retained_replay_intents: HashMap::from([(queue_key(&intent), intent.clone())]),
+            replay_cleanup: Some(cleanup),
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
@@ -1243,6 +1352,11 @@ mod tests {
             1,
             "an admitted responsibility must remain in the successor before proof"
         );
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            None,
+            "the committed replay source must not be reclaimed before the exact proof"
+        );
         runtime.discharge_durable_replay_anchors();
         assert!(
             !runtime.retained_replay_journal(),
@@ -1250,6 +1364,11 @@ mod tests {
         );
         assert!(runtime.dirty, "proof removal must be persisted by the next flush");
         assert!(runtime.snapshot().expect("discharged snapshot").0.is_empty());
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(cleanup),
+            "proof discharge must preserve the committed owner/sequence cleanup target"
+        );
         rustfs_common::mrf_channel::release_mrf_intent(&intent);
     }
 
@@ -1262,6 +1381,8 @@ mod tests {
         let runtime = MrfRuntime {
             queue,
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: true,
             journal_on_disk: true,
@@ -1271,6 +1392,8 @@ mod tests {
                 (queue_key(&accepted), accepted),
                 (queue_key(&pending), intent("snapshot-bucket", "pending", 0)),
             ]),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
             backoff_until: None,
         };
 
@@ -1300,12 +1423,16 @@ mod tests {
         let mut runtime = MrfRuntime {
             queue,
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: true,
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: Vec::new(),
             retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         assert!(runtime.snapshot().is_none(), "combined count must honor the queue ceiling");
@@ -1341,12 +1468,16 @@ mod tests {
             let mut runtime = MrfRuntime {
                 queue: MrfQueue::new(if count_limited { 1 } else { 2 }, retained.estimated_bytes()),
                 config: MrfConsumerConfig::default(),
+                checkpoint_owner: Uuid::new_v4(),
+                next_checkpoint_sequence: 1,
                 new_since_flush: 0,
                 dirty: false,
                 journal_on_disk: true,
                 retain_replay_journal: false,
                 durable_replay_anchors: vec![anchor],
                 retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
+                replay_cleanup: None,
+                runtime_checkpoint: None,
                 backoff_until: None,
             };
             if count_limited {
@@ -1393,12 +1524,16 @@ mod tests {
         let mut runtime = MrfRuntime {
             queue: MrfQueue::new(2, 4096),
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: false,
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: vec![anchor],
             retained_replay_intents: HashMap::from([(queue_key(&retained), retained.clone())]),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         let mut newer = retained.clone();
@@ -1428,6 +1563,31 @@ mod tests {
         assert_eq!(decode_journal(&runtime.snapshot().expect("new lease successor").0).0.len(), 2);
     }
 
+    #[test]
+    fn runtime_cleanup_defaults_to_legacy_for_runtime_written_journals() {
+        let runtime = MrfRuntime {
+            queue: MrfQueue::new(2, usize::MAX),
+            config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: true,
+            retain_replay_journal: false,
+            durable_replay_anchors: Vec::new(),
+            retained_replay_intents: HashMap::new(),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
+            backoff_until: None,
+        };
+
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(ReplayCleanup::Legacy),
+            "journals written by the runtime still use the legacy cleanup path"
+        );
+    }
+
     #[test]
     fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
         let unique = uuid::Uuid::new_v4();
@@ -1485,12 +1645,16 @@ mod tests {
         let runtime = MrfRuntime {
             queue: MrfQueue::new(1, 4096),
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: false,
             journal_on_disk: true,
             retain_replay_journal: true,
             durable_replay_anchors: Vec::new(),
             retained_replay_intents,
+            replay_cleanup: None,
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         let (snapshot, _) = runtime
@@ -1669,6 +1833,25 @@ mod tests {
         assert_eq!(truncated, corrupt.len());
     }
 
+    #[test]
+    fn journal_rejects_unknown_version_presence_flag_even_with_valid_crc() {
+        let mut versioned = intent("rollback-bucket", "object", 0);
+        versioned.version_id = Some([9; 16]);
+        let mut buf = Vec::new();
+        assert!(encode_intent(&versioned, &mut buf));
+
+        buf[12] = 2;
+        let crc_offset = buf.len() - 4;
+        let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        hasher.update(&buf[..crc_offset]);
+        let checksum = u32::try_from(hasher.finalize()).expect("CRC32 fits");
+        buf[crc_offset..].copy_from_slice(&checksum.to_le_bytes());
+
+        let (decoded, truncated) = decode_journal(&buf);
+        assert!(decoded.is_empty(), "unknown boolean encodings are not rollback-compatible payloads");
+        assert_eq!(truncated, buf.len());
+    }
+
     #[test]
     fn heal_request_mapping_follows_priority_matrix() {
         let decode = build_heal_request(&intent("b", "o", 0));

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -22,8 +22,9 @@
 //! responsibility through a newer durable snapshot or a verified repair proof.
 //! An unreadable commit path cannot prove that only legacy data exists. This
 //! explicit inspection API fails closed and never mutates recovery anchors.
-//! It is wired into the replay reader before writer activation, but the writer
-//! remains gated on ownership-aware handoff.
+//! The live consumer writes committed checkpoints alongside the scoped and
+//! legacy journal mirrors; cleanup remains gated by replay ownership and exact
+//! verified repair proof handoff.
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
@@ -356,8 +357,8 @@ fn validate_reusable_manifest_slot(existing: Option<&[u8]>, sequence: u64, paylo
 /// The writer is a narrow production primitive for the ownership-aware MRF
 /// handoff: it validates the whole journal payload, preserves the previous
 /// committed slot, and publishes the manifest only after the successor payload
-/// reaches the same disk. It does not delete legacy journals, tombstone older
-/// anchors, or activate the live consumer.
+/// reaches the same disk. It does not delete legacy journals or tombstone older
+/// anchors by itself; the consumer decides cleanup after replay handoff.
 pub async fn publish_committed_snapshot(
     disks: &[EcstoreDiskStore],
     owner: Uuid,
@@ -371,7 +372,10 @@ pub async fn publish_committed_snapshot(
     if owner.is_nil() || sequence == 0 || sequence == u64::MAX {
         return Err(SnapshotError::Corrupt);
     }
-    if payload.len() > limit || decode_journal(payload).1 != 0 {
+    if payload.len() > limit {
+        return Err(SnapshotError::TooLarge);
+    }
+    if decode_journal(payload).1 != 0 {
         return Err(SnapshotError::Corrupt);
     }
     let current = read_committed(disks, limit).await?;
@@ -555,31 +559,35 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed manifests whose sequence is no newer than
+/// Remove committed checkpoints from `owner` whose sequence is no newer than
 /// `committed_through`.
 ///
-/// Payload files are intentionally left as orphans after their manifest is
-/// removed. Readers cannot discover a payload without its matching manifest,
-/// and deleting manifests first prevents an older retained slot from becoming
-/// visible again after the newest replay has been fully discharged.
-pub async fn delete_committed_snapshots_through(committed_through: u64, max_bytes: usize) -> Result<bool, SnapshotError> {
+/// Cleanup is manifest-first so readers cannot rediscover an older payload
+/// after the newest replay has been fully discharged. The payload is removed
+/// only after the manifest and body were revalidated as one complete committed
+/// checkpoint; damaged, future, mismatched, or foreign-owner slots are retained.
+pub async fn delete_committed_snapshots_through(
+    owner: Uuid,
+    committed_through: u64,
+    max_bytes: usize,
+) -> Result<bool, SnapshotError> {
     let disks = super::journal_disks().await;
-    delete_committed_snapshots_through_on(&disks, committed_through, max_bytes).await
+    delete_committed_snapshots_through_on(&disks, owner, committed_through, max_bytes).await
 }
 
 async fn delete_committed_snapshots_through_on(
     disks: &[EcstoreDiskStore],
+    owner: Uuid,
     committed_through: u64,
     max_bytes: usize,
 ) -> Result<bool, SnapshotError> {
     if disks.is_empty() {
         return Err(SnapshotError::NoWritableReplica);
     }
-    let mut any_changed = false;
     let mut first_error = None;
     for disk in disks {
-        for path in MANIFEST_PATHS {
-            let existing = match read_bounded(disk, path, MANIFEST_LEN).await {
+        for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
+            let manifest_bytes = match read_bounded(disk, manifest_path, MANIFEST_LEN).await {
                 Ok(Some(existing)) => existing,
                 Ok(None) => continue,
                 Err(error) => {
@@ -589,7 +597,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            let manifest = match Manifest::decode(&existing, max_bytes) {
+            let manifest = match Manifest::decode(&manifest_bytes, max_bytes) {
                 Ok(manifest) => manifest,
                 Err(error) => {
                     if first_error.is_none() {
@@ -598,19 +606,61 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            if manifest.sequence > committed_through {
+            if manifest.owner != owner || manifest.sequence > committed_through {
+                continue;
+            }
+            let payload_bytes = match read_bounded(disk, payload_path, manifest.payload_len).await {
+                Ok(Some(payload)) => payload,
+                Ok(None) => {
+                    if first_error.is_none() {
+                        first_error = Some(SnapshotError::Corrupt);
+                    }
+                    continue;
+                }
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    continue;
+                }
+            };
+            if let Err(error) = CommittedSnapshot::decode(0, &manifest_bytes, payload_bytes.clone(), max_bytes) {
+                if first_error.is_none() {
+                    first_error = Some(error);
+                }
                 continue;
             }
             match EcstoreDiskAPI::compare_and_update_file(
                 disk.as_ref(),
                 RUSTFS_META_BUCKET,
-                path,
-                Some(EcstoreDiskBytes::from(existing)),
+                manifest_path,
+                Some(EcstoreDiskBytes::copy_from_slice(&manifest_bytes)),
                 None,
             )
             .await
             {
-                Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true,
+                Ok(EcstoreConditionalFileUpdate::Updated) => {
+                    match EcstoreDiskAPI::compare_and_update_file(
+                        disk.as_ref(),
+                        RUSTFS_META_BUCKET,
+                        payload_path,
+                        Some(EcstoreDiskBytes::copy_from_slice(&payload_bytes)),
+                        None,
+                    )
+                    .await
+                    {
+                        Ok(
+                            EcstoreConditionalFileUpdate::Updated
+                            | EcstoreConditionalFileUpdate::Missing
+                            | EcstoreConditionalFileUpdate::Mismatch,
+                        ) => {}
+                        Err(error) => {
+                            if first_error.is_none() {
+                                first_error = Some(SnapshotError::Disk(error));
+                            }
+                        }
+                    }
+                }
                 Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
                 Err(error) => {
                     if first_error.is_none() {
@@ -620,13 +670,7 @@ async fn delete_committed_snapshots_through_on(
             }
         }
     }
-    if any_changed {
-        Ok(true)
-    } else if let Some(error) = first_error {
-        Err(error)
-    } else {
-        Ok(true)
-    }
+    if let Some(error) = first_error { Err(error) } else { Ok(true) }
 }
 
 async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
@@ -1096,6 +1140,45 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn committed_snapshot_writer_capacity_failure_preserves_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let store = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        commit(&store, 0, owner, 1, &old).await;
+
+        let result = publish_committed_snapshot(std::slice::from_ref(&store), owner, 2, &next, next.len() - 1).await;
+
+        assert!(
+            matches!(result, Err(SnapshotError::TooLarge)),
+            "capacity failure must be reported separately from corruption: {result:?}"
+        );
+        let reopened = disk(&root, "disk").await;
+        let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+            .await
+            .expect("read previous committed snapshot")
+            .expect("old anchor remains committed");
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.slot(), 0);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("old manifest retained")
+                .as_ref(),
+            manifest(owner, 1, &old).as_slice()
+        );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "oversized successor payload must not be staged"
+        );
+    }
+
     #[tokio::test]
     async fn committed_snapshot_writer_does_not_overwrite_damaged_inactive_manifest() {
         let root = TempDir::new().expect("test directory");
@@ -1200,6 +1283,83 @@ mod tests {
         assert_eq!(recovered.manifest.sequence, 1);
     }
 
+    #[tokio::test]
+    async fn manifest_cas_failure_after_payload_write_keeps_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let store = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        let damaged_manifest = b"damaged successor manifest".to_vec();
+        commit(&store, 0, owner, 1, &old).await;
+
+        let expected_manifest = EcstoreDiskAPI::read_all(store.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+            .await
+            .ok();
+        assert_eq!(
+            cas_replace(&store, PAYLOAD_PATHS[1], &next, 4096)
+                .await
+                .expect("successor payload CAS"),
+            EcstoreConditionalFileUpdate::Updated
+        );
+        install(&store, MANIFEST_PATHS[1], &damaged_manifest).await;
+
+        let manifest_update = cas_replace_expected(&store, MANIFEST_PATHS[1], expected_manifest, &manifest(owner, 2, &next))
+            .await
+            .expect("successor manifest CAS");
+        assert_eq!(manifest_update, EcstoreConditionalFileUpdate::Mismatch);
+
+        let reopened = disk(&root, "disk").await;
+        let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+            .await
+            .expect("read committed snapshot after failed successor CAS")
+            .expect("previous committed anchor");
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.slot(), 0);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
+                .await
+                .expect("successor payload remains non-authoritative")
+                .as_ref(),
+            next.as_slice()
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                .await
+                .expect("failed successor manifest retained")
+                .as_ref(),
+            damaged_manifest.as_slice()
+        );
+    }
+
+    #[tokio::test]
+    async fn torn_successor_on_one_replica_does_not_hide_previous_anchor_on_peer() {
+        let root = TempDir::new().expect("test directory");
+        let first = disk(&root, "first").await;
+        let second = disk(&root, "second").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        let damaged_manifest = b"damaged successor manifest".to_vec();
+        commit(&first, 0, owner, 1, &old).await;
+        commit(&second, 0, owner, 1, &old).await;
+        install(&first, PAYLOAD_PATHS[1], &next).await;
+        install(&first, MANIFEST_PATHS[1], &damaged_manifest).await;
+
+        let mut stats = SnapshotReadStats::default();
+        let recovered = read_committed_with_stats(&[first, second], 4096, Some(&mut stats))
+            .await
+            .expect("read committed snapshot across torn successor")
+            .expect("previous committed anchor");
+
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(stats.file_reads, 5);
+        assert_eq!(stats.bytes_read, (MANIFEST_LEN * 2) + (old.len() * 2) + damaged_manifest.len());
+        assert_eq!(stats.peak_file_bytes, old.len().max(next.len()).max(MANIFEST_LEN));
+    }
+
     #[tokio::test]
     async fn manifest_cas_publication_transitions_from_legacy_without_losing_anchor() {
         let root = TempDir::new().expect("test directory");
@@ -1273,7 +1433,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
+    async fn committed_cleanup_removes_complete_checkpoint_at_or_below_sequence() {
         let root = TempDir::new().expect("test directory");
         let disk = disk(&root, "disk").await;
         let owner = Uuid::new_v4();
@@ -1283,7 +1443,7 @@ mod tests {
         commit(&disk, 1, owner, 4, &newer).await;
 
         assert!(
-            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), 3, 4096)
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096)
                 .await
                 .expect("delete old manifest"),
             "old committed manifest should be removed"
@@ -1295,12 +1455,19 @@ mod tests {
             ),
             "old manifest is gone, so the old payload cannot become visible again"
         );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "old payload should be reclaimed after its manifest is removed"
+        );
         assert_eq!(
-            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
                 .await
-                .expect("old payload orphan may remain")
+                .expect("newer payload retained")
                 .as_ref(),
-            older
+            newer
         );
         let recovered = read_committed(std::slice::from_ref(&disk), 4096)
             .await
@@ -1310,6 +1477,88 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_retains_manifest_when_payload_identity_mismatches() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let declared = payload("declared");
+        let actual = payload("actual");
+        install(&disk, MANIFEST_PATHS[0], &manifest(owner, 3, &declared)).await;
+        install(&disk, PAYLOAD_PATHS[0], &actual).await;
+
+        assert!(
+            matches!(
+                delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096).await,
+                Err(SnapshotError::Corrupt)
+            ),
+            "cleanup must fail closed when the committed body no longer matches its manifest"
+        );
+
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("mismatched manifest retained")
+                .as_ref(),
+            manifest(owner, 3, &declared)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+                .await
+                .expect("mismatched payload retained")
+                .as_ref(),
+            actual
+        );
+    }
+
+    #[tokio::test]
+    async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let replay_owner = Uuid::new_v4();
+        let other_owner = Uuid::new_v4();
+        let replay_payload = payload("replay-owner");
+        let other_payload = payload("other-owner");
+        commit(&disk, 0, replay_owner, 9, &replay_payload).await;
+        commit(&disk, 1, other_owner, 4, &other_payload).await;
+
+        assert!(
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), replay_owner, 9, 4096)
+                .await
+                .expect("delete replay-owner manifest"),
+            "the matched owner manifest should be removed"
+        );
+
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's manifest is reclaimed"
+        );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's payload is reclaimed"
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                .await
+                .expect("other owner manifest retained")
+                .as_ref(),
+            manifest(other_owner, 4, &other_payload)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
+                .await
+                .expect("other owner payload retained")
+                .as_ref(),
+            other_payload
+        );
+    }
+
     #[tokio::test]
     async fn legacy_import_requires_complete_consistent_replicas() {
         let root = TempDir::new().expect("test directory");

crates/heal/src/heal/task/heal_bucket.rs

@@ -328,18 +328,21 @@ impl HealTask {
             }
         }
 
-        let metadata_opts = HealOpts {
-            dry_run: self.options.dry_run,
-            scan_mode: self.options.scan_mode,
-            pool: self.options.pool_index,
-            set: self.options.set_index,
-            ..Default::default()
-        };
-        for result in self
-            .await_with_control(self.storage.heal_pool_metadata(&metadata_opts))
-            .await?
-        {
-            self.record_result_item(result).await;
+        if !self.options.dry_run {
+            let metadata_opts = HealOpts {
+                dry_run: self.options.dry_run,
+                recreate: self.options.recreate_missing,
+                scan_mode: self.options.scan_mode,
+                pool: self.options.pool_index,
+                set: self.options.set_index,
+                ..Default::default()
+            };
+            for result in self
+                .await_with_control(self.storage.heal_pool_metadata(&metadata_opts))
+                .await?
+            {
+                self.record_result_item(result).await;
+            }
         }
 
         if failed > 0 {

crates/heal/src/heal/task/heal_erasure_set.rs

@@ -451,6 +451,11 @@ impl HealTask {
             self.source,
         )
         .with_replacement_targets(replacement_targets, is_auto_replacement.then(|| self.id.clone()))
+        .with_pool_metadata_targets(if self.options.recreate_missing && !self.options.dry_run {
+            self.heal_endpoints.clone()
+        } else {
+            Vec::new()
+        })
         .with_replacement_identity_fence(replacement_target_identities.clone())
         .with_mainline_pacer(self.mainline_pacer.clone());
 

crates/heal/src/heal/task/tests.rs

@@ -14,6 +14,7 @@
 
 use super::super::{DiskOption, DiskStore, Endpoint, new_disk};
 use super::*;
+use crate::heal::POOL_META_NAME;
 use crate::heal::storage::HealStorageObjectResult;
 
 mod deferred_retry;
@@ -42,6 +43,7 @@ mod canonical_outcome {
     #[tokio::test(start_paused = true)]
     async fn cluster_retries_only_the_failed_listing_page() {
         let storage = Arc::new(MockStorage {
+            pool_metadata_required: true,
             recoverable_second_page_failures: Mutex::new(Some(1)),
             ..Default::default()
         });
@@ -66,7 +68,7 @@ mod canonical_outcome {
         assert_eq!(task.get_progress().await.objects_scanned, 2);
         assert_eq!(
             storage.heal_object_calls.lock().expect("object calls").as_slice(),
-            ["object-a", "object-b"]
+            ["object-a", "object-b", POOL_META_NAME]
         );
         assert_eq!(
             storage.listing_tokens.lock().expect("listing tokens").as_slice(),
@@ -2784,7 +2786,6 @@ async fn root_heal_pool_metadata_does_not_inherit_remove_or_no_lock() {
             HealOptions {
                 remove_corrupted: true,
                 no_lock: true,
-                dry_run: true,
                 ..Default::default()
             },
             HealPriority::Normal,
@@ -2794,11 +2795,11 @@ async fn root_heal_pool_metadata_does_not_inherit_remove_or_no_lock() {
 
     task.execute()
         .await
-        .expect("dry-run metadata inspection should be fenced and non-destructive");
+        .expect("metadata repair should retain its write fence and reject destructive options");
 
     let opts = storage.object_heal_opts.lock().expect("metadata options");
     assert_eq!(opts.len(), 1, "an empty user namespace must still inspect metadata");
-    assert!(opts[0].dry_run);
+    assert!(!opts[0].dry_run);
     assert!(!opts[0].remove);
     assert!(!opts[0].no_lock);
 }
@@ -2891,7 +2892,10 @@ async fn root_heal_pool_metadata_obeys_task_timeout() {
 
 #[tokio::test]
 async fn test_cluster_heal_visits_bucket_objects() {
-    let storage = Arc::new(MockStorage::default());
+    let storage = Arc::new(MockStorage {
+        pool_metadata_required: true,
+        ..Default::default()
+    });
     let request = HealRequest::new(
         HealType::Cluster,
         HealOptions {
@@ -2907,11 +2911,104 @@ async fn test_cluster_heal_visits_bucket_objects() {
 
     assert_eq!(
         storage.healed_objects.lock().unwrap().as_slice(),
-        ["object-a".to_string(), "object-b".to_string()]
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
     );
     assert!(matches!(task.get_status().await, HealTaskStatus::Completed));
 }
 
+#[tokio::test]
+async fn cluster_recreate_heals_pool_metadata_after_user_buckets() {
+    let storage = Arc::new(MockStorage {
+        pool_metadata_required: true,
+        ..Default::default()
+    });
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    task.execute()
+        .await
+        .expect("cluster recreate heal should include pool metadata");
+
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
+    );
+    let opts = storage.object_heal_opts.lock().expect("object opts");
+    assert!(opts.last().expect("pool metadata opts").recreate);
+}
+
+#[tokio::test]
+async fn cluster_recreate_fails_when_pool_metadata_heal_fails() {
+    let storage = Arc::new(MockStorage {
+        pool_metadata_required: true,
+        ..Default::default()
+    });
+    storage.heal_object_outcomes.lock().expect("object outcomes").insert(
+        POOL_META_NAME.to_string(),
+        VecDeque::from([MockHealObjectOutcome::ErrOther("pool metadata missing")]),
+    );
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    let err = task
+        .execute()
+        .await
+        .expect_err("cluster recreate heal must not hide pool metadata failure");
+
+    assert!(matches!(err, Error::Other(message) if message == "pool metadata missing"));
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
+    );
+}
+
+#[tokio::test]
+async fn cluster_dry_run_does_not_heal_pool_metadata() {
+    let storage = Arc::new(MockStorage {
+        pool_metadata_required: true,
+        ..Default::default()
+    });
+    let request = HealRequest::new(
+        HealType::Cluster,
+        HealOptions {
+            recursive: true,
+            dry_run: true,
+            recreate_missing: true,
+            timeout: None,
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage.clone());
+
+    task.execute()
+        .await
+        .expect("dry-run cluster heal should preserve existing coverage");
+
+    assert_eq!(
+        storage.heal_object_calls.lock().expect("object calls").as_slice(),
+        ["object-a".to_string(), "object-b".to_string()]
+    );
+}
+
 #[tokio::test]
 async fn object_heal_skips_dangling_delete_grace_without_failing_task() {
     let storage = Arc::new(MockStorage {

crates/heal/tests/mrf_pipeline_test.rs

@@ -51,6 +51,8 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
 const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin";
 const COMMITTED_PAYLOAD_REL: &str = ".heal-mrf-snapshot.0.bin";
 const COMMITTED_MANIFEST_REL: &str = ".heal-mrf-commit.0.bin";
+const COMMITTED_PAYLOAD_RELS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
+const COMMITTED_MANIFEST_RELS: [&str; 2] = [".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"];
 const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01";
 const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32;
 
@@ -235,6 +237,44 @@ fn journal_matches_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, exp
         .all(|path| std::fs::read(path.join(META_BUCKET).join(relative_path)).is_ok_and(|actual| actual == expected))
 }
 
+fn committed_checkpoint_matches_on_all_disks(disk_paths: &[PathBuf], sequence: u64, expected_payload: &[u8]) -> bool {
+    disk_paths.iter().all(|path| {
+        let root = path.join(META_BUCKET);
+        COMMITTED_PAYLOAD_RELS
+            .into_iter()
+            .zip(COMMITTED_MANIFEST_RELS)
+            .any(|(payload_rel, manifest_rel)| {
+                let Ok(payload) = std::fs::read(root.join(payload_rel)) else {
+                    return false;
+                };
+                if payload != expected_payload {
+                    return false;
+                }
+                let Ok(manifest) = std::fs::read(root.join(manifest_rel)) else {
+                    return false;
+                };
+                if manifest.len() != COMMITTED_MANIFEST_LEN || &manifest[..8] != COMMITTED_MAGIC || manifest[8] != 1 {
+                    return false;
+                }
+                let Ok(recorded_sequence) = <[u8; 8]>::try_from(&manifest[25..33]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(recorded_len) = <[u8; 8]>::try_from(&manifest[33..41]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(expected_len) = u64::try_from(expected_payload.len()) else {
+                    return false;
+                };
+                if recorded_sequence != sequence || recorded_len != expected_len {
+                    return false;
+                }
+                let payload_digest: [u8; 32] = Sha256::digest(expected_payload).into();
+                let manifest_digest: [u8; 32] = Sha256::digest(&manifest[..COMMITTED_MANIFEST_LEN - 32]).into();
+                payload_digest.as_slice() == &manifest[41..73] && manifest_digest.as_slice() == &manifest[73..]
+            })
+    })
+}
+
 async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
 where
     F: FnMut() -> Fut,
@@ -630,13 +670,14 @@ fn mrf_successor_flush_child_process_fixture() {
         expected_successor.extend(journal_record(1, "successor-bucket", "first-object", None, 0));
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before the delete phase"
+            "child process must publish the committed pending successor before the delete phase"
         );
     });
     std::process::exit(78);
@@ -678,13 +719,14 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
         expected_successor.extend(journal_record(1, "service-kill-bucket", "first-object", None, 0));
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before it can be killed"
+            "child process must publish the committed pending successor before it can be killed"
         );
         std::fs::write(&ready_path, b"ready").expect("write ready marker");
         loop {

docs/testing/ci-gates.md

@@ -289,9 +289,13 @@ measured durations, P3's pressure run needs at least two hours, and P1 needs a
 symbolized profile summary with resolved samples. Every G14 field and every
 performance gate's fields must also share one `measurement_window_id`, so EC8+4,
 multi-set/multi-pool, ABBA, throughput, and profiling artifacts cannot be
-stitched together from unrelated runs. Missing, synthetic, stale, tampered,
-undersized, or topology-mismatched evidence returns a compact blocked or invalid
-JSON result and a nonzero exit.
+stitched together from unrelated runs. P1 `profile_evidence` must bind every
+required profile artifact kind (`allocation-profile`, `flamegraph`,
+`rss-samples`, and `save-frequency`) with a relative path, artifact format,
+non-empty file, matching SHA256, and the same measurement window when a
+per-artifact window is declared. Missing, synthetic, stale, tampered, undersized,
+or topology-mismatched evidence returns a compact blocked or invalid JSON result
+and a nonzero exit.
 
 This command validates the evidence package; it does not create evidence. A
 handwritten JSON file, a synthetic harness pass, a single focused case, or a

scripts/check_test_wiring.py

@@ -23,7 +23,16 @@ try:
 except ModuleNotFoundError:
     import tomli as tomllib
 
-from scanner_abba import MAX_JSON_BYTES, digest, number, read_json, require, sha, write_json
+from scanner_abba import (
+    MAX_JSON_BYTES,
+    RELEASE_PROFILE_ARTIFACTS,
+    digest,
+    number,
+    read_json,
+    require,
+    sha,
+    write_json,
+)
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -68,6 +77,18 @@ SCANNER_HEAL_RELEASE_BUNDLE_REQUIRED_EVIDENCE_FIELDS = {
     "R-D": ("manager_disposition_evidence", "event_disposition_evidence", "ledger_disposition_evidence", "grace_handling"),
     "R-L": ("legacy_source_conflict_evidence", "migration_gap_evidence", "crash_safe_source_retirement_evidence"),
 }
+SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES = {
+    ("G03", "durable_root_publication_proof"): "durable-root-publication",
+    ("G03", "scoped_ack_request_identity"): "scoped-ack-request",
+    ("G03", "participating_peer_capability_snapshot"): "peer-capability-snapshot",
+    ("G03", "mixed_peer_ack_fallback_oracle"): "mixed-peer-ack-fallback",
+    ("G09", "mixed_version_reader_evidence"): "mixed-version-reader",
+    ("G09", "mixed_version_writer_evidence"): "mixed-version-writer",
+    ("G09", "rollback_payload_evidence"): "rollback-payload",
+    ("R-L", "legacy_source_conflict_evidence"): "legacy-source-conflict",
+    ("R-L", "migration_gap_evidence"): "migration-gap",
+    ("R-L", "crash_safe_source_retirement_evidence"): "crash-safe-source-retirement",
+}
 SCHEDULED_ALERT_WORKFLOWS = tuple(
     item["workflow"]
     for item in json.loads((ROOT / ".github/scheduled-validations.json").read_text())
@@ -956,6 +977,9 @@ def scanner_heal_oracle_names(root: Path) -> tuple[str, ...]:
     names = set()
     for case_id, requirement in cases.items():
         require(isinstance(case_id, str) and case_id, "invalid scanner/heal case identity")
+        lane = requirement.get("lane")
+        require(isinstance(lane, str) and re.fullmatch(r"[a-z0-9-]+", lane) is not None,
+                f"invalid nextest profile lane for {case_id}")
         oracle = requirement.get("oracle")
         require(isinstance(oracle, str) and oracle.endswith(".json"), f"invalid oracle for {case_id}")
         path = Path(oracle)
@@ -1338,8 +1362,14 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
     if gate in ("G03", "G09", "R-L"):
         versions = evidence.get("versions")
         require(isinstance(versions, list) and
-                len({version for version in versions if isinstance(version, str) and version.strip()}) >= 2,
+                len(set(versions)) >= 2 and
+                all(isinstance(version, str) and re.fullmatch(r"[0-9a-f]{40}", version) is not None
+                    for version in versions),
                 f"{gate}.{field} requires mixed-version evidence")
+        require(source_revision in versions, f"{gate}.{field} versions omit tested source revision")
+        expected_role = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
+        require(evidence.get("mixed_version_role") == expected_role,
+                f"{gate}.{field} mixed-version role must be {expected_role}")
     if gate in ("G04", "G07", "R-E", "R-L"):
         crash_points = evidence.get("crash_points")
         require(isinstance(crash_points, list) and crash_points,
@@ -1359,6 +1389,30 @@ def validate_release_bundle_artifact(bundle_path: Path, source_revision: str, ga
             evidence_integer(evidence.get("pools"), "G14 multi_pool_evidence.pools", 2, 1024)
     if field == "profile_evidence":
         evidence_integer(evidence.get("resolved_samples"), f"{gate}.{field}.resolved_samples", 1, 2**63 - 1)
+        profile_artifacts = evidence.get("profile_artifacts")
+        require(isinstance(profile_artifacts, dict), f"{gate}.{field} missing profile artifacts")
+        missing_artifacts = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(profile_artifacts))
+        require(not missing_artifacts,
+                f"{gate}.{field} missing profile artifacts: {', '.join(missing_artifacts)}")
+        unknown_artifacts = sorted(set(profile_artifacts) - set(RELEASE_PROFILE_ARTIFACTS))
+        require(not unknown_artifacts,
+                f"{gate}.{field} unknown profile artifacts: {', '.join(unknown_artifacts)}")
+        for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
+            item = profile_artifacts[artifact_kind]
+            require(isinstance(item, dict), f"{gate}.{field}.{artifact_kind} must be an object")
+            artifact_field = f"{field}.{artifact_kind}"
+            artifact_path = release_bundle_artifact_path(bundle_path, item.get("artifact"), gate, artifact_field)
+            require(artifact_path.stat().st_size > 0, f"{gate}.{artifact_field} artifact is empty")
+            require(sha(item.get("sha256")) and digest(artifact_path) == item["sha256"],
+                    f"{gate}.{artifact_field} artifact hash mismatch")
+            evidence_string(item.get("artifact_format"), f"{gate}.{artifact_field}.artifact_format",
+                            r"[A-Za-z0-9][A-Za-z0-9._+:-]{1,63}")
+            if "measurement_window_id" in item:
+                require(item["measurement_window_id"] == window_id,
+                        f"{gate}.{artifact_field} measurement window mismatch")
+            if "resolved_samples" in item:
+                evidence_integer(item.get("resolved_samples"), f"{gate}.{artifact_field}.resolved_samples",
+                                 0, 2**63 - 1)
     return window_id
 
 
@@ -1667,6 +1721,16 @@ class SelfTests(unittest.TestCase):
         finish_scanner_heal_receipt(run_dir, 0, root)
         return root, run_dir
 
+    def test_scanner_heal_case_lane_is_required_for_runner_profile(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, _ = self.scanner_heal_fixture(Path(tmp))
+            registry_path = root / ".config/scanner-heal-required-tests.json"
+            registry = read_json(registry_path)
+            del registry["cases"]["ec84-target-drive-restart"]["lane"]
+            write_json(registry_path, registry)
+            with self.assertRaisesRegex(ValueError, "invalid nextest profile lane for ec84-target-drive-restart"):
+                scanner_heal_oracle_names(root)
+
     def scanner_heal_release_bundle_fixture(self, directory: Path) -> tuple[Path, Path]:
         """Parser fixtures only; the bundle is not runtime evidence."""
         root, _ = self.scanner_heal_fixture(directory)
@@ -1704,7 +1768,8 @@ class SelfTests(unittest.TestCase):
                     evidence["duration_seconds"] = duration
                 evidence["finished_at"] = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z")
                 if gate in ("G03", "G09", "R-L"):
-                    evidence["versions"] = ["previous", "candidate"]
+                    evidence["versions"] = ["a" * 40, source_revision]
+                    evidence["mixed_version_role"] = SCANNER_HEAL_RELEASE_MIXED_VERSION_ROLES[(gate, field)]
                 if gate in ("G04", "G07", "R-E", "R-L"):
                     evidence["crash_points"] = ["before-commit"]
                 if gate == "G14" and field == "ec8_4_evidence":
@@ -1715,6 +1780,16 @@ class SelfTests(unittest.TestCase):
                     evidence["pools"] = 2
                 if field == "profile_evidence":
                     evidence["resolved_samples"] = 1
+                    artifacts = {}
+                    for artifact_kind in RELEASE_PROFILE_ARTIFACTS:
+                        artifact = artifact_dir / f"{gate}-{field}-{artifact_kind}.json"
+                        write_json(artifact, {"gate": gate, "field": field, "artifact": artifact_kind})
+                        artifacts[artifact_kind] = {
+                            "artifact": artifact.relative_to(bundle_dir).as_posix(),
+                            "sha256": digest(artifact),
+                            "artifact_format": "json",
+                        }
+                    evidence["profile_artifacts"] = artifacts
                 fields[field] = evidence
             gates[gate] = {
                 "status": "pass",
@@ -1768,7 +1843,31 @@ class SelfTests(unittest.TestCase):
             ("missing-duration", "P1", "cold_walk_share_measurement", lambda item: item.pop("duration_seconds"), "duration_seconds"),
             ("duration", "P3", "two_hour_pressure_measurement", lambda item: item.update({"duration_seconds": 7199}), "two hours"),
             ("profile", "P1", "profile_evidence", lambda item: item.pop("resolved_samples"), "resolved_samples"),
+            (
+                "profile-artifact",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"].pop("flamegraph"),
+                "missing profile artifacts",
+            ),
+            (
+                "profile-artifact-hash",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"]["rss-samples"].update({"sha256": "0" * 64}),
+                "artifact hash mismatch",
+            ),
+            (
+                "profile-artifact-window",
+                "P1",
+                "profile_evidence",
+                lambda item: item["profile_artifacts"]["save-frequency"].update(
+                    {"measurement_window_id": "p1-different-window"}
+                ),
+                "measurement window mismatch",
+            ),
             ("versions", "G09", "mixed_version_reader_evidence", lambda item: item.update({"versions": [1, 2]}), "mixed-version"),
+            ("stale-versions", "G09", "mixed_version_writer_evidence", lambda item: item.update({"versions": ["a" * 40, "c" * 40]}), "tested source revision"),
         ):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as tmp:
                 root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
@@ -1782,6 +1881,26 @@ class SelfTests(unittest.TestCase):
                 self.assertFalse(status["release_approved"])
                 self.assertTrue(any(expected in error for error in status["rejected_gates"][gate]))
 
+    def test_scanner_heal_release_bundle_requires_mixed_version_field_roles(self) -> None:
+        for gate, field, wrong_role in (
+            ("G03", "mixed_peer_ack_fallback_oracle", "mixed-version-reader"),
+            ("G09", "mixed_version_reader_evidence", "mixed-version-writer"),
+            ("G09", "mixed_version_writer_evidence", "mixed-version-reader"),
+            ("G09", "rollback_payload_evidence", "mixed-version-reader"),
+            ("R-L", "crash_safe_source_retirement_evidence", "migration-gap"),
+        ):
+            with self.subTest(gate=gate, field=field), tempfile.TemporaryDirectory() as tmp:
+                root, bundle = self.scanner_heal_release_bundle_fixture(Path(tmp))
+                data = read_json(bundle)
+                data["gates"][gate]["evidence_fields"][field]["mixed_version_role"] = wrong_role
+                write_json(bundle, data)
+
+                with mock.patch("subprocess.check_output", return_value="b" * 40):
+                    status = scanner_heal_release_bundle_status(root, bundle)
+                self.assertEqual(status["decision"], "blocked")
+                self.assertFalse(status["release_approved"])
+                self.assertTrue(any("mixed-version role" in error for error in status["rejected_gates"][gate]))
+
     def test_scanner_heal_release_bundle_requires_field_provenance(self) -> None:
         for fault, mutation, expected in (
             ("source", lambda item: item.update({"source_revision": "c" * 40}), "source revision mismatch"),

scripts/run_scanner_heal_evidence_case.sh

@@ -4,7 +4,7 @@ set -euo pipefail
 
 ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 PYTHON_BIN="${RUSTFS_PYTHON_BIN:-python3}"
-PROFILE="e2e-nightly"
+PROFILE=""
 CASE_ID="background-target-crash"
 RUN_DIR=""
 PLAN_ONLY=0
@@ -18,7 +18,7 @@ then validate the produced receipt, nextest listing, JUnit, and case oracle.
 
 Options:
   --case CASE          Registry case to run (default: background-target-crash)
-  --profile PROFILE   Nextest profile to use (default: e2e-nightly)
+  --profile PROFILE   Nextest profile to use (default: registry lane)
   --run-dir DIR       New evidence directory (default: target/scanner-heal-evidence/CASE-TIMESTAMP)
   --plan-only         Validate registry selection and print the exact filter without running cargo
   --self-test         Run lightweight CLI/registry checks without building Rust
@@ -31,6 +31,18 @@ port allocator when the default 20000..30000 test range is unavailable.
 USAGE
 }
 
+case_ids() {
+    "$PYTHON_BIN" - "$ROOT/.config/scanner-heal-required-tests.json" <<'PY'
+import json
+import pathlib
+import sys
+
+registry = json.loads(pathlib.Path(sys.argv[1]).read_text())
+for case_id in sorted(registry["cases"]):
+    print(case_id)
+PY
+}
+
 case_field() {
     local case_id="$1"
     local field="$2"
@@ -109,20 +121,24 @@ PY
 }
 
 run_self_test() {
-    local filter
-    filter="$(test_filter_for background-target-crash)"
-    case "$filter" in
-        *background_target_crash*) ;;
-        *)
-            echo "self-test failed: crash case filter missing" >&2
-            return 1
-            ;;
-    esac
     if "$0" --case release --plan-only >/dev/null 2>&1; then
         echo "self-test failed: release pseudo-case must not be runnable" >&2
         return 1
     fi
-    "$0" --case background-target-crash --plan-only >/dev/null
+    local case_id
+    while IFS= read -r case_id; do
+        local expected_filter expected_profile plan
+        expected_filter="$(test_filter_for "$case_id")"
+        expected_profile="$(case_field "$case_id" lane)"
+        plan="$("$0" --case "$case_id" --plan-only)"
+        if [[ "$plan" != *"case=$case_id"* ]] ||
+            [[ "$plan" != *"profile=$expected_profile"* ]] ||
+            [[ "$plan" != *"filter=$expected_filter"* ]] ||
+            [[ "$plan" != *"run_dir=$ROOT/target/scanner-heal-evidence/$case_id-"* ]]; then
+            echo "self-test failed: registry case plan mismatch for $case_id" >&2
+            return 1
+        fi
+    done < <(case_ids)
 }
 
 while [[ $# -gt 0 ]]; do
@@ -166,6 +182,9 @@ fi
 
 case_field "$CASE_ID" name >/dev/null
 TEST_FILTER="$(test_filter_for "$CASE_ID")"
+if [[ -z "$PROFILE" ]]; then
+    PROFILE="$(case_field "$CASE_ID" lane)"
+fi
 case "$CASE_ID" in
   background-target-crash|background-target-restart)
     export RUSTFS_HEAL_CHAOS_OBJECT_COUNT="${RUSTFS_HEAL_CHAOS_OBJECT_COUNT:-64}"

scripts/scanner_abba.py

@@ -173,6 +173,18 @@ def release_evidence_true(value, name):
     require(value is True, f"missing release_evidence.{name}")
 
 
+def release_evidence_exact_strings(value, expected, name):
+    require(isinstance(value, list) and all(isinstance(item, str) and item.strip() for item in value),
+            f"invalid release_evidence.{name}")
+    observed = set(value)
+    require(len(observed) == len(value), f"duplicate release_evidence.{name}")
+    missing = sorted(set(expected) - observed)
+    require(not missing, f"missing release_evidence.{name}: {', '.join(missing)}")
+    unknown = sorted(observed - set(expected))
+    require(not unknown, f"unknown release_evidence.{name}: {', '.join(unknown)}")
+    return value
+
+
 def validate_release_evidence_manifest(manifest):
     if manifest["evidence"] != "measured":
         return
@@ -210,17 +222,17 @@ def validate_release_evidence_manifest(manifest):
 
     crash = evidence.get("crash_restart")
     require(isinstance(crash, dict), "missing release_evidence.crash_restart")
-    fault_modes = crash.get("fault_modes")
-    require(
-        isinstance(fault_modes, list)
-        and all(mode in fault_modes for mode in RELEASE_FAULT_MODES)
-        and all(isinstance(mode, str) and mode.strip() for mode in fault_modes),
-        "missing release_evidence.crash_restart.fault_modes",
-    )
+    release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "crash_restart.fault_modes")
     release_evidence_true(crash.get("unclean_shutdown_marker"), "crash_restart.unclean_shutdown_marker")
 
     mixed = evidence.get("mixed_version")
     require(isinstance(mixed, dict), "missing release_evidence.mixed_version")
+    baseline_revision = manifest["baseline"]["revision"]
+    candidate_revision = manifest["candidate"]["revision"]
+    require(baseline_revision != candidate_revision,
+            "release_evidence.mixed_version requires distinct baseline and candidate revisions")
+    require(manifest["baseline"]["sha256"] != manifest["candidate"]["sha256"],
+            "release_evidence.mixed_version requires distinct baseline and candidate binaries")
     revisions = mixed.get("participating_revisions")
     require(
         isinstance(revisions, list)
@@ -229,20 +241,15 @@ def validate_release_evidence_manifest(manifest):
                 for revision in revisions),
         "invalid release_evidence.mixed_version.participating_revisions",
     )
-    for revision in (manifest["baseline"]["revision"], manifest["candidate"]["revision"]):
+    for revision in (baseline_revision, candidate_revision):
         require(revision in revisions, "release_evidence.mixed_version omits tested build revision")
     for key in ("reader", "writer", "rollback_payload"):
         require(mixed.get(key) is True, f"missing release_evidence.mixed_version.{key}")
 
     profile = evidence.get("profile")
     require(isinstance(profile, dict), "missing release_evidence.profile")
-    artifacts = profile.get("required_artifacts")
-    require(
-        isinstance(artifacts, list)
-        and all(item in artifacts for item in RELEASE_PROFILE_ARTIFACTS)
-        and all(isinstance(item, str) and item.strip() for item in artifacts),
-        "missing release_evidence.profile.required_artifacts",
-    )
+    release_evidence_exact_strings(profile.get("required_artifacts"), RELEASE_PROFILE_ARTIFACTS,
+                                   "profile.required_artifacts")
     for key in ("collector_config_sha256", "profiler_config_sha256"):
         require(sha(profile.get(key)), f"invalid release_evidence.profile.{key}")
 

scripts/test_scanner_abba.py

@@ -168,7 +168,14 @@ class ScannerAbbaTest(unittest.TestCase):
     def measured_manifest(self):
         manifest = copy.deepcopy(self.manifest)
         manifest.update(evidence="measured", duration_seconds=900)
-        manifest["candidate"]["revision"] = "b" * 40
+        candidate_binary = self.root / "candidate-python"
+        candidate_binary.write_bytes(self.binary.read_bytes() + b"\n")
+        candidate_binary.chmod(0o755)
+        manifest["candidate"] = {
+            "binary": str(candidate_binary),
+            "sha256": harness.digest(candidate_binary),
+            "revision": "b" * 40,
+        }
         manifest["release_evidence"] = {
             "topology": {
                 "nodes": 3,
@@ -548,6 +555,12 @@ class ScannerAbbaTest(unittest.TestCase):
             "missing crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
                 fault_modes=["process-restart"],
             ),
+            "unknown crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                fault_modes=["process-restart", "process-crash-restart", "kernel-panic"],
+            ),
+            "duplicate crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                fault_modes=["process-restart", "process-restart", "process-crash-restart"],
+            ),
             "clean crash marker": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
                 unclean_shutdown_marker=False,
             ),
@@ -555,9 +568,24 @@ class ScannerAbbaTest(unittest.TestCase):
             "missing candidate": lambda manifest: manifest["release_evidence"]["mixed_version"].update(
                 participating_revisions=["a" * 40, "c" * 40],
             ),
+            "same mixed revision": lambda manifest: manifest["candidate"].update(
+                revision=manifest["baseline"]["revision"],
+            ),
+            "same mixed binary": lambda manifest: manifest["candidate"].update(
+                binary=manifest["baseline"]["binary"],
+                sha256=manifest["baseline"]["sha256"],
+            ),
             "missing profile": lambda manifest: manifest["release_evidence"]["profile"].update(
                 required_artifacts=["allocation-profile", "flamegraph", "rss-samples"],
             ),
+            "unknown profile": lambda manifest: manifest["release_evidence"]["profile"].update(
+                required_artifacts=["allocation-profile", "flamegraph", "rss-samples", "save-frequency", "heapdump"],
+            ),
+            "duplicate profile": lambda manifest: manifest["release_evidence"]["profile"].update(
+                required_artifacts=[
+                    "allocation-profile", "flamegraph", "rss-samples", "save-frequency", "flamegraph",
+                ],
+            ),
             "bad profile hash": lambda manifest: manifest["release_evidence"]["profile"].update(
                 profiler_config_sha256="not-a-sha",
             ),