feat: actor monitoring with Down message notifications (Cycle 13)

Add MonitorRegistry for death watch subscriptions. Actors subscribe via
ctx.monitor(target) and receive a Down { addr, reason } message when the
target dies (stop or panic). Supports stacking, demonitor, and automatic
cleanup of dead watcher subscriptions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CLAUDE/notes/progress.md

@@ -2,7 +2,7 @@
 
 ## Current Stage: Phase 1 — Research + First Improvement Cycle
 
-### Status: Cycle 12 COMPLETE
+### Status: Cycle 13 COMPLETE
 
 ## Plan Overview
 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
@@ -124,6 +124,36 @@
   - `stop_nonexistent_actor_returns_error` — stop on bad address returns Err
 - **Result**: 82 tests pass, all workspace compiles
 
+### Cycle 13: Actor Monitoring / Death Watch
+- **Research**: Studied monitoring across Erlang (monitor/2, DOWN messages), Akka (watch/Terminated),
+  Ractor (link, SupervisionEvent), Actix (none), Kameo (link, on_link_died callback)
+  - Key finding: Erlang's unidirectional monitor + message delivery is the best fit for swactor
+    (reuses existing type-erased handler, zero trait changes, composable)
+  - Callbacks (Ractor/Kameo style) rejected: would require adding to AnyActor/ActorInterface traits
+  - Bidirectional links deferred: can layer on top of monitors later
+- **Implementation**: `MonitorRegistry` in delivery.rs + `Down`/`StopReason`/`MonitorRef` in actor.rs
+  - `MonitorRegistry`: `RwLock<HashMap<ActorAddress, Vec<(MonitorRef, ActorAddress)>>>` (watched→watchers)
+    + reverse `RwLock<HashMap<MonitorRef, ActorAddress>>` for O(1) demonitor
+  - `MonitorRef(u64)`: unique token from `AtomicU64` counter
+  - `Down { addr: ActorAddress, reason: StopReason }`: delivered as normal mailbox message
+  - `StopReason`: `Normal` (graceful stop) | `Panicked` (panic, not restartable)
+  - `ctx.monitor(target)` → `MonitorRef` — subscribe to death notifications
+  - `ctx.demonitor(mref)` — cancel a subscription
+  - `cleanup_dead` now returns `Vec<(ActorAddress, StopReason)>` instead of `Vec<ActorAddress>`
+  - After cleanup_dead: iterate dead actors, take_monitors from registry, route Down through normal
+    delivery (pool.deliver for same-worker, transfer_txs for cross-worker, inbox_registry for inboxes)
+  - Dead watcher cleanup: `remove_watcher()` strips monitor subscriptions for dead watchers
+  - Multiple monitors of same target produce independent notifications (stacking, like Erlang)
+- **Tests**: 7 new behavioral tests
+  - `monitor_notifies_on_graceful_stop` — Down{reason: Normal} on stop
+  - `monitor_notifies_on_panic` — Down{reason: Panicked} on panic
+  - `multiple_watchers_all_notified` — two watchers both get Down
+  - `demonitor_cancels_notification` — demonitor → no Down delivered
+  - `dead_watcher_does_not_receive_down` — dead watcher's monitors cleaned up
+  - `down_delivered_to_external_inbox` — Down forwarded through inbox
+  - `stacked_monitors_produce_multiple_notifications` — two monitors on same target → two Downs
+- **Result**: 113 tests pass (106 behavioral + 7 proptest), all workspace compiles, zero warnings
+
 ### Cycle 12: Named Actor Registry
 - **Research**: Studied named actor/service discovery across Erlang (register/2, whereis/1, global, pg),
   Actix (Registry, SystemRegistry — TypeId keys), Bastion (hierarchy-based), Ractor (String keys, DashMap,

src/actor.rs

@@ -137,6 +137,33 @@ where
     }
 }
 
+/// Unique token identifying a monitor subscription.
+///
+/// Returned by [`Ctx::monitor`] and used with [`Ctx::demonitor`] to cancel.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MonitorRef(pub(crate) u64);
+
+/// Reason an actor was removed from the runtime.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum StopReason {
+    /// Graceful stop (via `ctx.stop_self()` or `Runtime::stop_actor()`).
+    Normal,
+    /// Actor panicked and could not be restarted.
+    Panicked,
+}
+
+/// Death notification delivered as a normal message when a monitored actor dies.
+///
+/// Subscribe via [`Ctx::monitor`]. The `Down` message arrives in the watcher's
+/// regular `handle()` method — no special callback needed.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Down {
+    /// Address of the dead actor.
+    pub addr: ActorAddress,
+    /// Why it died.
+    pub reason: StopReason,
+}
+
 /// Internal sentinel message for graceful actor stop.
 /// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`.
 pub(crate) struct StopSignal;
@@ -182,6 +209,10 @@ pub trait ContextInner {
     fn where_is(&self, name: &str) -> Option<ActorAddress>;
     /// Register a name → address mapping. Returns `Err` if the name is taken.
     fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error>;
+    /// Subscribe to death notifications for `target`. Returns a MonitorRef for cancellation.
+    fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef;
+    /// Cancel a monitor subscription.
+    fn demonitor(&self, mref: MonitorRef);
 }
 
 /// Actor syscall interface — passed to `ActorInterface::handle()`.
@@ -276,6 +307,20 @@ impl<'a> Ctx<'a> {
         Ok(addr)
     }
 
+    /// Subscribe to death notifications for `target`.
+    ///
+    /// When `target` dies (stop or panic), a [`Down`] message is delivered to
+    /// this actor's mailbox as a normal message. Multiple monitors of the same
+    /// target create independent subscriptions.
+    pub fn monitor(&self, target: ActorAddress) -> MonitorRef {
+        self.inner.monitor(self.self_addr, target)
+    }
+
+    /// Cancel a previously created monitor subscription.
+    pub fn demonitor(&self, mref: MonitorRef) {
+        self.inner.demonitor(mref);
+    }
+
     /// Spawn a restartable actor. On panic, recreated via `factory` up to
     /// `max_restarts` times before permanent poisoning.
     pub fn spawn_restartable<A, F>(

src/delivery.rs

@@ -1,10 +1,10 @@
 use std::any::Any;
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, OnceLock, RwLock};
 use std::thread::Thread;
 
-use crate::actor::{ActorAddress, AnyActor, Message};
+use crate::actor::{ActorAddress, AnyActor, Message, MonitorRef};
 use crate::channel::Sender;
 use crate::config::RuntimeConfig;
 use crate::stats::WorkerStats;
@@ -190,6 +190,7 @@ pub(crate) struct TickContext<'a> {
     pub(crate) inbox_registry: &'a InboxRegistry,
     pub(crate) config: &'a RuntimeConfig,
     pub(crate) name_registry: &'a NameRegistry,
+    pub(crate) monitor_registry: &'a MonitorRegistry,
     pub(crate) stats_hook: Option<&'a dyn crate::stats::StatsHook>,
     /// Thread handles for waking parked workers on cross-worker sends.
     pub(crate) worker_threads: &'a [OnceLock<Thread>],
@@ -255,6 +256,82 @@ impl NameRegistry {
     }
 }
 
+// ─── Monitor Registry ────────────────────────────────────────────────────────
+
+/// Tracks monitor subscriptions: watched actor → list of (MonitorRef, watcher address).
+///
+/// Write-rare (monitor/demonitor/death), read at cleanup time.
+pub(crate) struct MonitorRegistry {
+    /// watched_addr → [(mref, watcher_addr)]
+    monitors: RwLock<HashMap<ActorAddress, Vec<(MonitorRef, ActorAddress)>>>,
+    /// mref → watched_addr (for O(1) demonitor)
+    ref_to_target: RwLock<HashMap<MonitorRef, ActorAddress>>,
+    next_ref: AtomicU64,
+}
+
+impl MonitorRegistry {
+    pub fn new() -> Self {
+        Self {
+            monitors: RwLock::new(HashMap::new()),
+            ref_to_target: RwLock::new(HashMap::new()),
+            next_ref: AtomicU64::new(1),
+        }
+    }
+
+    /// Register a monitor: `watcher` wants to know when `target` dies.
+    pub fn register(&self, watcher: ActorAddress, target: ActorAddress) -> MonitorRef {
+        let id = self.next_ref.fetch_add(1, Ordering::Relaxed);
+        let mref = MonitorRef(id);
+        self.monitors.write().unwrap()
+            .entry(target)
+            .or_default()
+            .push((mref, watcher));
+        self.ref_to_target.write().unwrap().insert(mref, target);
+        mref
+    }
+
+    /// Cancel a monitor by its ref.
+    pub fn deregister(&self, mref: MonitorRef) {
+        if let Some(target) = self.ref_to_target.write().unwrap().remove(&mref) {
+            let mut monitors = self.monitors.write().unwrap();
+            if let Some(watchers) = monitors.get_mut(&target) {
+                watchers.retain(|(r, _)| *r != mref);
+                if watchers.is_empty() {
+                    monitors.remove(&target);
+                }
+            }
+        }
+    }
+
+    /// Remove and return all monitors for a dead actor.
+    pub fn take_monitors(&self, target: &ActorAddress) -> Vec<(MonitorRef, ActorAddress)> {
+        let watchers = self.monitors.write().unwrap().remove(target).unwrap_or_default();
+        let mut ref_map = self.ref_to_target.write().unwrap();
+        for (mref, _) in &watchers {
+            ref_map.remove(mref);
+        }
+        watchers
+    }
+
+    /// Remove all monitor subscriptions where `addr` is the watcher (dead watcher cleanup).
+    pub fn remove_watcher(&self, addr: &ActorAddress) {
+        let mut monitors = self.monitors.write().unwrap();
+        let mut ref_map = self.ref_to_target.write().unwrap();
+        // Iterate all targets and remove entries where this addr is the watcher
+        monitors.retain(|_target, watchers| {
+            watchers.retain(|(mref, watcher)| {
+                if watcher == addr {
+                    ref_map.remove(mref);
+                    false
+                } else {
+                    true
+                }
+            });
+            !watchers.is_empty()
+        });
+    }
+}
+
 impl<'a> TickContext<'a> {
     /// Route a message whose destination is not in the local address map.
     /// Tries inbox registry, then remote transport, then falls back to inbox error.

src/runtime.rs

@@ -9,7 +9,7 @@ use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopS
 use crate::channel::{Receiver, Sender};
 // Re-export config types so existing code using `runtime::RuntimeConfig` still works
 pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
-use crate::delivery::{AddressMap, Envelope, InboxRegistry, NameRegistry, Placement, TickContext, WorkerId};
+use crate::delivery::{AddressMap, Envelope, InboxRegistry, MonitorRegistry, NameRegistry, Placement, TickContext, WorkerId};
 use crate::stats::{StatsHook, WorkerStats};
 // Re-export stats types so existing code using `runtime::*` still works
 pub use crate::stats::{RuntimeStats, WorkerInfo};
@@ -63,6 +63,7 @@ pub struct Runtime {
     address_map: Arc<AddressMap>,
     inbox_registry: Arc<InboxRegistry>,
     name_registry: Arc<NameRegistry>,
+    monitor_registry: Arc<MonitorRegistry>,
     transfer_txs: Vec<Sender<Envelope>>,
     spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
     placement: Placement,
@@ -121,6 +122,7 @@ impl Runtime {
         let address_map = Arc::new(AddressMap::with_capacity(config.max_actors));
         let inbox_registry = Arc::new(InboxRegistry::new());
         let name_registry = Arc::new(NameRegistry::new());
+        let monitor_registry = Arc::new(MonitorRegistry::new());
 
         let mut transfer_txs = Vec::with_capacity(num_workers);
         let mut spawn_txs = Vec::with_capacity(num_workers);
@@ -159,6 +161,7 @@ impl Runtime {
             address_map,
             inbox_registry,
             name_registry,
+            monitor_registry,
             transfer_txs,
             spawn_txs,
             placement,
@@ -292,6 +295,7 @@ impl Runtime {
             inbox_registry: &self.inbox_registry,
             config: &self.config,
             name_registry: &self.name_registry,
+            monitor_registry: &self.monitor_registry,
             stats_hook: self.stats_hook.as_deref(),
             worker_threads: &self.worker_threads,
             #[cfg(feature = "transport")]
@@ -500,4 +504,12 @@ impl ContextInner for Runtime {
     fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error> {
         self.name_registry.register(name, addr)
     }
+
+    fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef {
+        self.monitor_registry.register(watcher, target)
+    }
+
+    fn demonitor(&self, mref: crate::actor::MonitorRef) {
+        self.monitor_registry.deregister(mref);
+    }
 }

src/worker.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use std::thread;
 use std::time::Instant;
 
-use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopSignal, TimerRequest};
+use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest};
 use crate::channel::Receiver;
 use crate::config::MailboxOverflow;
 use crate::delivery::{Envelope, TickContext, WorkerId};
@@ -319,9 +319,9 @@ impl Worker {
             };
             let dead = self.pool.cleanup_dead(&cleanup_ctx);
             if !dead.is_empty() {
-                for addr in &dead {
-                    tc.address_map.remove(addr);
-                    tc.name_registry.unregister_by_addr(addr);
+                for &(addr, _) in &dead {
+                    tc.address_map.remove(&addr);
+                    tc.name_registry.unregister_by_addr(&addr);
                 }
                 // Re-publish num_actors after cleanup so stats reflect removal
                 self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed);
@@ -334,8 +334,34 @@ impl Worker {
             self.pool.deliver(&addr, msg);
         }
 
+        // Emit Down notifications for monitored dead actors
+        for &(addr, reason) in &dead {
+            let watchers = tc.monitor_registry.take_monitors(&addr);
+            for (_mref, watcher) in watchers {
+                let down = crate::actor::Down { addr, reason };
+                // Route through normal delivery path
+                if self.pool.contains(&watcher) {
+                    self.pool.deliver(&watcher, Box::new(down));
+                } else {
+                    match tc.address_map.lookup(&watcher) {
+                        Some(wid) => {
+                            tc.transfer_txs[wid.as_usize()]
+                                .send(Envelope::new(watcher, Box::new(down)));
+                            crate::runtime::notify_worker(tc.worker_threads, wid.as_usize());
+                        }
+                        None => {
+                            let _ = tc.inbox_registry.try_deliver(watcher, Box::new(down));
+                        }
+                    }
+                }
+            }
+            // Clean up any monitors the dead actor had placed on others
+            tc.monitor_registry.remove_watcher(&addr);
+        }
+
         // GC orphaned interval timers for actors that were just removed
-        self.timers.gc_dead_intervals(&dead);
+        let dead_addrs: Vec<ActorAddress> = dead.iter().map(|(a, _)| *a).collect();
+        self.timers.gc_dead_intervals(&dead_addrs);
 
         did_work
     }
@@ -427,6 +453,14 @@ impl ContextInner for WorkerContext<'_> {
     fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), crate::Error> {
         self.tc.name_registry.register(name, addr)
     }
+
+    fn monitor(&self, watcher: ActorAddress, target: ActorAddress) -> crate::actor::MonitorRef {
+        self.tc.monitor_registry.register(watcher, target)
+    }
+
+    fn demonitor(&self, mref: crate::actor::MonitorRef) {
+        self.tc.monitor_registry.deregister(mref);
+    }
 }
 
 struct ActorSlot {
@@ -626,19 +660,22 @@ impl ActorPool {
         self.actors.values().map(|slot| slot.mailbox.len()).sum()
     }
 
-    /// Remove poisoned and stopping actors, returning their addresses.
+    /// Remove poisoned and stopping actors, returning their addresses and stop reasons.
     /// Called after tick_all so the caller can clean up the address map.
     ///
     /// For stopping actors: calls `on_stop()` before removal (wrapped in catch_unwind).
     /// For poisoned actors: `on_stop()` is NOT called (state may be corrupt).
-    pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec<ActorAddress> {
-        let dead: Vec<ActorAddress> = self
+    pub fn cleanup_dead(&mut self, inner: &dyn ContextInner) -> Vec<(ActorAddress, StopReason)> {
+        let dead: Vec<(ActorAddress, StopReason)> = self
             .actors
             .iter()
             .filter(|(_, slot)| slot.poisoned || slot.stopping)
-            .map(|(&addr, _)| addr)
+            .map(|(&addr, slot)| {
+                let reason = if slot.poisoned { StopReason::Panicked } else { StopReason::Normal };
+                (addr, reason)
+            })
             .collect();
-        for &addr in &dead {
+        for &(addr, _) in &dead {
             if let Some(mut slot) = self.actors.remove(&addr) {
                 // Call on_stop for gracefully stopping actors only
                 if slot.stopping && !slot.poisoned {

tests/runtime_api.rs

@@ -1,7 +1,7 @@
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
-use swactor::actor::{ActorAddress, ActorInterface};
+use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason};
 use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
 
 // ── Messages ────────────────────────────────────────────────────────────────
@@ -3086,3 +3086,246 @@ fn ctx_spawn_named_registers_from_handler() {
     let child_addr = inbox.try_recv().expect("should receive child address");
     assert_eq!(rt.where_is("child"), Some(child_addr.0), "name registered from handler");
 }
+
+// ── Actor Monitoring / Death Watch ──────────────────────────────────────────
+
+/// An actor that monitors a target and forwards Down notifications to a reply address.
+struct WatcherActor {
+    watch_target: ActorAddress,
+    reply_to: ActorAddress,
+    mref: Option<MonitorRef>,
+}
+
+impl ActorInterface for WatcherActor {
+    type Incoming = Down;
+    type Response = ();
+    fn on_start(&mut self, ctx: &Ctx) {
+        self.mref = Some(ctx.monitor(self.watch_target));
+    }
+    fn handle(&mut self, ctx: &Ctx, msg: Down) {
+        // Forward the Down notification to the test inbox
+        ctx.send(self.reply_to, msg).unwrap();
+    }
+}
+
+/// Given actor A monitors actor B,
+/// when B is gracefully stopped,
+/// then A receives a Down { reason: Normal } message.
+#[test]
+fn monitor_notifies_on_graceful_stop() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PingPongActor).unwrap();
+    let _watcher = rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: *inbox.addr(),
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start → watcher sets up monitor
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // target receives StopSignal → cleanup_dead emits Down
+    rt.tick(); // watcher receives Down → forwards to inbox
+
+    let down = inbox.try_recv().expect("should receive Down notification");
+    assert_eq!(down.addr, target);
+    assert_eq!(down.reason, StopReason::Normal);
+}
+
+/// Given actor A monitors actor B,
+/// when B panics,
+/// then A receives a Down { reason: Panicked } message.
+#[test]
+fn monitor_notifies_on_panic() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PanicActor).unwrap();
+    let _watcher = rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: *inbox.addr(),
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start
+    rt.send_to(target, PanicMsg).unwrap();
+    rt.tick(); // target panics → cleanup_dead emits Down
+    rt.tick(); // watcher receives Down → forwards to inbox
+
+    let down = inbox.try_recv().expect("should receive Down on panic");
+    assert_eq!(down.addr, target);
+    assert_eq!(down.reason, StopReason::Panicked);
+}
+
+/// Given two actors both monitor the same target,
+/// when the target dies,
+/// then both watchers receive independent Down notifications.
+#[test]
+fn multiple_watchers_all_notified() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox1 = rt.new_inbox::<Down>().unwrap();
+    let inbox2 = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PingPongActor).unwrap();
+    rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: *inbox1.addr(),
+        mref: None,
+    }).unwrap();
+    rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: *inbox2.addr(),
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start for all
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // cleanup → Down emitted to both watchers
+    rt.tick(); // watchers forward Down to inboxes
+
+    assert!(inbox1.try_recv().is_some(), "watcher 1 should receive Down");
+    assert!(inbox2.try_recv().is_some(), "watcher 2 should receive Down");
+}
+
+/// An actor that demonitors in response to a Ping message.
+struct DemonitorActor {
+    watch_target: ActorAddress,
+    mref: Option<MonitorRef>,
+}
+
+impl ActorInterface for DemonitorActor {
+    type Incoming = Ping;
+    type Response = ();
+    fn on_start(&mut self, ctx: &Ctx) {
+        self.mref = Some(ctx.monitor(self.watch_target));
+    }
+    fn handle(&mut self, ctx: &Ctx, _msg: Ping) {
+        // Cancel the monitor
+        if let Some(mref) = self.mref.take() {
+            ctx.demonitor(mref);
+        }
+    }
+}
+
+/// Given actor A monitors actor B then demonitors,
+/// when B dies,
+/// then A does NOT receive a Down notification.
+#[test]
+fn demonitor_cancels_notification() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let down_inbox = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PingPongActor).unwrap();
+    let watcher = rt.spawn(DemonitorActor {
+        watch_target: target,
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start → monitor set up
+
+    // Trigger demonitor
+    rt.send_to(watcher, Ping { reply_to: ActorAddress::default() }).unwrap();
+    rt.tick(); // handle → demonitor
+
+    // Now kill the target
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // cleanup — no Down should be emitted
+    rt.tick(); // extra tick to be sure
+
+    assert!(down_inbox.try_recv().is_none(), "demonitored — should NOT receive Down");
+}
+
+/// Given actor A monitors B, and A dies before B,
+/// when B dies,
+/// then no Down is delivered (dead watcher cleaned up).
+#[test]
+fn dead_watcher_does_not_receive_down() {
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    let target = rt.spawn(PingPongActor).unwrap();
+    let watcher = rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: ActorAddress::default(), // won't matter, watcher dies first
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start → monitor set up
+    rt.stop_actor(watcher).unwrap();
+    rt.tick(); // watcher dies → its monitors are cleaned up
+
+    // Now kill the target — the dead watcher's subscription should be gone
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // cleanup — should not panic or try to deliver to dead watcher
+    // If we get here without panic, the test passes
+}
+
+/// Given an external inbox monitors via the runtime,
+/// when the target dies,
+/// then the inbox receives a Down message.
+#[test]
+fn down_delivered_to_external_inbox() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PingPongActor).unwrap();
+
+    // Set up a monitor from an actor that forwards Down to the inbox.
+    // The watcher is an actor, but the final recipient is the inbox.
+    let _watcher = rt.spawn(WatcherActor {
+        watch_target: target,
+        reply_to: *inbox.addr(),
+        mref: None,
+    }).unwrap();
+
+    rt.tick(); // on_start
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // cleanup → Down to watcher
+    rt.tick(); // watcher forwards to inbox
+
+    let down = inbox.try_recv().expect("inbox should receive forwarded Down");
+    assert_eq!(down.addr, target);
+    assert_eq!(down.reason, StopReason::Normal);
+}
+
+/// Given actor A monitors B with two independent monitors,
+/// when B dies,
+/// then A receives two Down messages (one per monitor).
+#[test]
+fn stacked_monitors_produce_multiple_notifications() {
+    /// An actor that creates two monitors on the same target.
+    struct DoubleWatcherActor {
+        target: ActorAddress,
+        reply_to: ActorAddress,
+    }
+
+    impl ActorInterface for DoubleWatcherActor {
+        type Incoming = Down;
+        type Response = ();
+        fn on_start(&mut self, ctx: &Ctx) {
+            ctx.monitor(self.target);
+            ctx.monitor(self.target);
+        }
+        fn handle(&mut self, ctx: &Ctx, msg: Down) {
+            ctx.send(self.reply_to, msg).unwrap();
+        }
+    }
+
+    let rt = Runtime::new(RuntimeConfig::default());
+    let inbox = rt.new_inbox::<Down>().unwrap();
+
+    let target = rt.spawn(PingPongActor).unwrap();
+    rt.spawn(DoubleWatcherActor {
+        target,
+        reply_to: *inbox.addr(),
+    }).unwrap();
+
+    rt.tick(); // on_start → 2 monitors
+    rt.stop_actor(target).unwrap();
+    rt.tick(); // cleanup → 2 Down messages to watcher
+    rt.tick(); // watcher forwards both to inbox
+
+    assert!(inbox.try_recv().is_some(), "first Down");
+    assert!(inbox.try_recv().is_some(), "second Down");
+    assert!(inbox.try_recv().is_none(), "no more");
+}