test(scanner): require exact ABBA release evidence sets

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

.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/e2e_test/src/distributed/heal_test.rs

@@ -12,12 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::harness::{DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_object, unique_bucket, wait_until};
+use super::harness::{
+    DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket,
+    wait_until,
+};
 use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post};
 use crate::common::init_logging;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
+use serde_json::Value;
+use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashSet};
+use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -25,6 +31,8 @@ const EC84_NODE_COUNT: usize = 3;
 const EC84_DRIVES_PER_NODE: usize = 4;
 const EC84_DATA_BLOCKS: usize = 8;
 const EC84_PARITY_BLOCKS: usize = 4;
+const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
+const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
 
 #[derive(Clone)]
 struct ExpectedShard {
@@ -33,6 +41,79 @@ struct ExpectedShard {
     baseline: VersionShardCensus,
 }
 
+struct ScannerHealEvidenceContext {
+    directory: PathBuf,
+    run: Value,
+}
+
+fn file_sha256(path: &Path) -> TestResult<String> {
+    let mut file = std::fs::File::open(path)?;
+    let mut digest = Sha256::new();
+    let mut buffer = [0_u8; 64 * 1024];
+    loop {
+        let read = file.read(&mut buffer)?;
+        if read == 0 {
+            break;
+        }
+        digest.update(&buffer[..read]);
+    }
+    Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect())
+}
+
+fn compiled_test_identity() -> Value {
+    serde_json::json!({
+        "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"),
+        "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false",
+        "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"),
+        "features": env!("RUSTFS_E2E_BUILD_FEATURES"),
+        "target": env!("RUSTFS_E2E_BUILD_TARGET"),
+        "profile": env!("RUSTFS_E2E_BUILD_PROFILE"),
+        "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"),
+    })
+}
+
+fn string_field<'a>(value: &'a Value, path: &str) -> TestResult<&'a str> {
+    let mut current = value;
+    for segment in path.split('.') {
+        current = current
+            .get(segment)
+            .ok_or_else(|| format!("scanner/heal run receipt missing {path}"))?;
+    }
+    current
+        .as_str()
+        .filter(|text| !text.is_empty())
+        .ok_or_else(|| format!("scanner/heal run receipt has invalid {path}").into())
+}
+
+fn scanner_heal_evidence_context() -> TestResult<Option<ScannerHealEvidenceContext>> {
+    let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else {
+        return Ok(None);
+    };
+    let directory = PathBuf::from(directory);
+    let receipt = directory.join("run.json");
+    if receipt.metadata()?.len() > 1024 * 1024 {
+        return Err("oversized scanner/heal execution receipt".into());
+    }
+    let run: Value = serde_json::from_slice(&std::fs::read(receipt)?)?;
+    let built = compiled_test_identity();
+    for key in ["source_revision", "dirty", "lock_blob", "features"] {
+        if built[key] != run["test_build"][key] {
+            return Err(format!("compiled test identity differs for {key}").into());
+        }
+    }
+    let binary_path = PathBuf::from(string_field(&run, "binary.path")?);
+    if file_sha256(&binary_path)? != string_field(&run, "binary.sha256")? {
+        return Err("server binary must match the run receipt".into());
+    }
+    if file_sha256(&std::env::current_exe()?)? != string_field(&run, "test_binary.sha256")? {
+        return Err("test executable must match the run receipt".into());
+    }
+    if directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE).exists() {
+        return Err("scanner/heal oracle already exists; create a new execution receipt".into());
+    }
+    Ok(Some(ScannerHealEvidenceContext { directory, run }))
+}
+
 fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     if census.data_blocks != Some(EC84_DATA_BLOCKS) || census.parity_blocks != Some(EC84_PARITY_BLOCKS) {
         return Err(format!("object {key} did not use EC8+4 geometry: {census:?}").into());
@@ -49,6 +130,76 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult {
     Ok(())
 }
 
+async fn write_scanner_heal_evidence(
+    context: ScannerHealEvidenceContext,
+    dist: &DistCluster,
+    bucket: &str,
+    expected: &[ExpectedShard],
+    outage_key: &str,
+    outage_body: &[u8],
+    replaced_drive: &Path,
+    pid_before: u32,
+    pid_after: u32,
+    node_listings: Vec<Vec<String>>,
+) -> TestResult {
+    let verifier = dist.client(0)?;
+    let mut objects = Vec::new();
+    for item in expected {
+        let actual = get_object_bytes(&verifier, bucket, &item.key).await?;
+        let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?;
+        objects.push(serde_json::json!({
+            "key": item.key,
+            "version_id": null,
+            "expected_bytes": item.body.len(),
+            "actual_bytes": actual.len(),
+            "expected_sha256": sha256_hex(&item.body),
+            "actual_sha256": sha256_hex(&actual),
+            "expected_physical": item.baseline,
+            "physical": physical,
+        }));
+    }
+    let actual = get_object_bytes(&verifier, bucket, outage_key).await?;
+    let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?;
+    objects.push(serde_json::json!({
+        "key": outage_key,
+        "version_id": null,
+        "expected_bytes": outage_body.len(),
+        "actual_bytes": actual.len(),
+        "expected_sha256": sha256_hex(outage_body),
+        "actual_sha256": sha256_hex(&actual),
+        "expected_physical": null,
+        "physical": physical,
+    }));
+
+    let evidence = serde_json::json!({
+        "schema": 1,
+        "case": EC84_TARGET_DRIVE_RESTART_CASE,
+        "evidence": "process-restart",
+        "run_id": string_field(&context.run, "run_id")?,
+        "source_revision": string_field(&context.run, "source_revision")?,
+        "test_build": compiled_test_identity(),
+        "binary_sha256": string_field(&context.run, "binary.sha256")?,
+        "test_binary_sha256": string_field(&context.run, "test_binary.sha256")?,
+        "topology": {"nodes": EC84_NODE_COUNT, "drives_per_node": EC84_DRIVES_PER_NODE},
+        "pid_before": pid_before,
+        "pid_after": pid_after,
+        "unclean_shutdown_marker": false,
+        "objects": objects,
+        "node_listings": node_listings,
+    });
+    let data = serde_json::to_vec(&evidence)?;
+    if data.len() > 1024 * 1024 {
+        return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into());
+    }
+    let mut output = std::fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(context.directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE))?;
+    output.write_all(&data)?;
+    output.sync_all()?;
+    Ok(())
+}
+
 fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> TestResult {
     for key in keys {
         let census = census_object_version_on_disk(drive, bucket, key, None)?;
@@ -87,6 +238,7 @@ async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<Ex
 #[tokio::test]
 async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_restart() -> TestResult {
     init_logging();
+    let evidence_context = scanner_heal_evidence_context()?;
     let mut dist = DistCluster::start_with_env(
         DistLayout::ThreeByFourEc84,
         &[
@@ -116,6 +268,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
 
     let format_path = replaced_drive.join(".rustfs.sys").join("format.json");
     let format_json = std::fs::read(&format_path)?;
+    let target_pid_before = dist.cluster.nodes[replaced_node]
+        .process
+        .as_ref()
+        .ok_or("target process is absent before graceful restart")?
+        .id();
     dist.cluster.stop_node_gracefully(replaced_node).await?;
     let retired_drive = PathBuf::from(format!("{}.retired", replaced_drive.display()));
     std::fs::rename(&replaced_drive, &retired_drive)?;
@@ -138,6 +295,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
         .await?;
 
     dist.cluster.start_node(replaced_node).await?;
+    let target_pid_after = dist.cluster.nodes[replaced_node]
+        .process
+        .as_ref()
+        .ok_or("target process is absent after restart")?
+        .id();
     let heal_body =
         r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
     let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
@@ -167,6 +329,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
         .chain(std::iter::once((outage_key.to_string(), outage_body.clone())))
         .collect::<BTreeMap<_, _>>();
     let expected_keys = inventory.keys().cloned().collect::<HashSet<_>>();
+    let mut node_listings = Vec::new();
     for node_index in 0..dist.cluster.nodes.len() {
         let client = dist.client(node_index)?;
         assert_inventory(&client, &bucket, &inventory).await?;
@@ -177,6 +340,24 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
             .filter_map(|object| object.key().map(str::to_owned))
             .collect::<HashSet<_>>();
         assert_eq!(observed, expected_keys, "node {node_index} listing diverged after EC8+4 heal");
+        let mut observed = observed.into_iter().collect::<Vec<_>>();
+        observed.sort();
+        node_listings.push(observed);
+    }
+    if let Some(context) = evidence_context {
+        write_scanner_heal_evidence(
+            context,
+            &dist,
+            &bucket,
+            &expected,
+            outage_key,
+            &outage_body,
+            &replaced_drive,
+            target_pid_before,
+            target_pid_after,
+            node_listings,
+        )
+        .await?;
     }
 
     Ok(())

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1567,7 +1567,10 @@ mod tests {
                 "Restored target endpoint forwarding"
             );
         } else {
-            if scenario == InterruptionScenario::BackgroundTargetRestart {
+            if matches!(
+                scenario,
+                InterruptionScenario::BackgroundTargetRestart | InterruptionScenario::BackgroundTargetRestartEc84
+            ) {
                 cluster.stop_node_gracefully(interruption_node).await?;
             } else {
                 cluster.stop_node(interruption_node)?;
@@ -1589,7 +1592,10 @@ mod tests {
             if background_enabled {
                 let marker_exists = unclean_shutdown_marker.is_file();
                 unclean_shutdown_marker_observed = Some(marker_exists);
-                let expected_marker = !matches!(scenario, InterruptionScenario::BackgroundTargetRestart);
+                let expected_marker = matches!(
+                    scenario,
+                    InterruptionScenario::BackgroundTargetCrash | InterruptionScenario::BackgroundTargetCrashEc84
+                );
                 assert!(
                     marker_exists == expected_marker,
                     "background restart/crash lane observed unexpected unclean-shutdown marker state"

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/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,10 +957,16 @@ impl ErasureSetHealer {
         resume_manager: &ResumeManager,
         checkpoint_manager: &CheckpointManager,
     ) -> Result<()> {
-        if self.replacement_task_id.is_none() {
+        let target_endpoints = if self.pool_metadata_target_endpoints.is_empty() {
+            self.target_endpoints.as_ref()
+        } else {
+            self.pool_metadata_target_endpoints.as_ref()
+        };
+        let target_scoped_recreate = !self.heal_opts.dry_run && self.heal_opts.recreate && !target_endpoints.is_empty();
+        if self.replacement_task_id.is_none() && !target_scoped_recreate {
             return Ok(());
         }
-        if self.target_endpoints.is_empty() {
+        if target_endpoints.is_empty() {
             return Err(Error::TaskExecutionFailed {
                 message: "Replacement pool metadata heal requires target endpoints".to_string(),
             });
@@ -978,17 +993,11 @@ impl ErasureSetHealer {
             .heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_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
-                    .replacement_targets_have_version(
-                        RUSTFS_META_BUCKET,
-                        POOL_META_NAME,
-                        None,
-                        &self.heal_opts,
-                        &self.target_endpoints,
-                    )
+                    .replacement_targets_have_version(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts, target_endpoints)
                     .await
                 {
                     Ok(true) => (object_size, Ok(())),
@@ -2762,6 +2771,68 @@ 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,
+                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));
+
+        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)]);
+    }
+
+    #[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

@@ -294,7 +294,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 {
@@ -718,7 +722,7 @@ fn replay_must_retain_journal(
 #[derive(Clone, Copy)]
 enum ReplayCleanup {
     Legacy,
-    Committed { sequence: u64 },
+    Committed { owner: Uuid, sequence: u64 },
 }
 
 struct ReplaySource {
@@ -731,6 +735,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(),
             },
         }));
@@ -754,18 +759,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
 }
@@ -1344,6 +1351,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

@@ -555,20 +555,25 @@ 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 manifests 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> {
+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> {
@@ -598,7 +603,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            if manifest.sequence > committed_through {
+            if manifest.owner != owner || manifest.sequence > committed_through {
                 continue;
             }
             match EcstoreDiskAPI::compare_and_update_file(
@@ -1200,6 +1205,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");
@@ -1283,7 +1365,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"
@@ -1310,6 +1392,47 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[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_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.rs

@@ -45,7 +45,7 @@ use tokio::sync::RwLock;
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
 
-use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
+use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, POOL_META_NAME, RUSTFS_META_BUCKET};
 
 #[cfg(test)]
 pub(crate) struct OutcomeFinishTestHook {

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

@@ -340,9 +340,88 @@ impl HealTask {
             return Err(self.record_batch_failure(failure).await);
         }
 
+        if self.options.recreate_missing && !self.options.dry_run {
+            self.heal_cluster_pool_metadata().await?;
+        }
+
         Ok(())
     }
 
+    async fn heal_cluster_pool_metadata(&self) -> Result<()> {
+        let heal_opts = HealOpts {
+            recursive: false,
+            dry_run: self.options.dry_run,
+            remove: false,
+            recreate: self.options.recreate_missing,
+            scan_mode: self.options.scan_mode,
+            update_parity: self.options.update_parity,
+            no_lock: self.options.no_lock,
+            read_repair: false,
+            pool: self.options.pool_index,
+            set: self.options.set_index,
+        };
+
+        let heal_result = self
+            .await_with_control(self.storage.heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &heal_opts))
+            .await;
+        match heal_result {
+            Ok((result, None)) => {
+                debug!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    drives_healed = result.drives_healed(),
+                    drives_total = result.drives_reported(),
+                    result = "pool_metadata_ok",
+                    "Heal cluster pool metadata repaired"
+                );
+                self.record_result_item(result).await;
+                Ok(())
+            }
+            Ok((result, Some(err))) => {
+                self.record_result_item(result).await;
+                warn!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    result = "pool_metadata_failed",
+                    error = %err,
+                    "Heal cluster pool metadata failed"
+                );
+                Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to heal cluster pool metadata: {err}"),
+                })
+            }
+            Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
+            Err(Error::TaskTimeout) => Err(Error::TaskTimeout),
+            Err(err) => {
+                warn!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_BUCKET_RESULT,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    task_id = %self.id,
+                    bucket = RUSTFS_META_BUCKET,
+                    object = POOL_META_NAME,
+                    result = "pool_metadata_failed",
+                    error = %err,
+                    "Heal cluster pool metadata failed"
+                );
+                Err(Error::TaskExecutionFailed {
+                    message: format!("Failed to heal cluster pool metadata: {err}"),
+                })
+            }
+        }
+    }
+
     pub(super) async fn heal_prefix(&self, bucket: &str, prefix: &str) -> Result<()> {
         debug!(
             target: "rustfs::heal::task",

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/heal_object.rs

@@ -162,11 +162,6 @@ impl HealTask {
             pool: self.options.pool_index,
             set: self.options.set_index,
         };
-        let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?;
-        let mut expected_identity =
-            self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
-        expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id;
-
         let mut expected_identity =
             self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
         expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?;

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

@@ -66,7 +66,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(),
@@ -2582,11 +2582,95 @@ 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::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::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::TaskExecutionFailed { .. }));
+    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::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 {

scripts/check_test_wiring.py

@@ -68,6 +68,24 @@ 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",
+}
+SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS = (
+    "allocation-profile",
+    "flamegraph",
+    "rss-samples",
+    "save-frequency",
+)
 SCHEDULED_ALERT_WORKFLOWS = tuple(
     item["workflow"]
     for item in json.loads((ROOT / ".github/scheduled-validations.json").read_text())
@@ -956,6 +974,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 +1359,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 +1386,15 @@ 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, list) and
+                len(set(profile_artifacts)) == len(profile_artifacts) and
+                all(isinstance(artifact, str) and re.fullmatch(r"[a-z0-9][a-z0-9-]{1,63}", artifact) is not None
+                    for artifact in profile_artifacts),
+                f"{gate}.{field} requires named profile artifacts")
+        missing_profile_artifacts = sorted(set(SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS) - set(profile_artifacts))
+        require(not missing_profile_artifacts,
+                f"{gate}.{field} missing profile artifacts: {', '.join(missing_profile_artifacts)}")
     return window_id
 
 
@@ -1667,6 +1703,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 +1750,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 +1762,7 @@ class SelfTests(unittest.TestCase):
                     evidence["pools"] = 2
                 if field == "profile_evidence":
                     evidence["resolved_samples"] = 1
+                    evidence["profile_artifacts"] = list(SCANNER_HEAL_RELEASE_PROFILE_ARTIFACTS)
                 fields[field] = evidence
             gates[gate] = {
                 "status": "pass",
@@ -1768,7 +1816,10 @@ 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.update({"profile_artifacts": ["allocation-profile", "rss-samples", "save-frequency"]}), "profile artifacts"),
+            ("duplicate-profile-artifact", "P1", "profile_evidence", lambda item: item.update({"profile_artifacts": ["allocation-profile", "allocation-profile", "flamegraph", "rss-samples", "save-frequency"]}), "named profile artifacts"),
             ("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 +1833,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,17 @@ def release_evidence_true(value, name):
     require(value is True, f"missing release_evidence.{name}")
 
 
+def release_evidence_exact_set(value, name, expected):
+    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}")
+    require(observed == set(expected), f"invalid release_evidence.{name}")
+
+
 def validate_release_evidence_manifest(manifest):
     if manifest["evidence"] != "measured":
         return
@@ -210,13 +221,7 @@ 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_set(crash.get("fault_modes"), "crash_restart.fault_modes", RELEASE_FAULT_MODES)
     release_evidence_true(crash.get("unclean_shutdown_marker"), "crash_restart.unclean_shutdown_marker")
 
     mixed = evidence.get("mixed_version")
@@ -236,13 +241,7 @@ def validate_release_evidence_manifest(manifest):
 
     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_set(profile.get("required_artifacts"), "profile.required_artifacts", RELEASE_PROFILE_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

@@ -548,6 +548,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", "power-cycle"],
+            ),
+            "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,
             ),
@@ -558,6 +564,12 @@ class ScannerAbbaTest(unittest.TestCase):
             "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", "heap-dump"],
+            ),
+            "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",
             ),