feat(scanner): observe canonical heal task terminals (#8851)

crates/heal-contracts/src/completion.rs

@@ -0,0 +1,105 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Ephemeral, bounded terminal observations. Delivery is never a heal commit gate.
+
+use crate::{
+    heal_channel::HealScanMode,
+    outcome::{HealExecutionOutcome, HealObjectReceipt},
+};
+use std::sync::{Arc, OnceLock};
+use tokio::sync::broadcast;
+
+pub const MAX_COMPLETION_IDENTITY_BYTES: usize = 4096;
+const COMPLETION_CHANNEL_CAPACITY: usize = 1024;
+
+#[derive(Debug, Clone)]
+pub struct HealCompletion {
+    /// The accepted owner, never a discarded merged request ID.
+    pub task_id: String,
+    pub scan_mode: HealScanMode,
+    pub dry_run: bool,
+    pub execution: Option<HealExecutionOutcome>,
+    /// Only an untruncated single-object result is retained. A missing result
+    /// does not mean absence, success, or payload integrity.
+    pub object: Option<HealObjectReceipt>,
+}
+
+fn sender() -> &'static broadcast::Sender<Arc<HealCompletion>> {
+    static SENDER: OnceLock<broadcast::Sender<Arc<HealCompletion>>> = OnceLock::new();
+    SENDER.get_or_init(|| broadcast::channel(COMPLETION_CHANNEL_CAPACITY).0)
+}
+
+pub fn has_heal_completion_observers() -> bool {
+    sender().receiver_count() > 0
+}
+
+/// Subscribe before admission so a fast worker cannot precede subscription.
+pub fn subscribe_heal_completions() -> broadcast::Receiver<Arc<HealCompletion>> {
+    sender().subscribe()
+}
+
+/// Oversized diagnostics and lagging consumers lose observations, never repairs.
+pub fn publish_heal_completion(completion: HealCompletion) {
+    let identity_bytes = completion.object.as_ref().map_or(0, |object| {
+        object
+            .identity
+            .bucket
+            .capacity()
+            .saturating_add(object.identity.object.capacity())
+            .saturating_add(object.identity.version_id.as_ref().map_or(0, String::capacity))
+    });
+    if completion.task_id.capacity().saturating_add(identity_bytes) <= MAX_COMPLETION_IDENTITY_BYTES {
+        let _ = sender().send(Arc::new(completion));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn completion_delivery_is_bounded_and_does_not_require_a_consumer() {
+        let mut receiver = subscribe_heal_completions();
+        for index in 0..=COMPLETION_CHANNEL_CAPACITY {
+            publish_heal_completion(HealCompletion {
+                task_id: format!("task-{index}"),
+                scan_mode: HealScanMode::Deep,
+                dry_run: false,
+                execution: Some(HealExecutionOutcome::Completed),
+                object: None,
+            });
+        }
+        assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Lagged(1))));
+        while receiver.try_recv().is_ok() {}
+        let mut oversized = String::with_capacity(MAX_COMPLETION_IDENTITY_BYTES + 1);
+        oversized.push_str("short-but-large-allocation");
+        publish_heal_completion(HealCompletion {
+            task_id: oversized,
+            scan_mode: HealScanMode::Deep,
+            dry_run: false,
+            execution: None,
+            object: None,
+        });
+        assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Empty)));
+        drop(receiver);
+        publish_heal_completion(HealCompletion {
+            task_id: "no-consumer".into(),
+            scan_mode: HealScanMode::Deep,
+            dry_run: false,
+            execution: Some(HealExecutionOutcome::Completed),
+            object: None,
+        });
+    }
+}

crates/heal-contracts/src/lib.rs

@@ -12,4 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod completion;
 pub mod heal_channel;
+pub mod outcome;

crates/heal-contracts/src/outcome.rs

@@ -0,0 +1,110 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Typed object results shared by producers and the heal executor.
+
+use serde::{Deserialize, Serialize};
+use std::time::SystemTime;
+use uuid::Uuid;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum HealObjectKind {
+    Object,
+    Metadata,
+    Decode,
+    DeleteMarkerPurge,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct HealObjectIdentity {
+    pub kind: HealObjectKind,
+    pub bucket: String,
+    pub object: String,
+    /// The requested version; None remains unresolved, never an absence proof.
+    pub version_id: Option<String>,
+    pub bucket_incarnation_id: Option<Uuid>,
+    pub pool_index: Option<usize>,
+    pub set_index: Option<usize>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum HealDeferredReason {
+    DanglingDeleteGrace,
+    RetiredMarkerProof,
+    TransientUsageCache,
+    TransientExistenceCheck,
+    Deadline,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum HealFailureClass {
+    Recoverable,
+    RetryExhausted,
+    Permanent,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(
+    tag = "state",
+    content = "details",
+    rename_all = "snake_case",
+    rename_all_fields = "camelCase",
+    deny_unknown_fields
+)]
+pub enum HealObjectDisposition {
+    /// The legacy storage response does not prove the requested check or commit.
+    Unknown,
+    Repaired,
+    VerifiedHealthy,
+    /// Authoritative metadata/presence proof only. This does not certify
+    /// payload integrity.
+    MetadataHealthy,
+    AuthoritativelyAbsent,
+    Deferred {
+        reason: HealDeferredReason,
+        retry_not_before: Option<SystemTime>,
+    },
+    Failed(HealFailureClass),
+    Cancelled,
+    DryRunObserved,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HealObjectReceipt {
+    pub identity: HealObjectIdentity,
+    pub disposition: HealObjectDisposition,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum HealAbortReason {
+    Cancelled,
+    Deadline,
+    Untraversable,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "state", content = "reason", rename_all = "snake_case")]
+pub enum HealExecutionOutcome {
+    #[default]
+    Pending,
+    Running,
+    Completed,
+    CompletedWithErrors,
+    Aborted(HealAbortReason),
+}

crates/heal/src/heal/erasure_healer/admin.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::heal::outcome::HealObjectReceiptExt;
 use crate::heal::outcome::{
     HealDeferredReason, HealFailureClass, HealObjectDisposition, HealObjectIdentity, HealObjectKind,
     HealObjectOutcome as CanonicalObjectOutcome, HealTaskOutcome,

crates/heal/src/heal/manager.rs

@@ -315,6 +315,27 @@ async fn publish_completed_heal(
         aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
     }
     prune_completed_heal_statuses(&mut retained);
+    drop(retained);
+    drop(aliases);
+    if terminal && rustfs_heal_contracts::completion::has_heal_completion_observers() {
+        // Publish the canonical terminal only after ownership settles. Retrying
+        // attempts and command success are not object completion observations.
+        let outcome = completed.outcome.as_deref();
+        let object = outcome
+            .filter(|outcome| completed.heal_type.is_per_object() && !outcome.objects_truncated && outcome.objects.len() == 1)
+            .and_then(|outcome| outcome.objects.front())
+            .map(|object| crate::heal::outcome::HealObjectReceipt {
+                identity: object.identity.clone(),
+                disposition: object.disposition.clone(),
+            });
+        rustfs_heal_contracts::completion::publish_heal_completion(rustfs_heal_contracts::completion::HealCompletion {
+            task_id: task_id.to_owned(),
+            scan_mode: completed.options.scan_mode,
+            dry_run: completed.options.dry_run,
+            execution: outcome.map(|outcome| outcome.execution),
+            object,
+        });
+    }
 }
 
 #[derive(Debug, Clone)]

crates/heal/src/heal/manager/tests.rs

@@ -30,6 +30,7 @@ pub(super) mod admin_overlap;
 mod root_recovery;
 use uuid::Uuid;
 mod auto_scan;
+mod completion;
 mod disposition;
 mod legacy_ambiguity_logging;
 mod running_mainline;

crates/heal/src/heal/manager/tests/completion.rs

@@ -0,0 +1,88 @@
+use super::*;
+use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome};
+use rustfs_heal_contracts::{completion::subscribe_heal_completions, heal_channel::HealScanMode};
+
+#[tokio::test]
+async fn canonical_completion_publishes_typed_objects_only_for_terminal_untruncated_results() {
+    let mut receiver = subscribe_heal_completions();
+    let completed_heals = Mutex::new(HashMap::new());
+    let aliases = Mutex::new(HashMap::new());
+    let id = Uuid::new_v4().to_string();
+    let alias = Uuid::new_v4().to_string();
+    aliases
+        .lock()
+        .await
+        .insert(alias.clone(), HealTaskAlias { task_id: id.clone() });
+    let mut completed = completed_retention_fixture(SystemTime::now());
+    completed.heal_type = HealType::Object {
+        bucket: "bucket".into(),
+        object: "object".into(),
+        version_id: None,
+    };
+    completed.options.scan_mode = HealScanMode::Normal;
+    let mut outcome = HealTaskOutcome::default();
+    outcome.record(HealObjectOutcome {
+        identity: HealObjectIdentity {
+            kind: HealObjectKind::Object,
+            bucket: "bucket".into(),
+            object: "object".into(),
+            version_id: None,
+            bucket_incarnation_id: Some(Uuid::from_u128(44)),
+            pool_index: Some(0),
+            set_index: Some(1),
+        },
+        disposition: HealObjectDisposition::Deferred {
+            reason: HealDeferredReason::Deadline,
+            retry_not_before: None,
+        },
+        detail: None,
+    });
+    outcome.finish(None);
+    completed.outcome = Some(Arc::new(outcome.clone()));
+    completed.status = HealTaskStatus::Retrying {
+        error: "retry".into(),
+        retry_attempt: 1,
+    };
+    publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false).await;
+    while let Ok(observed) = receiver.try_recv() {
+        assert_ne!(observed.task_id, id, "a retry attempt must not publish a terminal");
+    }
+    completed.status = HealTaskStatus::Completed;
+    publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), true).await;
+    let observed = tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            let event = receiver.recv().await.expect("terminal broadcaster stays open");
+            if event.task_id == id {
+                break event;
+            }
+        }
+    })
+    .await
+    .expect("canonical terminal must be published");
+    assert_eq!(observed.scan_mode, HealScanMode::Normal);
+    assert_eq!(observed.execution, Some(HealExecutionOutcome::Completed));
+    assert!(matches!(
+        observed.object.as_ref().expect("single typed outcome").disposition,
+        HealObjectDisposition::Deferred {
+            reason: HealDeferredReason::Deadline,
+            ..
+        }
+    ));
+    assert!(completed_heals.lock().await.contains_key(&alias), "existing alias retention is preserved");
+
+    outcome.objects_truncated = true;
+    completed.outcome = Some(Arc::new(outcome));
+    let truncated_id = Uuid::new_v4().to_string();
+    publish_completed_heal(&completed_heals, &aliases, &truncated_id, completed, true).await;
+    let truncated = tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            let event = receiver.recv().await.expect("terminal broadcaster stays open");
+            if event.task_id == truncated_id {
+                break event;
+            }
+        }
+    })
+    .await
+    .expect("truncated terminal must still be observable");
+    assert!(truncated.object.is_none(), "a retained tail cannot certify a whole result");
+}

crates/heal/src/heal/outcome.rs

@@ -17,95 +17,26 @@
 
 use rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL;
 use serde::{Deserialize, Serialize};
-use std::{collections::VecDeque, time::SystemTime};
+use std::collections::VecDeque;
+#[cfg(test)]
 use uuid::Uuid;
 
 const MAX_OUTCOME_ITEMS: usize = 128;
 const MAX_OUTCOME_BYTES: usize = 64 * 1024;
 const MAX_OUTCOME_DETAIL_BYTES: usize = 1024;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum HealObjectKind {
-    Object,
-    Metadata,
-    Decode,
-    DeleteMarkerPurge,
+pub use rustfs_heal_contracts::outcome::{
+    HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealFailureClass, HealObjectDisposition, HealObjectIdentity,
+    HealObjectKind, HealObjectReceipt,
+};
+
+/// Repair responsibility policy stays private to the heal executor.
+pub(crate) trait HealObjectReceiptExt {
+    fn verified_for(&self, expected: &HealObjectIdentity) -> bool;
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct HealObjectIdentity {
-    pub kind: HealObjectKind,
-    pub bucket: String,
-    pub object: String,
-    /// The requested version; None remains unresolved, never an absence proof.
-    pub version_id: Option<String>,
-    pub bucket_incarnation_id: Option<Uuid>,
-    pub pool_index: Option<usize>,
-    pub set_index: Option<usize>,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum HealDeferredReason {
-    DanglingDeleteGrace,
-    RetiredMarkerProof,
-    TransientUsageCache,
-    TransientExistenceCheck,
-    Deadline,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum HealFailureClass {
-    Recoverable,
-    RetryExhausted,
-    Permanent,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(
-    tag = "state",
-    content = "details",
-    rename_all = "snake_case",
-    rename_all_fields = "camelCase",
-    deny_unknown_fields
-)]
-pub enum HealObjectDisposition {
-    /// The legacy storage response does not prove the requested check or commit.
-    Unknown,
-    Repaired,
-    VerifiedHealthy,
-    /// Authoritative metadata/presence proof only. This does not certify
-    /// payload integrity.
-    MetadataHealthy,
-    AuthoritativelyAbsent,
-    Deferred {
-        reason: HealDeferredReason,
-        retry_not_before: Option<SystemTime>,
-    },
-    Failed(HealFailureClass),
-    Cancelled,
-    DryRunObserved,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct HealObjectOutcome {
-    pub identity: HealObjectIdentity,
-    pub disposition: HealObjectDisposition,
-    pub detail: Option<String>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct HealObjectReceipt {
-    pub identity: HealObjectIdentity,
-    pub disposition: HealObjectDisposition,
-}
-
-impl HealObjectReceipt {
-    pub(crate) fn verified_for(&self, expected: &HealObjectIdentity) -> bool {
+impl HealObjectReceiptExt for HealObjectReceipt {
+    fn verified_for(&self, expected: &HealObjectIdentity) -> bool {
         let disposition_verifies = match self.disposition {
             HealObjectDisposition::Repaired
             | HealObjectDisposition::VerifiedHealthy
@@ -127,6 +58,14 @@ impl HealObjectReceipt {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct HealObjectOutcome {
+    pub identity: HealObjectIdentity,
+    pub disposition: HealObjectDisposition,
+    pub detail: Option<String>,
+}
+
 impl HealObjectOutcome {
     fn retained_bytes(&self) -> usize {
         size_of::<Self>()
@@ -146,25 +85,6 @@ pub enum HealTraversalCoverage {
     Complete,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum HealAbortReason {
-    Cancelled,
-    Deadline,
-    Untraversable,
-}
-
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(tag = "state", content = "reason", rename_all = "snake_case")]
-pub enum HealExecutionOutcome {
-    #[default]
-    Pending,
-    Running,
-    Completed,
-    CompletedWithErrors,
-    Aborted(HealAbortReason),
-}
-
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct HealOutcomeCounters {

crates/heal/src/heal/task.rs

@@ -18,7 +18,7 @@ use crate::heal::{
     erasure_healer::target_outcomes_complete,
     outcome::{
         HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealFailureClass, HealObjectDisposition, HealObjectIdentity,
-        HealObjectKind, HealObjectOutcome, HealObjectReceipt, HealTaskOutcome,
+        HealObjectKind, HealObjectOutcome, HealObjectReceipt, HealObjectReceiptExt, HealTaskOutcome,
     },
     progress::HealProgress,
     resume::{

crates/scanner/src/heal_completion.rs

@@ -0,0 +1,320 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Live candidate diagnostics, deliberately separate from persisted retry hints
+//! and traversal coverage. Missing content/leader identities remain unverified.
+
+use metrics::{counter, gauge};
+use rustfs_heal_contracts::{
+    completion::{HealCompletion, MAX_COMPLETION_IDENTITY_BYTES, subscribe_heal_completions},
+    heal_channel::{HealAdmissionReceipt, HealChannelRequest, HealScanMode},
+    outcome::{HealExecutionOutcome, HealObjectDisposition, HealObjectIdentity, HealObjectKind},
+};
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+use tokio::sync::{broadcast, mpsc};
+
+const MAX_PENDING: usize = 1024;
+const MAX_RECENT: usize = 1024;
+const PENDING_TTL: Duration = Duration::from_secs(60 * 60);
+const RECENT_TTL: Duration = Duration::from_secs(10 * 60);
+
+#[derive(Debug)]
+pub(crate) struct Candidate {
+    identity: Option<HealObjectIdentity>,
+    mode: HealScanMode,
+    dry_run: bool,
+    first_seen: Instant,
+    count: u64,
+}
+
+impl Candidate {
+    fn from_request(request: &HealChannelRequest) -> Option<Self> {
+        let bytes = request
+            .bucket
+            .len()
+            .saturating_add(request.object_prefix.as_ref().map_or(0, String::len))
+            .saturating_add(request.object_version_id.as_ref().map_or(0, String::len));
+        if bytes > MAX_COMPLETION_IDENTITY_BYTES {
+            return None;
+        }
+        Some(Self {
+            identity: request.object_prefix.as_ref().map(|object| HealObjectIdentity {
+                kind: HealObjectKind::Object,
+                bucket: request.bucket.clone(),
+                object: object.clone(),
+                version_id: request.object_version_id.clone(),
+                bucket_incarnation_id: request.expected_bucket_incarnation_id,
+                pool_index: request.pool_index,
+                set_index: request.set_index,
+            }),
+            mode: request.scan_mode.unwrap_or_default(),
+            dry_run: request.dry_run.unwrap_or(false),
+            first_seen: Instant::now(),
+            count: 1,
+        })
+    }
+}
+
+struct Recent {
+    completion: Arc<HealCompletion>,
+    counted: bool,
+}
+
+#[derive(Default)]
+struct Observations {
+    pending: HashMap<String, Candidate>,
+    recent: HashMap<String, Recent>,
+    recent_order: VecDeque<(Instant, String)>,
+}
+
+#[derive(Debug)]
+struct Settlement {
+    candidates: u64,
+    first_task_observation: bool,
+    disposition: &'static str,
+    scope: &'static str,
+    mode: HealScanMode,
+}
+
+impl Settlement {
+    fn new(candidate: Candidate, recent: &mut Recent) -> Self {
+        let completion = &recent.completion;
+        let disposition = match completion.object.as_ref().map(|object| &object.disposition) {
+            Some(HealObjectDisposition::VerifiedHealthy) => "verified_healthy",
+            Some(HealObjectDisposition::MetadataHealthy) => "metadata_healthy",
+            Some(HealObjectDisposition::Repaired) => "repaired",
+            Some(HealObjectDisposition::AuthoritativelyAbsent) => "absent",
+            Some(HealObjectDisposition::Deferred { .. }) => "deferred",
+            Some(HealObjectDisposition::Failed(_)) => "failed",
+            Some(HealObjectDisposition::Cancelled) => "cancelled",
+            Some(HealObjectDisposition::DryRunObserved) => "dry_run",
+            Some(HealObjectDisposition::Unknown) | None => "unknown",
+        };
+        let scope = if candidate.dry_run || completion.dry_run {
+            "dry_run"
+        } else if completion.execution != Some(HealExecutionOutcome::Completed) {
+            "execution_incomplete"
+        } else if candidate.mode == HealScanMode::Deep && completion.scan_mode != HealScanMode::Deep {
+            "insufficient_scan"
+        } else {
+            match (candidate.identity.as_ref(), completion.object.as_ref()) {
+                (Some(expected), Some(actual)) if expected != &actual.identity => "identity_mismatch",
+                (Some(expected), Some(_))
+                    if expected.bucket_incarnation_id.is_some_and(|id| !id.is_nil())
+                        && expected.pool_index.is_some()
+                        && expected.set_index.is_some()
+                        && expected.version_id.as_ref().is_some_and(|version| !version.is_empty()) =>
+                {
+                    // Even an exact version and VerifiedHealthy do not bind the
+                    // candidate's content and leader epoch with today's contract.
+                    "content_and_leader_unknown"
+                }
+                _ => "identity_unknown",
+            }
+        };
+        let first_task_observation = !recent.counted;
+        recent.counted = true;
+        Self {
+            candidates: candidate.count,
+            first_task_observation,
+            disposition,
+            scope,
+            mode: candidate.mode,
+        }
+    }
+
+    fn record(self) {
+        counter!("rustfs_scanner_heal_candidate_terminal_total",
+            "disposition" => self.disposition, "scope" => self.scope, "scan_mode" => self.mode.as_str())
+        .increment(self.candidates);
+        if self.first_task_observation {
+            counter!("rustfs_scanner_heal_task_terminal_observations_total").increment(1);
+        }
+        // No terminal here can certify payload coverage. In particular repaired,
+        // metadata-healthy and authoritative absence never imply verified bytes.
+        record_unknown(self.scope, self.candidates);
+    }
+}
+
+fn record_unknown(reason: &'static str, count: u64) {
+    counter!("rustfs_scanner_heal_candidate_unknown_total", "reason" => reason).increment(count);
+}
+
+impl Observations {
+    fn admit(&mut self, task_id: String, candidate: Candidate) -> Option<Settlement> {
+        if let Some(recent) = self.recent.get_mut(&task_id) {
+            return Some(Settlement::new(candidate, recent));
+        }
+        if let Some(pending) = self.pending.get_mut(&task_id) {
+            pending.count = pending.count.saturating_add(candidate.count);
+            if pending.identity != candidate.identity {
+                pending.identity = None;
+            }
+            if candidate.mode == HealScanMode::Deep {
+                pending.mode = HealScanMode::Deep;
+            }
+            pending.dry_run |= candidate.dry_run;
+        } else if self.pending.len() < MAX_PENDING {
+            self.pending.insert(task_id, candidate);
+        } else {
+            record_unknown("pending_full", candidate.count);
+        }
+        None
+    }
+
+    fn complete(&mut self, completion: Arc<HealCompletion>, now: Instant) -> Option<Settlement> {
+        if self.recent.contains_key(&completion.task_id) {
+            return None;
+        }
+        let task_id = completion.task_id.clone();
+        let mut recent = Recent {
+            completion,
+            counted: false,
+        };
+        let settled = self
+            .pending
+            .remove(&task_id)
+            .map(|candidate| Settlement::new(candidate, &mut recent));
+        if self.recent.len() == MAX_RECENT {
+            self.evict_recent();
+        }
+        self.recent_order.push_back((now, task_id.clone()));
+        self.recent.insert(task_id, recent);
+        settled
+    }
+
+    fn evict_recent(&mut self) {
+        if let Some((_, id)) = self.recent_order.pop_front() {
+            self.recent.remove(&id);
+        }
+    }
+
+    fn lost(&mut self) {
+        for (_, candidate) in self.pending.drain() {
+            record_unknown("completion_lost", candidate.count);
+        }
+        self.recent.clear();
+        self.recent_order.clear();
+    }
+
+    fn maintain(&mut self, now: Instant) {
+        self.pending.retain(|_, candidate| {
+            if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
+                record_unknown("expired", candidate.count);
+                false
+            } else {
+                true
+            }
+        });
+        while self
+            .recent_order
+            .front()
+            .is_some_and(|(at, _)| now.saturating_duration_since(*at) >= RECENT_TTL)
+        {
+            self.evict_recent();
+        }
+        let oldest = self
+            .pending
+            .values()
+            .map(|candidate| now.saturating_duration_since(candidate.first_seen))
+            .max();
+        gauge!("rustfs_scanner_heal_pending_tasks").set(self.pending.len() as f64);
+        gauge!("rustfs_scanner_heal_pending_oldest_seconds").set(oldest.unwrap_or_default().as_secs_f64());
+    }
+}
+
+type AdmissionSender = mpsc::Sender<(String, Candidate)>;
+
+fn observer_from(registry: &Mutex<Option<AdmissionSender>>) -> AdmissionSender {
+    // This lock protects installation only; no I/O or await occurs while held.
+    let mut installed = registry.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if let Some(sender) = installed.as_ref().filter(|sender| !sender.is_closed()) {
+        return sender.clone();
+    }
+    let sender = {
+        let (sender, mut admissions) = mpsc::channel(MAX_PENDING);
+        let mut completions = subscribe_heal_completions();
+        tokio::spawn(async move {
+            let mut observations = Observations::default();
+            let mut tick = tokio::time::interval(Duration::from_secs(5));
+            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            loop {
+                let settlement = tokio::select! {
+                    candidate = admissions.recv() => match candidate {
+                        Some((id, candidate)) => observations.admit(id, candidate),
+                        None => break,
+                    },
+                    completion = completions.recv() => match completion {
+                        Ok(completion) => observations.complete(completion, Instant::now()),
+                        Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                            counter!("rustfs_scanner_heal_completion_lost_total").increment(skipped);
+                            observations.lost();
+                            None
+                        },
+                        Err(broadcast::error::RecvError::Closed) => { observations.lost(); break },
+                    },
+                    _ = tick.tick() => { observations.maintain(Instant::now()); None },
+                };
+                if let Some(settlement) = settlement {
+                    settlement.record();
+                }
+            }
+        });
+        sender
+    };
+    *installed = Some(sender.clone());
+    sender
+}
+
+fn observer() -> AdmissionSender {
+    static OBSERVER: Mutex<Option<AdmissionSender>> = Mutex::new(None);
+    observer_from(&OBSERVER)
+}
+
+/// Capture only the identity actually available at discovery, before dispatch.
+pub(crate) fn prepare(request: &HealChannelRequest) -> Option<Candidate> {
+    let _ = observer();
+    Candidate::from_request(request)
+}
+
+pub(crate) fn observe(candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
+    observe_on(&observer(), candidate, receipt);
+}
+
+fn observe_on(sender: &AdmissionSender, candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
+    match receipt {
+        Ok(receipt) if receipt.result.is_admitted() => {
+            let Some(candidate) = candidate else {
+                record_unknown("identity_too_large", 1);
+                return;
+            };
+            let task_id = &receipt.task_id;
+            if task_id.is_empty() || task_id.len() > 128 {
+                record_unknown("missing_task_id", 1);
+                return;
+            }
+            if sender.try_send((task_id.clone(), candidate)).is_err() {
+                record_unknown("observer_full_or_closed", 1);
+            }
+        }
+        _ => record_unknown("not_admitted", 1),
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/scanner/src/heal_completion/tests.rs

@@ -0,0 +1,364 @@
+use super::*;
+use rustfs_heal_contracts::outcome::{HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectReceipt};
+use uuid::Uuid;
+
+fn request() -> HealChannelRequest {
+    HealChannelRequest {
+        bucket: "bucket".into(),
+        object_prefix: Some("object".into()),
+        object_version_id: Some(Uuid::from_u128(11).to_string()),
+        expected_bucket_incarnation_id: Some(Uuid::from_u128(22)),
+        pool_index: Some(0),
+        set_index: Some(1),
+        scan_mode: Some(HealScanMode::Deep),
+        ..HealChannelRequest::new()
+    }
+}
+
+fn candidate() -> Candidate {
+    Candidate::from_request(&request()).expect("bounded candidate")
+}
+
+fn completion(task: &str, disposition: HealObjectDisposition) -> Arc<HealCompletion> {
+    Arc::new(HealCompletion {
+        task_id: task.into(),
+        scan_mode: HealScanMode::Deep,
+        dry_run: false,
+        execution: Some(HealExecutionOutcome::Completed),
+        object: Some(HealObjectReceipt {
+            identity: candidate().identity.expect("exact object identity"),
+            disposition,
+        }),
+    })
+}
+
+#[test]
+fn merged_candidates_close_only_on_the_canonical_task_once() {
+    let mut observations = Observations::default();
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    assert_eq!(observations.pending.len(), 1);
+    assert!(
+        observations
+            .complete(completion("discarded-id", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_none()
+    );
+    assert_eq!(observations.pending.len(), 1);
+    let terminal = completion("owner", HealObjectDisposition::VerifiedHealthy);
+    let settled = observations
+        .complete(terminal.clone(), Instant::now())
+        .expect("canonical completion settles both candidates");
+    assert_eq!(settled.candidates, 2);
+    assert!(settled.first_task_observation);
+    assert_eq!(settled.scope, "content_and_leader_unknown");
+    assert!(observations.pending.is_empty());
+    assert!(
+        observations.complete(terminal, Instant::now()).is_none(),
+        "duplicate completion is ignored"
+    );
+    let merged_late = observations
+        .admit("owner".into(), candidate())
+        .expect("terminal before merged admission is retained");
+    assert!(!merged_late.first_task_observation);
+    assert_eq!(merged_late.candidates, 1);
+}
+
+#[test]
+fn terminal_before_admission_is_observed_without_claiming_coverage() {
+    let mut observations = Observations::default();
+    assert!(
+        observations
+            .complete(completion("fast", HealObjectDisposition::Repaired), Instant::now())
+            .is_none()
+    );
+    let settled = observations
+        .admit("fast".into(), candidate())
+        .expect("fast terminal retained");
+    assert_eq!(settled.disposition, "repaired");
+    assert_eq!(settled.scope, "content_and_leader_unknown");
+    assert!(settled.first_task_observation);
+}
+
+#[test]
+fn replacement_bucket_version_or_location_cannot_match_an_old_result() {
+    for change in 0..4 {
+        let mut observed = candidate();
+        let identity = observed.identity.as_mut().expect("candidate identity");
+        match change {
+            0 => identity.bucket_incarnation_id = Some(Uuid::from_u128(23)),
+            1 => identity.version_id = Some(Uuid::from_u128(12).to_string()),
+            2 => identity.pool_index = Some(1),
+            _ => identity.set_index = Some(2),
+        }
+        let mut recent = Recent {
+            completion: completion("old", HealObjectDisposition::VerifiedHealthy),
+            counted: false,
+        };
+        assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_mismatch");
+    }
+}
+
+#[test]
+fn unversioned_overwrite_and_unknown_leader_never_certify_payload() {
+    let mut observed = candidate();
+    observed.identity.as_mut().expect("identity").version_id = None;
+    let mut terminal = (*completion("same-name", HealObjectDisposition::VerifiedHealthy)).clone();
+    terminal.object.as_mut().expect("receipt").identity.version_id = None;
+    let mut recent = Recent {
+        completion: Arc::new(terminal),
+        counted: false,
+    };
+    assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_unknown");
+    let mut recent = Recent {
+        completion: completion("old-leader", HealObjectDisposition::VerifiedHealthy),
+        counted: false,
+    };
+    assert_eq!(Settlement::new(candidate(), &mut recent).scope, "content_and_leader_unknown");
+}
+
+#[test]
+fn normal_metadata_and_unproven_repairs_do_not_satisfy_deep_candidates() {
+    for disposition in [HealObjectDisposition::MetadataHealthy, HealObjectDisposition::Repaired] {
+        let mut terminal = (*completion("normal", disposition)).clone();
+        terminal.scan_mode = HealScanMode::Normal;
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        assert_eq!(Settlement::new(candidate(), &mut recent).scope, "insufficient_scan");
+    }
+}
+
+#[test]
+fn command_success_does_not_hide_deferred_failed_cancelled_or_dry_run_objects() {
+    for (disposition, label) in [
+        (
+            HealObjectDisposition::Deferred {
+                reason: HealDeferredReason::Deadline,
+                retry_not_before: None,
+            },
+            "deferred",
+        ),
+        (HealObjectDisposition::Failed(HealFailureClass::Permanent), "failed"),
+        (HealObjectDisposition::Cancelled, "cancelled"),
+        (HealObjectDisposition::DryRunObserved, "dry_run"),
+        (HealObjectDisposition::AuthoritativelyAbsent, "absent"),
+    ] {
+        let mut recent = Recent {
+            completion: completion("completed-command", disposition),
+            counted: false,
+        };
+        let settled = Settlement::new(candidate(), &mut recent);
+        assert_eq!(settled.disposition, label);
+        assert_eq!(settled.scope, "content_and_leader_unknown");
+    }
+    let mut terminal = (*completion("cancelled", HealObjectDisposition::VerifiedHealthy)).clone();
+    terminal.execution = Some(HealExecutionOutcome::Aborted(HealAbortReason::Cancelled));
+    let mut recent = Recent {
+        completion: Arc::new(terminal),
+        counted: false,
+    };
+    assert_eq!(Settlement::new(candidate(), &mut recent).scope, "execution_incomplete");
+}
+
+#[test]
+fn missing_or_truncated_object_result_remains_unknown() {
+    let mut terminal = (*completion("truncated", HealObjectDisposition::VerifiedHealthy)).clone();
+    terminal.object = None;
+    let mut recent = Recent {
+        completion: Arc::new(terminal),
+        counted: false,
+    };
+    assert_eq!(Settlement::new(candidate(), &mut recent).scope, "identity_unknown");
+}
+
+#[test]
+fn overflow_loss_expiry_and_restart_are_bounded_and_do_not_restore_pending() {
+    let mut observations = Observations::default();
+    for index in 0..=MAX_PENDING {
+        observations.admit(format!("pending-{index}"), candidate());
+    }
+    assert_eq!(observations.pending.len(), MAX_PENDING);
+    assert!(!observations.pending.contains_key(&format!("pending-{MAX_PENDING}")));
+    observations.lost();
+    assert!(observations.pending.is_empty());
+    assert!(
+        observations
+            .complete(completion("pending-0", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_none()
+    );
+    observations.admit("expires".into(), candidate());
+    observations.maintain(Instant::now() + PENDING_TTL);
+    assert!(observations.pending.is_empty());
+    assert!(observations.recent.is_empty());
+    let mut restarted = Observations::default();
+    assert!(
+        restarted
+            .complete(completion("expires", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_none()
+    );
+}
+
+#[test]
+fn terminal_history_eviction_loses_evidence_without_inventing_a_completion() {
+    let mut observations = Observations::default();
+    for index in 0..=MAX_RECENT {
+        observations.complete(
+            completion(&format!("terminal-{index}"), HealObjectDisposition::VerifiedHealthy),
+            Instant::now(),
+        );
+    }
+    assert_eq!(observations.recent.len(), MAX_RECENT);
+    assert_eq!(observations.recent_order.len(), MAX_RECENT);
+    assert!(observations.admit("terminal-0".into(), candidate()).is_none());
+    assert!(observations.pending.contains_key("terminal-0"));
+}
+
+#[test]
+fn identity_size_is_bounded_before_cloning() {
+    let mut large = request();
+    large.object_prefix = Some("x".repeat(MAX_COMPLETION_IDENTITY_BYTES));
+    assert!(Candidate::from_request(&large).is_none());
+}
+
+#[test]
+fn observer_reinstalls_after_runtime_teardown() {
+    let registry = Mutex::new(None);
+    let first = {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("first runtime");
+        runtime.block_on(async { observer_from(&registry) })
+    };
+    assert!(first.is_closed(), "runtime teardown closes the old observer");
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("replacement runtime");
+    runtime.block_on(async {
+        let replacement = observer_from(&registry);
+        assert!(!replacement.is_closed(), "a stale global sender must not suppress a new observer");
+        replacement
+            .try_send(("new-runtime".into(), candidate()))
+            .expect("replacement observer admits work");
+    });
+}
+
+#[derive(Default)]
+struct CompletionRecorder(Mutex<HashMap<metrics::Key, Arc<std::sync::atomic::AtomicU64>>>);
+
+impl metrics::Recorder for CompletionRecorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        metrics::Counter::from_arc(
+            self.0
+                .lock()
+                .expect("completion recorder")
+                .entry(key.clone())
+                .or_default()
+                .clone(),
+        )
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        metrics::Histogram::noop()
+    }
+}
+
+impl CompletionRecorder {
+    fn total(&self, metric: &str) -> u64 {
+        self.0
+            .lock()
+            .expect("completion recorder")
+            .iter()
+            .filter(|(key, _)| key.name() == metric)
+            .map(|(_, value)| value.load(std::sync::atomic::Ordering::Relaxed))
+            .sum()
+    }
+
+    async fn wait_for_terminals(&self, expected: u64) {
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while self.total("rustfs_scanner_heal_candidate_terminal_total") < expected {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the live observer must settle admitted candidates");
+    }
+}
+
+// A local installation avoids sharing another test runtime's observer. The
+// current-thread runtime keeps its spawned observer under this local recorder.
+#[tokio::test]
+async fn live_observer_associates_early_completion_with_canonical_merged_receipts() {
+    use rustfs_heal_contracts::{completion::publish_heal_completion, heal_channel::HealAdmissionResult};
+
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let registry = Mutex::new(None);
+    let sender = observer_from(&registry);
+    let canonical_id = Uuid::new_v4().to_string();
+    let sentinel_id = Uuid::new_v4().to_string();
+    let source_request = request();
+    assert_ne!(source_request.id, canonical_id, "merged requests retain their owner's ID");
+
+    observe_on(
+        &sender,
+        Some(candidate()),
+        Ok(&HealAdmissionReceipt {
+            result: HealAdmissionResult::Accepted,
+            task_id: sentinel_id.clone(),
+        }),
+    );
+    publish_heal_completion((*completion(&canonical_id, HealObjectDisposition::VerifiedHealthy)).clone());
+    // FIFO on the real broadcast channel makes this terminal an acknowledgement
+    // that the earlier unmatched completion was consumed before its admission.
+    publish_heal_completion((*completion(&sentinel_id, HealObjectDisposition::MetadataHealthy)).clone());
+    recorder.wait_for_terminals(1).await;
+
+    let merged = HealAdmissionReceipt {
+        result: HealAdmissionResult::Merged,
+        task_id: canonical_id,
+    };
+    observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged));
+    recorder.wait_for_terminals(2).await;
+    observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged));
+    recorder.wait_for_terminals(3).await;
+
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_terminal_total"), 3);
+    assert_eq!(
+        recorder.total("rustfs_scanner_heal_task_terminal_observations_total"),
+        2,
+        "two merged candidates must add only one canonical task observation"
+    );
+    assert_eq!(
+        recorder.total("rustfs_scanner_heal_candidate_unknown_total"),
+        3,
+        "real typed terminals still lack content and leader proof"
+    );
+    let metrics = recorder.0.lock().expect("completion recorder");
+    let (_, verified_healthy) = metrics
+        .iter()
+        .find(|(key, _)| {
+            key.name() == "rustfs_scanner_heal_candidate_terminal_total"
+                && key
+                    .labels()
+                    .any(|label| label.key() == "disposition" && label.value() == "verified_healthy")
+        })
+        .expect("typed storage result must reach the real terminal metric");
+    assert_eq!(verified_healthy.load(std::sync::atomic::Ordering::Relaxed), 2);
+    for (key, _) in metrics
+        .iter()
+        .filter(|(key, _)| key.name() == "rustfs_scanner_heal_candidate_terminal_total")
+    {
+        assert!(
+            key.labels()
+                .any(|label| label.key() == "scope" && label.value() == "content_and_leader_unknown")
+        );
+    }
+}

crates/scanner/src/lib.rs

@@ -59,6 +59,7 @@ use uuid::Uuid;
 
 pub mod data_usage_define;
 pub mod error;
+mod heal_completion;
 pub mod prefix_usage;
 pub mod raw_page_index;
 mod remote_scanner;

crates/scanner/src/scanner_folder.rs

@@ -50,7 +50,7 @@ use rustfs_filemeta::{
 };
 use rustfs_heal_contracts::heal_channel::{
     HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest,
-    HealRequestSource, HealScanMode, send_heal_request_with_admission,
+    HealRequestSource, HealScanMode, send_heal_request_with_receipt,
 };
 use rustfs_scanner_metrics::metrics::{
     CloseDiskGuard, IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource,

crates/scanner/src/scanner_folder/item_actions.rs

@@ -480,8 +480,11 @@ pub(super) async fn send_scanner_heal_request(
 ) -> Result<HealAdmissionResult, ScannerError> {
     let priority = request.priority;
     let trace_context = scanner_heal_candidate_trace_context(&request);
-    match send_heal_request_with_admission(request).await {
-        Ok(result) => {
+    let candidate = crate::heal_completion::prepare(&request);
+    match send_heal_request_with_receipt(request).await {
+        Ok(receipt) => {
+            crate::heal_completion::observe(candidate, Ok(&receipt));
+            let result = receipt.result;
             record_heal_candidate_admission(candidate_type, priority, result);
             if let Some(trace_context) = trace_context.as_ref() {
                 emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace {
@@ -498,6 +501,7 @@ pub(super) async fn send_scanner_heal_request(
             Ok(result)
         }
         Err(err) => {
+            crate::heal_completion::observe(candidate, Err(()));
             counter!(
                 "rustfs_heal_candidate_enqueue_total",
                 "type" => candidate_type.to_string(),

crates/scanner/src/scanner_folder/tests.rs

@@ -2589,20 +2589,21 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
     let heal_starts_clone = heal_starts.clone();
     let healed_versions = Arc::new(Mutex::new(Vec::<Option<String>>::new()));
     let healed_versions_clone = healed_versions.clone();
-    let mut heal_rx =
-        rustfs_heal_contracts::heal_channel::init_heal_channel().expect("heal channel should initialize once for scanner tests");
+    let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels()
+        .expect("heal channels should initialize once for scanner tests");
     let _heal_responder = tokio::spawn(async move {
         while let Some(command) = heal_rx.recv().await {
-            if let rustfs_heal_contracts::heal_channel::HealChannelCommand::Start {
-                request, response_tx, ..
-            } = command
+            let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command;
             {
                 heal_starts_clone.fetch_add(1, Ordering::Relaxed);
                 healed_versions_clone
                     .lock()
                     .expect("heal version capture lock should not be poisoned")
                     .push(request.object_version_id);
-                let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
+                let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt {
+                    result: HealAdmissionResult::Accepted,
+                    task_id: request.id,
+                }));
             }
         }
     });

docs/operations/scanner-runtime-controls.md

@@ -502,6 +502,38 @@ The pacing gate holds neither namespace locks nor I/O/page permits while sleepin
 
 A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements.
 
+## Live Scanner Heal Completions
+
+Scanner admissions now retain the canonical heal task ID, including merged
+requests. Node-local metrics associate these live candidates with the executor's
+typed terminal result:
+
+| Metric | Meaning |
+|---|---|
+| `rustfs_scanner_heal_pending_tasks` | Canonical tasks awaiting a terminal observation. |
+| `rustfs_scanner_heal_pending_oldest_seconds` | Age of the oldest retained pending candidate; refreshed every five seconds. |
+| `rustfs_scanner_heal_candidate_terminal_total` | Associated candidates, labeled by object `disposition`, expected `scan_mode`, and unresolved `scope`. |
+| `rustfs_scanner_heal_task_terminal_observations_total` | Associated task observations, deduplicated within the bounded recent history. This is not an execution count. |
+| `rustfs_scanner_heal_candidate_unknown_total` | Candidates whose integrity remains unknown, labeled by reason. |
+| `rustfs_scanner_heal_completion_lost_total` | Terminal notifications skipped by a lagging observer. |
+
+These metrics use the configured metrics exporter. A task's `Completed` state
+can include a deferred object; the `disposition` label preserves that distinction.
+`MetadataHealthy`, `Repaired`, and authoritative absence never imply verified
+payload bytes. Even `VerifiedHealthy` cannot certify scanner coverage: the current
+contract does not bind the candidate's content identity and leader epoch. No
+verified-byte or full-bucket coverage counter is advanced here.
+
+The admission buffer, pending map, terminal channel, and recent history each hold
+at most 1,024 entries. Retained identities are capped at 4 KiB; terminal history
+expires after ten minutes and pending observations after one hour. Overflow,
+notification loss, missing identity, and expiry remain unknown. Queue cancellation
+or displacement paths without a terminal snapshot also age out as unknown.
+Process restart clears this diagnostic state; a restarted runtime reinstalls its
+observer. Counts and pending age are not a durable integrity ledger or a cold-range
+completion deadline. Existing periodic traversal and pending retry hints provide
+compensation independently.
+
 ## Pending Heal Hints
 
 The scanner's persisted pending-heal cache is a best-effort retry backstop, bounded to 10,000 hints per bucket and a 24-hour age limit. These limits do not authorize garbage collection of committed durable repair obligations. A durable owner must retain its independent replay record until verified object completion or an equivalent durable successor permits removal.