rnode: request exact driver shutdown

crates/rns-interface/src/android_usb.rs

@@ -16,7 +16,9 @@ use crate::android_usb_lifecycle::{
 };
 #[cfg(test)]
 use crate::kiss;
-use crate::rnode::{self, RNodeRuntimeReason, RNodeTransportClass, SpawnedRNodeInterface};
+use crate::rnode::{
+    self, RNodeDriverShutdown, RNodeRuntimeReason, RNodeTransportClass, SpawnedRNodeInterface,
+};
 use crate::rnode_protocol::RNodeProtocolTarget;
 use crate::traits::{
     InterfaceDirection, InterfaceError, InterfaceHandle, InterfaceId, InterfaceMode,
@@ -66,6 +68,7 @@ fn android_usb_rnode_stop_registry() -> &'static AndroidUsbRNodeStopRegistry {
 
 struct AndroidUsbRNodeStopRegistryGuard {
     id: InterfaceId,
+    stop_tx: mpsc::Sender<()>,
     status: watch::Receiver<AndroidUsbShutdownStatus>,
 }
 
@@ -79,10 +82,15 @@ impl Drop for AndroidUsbRNodeStopRegistryGuard {
             // quarantine separately blocks a competing reopen.
             return;
         }
-        android_usb_rnode_stop_registry()
+        let mut registry = android_usb_rnode_stop_registry()
             .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .remove(&self.id);
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let owns_entry = registry
+            .get(&self.id)
+            .is_some_and(|registered| registered.stop_tx.same_channel(&self.stop_tx));
+        if owns_entry {
+            registry.remove(&self.id);
+        }
     }
 }
 
@@ -95,9 +103,16 @@ fn register_android_usb_rnode_stop(
     android_usb_rnode_stop_registry()
         .lock()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
-        .insert(id, AndroidUsbStopHandle { stop_tx, status });
+        .insert(
+            id,
+            AndroidUsbStopHandle {
+                stop_tx: stop_tx.clone(),
+                status,
+            },
+        );
     AndroidUsbRNodeStopRegistryGuard {
         id,
+        stop_tx,
         status: guard_status,
     }
 }
@@ -124,8 +139,12 @@ fn request_android_usb_rnode_stop(id: InterfaceId) -> Option<AndroidUsbStopHandl
     Some(stop_handle)
 }
 
-/// Compatibility stop request. The runtime uses the acknowledged async form
-/// below before deregistering/aborting the outer driver task.
+/// Compatibility stop request for the currently registered Android USB RNode.
+///
+/// New owners should retain [`crate::rnode::RNodeDriverHandle`] and call
+/// [`crate::rnode::RNodeDriverHandle::request_shutdown`] so later ID reuse
+/// cannot redirect the request. The acknowledged compatibility form below
+/// remains available to existing runtime callers.
 pub fn stop_android_usb_rnode_interface(id: InterfaceId) {
     let _ = request_android_usb_rnode_stop(id);
 }
@@ -1154,15 +1173,16 @@ pub async fn spawn_android_usb_rnode_interface_with_driver(
         )));
     }
 
-    let (mut snapshot_publisher, driver) =
-        rnode::new_rnode_driver_observation(RNodeTransportClass::Usb);
-    snapshot_publisher.connection_established();
-
     let shared_txb = Arc::new(AtomicU64::new(0));
     let shared_rxb = Arc::new(AtomicU64::new(0));
 
     let (tx, app_rx) = mpsc::channel::<Bytes>(64);
     let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
+    let (mut snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown(
+        RNodeTransportClass::Usb,
+        RNodeDriverShutdown::from_stop_sender(stop_tx.clone()),
+    );
+    snapshot_publisher.connection_established();
     let (shutdown_status_tx, shutdown_status_rx) =
         watch::channel(AndroidUsbShutdownStatus::Running);
     let stop_guard = register_android_usb_rnode_stop(id, stop_tx, shutdown_status_rx);
@@ -1427,6 +1447,47 @@ mod tests {
         let _usb_transport: RNodeTransportClass = RNodeTransportClass::Usb;
     }
 
+    #[test]
+    fn android_usb_exact_shutdown_and_cleanup_resist_same_id_aba() {
+        let id: InterfaceId = 0xA11D_0001;
+        let (old_tx, mut old_rx) = mpsc::channel::<()>(2);
+        let (_old_status_tx, old_status_rx) = watch::channel(AndroidUsbShutdownStatus::Running);
+        let (_old_publisher, old_driver) = rnode::new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Usb,
+            RNodeDriverShutdown::from_stop_sender(old_tx.clone()),
+        );
+        let old_guard = register_android_usb_rnode_stop(id, old_tx, old_status_rx);
+
+        let (new_tx, mut new_rx) = mpsc::channel::<()>(2);
+        let (_new_status_tx, new_status_rx) = watch::channel(AndroidUsbShutdownStatus::Running);
+        let (_new_publisher, _new_driver) = rnode::new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Usb,
+            RNodeDriverShutdown::from_stop_sender(new_tx.clone()),
+        );
+        let new_guard = register_android_usb_rnode_stop(id, new_tx, new_status_rx);
+
+        drop(old_guard);
+        old_driver.request_shutdown();
+        assert!(old_rx.try_recv().is_ok());
+        assert!(
+            matches!(new_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+            "the retired handle must not stop the newer same-ID USB driver"
+        );
+
+        stop_android_usb_rnode_interface(id);
+        assert!(
+            new_rx.try_recv().is_ok(),
+            "retired guard cleanup must preserve the newer compatibility entry"
+        );
+        drop(new_guard);
+        assert!(
+            !android_usb_rnode_stop_registry()
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .contains_key(&id)
+        );
+    }
+
     fn fragmented_decode(wire: &[u8]) -> Vec<(u8, Vec<u8>)> {
         let mut deframer = kiss::RawKissDeframer::new();
         let fragment_widths = [1, 2, 1, 3, 2, 4];

crates/rns-interface/src/ble_rnode.rs

@@ -29,8 +29,8 @@ pub fn is_btleplug_initialized() -> bool {
 
 use crate::kiss;
 use crate::rnode::{
-    self, RNodeResponse, RNodeRuntimeReason, RNodeSnapshotPublisher, RNodeTransportClass,
-    SpawnedRNodeInterface,
+    self, RNodeDriverShutdown, RNodeResponse, RNodeRuntimeReason, RNodeSnapshotPublisher,
+    RNodeTransportClass, SpawnedRNodeInterface,
 };
 use crate::rnode_protocol::{RNodeProtocolState, RNodeProtocolTarget};
 use crate::traits::{
@@ -76,14 +76,23 @@ fn register_running(id: InterfaceId) -> Arc<AtomicBool> {
     flag
 }
 
-fn unregister_running(id: InterfaceId) {
+fn unregister_running(id: InterfaceId, running: &Arc<AtomicBool>) {
     if let Ok(mut map) = running_map().lock() {
-        map.remove(&id);
+        let owns_entry = map
+            .get(&id)
+            .is_some_and(|registered| Arc::ptr_eq(registered, running));
+        if owns_entry {
+            map.remove(&id);
+        }
     }
 }
 
-/// Idempotent. Safe to call before or after deregistering from the
-/// transport actor.
+/// Compatibility facade requesting shutdown of the currently registered BLE
+/// RNode for `id`.
+///
+/// New owners should retain [`crate::rnode::RNodeDriverHandle`] and call
+/// [`crate::rnode::RNodeDriverHandle::request_shutdown`] so later ID reuse
+/// cannot redirect the request.
 pub fn stop_ble_rnode_interface(id: InterfaceId) {
     if let Ok(map) = running_map().lock() {
         if let Some(flag) = map.get(&id) {
@@ -1300,8 +1309,6 @@ pub async fn spawn_ble_rnode_interface_with_driver(
         config.coding_rate,
         config.tx_power,
     );
-    let (snapshot_publisher, driver) =
-        rnode::new_rnode_driver_observation(RNodeTransportClass::Ble);
     let online = Arc::new(AtomicBool::new(false));
     let online_handle = online.clone();
     let shared_rxb = Arc::new(AtomicU64::new(0));
@@ -1311,6 +1318,10 @@ pub async fn spawn_ble_rnode_interface_with_driver(
     let (tx, rx) = mpsc::channel::<Bytes>(256);
     let rx = Arc::new(tokio::sync::Mutex::new(rx));
     let running = register_running(id);
+    let (snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown(
+        RNodeTransportClass::Ble,
+        RNodeDriverShutdown::from_running_flag(running.clone()),
+    );
 
     let bitrate = rnode::calculate_bitrate(
         config.spreading_factor,
@@ -1336,13 +1347,13 @@ pub async fn spawn_ble_rnode_interface_with_driver(
 
         // Drop guard: every early return must clear the running-flag map
         // entry, or stale entries confuse later spawns reusing the id.
-        struct Cleanup(InterfaceId);
+        struct Cleanup(InterfaceId, Arc<AtomicBool>);
         impl Drop for Cleanup {
             fn drop(&mut self) {
-                unregister_running(self.0);
+                unregister_running(self.0, &self.1);
             }
         }
-        let _cleanup = Cleanup(id);
+        let _cleanup = Cleanup(id, running_task.clone());
 
         loop {
             if !running_task.load(Ordering::SeqCst) {
@@ -1763,8 +1774,6 @@ pub async fn spawn_ble_rnode_interface_native_with_driver(
         config.coding_rate,
         config.tx_power,
     );
-    let (snapshot_publisher, driver) =
-        rnode::new_rnode_driver_observation(RNodeTransportClass::Ble);
     let online = Arc::new(AtomicBool::new(false));
     let online_handle = online.clone();
     let shared_rxb = Arc::new(AtomicU64::new(0));
@@ -1789,6 +1798,10 @@ pub async fn spawn_ble_rnode_interface_native_with_driver(
     let ble_uri = config.ble_uri.clone();
     let log_name = name.clone();
     let running = register_running(id);
+    let (snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown(
+        RNodeTransportClass::Ble,
+        RNodeDriverShutdown::from_running_flag(running.clone()),
+    );
     let running_task = running.clone();
 
     let read_task = tokio::spawn(async move {
@@ -1797,13 +1810,13 @@ pub async fn spawn_ble_rnode_interface_native_with_driver(
         let mut backoff = RECONNECT_WAIT;
         let mut initial_attempt = true;
 
-        struct Cleanup(InterfaceId);
+        struct Cleanup(InterfaceId, Arc<AtomicBool>);
         impl Drop for Cleanup {
             fn drop(&mut self) {
-                unregister_running(self.0);
+                unregister_running(self.0, &self.1);
             }
         }
-        let _cleanup = Cleanup(id);
+        let _cleanup = Cleanup(id, running_task.clone());
 
         loop {
             if !running_task.load(Ordering::SeqCst) {
@@ -2218,7 +2231,7 @@ mod tests {
         let flag = register_running(id);
         assert!(is_registered(id));
         assert!(flag.load(Ordering::SeqCst));
-        unregister_running(id);
+        unregister_running(id, &flag);
         assert!(!is_registered(id));
     }
 
@@ -2230,7 +2243,7 @@ mod tests {
         stop_ble_rnode_interface(id);
         assert!(!flag.load(Ordering::SeqCst));
         // Map entry survives until the owning task's Drop runs; clean up.
-        unregister_running(id);
+        unregister_running(id, &flag);
     }
 
     #[test]
@@ -2241,6 +2254,41 @@ mod tests {
         assert!(!is_registered(id));
     }
 
+    #[test]
+    fn test_ble_exact_shutdown_and_cleanup_resist_same_id_aba() {
+        let id: InterfaceId = 0xDEAD_BEEF_0000_0004;
+        let old_running = register_running(id);
+        let (_old_publisher, old_driver) = rnode::new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Ble,
+            RNodeDriverShutdown::from_running_flag(old_running.clone()),
+        );
+
+        let new_running = register_running(id);
+        let (_new_publisher, _new_driver) = rnode::new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Ble,
+            RNodeDriverShutdown::from_running_flag(new_running.clone()),
+        );
+
+        unregister_running(id, &old_running);
+        assert!(
+            is_registered(id),
+            "retired task cleanup must preserve the newer compatibility entry"
+        );
+        old_driver.request_shutdown();
+        assert!(!old_running.load(Ordering::SeqCst));
+        assert!(
+            new_running.load(Ordering::SeqCst),
+            "the retired handle must not stop the newer same-ID BLE driver"
+        );
+
+        stop_ble_rnode_interface(id);
+        assert!(
+            !new_running.load(Ordering::SeqCst),
+            "the compatibility facade must still stop the current registration"
+        );
+        unregister_running(id, &new_running);
+    }
+
     #[tokio::test]
     async fn test_wait_or_shutdown_returns_false_when_flag_stays_set() {
         // Short real-time wait; flag stays true → full duration elapses.

crates/rns-interface/src/rnode.rs

@@ -30,14 +30,17 @@ use crate::{rnode_protocol::RNodeProtocolTarget, traits::InterfaceDirection};
 use std::collections::HashMap;
 use std::sync::Arc;
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::{AtomicBool, Ordering};
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
 use std::sync::{Mutex, OnceLock};
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
 use std::time::Duration;
-use tokio::sync::watch;
+#[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))]
+use tokio::sync::mpsc;
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
+use tokio::sync::watch;
 
 pub const CMD_FREQUENCY: u8 = 0x01;
 pub const CMD_BANDWIDTH: u8 = 0x02;
@@ -425,10 +428,93 @@ fn project_rnode_protocol_effect(
     *snapshot != before
 }
 
-/// Cloneable, observation-only handle for a generic RNode driver.
+enum RNodeDriverShutdownSignal {
+    #[cfg(test)]
+    InertTest,
+    #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))]
+    StopSender(mpsc::Sender<()>),
+    #[cfg(feature = "ble")]
+    RunningFlag(Arc<AtomicBool>),
+}
+
+struct RNodeDriverShutdownInner {
+    requested: AtomicBool,
+    signal: RNodeDriverShutdownSignal,
+}
+
+/// Exact-instance shutdown request shared by every clone of a driver handle.
+///
+/// This stays crate-private so observed drivers can expose only the one
+/// bounded lifecycle action, never arbitrary RNode controls.
+#[derive(Clone)]
+pub(crate) struct RNodeDriverShutdown {
+    inner: Arc<RNodeDriverShutdownInner>,
+}
+
+impl RNodeDriverShutdown {
+    #[cfg(test)]
+    fn inert_test() -> Self {
+        Self::new(RNodeDriverShutdownSignal::InertTest)
+    }
+
+    #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))]
+    pub(crate) fn from_stop_sender(stop_tx: mpsc::Sender<()>) -> Self {
+        Self::new(RNodeDriverShutdownSignal::StopSender(stop_tx))
+    }
+
+    #[cfg(feature = "ble")]
+    pub(crate) fn from_running_flag(running: Arc<AtomicBool>) -> Self {
+        Self::new(RNodeDriverShutdownSignal::RunningFlag(running))
+    }
+
+    #[cfg(any(
+        feature = "serial",
+        feature = "rnode-tcp",
+        feature = "ble",
+        target_os = "android",
+        test
+    ))]
+    fn new(signal: RNodeDriverShutdownSignal) -> Self {
+        Self {
+            inner: Arc::new(RNodeDriverShutdownInner {
+                requested: AtomicBool::new(false),
+                signal,
+            }),
+        }
+    }
+
+    fn request(&self) {
+        if self.inner.requested.swap(true, Ordering::SeqCst) {
+            return;
+        }
+        match &self.inner.signal {
+            #[cfg(test)]
+            RNodeDriverShutdownSignal::InertTest => {}
+            #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))]
+            RNodeDriverShutdownSignal::StopSender(stop_tx) => {
+                let _ = stop_tx.try_send(());
+            }
+            #[cfg(feature = "ble")]
+            RNodeDriverShutdownSignal::RunningFlag(running) => {
+                running.store(false, Ordering::SeqCst);
+            }
+            #[cfg(not(any(
+                feature = "serial",
+                feature = "rnode-tcp",
+                feature = "ble",
+                target_os = "android",
+                test
+            )))]
+            _ => unreachable!("no RNode driver transport can construct a shutdown primitive"),
+        }
+    }
+}
+
+/// Cloneable, privacy-safe lifecycle handle for one exact RNode driver.
 #[derive(Clone)]
 pub struct RNodeDriverHandle {
     state: watch::Receiver<Arc<RNodeRuntimeSnapshot>>,
+    shutdown: RNodeDriverShutdown,
 }
 
 /// Clone-only subscription to generic RNode driver observations.
@@ -453,6 +539,16 @@ impl RNodeDriverHandle {
             state: self.state.clone(),
         }
     }
+
+    /// Request shutdown of this exact spawned driver instance.
+    ///
+    /// The request is idempotent across every clone. It never resolves an
+    /// interface ID or another global registry entry, so later reuse of the
+    /// same ID cannot redirect it to a different session. Completion remains
+    /// owned by the spawned interface task and its normal join path.
+    pub fn request_shutdown(&self) {
+        self.shutdown.request();
+    }
 }
 
 impl RNodeDriverSubscription {
@@ -668,6 +764,13 @@ impl Drop for RNodeSnapshotPublisher {
     }
 }
 
+#[cfg(test)]
+pub(crate) fn new_rnode_driver_observation(
+    transport: RNodeTransportClass,
+) -> (RNodeSnapshotPublisher, RNodeDriverHandle) {
+    new_rnode_driver_observation_with_shutdown(transport, RNodeDriverShutdown::inert_test())
+}
+
 #[cfg(any(
     feature = "serial",
     feature = "rnode-tcp",
@@ -675,13 +778,17 @@ impl Drop for RNodeSnapshotPublisher {
     target_os = "android",
     test
 ))]
-pub(crate) fn new_rnode_driver_observation(
+pub(crate) fn new_rnode_driver_observation_with_shutdown(
     transport: RNodeTransportClass,
+    shutdown: RNodeDriverShutdown,
 ) -> (RNodeSnapshotPublisher, RNodeDriverHandle) {
     let (state_tx, state_rx) = watch::channel(Arc::new(RNodeRuntimeSnapshot::initial(transport)));
     (
         RNodeSnapshotPublisher::new(state_tx),
-        RNodeDriverHandle { state: state_rx },
+        RNodeDriverHandle {
+            state: state_rx,
+            shutdown,
+        },
     )
 }
 
@@ -697,15 +804,21 @@ fn rnode_stop_registry() -> &'static RNodeStopRegistry {
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
 struct RNodeStopRegistryGuard {
     id: InterfaceId,
+    stop_tx: mpsc::Sender<()>,
 }
 
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
 impl Drop for RNodeStopRegistryGuard {
     fn drop(&mut self) {
-        rnode_stop_registry()
+        let mut registry = rnode_stop_registry()
             .lock()
-            .expect("rnode_stop_registry mutex poisoned")
-            .remove(&self.id);
+            .expect("rnode_stop_registry mutex poisoned");
+        let owns_entry = registry
+            .get(&self.id)
+            .is_some_and(|registered| registered.same_channel(&self.stop_tx));
+        if owns_entry {
+            registry.remove(&self.id);
+        }
     }
 }
 
@@ -714,12 +827,16 @@ fn register_rnode_stop(id: InterfaceId, stop_tx: mpsc::Sender<()>) -> RNodeStopR
     rnode_stop_registry()
         .lock()
         .expect("rnode_stop_registry mutex poisoned")
-        .insert(id, stop_tx);
-    RNodeStopRegistryGuard { id }
+        .insert(id, stop_tx.clone());
+    RNodeStopRegistryGuard { id, stop_tx }
 }
 
-/// Ask a serial/TCP RNode interface to send upstream's detach sequence before
-/// runtime teardown aborts the task. Idempotent; unknown ids are ignored.
+/// Compatibility facade requesting shutdown of the currently registered
+/// serial/TCP RNode for `id`.
+///
+/// New owners should retain [`RNodeDriverHandle`] and call
+/// [`RNodeDriverHandle::request_shutdown`] so later ID reuse cannot redirect
+/// the request. Unknown IDs are ignored.
 #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
 pub fn stop_rnode_interface(id: InterfaceId) {
     let stop_tx = rnode_stop_registry()
@@ -2406,7 +2523,6 @@ pub async fn spawn_rnode_interface_with_driver(
     };
 
     let port = open_configured_rnode_stream(&config, &port_cfg).await?;
-    let (initial_snapshot_publisher, driver) = new_rnode_driver_observation(transport);
 
     let bitrate = calculate_bitrate(
         config.spreading_factor,
@@ -2449,6 +2565,10 @@ pub async fn spawn_rnode_interface_with_driver(
         start_rnode_generation(port, &config, id, &online, &shared_txb, &beacon).await?;
 
     let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1);
+    let (initial_snapshot_publisher, driver) = new_rnode_driver_observation_with_shutdown(
+        transport,
+        RNodeDriverShutdown::from_stop_sender(stop_tx.clone()),
+    );
     let stop_guard = register_rnode_stop(id, stop_tx);
     let online_r = online.clone();
     let rxb_r = shared_rxb.clone();
@@ -3297,6 +3417,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_driver_shutdown_is_idempotent_across_clones() {
+        let (stop_tx, mut stop_rx) = mpsc::channel::<()>(4);
+        let (_publisher, driver) = new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Tcp,
+            RNodeDriverShutdown::from_stop_sender(stop_tx),
+        );
+        let clone = driver.clone();
+
+        driver.request_shutdown();
+        clone.request_shutdown();
+        driver.request_shutdown();
+
+        assert!(stop_rx.try_recv().is_ok());
+        assert!(
+            matches!(stop_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+            "all handle clones must share one shutdown request"
+        );
+    }
+
     #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
     #[test]
     fn test_stop_rnode_interface_signals_registered_driver() {
@@ -3311,6 +3451,40 @@ mod tests {
         stop_rnode_interface(id);
     }
 
+    #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
+    #[test]
+    fn test_exact_shutdown_and_registry_cleanup_resist_same_id_aba() {
+        let id = 0x0BAD_5701;
+        let (old_tx, mut old_rx) = mpsc::channel::<()>(2);
+        let (_old_publisher, old_driver) = new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Tcp,
+            RNodeDriverShutdown::from_stop_sender(old_tx.clone()),
+        );
+        let old_guard = register_rnode_stop(id, old_tx);
+
+        let (new_tx, mut new_rx) = mpsc::channel::<()>(2);
+        let (_new_publisher, _new_driver) = new_rnode_driver_observation_with_shutdown(
+            RNodeTransportClass::Tcp,
+            RNodeDriverShutdown::from_stop_sender(new_tx.clone()),
+        );
+        let new_guard = register_rnode_stop(id, new_tx);
+
+        drop(old_guard);
+        old_driver.request_shutdown();
+        assert!(old_rx.try_recv().is_ok());
+        assert!(
+            matches!(new_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+            "the retired handle must not stop the newer same-ID driver"
+        );
+
+        stop_rnode_interface(id);
+        assert!(
+            new_rx.try_recv().is_ok(),
+            "retired guard cleanup must preserve the newer compatibility entry"
+        );
+        drop(new_guard);
+    }
+
     #[cfg(any(feature = "serial", feature = "rnode-tcp"))]
     #[tokio::test]
     async fn test_rnode_writer_startup_has_two_flush_acked_stages() {
@@ -4909,7 +5083,10 @@ mod tests {
         assert_eq!(snapshot.phase, RNodeRuntimePhase::Ready);
 
         let (state_tx, state_rx) = watch::channel(Arc::new(snapshot.clone()));
-        let driver = RNodeDriverHandle { state: state_rx };
+        let driver = RNodeDriverHandle {
+            state: state_rx,
+            shutdown: RNodeDriverShutdown::inert_test(),
+        };
         let publisher = RNodeSnapshotPublisher::new(state_tx);
         let subscription = driver.watch();
         let retained = subscription.snapshot();