diff --git a/crates/lifecycle/src/core.rs b/crates/lifecycle/src/core.rs
index 75d20fd18..d125723ec 100644
--- a/crates/lifecycle/src/core.rs
+++ b/crates/lifecycle/src/core.rs
@@ -67,6 +67,7 @@ const ERR_LIFECYCLE_RULE_MUST_HAVE_ACTION: &str = "Rule must have at least one o
 const ERR_LIFECYCLE_PREFIX_FILTER_CONFLICT: &str = "Legacy Prefix and Filter cannot both be present in a lifecycle rule. Use Filter.Prefix instead of the top-level Prefix element.";
 const ERR_LIFECYCLE_INVALID_NEWER_NONCURRENT_VERSIONS: &str = "'NewerNoncurrentVersions' must be a non-negative integer";
 const ERR_LIFECYCLE_FILTER_AND_TOO_FEW_PREDICATES: &str = "Filter And must contain at least two predicates";
+const ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES: &str = "Filter has too many predicates";
 const ERR_LIFECYCLE_FILTER_DUPLICATE_TAG_KEY: &str = "Filter must not repeat a tag key";
 const ERR_LIFECYCLE_FILTER_INVALID_TAG: &str = "Tag key must be 1-128 characters and tag value must be at most 256 characters";
 const ERR_LIFECYCLE_FILTER_NEGATIVE_SIZE: &str = "ObjectSizeGreaterThan and ObjectSizeLessThan must not be negative";
@@ -286,11 +287,14 @@ impl RuleValidate for LifecycleRule {
 /// `Filter` as "applies to every object in the bucket", and rejecting it would
 /// break the most common way to write an unconditional rule.
 fn validate_lifecycle_filter(filter: &LifecycleRuleFilter) -> Result<(), std::io::Error> {
-    // AWS S3 allows multiple top-level predicates (Prefix, Tag, ObjectSize*)
-    // as siblings in Filter without an explicit And wrapper — botocore sends
-    // this layout when a rule combines Prefix with Tag. The evaluation code
-    // already treats sibling predicates as implicit AND, so we validate each
-    // predicate individually rather than enforcing a single-predicate limit.
+    let top_level_predicates = usize::from(filter.prefix.is_some())
+        + usize::from(filter.tag.is_some())
+        + usize::from(filter.object_size_greater_than.is_some())
+        + usize::from(filter.object_size_less_than.is_some())
+        + usize::from(filter.and.is_some());
+    if top_level_predicates > 1 {
+        return Err(malformed_xml_error(ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES));
+    }
 
     if let Some(tag) = filter.tag.as_ref() {
         validate_lifecycle_tag(tag)?;
@@ -4713,7 +4717,7 @@ mod tests {
                     tag: Some(tag("env", "prod")),
                     ..Default::default()
                 },
-                expected: None,
+                expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
             },
             Case {
                 name: "prefix alongside And",
@@ -4726,7 +4730,7 @@ mod tests {
                     }),
                     ..Default::default()
                 },
-                expected: None,
+                expected: Some((ERR_LIFECYCLE_FILTER_TOO_MANY_PREDICATES, LIFECYCLE_MALFORMED_XML_ERROR_KIND)),
             },
             Case {
                 name: "And with a single member",
diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs
index 65f76ac2a..afd54825d 100644
--- a/crates/scanner/src/scanner.rs
+++ b/crates/scanner/src/scanner.rs
@@ -1910,10 +1910,10 @@ where
         .as_ref()
         .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone()));
     let remote_lease_release_safe = Arc::new(AtomicBool::new(true));
-    let mut usage_persist_outcome = match publication_defer_reason {
+    let mut usage_publication_result = match publication_defer_reason {
         Some(reason) => {
             drop(receiver);
-            DataUsagePersistOutcome::Deferred(reason)
+            DataUsagePublicationResult::from(DataUsagePersistOutcome::Deferred(reason))
         }
         None => {
             // ScannerIO emits its complete or observational update only after
@@ -1928,6 +1928,11 @@ where
                 .as_ref()
                 .map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect())
                 .unwrap_or_default();
+            let ack_expectation = scan_result
+                .as_ref()
+                .ok()
+                .filter(|result| result.has_dirty_usage_to_acknowledge())
+                .and_then(ScannerCycleResult::publication_expectation);
             let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
                 store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
                     ctx_clone,
@@ -1940,6 +1945,7 @@ where
                         remote_lease_deadline,
                         remote_lease_fence,
                     )
+                    .with_ack_expectation(ack_expectation)
                     .with_remote_lease_tokens(remote_lease_tokens)
                     .with_lease_release_flag(remote_lease_release_safe_for_task),
                     move || {
@@ -1973,7 +1979,7 @@ where
                         error = %err,
                         "Scanner data usage persistence task failed"
                     );
-                    DataUsagePersistOutcome::Failed
+                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
                 }
                 DataUsagePersistTaskResult::Cancelled => {
                     debug!(
@@ -1985,7 +1991,7 @@ where
                         state = "usage_persist_task_cancelled",
                         "Scanner data usage persistence task cancelled"
                     );
-                    DataUsagePersistOutcome::Failed
+                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
                 }
                 DataUsagePersistTaskResult::TimedOut => {
                     error!(
@@ -1998,11 +2004,12 @@ where
                         state = "usage_persist_task_timed_out",
                         "Scanner data usage persistence task timed out"
                     );
-                    DataUsagePersistOutcome::Failed
+                    DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed)
                 }
             }
         }
     };
+    let mut usage_persist_outcome = usage_publication_result.outcome();
     let lease_expired = remote_publication_leases
         .as_ref()
         .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid()));
@@ -2202,8 +2209,9 @@ where
         };
     }
 
+    usage_publication_result.restrict_outcome(usage_persist_outcome);
     let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) =
-        finalize_scanner_cycle_result(scan_cycle_result, usage_persist_outcome);
+        finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result);
     let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() {
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
@@ -3437,21 +3445,35 @@ fn scanner_cycle_completion_outcome(
 
 fn finalize_scanner_cycle_result(
     scan_cycle_result: crate::scanner_io::ScannerCycleResult,
-    usage_persist_outcome: DataUsagePersistOutcome,
+    publication: DataUsagePublicationResult,
 ) -> (ScannerCycleOutcome, bool, Vec<ScannerDirtyUsageAcknowledgement>) {
+    let (usage_persist_outcome, proof) = publication.into_parts();
     let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome);
-    let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work();
     let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete
         && matches!(
             usage_persist_outcome,
             DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable
-        );
+        )
+        && scan_cycle_result.publication_expectation().as_ref().is_some_and(|expected| {
+            proof
+                .as_ref()
+                .is_some_and(|proof| proof.verified_version_for(expected).is_some())
+        });
+    let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work()
+        || (scan_cycle_result.has_dirty_usage_to_acknowledge() && !durable_complete_snapshot);
     let remote_dirty_usage_acknowledgements = if durable_complete_snapshot {
-        scan_cycle_result.acknowledge_durable_usage()
+        match proof {
+            Some(proof) => scan_cycle_result.acknowledge_durable_usage(&proof),
+            None => Vec::new(),
+        }
     } else {
         Vec::new()
     };
-    (completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements)
+    (
+        completion_outcome,
+        pending_maintenance_work || crate::scanner_io::dirty_usage_buckets_pending(),
+        remote_dirty_usage_acknowledgements,
+    )
 }
 
 fn scanner_cycle_completion_outcome_for_result(
@@ -3551,6 +3573,7 @@ use activity::*;
 use backlog::*;
 use cycle_state::*;
 use leadership::*;
+pub(crate) use usage_store::RootPublicationProof;
 use usage_store::*;
 
 pub use activity::scanner_topology_digest;
diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs
index 04d548404..cfcd37fe9 100644
--- a/crates/scanner/src/scanner/tests.rs
+++ b/crates/scanner/src/scanner/tests.rs
@@ -36,7 +36,10 @@ use tokio::time::{Duration, advance};
 
 const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
 
+mod quota_reset_preservation;
+
 mod recovery_control;
+mod scoped_ack_publication;
 
 async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
     setup_scanner_cycle_store_with_usage_baseline(true).await
@@ -5288,6 +5291,103 @@ async fn scanner_usage_state_reset_resumes_every_cleanup_boundary_without_rewrit
     }
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_usage_state_reset_resumes_real_store_cleanup_boundaries_after_reopen() {
+    let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str();
+    let cleanup_paths = [
+        format!("{primary_path}.bkp"),
+        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
+        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str().to_string(),
+    ];
+
+    for completed in 0..=cleanup_paths.len() {
+        let (_temp_dir, store) = setup_scanner_cycle_store().await;
+        let cycle = CurrentCycle {
+            current: 12,
+            next: 42,
+            cycle_completed: vec![Utc::now()],
+            started: Utc::now(),
+        };
+        save_config(
+            store.clone(),
+            DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+            encode_scanner_cycle_state(&cycle, 3).expect("cycle state should encode"),
+        )
+        .await
+        .expect("cycle state should persist");
+        let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::UNIX_EPOCH, Some(3));
+        save_config(
+            store.clone(),
+            primary_path,
+            serde_json::to_vec(&marker).expect("usage reset marker should encode"),
+        )
+        .await
+        .expect("usage reset marker should persist");
+
+        for path in cleanup_paths.iter().skip(completed) {
+            let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+            usage.scanner_epoch = Some(1);
+            usage.scanner_cycle = Some(12);
+            save_config(store.clone(), path, serde_json::to_vec(&usage).expect("cleanup slot should encode"))
+                .await
+                .expect("cleanup slot should persist");
+        }
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            save_config(store.clone(), path, b"retain".to_vec())
+                .await
+                .expect("unrelated state should persist before reopen");
+        }
+
+        let restarted = restart_scanner_cycle_store_from(&store).await;
+        let intent_before = read_config_with_revision(restarted.clone(), primary_path)
+            .await
+            .expect("reopened reset intent should be readable");
+
+        let result = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), restarted.clone())
+            .await
+            .expect("reopened usage reset should complete");
+
+        assert_eq!(result.leader_epoch, 3, "boundary {completed}");
+        assert_eq!(result.next_cycle, 42, "boundary {completed}");
+        assert_eq!(result.reset_paths.len(), cleanup_paths.len() + 1 - completed, "boundary {completed}");
+        assert_eq!(
+            read_config_with_revision(restarted.clone(), primary_path)
+                .await
+                .expect("completed reset intent should remain readable"),
+            intent_before,
+            "boundary {completed}: resumed cleanup must not rewrite the reset intent"
+        );
+
+        let (floor, state) = persisted_usage_floor_for_startup(restarted.clone(), false)
+            .await
+            .expect("completed reset marker should remain resumable");
+        assert_eq!(floor.leader_epoch, 3, "boundary {completed}");
+        assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending, "boundary {completed}");
+        assert!(
+            persisted_usage_floor(restarted.clone()).await.is_err(),
+            "boundary {completed}: bootstrap marker must not become an authoritative floor"
+        );
+
+        for path in &cleanup_paths {
+            assert!(
+                matches!(read_config(restarted.clone(), path).await, Err(EcstoreError::ConfigNotFound)),
+                "boundary {completed}: reset should remove stale usage slot {path}"
+            );
+        }
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            assert_eq!(
+                read_config(restarted.clone(), path)
+                    .await
+                    .expect("unrelated state should survive reopened reset"),
+                b"retain",
+                "boundary {completed}: reset must preserve non-scanner-state config"
+            );
+        }
+    }
+}
+
 #[tokio::test]
 async fn scanner_usage_state_reset_stops_usage_fence_after_owner_loss() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -6184,7 +6284,7 @@ async fn coordinator_classifies_an_expired_publication_lease() {
         .await;
 
     assert_eq!(
-        outcome,
+        outcome.outcome(),
         DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
     );
     assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT");
@@ -7325,7 +7425,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
 
 #[test]
 #[serial]
-fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
+fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() {
     crate::scanner_io::clear_dirty_usage_bucket("photos");
     crate::scanner_io::record_dirty_usage_bucket("photos");
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
@@ -7337,17 +7437,19 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
     };
     let unsaved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot.clone()))
         .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate);
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate.into());
     assert_eq!(outcome, ScannerCycleOutcome::Failed);
     assert!(acknowledgements.is_empty());
     assert!(crate::scanner_io::dirty_usage_buckets_pending());
 
     let saved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot))
-        .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved);
+        .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement]);
+    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved.into());
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
-    assert_eq!(acknowledgements, vec![remote_acknowledgement]);
-    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
+    assert!(acknowledgements.is_empty());
+    assert!(pending);
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
 }
 
 #[test]
@@ -7359,7 +7461,7 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
     let deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
 
     let (outcome, _, acknowledgements) =
-        finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+        finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement).into());
 
     assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
     assert!(acknowledgements.is_empty());
@@ -7379,7 +7481,7 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
     )
     .with_observational_snapshot_published(true);
 
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved);
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved.into());
 
     assert_eq!(outcome, ScannerCycleOutcome::Partial);
     assert!(acknowledgements.is_empty());
@@ -7415,17 +7517,20 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
 
 #[test]
 #[serial]
-fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
+fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() {
     crate::scanner_io::clear_dirty_usage_bucket("photos");
     crate::scanner_io::record_dirty_usage_bucket("photos");
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable);
+    let (outcome, pending, acknowledgements) =
+        finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable.into());
 
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
     assert!(acknowledgements.is_empty());
-    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
+    assert!(pending);
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
 }
 
 #[test]
@@ -7436,7 +7541,8 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable);
+    let (outcome, _, acknowledgements) =
+        finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable.into());
 
     assert_eq!(outcome, ScannerCycleOutcome::Completed);
     assert!(acknowledgements.is_empty());
@@ -7452,7 +7558,7 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
     let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
 
     let superseded = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Superseded, Some(dirty_snapshot));
-    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved);
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved.into());
 
     assert_eq!(outcome, ScannerCycleOutcome::Superseded);
     assert!(acknowledgements.is_empty());
@@ -8881,7 +8987,7 @@ fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acqui
     ]);
     let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(
         result,
-        DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")),
+        DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")).into(),
     );
     assert_eq!(
         outcome,
diff --git a/crates/scanner/src/scanner/tests/quota_reset_preservation.rs b/crates/scanner/src/scanner/tests/quota_reset_preservation.rs
new file mode 100644
index 000000000..a1de2efc4
--- /dev/null
+++ b/crates/scanner/src/scanner/tests/quota_reset_preservation.rs
@@ -0,0 +1,148 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use crate::storage_api::owner::ObjectOperations as _;
+
+const BUCKET: &str = "quota-reset-preservation";
+const OPERATION: &str = "00000000-0000-0000-0000-000000000002";
+
+async fn reservation_fixture() -> (tempfile::TempDir, Arc<ECStore>, Uuid, String, Vec<u8>) {
+    let (directory, store) = setup_scanner_cycle_store().await;
+    store
+        .make_bucket(BUCKET, &crate::storage_api::scan::MakeBucketOptions::default())
+        .await
+        .expect("create the reservation fixture bucket through its owner");
+    let incarnation = store
+        .bucket_incarnation_id_from_disk(BUCKET)
+        .await
+        .expect("durable bucket incarnation");
+    assert!(!incarnation.is_nil());
+    let path = format!("config/quota-ledger/{BUCKET}.json");
+    let bytes = serde_json::to_vec(&serde_json::json!({
+        "version": 1,
+        "bucket_incarnation": incarnation,
+        "quota_revision_unix_nanos": 1,
+        "accounted_usage": 100,
+        "reservations": {
+            OPERATION: {
+                "object": "pending-object",
+                "old_size": 0,
+                "new_size": 64,
+                "created_at": 1,
+                "pool_index": 0,
+                "set_index": 0,
+                "commit_started": true
+            }
+        }
+    }))
+    .expect("encode the committed reservation fixture");
+    save_config(store.clone(), &path, bytes.clone())
+        .await
+        .expect("persist reservation bytes through the real storage owner");
+    (directory, store, incarnation, path, bytes)
+}
+
+async fn assert_reservation_retained(store: &Arc<ECStore>, path: &str, expected: &[u8], incarnation: Uuid) {
+    let bytes = read_config(store.clone(), path)
+        .await
+        .expect("read the actual reservation ledger");
+    assert_eq!(bytes, expected, "scanner reset must not rewrite the reservation ledger");
+    let ledger: serde_json::Value = serde_json::from_slice(&bytes).expect("persisted ledger JSON");
+    assert_eq!(ledger["version"], 1);
+    assert_eq!(ledger["bucket_incarnation"], incarnation.to_string());
+    assert_eq!(ledger["accounted_usage"], 100);
+    let reservations = ledger["reservations"].as_object().expect("reservation map");
+    assert_eq!(reservations.len(), 1);
+    let pending = &reservations[OPERATION];
+    assert_eq!(pending["old_size"], 0);
+    assert_eq!(pending["new_size"], 64);
+    assert_eq!(pending["commit_started"], true);
+    assert_eq!(
+        store
+            .bucket_incarnation_id_from_disk(BUCKET)
+            .await
+            .expect("owner incarnation after restart"),
+        incarnation
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn quota_reset_preservation_survives_storage_owner_reconstruction() {
+    let (_directory, store, incarnation, path, bytes) = reservation_fixture().await;
+    let reset = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
+        .await
+        .expect("reset scanner usage through the fenced production entry");
+    assert_eq!(reset.usage_state, "bootstrap-pending");
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    assert!(
+        !Arc::ptr_eq(&store, &restarted),
+        "the assertion must read through a newly constructed ECStore"
+    );
+    assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
+    let usage = read_config(restarted.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("read reset usage through the reconstructed owner");
+    let usage: DataUsageInfo = serde_json::from_slice(&usage).expect("bootstrap usage JSON");
+    assert!(data_usage_info_is_bootstrap_pending(&usage));
+    assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
+}
+
+#[tokio::test]
+#[serial]
+async fn quota_reset_preservation_unknown_protocol_rejects_put_after_restart() {
+    for quota_shape in ["zero", "null", "missing"] {
+        let (_directory, store, incarnation, path, bytes) = reservation_fixture().await;
+        let mut quota = serde_json::json!({
+            "quota_type": "Hard",
+            "reservation_protocol": 2,
+            "reservation_quota": 1024
+        });
+        match quota_shape {
+            "zero" => quota["quota"] = serde_json::json!(0),
+            "null" => quota["quota"] = serde_json::Value::Null,
+            "missing" => {}
+            _ => unreachable!("fixed quota shapes"),
+        }
+        let unknown_quota = serde_json::to_vec(&quota).expect("unknown but syntactically valid quota protocol");
+        store
+            .update_bucket_metadata_config(BUCKET, rustfs_config::QUOTA_CONFIG_FILE, unknown_quota)
+            .await
+            .expect("persist a future protocol using the real metadata owner");
+        assert_eq!(
+            store
+                .bucket_incarnation_id_from_disk(BUCKET)
+                .await
+                .expect("same metadata owner incarnation"),
+            incarnation
+        );
+        reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone())
+            .await
+            .expect("scanner reset must not change quota metadata");
+        let restarted = restart_scanner_cycle_store_from(&store).await;
+        assert!(!Arc::ptr_eq(&store, &restarted));
+        assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
+        let mut reader = PutObjReader::from_vec(b"must-not-commit".to_vec());
+        let result = restarted.pools[0].disk_set[0]
+            .put_object(BUCKET, "rejected-object", &mut reader, &ObjectOptions::default())
+            .await;
+        let error = match result {
+            Err(error) => error,
+            Ok(_) => panic!("unknown reservation protocol with quota={quota_shape} must not admit a PUT"),
+        };
+        assert!(
+            matches!(error, EcstoreError::PartMissingOrCorrupt),
+            "unexpected protocol rejection: {error}"
+        );
+        let missing = restarted.pools[0].disk_set[0]
+            .get_object_info(BUCKET, "rejected-object", &ObjectOptions::default())
+            .await
+            .expect_err("the rejected PUT must not create an object");
+        assert!(
+            matches!(missing, EcstoreError::FileNotFound | EcstoreError::ObjectNotFound(_, _)),
+            "object absence must not be confused with another storage failure: {missing}"
+        );
+        assert_reservation_retained(&restarted, &path, &bytes, incarnation).await;
+    }
+}
diff --git a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs
new file mode 100644
index 000000000..39deae1e5
--- /dev/null
+++ b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs
@@ -0,0 +1,490 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::super::usage_store::DataUsagePublicationResult;
+use super::*;
+use crate::scanner_io::ScannerBucketScanScope;
+use rustfs_utils::path::path_join_buf;
+use sha2::Digest;
+use std::time::SystemTime;
+
+const PROOF_BUCKET: &str = "publication-proof-bucket";
+const PROOF_EPOCH: u64 = 7;
+const PROOF_CYCLE: u64 = 11;
+
+async fn settle_namespace_commits(store: &ECStore) {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        while store.scanner_data_usage_publication_blocked().await {
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+    })
+    .await
+    .expect("fixture namespace commits must settle before collecting complete coverage");
+}
+
+async fn complete_candidate(store: &Arc<ECStore>, cycle: u64) -> (crate::scanner_io::ScannerCycleResult, DataUsageInfo) {
+    settle_namespace_commits(store).await;
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &ctx,
+        ScannerCycleBudgetConfig {
+            max_objects: Some(8),
+            ..Default::default()
+        },
+    );
+    let (updates, mut receiver) = mpsc::channel(1);
+    let result = crate::scanner_io::nsscanner_with_storage_status_scoped(
+        store.as_ref(),
+        crate::scanner_io::ScannerCycleRequest {
+            ctx,
+            budget,
+            updates,
+            want_cycle: cycle,
+            leader_epoch: PROOF_EPOCH,
+            scan_mode: HealScanMode::Normal,
+            scan_scope: ScannerBucketScanScope::default(),
+            persisted_usage_baseline: None,
+            observed_usage_candidate: None,
+            requires_full_scan: true,
+            service_cohort: None,
+            resolved_scope_observer: None,
+        },
+    )
+    .await
+    .expect("real scanner must produce the fixture candidate");
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    let candidate = receiver.recv().await.expect("complete scanner snapshot");
+    assert!(candidate.usage_snapshot_complete);
+    assert_eq!(candidate.usage_snapshot_converged, Some(true));
+    assert_eq!(candidate.scanner_cycle, Some(cycle));
+    assert_eq!(candidate.scanner_epoch, Some(PROOF_EPOCH));
+    (result, candidate)
+}
+
+async fn candidate_store() -> (tempfile::TempDir, Arc<ECStore>) {
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+    let (directory, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
+    store
+        .make_bucket(PROOF_BUCKET, &crate::storage_api::scan::MakeBucketOptions::default())
+        .await
+        .expect("create proof fixture bucket through the owner");
+    let mut reader = PutObjReader::from_vec(b"proof".to_vec());
+    store.pools[0].disk_set[0]
+        .put_object(PROOF_BUCKET, "initial", &mut reader, &ObjectOptions::default())
+        .await
+        .expect("persist fixture object through the owner");
+    crate::scanner_io::record_dirty_usage_bucket(PROOF_BUCKET);
+    settle_namespace_commits(&store).await;
+    (directory, store)
+}
+
+async fn read_root(store: &Arc<ECStore>) -> (Option<Vec<u8>>, DataUsageCacheRevision) {
+    read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("read actual v2 root bytes and revision")
+}
+
+async fn publish_candidate(
+    store: &Arc<ECStore>,
+    scan: &crate::scanner_io::ScannerCycleResult,
+    candidate: DataUsageInfo,
+    baseline: Option<DataUsagePersistBaseline>,
+) -> DataUsagePublicationResult {
+    let expectation = scan.publication_expectation();
+    assert!(expectation.is_some(), "only a real complete scan may supply the expectation");
+    let (sender, receiver) = mpsc::channel(1);
+    sender.send(candidate).await.expect("enqueue the real scan candidate");
+    drop(sender);
+    store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence(
+        CancellationToken::new(),
+        store.clone(),
+        receiver,
+        Some(PROOF_EPOCH),
+        baseline,
+        ScannerPublicationFence::new(scan.publication_epoch(), None, None).with_ack_expectation(expectation),
+        || async { None },
+    )
+    .await
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_companion_only_does_not_authorize_root_ack() {
+    for companion in [
+        format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        LEGACY_DATA_USAGE_OBJ_NAME_PATH.to_string(),
+        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+    ] {
+        let (_directory, store) = candidate_store().await;
+        let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+        let bytes = serde_json::to_vec(&candidate).expect("actual candidate JSON");
+        save_config(store.clone(), &companion, bytes.clone())
+            .await
+            .expect("persist the companion on real disks");
+        let baseline = read_data_usage_persist_baseline(store.clone())
+            .await
+            .expect("companion fallback baseline");
+        assert_eq!(baseline.data.as_deref(), Some(bytes.as_slice()));
+        assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
+        assert_eq!(read_root(&store).await.0, None);
+        let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
+
+        let publication = publish_candidate(&store, &scan, candidate, Some(baseline)).await;
+        assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
+        let (_, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
+        assert!(pending, "unacknowledged durable companion work must remain pending");
+        assert!(acknowledgements.is_empty());
+        assert_eq!(
+            crate::scanner_io::dirty_usage_buckets_for_tests(),
+            dirty,
+            "a companion is not the v2 root target"
+        );
+        assert_eq!(read_root(&store).await, (None, DataUsageCacheRevision::Missing));
+        assert_eq!(read_config(store.clone(), &companion).await.expect("companion retained"), bytes);
+    }
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_actual_root_readback_accepts_semantic_json_equivalence() {
+    let (_directory, store) = candidate_store().await;
+    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    let canonical = serde_json::to_vec(&candidate).expect("candidate encoding");
+    let mut value = serde_json::to_value(&candidate).expect("candidate value");
+    value
+        .as_object_mut()
+        .expect("usage object")
+        .insert("fixture_unknown_field".into(), serde_json::json!({"retained": true}));
+    let different_bytes = serde_json::to_vec_pretty(&value).expect("noncanonical primary JSON");
+    assert_ne!(different_bytes, canonical);
+    assert_eq!(
+        serde_json::from_slice::<DataUsageInfo>(&different_bytes).expect("semantic primary"),
+        candidate
+    );
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), different_bytes.clone())
+        .await
+        .expect("persist actual primary representation");
+    let before = read_root(&store).await;
+    assert!(matches!(&before.1, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
+    let baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("real primary revision");
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+
+    let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
+    assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
+    let (_, proof) = publication.into_parts();
+    let proof = proof.expect("actual primary readback must produce its own root proof");
+    let expected = scan.publication_expectation().expect("real scan expectation");
+    let (etag, raw_digest) = proof.verified_version_for(&expected).expect("proof must bind this candidate");
+    let DataUsageCacheRevision::Etag(expected_etag) = &before.1 else { panic!("actual root ETag") };
+    assert_eq!(etag, expected_etag);
+    let expected_digest: [u8; 32] = sha2::Sha256::digest(&different_bytes).into();
+    assert_eq!(
+        *raw_digest, expected_digest,
+        "proof must record actual bytes, not reserialized candidate bytes"
+    );
+    // Obtain another proof through the same real readback path rather than
+    // fabricating a publication result from the inspected proof above.
+    let publication = publish_candidate(&store, &scan, candidate, None).await;
+    let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
+    assert_eq!(outcome, ScannerCycleOutcome::Completed);
+    assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
+    assert!(
+        !crate::scanner_io::dirty_usage_buckets_pending(),
+        "actual root bytes plus a real revision authorize this scan"
+    );
+    assert_eq!(read_root(&store).await, before, "readback must not rewrite unknown fields or whitespace");
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_successful_root_cas_authorizes_its_scan() {
+    let (_directory, store) = candidate_store().await;
+    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    let baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("initial root revision");
+    assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
+    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
+    let (bytes, revision) = read_root(&store).await;
+    assert!(matches!(revision, DataUsageCacheRevision::Etag(etag) if !etag.is_empty()));
+    assert_eq!(
+        serde_json::from_slice::<DataUsageInfo>(&bytes.expect("actual saved root")).expect("root JSON"),
+        candidate
+    );
+    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
+    assert_eq!(outcome, ScannerCycleOutcome::Completed);
+    assert!(!pending);
+    assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets");
+    assert!(
+        !crate::scanner_io::dirty_usage_buckets_pending(),
+        "the real root CAS must authorize its matching scan"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_proof() {
+    let (_directory, store) = candidate_store().await;
+    let bootstrap = scanner_usage_bootstrap_marker(SystemTime::now(), Some(PROOF_EPOCH));
+    save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&bootstrap).expect("bootstrap root encoding"),
+    )
+    .await
+    .expect("persist authoritative bootstrap root");
+    let (prior_scan, mut observed_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    // Seed a complete but unconverged observation from real scanner coverage;
+    // the production writer attaches its authoritative baseline identity.
+    observed_candidate.usage_snapshot_converged = Some(false);
+    let observation = publish_candidate(&store, &prior_scan, observed_candidate, None).await;
+    let (outcome, proof) = observation.into_parts();
+    assert_eq!(outcome, DataUsagePersistOutcome::Saved);
+    assert!(proof.is_none(), "an observational write cannot authorize a root ACK");
+    let (root_before, revision_before) = read_root(&store).await;
+    let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("read real persisted observation");
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+    let (updates, mut receiver) = mpsc::channel(1);
+    let (observer, selected) = tokio::sync::oneshot::channel();
+    let scan = crate::scanner_io::nsscanner_with_storage_status_scoped(
+        store.as_ref(),
+        crate::scanner_io::ScannerCycleRequest {
+            ctx,
+            budget,
+            updates,
+            want_cycle: PROOF_CYCLE + 1,
+            leader_epoch: PROOF_EPOCH,
+            scan_mode: HealScanMode::Normal,
+            scan_scope: ScannerBucketScanScope::default(),
+            persisted_usage_baseline: root_before.clone().map(Bytes::from),
+            observed_usage_candidate: Some(Bytes::from(observed)),
+            requires_full_scan: false,
+            service_cohort: None,
+            resolved_scope_observer: Some(observer),
+        },
+    )
+    .await
+    .expect("observation-backed scope must run through the real scanner");
+    let scope = selected.await.expect("production resolver decision");
+    assert_eq!(scope.selected_buckets_for_tests(), Some(&HashSet::from([PROOF_BUCKET.to_string()])));
+    assert_eq!(scan.status, ScannerCycleStatus::Complete);
+    let expectation = scan.publication_expectation().expect("reused coverage must be revalidated");
+    assert!(
+        !expectation.same_candidate(&prior_scan.publication_expectation().expect("prior real candidate")),
+        "the observation cannot transfer the previous scan's expectation"
+    );
+    assert_eq!(read_root(&store).await, (root_before, revision_before));
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    let candidate = receiver.recv().await.expect("new validated root candidate");
+    assert_eq!(candidate.scanner_cycle, Some(PROOF_CYCLE + 1));
+    assert_eq!(candidate.usage_snapshot_converged, Some(true));
+    let publication = publish_candidate(&store, &scan, candidate, None).await;
+    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
+    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
+    assert_eq!(outcome, ScannerCycleOutcome::Completed);
+    assert!(!pending);
+    assert!(acknowledgements.is_empty());
+    assert!(!crate::scanner_io::dirty_usage_buckets_pending());
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_stale_root_cas_keeps_dirty_after_bucket_save() {
+    let (_directory, store) = candidate_store().await;
+    let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    let mut bucket_cache = DataUsageCache::default();
+    bucket_cache
+        .load(store.pools[0].disk_set[0].clone(), &path_join_buf(&[PROOF_BUCKET, DATA_USAGE_CACHE_NAME]))
+        .await
+        .expect("real bucket checkpoint must be persisted before root publication");
+    assert!(bucket_cache.info.snapshot_complete);
+    assert_eq!(
+        bucket_cache
+            .checked_flatten(PROOF_BUCKET)
+            .expect("persisted bucket root")
+            .objects,
+        1
+    );
+    let stale_baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("missing root revision");
+    assert_eq!(stale_baseline.revision, DataUsageCacheRevision::Missing);
+    let mut competing = candidate.clone();
+    competing.scanner_epoch = Some(PROOF_EPOCH + 1);
+    competing.scanner_cycle = Some(PROOF_CYCLE + 1);
+    for state in &mut competing.usage_snapshot_set_states {
+        state.scanner_epoch = Some(PROOF_EPOCH + 1);
+        state.scanner_cycle = Some(PROOF_CYCLE + 1);
+    }
+    let competing_bytes = serde_json::to_vec(&competing).expect("competing root");
+    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), competing_bytes.clone())
+        .await
+        .expect("another publisher wins the actual root slot");
+    let before = read_root(&store).await;
+    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
+
+    let publication = publish_candidate(&store, &scan, candidate, Some(stale_baseline)).await;
+    assert_eq!(
+        publication.outcome(),
+        DataUsagePersistOutcome::Current,
+        "the old missing revision loses CAS and reconciles the newer root"
+    );
+    let (_, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication);
+    assert!(acknowledgements.is_empty());
+    assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
+    assert_eq!(
+        read_root(&store).await,
+        before,
+        "bucket durability must not authorize replacing the winning root"
+    );
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_cannot_transfer_proof_between_real_scan_results() {
+    let (_directory, store) = candidate_store().await;
+    let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    let (second_scan, second_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    assert_eq!(first_candidate.scanner_epoch, second_candidate.scanner_epoch);
+    assert_eq!(first_candidate.scanner_cycle, second_candidate.scanner_cycle);
+    assert_eq!(first_candidate.objects_total_count, second_candidate.objects_total_count);
+    let baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("initial root revision");
+    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
+    let publication = publish_candidate(&store, &first_scan, first_candidate, Some(baseline)).await;
+    assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved);
+    assert!(read_root(&store).await.0.is_some(), "the first scan really published its root");
+
+    let (_, pending, acknowledgements) = finalize_scanner_cycle_result(second_scan, publication);
+    assert!(pending, "another scan's publication must not finish this scan's dirty maintenance work");
+    assert!(acknowledgements.is_empty());
+    assert_eq!(
+        crate::scanner_io::dirty_usage_buckets_for_tests(),
+        dirty,
+        "same counters and cycle cannot transfer another scan's proof"
+    );
+
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_stale_baseline_cannot_prove_a_replaced_root() {
+    let (_directory, store) = candidate_store().await;
+    let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+    save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&first_candidate).expect("first candidate"),
+    )
+    .await
+    .expect("persist the first candidate on real disks");
+    let stale_baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("capture the genuine first root revision");
+    let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
+    let mut reader = PutObjReader::from_vec(b"second".to_vec());
+    store.pools[0].disk_set[0]
+        .put_object(PROOF_BUCKET, "second", &mut reader, &ObjectOptions::default())
+        .await
+        .expect("commit a real namespace change");
+    assert_eq!(
+        crate::scanner_io::dirty_usage_buckets_for_tests(),
+        dirty,
+        "direct storage writes leave this fixture's scanner hint generation unchanged"
+    );
+    let (_, replacement) = complete_candidate(&store, PROOF_CYCLE).await;
+    assert_eq!(first_candidate.scanner_epoch, replacement.scanner_epoch);
+    assert_eq!(first_candidate.scanner_cycle, replacement.scanner_cycle);
+    assert_eq!((first_candidate.objects_total_count, replacement.objects_total_count), (1, 2));
+    save_config(
+        store.clone(),
+        DATA_USAGE_OBJ_NAME_PATH.as_str(),
+        serde_json::to_vec(&replacement).expect("replacement candidate"),
+    )
+    .await
+    .expect("publish the replacement root");
+    let current = read_root(&store).await;
+    assert_ne!(current.1, stale_baseline.revision);
+
+    // The supplied baseline still equals candidate A, but the actual target
+    // now contains B. Compatibility's AlreadyDurable outcome is not proof.
+    let publication = publish_candidate(&store, &first_scan, first_candidate, Some(stale_baseline)).await;
+    assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable);
+    let (_, pending, acknowledgements) = finalize_scanner_cycle_result(first_scan, publication);
+    assert!(pending);
+    assert!(acknowledgements.is_empty());
+    assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty);
+    assert_eq!(read_root(&store).await, current);
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish() {
+    for mutation in ["remote_ack_target", "publication_epoch", "remote_lease_targets"] {
+        let (_directory, store) = candidate_store().await;
+        let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
+        let baseline = read_data_usage_persist_baseline(store.clone())
+            .await
+            .expect("initial root revision");
+        let dirty = crate::scanner_io::dirty_usage_buckets_for_tests();
+        let changed_generation = dirty
+            .get(PROOF_BUCKET)
+            .expect("the real scan has dirty work")
+            .checked_add(1)
+            .expect("bounded fixture generation");
+        let changed_epoch = scan
+            .publication_epoch()
+            .expect("real scan publication epoch")
+            .checked_add(1)
+            .expect("bounded fixture epoch");
+        let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await;
+        assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved, "{mutation}");
+        let root_before = read_root(&store).await;
+        assert_eq!(
+            serde_json::from_slice::<DataUsageInfo>(root_before.0.as_deref().expect("actual saved root"))
+                .expect("persisted root JSON"),
+            candidate,
+            "{mutation}: the original candidate really reached root storage"
+        );
+
+        let changed = match mutation {
+            "remote_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement {
+                host: "proof-peer:9000".to_string(),
+                instance_id: crate::scanner_activity_epoch().to_string(),
+                generation: changed_generation,
+            }]),
+            "publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
+            "remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(
+                "proof-peer:9000".to_string(),
+                crate::scanner_activity_epoch().to_string(),
+                changed_generation,
+            )]),
+            _ => unreachable!("fixed mutation cases"),
+        };
+        let (_, pending, acknowledgements) = finalize_scanner_cycle_result(changed, publication);
+        assert!(
+            acknowledgements.is_empty(),
+            "{mutation}: the old root proof must not authorize changed ACK work"
+        );
+        assert!(pending, "{mutation}: changed maintenance work must remain pending");
+        assert_eq!(
+            crate::scanner_io::dirty_usage_buckets_for_tests(),
+            dirty,
+            "{mutation}: the changed scan must not clear local dirty work"
+        );
+        assert_eq!(read_root(&store).await, root_before, "{mutation}: the durable original root is retained");
+    }
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs
index 08a3c9996..5c4727732 100644
--- a/crates/scanner/src/scanner/usage_store.rs
+++ b/crates/scanner/src/scanner/usage_store.rs
@@ -34,6 +34,139 @@ pub(super) enum DataUsagePersistOutcome {
     Failed,
 }
 
+#[derive(Debug)]
+pub(crate) struct RootPublicationProof {
+    candidate: crate::scanner_io::ScannerPublicationExpectation,
+    root_version: (String, [u8; 32]),
+}
+
+impl RootPublicationProof {
+    pub(crate) fn verified_version_for(
+        &self,
+        expected: &crate::scanner_io::ScannerPublicationExpectation,
+    ) -> Option<&(String, [u8; 32])> {
+        self.candidate.same_candidate(expected).then_some(&self.root_version)
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct DataUsagePublicationResult {
+    outcome: DataUsagePersistOutcome,
+    proof: Option<RootPublicationProof>,
+}
+
+impl From<DataUsagePersistOutcome> for DataUsagePublicationResult {
+    fn from(outcome: DataUsagePersistOutcome) -> Self {
+        Self { outcome, proof: None }
+    }
+}
+
+impl DataUsagePublicationResult {
+    pub(super) fn outcome(&self) -> DataUsagePersistOutcome {
+        self.outcome
+    }
+    pub(super) fn restrict_outcome(&mut self, outcome: DataUsagePersistOutcome) {
+        if outcome != self.outcome {
+            self.proof = None;
+        }
+        self.outcome = outcome;
+    }
+    pub(super) fn into_parts(self) -> (DataUsagePersistOutcome, Option<RootPublicationProof>) {
+        (self.outcome, self.proof)
+    }
+}
+
+fn root_ack_write_is_confirmed<T, E>(
+    result: &std::result::Result<T, E>,
+    state: Option<ScannerPublicationCommitState>,
+    written_etag: Option<&str>,
+) -> bool {
+    result.is_ok() && state == Some(ScannerPublicationCommitState::Committed) && written_etag.is_some_and(|etag| !etag.is_empty())
+}
+
+#[cfg(test)]
+mod root_publication_confirmation_tests {
+    use super::*;
+
+    #[test]
+    fn root_publication_confirmation_requires_committed_state_and_write_revision() {
+        let saved = Ok::<(), ()>(());
+        for state in [
+            None,
+            Some(ScannerPublicationCommitState::Admitted),
+            Some(ScannerPublicationCommitState::InFlight),
+            Some(ScannerPublicationCommitState::AbortedBeforeCommit),
+            Some(ScannerPublicationCommitState::Indeterminate),
+        ] {
+            assert!(!root_ack_write_is_confirmed(&saved, state, Some("revision")));
+        }
+        for etag in [None, Some("")] {
+            assert!(!root_ack_write_is_confirmed(&saved, Some(ScannerPublicationCommitState::Committed), etag));
+        }
+        assert!(root_ack_write_is_confirmed(
+            &saved,
+            Some(ScannerPublicationCommitState::Committed),
+            Some("revision")
+        ));
+    }
+
+    #[test]
+    fn root_publication_confirmation_does_not_carry_state_across_cas_attempts() {
+        let attempts = [
+            (Err(()), Some(ScannerPublicationCommitState::Committed), Some("first")),
+            (Ok(()), Some(ScannerPublicationCommitState::AbortedBeforeCommit), Some("second")),
+            (Ok(()), None, Some("legacy")),
+            (Ok(()), Some(ScannerPublicationCommitState::Committed), Some("confirmed")),
+        ];
+        let confirmations = attempts
+            .iter()
+            .map(|(result, state, etag)| root_ack_write_is_confirmed(result, *state, *etag))
+            .collect::<Vec<_>>();
+        assert_eq!(confirmations, [false, false, false, true]);
+    }
+}
+
+async fn read_root_publication_proof<S: ScannerObjectIO + ScannerConfigObjectDelete>(
+    store: Arc<S>,
+    ctx: &CancellationToken,
+    deadline: tokio::time::Instant,
+    epoch: u64,
+    expected: &crate::scanner_io::ScannerPublicationExpectation,
+    candidate: &DataUsageInfo,
+    written_etag: Option<&str>,
+) -> Option<RootPublicationProof> {
+    let read = async {
+        let _admission = scanner_publication_admission_for_epoch(store.clone(), epoch).await?;
+        let (bytes, revision) = read_config_with_revision(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .ok()?;
+        let bytes = bytes?;
+        let DataUsageCacheRevision::Etag(etag) = revision else {
+            return None;
+        };
+        if etag.is_empty() || written_etag.is_some_and(|written| written != etag) {
+            return None;
+        }
+        let persisted: DataUsageInfo = serde_json::from_slice(&bytes).ok()?;
+        if &persisted != candidate {
+            return None;
+        }
+        let root_digest = Sha256::digest(&bytes).into();
+        if ctx.is_cancelled() || tokio::time::Instant::now() >= deadline {
+            return None;
+        }
+        Some(RootPublicationProof {
+            candidate: expected.clone(),
+            root_version: (etag, root_digest),
+        })
+    };
+    tokio::select! {
+        biased;
+        _ = ctx.cancelled() => None,
+        result = tokio::time::timeout_at(deadline, read) => result.ok().flatten(),
+    }
+}
+
 fn remote_lease_expired(deadline: Option<std::time::Instant>) -> bool {
     deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
 }
@@ -166,6 +299,7 @@ pub(super) struct ScannerPublicationFence {
     pub(super) scanner_publication_lease_fence: Option<String>,
     pub(super) remote_lease_tokens: Vec<Uuid>,
     pub(super) lease_release_safe: Arc<AtomicBool>,
+    pub(super) ack_expectation: Option<crate::scanner_io::ScannerPublicationExpectation>,
 }
 
 impl ScannerPublicationFence {
@@ -180,6 +314,7 @@ impl ScannerPublicationFence {
             scanner_publication_lease_fence,
             remote_lease_tokens: Vec::new(),
             lease_release_safe: Arc::new(AtomicBool::new(true)),
+            ack_expectation: None,
         }
     }
 
@@ -192,21 +327,26 @@ impl ScannerPublicationFence {
         self.lease_release_safe = lease_release_safe;
         self
     }
+
+    pub(super) fn with_ack_expectation(mut self, expected: Option<crate::scanner_io::ScannerPublicationExpectation>) -> Self {
+        self.ack_expectation = expected;
+        self
+    }
 }
 
 #[derive(Debug)]
-pub(super) enum DataUsagePersistTaskResult {
-    Completed(DataUsagePersistOutcome),
+pub(super) enum DataUsagePersistTaskResult<T = DataUsagePersistOutcome> {
+    Completed(T),
     Cancelled,
     TimedOut,
     JoinFailed(tokio::task::JoinError),
 }
 
-pub(super) async fn wait_for_data_usage_persist_task(
+pub(super) async fn wait_for_data_usage_persist_task<T>(
     ctx: &CancellationToken,
-    task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
+    task: &mut AbortOnDropHandle<T>,
     timeout: Duration,
-) -> DataUsagePersistTaskResult {
+) -> DataUsagePersistTaskResult<T> {
     tokio::select! {
         biased;
         result = &mut *task => match result {
@@ -320,6 +460,7 @@ where
         route_probe,
     )
     .await
+    .outcome()
 }
 
 pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence<
@@ -333,7 +474,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
     initial_baseline: Option<DataUsagePersistBaseline>,
     publication_fence: ScannerPublicationFence,
     route_probe: F,
-) -> DataUsagePersistOutcome
+) -> DataUsagePublicationResult
 where
     F: Fn() -> Fut + Send + Sync,
     Fut: Future<Output = Option<ScannerCycleDeferReason>> + Send,
@@ -344,11 +485,15 @@ where
         scanner_publication_lease_fence,
         remote_lease_tokens,
         lease_release_safe,
+        ack_expectation,
     } = publication_fence;
+    let ack_deadline = scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline);
     let mut outcome = DataUsagePersistOutcome::NoUpdate;
+    let mut proof = None;
     let mut next_baseline = initial_baseline;
 
     'updates: while let Some(mut data_usage_info) = receiver.recv().await {
+        proof = None;
         let _activity_guard = ScannerActivityGuard::new();
         if ctx.is_cancelled() {
             break;
@@ -523,10 +668,14 @@ where
                 continue;
             }
         };
-        let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower));
+        let data_digest: [u8; 32] = Sha256::digest(&data).into();
+        let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(data_digest, hex_simd::AsciiCase::Lower));
         let data = Bytes::from(data);
         let backup_due = !observational && data_usage_backup_due(&data_usage_info);
         let mut cas_retry = 0usize;
+        let mut ack_epoch = None;
+        let mut write_confirmed = false;
+        let mut written_etag = None;
         let save_outcome = loop {
             if ctx.is_cancelled() {
                 break 'updates;
@@ -557,6 +706,7 @@ where
             } else {
                 None
             };
+            ack_epoch = Some(publication_epoch_for_save);
             let (existing_data, revision) = match baseline {
                 Some(baseline) => (baseline.data, baseline.revision),
                 None => match read_config_with_revision(storeapi.clone(), target_path).await {
@@ -645,7 +795,7 @@ where
             }
 
             let done_save = Metrics::time(Metric::SaveUsage);
-            let save_result = {
+            let (save_result, commit_state) = {
                 let publication_scope = storeapi
                     .scanner_data_usage_publication_commit_scope_with_release_flag(
                         publication_epoch_for_save,
@@ -681,24 +831,33 @@ where
                 .await;
                 drop(legacy_publication_admission);
                 if let Some(scope) = publication_scope {
-                    match scope.wait_for_completion().await {
-                        ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => {
-                            save_result
-                        }
+                    let state = scope.wait_for_completion().await;
+                    let result = match state {
+                        ScannerPublicationCommitState::Committed => save_result,
+                        ScannerPublicationCommitState::AbortedBeforeCommit => save_result,
                         ScannerPublicationCommitState::Indeterminate
                         | ScannerPublicationCommitState::Admitted
                         | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other(
                             "scanner publication commit scope did not reach a safe terminal state",
                         )),
-                    }
+                    };
+                    (result, Some(state))
                 } else {
-                    save_result
+                    (save_result, None)
                 }
             };
             done_save();
 
+            let attempt_confirmed = root_ack_write_is_confirmed(
+                &save_result,
+                commit_state,
+                save_result.as_ref().ok().and_then(|info| info.etag.as_deref()),
+            );
+
             match save_result {
                 Ok(object_info) => {
+                    write_confirmed = attempt_confirmed;
+                    written_etag = object_info.etag.as_ref().filter(|etag| !etag.is_empty()).cloned();
                     if !observational {
                         next_baseline = object_info
                             .etag
@@ -909,9 +1068,27 @@ where
                 break 'updates;
             }
         }
+        if !observational
+            && data_usage_info.usage_snapshot_converged == Some(true)
+            && matches!(outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable)
+            && (outcome == DataUsagePersistOutcome::AlreadyDurable || write_confirmed)
+            && let (Some(expected), Some(epoch)) = (ack_expectation.as_ref(), ack_epoch)
+            && expected.matches_encoded_candidate(&data_digest)
+        {
+            proof = read_root_publication_proof(
+                storeapi.clone(),
+                &ctx,
+                ack_deadline,
+                epoch,
+                expected,
+                &data_usage_info,
+                written_etag.as_deref(),
+            )
+            .await;
+        }
     }
 
-    outcome
+    DataUsagePublicationResult { outcome, proof }
 }
 
 async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease(
diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs
index e0adc45a6..b51e5d2f8 100644
--- a/crates/scanner/src/scanner_io.rs
+++ b/crates/scanner/src/scanner_io.rs
@@ -114,6 +114,11 @@ pub(crate) struct ScannerBucketScanScope {
 }
 
 impl ScannerBucketScanScope {
+    #[cfg(test)]
+    pub(crate) fn selected_buckets_for_tests(&self) -> Option<&HashSet<String>> {
+        self.selected_buckets.as_deref()
+    }
+
     fn is_default(&self) -> bool {
         self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
     }
@@ -218,8 +223,8 @@ fn complete_scanner_cache_snapshot_plan_digest(
 
 fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
     let authoritative = serde_json::from_slice::<DataUsageInfo>(proof.authoritative_data?).ok()?;
-    if let Some(plan_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
-        return Some(plan_digest);
+    if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) {
+        return Some(validated_digest);
     }
 
     // A complete but superseded observation may reuse its per-set cache only
@@ -896,6 +901,7 @@ pub(crate) struct ScannerCycleResult {
     failed_dirty_usage: bool,
     pending_maintenance_work: bool,
     required_cycle_floor: Option<u64>,
+    publication_expectation: Option<ScannerPublicationExpectation>,
 }
 
 impl ScannerCycleResult {
@@ -911,10 +917,12 @@ impl ScannerCycleResult {
             failed_dirty_usage: false,
             pending_maintenance_work: false,
             required_cycle_floor: None,
+            publication_expectation: None,
         }
     }
 
     pub(crate) fn with_publication_epoch(mut self, publication_epoch: Option<u64>) -> Self {
+        self.publication_expectation = None;
         self.publication_epoch = publication_epoch;
         self
     }
@@ -924,6 +932,7 @@ impl ScannerCycleResult {
     }
 
     fn with_activity_digest(mut self, activity_digest: [u8; 32]) -> Self {
+        self.publication_expectation = None;
         self.activity_digest = Some(activity_digest);
         self
     }
@@ -933,6 +942,7 @@ impl ScannerCycleResult {
     }
 
     pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self {
+        self.publication_expectation = None;
         self.observational_snapshot_published = published;
         self
     }
@@ -942,16 +952,19 @@ impl ScannerCycleResult {
     }
 
     fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self {
+        self.publication_expectation = None;
         self.failed_dirty_usage = failed_dirty_usage;
         self
     }
 
     fn with_pending_maintenance_work(mut self, pending_maintenance_work: bool) -> Self {
+        self.publication_expectation = None;
         self.pending_maintenance_work = pending_maintenance_work;
         self
     }
 
     fn with_required_cycle_floor(mut self, required_cycle_floor: Option<u64>) -> Self {
+        self.publication_expectation = None;
         self.required_cycle_floor = required_cycle_floor;
         self
     }
@@ -960,11 +973,13 @@ impl ScannerCycleResult {
         mut self,
         acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
     ) -> Self {
+        self.publication_expectation = None;
         self.remote_dirty_usage_acknowledgements = acknowledgements;
         self
     }
 
     pub(crate) fn with_remote_publication_lease_targets(mut self, targets: Vec<(String, String, u64)>) -> Self {
+        self.publication_expectation = None;
         self.remote_publication_lease_targets = targets;
         self
     }
@@ -973,7 +988,32 @@ impl ScannerCycleResult {
         &self.remote_publication_lease_targets
     }
 
-    pub(crate) fn acknowledge_durable_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
+    pub(crate) fn publication_expectation(&self) -> Option<ScannerPublicationExpectation> {
+        self.publication_expectation.clone()
+    }
+
+    fn with_publication_expectation(mut self, expectation: Option<ScannerPublicationExpectation>) -> Self {
+        // Seal only after all coverage and acknowledgement inputs are final.
+        self.publication_expectation = expectation;
+        self
+    }
+
+    pub(crate) fn acknowledge_durable_usage(
+        self,
+        proof: &crate::scanner::RootPublicationProof,
+    ) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
+        if self.status != ScannerCycleStatus::Complete
+            || self
+                .publication_expectation
+                .as_ref()
+                .is_none_or(|expected| proof.verified_version_for(expected).is_none())
+        {
+            return Vec::new();
+        }
+        self.clear_verified_usage()
+    }
+
+    fn clear_verified_usage(self) -> Vec<crate::scanner::ScannerDirtyUsageAcknowledgement> {
         if let Some(snapshot) = self.dirty_usage_clear {
             clear_dirty_usage_buckets(&snapshot);
         }
@@ -1013,6 +1053,7 @@ mod publish_gate_tests;
 #[cfg(test)]
 mod tests;
 
+pub(crate) use cache::ScannerPublicationExpectation;
 use cache::*;
 use dirty_usage::*;
 use guards::*;
diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs
index 5e47128b8..328b91ce9 100644
--- a/crates/scanner/src/scanner_io/cache.rs
+++ b/crates/scanner/src/scanner_io/cache.rs
@@ -308,6 +308,86 @@ impl<'a> ValidatedScannerSnapshot<'a> {
     }
 }
 
+#[derive(Clone, Debug)]
+pub(crate) struct ScannerPublicationExpectation {
+    candidate: Arc<([u8; 32], DataUsageScanPlanDigest)>,
+}
+
+impl ScannerPublicationExpectation {
+    pub(crate) fn matches_encoded_candidate(&self, digest: &[u8; 32]) -> bool {
+        &self.candidate.0 == digest
+    }
+
+    pub(crate) fn same_candidate(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.candidate, &other.candidate) && self.candidate.1 == other.candidate.1
+    }
+}
+
+pub(super) struct ValidatedUsageCandidate {
+    data: DataUsageInfo,
+    #[cfg(test)]
+    last_update: SystemTime,
+    coverage_digest: DataUsageScanPlanDigest,
+}
+
+pub(super) fn empty_namespace_usage_candidate(
+    all_buckets: &[BucketInfo],
+    sources: &HashSet<DataUsageCacheSource>,
+    buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
+    identity: ScannerSnapshotIdentity,
+) -> Option<ValidatedUsageCandidate> {
+    if !all_buckets.is_empty()
+        || sources.is_empty()
+        || sources.len() != buckets_by_source.len()
+        || sources
+            .iter()
+            .any(|source| buckets_by_source.get(source).is_none_or(|buckets| !buckets.is_empty()))
+    {
+        return None;
+    }
+    let last_update = SystemTime::now();
+    Some(ValidatedUsageCandidate {
+        data: DataUsageInfo {
+            last_update: Some(last_update),
+            scanner_cycle: Some(identity.cycle),
+            scanner_epoch: Some(identity.leader_epoch),
+            usage_snapshot_complete: true,
+            ..Default::default()
+        },
+        #[cfg(test)]
+        last_update,
+        coverage_digest: identity.coverage_digest,
+    })
+}
+
+impl ValidatedUsageCandidate {
+    pub(super) fn prepare(mut self, status: ScannerCycleStatus) -> (DataUsageInfo, Option<ScannerPublicationExpectation>) {
+        self.data.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete);
+        let expectation = if status == ScannerCycleStatus::Complete {
+            struct DigestWriter(Sha256);
+            impl std::io::Write for DigestWriter {
+                fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                    self.0.update(bytes);
+                    Ok(bytes.len())
+                }
+                fn flush(&mut self) -> std::io::Result<()> {
+                    Ok(())
+                }
+            }
+            let mut writer = DigestWriter(Sha256::new());
+            serde_json::to_writer(&mut writer, &self.data)
+                .ok()
+                .map(|()| ScannerPublicationExpectation {
+                    candidate: Arc::new((writer.0.finalize().into(), self.coverage_digest)),
+                })
+        } else {
+            None
+        };
+        (self.data, expectation)
+    }
+}
+
+#[cfg(test)]
 pub(super) fn completed_data_usage_info(
     results: &[DataUsageCache],
     scope: &ScannerSnapshotScope<'_>,
@@ -316,6 +396,18 @@ pub(super) fn completed_data_usage_info(
     budget_elapsed: bool,
     cancelled: bool,
 ) -> Option<(DataUsageInfo, SystemTime)> {
+    completed_usage_candidate(results, scope, tier_registry_names, bucket_plan_complete, budget_elapsed, cancelled)
+        .map(|candidate| (candidate.data, candidate.last_update))
+}
+
+pub(super) fn completed_usage_candidate(
+    results: &[DataUsageCache],
+    scope: &ScannerSnapshotScope<'_>,
+    tier_registry_names: &[String],
+    bucket_plan_complete: bool,
+    budget_elapsed: bool,
+    cancelled: bool,
+) -> Option<ValidatedUsageCandidate> {
     if !bucket_plan_complete {
         return None;
     }
@@ -393,7 +485,12 @@ pub(super) fn completed_data_usage_info(
         usage_snapshot_set_states,
         ..Default::default()
     };
-    Some((data_usage_info, merged_last_update))
+    Some(ValidatedUsageCandidate {
+        data: data_usage_info,
+        #[cfg(test)]
+        last_update: merged_last_update,
+        coverage_digest: scope.identity.coverage_digest,
+    })
 }
 
 fn tier_accounting_proof_is_publishable(
diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs
index 4eefe2910..d4094a311 100644
--- a/crates/scanner/src/scanner_io/io_cycle.rs
+++ b/crates/scanner/src/scanner_io/io_cycle.rs
@@ -338,12 +338,21 @@ where
             dirty_usage_status,
             activity_status,
         );
-        let empty_usage = DataUsageInfo {
-            last_update: Some(SystemTime::now()),
-            scanner_cycle: Some(want_cycle),
-            usage_snapshot_complete: true,
-            ..Default::default()
+        let Some(candidate) = empty_namespace_usage_candidate(
+            &all_buckets,
+            &expected_sources,
+            &buckets_by_source,
+            ScannerSnapshotIdentity {
+                cycle: want_cycle,
+                leader_epoch,
+                plan_digest: scan_plan_digest,
+                coverage_digest: bucket_coverage_digest,
+                tier_registry_generation: Some(tier_registry_generation),
+            },
+        ) else {
+            return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
         };
+        let (empty_usage, publication_expectation) = candidate.prepare(status);
         let observational_snapshot_published = if should_publish_observational_snapshot(status) {
             publish_observational_snapshot(&updates, empty_usage).await?
         } else {
@@ -366,7 +375,8 @@ where
             .with_activity_digest(activity_digest)
             .with_observational_snapshot_published(observational_snapshot_published)
             .with_remote_publication_lease_targets(remote_publication_lease_targets)
-            .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements));
+            .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
+            .with_publication_expectation(publication_expectation));
     }
 
     let total_results = expected_sources.len();
@@ -595,7 +605,7 @@ where
     let (activity_status, remote_publication_lease_targets) =
         scanner_cycle_activity_status(store, distributed, &activity_before).await;
     let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
-    let completed_usage = completed_data_usage_info(
+    let completed_usage = completed_usage_candidate(
         &results,
         &ScannerSnapshotScope {
             sources: &expected_sources,
@@ -636,7 +646,10 @@ where
         dirty_usage_status,
         activity_status,
     );
-    let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage {
+    let mut publication_expectation = None;
+    let observational_snapshot_published = if let Some(candidate) = completed_usage {
+        let (data_usage_info, expectation) = candidate.prepare(cycle_status);
+        publication_expectation = expectation;
         if should_publish_observational_snapshot(cycle_status) {
             publish_observational_snapshot(&updates, data_usage_info).await?
         } else {
@@ -674,5 +687,6 @@ where
         .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
         .with_failed_dirty_usage(!failed_buckets.is_empty())
         .with_pending_maintenance_work(pending_maintenance_work)
-        .with_required_cycle_floor(required_cycle_floor))
+        .with_required_cycle_floor(required_cycle_floor)
+        .with_publication_expectation(publication_expectation))
 }
diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs
index e33a893d0..1d58dc88e 100644
--- a/crates/scanner/src/scanner_io/tests.rs
+++ b/crates/scanner/src/scanner_io/tests.rs
@@ -462,6 +462,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                 .await
                 .expect("fixture rename tail should finish before the usage scan");
         }
+        wait_for_namespace_commit_tails(&store).await;
         let ctx = CancellationToken::new();
         let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
         let (updates, receiver) = mpsc::channel(1);
@@ -511,15 +512,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
             .put_object("cold-bucket", "new", &mut reader, &ScannerObjectOptions::default())
             .await
             .expect("new cold object should persist");
-        let lock = store.pools[0].disk_set[0]
-            .new_ns_lock("cold-bucket", "new")
-            .await
-            .expect("fixture namespace lock should be created");
-        let _settled = lock
-            .get_write_lock(Duration::from_secs(30))
-            .await
-            .expect("fixture rename tail should finish before the usage scan");
-        drop(_settled);
+        wait_for_namespace_commit_tails(&store).await;
         record_dirty_usage_bucket("hot-bucket");
         if scan_mode == HealScanMode::Normal && !requires_full_scan {
             record_dirty_usage_bucket("cold-bucket");
@@ -1011,8 +1004,8 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
     assert!(dirty_usage_buckets().contains_key("temporarily-omitted"));
     assert_eq!(dirty_usage_snapshot_status(&snapshot), DirtyUsageSnapshotStatus::Current);
 
-    let acknowledgements = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()))
-        .acknowledge_durable_usage();
+    let acknowledgements =
+        ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())).clear_verified_usage();
     assert!(acknowledgements.is_empty());
     assert!(!dirty_usage_buckets().contains_key("temporarily-omitted"));
     clear_dirty_usage_buckets_for_tests();
@@ -1118,7 +1111,7 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
     assert!(dirty_usage_buckets().contains_key("photos"));
 
     let confirmed = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone()));
-    let acknowledgements = confirmed.acknowledge_durable_usage();
+    let acknowledgements = confirmed.clear_verified_usage();
     assert!(acknowledgements.is_empty());
     assert!(!dirty_usage_buckets().contains_key("photos"));
     clear_dirty_usage_buckets_for_tests();
diff --git a/docs/architecture/s3-compatibility-matrix.md b/docs/architecture/s3-compatibility-matrix.md
index 9393e129b..0267db33a 100644
--- a/docs/architecture/s3-compatibility-matrix.md
+++ b/docs/architecture/s3-compatibility-matrix.md
@@ -55,6 +55,14 @@ Standard S3 areas that must not be described as complete:
 
 `excluded_tests.txt` holds tests that must not block the compatibility gate: vendor-specific or non-portable behavior, and intentionally unsupported product behavior such as ACL authorization.
 
+## Intentional Deviations From AWS S3
+
+Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/{object}/xl.meta`), the same layout MinIO uses. The rules below exist to keep that layout unambiguous and are not compatibility gaps to close; clients that need the AWS behavior must adapt on their side.
+
+| Behavior | RustFS | AWS S3 | Why |
+|---|---|---|---|
+| Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
+
 ## Update Rule
 
 When a feature starts passing, move its test entries from `unimplemented_tests.txt` to `implemented_tests.txt` and update the row here in the same PR. Do not change README wording beyond the supported coverage. Handler-level status (missing, stubbed, or diverging endpoints) is tracked in [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md).
diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs
index d511470d0..a478a5778 100644
--- a/rustfs/src/admin/handlers/ilm_transition.rs
+++ b/rustfs/src/admin/handlers/ilm_transition.rs
@@ -470,6 +470,14 @@ async fn authorize_manual_transition_request(req: &S3Request<Body>) -> S3Result<
 /// The credential pre-check keeps this endpoint family's historical
 /// missing-credentials message (the shared gate reports "get cred failed") and
 /// still yields the masked actor every transition audit log records.
+async fn authorize_recovery_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
+    if req.credentials.is_none() {
+        return Err(admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"));
+    }
+    let credentials = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    Ok(recovery_actor_sha256(&credentials))
+}
+
 async fn authorize_transition_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
     let Some(input_cred) = req.credentials.as_ref() else {
         return Err(s3_error!(InvalidRequest, "authentication required"));
@@ -481,14 +489,6 @@ async fn authorize_transition_admin_request(req: &S3Request<Body>, action: Admin
     Ok(actor)
 }
 
-async fn authorize_recovery_admin_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<String> {
-    if req.credentials.is_none() {
-        return Err(admin_s3_error(AdminS3ErrorCode::InvalidRequest, "authentication required"));
-    }
-    let credentials = authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
-    Ok(recovery_actor_sha256(&credentials))
-}
-
 fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uuid> {
     Uuid::parse_str(params.get("transaction_id").unwrap_or(""))
         .map_err(|_| s3_error!(InvalidArgument, "invalid transition transaction id"))
@@ -1498,23 +1498,6 @@ impl Operation for ManualTransitionJobCancelHandler {
     }
 }
 
-pub struct TransitionReconcileInspectHandler {}
-
-#[async_trait::async_trait]
-impl Operation for TransitionReconcileInspectHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
-        let transaction_id = transition_transaction_id_from_params(&params)?;
-        let Some(store) = object_store_from_extensions(&req.extensions) else {
-            return Err(s3_error!(InternalError, "object store is not initialized"));
-        };
-        let status = inspect_transition_transaction_for_operator(store, transaction_id)
-            .await
-            .map_err(map_transition_operator_error)?;
-        json_response(StatusCode::OK, &status)
-    }
-}
-
 pub struct IlmRecoveryControlListHandler {}
 
 #[async_trait::async_trait]
@@ -1634,6 +1617,23 @@ impl Operation for IlmRecoveryExportDownloadHandler {
     }
 }
 
+pub struct TransitionReconcileInspectHandler {}
+
+#[async_trait::async_trait]
+impl Operation for TransitionReconcileInspectHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
+        let transaction_id = transition_transaction_id_from_params(&params)?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(s3_error!(InternalError, "object store is not initialized"));
+        };
+        let status = inspect_transition_transaction_for_operator(store, transaction_id)
+            .await
+            .map_err(map_transition_operator_error)?;
+        json_response(StatusCode::OK, &status)
+    }
+}
+
 pub struct TransitionReconcileApplyHandler {}
 
 #[async_trait::async_trait]
@@ -2153,7 +2153,7 @@ mod tests {
         let inspect = src
             .split("impl Operation for TransitionReconcileInspectHandler")
             .nth(1)
-            .and_then(|block| block.split("impl Operation for TransitionReconcileApplyHandler").next())
+            .and_then(|block| block.split("pub struct IlmRecoveryControlListHandler").next())
             .expect("inspect handler block");
         assert!(inspect.contains("AdminAction::ListTierAction"));
         assert!(!inspect.contains("AdminAction::SetTierAction"));
@@ -2502,7 +2502,7 @@ mod tests {
         let wrapper = extract_block_between_markers(
             production,
             "async fn authorize_transition_admin_request",
-            "fn transition_transaction_id_from_params",
+            "async fn authorize_recovery_admin_request",
         );
 
         assert_eq!(
diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs
index 1cb5a8be9..d77531e1f 100644
--- a/rustfs/src/app/bucket_usecase.rs
+++ b/rustfs/src/app/bucket_usecase.rs
@@ -4070,14 +4070,13 @@ mod tests {
                     abort_incomplete_multipart_upload: None,
                     del_marker_expiration: None,
                     filter: Some(s3s::dto::LifecycleRuleFilter {
-                        prefix: Some("logs/".to_string()),
-                        tag: Some(s3s::dto::Tag {
-                            key: Some("env".to_string()),
-                            value: Some("prod".to_string()),
+                        and: Some(s3s::dto::LifecycleRuleAndOperator {
+                            prefix: Some("logs/".to_string()),
+                            ..Default::default()
                         }),
                         ..Default::default()
                     }),
-                    id: Some("two-predicates".to_string()),
+                    id: Some("one-member-and".to_string()),
                     noncurrent_version_expiration: None,
                     noncurrent_version_transitions: None,
                     prefix: None,
@@ -4087,7 +4086,7 @@ mod tests {
             &ObjectLockConfiguration::default(),
         )
         .await
-        .expect_err("a Filter with two predicates is a schema violation");
+        .expect_err("a Filter And with one predicate is a schema violation");
         assert_eq!(*lifecycle_validation_error(&malformed).code(), S3ErrorCode::MalformedXML);
 
         let invalid_value = validate_lifecycle_config(
diff --git a/scripts/compat/heal-outcome/README.md b/scripts/compat/heal-outcome/README.md
deleted file mode 100644
index 58b215910..000000000
--- a/scripts/compat/heal-outcome/README.md
+++ /dev/null
@@ -1,33 +0,0 @@
-# Legacy Heal Outcome Compatibility
-
-This fixture executes the real `madmin-go` HTTP decoder and a pinned `mc` binary against synthetic v3 responses. Rust owner, admin adapter, and SDK tests validate the same JSON cases in `crates/madmin/tests/fixtures/heal-outcome-v3.json`. It does not run a storage repair or prove distributed recovery.
-
-Pinned primary sources:
-
-- `mc` release `RELEASE.2025-08-13T08-35-41Z`, commit `7394ce0dd2a80935aded936b09fa12cbb3cb8096`: [polling implementation](https://github.com/minio/mc/blob/7394ce0dd2a80935aded936b09fa12cbb3cb8096/cmd/admin-heal-ui.go#L414).
-- Its `madmin-go/v3` dependency is `v3.0.107-0.20250415152934-4b504b82db63`: [decoder and response type](https://github.com/minio/madmin-go/blob/4b504b82db633e978a57d49443b2be75824244c3/heal-commands.go#L101).
-
-The old decoder ignores additional JSON fields. The old poller returns success for `finished` without examining `detail`; only `stopped` returns a terminal error. Consequently `completed_with_errors` retains its canonical outcome and complete traversal coverage, but uses legacy summary `stopped`. `completed` describes execution only: unknown storage receipts remain `unknown`, not `repaired`.
-
-Run from the repository root with an isolated tool cache and binary directory:
-
-```sh
-(
-set -eu
-compat_dir=$(mktemp -d)
-trap 'rm -rf "$compat_dir"' EXIT
-export GOPATH="$compat_dir/gopath" GOMODCACHE="$compat_dir/mod" GOCACHE="$compat_dir/cache" GOBIN="$compat_dir/bin"
-export CGO_ENABLED=0 GOTOOLCHAIN=local GOMAXPROCS=2
-go install github.com/minio/mc@v0.0.0-20250813083541-7394ce0dd2a8
-cd scripts/compat/heal-outcome
-MC_BINARY="$compat_dir/bin/mc" NO_PROXY=127.0.0.1,localhost go test -mod=readonly -p 2 -count=1 -v ./...
-)
-```
-
-The subshell keeps the calling shell unchanged. `MC_BINARY` is mandatory and its Go build metadata must identify the pinned commit. Each subprocess gets a temporary mc configuration directory and synthetic credentials; it never edits the user's mc configuration. Loopback socket permission is required. The Go tests do not skip unavailable prerequisites.
-
-Six cases cover completed traversal, unknown repair proof, completed traversal with failures, cancellation, deadline, and untraversable listing. Two receiver cases carry a remote `finished` summary that contradicts an aborted or completed-with-errors outcome. The admin test applies the heal owner's wire validator to each `remoteResponse` and must produce the corresponding public `response`; the old CLI must then exit with an error. Unknown extension fields remain intact. Unknown or missing execution fields cannot validate a successful summary.
-
-New counters do not replace or reinterpret legacy progress. Outcome is a cumulative snapshot, not a page delta; `sinceSeq` only pages legacy result items. Result cursors and truncation markers remain separate from execution and traversal coverage.
-
-Two existing CLI limitations remain explicit: this mc does not terminate on `notFound`, and `-f` polling sends `forceStart` together with `clientToken`, a combination the RustFS v3 request contract rejects. The fixture uses the standard non-force polling flow. Neither limitation is hidden by emitting a new summary string or reporting a missing task as completed.
diff --git a/scripts/compat/heal-outcome/compat_test.go b/scripts/compat/heal-outcome/compat_test.go
deleted file mode 100644
index 8178ee27f..000000000
--- a/scripts/compat/heal-outcome/compat_test.go
+++ /dev/null
@@ -1,137 +0,0 @@
-// Copyright 2026 RustFS Team
-// Licensed under the Apache License, Version 2.0.
-
-package compat_test
-
-import (
-	"context"
-	"debug/buildinfo"
-	"encoding/json"
-	"fmt"
-	"net/http"
-	"net/http/httptest"
-	"os"
-	"os/exec"
-	"path/filepath"
-	"strings"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	madmin "github.com/minio/madmin-go/v3"
-)
-
-type fixture struct {
-	Name     string          `json:"name"`
-	CLIExit  int             `json:"cliExit"`
-	Response json.RawMessage `json:"response"`
-}
-
-func fixtures(t *testing.T) []fixture {
-	t.Helper()
-	data, err := os.ReadFile("../../../crates/madmin/tests/fixtures/heal-outcome-v3.json")
-	if err != nil {
-		t.Fatal(err)
-	}
-	var cases []fixture
-	if err := json.Unmarshal(data, &cases); err != nil {
-		t.Fatal(err)
-	}
-	if len(cases) != 8 {
-		t.Fatalf("expected eight owner/receiver-validated fixtures, got %d", len(cases))
-	}
-	return cases
-}
-
-func fixtureServer(t *testing.T, response []byte, polls *atomic.Int32) *httptest.Server {
-	t.Helper()
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodPost || !strings.HasPrefix(r.URL.Path, "/minio/admin/v3/heal/") {
-			t.Errorf("unexpected client request %s %s", r.Method, r.URL.Path)
-			http.Error(w, "unexpected request", http.StatusBadRequest)
-			return
-		}
-		w.Header().Set("Content-Type", "application/json")
-		if r.URL.Query().Get("clientToken") == "" {
-			fmt.Fprint(w, `{"clientToken":"fixture-token","clientAddress":"","startTime":"2026-01-01T00:00:00Z"}`)
-			return
-		}
-		polls.Add(1)
-		w.Write(response)
-	}))
-	t.Cleanup(server.Close)
-	return server
-}
-
-func TestLegacyMadminDecoder(t *testing.T) {
-	for _, f := range fixtures(t) {
-		t.Run(f.Name, func(t *testing.T) {
-			var polls atomic.Int32
-			server := fixtureServer(t, f.Response, &polls)
-			client, err := madmin.New(strings.TrimPrefix(server.URL, "http://"), "fixture-access", "fixture-secret", false)
-			if err != nil {
-				t.Fatal(err)
-			}
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			defer cancel()
-			_, status, err := client.Heal(ctx, "bucket", "", madmin.HealOpts{}, "fixture-token", false, false)
-			if err != nil {
-				t.Fatal(err)
-			}
-			var expected struct {
-				Summary string `json:"summary"`
-				Detail  string `json:"detail"`
-			}
-			if err := json.Unmarshal(f.Response, &expected); err != nil {
-				t.Fatal(err)
-			}
-			if status.Summary != expected.Summary || status.FailureDetail != expected.Detail || polls.Load() != 1 {
-				t.Fatalf("decoder changed legacy fields: %+v, polls=%d", status, polls.Load())
-			}
-		})
-	}
-}
-
-func TestLegacyMCPoll(t *testing.T) {
-	binary := os.Getenv("MC_BINARY")
-	if binary == "" {
-		t.Fatal("MC_BINARY must point to the pinned mc release; this check cannot be skipped")
-	}
-	info, err := buildinfo.ReadFile(binary)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if info.Main.Path != "github.com/minio/mc" || !strings.Contains(info.Main.Version, "7394ce0dd2a8") {
-		t.Fatalf("expected mc RELEASE.2025-08-13T08-35-41Z (7394ce0dd2a8), got %+v", info.Main)
-	}
-	for _, f := range fixtures(t) {
-		t.Run(f.Name, func(t *testing.T) {
-			var polls atomic.Int32
-			server := fixtureServer(t, f.Response, &polls)
-			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-			defer cancel()
-			cmd := exec.CommandContext(ctx, binary, "--config-dir", filepath.Join(t.TempDir(), "mc"), "--json", "admin", "heal", "--recursive", "w23/bucket")
-			cmd.Env = append(os.Environ(), "MC_HOST_w23="+strings.Replace(server.URL, "http://", "http://fixture-access:fixture-secret@", 1), "MC_NO_COLOR=1")
-			output, err := cmd.CombinedOutput()
-			if ctx.Err() != nil {
-				t.Fatalf("legacy poll did not terminate: %s", output)
-			}
-			exit := 0
-			if err != nil {
-				var ok bool
-				var status *exec.ExitError
-				status, ok = err.(*exec.ExitError)
-				if !ok {
-					t.Fatal(err)
-				}
-				exit = status.ExitCode()
-			}
-			if exit != f.CLIExit || polls.Load() != 1 {
-				t.Fatalf("exit=%d expected=%d polls=%d output=%s", exit, f.CLIExit, polls.Load(), output)
-			}
-			if f.CLIExit != 0 && !strings.Contains(string(output), "Heal had an error") {
-				t.Fatalf("failure was not the expected legacy terminal result: %s", output)
-			}
-		})
-	}
-}
diff --git a/scripts/compat/heal-outcome/go.mod b/scripts/compat/heal-outcome/go.mod
deleted file mode 100644
index f7b5bfd1d..000000000
--- a/scripts/compat/heal-outcome/go.mod
+++ /dev/null
@@ -1,42 +0,0 @@
-module rustfs.local/heal-outcome-compat
-
-go 1.25.0
-
-require github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63
-
-require (
-	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/dustin/go-humanize v1.0.1 // indirect
-	github.com/go-ini/ini v1.67.0 // indirect
-	github.com/go-ole/go-ole v1.3.0 // indirect
-	github.com/goccy/go-json v0.10.5 // indirect
-	github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
-	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
-	github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
-	github.com/minio/md5-simd v1.1.2 // indirect
-	github.com/minio/minio-go/v7 v7.0.90 // indirect
-	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect
-	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.63.0 // indirect
-	github.com/prometheus/procfs v0.16.0 // indirect
-	github.com/prometheus/prom2json v1.4.2 // indirect
-	github.com/prometheus/prometheus v0.303.0 // indirect
-	github.com/rs/xid v1.6.0 // indirect
-	github.com/safchain/ethtool v0.5.10 // indirect
-	github.com/secure-io/sio-go v0.3.1 // indirect
-	github.com/shirou/gopsutil/v3 v3.24.5 // indirect
-	github.com/shoenig/go-m1cpu v0.1.6 // indirect
-	github.com/tinylib/msgp v1.2.5 // indirect
-	github.com/tklauser/go-sysconf v0.3.15 // indirect
-	github.com/tklauser/numcpus v0.10.0 // indirect
-	github.com/yusufpapurcu/wmi v1.2.4 // indirect
-	golang.org/x/crypto v0.52.0 // indirect
-	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.13.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
-)
diff --git a/scripts/compat/heal-outcome/go.sum b/scripts/compat/heal-outcome/go.sum
deleted file mode 100644
index 9e3af9f58..000000000
--- a/scripts/compat/heal-outcome/go.sum
+++ /dev/null
@@ -1,97 +0,0 @@
-github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
-github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
-github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
-github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
-github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A=
-github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
-github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
-github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
-github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
-github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
-github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
-github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI=
-github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
-github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
-github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
-github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
-github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
-github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
-github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
-github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
-github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35 h1:PpXWgLPs+Fqr325bN2FD2ISlRRztXibcX6e8f5FR5Dc=
-github.com/lufia/plan9stats v0.0.0-20250317134145-8bc96cf8fc35/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
-github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
-github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
-github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63 h1:ktN/FrMuM9sjvjIbPZYRKeHEzBDOXQdpYUDiNO0CutE=
-github.com/minio/madmin-go/v3 v3.0.107-0.20250415152934-4b504b82db63/go.mod h1:U0bL6ip4yKFwvo0keonUcWFQp0Hd462tOLLeVyPzWmE=
-github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
-github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
-github.com/minio/minio-go/v7 v7.0.90 h1:TmSj1083wtAD0kEYTx7a5pFsv3iRYMsOJ6A4crjA1lE=
-github.com/minio/minio-go/v7 v7.0.90/go.mod h1:uvMUcGrpgeSAAI6+sD3818508nUyMULw94j2Nxku/Go=
-github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
-github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
-github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c h1:dAMKvw0MlJT1GshSTtih8C2gDs04w8dReiOGXrGLNoY=
-github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
-github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
-github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU=
-github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
-github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
-github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
-github.com/prometheus/common v0.63.0 h1:YR/EIY1o3mEFP/kZCD7iDMnLPlGyuU2Gb3HIcXnA98k=
-github.com/prometheus/common v0.63.0/go.mod h1:VVFF/fBIoToEnWRVkYoXEkq3R3paCoxG9PXP74SnV18=
-github.com/prometheus/procfs v0.16.0 h1:xh6oHhKwnOJKMYiYBDWmkHqQPyiY40sny36Cmx2bbsM=
-github.com/prometheus/procfs v0.16.0/go.mod h1:8veyXUu3nGP7oaCxhX6yeaM5u4stL2FeMXnCqhDthZg=
-github.com/prometheus/prom2json v1.4.2 h1:PxCTM+Whqi/eykO1MKsEL0p/zMpxp9ybpsmdFamw6po=
-github.com/prometheus/prom2json v1.4.2/go.mod h1:zuvPm7u3epZSbXPWHny6G+o8ETgu6eAK3oPr6yFkRWE=
-github.com/prometheus/prometheus v0.303.0 h1:wsNNsbd4EycMCphYnTmNY9JASBVbp7NWwJna857cGpA=
-github.com/prometheus/prometheus v0.303.0/go.mod h1:8PMRi+Fk1WzopMDeb0/6hbNs9nV6zgySkU/zds5Lu3o=
-github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
-github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
-github.com/safchain/ethtool v0.5.10 h1:Im294gZtuf4pSGJRAOGKaASNi3wMeFaGaWuSaomedpc=
-github.com/safchain/ethtool v0.5.10/go.mod h1:w9jh2Lx7YBR4UwzLkzCmWl85UY0W2uZdd7/DckVE5+c=
-github.com/secure-io/sio-go v0.3.1 h1:dNvY9awjabXTYGsTF1PiCySl9Ltofk9GA3VdWlo7rRc=
-github.com/secure-io/sio-go v0.3.1/go.mod h1:+xbkjDzPjwh4Axd07pRKSNriS9SCiYksWnZqdnfpQxs=
-github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI=
-github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk=
-github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
-github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
-github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
-github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
-github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/tinylib/msgp v1.2.5 h1:WeQg1whrXRFiZusidTQqzETkRpGjFjcIhW6uqWH09po=
-github.com/tinylib/msgp v1.2.5/go.mod h1:ykjzy2wzgrlvpDCRc4LA8UXy6D8bzMSuAF3WD57Gok0=
-github.com/tklauser/go-sysconf v0.3.15 h1:VE89k0criAymJ/Os65CSn1IXaol+1wrsFHEB8Ol49K4=
-github.com/tklauser/go-sysconf v0.3.15/go.mod h1:Dmjwr6tYFIseJw7a3dRLJfsHAMXZ3nEnL/aZY+0IuI4=
-github.com/tklauser/numcpus v0.10.0 h1:18njr6LDBk1zuna922MgdjQuJFjrdppsZG60sHGfjso=
-github.com/tklauser/numcpus v0.10.0/go.mod h1:BiTKazU708GQTYF4mB+cmlpT2Is1gLk7XVuEeem8LsQ=
-github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
-github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
-golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
-golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
-golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
-golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
-golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
-google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
-gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
-gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=