fix(heal): prototype owner observation identity

Cargo.lock

@@ -9858,6 +9858,7 @@ version = "1.0.2"
 dependencies = [
  "serde",
  "serde_json",
+ "thiserror 2.0.21",
  "tokio",
  "uuid",
 ]

crates/heal-contracts/Cargo.toml

@@ -29,6 +29,7 @@ workspace = true
 
 [dependencies]
 serde = { workspace = true, features = ["derive"] }
+thiserror.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }
 

crates/heal-contracts/src/completion.rs

@@ -18,12 +18,25 @@ use crate::{
     heal_channel::HealScanMode,
     outcome::{HealExecutionOutcome, HealObjectReceipt},
 };
-use std::sync::{Arc, OnceLock};
+use std::sync::{
+    Arc, OnceLock,
+    atomic::{AtomicU64, Ordering},
+};
 use tokio::sync::broadcast;
 
 pub const MAX_COMPLETION_IDENTITY_BYTES: usize = 4096;
 const COMPLETION_CHANNEL_CAPACITY: usize = 1024;
 
+/// A process-local observation identity. It is never persisted or sent over RPC.
+pub type HealObservationId = u64;
+
+/// Allocate one identity per accepted owner. Exhaustion is reported as unknown.
+pub fn new_heal_observation_id() -> Option<HealObservationId> {
+    static NEXT: AtomicU64 = AtomicU64::new(1);
+    NEXT.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| value.checked_add(1))
+        .ok()
+}
+
 #[derive(Debug, Clone)]
 pub struct HealCompletion {
     /// The accepted owner, never a discarded merged request ID.
@@ -36,33 +49,84 @@ pub struct HealCompletion {
     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)
+/// A completion paired with an optional in-process owner identity.
+#[derive(Debug, Clone)]
+pub struct ObservedHealCompletion {
+    pub completion: Arc<HealCompletion>,
+    pub observation_id: Option<HealObservationId>,
+}
+
+impl std::ops::Deref for ObservedHealCompletion {
+    type Target = HealCompletion;
+
+    fn deref(&self) -> &Self::Target {
+        &self.completion
+    }
+}
+
+struct CompletionBroadcaster {
+    legacy: broadcast::Sender<Arc<HealCompletion>>,
+    observed: broadcast::Sender<Arc<ObservedHealCompletion>>,
+}
+
+impl CompletionBroadcaster {
+    fn new(capacity: usize) -> Self {
+        Self {
+            legacy: broadcast::channel(capacity).0,
+            observed: broadcast::channel(capacity).0,
+        }
+    }
+
+    fn has_observers(&self) -> bool {
+        self.legacy.receiver_count() > 0 || self.observed.receiver_count() > 0
+    }
+
+    fn publish(&self, completion: HealCompletion, observation_id: Option<HealObservationId>) {
+        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 completion = Arc::new(completion);
+            let _ = self.legacy.send(Arc::clone(&completion));
+            let _ = self.observed.send(Arc::new(ObservedHealCompletion {
+                completion,
+                observation_id,
+            }));
+        }
+    }
+}
+
+fn broadcaster() -> &'static CompletionBroadcaster {
+    static BROADCASTER: OnceLock<CompletionBroadcaster> = OnceLock::new();
+    BROADCASTER.get_or_init(|| CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY))
 }
 
 pub fn has_heal_completion_observers() -> bool {
-    sender().receiver_count() > 0
+    broadcaster().has_observers()
 }
 
 /// Subscribe before admission so a fast worker cannot precede subscription.
 pub fn subscribe_heal_completions() -> broadcast::Receiver<Arc<HealCompletion>> {
-    sender().subscribe()
+    broadcaster().legacy.subscribe()
+}
+
+pub fn subscribe_observed_heal_completions() -> broadcast::Receiver<Arc<ObservedHealCompletion>> {
+    broadcaster().observed.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));
-    }
+    publish_observed_heal_completion(completion, None);
+}
+
+/// Each channel is FIFO independently; relative delivery across channels is not ordered.
+pub fn publish_observed_heal_completion(completion: HealCompletion, observation_id: Option<HealObservationId>) {
+    broadcaster().publish(completion, observation_id);
 }
 
 #[cfg(test)]
@@ -70,36 +134,155 @@ mod tests {
     use super::*;
 
     #[test]
-    fn completion_delivery_is_bounded_and_does_not_require_a_consumer() {
-        let mut receiver = subscribe_heal_completions();
+    fn public_completion_facades_share_receipts_and_preserve_missing_tokens() {
+        // Other completion tests use private broadcasters, keeping this global
+        // facade test isolated from their capacity and lag checks.
+        let mut legacy = subscribe_heal_completions();
+        let mut observed = subscribe_observed_heal_completions();
+        let completion = HealCompletion {
+            task_id: uuid::Uuid::new_v4().to_string(),
+            scan_mode: HealScanMode::Deep,
+            dry_run: false,
+            execution: None,
+            object: None,
+        };
+        publish_heal_completion(completion.clone());
+        let legacy_event = legacy.try_recv().expect("legacy subscriber receives old API event");
+        let observed_event = observed.try_recv().expect("observer receives old API event");
+        assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
+        assert_eq!(legacy_event.task_id, completion.task_id);
+        assert_eq!(observed_event.observation_id, None);
+
+        publish_observed_heal_completion(completion, Some(51));
+        let legacy_event = legacy.try_recv().expect("legacy subscriber receives observed API event");
+        let observed_event = observed.try_recv().expect("observer receives observed API event");
+        assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
+        assert_eq!(observed_event.observation_id, Some(51));
+    }
+
+    #[test]
+    fn legacy_and_observed_subscribers_receive_the_same_completion_in_each_channel_order() {
+        let id = uuid::Uuid::new_v4().to_string();
+        let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
+        let mut legacy = bus.legacy.subscribe();
+        let mut observed = bus.observed.subscribe();
+        assert!(bus.has_observers());
+        for (suffix, token) in [("first", Some(41)), ("second", None)] {
+            bus.publish(
+                HealCompletion {
+                    task_id: format!("{id}-{suffix}"),
+                    scan_mode: HealScanMode::Normal,
+                    dry_run: false,
+                    execution: None,
+                    object: None,
+                },
+                token,
+            );
+        }
+        for (suffix, token) in [("first", Some(41)), ("second", None)] {
+            let expected = format!("{id}-{suffix}");
+            let legacy_event = loop {
+                let event = legacy.try_recv().expect("legacy completion");
+                if event.task_id == expected {
+                    break event;
+                }
+            };
+            let observed_event = loop {
+                let event = observed.try_recv().expect("observed completion");
+                if event.task_id == expected {
+                    break event;
+                }
+            };
+            assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
+            assert_eq!(observed_event.observation_id, token);
+        }
+        drop(legacy);
+        assert!(bus.has_observers(), "observed-only subscribers require completion construction");
+        let completion = HealCompletion {
+            task_id: id.clone(),
+            scan_mode: HealScanMode::Deep,
+            dry_run: false,
+            execution: None,
+            object: None,
+        };
+        bus.publish(completion, None);
+        let observed_event = loop {
+            let event = observed.try_recv().expect("legacy publication reaches observed channel");
+            if event.task_id == id {
+                break event;
+            }
+        };
+        assert_eq!(observed_event.observation_id, None);
+    }
+
+    #[test]
+    fn legacy_and_observed_channels_report_lag_independently() {
+        let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
+        let mut legacy = bus.legacy.subscribe();
+        let mut observed = bus.observed.subscribe();
         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,
-            });
+            bus.publish(
+                HealCompletion {
+                    task_id: format!("dual-lag-{index}"),
+                    scan_mode: HealScanMode::Normal,
+                    dry_run: false,
+                    execution: None,
+                    object: None,
+                },
+                Some(42),
+            );
+        }
+        assert!(matches!(legacy.try_recv(), Err(broadcast::error::TryRecvError::Lagged(_))));
+        assert!(matches!(observed.try_recv(), Err(broadcast::error::TryRecvError::Lagged(_))));
+    }
+
+    #[test]
+    fn completion_delivery_is_bounded_and_does_not_require_a_consumer() {
+        let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
+        assert!(!bus.has_observers());
+        let mut receiver = bus.legacy.subscribe();
+        assert!(bus.has_observers());
+        for index in 0..=COMPLETION_CHANNEL_CAPACITY {
+            bus.publish(
+                HealCompletion {
+                    task_id: format!("task-{index}"),
+                    scan_mode: HealScanMode::Deep,
+                    dry_run: false,
+                    execution: Some(HealExecutionOutcome::Completed),
+                    object: None,
+                },
+                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,
-        });
+        let mut observed = bus.observed.subscribe();
+        bus.publish(
+            HealCompletion {
+                task_id: oversized,
+                scan_mode: HealScanMode::Deep,
+                dry_run: false,
+                execution: None,
+                object: None,
+            },
+            None,
+        );
         assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Empty)));
+        assert!(matches!(observed.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,
-        });
+        drop(observed);
+        assert!(!bus.has_observers());
+        bus.publish(
+            HealCompletion {
+                task_id: "no-consumer".into(),
+                scan_mode: HealScanMode::Deep,
+                dry_run: false,
+                execution: Some(HealExecutionOutcome::Completed),
+                object: None,
+            },
+            None,
+        );
     }
 }

crates/heal-contracts/src/heal_channel.rs

@@ -271,6 +271,39 @@ pub struct HealAdmissionReceipt {
     pub task_id: String,
 }
 
+/// Optional process-local observation metadata for a canonical receipt.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObservedHealAdmissionReceipt {
+    pub receipt: HealAdmissionReceipt,
+    /// Process-local owner identity; absent for remote or unproven owners.
+    pub observation_id: Option<crate::completion::HealObservationId>,
+}
+
+/// Errors from process-local observed admission and its channel initialization.
+#[derive(Debug, thiserror::Error)]
+pub enum HealObservationChannelError {
+    #[error("Heal channel sender already initialized")]
+    AlreadyInitialized,
+    #[error("Heal channel not initialized")]
+    NotInitialized,
+    #[error("Heal receipt command channel closed")]
+    LegacyCommandClosed,
+    #[error("Failed to receive heal admission response: {0}")]
+    LegacyResponseClosed(#[source] oneshot::error::RecvError),
+    #[error("Failed to receive observed heal admission response: {0}")]
+    ObservedResponseClosed(#[source] oneshot::error::RecvError),
+    #[error("{0}")]
+    AdmissionFailed(String),
+}
+
+impl std::ops::Deref for ObservedHealAdmissionReceipt {
+    type Target = HealAdmissionReceipt;
+
+    fn deref(&self) -> &Self::Target {
+        &self.receipt
+    }
+}
+
 impl HealAdmissionResult {
     pub fn result_label(self) -> &'static str {
         match self {
@@ -447,9 +480,19 @@ pub struct HealReceiptCommand {
 /// Canonical-receipt command receiver.
 pub type HealReceiptReceiver = mpsc::UnboundedReceiver<HealReceiptCommand>;
 
+/// Process-local admission command carrying an observed receipt.
+#[derive(Debug)]
+pub struct ObservedHealReceiptCommand {
+    pub request: HealChannelRequest,
+    pub response_tx: oneshot::Sender<Result<ObservedHealAdmissionReceipt, HealObservationChannelError>>,
+}
+
+pub type ObservedHealReceiptReceiver = mpsc::UnboundedReceiver<ObservedHealReceiptCommand>;
+
 struct HealChannelSenders {
     command: HealChannelSender,
     receipt: mpsc::UnboundedSender<HealReceiptCommand>,
+    observed: Option<mpsc::UnboundedSender<ObservedHealReceiptCommand>>,
 }
 
 /// Global heal channel sender
@@ -469,12 +512,31 @@ pub fn init_heal_channel() -> Result<HealChannelReceiver, &'static str> {
 
 /// Initialize the legacy command and canonical-receipt channels atomically.
 pub fn init_heal_channels() -> Result<(HealChannelReceiver, HealReceiptReceiver), &'static str> {
+    let (command_receiver, receipt_receiver, _) =
+        init_heal_channels_inner(false).map_err(|_| "Heal channel sender already initialized")?;
+    Ok((command_receiver, receipt_receiver))
+}
+
+/// Initialize an additional local observation channel while preserving the legacy channels.
+pub fn init_heal_channels_with_observations()
+-> Result<(HealChannelReceiver, HealReceiptReceiver, ObservedHealReceiptReceiver), HealObservationChannelError> {
+    init_heal_channels_inner(true)
+}
+
+fn init_heal_channels_inner(
+    with_observations: bool,
+) -> Result<(HealChannelReceiver, HealReceiptReceiver, ObservedHealReceiptReceiver), HealObservationChannelError> {
     let (command, command_receiver) = mpsc::unbounded_channel();
     let (receipt, receipt_receiver) = mpsc::unbounded_channel();
+    let (observed, observed_receiver) = mpsc::unbounded_channel();
     GLOBAL_HEAL_CHANNEL_SENDERS
-        .set(HealChannelSenders { command, receipt })
-        .map_err(|_| "Heal channel sender already initialized")?;
-    Ok((command_receiver, receipt_receiver))
+        .set(HealChannelSenders {
+            command,
+            receipt,
+            observed: with_observations.then_some(observed),
+        })
+        .map_err(|_| HealObservationChannelError::AlreadyInitialized)?;
+    Ok((command_receiver, receipt_receiver, observed_receiver))
 }
 
 /// Get global heal channel sender
@@ -515,10 +577,17 @@ pub fn subscribe_heal_responses() -> broadcast::Receiver<HealChannelResponse> {
 
 /// Send heal start request and wait for structured admission feedback.
 pub async fn send_heal_request_with_receipt(request: HealChannelRequest) -> Result<HealAdmissionReceipt, String> {
-    let (response_tx, response_rx) = oneshot::channel();
     let senders = GLOBAL_HEAL_CHANNEL_SENDERS
         .get()
         .ok_or_else(|| "Heal channel not initialized".to_string())?;
+    send_heal_request_with_receipt_on(senders, request).await
+}
+
+async fn send_heal_request_with_receipt_on(
+    senders: &HealChannelSenders,
+    request: HealChannelRequest,
+) -> Result<HealAdmissionReceipt, String> {
+    let (response_tx, response_rx) = oneshot::channel();
     senders
         .receipt
         .send(HealReceiptCommand { request, response_tx })
@@ -528,6 +597,46 @@ pub async fn send_heal_request_with_receipt(request: HealChannelRequest) -> Resu
         .map_err(|e| format!("Failed to receive heal admission response: {e}"))?
 }
 
+/// A missing or closed observation channel falls back only before a command is accepted.
+pub async fn send_heal_request_with_observation(
+    request: HealChannelRequest,
+) -> Result<ObservedHealAdmissionReceipt, HealObservationChannelError> {
+    let senders = GLOBAL_HEAL_CHANNEL_SENDERS
+        .get()
+        .ok_or(HealObservationChannelError::NotInitialized)?;
+    send_heal_request_with_observation_on(senders, request).await
+}
+
+async fn send_heal_request_with_observation_on(
+    senders: &HealChannelSenders,
+    mut request: HealChannelRequest,
+) -> Result<ObservedHealAdmissionReceipt, HealObservationChannelError> {
+    if let Some(observed) = &senders.observed {
+        let (response_tx, response_rx) = oneshot::channel();
+        match observed.send(ObservedHealReceiptCommand { request, response_tx }) {
+            Ok(()) => {
+                return response_rx
+                    .await
+                    .map_err(HealObservationChannelError::ObservedResponseClosed)?;
+            }
+            Err(error) => request = error.0.request,
+        }
+    }
+    let (response_tx, response_rx) = oneshot::channel();
+    senders
+        .receipt
+        .send(HealReceiptCommand { request, response_tx })
+        .map_err(|_| HealObservationChannelError::LegacyCommandClosed)?;
+    let receipt = response_rx
+        .await
+        .map_err(HealObservationChannelError::LegacyResponseClosed)?
+        .map_err(HealObservationChannelError::AdmissionFailed)?;
+    Ok(ObservedHealAdmissionReceipt {
+        receipt,
+        observation_id: None,
+    })
+}
+
 /// Send heal start request and wait for structured admission feedback.
 pub async fn send_heal_request_with_admission(request: HealChannelRequest) -> Result<HealAdmissionResult, String> {
     let (response_tx, response_rx) = oneshot::channel();
@@ -775,6 +884,109 @@ mod auto_heal_disk_request_tests {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn observed_receipt_falls_back_only_when_no_observed_command_was_accepted() {
+        for closed in [false, true] {
+            let (command, _command_rx) = mpsc::unbounded_channel();
+            let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
+            let observed = if closed {
+                let (sender, receiver) = mpsc::unbounded_channel();
+                drop(receiver);
+                Some(sender)
+            } else {
+                None
+            };
+            let senders = HealChannelSenders {
+                command,
+                receipt,
+                observed,
+            };
+            let request = HealChannelRequest::new();
+            let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, request).await });
+            let command = receipt_rx.recv().await.expect("legacy receipt command");
+            command
+                .response_tx
+                .send(Ok(HealAdmissionReceipt {
+                    result: HealAdmissionResult::Accepted,
+                    task_id: "owner".into(),
+                }))
+                .expect("caller waiting");
+            let received = task.await.expect("sender task").expect("legacy admission");
+            assert_eq!(received.task_id, "owner");
+            assert_eq!(received.observation_id, None);
+        }
+    }
+
+    #[tokio::test]
+    async fn lost_observed_reply_never_resends_the_request() {
+        let (command, _command_rx) = mpsc::unbounded_channel();
+        let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
+        let keep_receipt_open = receipt.clone();
+        let (observed, mut observed_rx) = mpsc::unbounded_channel();
+        let senders = HealChannelSenders {
+            command,
+            receipt,
+            observed: Some(observed),
+        };
+        let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, HealChannelRequest::new()).await });
+        let command = observed_rx.recv().await.expect("accepted observed command");
+        drop(command.response_tx);
+        let error = task
+            .await
+            .expect("sender task")
+            .expect_err("lost response must remain an error");
+        assert!(matches!(&error, HealObservationChannelError::ObservedResponseClosed(_)));
+        assert!(std::error::Error::source(&error).is_some());
+        assert!(matches!(receipt_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)));
+        drop(keep_receipt_open);
+    }
+
+    #[tokio::test]
+    async fn fallback_reports_a_typed_error_if_legacy_channel_is_closed() {
+        let (command, _command_rx) = mpsc::unbounded_channel();
+        let (receipt, receipt_rx) = mpsc::unbounded_channel();
+        drop(receipt_rx);
+        let senders = HealChannelSenders {
+            command,
+            receipt,
+            observed: None,
+        };
+        let error = send_heal_request_with_observation_on(&senders, HealChannelRequest::new())
+            .await
+            .expect_err("closed legacy channel must not accept work");
+        assert!(matches!(error, HealObservationChannelError::LegacyCommandClosed));
+    }
+
+    #[tokio::test]
+    async fn observed_receipt_keeps_the_owner_token_from_the_processor() {
+        let (command, _command_rx) = mpsc::unbounded_channel();
+        let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
+        let keep_receipt_open = receipt.clone();
+        let (observed, mut observed_rx) = mpsc::unbounded_channel();
+        let senders = HealChannelSenders {
+            command,
+            receipt,
+            observed: Some(observed),
+        };
+        let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, HealChannelRequest::new()).await });
+        let command = observed_rx.recv().await.expect("observed command");
+        command
+            .response_tx
+            .send(Ok(ObservedHealAdmissionReceipt {
+                receipt: HealAdmissionReceipt {
+                    result: HealAdmissionResult::Merged,
+                    task_id: "owner".into(),
+                },
+                observation_id: Some(37),
+            }))
+            .expect("caller waiting");
+        let received = task.await.expect("sender task").expect("observed admission");
+        assert_eq!(received.task_id, "owner");
+        assert_eq!(received.observation_id, Some(37));
+        assert!(matches!(receipt_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)));
+        drop(keep_receipt_open);
+    }
+
     #[test]
     fn heal_request_source_labels_are_stable() {
         assert_eq!(HealRequestSource::Scanner.as_str(), "scanner");

crates/heal/src/heal/channel.rs

@@ -21,8 +21,8 @@ use crate::heal::{
 use crate::{Error, Result};
 use rustfs_heal_contracts::heal_channel::{
     HealAdmissionReceipt, HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest,
-    HealChannelResponse, HealOpts, HealReceiptCommand, HealReceiptReceiver, HealRequestSource, HealScanMode,
-    publish_heal_response,
+    HealChannelResponse, HealObservationChannelError, HealOpts, HealReceiptCommand, HealReceiptReceiver, HealRequestSource,
+    HealScanMode, ObservedHealAdmissionReceipt, ObservedHealReceiptCommand, ObservedHealReceiptReceiver, publish_heal_response,
 };
 use rustfs_madmin::heal_commands::HealResultItem;
 use serde::Serialize;
@@ -30,6 +30,24 @@ use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, error, info};
 
+enum StartResponseSender {
+    Legacy(oneshot::Sender<std::result::Result<HealAdmissionReceipt, String>>),
+    Observed(oneshot::Sender<std::result::Result<ObservedHealAdmissionReceipt, HealObservationChannelError>>),
+}
+
+impl StartResponseSender {
+    fn send(self, result: std::result::Result<ObservedHealAdmissionReceipt, String>) {
+        match self {
+            Self::Legacy(sender) => {
+                let _ = sender.send(result.map(|observed| observed.receipt));
+            }
+            Self::Observed(sender) => {
+                let _ = sender.send(result.map_err(HealObservationChannelError::AdmissionFailed));
+            }
+        }
+    }
+}
+
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_CHANNEL: &str = "channel";
 const EVENT_HEAL_CHANNEL_STATE: &str = "heal_channel_state";
@@ -215,9 +233,22 @@ impl HealChannelProcessor {
 
     /// Start processing legacy and canonical-receipt heal channel requests.
     pub async fn start_with_receipts(
+        &mut self,
+        receiver: HealChannelReceiver,
+        receipt_receiver: HealReceiptReceiver,
+    ) -> Result<()> {
+        let (sender, receiver_observed) = mpsc::unbounded_channel();
+        drop(sender);
+        self.start_with_observed_receipts(receiver, receipt_receiver, receiver_observed)
+            .await
+    }
+
+    /// The three command channels are FIFO independently, with no ordering across channels.
+    pub async fn start_with_observed_receipts(
         &mut self,
         mut receiver: HealChannelReceiver,
         mut receipt_receiver: HealReceiptReceiver,
+        mut observed_receiver: ObservedHealReceiptReceiver,
     ) -> Result<()> {
         info!(
             target: "rustfs::heal::channel",
@@ -229,6 +260,7 @@ impl HealChannelProcessor {
         );
 
         let mut receipt_channel_open = true;
+        let mut observed_channel_open = true;
         loop {
             tokio::select! {
                 command = receiver.recv() => {
@@ -276,6 +308,23 @@ impl HealChannelProcessor {
                         );
                     }
                 }
+                command = observed_receiver.recv(), if observed_channel_open => {
+                    let Some(ObservedHealReceiptCommand { request, response_tx }) = command else {
+                        observed_channel_open = false;
+                        continue;
+                    };
+                    if let Err(e) = self.process_start_request_inner(request, false, true, StartResponseSender::Observed(response_tx)).await {
+                        error!(
+                            target: "rustfs::heal::channel",
+                            event = EVENT_HEAL_CHANNEL_REQUEST,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_CHANNEL,
+                            state = "receipt_process_failed",
+                            error = %e,
+                            "Heal receipt request processing failed"
+                        );
+                    }
+                }
                 response = self.response_receiver.recv() => {
                     if let Some(response) = response {
                         // Handle response if needed
@@ -344,10 +393,21 @@ impl HealChannelProcessor {
     /// Process start request
     async fn process_start_request(
         &self,
-        mut request: HealChannelRequest,
+        request: HealChannelRequest,
         preserve_alias: bool,
         publish_canonical_id: bool,
         response_tx: oneshot::Sender<std::result::Result<HealAdmissionReceipt, String>>,
+    ) -> Result<()> {
+        self.process_start_request_inner(request, preserve_alias, publish_canonical_id, StartResponseSender::Legacy(response_tx))
+            .await
+    }
+
+    async fn process_start_request_inner(
+        &self,
+        mut request: HealChannelRequest,
+        preserve_alias: bool,
+        publish_canonical_id: bool,
+        response_tx: StartResponseSender,
     ) -> Result<()> {
         if request.id.is_empty() {
             request.id = uuid::Uuid::new_v4().to_string();
@@ -369,7 +429,7 @@ impl HealChannelProcessor {
             Ok(heal_request) => heal_request,
             Err(err) => {
                 let error_text = err.to_string();
-                let _ = response_tx.send(Err(error_text.clone()));
+                response_tx.send(Err(error_text.clone()));
                 self.publish_response(HealChannelResponse {
                     request_id: request.id,
                     success: false,
@@ -383,7 +443,7 @@ impl HealChannelProcessor {
         // Submit to heal manager
         match self
             .heal_manager
-            .submit_heal_request_with_receipt_and_alias(heal_request, preserve_alias)
+            .submit_heal_request_with_observation_and_alias(heal_request, preserve_alias)
             .await
         {
             Ok(receipt) => {
@@ -405,7 +465,7 @@ impl HealChannelProcessor {
                     request.id.clone()
                 };
                 self.publish_response(admission_response(response_id, receipt.result));
-                let _ = response_tx.send(Ok(receipt));
+                response_tx.send(Ok(receipt));
             }
             Err(e) => {
                 let error_text = e.to_string();
@@ -419,7 +479,7 @@ impl HealChannelProcessor {
                     error = %error_text,
                     "Heal start submission failed"
                 );
-                let _ = response_tx.send(Err(error_text.clone()));
+                response_tx.send(Err(error_text.clone()));
 
                 // Send error response
                 let response = HealChannelResponse {
@@ -1748,6 +1808,84 @@ mod tests {
             .expect("legacy alias should cancel the canonical task");
     }
 
+    #[tokio::test]
+    async fn observed_and_legacy_channels_share_the_real_admission_owner() {
+        let manager = create_test_heal_manager();
+        let mut processor = HealChannelProcessor::new(manager);
+        let (command_tx, command_rx) = mpsc::unbounded_channel();
+        let (receipt_tx, receipt_rx) = mpsc::unbounded_channel();
+        let (observed_tx, observed_rx) = mpsc::unbounded_channel();
+        let task = tokio::spawn(async move {
+            processor
+                .start_with_observed_receipts(command_rx, receipt_rx, observed_rx)
+                .await
+        });
+        let request = HealChannelRequest {
+            id: "observed-owner".into(),
+            bucket: "observed-owner-bucket".into(),
+            object_prefix: Some("object".into()),
+            source: HealRequestSource::Internal,
+            ..Default::default()
+        };
+        let (response_tx, response_rx) = oneshot::channel();
+        observed_tx
+            .send(ObservedHealReceiptCommand {
+                request: request.clone(),
+                response_tx,
+            })
+            .expect("observed command");
+        let first = tokio::time::timeout(Duration::from_secs(1), response_rx)
+            .await
+            .expect("observed response")
+            .expect("sender")
+            .expect("admitted");
+        assert_eq!(first.result, HealAdmissionResult::Accepted);
+        assert!(first.observation_id.is_some());
+
+        let mut duplicate = request.clone();
+        duplicate.id = "legacy-observed-duplicate".into();
+        let (response_tx, response_rx) = oneshot::channel();
+        receipt_tx
+            .send(HealReceiptCommand {
+                request: duplicate,
+                response_tx,
+            })
+            .expect("legacy command");
+        let legacy = tokio::time::timeout(Duration::from_secs(1), response_rx)
+            .await
+            .expect("legacy response")
+            .expect("sender")
+            .expect("merged");
+        assert_eq!(legacy.result, HealAdmissionResult::Merged);
+        assert_eq!(legacy.task_id, first.task_id);
+
+        let mut duplicate = request;
+        duplicate.id = "observed-duplicate".into();
+        let (response_tx, response_rx) = oneshot::channel();
+        observed_tx
+            .send(ObservedHealReceiptCommand {
+                request: duplicate,
+                response_tx,
+            })
+            .expect("observed duplicate");
+        let merged = tokio::time::timeout(Duration::from_secs(1), response_rx)
+            .await
+            .expect("observed response")
+            .expect("sender")
+            .expect("merged");
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, first.task_id);
+        assert_eq!(merged.observation_id, first.observation_id);
+        drop(command_tx);
+        assert!(
+            tokio::time::timeout(Duration::from_secs(1), task)
+                .await
+                .expect("processor stops")
+                .expect("join")
+                .is_ok()
+        );
+    }
+
     #[tokio::test]
     async fn test_public_legacy_start_processes_commands_with_receipt_channel_closed() {
         let manager = create_test_heal_manager();

crates/heal/src/heal/manager.rs

@@ -29,7 +29,7 @@ use rustfs_concurrency::workload::{ForegroundPressure, foreground_pressure};
 #[cfg(test)]
 use rustfs_concurrency::{AdmissionState, WorkloadClass};
 use rustfs_heal_contracts::heal_channel::{
-    HealAdmissionDropReason, HealAdmissionReceipt, HealAdmissionResult, HealRequestSource,
+    HealAdmissionDropReason, HealAdmissionReceipt, HealAdmissionResult, HealRequestSource, ObservedHealAdmissionReceipt,
 };
 use rustfs_madmin::heal_commands::HealResultItem;
 #[cfg(test)]
@@ -297,6 +297,7 @@ async fn publish_completed_heal(
     task_id: &str,
     completed: CompletedHealStatus,
     terminal: bool,
+    observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
 ) {
     let completed = Arc::new(completed);
     completed.retained_bytes();
@@ -328,13 +329,16 @@ async fn publish_completed_heal(
                 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,
-        });
+        rustfs_heal_contracts::completion::publish_observed_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,
+            },
+            observation_id,
+        );
     }
 }
 
@@ -1292,6 +1296,7 @@ impl HealManager {
         config: &HealConfig,
         context: &'static str,
         replacement_targets_frozen: bool,
+        observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
     ) -> HealAdmissionDecision {
         let replacement_targets_frozen = replacement_targets_frozen || request.retry_attempts != 0;
         let queue_len = queue.len();
@@ -1328,7 +1333,7 @@ impl HealManager {
                 let priority = request.priority;
                 let source = request.source;
                 if let Some(displaced) =
-                    queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced)
+                    queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced, observation_id)
                 {
                     publish_heal_queue_length(queue);
                     Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
@@ -1479,7 +1484,7 @@ impl HealManager {
         let priority = request.priority;
         let source = request.source;
 
-        match queue.push_with_replacement_scope(request, replacement_targets_frozen) {
+        match queue.push_with_replacement_scope(request, replacement_targets_frozen, observation_id) {
             QueuePushOutcome::Accepted => {
                 publish_heal_queue_length(queue);
                 Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
@@ -1951,6 +1956,17 @@ impl HealManager {
         request: HealRequest,
         preserve_alias: bool,
     ) -> Result<HealAdmissionReceipt> {
+        Ok(self
+            .submit_heal_request_with_observation_and_alias(request, preserve_alias)
+            .await?
+            .receipt)
+    }
+
+    pub(crate) async fn submit_heal_request_with_observation_and_alias(
+        &self,
+        request: HealRequest,
+        preserve_alias: bool,
+    ) -> Result<ObservedHealAdmissionReceipt> {
         self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, None)
             .await
     }
@@ -1991,8 +2007,10 @@ impl HealManager {
         mrf_notice_target: MrfRepairNoticeTarget,
     ) -> Result<HealAdmissionReceipt> {
         let preserve_alias = !mrf_notice_target.kind.is_durable();
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target))
-            .await
+        Ok(self
+            .submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target))
+            .await?
+            .receipt)
     }
 
     #[cfg(test)]
@@ -2058,7 +2076,7 @@ impl HealManager {
         preserve_alias: bool,
         accept_same_request_id_replay: bool,
         mrf_notice_target: Option<MrfRepairNoticeTarget>,
-    ) -> Result<HealAdmissionReceipt> {
+    ) -> Result<ObservedHealAdmissionReceipt> {
         let admission_start = Instant::now();
         if request.id.is_empty() {
             request.id = uuid::Uuid::new_v4().to_string();
@@ -2121,17 +2139,19 @@ impl HealManager {
 
         let request_id_admission = active_heals
             .get(&request.id)
-            .map(|task| (request_matches_task(&request, task), "active"))
+            .map(|task| (request_matches_task(&request, task), "active", task.observation_id))
             .or_else(|| {
+                // ID uniqueness across queue keys is not indexed. Avoid a
+                // second heap scan; a same-ID queued receipt is unproven.
                 queue
                     .requests()
                     .find(|queued| queued.id == request.id)
-                    .map(|queued| (request_matches_request(&request, queued), "queued"))
+                    .map(|queued| (request_matches_request(&request, queued), "queued", None))
             })
             .or_else(|| {
                 retrying_heals
                     .get(&request.id)
-                    .map(|retrying| (request_matches_request(&request, &retrying.request), "retrying"))
+                    .map(|retrying| (request_matches_request(&request, &retrying.request), "retrying", retrying.observation_id))
             })
             // A durable-only owner must be resumed by recovery, not overwritten
             // with a new execution budget by a replay before recovery completes.
@@ -2139,9 +2159,9 @@ impl HealManager {
                 durable_owners
                     .iter()
                     .find(|owner| owner.id == request.id)
-                    .map(|_| (false, "durable"))
+                    .map(|_| (false, "durable", None))
             });
-        if let Some((matches_existing, duplicate_state)) = request_id_admission {
+        if let Some((matches_existing, duplicate_state, observation_id)) = request_id_admission {
             let admission = if matches_existing {
                 if accept_same_request_id_replay {
                     HealAdmissionResult::Accepted
@@ -2183,9 +2203,12 @@ impl HealManager {
                 reason = admission.reason_label(),
                 "Heal queue admission reused an existing request id"
             );
-            return Ok(HealAdmissionReceipt {
-                result: admission,
-                task_id: request.id,
+            return Ok(ObservedHealAdmissionReceipt {
+                receipt: HealAdmissionReceipt {
+                    result: admission,
+                    task_id: request.id,
+                },
+                observation_id,
             });
         }
 
@@ -2284,15 +2307,23 @@ impl HealManager {
 
         let duplicate = (!request.force_start && rejection.is_none()).then(|| {
             active_heal_for_dedup_key(&active_heals, &dedup_key)
-                .map(|(task_id, _)| (task_id, "active"))
+                .map(|(task_id, _)| {
+                    let observation_id = active_heals.get(&task_id).and_then(|task| task.observation_id);
+                    (task_id, "active", observation_id)
+                })
                 .or_else(|| {
                     queue
                         .queued_request_id_for_dedup_key(&dedup_key)
-                        .map(|queued_id| (queued_id.to_string(), "queued"))
+                        .map(|queued_id| (queued_id.to_string(), "queued", queue.queued_observation_for_dedup_key(&dedup_key)))
+                })
+                .or_else(|| {
+                    retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| {
+                        let observation_id = retrying_heals.get(&task_id).and_then(|retry| retry.observation_id);
+                        (task_id, "retrying", observation_id)
+                    })
                 })
-                .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
         });
-        if let Some((merged_task_id, duplicate_state)) = duplicate.flatten() {
+        if let Some((merged_task_id, duplicate_state, observation_id)) = duplicate.flatten() {
             // HS-06: under the minio_error overlap policy an exact duplicate
             // admin start reports the typed AlreadyRunning rejection instead
             // of the silent merge (MinIO's ErrHealAlreadyRunning).
@@ -2400,9 +2431,12 @@ impl HealManager {
                 HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
             }
 
-            return Ok(HealAdmissionReceipt {
-                result: admission,
-                task_id: merged_task_id,
+            return Ok(ObservedHealAdmissionReceipt {
+                receipt: HealAdmissionReceipt {
+                    result: admission,
+                    task_id: merged_task_id,
+                },
+                observation_id,
             });
         }
 
@@ -2433,9 +2467,12 @@ impl HealManager {
                 result = "overlap_rejected",
                 "Admin heal start rejected by overlap policy"
             );
-            return Ok(HealAdmissionReceipt {
-                result: HealAdmissionResult::Dropped(reason),
-                task_id: overlap_task_id,
+            return Ok(ObservedHealAdmissionReceipt {
+                receipt: HealAdmissionReceipt {
+                    result: HealAdmissionResult::Dropped(reason),
+                    task_id: overlap_task_id,
+                },
+                observation_id: None,
             });
         }
 
@@ -2477,9 +2514,12 @@ impl HealManager {
                     result = "dropped_by_disposition",
                     "Heal queue admission answered from the disposition memory"
                 );
-                return Ok(HealAdmissionReceipt {
-                    result: admission,
-                    task_id: recorded_task_id,
+                return Ok(ObservedHealAdmissionReceipt {
+                    receipt: HealAdmissionReceipt {
+                        result: admission,
+                        task_id: recorded_task_id,
+                    },
+                    observation_id: None,
                 });
             }
         }
@@ -2507,17 +2547,24 @@ impl HealManager {
                 .lock()
                 .unwrap_or_else(|poisoned| poisoned.into_inner())
                 .contains_key(&request.id);
-        let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit", recovered_replacement);
+        let mut observation_id = rustfs_heal_contracts::completion::new_heal_observation_id();
+        let admission_decision =
+            Self::admit_request_to_queue(&mut queue, request, &config, "submit", recovered_replacement, observation_id);
         let admission = admission_decision.result;
         if durable_handoff_required && !admission.is_admitted() {
             self.root_recovery
                 .remove(&request_id, &request_heal_type, request_source)
                 .await?;
         }
-        if admission == HealAdmissionResult::Merged
-            && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
-        {
-            task_id = queued_id.to_owned();
+        if admission == HealAdmissionResult::Merged {
+            observation_id = None;
+            if let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key) {
+                task_id = queued_id.to_owned();
+                observation_id = queue.queued_observation_for_dedup_key(&dedup_key);
+            }
+        }
+        if !admission.is_admitted() {
+            observation_id = None;
         }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
         let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
@@ -2563,9 +2610,12 @@ impl HealManager {
             lock_timing,
         });
 
-        Ok(HealAdmissionReceipt {
-            result: admission,
-            task_id,
+        Ok(ObservedHealAdmissionReceipt {
+            receipt: HealAdmissionReceipt {
+                result: admission,
+                task_id,
+            },
+            observation_id,
         })
     }
 
@@ -2887,7 +2937,15 @@ impl HealManager {
                     .await?;
                 self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
                 task.cancel().await?;
-                publish_completed_heal(&self.completed_heals, &self.task_aliases, &canonical_task_id, completed, true).await;
+                publish_completed_heal(
+                    &self.completed_heals,
+                    &self.task_aliases,
+                    &canonical_task_id,
+                    completed,
+                    true,
+                    task.observation_id,
+                )
+                .await;
                 active_heals.remove(&canonical_task_id);
                 publish_active_heal_count(&active_heals);
                 info!(
@@ -2998,7 +3056,15 @@ impl HealManager {
                         .await?;
                     self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
                     task.cancel().await?;
-                    publish_completed_heal(&self.completed_heals, &self.task_aliases, task_id, completed, true).await;
+                    publish_completed_heal(
+                        &self.completed_heals,
+                        &self.task_aliases,
+                        task_id,
+                        completed,
+                        true,
+                        task.observation_id,
+                    )
+                    .await;
                 }
                 active_heals.remove(task_id);
                 cancelled += 1;

crates/heal/src/heal/manager/auto_scan.rs

@@ -693,7 +693,7 @@ impl HealManager {
                             let config = config.read().await;
                             let mut queue = heal_queue.lock().await;
                             let admission_decision =
-                                Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan", recovered_replacement);
+                                Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan", recovered_replacement, rustfs_heal_contracts::completion::new_heal_observation_id());
                             let admission = admission_decision.result;
                             let should_notify =
                                 matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;

crates/heal/src/heal/manager/queue.rs

@@ -22,6 +22,8 @@ use super::*;
 pub(super) struct DedupKeyEntry {
     pub(super) refcount: usize,
     pub(super) representative_request_id: String,
+    representative_sequence: u64,
+    representative_observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
 }
 
 /// Priority queue wrapper for heal requests
@@ -70,10 +72,25 @@ pub(super) struct PriorityQueueItem {
     pub(super) sequence: u64,
     pub(super) dedup_key: String,
     pub(super) request: HealRequest,
+    observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
     /// Execution or durable recovery fixed these targets; grace waits can have zero retries.
     pub(super) replacement_targets_frozen: bool,
 }
 
+/// The in-process identity moves with the queue item, including forced duplicates.
+pub(super) struct QueuedHeal {
+    pub(super) request: HealRequest,
+    pub(super) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
+}
+
+impl std::ops::Deref for QueuedHeal {
+    type Target = HealRequest;
+
+    fn deref(&self) -> &Self::Target {
+        &self.request
+    }
+}
+
 impl Eq for PriorityQueueItem {}
 
 impl PartialEq for PriorityQueueItem {
@@ -283,6 +300,7 @@ pub(super) struct HealTaskAlias {
 #[derive(Debug, Clone)]
 pub(super) struct RetryingHeal {
     pub(super) request: HealRequest,
+    pub(super) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
     pub(super) error: String,
     pub(super) cancel_token: CancellationToken,
 }
@@ -307,7 +325,7 @@ impl PriorityHealQueue {
     #[cfg(test)]
     pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
             self.remove_best_effort_count(item.request.source);
             item.request
         })
@@ -318,10 +336,15 @@ impl PriorityHealQueue {
     }
 
     pub(super) fn push(&mut self, request: HealRequest) -> QueuePushOutcome {
-        self.push_with_replacement_scope(request, false)
+        self.push_with_replacement_scope(request, false, rustfs_heal_contracts::completion::new_heal_observation_id())
     }
 
-    pub(super) fn push_with_replacement_scope(&mut self, request: HealRequest, frozen: bool) -> QueuePushOutcome {
+    pub(super) fn push_with_replacement_scope(
+        &mut self,
+        request: HealRequest,
+        frozen: bool,
+        observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
+    ) -> QueuePushOutcome {
         let key = Self::make_dedup_key(&request);
 
         // Check for duplicates unless the caller explicitly forces admission.
@@ -332,20 +355,24 @@ impl PriorityHealQueue {
         // also reserves the dedup key for later non-forced duplicates. The first
         // request that opens the key becomes the named representative for merge
         // receipts (taken before `request` moves into the heap).
+        self.sequence += 1;
+        let sequence = self.sequence;
         self.dedup_keys
             .entry(key.clone())
             .or_insert_with(|| DedupKeyEntry {
                 refcount: 0,
                 representative_request_id: request.id.clone(),
+                representative_sequence: sequence,
+                representative_observation_id: observation_id,
             })
             .refcount += 1;
-        self.sequence += 1;
         let source = request.source;
         self.heap.push(PriorityQueueItem {
             priority: request.priority,
             sequence: self.sequence,
             dedup_key: key,
             request,
+            observation_id,
             replacement_targets_frozen: frozen,
         });
         if is_best_effort_source(source) {
@@ -371,7 +398,12 @@ impl PriorityHealQueue {
         self.can_displace_lower_priority_where(priority, |_| true)
     }
 
-    pub(super) fn push_displacing_lower_priority_where<F>(&mut self, request: HealRequest, can_displace: F) -> Option<HealRequest>
+    pub(super) fn push_displacing_lower_priority_where<F>(
+        &mut self,
+        request: HealRequest,
+        can_displace: F,
+        observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
+    ) -> Option<HealRequest>
     where
         F: Fn(&HealRequest) -> bool,
     {
@@ -402,7 +434,7 @@ impl PriorityHealQueue {
         self.heap = retained;
 
         let displaced = displaced.map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
             self.remove_best_effort_count(item.request.source);
             self.refresh_dedup_representative(&item.dedup_key);
             item.request
@@ -413,7 +445,7 @@ impl PriorityHealQueue {
             // into `debug_assert_eq!` — in release builds (`debug_assertions` off) the whole
             // macro, including its argument expression, is compiled out, which would silently
             // drop the new high-priority request after having already evicted a queued item.
-            let outcome = self.push(request);
+            let outcome = self.push_with_replacement_scope(request, false, observation_id);
             debug_assert_eq!(outcome, QueuePushOutcome::Accepted);
         }
 
@@ -422,7 +454,7 @@ impl PriorityHealQueue {
 
     #[cfg(test)]
     pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option<HealRequest> {
-        self.push_displacing_lower_priority_where(request, |_| true)
+        self.push_displacing_lower_priority_where(request, |_| true, rustfs_heal_contracts::completion::new_heal_observation_id())
     }
 
     /// Get statistics about queue contents by priority
@@ -447,13 +479,13 @@ impl PriorityHealQueue {
     #[cfg(test)]
     pub(super) fn pop(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
             self.remove_best_effort_count(item.request.source);
             item.request
         })
     }
 
-    pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
+    pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<QueuedHeal>, Vec<String>)
     where
         F: Fn(&HealRequest) -> bool,
         G: Fn(&HealRequest) -> Option<String>,
@@ -477,15 +509,18 @@ impl PriorityHealQueue {
 
         (
             selected.map(|item| {
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
                 self.remove_best_effort_count(item.request.source);
-                item.request
+                QueuedHeal {
+                    request: item.request,
+                    observation_id: item.observation_id,
+                }
             }),
             skipped,
         )
     }
 
-    pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
+    pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<QueuedHeal>, Vec<String>)
     where
         F: Fn(&HealRequest) -> bool,
         G: Fn(&HealRequest) -> Option<String>,
@@ -555,10 +590,13 @@ impl PriorityHealQueue {
 
         let chosen = chosen.map(|item| {
             let request = item.request;
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
             self.remove_best_effort_count(request.source);
             self.record_dispatch(&request);
-            request
+            QueuedHeal {
+                request,
+                observation_id: item.observation_id,
+            }
         });
         (chosen, skipped)
     }
@@ -667,12 +705,15 @@ impl PriorityHealQueue {
         }
     }
 
-    pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, key: &str) {
+    pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, key: &str, sequence: u64) {
         if let Some(entry) = dedup_keys.get_mut(key) {
             if entry.refcount <= 1 {
                 dedup_keys.remove(key);
             } else {
                 entry.refcount -= 1;
+                if entry.representative_sequence == sequence {
+                    entry.representative_observation_id = None;
+                }
             }
         }
     }
@@ -747,6 +788,23 @@ impl PriorityHealQueue {
         self.dedup_keys.get(key).map(|entry| entry.representative_request_id.as_str())
     }
 
+    pub(super) fn queued_observation_for_dedup_key(
+        &self,
+        key: &str,
+    ) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
+        self.dedup_keys.get(key).and_then(|entry| entry.representative_observation_id)
+    }
+
+    #[cfg(test)]
+    pub(super) fn queued_observation_for_request_id(
+        &self,
+        id: &str,
+    ) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
+        let mut matching = self.heap.iter().filter(|item| item.request.id == id);
+        let observation_id = matching.next()?.observation_id;
+        if matching.next().is_some() { None } else { observation_id }
+    }
+
     /// Re-elect the representative for `key` from the queue entries holding
     /// it. Needed after a holder leaves the queue *without* becoming active
     /// (canceled by id, or displaced): the former opener may be the request
@@ -764,10 +822,12 @@ impl PriorityHealQueue {
             .heap
             .iter()
             .find(|item| item.dedup_key == key)
-            .map(|item| item.request.id.clone())
+            .map(|item| (item.request.id.clone(), item.sequence, item.observation_id))
             && let Some(entry) = self.dedup_keys.get_mut(key)
         {
-            entry.representative_request_id = id;
+            entry.representative_request_id = id.0;
+            entry.representative_sequence = id.1;
+            entry.representative_observation_id = id.2;
         }
     }
 
@@ -786,7 +846,7 @@ impl PriorityHealQueue {
         while let Some(item) = self.heap.pop() {
             if removed.is_none() && item.request.id == request_id {
                 let key = item.dedup_key.clone();
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key, item.sequence);
                 self.remove_best_effort_count(item.request.source);
                 affected_key = Some(key);
                 removed = Some(item.request);
@@ -812,7 +872,7 @@ impl PriorityHealQueue {
 
         while let Some(item) = self.heap.pop() {
             if should_remove(&item.request) {
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key, item.sequence);
                 self.remove_best_effort_count(item.request.source);
                 affected_keys.push(item.dedup_key);
                 removed.push(item.request);

crates/heal/src/heal/manager/scheduler.rs

@@ -15,6 +15,27 @@ use super::disposition::record_task_disposition;
 /// The heal scheduler: queue consumption loop and its skip/metric helpers.
 use super::*;
 
+pub(super) fn current_task_observation_id(
+    active: &HashMap<String, Arc<HealTask>>,
+    task: &Arc<HealTask>,
+) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
+    active
+        .get(&task.id)
+        .filter(|owner| Arc::ptr_eq(owner, task))
+        .and_then(|_| task.observation_id)
+}
+
+pub(super) fn current_retry_observation_id(
+    retrying: &HashMap<String, RetryingHeal>,
+    task_id: &str,
+    cancel_token: &CancellationToken,
+) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
+    retrying
+        .get(task_id)
+        .filter(|owner| &owner.cancel_token == cancel_token)
+        .and_then(|owner| owner.observation_id)
+}
+
 pub(super) fn retry_admission_delay(
     grace_deadline: Option<u64>,
     ordinary_delay: Duration,
@@ -211,7 +232,9 @@ impl HealManager {
                 queue.pop_runnable_with_fairness(|_| true, |_| None).0
             };
 
-            if let Some(mut request) = selected_request {
+            if let Some(queued) = selected_request {
+                let observation_id = queued.observation_id;
+                let mut request = queued.request;
                 request.options.timeout.get_or_insert(config.task_timeout);
                 let task_priority = request.priority;
                 let task_type_label = heal_request_type_label(&request).to_string();
@@ -229,6 +252,7 @@ impl HealManager {
                 let mainline_pacer = running_mainline_pacer(&request, &config, workload_provider);
                 let task = Arc::new(
                     HealTask::from_replacement_recovery_request(request, storage.clone(), replacement_resume_endpoint)
+                        .with_observation_id(observation_id)
                         .with_mainline_pacer(mainline_pacer)
                         .with_admin_recovery(root_recovery.clone()),
                 );
@@ -336,6 +360,9 @@ impl HealManager {
                     tests::pause_completed_retention_before_publish(&task_id, &completed_status).await;
                     let mut active_heals_guard = active_heals_clone.lock().await;
                     let owns_completion = active_heals_guard.contains_key(&task_id);
+                    // The legacy scheduler checks only the ID; observation needs the
+                    // actual task object because IDs can be reused concurrently.
+                    let observation_id = current_task_observation_id(&active_heals_guard, &task);
                     let cancelled_snapshot = if owns_completion {
                         None
                     } else {
@@ -428,6 +455,7 @@ impl HealManager {
                                 request.id.clone(),
                                 RetryingHeal {
                                     request: request.clone(),
+                                    observation_id,
                                     error: error.clone(),
                                     cancel_token: cancel_token.clone(),
                                 },
@@ -446,6 +474,7 @@ impl HealManager {
                             &task_id,
                             completed_status_entry,
                             terminal_completion,
+                            observation_id,
                         )
                         .await;
                     }
@@ -593,8 +622,18 @@ impl HealManager {
                                 if retry_cancel_token.is_cancelled() || !retrying.contains_key(&retry_request_id) {
                                     return;
                                 }
-                                let admission_decision =
-                                    Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry", true);
+                                // A later retry can reuse the ID. Only the same
+                                // cancellation token proves this is its owner.
+                                let retry_observation_id =
+                                    current_retry_observation_id(&retrying, &retry_request_id, &retry_cancel_token);
+                                let admission_decision = Self::admit_request_to_queue(
+                                    &mut queue,
+                                    retry_request.clone(),
+                                    &retry_config,
+                                    "retry",
+                                    true,
+                                    retry_observation_id,
+                                );
                                 let admission = admission_decision.result;
                                 let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                     && retry_config.event_driven_scheduler_enable;

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

@@ -935,6 +935,7 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() {
 
 #[tokio::test]
 async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
+    let mut terminals = rustfs_heal_contracts::completion::subscribe_observed_heal_completions();
     let manager = HealManager::new_without_root_recovery_for_test(
         Arc::new(MockStorage),
         Some(HealConfig {
@@ -948,10 +949,11 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
     request.options.timeout = Some(Duration::from_secs(60));
     let task_id = request.id.clone();
     let created_at = request.created_at;
-    manager
-        .submit_heal_request(request)
+    let receipt = manager
+        .submit_heal_request_with_observation_and_alias(request, false)
         .await
         .expect("admit explicit-budget task");
+    let observation_id = receipt.observation_id.expect("accepted owner has an observation identity");
 
     let mut previous_remaining = Duration::from_secs(60);
     for expected_attempt in 1..=3 {
@@ -967,7 +969,9 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
         .await
         .expect("recoverable executor failure should enter retry backoff");
 
-        let retry = manager.retrying_heals.lock().await[&task_id].request.clone();
+        let retry_owner = manager.retrying_heals.lock().await[&task_id].clone();
+        assert_eq!(retry_owner.observation_id, Some(observation_id));
+        let retry = retry_owner.request;
         assert_eq!(retry.id, task_id);
         assert_eq!(retry.created_at, created_at);
         assert_eq!(retry.source, HealRequestSource::Admin);
@@ -993,6 +997,10 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
         .await
         .expect("retry backoff should requeue the same task id");
         assert_eq!(manager.operations_snapshot().await.queue_length, 1);
+        assert_eq!(
+            manager.heal_queue.lock().await.queued_observation_for_request_id(&task_id),
+            Some(observation_id)
+        );
     }
 
     process_manager_queue_once(&manager).await;
@@ -1013,6 +1021,17 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
     );
     assert_eq!(manager.operations_snapshot().await.queue_length, 0);
     assert_eq!(manager.operations_snapshot().await.retrying_tasks, 0);
+    let terminal = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let terminal = terminals.recv().await.expect("terminal channel remains open");
+            if terminal.task_id == task_id {
+                break terminal;
+            }
+        }
+    })
+    .await
+    .expect("retry exhaustion emits one terminal");
+    assert_eq!(terminal.observation_id, Some(observation_id));
 }
 
 struct ManagerRecoveryTestHook {
@@ -1206,6 +1225,32 @@ fn test_push_displacing_lower_priority_actually_enqueues_new_request() {
     assert_eq!(queue.len(), 0);
 }
 
+#[test]
+fn queued_observation_tracks_the_holder_sequence_when_ids_repeat() {
+    let mut queue = PriorityHealQueue::new();
+    let first = bucket_request("same-owner-key", HealPriority::Normal, HealRequestSource::Internal);
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let mut second = first.clone();
+    second.force_start = true;
+    assert_eq!(queue.push_with_replacement_scope(first, false, Some(11)), QueuePushOutcome::Accepted);
+    assert_eq!(queue.push_with_replacement_scope(second, false, Some(12)), QueuePushOutcome::Accepted);
+    assert_eq!(queue.queued_observation_for_dedup_key(&key), Some(11));
+    assert_eq!(
+        queue.queued_observation_for_request_id(&queue.heap.peek().expect("holder").request.id),
+        None
+    );
+
+    let (first, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
+    assert_eq!(first.expect("first holder").observation_id, Some(11));
+    assert_eq!(
+        queue.queued_observation_for_dedup_key(&key),
+        None,
+        "a departed representative is unproven"
+    );
+    let (second, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
+    assert_eq!(second.expect("second holder").observation_id, Some(12));
+}
+
 #[test]
 fn queued_request_id_for_dedup_key_tracks_the_representative() {
     let mut queue = PriorityHealQueue::new();
@@ -1214,6 +1259,7 @@ fn queued_request_id_for_dedup_key_tracks_the_representative() {
     let first_id = first.id.clone();
     let first_key = PriorityHealQueue::make_dedup_key(&first);
     assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+    assert!(queue.queued_observation_for_dedup_key(&first_key).is_some());
 
     // A forced duplicate of the same target opens a second entry under
     // the same key; the representative stays the request that opened it.
@@ -1341,6 +1387,7 @@ async fn durable_mrf_anchor_snapshot_follows_live_owners_and_ignores_stale_notic
     manager.retrying_heals.lock().await.insert(
         "retry-merged-active".into(),
         RetryingHeal {
+            observation_id: None,
             request: retry_request,
             error: "retry pending".into(),
             cancel_token: CancellationToken::new(),
@@ -2590,6 +2637,7 @@ async fn automatic_replacement_frozen_targets_reject_late_disks_but_cover_duplic
                 manager.retrying_heals.lock().await.insert(
                     owner_id.clone(),
                     RetryingHeal {
+                        observation_id: None,
                         request: owner,
                         error: "target unavailable".to_string(),
                         cancel_token: CancellationToken::new(),
@@ -2601,7 +2649,7 @@ async fn automatic_replacement_frozen_targets_reject_late_disks_but_cover_duplic
                 let mut queue = manager.heal_queue.lock().await;
                 let context = if state == "grace" { "retry" } else { "auto_scan" };
                 assert_eq!(
-                    HealManager::admit_request_to_queue(&mut queue, owner, &config, context, true).result,
+                    HealManager::admit_request_to_queue(&mut queue, owner, &config, context, true, None).result,
                     HealAdmissionResult::Accepted
                 );
                 assert!(
@@ -2725,7 +2773,7 @@ fn automatic_replacement_direct_admission_checks_frozen_targets_before_capacity(
     let owner_id = owner.id.clone();
     assert_eq!(queue.push(owner), QueuePushOutcome::Accepted);
     let admit = |queue: &mut PriorityHealQueue, request, frozen| {
-        HealManager::admit_request_to_queue(queue, request, &config, "auto_scan", frozen).result
+        HealManager::admit_request_to_queue(queue, request, &config, "auto_scan", frozen, None).result
     };
     assert_eq!(
         admit(&mut queue, automatic_replacement_request(&["replacement-b"]), false),
@@ -3958,6 +4006,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
     manager.retrying_heals.lock().await.insert(
         task_id.clone(),
         RetryingHeal {
+            observation_id: None,
             request: request.clone(),
             error: "Lock acquisition timeout".to_string(),
             cancel_token: cancel_token.clone(),
@@ -4510,6 +4559,7 @@ async fn test_operations_snapshot_counts_retry_backoff_as_owned_work() {
     manager.retrying_heals.lock().await.insert(
         task_id,
         RetryingHeal {
+            observation_id: None,
             request,
             error: "transient".to_string(),
             cancel_token: CancellationToken::new(),
@@ -4789,6 +4839,7 @@ async fn test_get_task_report_preserves_retrying_options() {
     manager.retrying_heals.lock().await.insert(
         task_id.clone(),
         RetryingHeal {
+            observation_id: None,
             request,
             error: "transient".to_string(),
             cancel_token: CancellationToken::new(),

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

@@ -37,6 +37,7 @@ async fn install_owner(manager: &HealManager, request: HealRequest, state: Owner
             manager.retrying_heals.lock().await.insert(
                 id.clone(),
                 RetryingHeal {
+                    observation_id: None,
                     request,
                     error: "recoverable fixture error".to_string(),
                     cancel_token: CancellationToken::new(),

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

@@ -1,6 +1,208 @@
 use super::*;
 use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome};
-use rustfs_heal_contracts::{completion::subscribe_heal_completions, heal_channel::HealScanMode};
+use rustfs_heal_contracts::{
+    completion::{subscribe_heal_completions, subscribe_observed_heal_completions},
+    heal_channel::HealScanMode,
+};
+
+#[test]
+fn observation_requires_the_current_task_or_retry_object() {
+    let request = bucket_request("observation-object-owner", HealPriority::Normal, HealRequestSource::Internal);
+    let id = request.id.clone();
+    let stale = Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(301)));
+    let current = Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(302)));
+    let active = HashMap::from([(id.clone(), Arc::clone(&current))]);
+    assert_eq!(current_task_observation_id(&active, &stale), None);
+    assert_eq!(current_task_observation_id(&active, &current), Some(302));
+
+    let current_token = CancellationToken::new();
+    let stale_token = CancellationToken::new();
+    let retrying = HashMap::from([(
+        id.clone(),
+        RetryingHeal {
+            request,
+            observation_id: Some(303),
+            error: "retrying".into(),
+            cancel_token: current_token.clone(),
+        },
+    )]);
+    assert_eq!(current_retry_observation_id(&retrying, &id, &stale_token), None);
+    assert_eq!(current_retry_observation_id(&retrying, &id, &current_token), Some(303));
+}
+
+#[tokio::test]
+async fn queued_same_id_replay_is_unknown_but_canonical_merge_retains_owner_identity() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let request = bucket_request("queued-id-replay-observation", HealPriority::Normal, HealRequestSource::Internal);
+    let first = manager
+        .submit_heal_request_with_observation_and_alias(request.clone(), false)
+        .await
+        .expect("first owner");
+    assert!(first.observation_id.is_some());
+    let replay = manager
+        .submit_heal_request_with_observation_and_alias(request, false)
+        .await
+        .expect("same ID replay");
+    assert_eq!(replay.result, HealAdmissionResult::Accepted);
+    assert_eq!(replay.task_id, first.task_id);
+    assert_eq!(replay.observation_id, None);
+
+    let duplicate = bucket_request("queued-id-replay-observation", HealPriority::Normal, HealRequestSource::Scanner);
+    let merged = manager
+        .submit_heal_request_with_observation_and_alias(duplicate, false)
+        .await
+        .expect("canonical merge");
+    assert_eq!(merged.result, HealAdmissionResult::Merged);
+    assert_eq!(merged.task_id, first.task_id);
+    assert_eq!(merged.observation_id, first.observation_id);
+}
+
+#[tokio::test]
+async fn merged_receipt_uses_live_active_or_retrying_owner_identity() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let request = bucket_request("owner-observation-states", HealPriority::Normal, HealRequestSource::Internal);
+    let id = request.id.clone();
+    manager.active_heals.lock().await.insert(
+        id.clone(),
+        Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(201))),
+    );
+    let scanner = bucket_request("owner-observation-states", HealPriority::Normal, HealRequestSource::Scanner);
+    let active = manager
+        .submit_heal_request_with_observation_and_alias(scanner.clone(), false)
+        .await
+        .expect("active merge");
+    assert_eq!(active.result, HealAdmissionResult::Merged);
+    assert_eq!(active.task_id, id);
+    assert_eq!(active.observation_id, Some(201));
+    manager.active_heals.lock().await.remove(&id);
+    manager.retrying_heals.lock().await.insert(
+        id.clone(),
+        RetryingHeal {
+            request,
+            observation_id: Some(202),
+            error: "retrying".into(),
+            cancel_token: CancellationToken::new(),
+        },
+    );
+    let mut scanner_retry = scanner;
+    scanner_retry.id = Uuid::new_v4().to_string();
+    let retrying = manager
+        .submit_heal_request_with_observation_and_alias(scanner_retry, false)
+        .await
+        .expect("retry merge");
+    assert_eq!(retrying.result, HealAdmissionResult::Merged);
+    assert_eq!(retrying.task_id, id);
+    assert_eq!(retrying.observation_id, Some(202));
+}
+
+#[tokio::test]
+async fn cancelled_or_displaced_queue_owner_does_not_reuse_its_observation_identity() {
+    for displace in [false, true] {
+        let manager = HealManager::new_without_root_recovery_for_test(
+            Arc::new(MockStorage),
+            Some(HealConfig {
+                queue_size: 1,
+                event_driven_scheduler_enable: false,
+                ..Default::default()
+            }),
+        );
+        let request = bucket_request("removed-observation-owner", HealPriority::Low, HealRequestSource::Scanner);
+        let old = manager
+            .submit_heal_request_with_observation_and_alias(request.clone(), false)
+            .await
+            .expect("first owner");
+        assert_eq!(old.result, HealAdmissionResult::Accepted);
+        assert!(old.observation_id.is_some());
+        if displace {
+            let urgent = bucket_request("displacing-owner", HealPriority::Urgent, HealRequestSource::Internal);
+            let urgent_id = urgent.id.clone();
+            let admitted = manager
+                .submit_heal_request_with_observation_and_alias(urgent, false)
+                .await
+                .expect("displacing owner");
+            assert_eq!(admitted.result, HealAdmissionResult::Accepted);
+            assert!(
+                manager
+                    .heal_queue
+                    .lock()
+                    .await
+                    .requests()
+                    .all(|queued| queued.id != old.task_id)
+            );
+            manager.cancel_task(&urgent_id).await.expect("free queue after displacement");
+        } else {
+            manager.cancel_task(&old.task_id).await.expect("cancel queued owner");
+        }
+        let next = manager
+            .submit_heal_request_with_observation_and_alias(request, false)
+            .await
+            .expect("readmit same id");
+        assert_eq!(next.result, HealAdmissionResult::Accepted);
+        assert_eq!(next.task_id, old.task_id);
+        assert!(next.observation_id.is_some());
+        assert_ne!(next.observation_id, old.observation_id);
+    }
+}
+
+#[tokio::test]
+async fn reused_id_has_distinct_owner_generations_in_receipts_and_terminals() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let mut receiver = subscribe_observed_heal_completions();
+    let request = bucket_request("observation-id-reuse", HealPriority::Normal, HealRequestSource::Internal);
+
+    let first = manager
+        .submit_heal_request_with_observation_and_alias(request.clone(), false)
+        .await
+        .expect("first owner");
+    let mut scanner = bucket_request("observation-id-reuse", HealPriority::Normal, HealRequestSource::Scanner);
+    let merged = manager
+        .submit_heal_request_with_observation_and_alias(scanner.clone(), false)
+        .await
+        .expect("first merge");
+    assert_eq!(merged.result, HealAdmissionResult::Merged);
+    assert_eq!(merged.task_id, first.task_id);
+    assert!(first.observation_id.is_some());
+    assert_eq!(merged.observation_id, first.observation_id);
+    process_manager_queue_once(&manager).await;
+    let first_terminal = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let terminal = receiver.recv().await.expect("terminal channel");
+            if terminal.task_id == first.task_id {
+                break terminal;
+            }
+        }
+    })
+    .await
+    .expect("first terminal");
+    assert_eq!(first_terminal.observation_id, first.observation_id);
+
+    let second = manager
+        .submit_heal_request_with_observation_and_alias(request, false)
+        .await
+        .expect("reused id");
+    assert_eq!(second.result, HealAdmissionResult::Accepted);
+    assert_eq!(second.task_id, first.task_id);
+    assert_ne!(second.observation_id, first.observation_id);
+    scanner.id = Uuid::new_v4().to_string();
+    let merged_again = manager
+        .submit_heal_request_with_observation_and_alias(scanner, false)
+        .await
+        .expect("second merge");
+    assert_eq!(merged_again.result, HealAdmissionResult::Merged);
+    assert_eq!(merged_again.observation_id, second.observation_id);
+    process_manager_queue_once(&manager).await;
+    let second_terminal = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            let terminal = receiver.recv().await.expect("terminal channel");
+            if terminal.task_id == second.task_id && terminal.observation_id == second.observation_id {
+                break terminal;
+            }
+        }
+    })
+    .await
+    .expect("second terminal");
+    assert_eq!(second_terminal.observation_id, second.observation_id);
+}
 
 #[tokio::test]
 async fn canonical_completion_publishes_typed_objects_only_for_terminal_untruncated_results() {
@@ -43,12 +245,12 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
         error: "retry".into(),
         retry_attempt: 1,
     };
-    publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false).await;
+    publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false, Some(1)).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;
+    publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), true, Some(1)).await;
     let observed = tokio::time::timeout(Duration::from_secs(1), async {
         loop {
             let event = receiver.recv().await.expect("terminal broadcaster stays open");
@@ -73,7 +275,7 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
     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;
+    publish_completed_heal(&completed_heals, &aliases, &truncated_id, completed, true, Some(1)).await;
     let truncated = tokio::time::timeout(Duration::from_secs(1), async {
         loop {
             let event = receiver.recv().await.expect("terminal broadcaster stays open");

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

@@ -448,6 +448,7 @@ async fn root_recovery_retry_cancellation_preserves_previous_attempt_outcome() {
     manager.retrying_heals.lock().await.insert(
         request.id.clone(),
         RetryingHeal {
+            observation_id: None,
             request: request.clone(),
             error: "retryable".to_string(),
             cancel_token: CancellationToken::new(),
@@ -2153,6 +2154,30 @@ async fn root_recovery_cluster_rejects_inventory_without_checkpoint_headroom() {
     );
 }
 
+#[tokio::test]
+async fn root_recovery_replayed_queue_owner_gets_a_fresh_in_process_observation_identity() {
+    let (_temp, disk) = recovery_disk().await;
+    let manager = recovery_manager(vec![disk.clone()]);
+    let request = root_request();
+    let first = manager
+        .submit_heal_request_with_observation_and_alias(request.clone(), false)
+        .await
+        .expect("persisted admission");
+    assert_eq!(first.result, HealAdmissionResult::Accepted);
+    assert!(first.observation_id.is_some());
+    drop(manager);
+
+    let replayed = recovery_manager(vec![disk]);
+    replayed.replay_root_heals().await.expect("replay persisted root");
+    let recovered_id = replayed
+        .heal_queue
+        .lock()
+        .await
+        .queued_observation_for_request_id(&request.id);
+    assert!(recovered_id.is_some());
+    assert_ne!(recovered_id, first.observation_id);
+}
+
 #[tokio::test]
 async fn root_recovery_queued_owner_is_not_priority_displaced() {
     let (_temp, disk) = recovery_disk().await;

crates/heal/src/heal/task.rs

@@ -413,6 +413,8 @@ pub struct HealResultWindow {
 pub struct HealTask {
     /// Task ID
     pub id: String,
+    /// Process-local diagnostic identity; never persisted or part of a request.
+    pub(crate) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
     /// Heal type
     pub heal_type: HealType,
     pub bucket_incarnation_id: Option<Uuid>,
@@ -494,6 +496,7 @@ impl HealTask {
     pub fn from_request(request: HealRequest, storage: Arc<dyn HealStorageAPI>) -> Self {
         Self {
             id: request.id,
+            observation_id: None,
             heal_type: request.heal_type,
             bucket_incarnation_id: request.bucket_incarnation_id,
             expected_mrf_bucket_incarnation_id: request.expected_mrf_bucket_incarnation_id,
@@ -529,6 +532,14 @@ impl HealTask {
         }
     }
 
+    pub(crate) fn with_observation_id(
+        mut self,
+        observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
+    ) -> Self {
+        self.observation_id = observation_id;
+        self
+    }
+
     pub fn retry_request(&self) -> HealRequest {
         HealRequest {
             id: self.id.clone(),

crates/heal/src/lib.rs

@@ -280,12 +280,12 @@ pub async fn init_heal_manager_with_workload_provider(
             .is_some_and(|hook| hook.fail_channel_init.load(Ordering::SeqCst));
         #[cfg(not(test))]
         let force_channel_failure = false;
-        let channel_receiver = if force_channel_failure {
-            Err("forced heal channel initialization failure")
-        } else {
-            rustfs_heal_contracts::heal_channel::init_heal_channels()
-        };
-        let (receiver, receipt_receiver) = match channel_receiver {
+        if force_channel_failure {
+            stop_initializing_manager(&heal_manager).await?;
+            return Err(Error::Config("forced heal channel initialization failure".to_string()));
+        }
+        let channel_receiver = rustfs_heal_contracts::heal_channel::init_heal_channels_with_observations();
+        let (receiver, receipt_receiver, observed_receiver) = match channel_receiver {
             Ok(receivers) => receivers,
             Err(err) => {
                 stop_initializing_manager(&heal_manager).await?;
@@ -302,7 +302,10 @@ pub async fn init_heal_manager_with_workload_provider(
 
         tokio::spawn(async move {
             let mut processor = channel_processor.lock().await;
-            if let Err(e) = processor.start_with_receipts(receiver, receipt_receiver).await {
+            if let Err(e) = processor
+                .start_with_observed_receipts(receiver, receipt_receiver, observed_receiver)
+                .await
+            {
                 error!(
                     target: "rustfs::heal",
                     event = EVENT_HEAL_RUNTIME_STATE,

crates/scanner/src/heal_completion.rs

@@ -17,8 +17,8 @@
 
 use metrics::{counter, gauge};
 use rustfs_heal_contracts::{
-    completion::{HealCompletion, MAX_COMPLETION_IDENTITY_BYTES, subscribe_heal_completions},
-    heal_channel::{HealAdmissionReceipt, HealChannelRequest, HealScanMode},
+    completion::{MAX_COMPLETION_IDENTITY_BYTES, ObservedHealCompletion, subscribe_observed_heal_completions},
+    heal_channel::{HealChannelRequest, HealObservationChannelError, HealScanMode, ObservedHealAdmissionReceipt},
     outcome::{HealAbortReason, HealExecutionOutcome, HealObjectDisposition, HealObjectIdentity, HealObjectKind},
 };
 use serde::Serialize;
@@ -125,15 +125,21 @@ impl Candidate {
 }
 
 struct Recent {
-    completion: Arc<HealCompletion>,
+    completion: Arc<ObservedHealCompletion>,
     counted: bool,
 }
 
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+struct ObservationKey {
+    task_id: String,
+    observation_id: rustfs_heal_contracts::completion::HealObservationId,
+}
+
 #[derive(Default)]
 struct Observations {
-    pending: HashMap<String, Candidate>,
-    recent: HashMap<String, Recent>,
-    recent_order: VecDeque<(Instant, String)>,
+    pending: HashMap<ObservationKey, Candidate>,
+    recent: HashMap<ObservationKey, Recent>,
+    recent_order: VecDeque<(Instant, ObservationKey)>,
     // All keys come from fixed execution/disposition/scope/mode labels, never task or object identities.
     terminal_counts: BTreeMap<(&'static str, &'static str, &'static str, &'static str), u64>,
     terminal_tasks: u64,
@@ -240,13 +246,13 @@ impl Observations {
         self.record_unknown(settlement.scope, settlement.candidates);
     }
 
-    fn admit(&mut self, task_id: String, candidate: Candidate) -> Option<Settlement> {
-        if let Some(recent) = self.recent.get_mut(&task_id) {
+    fn admit(&mut self, key: ObservationKey, candidate: Candidate) -> Option<Settlement> {
+        if let Some(recent) = self.recent.get_mut(&key) {
             let settlement = Settlement::new(candidate, recent);
             self.record_settlement(&settlement);
             return Some(settlement);
         }
-        if let Some(pending) = self.pending.get_mut(&task_id) {
+        if let Some(pending) = self.pending.get_mut(&key) {
             pending.count = pending.count.saturating_add(candidate.count);
             if pending.identity != candidate.identity {
                 pending.identity = None;
@@ -256,31 +262,34 @@ impl Observations {
             }
             pending.dry_run |= candidate.dry_run;
         } else if self.pending.len() < MAX_PENDING {
-            self.pending.insert(task_id, candidate);
+            self.pending.insert(key, candidate);
         } else {
             self.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) {
+    fn complete(&mut self, completion: Arc<ObservedHealCompletion>, now: Instant) -> Option<Settlement> {
+        let key = ObservationKey {
+            task_id: completion.task_id.clone(),
+            observation_id: completion.observation_id?,
+        };
+        if self.recent.contains_key(&key) {
             return None;
         }
-        let task_id = completion.task_id.clone();
         let mut recent = Recent {
             completion,
             counted: false,
         };
         let settled = self
             .pending
-            .remove(&task_id)
+            .remove(&key)
             .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);
+        self.recent_order.push_back((now, key.clone()));
+        self.recent.insert(key, recent);
         if let Some(settlement) = &settled {
             self.record_settlement(settlement);
         }
@@ -341,7 +350,7 @@ impl Observations {
     }
 }
 
-type AdmissionSender = mpsc::Sender<(String, Candidate)>;
+type AdmissionSender = mpsc::Sender<(ObservationKey, Candidate)>;
 
 #[derive(Clone)]
 struct ObserverHandle {
@@ -371,7 +380,7 @@ fn observer_from(registry: &Mutex<Option<ObserverHandle>>) -> ObserverHandle {
     let state = Arc::new(Mutex::new(Observations::default()));
     let handle = {
         let (sender, mut admissions) = mpsc::channel(MAX_PENDING);
-        let mut completions = subscribe_heal_completions();
+        let mut completions = subscribe_observed_heal_completions();
         let observations = state.clone();
         tokio::spawn(async move {
             let mut tick = tokio::time::interval(Duration::from_secs(5));
@@ -489,11 +498,18 @@ pub(crate) fn prepare(request: &HealChannelRequest) -> Option<Candidate> {
     Candidate::from_request(request)
 }
 
-pub(crate) fn observe(candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
+pub(crate) fn observe(
+    candidate: Option<Candidate>,
+    receipt: Result<&ObservedHealAdmissionReceipt, &HealObservationChannelError>,
+) {
     observe_on(&observer(), candidate, receipt);
 }
 
-fn observe_on(observer: &ObserverHandle, candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
+fn observe_on(
+    observer: &ObserverHandle,
+    candidate: Option<Candidate>,
+    receipt: Result<&ObservedHealAdmissionReceipt, &HealObservationChannelError>,
+) {
     match receipt {
         Ok(receipt) if receipt.result.is_admitted() => {
             let Some(candidate) = candidate else {
@@ -505,11 +521,20 @@ fn observe_on(observer: &ObserverHandle, candidate: Option<Candidate>, receipt:
                 observer.record_unknown("missing_task_id", 1);
                 return;
             }
-            if observer.sender.try_send((task_id.clone(), candidate)).is_err() {
+            let Some(observation_id) = receipt.observation_id else {
+                observer.record_unknown("owner_generation_unknown", candidate.count);
+                return;
+            };
+            let key = ObservationKey {
+                task_id: task_id.clone(),
+                observation_id,
+            };
+            if observer.sender.try_send((key, candidate)).is_err() {
                 observer.record_unknown("observer_full_or_closed", 1);
             }
         }
-        _ => observer.record_unknown("not_admitted", 1),
+        Ok(_) => observer.record_unknown("not_admitted", 1),
+        Err(_) => observer.record_unknown("admission_unknown", 1),
     }
 }
 

crates/scanner/src/heal_completion/tests.rs

@@ -1,5 +1,9 @@
 use super::*;
-use rustfs_heal_contracts::outcome::{HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectReceipt};
+use rustfs_heal_contracts::{
+    completion::{HealCompletion, publish_observed_heal_completion},
+    heal_channel::HealAdmissionReceipt,
+    outcome::{HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectReceipt},
+};
 use uuid::Uuid;
 
 fn request() -> HealChannelRequest {
@@ -19,24 +23,38 @@ 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,
+fn key(task_id: &str) -> ObservationKey {
+    ObservationKey {
+        task_id: task_id.into(),
+        observation_id: 1,
+    }
+}
+
+fn completion(task: &str, disposition: HealObjectDisposition) -> Arc<ObservedHealCompletion> {
+    Arc::new(ObservedHealCompletion {
+        observation_id: Some(1),
+        completion: 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,
+            }),
         }),
     })
 }
 
+fn publish_terminal(observed: Arc<ObservedHealCompletion>) {
+    publish_observed_heal_completion((*observed.completion).clone(), observed.observation_id);
+}
+
 #[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!(observations.admit(key("owner"), candidate()).is_none());
+    assert!(observations.admit(key("owner"), candidate()).is_none());
     assert_eq!(observations.pending.len(), 1);
     assert!(
         observations
@@ -57,7 +75,7 @@ fn merged_candidates_close_only_on_the_canonical_task_once() {
         "duplicate completion is ignored"
     );
     let merged_late = observations
-        .admit("owner".into(), candidate())
+        .admit(key("owner"), candidate())
         .expect("terminal before merged admission is retained");
     assert!(!merged_late.first_task_observation);
     assert_eq!(merged_late.candidates, 1);
@@ -71,14 +89,149 @@ fn terminal_before_admission_is_observed_without_claiming_coverage() {
             .complete(completion("fast", HealObjectDisposition::Repaired), Instant::now())
             .is_none()
     );
-    let settled = observations
-        .admit("fast".into(), candidate())
-        .expect("fast terminal retained");
+    let settled = observations.admit(key("fast"), 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 status_and_pending_metric_count_each_owner_generation() {
+    #[derive(Default)]
+    struct GaugeValue(Mutex<f64>);
+
+    impl metrics::GaugeFn for GaugeValue {
+        fn increment(&self, value: f64) {
+            *self.0.lock().expect("pending gauge") += value;
+        }
+        fn decrement(&self, value: f64) {
+            *self.0.lock().expect("pending gauge") -= value;
+        }
+        fn set(&self, value: f64) {
+            *self.0.lock().expect("pending gauge") = value;
+        }
+    }
+
+    #[derive(Default)]
+    struct PendingRecorder(Arc<GaugeValue>);
+
+    impl metrics::Recorder for PendingRecorder {
+        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, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+            metrics::Counter::noop()
+        }
+        fn register_gauge(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+            if key.name() == "rustfs_scanner_heal_pending_tasks" {
+                metrics::Gauge::from_arc(Arc::clone(&self.0))
+            } else {
+                metrics::Gauge::noop()
+            }
+        }
+        fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+            metrics::Histogram::noop()
+        }
+    }
+
+    let recorder = PendingRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let (sender, _receiver) = mpsc::channel(1);
+    let observer = ObserverHandle {
+        sender,
+        state: Arc::new(Mutex::new(Observations::default())),
+        generation: 1,
+        started_at_unix_secs: None,
+    };
+    let registry = Mutex::new(Some(observer.clone()));
+    let first = key("same-task");
+    let second = ObservationKey {
+        task_id: "same-task".into(),
+        observation_id: 2,
+    };
+    {
+        let mut observations = observer.state.lock().expect("observations");
+        assert!(observations.admit(first.clone(), candidate()).is_none());
+        assert!(observations.admit(first, candidate()).is_none());
+        assert!(observations.admit(second, candidate()).is_none());
+        observations.maintain(Instant::now());
+    }
+    let status = observation_status_from(&registry);
+    assert_eq!(status.retained_pending_tasks, Some(2));
+    assert_eq!(status.retained_pending_candidates, Some(3));
+    assert_eq!(*recorder.0.0.lock().expect("pending gauge"), 2.0);
+    let json = serde_json::to_value(&status).expect("serialize observation status");
+    assert_eq!(json["retained_pending_tasks"], 2);
+    assert_eq!(json["retained_pending_candidates"], 3);
+
+    {
+        let mut observations = observer.state.lock().expect("observations");
+        let settlement = observations
+            .complete(completion("same-task", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .expect("the first generation settles its two candidates");
+        assert_eq!(settlement.candidates, 2);
+        observations.maintain(Instant::now());
+    }
+    let status = observation_status_from(&registry);
+    assert_eq!(status.retained_pending_tasks, Some(1));
+    assert_eq!(status.retained_pending_candidates, Some(1));
+    assert_eq!(*recorder.0.0.lock().expect("pending gauge"), 1.0);
+}
+
+#[test]
+fn reused_task_id_keeps_terminal_observations_on_their_owner_generation() {
+    let mut observations = Observations::default();
+    let old_key = key("owner");
+    assert!(observations.admit(old_key.clone(), candidate()).is_none());
+    assert!(
+        observations
+            .complete(completion("owner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_some()
+    );
+
+    let new_key = ObservationKey {
+        task_id: "owner".into(),
+        observation_id: 2,
+    };
+    assert!(
+        observations.admit(new_key.clone(), candidate()).is_none(),
+        "the old terminal cannot settle a new owner"
+    );
+    let mut new_terminal = (*completion("owner", HealObjectDisposition::Repaired)).clone();
+    new_terminal.observation_id = Some(2);
+    let settled = observations
+        .complete(Arc::new(new_terminal), Instant::now())
+        .expect("the new owner settles itself");
+    assert_eq!(settled.disposition, "repaired");
+    assert!(observations.pending.is_empty());
+
+    let fast_key = ObservationKey {
+        task_id: "owner".into(),
+        observation_id: 3,
+    };
+    let mut fast_terminal = (*completion("owner", HealObjectDisposition::MetadataHealthy)).clone();
+    fast_terminal.observation_id = Some(3);
+    assert!(observations.complete(Arc::new(fast_terminal), Instant::now()).is_none());
+    assert!(
+        observations.admit(fast_key, candidate()).is_some(),
+        "a fast terminal stays available to its owner"
+    );
+}
+
+#[test]
+fn removed_owner_remains_pending_then_expires_unknown_after_id_reuse() {
+    let mut observations = Observations::default();
+    let old = key("removed-owner");
+    assert!(observations.admit(old.clone(), candidate()).is_none());
+    let mut new_terminal = (*completion("removed-owner", HealObjectDisposition::VerifiedHealthy)).clone();
+    new_terminal.observation_id = Some(2);
+    assert!(observations.complete(Arc::new(new_terminal), Instant::now()).is_none());
+    assert!(observations.pending.contains_key(&old));
+    observations.maintain(Instant::now() + PENDING_TTL);
+    assert!(!observations.pending.contains_key(&old));
+    assert_eq!(observations.unknown_counts.get("expired"), Some(&1));
+}
+
 #[test]
 fn replacement_bucket_version_or_location_cannot_match_an_old_result() {
     for change in 0..4 {
@@ -103,7 +256,12 @@ 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;
+    Arc::make_mut(&mut terminal.completion)
+        .object
+        .as_mut()
+        .expect("receipt")
+        .identity
+        .version_id = None;
     let mut recent = Recent {
         completion: Arc::new(terminal),
         counted: false,
@@ -120,7 +278,7 @@ fn unversioned_overwrite_and_unknown_leader_never_certify_payload() {
 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;
+        Arc::make_mut(&mut terminal.completion).scan_mode = HealScanMode::Normal;
         let mut recent = Recent {
             completion: Arc::new(terminal),
             counted: false,
@@ -153,7 +311,7 @@ fn command_success_does_not_hide_deferred_failed_cancelled_or_dry_run_objects()
         assert_eq!(settled.scope, "content_and_leader_unknown");
     }
     let mut terminal = (*completion("cancelled", HealObjectDisposition::VerifiedHealthy)).clone();
-    terminal.execution = Some(HealExecutionOutcome::Aborted(HealAbortReason::Cancelled));
+    Arc::make_mut(&mut terminal.completion).execution = Some(HealExecutionOutcome::Aborted(HealAbortReason::Cancelled));
     let mut recent = Recent {
         completion: Arc::new(terminal),
         counted: false,
@@ -161,7 +319,7 @@ fn command_success_does_not_hide_deferred_failed_cancelled_or_dry_run_objects()
     assert_eq!(Settlement::new(candidate(), &mut recent).scope, "execution_incomplete");
     assert_eq!(Settlement::new(candidate(), &mut recent).execution, "aborted_cancelled");
     let mut errors = (*completion("errors", HealObjectDisposition::Repaired)).clone();
-    errors.execution = Some(HealExecutionOutcome::CompletedWithErrors);
+    Arc::make_mut(&mut errors.completion).execution = Some(HealExecutionOutcome::CompletedWithErrors);
     let mut recent = Recent {
         completion: Arc::new(errors),
         counted: false,
@@ -174,7 +332,7 @@ fn command_success_does_not_hide_deferred_failed_cancelled_or_dry_run_objects()
 #[test]
 fn missing_or_truncated_object_result_remains_unknown() {
     let mut terminal = (*completion("truncated", HealObjectDisposition::VerifiedHealthy)).clone();
-    terminal.object = None;
+    Arc::make_mut(&mut terminal.completion).object = None;
     let mut recent = Recent {
         completion: Arc::new(terminal),
         counted: false,
@@ -186,10 +344,10 @@ fn missing_or_truncated_object_result_remains_unknown() {
 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());
+        observations.admit(key(&format!("pending-{index}")), candidate());
     }
     assert_eq!(observations.pending.len(), MAX_PENDING);
-    assert!(!observations.pending.contains_key(&format!("pending-{MAX_PENDING}")));
+    assert!(!observations.pending.contains_key(&key(&format!("pending-{MAX_PENDING}"))));
     observations.lost();
     assert!(observations.pending.is_empty());
     assert!(
@@ -197,7 +355,7 @@ fn overflow_loss_expiry_and_restart_are_bounded_and_do_not_restore_pending() {
             .complete(completion("pending-0", HealObjectDisposition::VerifiedHealthy), Instant::now())
             .is_none()
     );
-    observations.admit("expires".into(), candidate());
+    observations.admit(key("expires"), candidate());
     observations.maintain(Instant::now() + PENDING_TTL);
     assert!(observations.pending.is_empty());
     assert!(observations.recent.is_empty());
@@ -220,8 +378,8 @@ fn terminal_history_eviction_loses_evidence_without_inventing_a_completion() {
     }
     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"));
+    assert!(observations.admit(key("terminal-0"), candidate()).is_none());
+    assert!(observations.pending.contains_key(&key("terminal-0")));
 }
 
 #[test]
@@ -258,7 +416,7 @@ fn observer_reinstalls_after_runtime_teardown() {
         assert!(observation_status_from(&registry).unknown_candidates.is_empty());
         replacement
             .sender
-            .try_send(("new-runtime".into(), candidate()))
+            .try_send((key("new-runtime"), candidate()))
             .expect("replacement observer admits work");
     });
 }
@@ -294,10 +452,10 @@ fn observer_status_preserves_loss_expiry_and_rejected_admissions() {
     {
         let mut state = observer.state.lock().expect("observations");
         for index in 0..=MAX_PENDING {
-            let _ = state.admit(format!("pending-{index}"), candidate());
+            let _ = state.admit(key(&format!("pending-{index}")), candidate());
         }
         state.lagged(3);
-        let _ = state.admit("expired".into(), candidate());
+        let _ = state.admit(key("expired"), candidate());
         state.maintain(Instant::now() + PENDING_TTL);
     }
     let status = observation_status_from(&registry);
@@ -313,20 +471,37 @@ fn observer_status_preserves_loss_expiry_and_rejected_admissions() {
     assert_eq!(unknown["completion_lost"], u64::try_from(MAX_PENDING).expect("bounded pending count"));
     assert_eq!(unknown["expired"], 1);
 
-    let admitted = HealAdmissionReceipt {
-        result: HealAdmissionResult::Accepted,
-        task_id: "canonical".into(),
+    let admitted = ObservedHealAdmissionReceipt {
+        receipt: HealAdmissionReceipt {
+            result: HealAdmissionResult::Accepted,
+            task_id: "canonical".into(),
+        },
+        observation_id: Some(1),
     };
+    observe_on(
+        &observer,
+        Some(candidate()),
+        Ok(&ObservedHealAdmissionReceipt {
+            receipt: HealAdmissionReceipt {
+                result: HealAdmissionResult::Merged,
+                task_id: "unproven-owner".into(),
+            },
+            observation_id: None,
+        }),
+    );
     observe_on(&observer, None, Ok(&admitted));
     observe_on(
         &observer,
         Some(candidate()),
-        Ok(&HealAdmissionReceipt {
-            result: HealAdmissionResult::Accepted,
-            task_id: String::new(),
+        Ok(&ObservedHealAdmissionReceipt {
+            receipt: HealAdmissionReceipt {
+                result: HealAdmissionResult::Accepted,
+                task_id: String::new(),
+            },
+            observation_id: Some(1),
         }),
     );
-    observe_on(&observer, Some(candidate()), Err(()));
+    observe_on(&observer, Some(candidate()), Err(&HealObservationChannelError::NotInitialized));
     observe_on(&observer, Some(candidate()), Ok(&admitted));
     assert_eq!(observation_status_from(&registry).queued_admissions, Some(1));
     observe_on(&observer, Some(candidate()), Ok(&admitted));
@@ -341,14 +516,98 @@ fn observer_status_preserves_loss_expiry_and_rejected_admissions() {
         .map(|row| (row.reason, row.count))
         .collect::<BTreeMap<_, _>>();
     assert_eq!(unknown["identity_too_large"], 1);
+    assert_eq!(unknown["owner_generation_unknown"], 1);
     assert_eq!(unknown["missing_task_id"], 1);
-    assert_eq!(unknown["not_admitted"], 1);
+    assert_eq!(unknown["admission_unknown"], 1);
     assert_eq!(unknown["observer_full_or_closed"], 2);
 }
 
+#[test]
+fn explicit_rejections_are_reported_not_admitted() {
+    use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
+
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let (sender, _receiver) = mpsc::channel(1);
+    let observer = ObserverHandle {
+        sender,
+        state: Arc::new(Mutex::new(Observations::default())),
+        generation: 1,
+        started_at_unix_secs: None,
+    };
+    let registry = Mutex::new(Some(observer.clone()));
+    for result in [
+        HealAdmissionResult::Full,
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull),
+    ] {
+        let receipt = ObservedHealAdmissionReceipt {
+            receipt: HealAdmissionReceipt {
+                result,
+                task_id: String::new(),
+            },
+            observation_id: None,
+        };
+        observe_on(&observer, Some(candidate()), Ok(&receipt));
+    }
+    let status = observation_status_from(&registry);
+    assert_eq!(status.integrity_state, "unknown");
+    assert_eq!(status.retained_pending_candidates, Some(0));
+    assert_eq!(status.unknown_candidates.iter().map(|row| row.count).sum::<u64>(), 2);
+    assert!(
+        status
+            .unknown_candidates
+            .iter()
+            .any(|row| row.reason == "not_admitted" && row.count == 2)
+    );
+    assert_eq!(recorder.reason_total("not_admitted"), 2);
+    assert_eq!(recorder.reason_total("admission_unknown"), 0);
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_unknown_total"), 2);
+}
+
+#[tokio::test]
+async fn admission_errors_without_decisions_are_reported_unknown() {
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let (sender, _receiver) = mpsc::channel(1);
+    let observer = ObserverHandle {
+        sender,
+        state: Arc::new(Mutex::new(Observations::default())),
+        generation: 1,
+        started_at_unix_secs: None,
+    };
+    let registry = Mutex::new(Some(observer.clone()));
+    let (sender, observed_reply) = tokio::sync::oneshot::channel::<()>();
+    drop(sender);
+    let (sender, legacy_reply) = tokio::sync::oneshot::channel::<()>();
+    drop(sender);
+    for error in [
+        HealObservationChannelError::NotInitialized,
+        HealObservationChannelError::LegacyCommandClosed,
+        HealObservationChannelError::ObservedResponseClosed(observed_reply.await.expect_err("reply was dropped")),
+        HealObservationChannelError::LegacyResponseClosed(legacy_reply.await.expect_err("reply was dropped")),
+        HealObservationChannelError::AdmissionFailed("unconfirmed processor error".into()),
+    ] {
+        observe_on(&observer, Some(candidate()), Err(&error));
+    }
+
+    let status = observation_status_from(&registry);
+    assert_eq!(status.integrity_state, "unknown");
+    assert_eq!(status.retained_pending_candidates, Some(0));
+    assert_eq!(status.unknown_candidates.iter().map(|row| row.count).sum::<u64>(), 5);
+    assert!(
+        status
+            .unknown_candidates
+            .iter()
+            .any(|row| row.reason == "admission_unknown" && row.count == 5)
+    );
+    assert_eq!(recorder.reason_total("admission_unknown"), 5);
+    assert_eq!(recorder.reason_total("not_admitted"), 0);
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_unknown_total"), 5);
+}
+
 #[tokio::test]
 async fn observer_status_tracks_canonical_terminals_without_a_metrics_recorder() {
-    use rustfs_heal_contracts::{completion::publish_heal_completion, heal_channel::HealAdmissionResult};
+    use rustfs_heal_contracts::heal_channel::HealAdmissionResult;
 
     let recorder = metrics::NoopRecorder;
     let _recorder = metrics::set_default_local_recorder(&recorder);
@@ -361,19 +620,23 @@ async fn observer_status_tracks_canonical_terminals_without_a_metrics_recorder()
         (canonical.clone(), HealAdmissionResult::Merged),
         (deferred.clone(), HealAdmissionResult::Accepted),
     ] {
-        observe_on(&observer, Some(candidate()), Ok(&HealAdmissionReceipt { result, task_id }));
+        observe_on(
+            &observer,
+            Some(candidate()),
+            Ok(&ObservedHealAdmissionReceipt {
+                receipt: HealAdmissionReceipt { result, task_id },
+                observation_id: Some(1),
+            }),
+        );
     }
-    publish_heal_completion((*completion(&canonical, HealObjectDisposition::VerifiedHealthy)).clone());
-    publish_heal_completion(
-        (*completion(
-            &deferred,
-            HealObjectDisposition::Deferred {
-                reason: HealDeferredReason::Deadline,
-                retry_not_before: None,
-            },
-        ))
-        .clone(),
-    );
+    publish_terminal(completion(&canonical, HealObjectDisposition::VerifiedHealthy));
+    publish_terminal(completion(
+        &deferred,
+        HealObjectDisposition::Deferred {
+            reason: HealDeferredReason::Deadline,
+            retry_not_before: None,
+        },
+    ));
     tokio::time::timeout(Duration::from_secs(2), async {
         loop {
             let status = observation_status_from(&registry);
@@ -447,6 +710,19 @@ impl CompletionRecorder {
             .sum()
     }
 
+    fn reason_total(&self, reason: &str) -> u64 {
+        self.0
+            .lock()
+            .expect("completion recorder")
+            .iter()
+            .filter(|(key, _)| {
+                key.name() == "rustfs_scanner_heal_candidate_unknown_total"
+                    && key.labels().any(|label| label.key() == "reason" && label.value() == reason)
+            })
+            .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 {
@@ -462,7 +738,7 @@ impl CompletionRecorder {
 // 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};
+    use rustfs_heal_contracts::heal_channel::HealAdmissionResult;
 
     let recorder = CompletionRecorder::default();
     let _recorder = metrics::set_default_local_recorder(&recorder);
@@ -476,20 +752,26 @@ async fn live_observer_associates_early_completion_with_canonical_merged_receipt
     observe_on(
         &sender,
         Some(candidate()),
-        Ok(&HealAdmissionReceipt {
-            result: HealAdmissionResult::Accepted,
-            task_id: sentinel_id.clone(),
+        Ok(&ObservedHealAdmissionReceipt {
+            receipt: HealAdmissionReceipt {
+                result: HealAdmissionResult::Accepted,
+                task_id: sentinel_id.clone(),
+            },
+            observation_id: Some(1),
         }),
     );
-    publish_heal_completion((*completion(&canonical_id, HealObjectDisposition::VerifiedHealthy)).clone());
+    publish_terminal(completion(&canonical_id, HealObjectDisposition::VerifiedHealthy));
     // 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());
+    publish_terminal(completion(&sentinel_id, HealObjectDisposition::MetadataHealthy));
     recorder.wait_for_terminals(1).await;
 
-    let merged = HealAdmissionReceipt {
-        result: HealAdmissionResult::Merged,
-        task_id: canonical_id,
+    let merged = ObservedHealAdmissionReceipt {
+        receipt: HealAdmissionReceipt {
+            result: HealAdmissionResult::Merged,
+            task_id: canonical_id,
+        },
+        observation_id: Some(1),
     };
     observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged));
     recorder.wait_for_terminals(2).await;

crates/scanner/src/scanner_folder.rs

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

crates/scanner/src/scanner_folder/item_actions.rs

@@ -481,7 +481,7 @@ pub(super) async fn send_scanner_heal_request(
     let priority = request.priority;
     let trace_context = scanner_heal_candidate_trace_context(&request);
     let candidate = crate::heal_completion::prepare(&request);
-    match send_heal_request_with_receipt(request).await {
+    match send_heal_request_with_observation(request).await {
         Ok(receipt) => {
             crate::heal_completion::observe(candidate, Ok(&receipt));
             let result = receipt.result;
@@ -501,7 +501,8 @@ pub(super) async fn send_scanner_heal_request(
             Ok(result)
         }
         Err(err) => {
-            crate::heal_completion::observe(candidate, Err(()));
+            crate::heal_completion::observe(candidate, Err(&err));
+            let err = err.to_string();
             counter!(
                 "rustfs_heal_candidate_enqueue_total",
                 "type" => candidate_type.to_string(),

docs/operations/scanner-runtime-controls.md

@@ -623,7 +623,7 @@ typed terminal result:
 
 | Metric | Meaning |
 |---|---|
-| `rustfs_scanner_heal_pending_tasks` | Canonical tasks awaiting a terminal observation. |
+| `rustfs_scanner_heal_pending_tasks` | Pending task ID and owner-generation pairs 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. |
@@ -640,16 +640,25 @@ verified-byte or full-bucket coverage counter is advanced here.
 The scanner status response also exposes `heal_observation` without requiring
 a metrics exporter. Reading status does not start an observer. `state` is
 `not_started`, `active`, or `stopped`; unavailable pending counts are null.
-`retained_pending_tasks` counts canonical IDs in the existing bounded map,
-while `retained_pending_candidates` counts their associated discoveries.
+`retained_pending_tasks` counts distinct canonical task ID and process-local
+owner-generation pairs in the bounded map. Different generations of the same
+task ID count separately. `retained_pending_candidates` counts their associated
+discoveries. The JSON field names are unchanged. The pending-tasks metric uses
+the same pair count when periodically refreshed, so it can lag the status.
 `queued_admissions` samples the local admission channel and is not an atomic
 cluster backlog. `terminal_task_observations_total` keeps the same bounded
 deduplication semantics as the exporter counter.
 
 `terminal_candidates` preserves execution, object disposition, scan mode, and
 unresolved scope. `unknown_candidates` includes admission rejection, overflow,
-loss, expiry, and missing terminal proof. These categories overlap: a completed
-task with a repaired or verified-healthy object can still have unknown scanner
+loss, expiry, and missing terminal proof. The `not_admitted` reason covers an
+explicit `Full` or `Dropped` decision. The `admission_unknown` reason means no
+decision was received because of a channel or processor error. A lost reply
+does not establish whether work was admitted; a failure known to occur before
+enqueue separately rules out submission. Both reasons remain part of the
+unknown total. Monitoring that filters by reason must include the new
+`admission_unknown` label to retain those errors. These categories overlap: a
+completed task with a repaired or verified-healthy object can still have unknown scanner
 integrity. `integrity_state` therefore remains `unknown`, including when retained
 pending is zero. `lost_completion_notifications` counts skipped messages, not
 objects. Counts belong to the named local observer `generation`; runtime