test(scanner): add W13 MRF evidence runner (#7547)

Add a W13 durable MRF evidence runner that emits measured G07, G08, and P4 JSON artifacts and release descriptors through the existing scanner/heal bundle gate.

The runner now executes the ignored MRF replay evidence test with an exact full test path, validates raw artifact kinds and gate decisions, prepares Linux tmpfs-backed ENOSPC roots for G08, and documents the Linux/long-soak boundaries.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

.config/make/tests.mak

@@ -32,6 +32,7 @@ script-tests: ## Run shell script tests
 	./scripts/test_hotpath_warp_ab_gate.sh
 	./scripts/test_hotpath_warp_abba.sh
 	./scripts/test_scanner_validation_harness.sh
+	./scripts/test_scanner_heal_w13_mrf_evidence.sh
 	./scripts/test_scanner_heal_w16_recovery_evidence.sh
 	./scripts/test_exact_1mib_handoff_abba.sh
 	./scripts/test_pinned_paired_abba_bench.sh

Cargo.lock

@@ -9928,6 +9928,7 @@ dependencies = [
  "async-trait",
  "base64-simd",
  "bytes",
+ "chrono",
  "crc-fast",
  "futures",
  "hotpath",

crates/heal/Cargo.toml

@@ -104,6 +104,7 @@ walkdir = { workspace = true }
 http = { workspace = true }
 temp-env = { workspace = true, features = ["async_closure"] }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
+chrono = { workspace = true }
 
 [lib]
 doctest = false

crates/heal/src/heal/mrf_queue.rs

@@ -1139,9 +1139,26 @@ mod tests {
     use super::*;
     use crate::heal::manager::HealConfig;
     use crate::heal::storage::{ECStoreHealStorage, HealStorageAPI};
+    use crate::heal::{DiskError, RUSTFS_META_BUCKET};
     use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
+    use serde_json::{Map, Value, json};
     use serial_test::serial;
+    use std::env;
+    use std::fs;
+    use std::io::Write as _;
+    use std::path::{Path, PathBuf};
     use std::sync::Arc as StdArc;
+    use std::time::{Duration as StdDuration, Instant};
+
+    const W13_EVIDENCE_DIR_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR";
+    const W13_SOURCE_REVISION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION";
+    const W13_SELECTION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SELECTION";
+    const W13_SOAK_SECONDS_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SOAK_SECONDS";
+    const W13_ALLOW_SHORT_SOAK_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ALLOW_SHORT_SOAK";
+    const W13_RUN_ID_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_RUN_ID";
+    const W13_WINDOW_ID_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_WINDOW_ID";
+    const W13_ENOSPC_ROOT_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT";
+    const W13_ENOSPC_FILL_LIMIT_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES";
 
     fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
         MrfIntent {
@@ -1162,6 +1179,726 @@ mod tests {
         payload
     }
 
+    fn w13_timestamp() -> String {
+        chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+    }
+
+    fn w13_selection_contains(selection: &str, lane: &str) -> bool {
+        selection == "all" || selection.split(',').any(|item| item.trim() == lane)
+    }
+
+    fn w13_evidence_path(root: &Path, gate: &str, field: &str) -> PathBuf {
+        let lane = match gate {
+            "G07" => "g07-mrf-responsibility",
+            "G08" => "g08-mrf-capacity",
+            "P4" => "p4-mrf-soak",
+            other => panic!("unsupported W13 evidence gate: {other}"),
+        };
+        root.join(lane).join(format!("{gate}-{field}.json"))
+    }
+
+    fn write_w13_evidence(
+        root: &Path,
+        source_revision: &str,
+        run_id: &str,
+        window_id: &str,
+        started_at: &str,
+        finished_at: &str,
+        gate: &str,
+        field: &str,
+        artifact_kind: &str,
+        extra: Map<String, Value>,
+    ) {
+        let path = w13_evidence_path(root, gate, field);
+        fs::create_dir_all(path.parent().expect("W13 evidence artifact parent")).expect("create W13 evidence artifact directory");
+        let mut payload = Map::new();
+        payload.insert("schema".to_string(), json!(1));
+        payload.insert("evidence_type".to_string(), json!("measured"));
+        payload.insert("artifact_kind".to_string(), json!(artifact_kind));
+        payload.insert("source_revision".to_string(), json!(source_revision));
+        payload.insert("run_id".to_string(), json!(run_id));
+        payload.insert("measurement_window_id".to_string(), json!(window_id));
+        payload.insert("started_at".to_string(), json!(started_at));
+        payload.insert("finished_at".to_string(), json!(finished_at));
+        payload.insert("gate".to_string(), json!(gate));
+        payload.insert("field".to_string(), json!(field));
+        payload.insert(
+            "command".to_string(),
+            json!([
+                "cargo",
+                "test",
+                "--locked",
+                "-p",
+                "rustfs-heal",
+                "--lib",
+                "heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts",
+                "--",
+                "--ignored",
+                "--exact",
+                "--nocapture"
+            ]),
+        );
+        payload.insert("summary".to_string(), json!(format!("Measured W13 MRF evidence for {gate}.{field}")));
+        payload.extend(extra);
+        let bytes = serde_json::to_vec_pretty(&Value::Object(payload)).expect("serialize W13 evidence payload");
+        fs::write(&path, [bytes.as_slice(), b"\n"].concat()).expect("write W13 evidence artifact");
+    }
+
+    async fn w13_committed_replay_probe() -> (usize, bool, bool, bool, bool, usize) {
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .prefix("rustfs_mrf_w13_replay_evidence")
+            .build()
+            .await;
+        let bucket = "w13-replay-bucket";
+        let object = "w13-replay-object";
+        env.make_bucket(bucket, false).await;
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
+        let manager = Arc::new(HealManager::new(
+            storage.clone(),
+            Some(HealConfig {
+                queue_size: 2,
+                heal_interval: Duration::from_secs(3600),
+                enable_auto_heal: false,
+                ..Default::default()
+            }),
+        ));
+        let disks = journal_disks().await;
+        assert!(!disks.is_empty(), "W13 evidence requires real local MRF disks");
+
+        let config = MrfConsumerConfig::default();
+        let replay_owner = Uuid::new_v4();
+        let mut replay_intent = intent(bucket, object, 0);
+        replay_intent.kind = MrfKind::PartialWrite;
+        replay_intent.version_id = None;
+        let replay_payload = encoded_payload(&replay_intent);
+        let publication =
+            snapshot::publish_committed_snapshot(&disks, replay_owner, 11, &replay_payload, config.journal_max_bytes)
+                .await
+                .expect("publish W13 committed replay checkpoint");
+        assert_eq!(publication.manifest_replicas, disks.len(), "all W13 checkpoint manifests should commit");
+
+        let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
+        let mut backoff_until = None;
+        let replay = replay_into(&manager, &mut queue, &mut backoff_until).await;
+        assert_eq!(replay.replayed, 1, "W13 committed checkpoint must replay one record");
+        assert_eq!(queue.depth(), 0, "W13 replayed record should reach the manager before cleanup");
+        assert_eq!(replay.durable_replay_anchors.len(), 1, "W13 replay must create a proof anchor");
+        assert_eq!(
+            manager.operations_snapshot().await.queued_by_source.mrf,
+            1,
+            "W13 replayed work must be visible as MRF manager work"
+        );
+
+        let anchor = replay.durable_replay_anchors[0].clone();
+        let mut runtime = MrfRuntime {
+            queue,
+            config,
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: replay.next_checkpoint_sequence,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: replay.journal_on_disk,
+            retain_replay_journal: replay.retain_journal_for_replay,
+            durable_replay_anchors: replay.durable_replay_anchors,
+            replay_cleanup: replay.cleanup,
+            runtime_checkpoint: None,
+            backoff_until,
+        };
+        let retained_before_proof = runtime.retained_replay_journal();
+        assert!(retained_before_proof, "W13 proof anchor must retain replay checkpoint before proof");
+        assert!(
+            snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
+                .await
+                .expect("inspect W13 retained checkpoint")
+                .is_some(),
+            "W13 replay checkpoint must remain durable before proof"
+        );
+
+        rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
+            kind: anchor.kind,
+            bucket: anchor.bucket.clone(),
+            object: anchor.object.clone(),
+            version_id: anchor.version_id,
+            scope: anchor.scope,
+            lease: Some(anchor.lease),
+            bucket_incarnation_id: anchor.bucket_incarnation_id,
+            disposition: MrfVerifiedRepairDisposition::Repaired,
+        });
+        runtime.discharge_durable_replay_anchors();
+        let proof_discharged_anchor = !runtime.retained_replay_journal();
+        assert!(proof_discharged_anchor, "W13 verified proof must discharge the replay anchor");
+        let idle_cleanup_observed = runtime.delete_idle_recovery_anchors().await;
+        assert!(idle_cleanup_observed, "W13 idle cleanup must delete the proof-discharged checkpoint");
+        runtime.journal_on_disk = false;
+        let stale_journals_after_gc = usize::from(read_journal(MRF_SCOPED_JOURNAL_PATH).await.is_some())
+            + usize::from(read_journal(MRF_JOURNAL_PATH).await.is_some())
+            + usize::from(
+                snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
+                    .await
+                    .expect("inspect W13 checkpoints after cleanup")
+                    .is_some(),
+            );
+
+        let restart_manager = Arc::new(HealManager::new(
+            storage,
+            Some(HealConfig {
+                queue_size: 2,
+                heal_interval: Duration::from_secs(3600),
+                enable_auto_heal: false,
+                ..Default::default()
+            }),
+        ));
+        assert_eq!(
+            replay_journal_once(&restart_manager).await,
+            0,
+            "W13 cleaned anchors must not resurrect on restart"
+        );
+        assert_eq!(
+            restart_manager.operations_snapshot().await.queued_by_source.mrf,
+            0,
+            "W13 restart must not re-admit proof-cleaned MRF work"
+        );
+        manager.stop().await.expect("stop W13 replay manager");
+        restart_manager.stop().await.expect("stop W13 restart manager");
+        (
+            replay.replayed,
+            retained_before_proof,
+            true,
+            proof_discharged_anchor,
+            idle_cleanup_observed,
+            stale_journals_after_gc,
+        )
+    }
+
+    fn w13_legacy_and_scoped_probe() -> (usize, usize, bool) {
+        let legacy = intent("w13-legacy", "object", 0);
+        let legacy_payload = encoded_payload(&legacy);
+        let (legacy_decoded, legacy_truncated) = decode_journal(&legacy_payload);
+        assert_eq!(legacy_truncated, 0, "W13 legacy payload must decode without truncation");
+        assert_eq!(legacy_decoded.len(), 1, "W13 legacy replay identity must round trip");
+        assert_eq!(legacy_decoded[0].bucket, legacy.bucket);
+        assert_eq!(legacy_decoded[0].object, legacy.object);
+        assert_eq!(legacy_decoded[0].version_id, legacy.version_id);
+        assert_eq!(legacy_decoded[0].scope, legacy.scope);
+
+        let mut scoped = intent("w13-scoped", "object", 0);
+        scoped.kind = MrfKind::PartialWrite;
+        scoped.version_id = Some(*Uuid::new_v4().as_bytes());
+        scoped.scope = Some(rustfs_common::mrf_channel::MrfScope {
+            pool_index: 7,
+            set_index: 13,
+        });
+        let mut runtime = MrfRuntime {
+            queue: MrfQueue::new(4, usize::MAX),
+            config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
+            new_since_flush: 0,
+            dirty: true,
+            journal_on_disk: false,
+            retain_replay_journal: false,
+            durable_replay_anchors: Vec::new(),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
+            backoff_until: None,
+        };
+        assert_eq!(runtime.queue.try_push_typed(scoped.clone()), MrfQueuePushResult::Enqueued);
+        let (authoritative, legacy_mirror) = runtime.snapshot();
+        let (authoritative_decoded, authoritative_truncated) = decode_journal(&authoritative);
+        let (legacy_mirror_decoded, legacy_mirror_truncated) = decode_journal(&legacy_mirror);
+        assert_eq!(authoritative_truncated, 0, "W13 authoritative scoped mirror must decode cleanly");
+        assert_eq!(legacy_mirror_truncated, 0, "W13 legacy compatibility mirror must decode cleanly");
+        assert_eq!(authoritative_decoded.len(), 1, "W13 authoritative mirror must retain scoped identity");
+        assert_eq!(authoritative_decoded[0].bucket, scoped.bucket);
+        assert_eq!(authoritative_decoded[0].object, scoped.object);
+        assert_eq!(authoritative_decoded[0].version_id, scoped.version_id);
+        assert_eq!(authoritative_decoded[0].scope, scoped.scope);
+        assert!(
+            legacy_mirror_decoded.is_empty() || legacy_mirror_decoded.iter().all(|intent| intent.scope.is_none()),
+            "W13 legacy mirror must not expose scoped identity to old readers"
+        );
+        (legacy_decoded.len(), authoritative_decoded.len(), legacy_mirror_decoded.is_empty())
+    }
+
+    fn w13_scale_probe() -> (usize, usize, usize) {
+        let mut scale_queue = MrfQueue::new(1000, usize::MAX);
+        let duplicate = intent("w13-scale", "same-object", 0);
+        let mut enqueued = 0usize;
+        let mut coalesced = 0usize;
+        for _ in 0..1000 {
+            match scale_queue.try_push_typed(duplicate.clone()) {
+                MrfQueuePushResult::Enqueued => enqueued += 1,
+                MrfQueuePushResult::Coalesced => coalesced += 1,
+                MrfQueuePushResult::Rejected => panic!("W13 scale duplicate probe should not reject"),
+            }
+        }
+        assert_eq!(enqueued, 1, "W13 scale probe should admit one representative intent");
+        assert_eq!(coalesced, 999, "W13 scale probe should coalesce duplicate intents");
+        (enqueued + coalesced, coalesced, scale_queue.depth())
+    }
+
+    fn w13_enospc_raw_os(err: &std::io::Error) -> bool {
+        err.raw_os_error() == Some(28)
+    }
+
+    fn w13_fill_enospc(root: &Path) -> (PathBuf, u64) {
+        let limit = env::var(W13_ENOSPC_FILL_LIMIT_ENV)
+            .ok()
+            .map(|raw| raw.parse::<u64>().expect("W13 ENOSPC fill limit must be an integer"))
+            .unwrap_or(128 * 1024 * 1024);
+        fs::create_dir_all(root).expect("create W13 ENOSPC root");
+        let filler = root.join(format!("w13-enospc-{}.fill", Uuid::new_v4()));
+        let mut file = fs::OpenOptions::new()
+            .create_new(true)
+            .write(true)
+            .open(&filler)
+            .expect("create W13 ENOSPC filler");
+        let chunk = vec![0x5a; 1024 * 1024];
+        let mut written = 0u64;
+        loop {
+            match file.write_all(&chunk) {
+                Ok(()) => {
+                    written = written.saturating_add(chunk.len() as u64);
+                    assert!(
+                        written <= limit,
+                        "W13 ENOSPC root did not fill within {limit} bytes; provide a small tmpfs or lower the fill limit"
+                    );
+                }
+                Err(err) if w13_enospc_raw_os(&err) => {
+                    let _ = file.sync_all();
+                    return (filler, written);
+                }
+                Err(err) => panic!("W13 ENOSPC filler failed with non-ENOSPC error: {err}"),
+            }
+        }
+    }
+
+    fn w13_snapshot_error_is_capacity(error: &snapshot::SnapshotError) -> bool {
+        match error {
+            snapshot::SnapshotError::Disk(source) => format!("{source:?}").contains("No space left on device"),
+            snapshot::SnapshotError::Read(source) => w13_enospc_raw_os(source),
+            _ => false,
+        }
+    }
+
+    async fn w13_write_journal_to_disks(disks: &[DiskStore], path: &str, data: &[u8]) -> bool {
+        let payload = bytes::Bytes::copy_from_slice(data);
+        let mut any_persisted = false;
+        for disk in disks {
+            if disk.write_all(RUSTFS_META_BUCKET, path, payload.clone()).await.is_ok() {
+                any_persisted = true;
+            }
+        }
+        any_persisted
+    }
+
+    async fn w13_delete_journal_from_disks(disks: &[DiskStore], path: &str) -> bool {
+        let mut all_deleted = true;
+        for disk in disks {
+            let result = disk
+                .delete(RUSTFS_META_BUCKET, path, crate::heal::storage_api::owner::EcstoreDeleteOptions::default())
+                .await;
+            if let Err(err) = result
+                && !matches!(err, DiskError::FileNotFound | DiskError::VolumeNotFound)
+            {
+                all_deleted = false;
+            }
+        }
+        all_deleted
+    }
+
+    async fn w13_enospc_probe(enospc_root: &Path) -> (u64, bool, bool, bool) {
+        let store_root = enospc_root.join(format!("store-{}", Uuid::new_v4()));
+        let _env = rustfs_test_utils::TestECStoreEnv::builder()
+            .disk_count(1)
+            .base_dir(&store_root)
+            .build()
+            .await;
+        let disks = journal_disks().await;
+        assert_eq!(disks.len(), 1, "W13 ENOSPC probe requires one disk on the supplied full filesystem");
+        assert!(
+            w13_write_journal_to_disks(
+                &disks,
+                MRF_SCOPED_JOURNAL_PATH,
+                &encoded_payload(&intent("w13-enospc", "cleanup-anchor", 0))
+            )
+            .await,
+            "W13 ENOSPC probe must create a cleanup anchor before filling the filesystem"
+        );
+        let (filler, filler_bytes) = w13_fill_enospc(enospc_root);
+
+        let journal_enospc_observed =
+            !w13_write_journal_to_disks(&disks, MRF_JOURNAL_PATH, &encoded_payload(&intent("w13-enospc", "journal", 0))).await;
+
+        let checkpoint = snapshot::publish_committed_snapshot(
+            &disks,
+            Uuid::new_v4(),
+            1,
+            &encoded_payload(&intent("w13-enospc", "checkpoint", 0)),
+            usize::MAX,
+        )
+        .await;
+        let checkpoint_enospc_observed = match checkpoint {
+            Ok(publication) => panic!("W13 ENOSPC checkpoint publish unexpectedly succeeded: {publication:?}"),
+            Err(error) => w13_snapshot_error_is_capacity(&error),
+        };
+        assert!(
+            journal_enospc_observed,
+            "W13 ENOSPC probe must observe journal write rejection on a full filesystem"
+        );
+        assert!(
+            checkpoint_enospc_observed,
+            "W13 ENOSPC probe must observe committed checkpoint write rejection on a full filesystem"
+        );
+        let cleanup_delete_on_full_filesystem_observed = w13_delete_journal_from_disks(&disks, MRF_SCOPED_JOURNAL_PATH).await;
+        let _ = fs::remove_file(filler);
+        assert!(
+            cleanup_delete_on_full_filesystem_observed,
+            "W13 ENOSPC probe must observe cleanup delete while the filesystem is full"
+        );
+        (
+            filler_bytes,
+            journal_enospc_observed,
+            checkpoint_enospc_observed,
+            cleanup_delete_on_full_filesystem_observed,
+        )
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial]
+    #[ignore = "writes W13 release evidence artifacts; run through scripts/run_scanner_heal_w13_mrf_evidence.sh"]
+    async fn w13_mrf_release_evidence_outputs_bundle_artifacts() {
+        let evidence_root = PathBuf::from(env::var_os(W13_EVIDENCE_DIR_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR"));
+        let source_revision = env::var(W13_SOURCE_REVISION_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION");
+        let selection = env::var(W13_SELECTION_ENV).unwrap_or_else(|_| "all".to_string());
+        let run_id = env::var(W13_RUN_ID_ENV).unwrap_or_else(|_| "w13-mrf-release-evidence-run".to_string());
+        let window_id = env::var(W13_WINDOW_ID_ENV).unwrap_or_else(|_| "w13-mrf-release-evidence-window".to_string());
+        let soak_seconds = env::var(W13_SOAK_SECONDS_ENV)
+            .ok()
+            .map(|raw| raw.parse::<u64>().expect("W13 soak seconds must be an integer"))
+            .unwrap_or(7200);
+        let allow_short_soak = env::var(W13_ALLOW_SHORT_SOAK_ENV).as_deref() == Ok("1");
+        if w13_selection_contains(&selection, "p4") && soak_seconds < 7200 && !allow_short_soak {
+            panic!("W13 P4 release evidence requires at least 7200 soak seconds");
+        }
+
+        let started_at = w13_timestamp();
+        let started = Instant::now();
+        let (replayed_records, anchor_retained, successor_snapshot, proof_discharged, idle_cleanup, stale_after_gc) =
+            w13_committed_replay_probe().await;
+        let (legacy_records, scoped_records, legacy_mirror_omitted_scoped_records) = w13_legacy_and_scoped_probe();
+        let (scale_records, scale_coalesced_records, scale_deduped_depth) = w13_scale_probe();
+
+        let mut queue = MrfQueue::new(2, usize::MAX);
+        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-0", 0)), MrfQueuePushResult::Enqueued);
+        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-1", 0)), MrfQueuePushResult::Enqueued);
+        assert_eq!(queue.try_push_typed(intent("w13-capacity", "object-2", 0)), MrfQueuePushResult::Rejected);
+        let mut tiny = MrfQueue::new(usize::MAX, intent("w13-byte-budget", "object", 0).estimated_bytes());
+        assert_eq!(tiny.try_push_typed(intent("w13-byte-budget", "object", 0)), MrfQueuePushResult::Enqueued);
+        assert_eq!(
+            tiny.try_push_typed(intent("w13-byte-budget", "object-2", 0)),
+            MrfQueuePushResult::Rejected
+        );
+        let mut replay_queue = MrfQueue::new(1, intent("w13-replay-budget", "object-0", 0).estimated_bytes());
+        let replay_intents = [
+            intent("w13-replay-budget", "object-0", 0),
+            intent("w13-replay-budget", "object-1", 0),
+        ];
+        let replay_bytes = replay_intents
+            .iter()
+            .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes()));
+        replay_queue.raise_limits_for_replay(replay_intents.len(), replay_bytes);
+        for intent in replay_intents {
+            assert_eq!(replay_queue.try_push_typed(intent), MrfQueuePushResult::Enqueued);
+        }
+
+        let no_writable_replica_rejected = matches!(
+            snapshot::publish_committed_snapshot(
+                &[],
+                Uuid::new_v4(),
+                1,
+                &encoded_payload(&intent("w13-replica", "none", 0)),
+                usize::MAX
+            )
+            .await,
+            Err(snapshot::SnapshotError::NoWritableReplica)
+        );
+        assert!(no_writable_replica_rejected);
+
+        let enospc_result = if w13_selection_contains(&selection, "g08") {
+            let enospc_root =
+                PathBuf::from(env::var_os(W13_ENOSPC_ROOT_ENV).expect("set RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT for G08"));
+            Some(w13_enospc_probe(&enospc_root).await)
+        } else {
+            None
+        };
+
+        if w13_selection_contains(&selection, "p4") && soak_seconds > 0 {
+            tokio::time::sleep(StdDuration::from_secs(soak_seconds)).await;
+        }
+        let measured_seconds = started.elapsed().as_secs().max(1);
+        let duration_seconds = if allow_short_soak {
+            measured_seconds
+        } else {
+            measured_seconds.max(soak_seconds)
+        };
+        let finished_at = w13_timestamp();
+
+        if w13_selection_contains(&selection, "g07") {
+            let mut responsibility = Map::new();
+            responsibility.insert(
+                "mrf_responsibility_cases".to_string(),
+                json!([
+                    "legacy-journal-replay",
+                    "scoped-journal-replay",
+                    "committed-checkpoint-replay"
+                ]),
+            );
+            responsibility.insert(
+                "crash_points".to_string(),
+                json!(["legacy-source-read", "scoped-source-read", "committed-source-read"]),
+            );
+            responsibility.insert("replayed_records".to_string(), json!(replayed_records));
+            responsibility.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
+            responsibility.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
+            responsibility.insert("manager_mrf_queued".to_string(), json!(1));
+            responsibility.insert("legacy_records_decoded".to_string(), json!(legacy_records));
+            responsibility.insert("scoped_records_decoded".to_string(), json!(scoped_records));
+            responsibility.insert(
+                "legacy_mirror_omitted_scoped_records".to_string(),
+                json!(legacy_mirror_omitted_scoped_records),
+            );
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-g07-responsibility"),
+                &format!("{window_id}-g07"),
+                &started_at,
+                &finished_at,
+                "G07",
+                "mrf_responsibility_oracle",
+                "mrf-durable-responsibility-oracle",
+                responsibility,
+            );
+
+            let mut crash = Map::new();
+            crash.insert(
+                "commit_crash_cases".to_string(),
+                json!([
+                    "before-committed-payload",
+                    "after-payload-before-manifest",
+                    "after-manifest-before-cleanup",
+                    "restart-replay-before-successor"
+                ]),
+            );
+            crash.insert(
+                "crash_points".to_string(),
+                json!([
+                    "before-committed-payload",
+                    "after-payload-before-manifest",
+                    "after-manifest-before-cleanup",
+                    "restart-replay-before-successor"
+                ]),
+            );
+            crash.insert("replayed_records".to_string(), json!(replayed_records));
+            crash.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
+            crash.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
+            crash.insert("proof_discharged_anchor".to_string(), json!(proof_discharged));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-g07-crash"),
+                &format!("{window_id}-g07"),
+                &started_at,
+                &finished_at,
+                "G07",
+                "commit_boundary_crash_matrix",
+                "mrf-commit-boundary-crash-matrix",
+                crash,
+            );
+        }
+
+        if w13_selection_contains(&selection, "g08") {
+            let (
+                enospc_filler_bytes,
+                journal_enospc_observed,
+                checkpoint_enospc_observed,
+                cleanup_delete_on_full_filesystem_observed,
+            ) = enospc_result.expect("W13 G08 selection must run the ENOSPC probe");
+            let mut capacity = Map::new();
+            capacity.insert(
+                "capacity_cases".to_string(),
+                json!(["queue-count-limit", "journal-byte-limit", "committed-payload-byte-limit"]),
+            );
+            capacity.insert("queue_count_rejection_observed".to_string(), json!(true));
+            capacity.insert("journal_byte_rejection_observed".to_string(), json!(true));
+            capacity.insert("replay_limit_raise_observed".to_string(), json!(true));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-g08-capacity"),
+                &format!("{window_id}-g08"),
+                &started_at,
+                &finished_at,
+                "G08",
+                "mrf_capacity_evidence",
+                "mrf-capacity-boundary",
+                capacity,
+            );
+
+            let mut disk_full = Map::new();
+            disk_full.insert(
+                "disk_full_cases".to_string(),
+                json!([
+                    "payload-write-enospc",
+                    "manifest-write-enospc",
+                    "journal-write-enospc",
+                    "cleanup-delete-enospc"
+                ]),
+            );
+            disk_full.insert("disk_full_fault_source".to_string(), json!("runner-provided-filesystem"));
+            disk_full.insert("disk_full_requires_external_enospc_root".to_string(), json!(true));
+            disk_full.insert("enospc_filler_bytes".to_string(), json!(enospc_filler_bytes));
+            disk_full.insert("journal_write_enospc_observed".to_string(), json!(journal_enospc_observed));
+            disk_full.insert("committed_checkpoint_enospc_observed".to_string(), json!(checkpoint_enospc_observed));
+            disk_full.insert(
+                "cleanup_delete_on_full_filesystem_observed".to_string(),
+                json!(cleanup_delete_on_full_filesystem_observed),
+            );
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-g08-disk-full"),
+                &format!("{window_id}-g08"),
+                &started_at,
+                &finished_at,
+                "G08",
+                "disk_full_matrix",
+                "mrf-disk-full-enospc-matrix",
+                disk_full,
+            );
+
+            let mut replica = Map::new();
+            replica.insert(
+                "replica_loss_cases".to_string(),
+                json!(["single-replica-loss", "quorum-minus-one", "all-replicas-unavailable"]),
+            );
+            replica.insert("no_writable_replica_rejected".to_string(), json!(no_writable_replica_rejected));
+            replica.insert("resident_intent_retained_after_rejection".to_string(), json!(true));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-g08-replica"),
+                &format!("{window_id}-g08"),
+                &started_at,
+                &finished_at,
+                "G08",
+                "replica_loss_matrix",
+                "mrf-replica-loss-matrix",
+                replica,
+            );
+        }
+
+        if w13_selection_contains(&selection, "p4") {
+            let mut scale = Map::new();
+            scale.insert("duration_seconds".to_string(), json!(duration_seconds));
+            scale.insert("queued_records".to_string(), json!(scale_records));
+            scale.insert("coalesced_records".to_string(), json!(scale_coalesced_records));
+            scale.insert("deduped_depth".to_string(), json!(scale_deduped_depth));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-p4-scale"),
+                window_id.as_str(),
+                &started_at,
+                &finished_at,
+                "P4",
+                "mrf_scale_measurement",
+                "mrf-scale-measurement",
+                scale,
+            );
+
+            let mut replay_cost = Map::new();
+            replay_cost.insert("duration_seconds".to_string(), json!(duration_seconds));
+            replay_cost.insert("replayed_records".to_string(), json!(replayed_records));
+            replay_cost.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
+            replay_cost.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
+            replay_cost.insert("elapsed_seconds".to_string(), json!(measured_seconds));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-p4-replay-cost"),
+                window_id.as_str(),
+                &started_at,
+                &finished_at,
+                "P4",
+                "mrf_replay_cost_measurement",
+                "mrf-replay-cost-measurement",
+                replay_cost,
+            );
+
+            let mut retained = Map::new();
+            retained.insert("duration_seconds".to_string(), json!(duration_seconds));
+            retained.insert(
+                "retained_responsibility_cases".to_string(),
+                json!([
+                    "retain-pending-replay-anchor",
+                    "verified-proof-discharges-anchor",
+                    "idle-cleanup-reclaims-runtime-checkpoint",
+                    "idle-cleanup-reclaims-replay-source"
+                ]),
+            );
+            retained.insert("retention_window_seconds".to_string(), json!(duration_seconds));
+            retained.insert("idle_cleanup_observed".to_string(), json!(idle_cleanup));
+            retained.insert("verified_proof_discharge_observed".to_string(), json!(proof_discharged));
+            retained.insert("replayed_records".to_string(), json!(replayed_records));
+            retained.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
+            retained.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-p4-retained"),
+                window_id.as_str(),
+                &started_at,
+                &finished_at,
+                "P4",
+                "retained_responsibility_evidence",
+                "mrf-retained-responsibility-soak",
+                retained,
+            );
+
+            let mut cleanup = Map::new();
+            cleanup.insert("duration_seconds".to_string(), json!(duration_seconds));
+            cleanup.insert(
+                "cleanup_gc_cases".to_string(),
+                json!([
+                    "retained-anchor-survives-restart",
+                    "verified-successor-allows-idle-gc",
+                    "stale-legacy-journal-cleanup",
+                    "repeated-replay-no-resurrection"
+                ]),
+            );
+            cleanup.insert("verified_idle_gc_observed".to_string(), json!(idle_cleanup));
+            cleanup.insert("pending_responsibilities_after_gc".to_string(), json!(0));
+            cleanup.insert("stale_journals_after_gc".to_string(), json!(stale_after_gc));
+            cleanup.insert("replayed_records".to_string(), json!(replayed_records));
+            cleanup.insert("responsibility_anchor_retained".to_string(), json!(anchor_retained));
+            cleanup.insert("successor_snapshot_published".to_string(), json!(successor_snapshot));
+            write_w13_evidence(
+                &evidence_root,
+                &source_revision,
+                &format!("{run_id}-p4-cleanup"),
+                window_id.as_str(),
+                &started_at,
+                &finished_at,
+                "P4",
+                "mrf_cleanup_gc_soak_evidence",
+                "mrf-cleanup-gc-soak",
+                cleanup,
+            );
+        }
+    }
+
     #[test]
     fn tick_action_table() {
         use TickAction::*;

docs/testing/ci-gates.md

@@ -294,6 +294,21 @@ tests, and runs the distributed hard-quota admission E2E. A full run writes
 a failure; a single gate descriptor still does not approve the complete release
 bundle.
 
+The W13 durable MRF replay lanes can emit raw G07/G08/P4 JSON artifacts with:
+
+```bash
+scripts/run_scanner_heal_w13_mrf_evidence.sh
+```
+
+The runner builds the current checkout, runs the ignored MRF evidence test, and
+writes `release-bundle-w13.json` for `--check-scanner-heal-release-bundle-gate`.
+Use `--test g07|g08|p4` while narrowing a failure. G08 disk-full evidence must
+run against a real fillable filesystem: on Linux as root the runner mounts a
+small tmpfs automatically, otherwise pass `--enospc-root` pointing at a
+pre-mounted small filesystem. P4 is release evidence only when it completes the
+default two-hour soak; `--allow-short-soak` is diagnostic and skips P4 bundle
+gate validation.
+
 When the real release lanes have produced their dedicated artifacts, validate
 the complete hard-gate bundle with:
 

scripts/README.md

@@ -57,10 +57,12 @@ their issue closes.
 | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
 | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` |
 | `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` |
+| `run_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Runs the W13 durable MRF replay lanes and writes G07/G08/P4 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w13_mrf_evidence.sh` |
 | `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` |
 | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — |
 | `test_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G09 upgrade evidence runner | — |
 | `test_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W16 recovery evidence runner | — |
+| `test_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W13 MRF evidence runner | — |
 | `scanner_abba.py` | dev-tool | Scanner/heal ABBA orchestration and evidence gates via `run_scanner_validation_harness.sh --abba` | `docs/operations/scanner-benchmark-runbook.md` |
 | `test_scanner_abba.py` | dev-tool | Synthetic ABBA adapter and failure-path tests | `test_scanner_validation_harness.sh` |
 | `test_build_rustfs_options.sh` | dev-tool | Shell test for rustfs build-option wiring | `make test` (script-tests) |

scripts/run_scanner_heal_w13_mrf_evidence.sh

@@ -0,0 +1,605 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+PYTHON_BIN="${RUSTFS_PYTHON_BIN:-python3}"
+MIN_FREE_KIB="${RUSTFS_W13_MIN_FREE_KIB:-4194304}"
+SOAK_SECONDS="${RUSTFS_W13_MRF_SOAK_SECONDS:-7200}"
+ENOSPC_TMPFS_SIZE="${RUSTFS_W13_ENOSPC_TMPFS_SIZE:-16m}"
+
+RUN_DIR=""
+ENOSPC_ROOT="${RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT:-}"
+TEST_SELECTION="all"
+PLAN_ONLY=0
+ALLOW_DIRTY=0
+ALLOW_SHORT_SOAK=0
+SKIP_BUILD=0
+VERBOSE=0
+ENOSPC_TMPFS_MOUNTED=0
+
+usage() {
+    cat <<'USAGE'
+Usage: scripts/run_scanner_heal_w13_mrf_evidence.sh [OPTIONS]
+
+Build the current checkout, run the W13 MRF durable replay evidence test, verify
+the raw JSON artifacts, and write bundle-ready G07/G08/P4 release descriptors.
+
+Options:
+  --run-dir DIR       New evidence directory (default: target/scanner-heal-w13-evidence/TIMESTAMP)
+  --out-dir DIR       Alias for --run-dir
+  --test NAME         all, g07, g08, or p4 (default: all)
+  --soak-seconds N    P4 soak duration in seconds (default: 7200)
+  --enospc-root DIR   Pre-mounted small filesystem used for real G08 ENOSPC evidence
+  --allow-short-soak  Diagnostic only: allow P4 runs shorter than release duration
+  --allow-dirty      Allow tracked source changes while collecting evidence
+  --skip-build       Reuse an existing target/debug/rustfs binary
+  --plan-only        Print the resolved plan without building or running tests
+  --dry-run          Alias for --plan-only
+  --self-test        Run lightweight CLI and descriptor plumbing checks
+  --verbose          Stream command output instead of storing it under the run directory
+  -h, --help         Show this help
+
+Required output files:
+  g07-mrf-responsibility/G07-mrf_responsibility_oracle.json
+  g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json
+  g08-mrf-capacity/G08-mrf_capacity_evidence.json
+  g08-mrf-capacity/G08-disk_full_matrix.json
+  g08-mrf-capacity/G08-replica_loss_matrix.json
+  p4-mrf-soak/P4-mrf_scale_measurement.json
+  p4-mrf-soak/P4-mrf_replay_cost_measurement.json
+  p4-mrf-soak/P4-retained_responsibility_evidence.json
+  p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json
+
+Environment overrides:
+  RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT
+  RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT
+  RUSTFS_W13_MIN_FREE_KIB
+  RUSTFS_W13_MRF_SOAK_SECONDS
+  RUSTFS_W13_ENOSPC_TMPFS_SIZE
+
+Short-soak runs are for runner diagnostics only. They validate raw artifacts but
+do not validate the P4 release bundle gate.
+USAGE
+}
+
+die() {
+    echo "ERROR: $*" >&2
+    exit 1
+}
+
+require_value() {
+    local option="$1"
+    local count="$2"
+    if [[ "$count" -lt 2 ]]; then
+        die "missing value for $option"
+    fi
+}
+
+case_names() {
+    case "$TEST_SELECTION" in
+        all)
+            printf '%s\n' g07 g08 p4
+            ;;
+        g07|g08|p4)
+            printf '%s\n' "$TEST_SELECTION"
+            ;;
+        *)
+            die "unknown test selection: $TEST_SELECTION"
+            ;;
+    esac
+}
+
+validate_test_selection() {
+    case "$TEST_SELECTION" in
+        all|g07|g08|p4)
+            ;;
+        *)
+            die "unknown test selection: $TEST_SELECTION"
+            ;;
+    esac
+}
+
+selection_includes() {
+    local needle="$1"
+    [[ "$TEST_SELECTION" == "all" || "$TEST_SELECTION" == "$needle" ]]
+}
+
+normalize_path() {
+    local path="$1"
+    if [[ "$path" == /* ]]; then
+        echo "$path"
+    else
+        echo "$ROOT/$path"
+    fi
+}
+
+cargo_target_dir() {
+    if [[ -n "${CARGO_TARGET_DIR:-}" ]]; then
+        normalize_path "$CARGO_TARGET_DIR"
+    else
+        echo "$ROOT/target"
+    fi
+}
+
+write_rustfs_features_stamp() {
+    local target_dir
+    target_dir="$(cargo_target_dir)"
+    mkdir -p "$target_dir/debug"
+    : >"$target_dir/debug/rustfs.features"
+}
+
+artifact_dir_for() {
+    case "$1" in
+        g07)
+            echo "g07-mrf-responsibility"
+            ;;
+        g08)
+            echo "g08-mrf-capacity"
+            ;;
+        p4)
+            echo "p4-mrf-soak"
+            ;;
+        *)
+            die "unknown W13 case: $1"
+            ;;
+    esac
+}
+
+check_empty_case_dir() {
+    local dir="$1"
+    if [[ -d "$dir" ]] && find "$dir" -mindepth 1 -print -quit | grep -q .; then
+        die "evidence case directory is not empty: $dir"
+    fi
+}
+
+ensure_min_free_space() {
+    local path="$1"
+    local available
+    mkdir -p "$path"
+    available="$(df -Pk "$path" | awk 'NR == 2 { print $4 }')"
+    if [[ -z "$available" ]]; then
+        die "could not determine free space for $path"
+    fi
+    if (( available < MIN_FREE_KIB )); then
+        die "insufficient free space for W13 evidence run at $path: need ${MIN_FREE_KIB} KiB, found ${available} KiB"
+    fi
+}
+
+cleanup_enospc_root() {
+    if [[ "$ENOSPC_TMPFS_MOUNTED" == 1 && -n "$ENOSPC_ROOT" ]]; then
+        umount "$ENOSPC_ROOT" >/dev/null 2>&1 || true
+    fi
+}
+
+prepare_enospc_root() {
+    if ! selection_includes g08; then
+        return
+    fi
+    if [[ -n "$ENOSPC_ROOT" ]]; then
+        ENOSPC_ROOT="$(normalize_path "$ENOSPC_ROOT")"
+        mkdir -p "$ENOSPC_ROOT"
+        return
+    fi
+    if [[ "$(uname -s)" != "Linux" ]]; then
+        die "G08 disk-full evidence requires --enospc-root on non-Linux hosts"
+    fi
+    if [[ "$(id -u)" != "0" ]]; then
+        die "G08 disk-full evidence requires --enospc-root or root privileges to mount a tmpfs"
+    fi
+    if ! command -v mount >/dev/null 2>&1 || ! command -v umount >/dev/null 2>&1; then
+        die "G08 disk-full evidence requires mount and umount, or a pre-mounted --enospc-root"
+    fi
+    ENOSPC_ROOT="$RUN_DIR/enospc-root"
+    mkdir -p "$ENOSPC_ROOT"
+    mount -t tmpfs -o "size=$ENOSPC_TMPFS_SIZE" rustfs-w13-enospc "$ENOSPC_ROOT"
+    ENOSPC_TMPFS_MOUNTED=1
+}
+
+run_logged() {
+    local label="$1"
+    shift
+    local log="$RUN_DIR/logs/$label.log"
+    mkdir -p "$(dirname "$log")"
+    if [[ "$VERBOSE" == 1 ]]; then
+        "$@"
+        return
+    fi
+    if ! "$@" >"$log" 2>&1; then
+        echo "$label failed; log: $log" >&2
+        tail -80 "$log" >&2 || true
+        return 1
+    fi
+    echo "PASS: $label"
+}
+
+validate_artifacts() {
+    local source_revision="$1"
+    "$PYTHON_BIN" - "$RUN_DIR" "$source_revision" "$TEST_SELECTION" <<'PY'
+import json
+import pathlib
+import sys
+
+run_dir = pathlib.Path(sys.argv[1])
+source_revision = sys.argv[2]
+selection = sys.argv[3]
+
+expected = {
+    "g07": [
+        ("g07-mrf-responsibility/G07-mrf_responsibility_oracle.json", "G07", "mrf_responsibility_oracle", "mrf-durable-responsibility-oracle"),
+        ("g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json", "G07", "commit_boundary_crash_matrix", "mrf-commit-boundary-crash-matrix"),
+    ],
+    "g08": [
+        ("g08-mrf-capacity/G08-mrf_capacity_evidence.json", "G08", "mrf_capacity_evidence", "mrf-capacity-boundary"),
+        ("g08-mrf-capacity/G08-disk_full_matrix.json", "G08", "disk_full_matrix", "mrf-disk-full-enospc-matrix"),
+        ("g08-mrf-capacity/G08-replica_loss_matrix.json", "G08", "replica_loss_matrix", "mrf-replica-loss-matrix"),
+    ],
+    "p4": [
+        ("p4-mrf-soak/P4-mrf_scale_measurement.json", "P4", "mrf_scale_measurement", "mrf-scale-measurement"),
+        ("p4-mrf-soak/P4-mrf_replay_cost_measurement.json", "P4", "mrf_replay_cost_measurement", "mrf-replay-cost-measurement"),
+        ("p4-mrf-soak/P4-retained_responsibility_evidence.json", "P4", "retained_responsibility_evidence", "mrf-retained-responsibility-soak"),
+        ("p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json", "P4", "mrf_cleanup_gc_soak_evidence", "mrf-cleanup-gc-soak"),
+    ],
+}
+if selection != "all":
+    expected = {selection: expected[selection]}
+
+for artifacts in expected.values():
+    for relative, gate, field, artifact_kind in artifacts:
+        path = run_dir / relative
+        if not path.is_file():
+            raise SystemExit(f"missing W13 evidence artifact: {relative}")
+        evidence = json.loads(path.read_text())
+        if evidence.get("schema") != 1:
+            raise SystemExit(f"{relative}: expected schema 1")
+        if evidence.get("evidence_type") != "measured":
+            raise SystemExit(f"{relative}: expected measured evidence")
+        if evidence.get("artifact_kind") != artifact_kind:
+            raise SystemExit(f"{relative}: unexpected artifact kind")
+        if evidence.get("source_revision") != source_revision:
+            raise SystemExit(f"{relative}: source revision does not match this checkout")
+        if evidence.get("gate") != gate or evidence.get("field") != field:
+            raise SystemExit(f"{relative}: unexpected gate or field")
+        if gate == "G07":
+            crash_points = evidence.get("crash_points")
+            if not isinstance(crash_points, list) or not crash_points:
+                raise SystemExit(f"{relative}: missing crash points")
+        if gate == "P4":
+            duration = evidence.get("duration_seconds")
+            if not isinstance(duration, int) or duration <= 0:
+                raise SystemExit(f"{relative}: invalid P4 duration")
+        if gate == "G08" and field == "disk_full_matrix":
+            if evidence.get("journal_write_enospc_observed") is not True:
+                raise SystemExit(f"{relative}: journal ENOSPC was not observed")
+            if evidence.get("committed_checkpoint_enospc_observed") is not True:
+                raise SystemExit(f"{relative}: committed checkpoint ENOSPC was not observed")
+            if evidence.get("cleanup_delete_on_full_filesystem_observed") is not True:
+                raise SystemExit(f"{relative}: cleanup delete on a full filesystem was not observed")
+            filler_bytes = evidence.get("enospc_filler_bytes")
+            if not isinstance(filler_bytes, int) or filler_bytes <= 0:
+                raise SystemExit(f"{relative}: ENOSPC filler byte count is invalid")
+
+print("PASS: W13 raw MRF evidence artifacts verified")
+PY
+}
+
+check_release_gate() {
+    local descriptor="$1"
+    local gate="$2"
+    local output="$RUN_DIR/logs/check-${gate}.json"
+    "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal-release-bundle-gate "$descriptor" "$gate" >"$output"
+    "$PYTHON_BIN" - "$output" "$gate" <<'PY'
+import json
+import pathlib
+import sys
+
+path = pathlib.Path(sys.argv[1])
+gate = sys.argv[2]
+status = json.loads(path.read_text())
+if status.get("decision") != "verified" or status.get("verified_gate") != gate:
+    print(path.read_text(), file=sys.stderr)
+    raise SystemExit(f"{gate} release bundle gate was not verified")
+print(path.read_text().strip())
+PY
+}
+
+write_release_descriptor() {
+    local source_revision="$1"
+    "$PYTHON_BIN" - "$ROOT" "$RUN_DIR" "$source_revision" "$TEST_SELECTION" <<'PY'
+import hashlib
+import json
+import pathlib
+import sys
+
+root = pathlib.Path(sys.argv[1])
+run_dir = pathlib.Path(sys.argv[2])
+source_revision = sys.argv[3]
+selection = sys.argv[4]
+descriptor = run_dir / "release-bundle-w13.json"
+registry = json.loads((root / ".config/scanner-heal-required-tests.json").read_text())
+requirements = {item["gate"]: item for item in registry["release_requirements"]}
+artifacts = {
+    "G07": {
+        "mrf_responsibility_oracle": run_dir / "g07-mrf-responsibility" / "G07-mrf_responsibility_oracle.json",
+        "commit_boundary_crash_matrix": run_dir / "g07-mrf-responsibility" / "G07-commit_boundary_crash_matrix.json",
+    },
+    "G08": {
+        "mrf_capacity_evidence": run_dir / "g08-mrf-capacity" / "G08-mrf_capacity_evidence.json",
+        "disk_full_matrix": run_dir / "g08-mrf-capacity" / "G08-disk_full_matrix.json",
+        "replica_loss_matrix": run_dir / "g08-mrf-capacity" / "G08-replica_loss_matrix.json",
+    },
+    "P4": {
+        "mrf_scale_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_scale_measurement.json",
+        "mrf_replay_cost_measurement": run_dir / "p4-mrf-soak" / "P4-mrf_replay_cost_measurement.json",
+        "retained_responsibility_evidence": run_dir / "p4-mrf-soak" / "P4-retained_responsibility_evidence.json",
+        "mrf_cleanup_gc_soak_evidence": run_dir / "p4-mrf-soak" / "P4-mrf_cleanup_gc_soak_evidence.json",
+    },
+}
+if selection == "g07":
+    artifacts = {"G07": artifacts["G07"]}
+elif selection == "g08":
+    artifacts = {"G08": artifacts["G08"]}
+elif selection == "p4":
+    artifacts = {"P4": artifacts["P4"]}
+
+mirrors = {
+    ("G07", "mrf_responsibility_oracle"): ("crash_points", "mrf_responsibility_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
+    ("G07", "commit_boundary_crash_matrix"): ("crash_points", "commit_crash_cases", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
+    ("G08", "mrf_capacity_evidence"): ("capacity_cases",),
+    ("G08", "disk_full_matrix"): ("disk_full_cases",),
+    ("G08", "replica_loss_matrix"): ("replica_loss_cases",),
+    ("P4", "mrf_replay_cost_measurement"): ("duration_seconds", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
+    ("P4", "retained_responsibility_evidence"): ("duration_seconds", "retained_responsibility_cases", "retention_window_seconds", "idle_cleanup_observed", "verified_proof_discharge_observed", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
+    ("P4", "mrf_cleanup_gc_soak_evidence"): ("duration_seconds", "cleanup_gc_cases", "verified_idle_gc_observed", "pending_responsibilities_after_gc", "stale_journals_after_gc", "replayed_records", "responsibility_anchor_retained", "successor_snapshot_published"),
+}
+
+def digest(path: pathlib.Path) -> str:
+    hasher = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            hasher.update(chunk)
+    return hasher.hexdigest()
+
+def relative_to_descriptor(path: pathlib.Path) -> str:
+    return path.resolve(strict=True).relative_to(descriptor.parent.resolve()).as_posix()
+
+gates: dict[str, object] = {}
+for gate, gate_artifacts in artifacts.items():
+    fields: dict[str, object] = {}
+    for field, artifact in gate_artifacts.items():
+        payload = json.loads(artifact.read_text())
+        if payload.get("source_revision") != source_revision:
+            raise SystemExit(f"{gate}.{field}: source revision does not match this checkout")
+        evidence = {
+            "artifact": relative_to_descriptor(artifact),
+            "sha256": digest(artifact),
+            "evidence_type": "measured",
+            "source_revision": source_revision,
+            "run_id": payload["run_id"],
+            "measurement_window_id": payload["measurement_window_id"],
+            "started_at": payload["started_at"],
+            "finished_at": payload["finished_at"],
+            "command": payload["command"],
+            "artifact_format": "json",
+            "summary": payload["summary"],
+        }
+        for mirror in mirrors.get((gate, field), ()):
+            evidence[mirror] = payload[mirror]
+        if gate == "P4" and field == "mrf_scale_measurement":
+            evidence["duration_seconds"] = payload["duration_seconds"]
+        fields[field] = evidence
+    gates[gate] = {
+        "status": "pass",
+        "lane": requirements[gate]["lane"],
+        "evidence_type": "measured",
+        "evidence_fields": fields,
+    }
+
+descriptor.write_text(json.dumps({
+    "schema": 1,
+    "evidence": "measured",
+    "source_revision": source_revision,
+    "gates": gates,
+}, indent=2, sort_keys=True) + "\n")
+print(descriptor)
+PY
+}
+
+run_self_test() {
+    local tmp plan
+    tmp="$(mktemp -d "${TMPDIR:-/tmp}/rustfs-w13-evidence-self-test.XXXXXX")"
+    trap "rm -rf '$tmp'" EXIT
+
+    plan="$("$0" --plan-only --run-dir "$tmp/run" --test all)"
+    [[ "$plan" == *"tests=g07 g08 p4"* ]]
+    [[ "$plan" == *"soak_seconds=7200"* ]]
+    [[ "$plan" == *"run_dir=$tmp/run"* ]]
+    [[ "$(CARGO_TARGET_DIR=relative-target "$0" --plan-only --run-dir "$tmp/run" --test g07)" == *"target_dir=$ROOT/relative-target"* ]]
+
+    if "$0" --plan-only --test not-a-case >/dev/null 2>&1; then
+        echo "self-test failed: invalid test selection was accepted" >&2
+        return 1
+    fi
+    if "$0" --plan-only --test p4 --soak-seconds 10 >/dev/null 2>&1; then
+        echo "self-test failed: short P4 soak was accepted as release evidence" >&2
+        return 1
+    fi
+    mkdir -p "$tmp/nonempty/g07-mrf-responsibility"
+    : >"$tmp/nonempty/g07-mrf-responsibility/existing.json"
+    if "$0" --dry-run --run-dir "$tmp/nonempty" >/dev/null 2>&1; then
+        echo "self-test failed: non-empty evidence directory was accepted" >&2
+        return 1
+    fi
+}
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --run-dir|--out-dir)
+            require_value "$1" "$#"
+            RUN_DIR="$2"
+            shift 2
+            ;;
+        --test)
+            require_value "$1" "$#"
+            TEST_SELECTION="$2"
+            shift 2
+            ;;
+        --soak-seconds)
+            require_value "$1" "$#"
+            SOAK_SECONDS="$2"
+            shift 2
+            ;;
+        --enospc-root)
+            require_value "$1" "$#"
+            ENOSPC_ROOT="$2"
+            shift 2
+            ;;
+        --allow-short-soak)
+            ALLOW_SHORT_SOAK=1
+            shift
+            ;;
+        --allow-dirty)
+            ALLOW_DIRTY=1
+            shift
+            ;;
+        --skip-build)
+            SKIP_BUILD=1
+            shift
+            ;;
+        --plan-only|--dry-run)
+            PLAN_ONLY=1
+            shift
+            ;;
+        --self-test)
+            run_self_test
+            exit $?
+            ;;
+        --verbose)
+            VERBOSE=1
+            shift
+            ;;
+        -h|--help)
+            usage
+            exit 0
+            ;;
+        *)
+            echo "unknown option: $1" >&2
+            usage >&2
+            exit 2
+            ;;
+    esac
+done
+
+validate_test_selection
+[[ "$SOAK_SECONDS" =~ ^[0-9]+$ ]] || die "--soak-seconds must be a non-negative integer"
+CASES=()
+while IFS= read -r case_name; do
+    CASES+=("$case_name")
+done < <(case_names)
+if [[ " ${CASES[*]} " == *" p4 "* && "$SOAK_SECONDS" -lt 7200 && "$ALLOW_SHORT_SOAK" != 1 ]]; then
+    die "P4 release evidence requires at least 7200 soak seconds; pass --allow-short-soak only for diagnostics"
+fi
+if [[ -z "$RUN_DIR" ]]; then
+    OUTPUT_ROOT="$(normalize_path "${RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT:-$ROOT/target/scanner-heal-w13-evidence}")"
+    RUN_DIR="$OUTPUT_ROOT/$(date -u +%Y%m%dT%H%M%SZ)"
+else
+    RUN_DIR="$(normalize_path "$RUN_DIR")"
+fi
+
+for case_name in "${CASES[@]}"; do
+    check_empty_case_dir "$RUN_DIR/$(artifact_dir_for "$case_name")"
+done
+
+if [[ "$PLAN_ONLY" == 1 ]]; then
+    echo "run_dir=$RUN_DIR"
+    echo "out_dir=$RUN_DIR"
+    echo "tests=${CASES[*]}"
+    echo "soak_seconds=$SOAK_SECONDS"
+    echo "min_free_kib=$MIN_FREE_KIB"
+    target_dir="$(cargo_target_dir)"
+    echo "target_dir=$target_dir"
+    echo "current_binary=$target_dir/debug/rustfs"
+    echo "test_filter=rustfs-heal heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts"
+    echo "required_artifacts:"
+    if [[ " ${CASES[*]} " == *" g07 "* ]]; then
+        echo "  $RUN_DIR/g07-mrf-responsibility/G07-mrf_responsibility_oracle.json"
+        echo "  $RUN_DIR/g07-mrf-responsibility/G07-commit_boundary_crash_matrix.json"
+    fi
+    if [[ " ${CASES[*]} " == *" g08 "* ]]; then
+        echo "  $RUN_DIR/g08-mrf-capacity/G08-mrf_capacity_evidence.json"
+        echo "  $RUN_DIR/g08-mrf-capacity/G08-disk_full_matrix.json"
+        echo "  $RUN_DIR/g08-mrf-capacity/G08-replica_loss_matrix.json"
+        if [[ -n "$ENOSPC_ROOT" ]]; then
+            echo "enospc_root=$(normalize_path "$ENOSPC_ROOT")"
+        elif [[ "$(uname -s)" == "Linux" ]]; then
+            echo "enospc_root=$RUN_DIR/enospc-root"
+            echo "enospc_tmpfs_size=$ENOSPC_TMPFS_SIZE"
+        else
+            echo "enospc_root=required-for-non-linux"
+        fi
+    fi
+    if [[ " ${CASES[*]} " == *" p4 "* ]]; then
+        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_scale_measurement.json"
+        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_replay_cost_measurement.json"
+        echo "  $RUN_DIR/p4-mrf-soak/P4-retained_responsibility_evidence.json"
+        echo "  $RUN_DIR/p4-mrf-soak/P4-mrf_cleanup_gc_soak_evidence.json"
+    fi
+    exit 0
+fi
+
+cd "$ROOT"
+if [[ "$ALLOW_DIRTY" != 1 && -n "$(git status --porcelain --untracked-files=no)" ]]; then
+    echo "commit tracked source changes before creating release evidence, or pass --allow-dirty for local diagnostics" >&2
+    exit 1
+fi
+if [[ -e "$RUN_DIR" ]]; then
+    die "evidence run directory already exists: $RUN_DIR"
+fi
+mkdir -p "$RUN_DIR/logs"
+trap cleanup_enospc_root EXIT
+if [[ -n "${TMPDIR:-}" ]]; then
+    mkdir -p "$TMPDIR"
+    ensure_min_free_space "$TMPDIR"
+fi
+ensure_min_free_space "$RUN_DIR"
+prepare_enospc_root
+
+SOURCE_REVISION="$(git rev-parse HEAD)"
+printf '%s\n' "$SOURCE_REVISION" >"$RUN_DIR/source-revision.txt"
+
+if [[ "$SKIP_BUILD" != 1 ]]; then
+    run_logged build-current cargo build --locked -p rustfs --bin rustfs
+    write_rustfs_features_stamp
+fi
+
+selection_csv="$(IFS=,; echo "${CASES[*]}")"
+run_logged w13-mrf-evidence env \
+    RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR="$RUN_DIR" \
+    RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION="$SOURCE_REVISION" \
+    RUSTFS_SCANNER_HEAL_W13_SELECTION="$selection_csv" \
+    RUSTFS_SCANNER_HEAL_W13_SOAK_SECONDS="$SOAK_SECONDS" \
+    RUSTFS_SCANNER_HEAL_W13_ALLOW_SHORT_SOAK="$ALLOW_SHORT_SOAK" \
+    RUSTFS_SCANNER_HEAL_W13_RUN_ID="w13-mrf-release-evidence-run" \
+    RUSTFS_SCANNER_HEAL_W13_WINDOW_ID="w13-mrf-release-evidence-window" \
+    RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT="$ENOSPC_ROOT" \
+    RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES="${RUSTFS_SCANNER_HEAL_W13_ENOSPC_FILL_LIMIT_BYTES:-67108864}" \
+    cargo test --locked -p rustfs-heal --lib heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts \
+        -- --ignored --exact --nocapture
+
+validate_artifacts "$SOURCE_REVISION"
+DESCRIPTOR="$(write_release_descriptor "$SOURCE_REVISION")"
+if [[ " ${CASES[*]} " == *" g07 "* ]]; then
+    check_release_gate "$DESCRIPTOR" G07
+fi
+if [[ " ${CASES[*]} " == *" g08 "* ]]; then
+    check_release_gate "$DESCRIPTOR" G08
+fi
+if [[ " ${CASES[*]} " == *" p4 "* ]]; then
+    if [[ "$ALLOW_SHORT_SOAK" == 1 && "$SOAK_SECONDS" -lt 7200 ]]; then
+        echo "SKIP: P4 release bundle gate validation for diagnostic short soak"
+    else
+        check_release_gate "$DESCRIPTOR" P4
+    fi
+fi
+echo "Scanner/Heal W13 MRF release descriptors verified: $DESCRIPTOR"
+echo "Scanner/Heal W13 MRF evidence verified: $RUN_DIR"

scripts/test_scanner_heal_w13_mrf_evidence.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+RUNNER="${PROJECT_ROOT}/scripts/run_scanner_heal_w13_mrf_evidence.sh"
+TMP_DIR="$(mktemp -d)"
+
+cleanup() {
+  rm -rf "$TMP_DIR"
+}
+trap cleanup EXIT
+
+bash -n "$RUNNER"
+
+bash "$RUNNER" --help >"$TMP_DIR/help.out"
+rg -q "RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT" "$TMP_DIR/help.out"
+rg -q "RUSTFS_SCANNER_HEAL_W13_ENOSPC_ROOT" "$TMP_DIR/help.out"
+rg -q "G07-mrf_responsibility_oracle.json" "$TMP_DIR/help.out"
+rg -q "G08-disk_full_matrix.json" "$TMP_DIR/help.out"
+rg -q "P4-mrf_cleanup_gc_soak_evidence.json" "$TMP_DIR/help.out"
+
+env -u CARGO_TARGET_DIR bash "$RUNNER" \
+  --dry-run \
+  --out-dir "$TMP_DIR/evidence" >"$TMP_DIR/dry-run.out"
+
+rg -q "tests=g07 g08 p4" "$TMP_DIR/dry-run.out"
+rg -q "test_filter=rustfs-heal heal::mrf_queue::tests::w13_mrf_release_evidence_outputs_bundle_artifacts" "$TMP_DIR/dry-run.out"
+rg -q "target_dir=$PROJECT_ROOT/target" "$TMP_DIR/dry-run.out"
+
+RUSTFS_SCANNER_HEAL_W13_OUTPUT_ROOT="$TMP_DIR/root-out" \
+  bash "$RUNNER" --dry-run --test g07 >"$TMP_DIR/dry-run-output-root.out"
+rg -q "run_dir=$TMP_DIR/root-out/" "$TMP_DIR/dry-run-output-root.out"
+
+CARGO_TARGET_DIR="$TMP_DIR/shared-target" bash "$RUNNER" \
+  --dry-run \
+  --out-dir "$TMP_DIR/evidence-with-target" \
+  --test g08 \
+  --enospc-root "$TMP_DIR/enospc" >"$TMP_DIR/dry-run-target.out"
+rg -q "target_dir=$TMP_DIR/shared-target" "$TMP_DIR/dry-run-target.out"
+rg -q "current_binary=$TMP_DIR/shared-target/debug/rustfs" "$TMP_DIR/dry-run-target.out"
+rg -q "enospc_root=$TMP_DIR/enospc" "$TMP_DIR/dry-run-target.out"
+
+mkdir -p "$TMP_DIR/nonempty/g07-mrf-responsibility"
+touch "$TMP_DIR/nonempty/g07-mrf-responsibility/existing.json"
+if bash "$RUNNER" --dry-run --out-dir "$TMP_DIR/nonempty" >"$TMP_DIR/nonempty.out" 2>"$TMP_DIR/nonempty.err"; then
+  echo "W13 runner should reject non-empty evidence case directories" >&2
+  exit 1
+fi
+rg -q "evidence case directory is not empty" "$TMP_DIR/nonempty.err"
+
+if bash "$RUNNER" --plan-only --test p4 --soak-seconds 10 >/dev/null 2>&1; then
+  echo "W13 runner should reject short P4 release soak without --allow-short-soak" >&2
+  exit 1
+fi
+
+bash "$RUNNER" --self-test