diff --git a/src/pg.rs b/src/pg.rs
index 85ee2f8..dcf257b 100644
--- a/src/pg.rs
+++ b/src/pg.rs
@@ -4,26 +4,32 @@
 //! registry is a *bimap*: at most one pid per name. A group is the opposite —
 //! many pids per name, the same pid in many groups — so it cannot be a
 //! generalization of the bimap; it is its own keyspace sharing only the
-//! liveness helper and the lock *class*.
+//! liveness signal source (monitors) and the lock *class*.
 //!
-//! ## Cleanup is eager (unlike the registry)
+//! ## Cleanup is eager and monitor-driven (unlike the registry)
 //!
 //! The registry prunes a stale binding lazily, on contact, because it only
-//! ever resolves one binding at a time. A group is *iterated* — `members`
-//! fans out to every member — so it must not carry dead members across a
-//! broadcast. Removal is therefore eager and monitor-driven: each `join`
-//! installs a `monitor(pid)` and the one-shot `Down` evicts the member
-//! (wired in Phase 2). A generation-checked liveness read stays on the read
-//! path as a belt-and-braces backstop (Phase 3), matching the registry's
-//! prune-on-contact guard.
+//! ever resolves one binding at a time. A group is *iterated* — `members` fans
+//! out to every member — so it must not carry dead members across a broadcast.
+//! Each [`join`] installs a `monitor(pid)` and keeps the resulting [`Monitor`]
+//! *alongside the group entry*; the actor's one-shot `Down` lands in that
+//! monitor's channel when it dies. Every group operation [`reap`s][reap] the
+//! group it touches first — draining each membership's monitor with a
+//! non-blocking `try_recv` — and on the first sign of death sweeps that pid out
+//! of *every* group via the predicate primitive. So the set is self-pruning on
+//! contact rather than filtered on read; Phase 3 adds a generation-checked
+//! liveness read as a belt-and-braces backstop on the read path.
+//!
+//! [reap]: ProcessGroups::reap_group
 //!
 //! ## Eviction is one dumb primitive
 //!
 //! [`ProcessGroups::remove_where`] removes every member matching a predicate
-//! from every group. The primitive never knows *why* a member leaves — that
-//! is the caller's concern. Its first caller is the monitor death hook (this
-//! RFC); the later `evict_incarnation(node, inc)` sweep (RFC 010) reuses the
-//! same predicate path, which is the whole reason to shape it as a predicate.
+//! from every group. The primitive never knows *why* a member leaves — that is
+//! the caller's concern. Its first caller is the monitor death hook (this RFC,
+//! via `reap_group`); the later `evict_incarnation(node, inc)` sweep (RFC 010)
+//! reuses the same predicate path, which is the whole reason to shape it as a
+//! predicate.
 //!
 //! ## Identity is cluster-shaped from the first commit
 //!
@@ -38,11 +44,23 @@
 //!
 //! ## Locking
 //!
-//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`
-//! exactly: never held with any other lock, and the eviction path keeps it
-//! off the send path (it does not send under the lock). Liveness checks under
-//! it read only the atomic slot word, so the leaf rule holds trivially.
+//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`. The
+//! group lock is never held with another Leaf (it never touches the registry
+//! or a slot's cold lock). The two places that *do* need another lock are kept
+//! off the group-lock path:
+//!
+//!   - `monitor()` / `demonitor()` acquire the target's cold lock (Leaf) and
+//!     so run *before* / *after* the group lock, never under it.
+//!   - draining a monitor with `try_recv` takes the channel's Channel-class
+//!     lock — permitted *under* a Leaf by the lock order (`raw_mutex.rs`), and
+//!     a channel critical section only does the lock-free unpark protocol, so
+//!     no Leaf is ever nested under it.
+//!
+//! Evicted and rejected [`Monitor`]s are dropped only *after* the group lock is
+//! released, so a receiver-drop never runs a wakeup under the lock (same
+//! discipline as `demonitor`).
 
+use crate::monitor::{demonitor, monitor, Monitor};
 use crate::pid::Pid;
 use crate::scheduler::with_runtime;
 use std::collections::HashMap;
@@ -118,12 +136,21 @@ pub struct Member {
     pub pid: Pid,
 }
 
+/// One membership: a [`Member`] and the [`Monitor`] that watches its liveness.
+/// The monitor lives *alongside* the group entry (RFC 012) so a group is
+/// self-contained: draining the membership tells us whether the member is
+/// still alive, and dropping the membership drops its monitor.
+struct Membership {
+    member: Member,
+    monitor: Monitor,
+}
+
 /// The store: `name → multiset<Member>`. Within a single group a `Member`
 /// appears at most once (`join` is idempotent); the *multiset* framing is for
 /// cluster-readiness — the same pid is freely a member of many groups, and the
 /// width admits multiples in general. Held under one Leaf-class `RawMutex`.
 pub(crate) struct ProcessGroups {
-    groups: HashMap<String, Vec<Member>>,
+    groups: HashMap<String, Vec<Membership>>,
 }
 
 impl ProcessGroups {
@@ -131,50 +158,93 @@ impl ProcessGroups {
         Self { groups: HashMap::new() }
     }
 
-    /// Add `member` to `group`. Idempotent within a group: returns `true` if
-    /// it was newly inserted, `false` if it was already present.
-    pub(crate) fn join(&mut self, group: &str, member: Member) -> bool {
+    /// Insert `ms` into `group`. Idempotent on the *member*: if the member is
+    /// already present the new membership is handed back (`Some`) so the caller
+    /// can tear its now-redundant monitor down outside the lock; `None` means
+    /// it was inserted.
+    fn join(&mut self, group: &str, ms: Membership) -> Option<Membership> {
         let v = self.groups.entry(group.to_owned()).or_default();
-        if v.contains(&member) {
-            return false;
+        if v.iter().any(|e| e.member == ms.member) {
+            return Some(ms);
         }
-        v.push(member);
-        true
+        v.push(ms);
+        None
     }
 
-    /// Remove `member` from `group`. Returns whether anything was removed.
-    /// An emptied group is pruned so the keyspace does not leak empty vecs.
-    pub(crate) fn leave(&mut self, group: &str, member: Member) -> bool {
-        let Some(v) = self.groups.get_mut(group) else {
-            return false;
-        };
-        let before = v.len();
-        v.retain(|m| *m != member);
-        let removed = v.len() != before;
+    /// Remove `member`'s membership from `group`, returning it (so the caller
+    /// can `demonitor` it outside the lock). An emptied group is pruned.
+    fn leave(&mut self, group: &str, member: Member) -> Option<Membership> {
+        let v = self.groups.get_mut(group)?;
+        let pos = v.iter().position(|e| e.member == member)?;
+        let removed = v.remove(pos);
         if v.is_empty() {
             self.groups.remove(group);
         }
-        removed
+        Some(removed)
     }
 
     /// The one dumb eviction primitive: drop every member matching `pred` from
-    /// every group, pruning emptied groups. The primitive does not know *why*
-    /// a member leaves. First caller: the monitor death hook (Phase 2). Later:
-    /// `evict_incarnation` (RFC 010) over the same path.
-    //
-    // No non-test caller until Phase 2 wires the death hook to it.
-    #[allow(dead_code)]
-    pub(crate) fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) {
+    /// every group, pruning emptied groups, and return the evicted memberships'
+    /// monitors for the caller to drop outside the lock. The primitive does not
+    /// know *why* a member leaves. Callers: the death hook (`reap_group`) and,
+    /// later, `evict_incarnation` (RFC 010), over the same path. Insertion
+    /// order within a group is preserved (`members` / `pick` are order-stable).
+    fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) -> Vec<Monitor> {
+        let mut evicted = Vec::new();
         self.groups.retain(|_, v| {
-            v.retain(|m| !pred(m));
+            let mut i = 0;
+            while i < v.len() {
+                if pred(&v[i].member) {
+                    evicted.push(v.remove(i).monitor);
+                } else {
+                    i += 1;
+                }
+            }
             !v.is_empty()
         });
+        evicted
     }
 
-    /// Raw enumeration of a group's members — no liveness filtering (the
-    /// public `members` layers that on in Phase 3). Absent group → empty.
-    pub(crate) fn members_of(&self, group: &str) -> Vec<Member> {
-        self.groups.get(group).cloned().unwrap_or_default()
+    /// Drain-on-contact death hook. Drains every membership monitor in `group`
+    /// with a non-blocking `try_recv`: a delivered `Down` (any reason) or a
+    /// closed channel means that member is dead. On the first death detected,
+    /// sweep *all* of the dead pids out of *every* group via [`remove_where`]
+    /// — a death is removed from each group it joined, not just the one being
+    /// touched. Returns the evicted monitors to drop outside the lock.
+    fn reap_group(&mut self, group: &str) -> Vec<Monitor> {
+        let dead: Vec<Pid> = {
+            let Some(v) = self.groups.get(group) else {
+                return Vec::new();
+            };
+            v.iter()
+                .filter_map(|e| match e.monitor.rx.try_recv() {
+                    // A Down arrived, or the channel closed and drained: dead.
+                    Ok(Some(_)) | Err(_) => Some(e.member.pid),
+                    // Empty but open — the sender still lives in the slot: alive.
+                    Ok(None) => None,
+                })
+                .collect()
+        };
+        if dead.is_empty() {
+            return Vec::new();
+        }
+        self.remove_where(|m| dead.contains(&m.pid))
+    }
+
+    /// Raw enumeration of a group's members — no liveness filtering (Phase 3
+    /// layers a generation-checked backstop on the public `members`). Absent
+    /// group → empty.
+    fn members_of(&self, group: &str) -> Vec<Member> {
+        self.groups
+            .get(group)
+            .map(|v| v.iter().map(|e| e.member).collect())
+            .unwrap_or_default()
+    }
+
+    /// The first member of `group`, if any — backing the stateless first-live
+    /// `pick` without cloning the whole group.
+    fn first_member(&self, group: &str) -> Option<Member> {
+        self.groups.get(group).and_then(|v| v.first().map(|e| e.member))
     }
 }
 
@@ -184,62 +254,107 @@ fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member {
 }
 
 /// Add `pid` to `group`. The same pid may join many groups; within one group a
-/// pid is a member at most once (idempotent). Returns `true` if this call
-/// newly added the membership.
+/// pid is a member at most once (idempotent). Returns `true` if this call newly
+/// added the membership, `false` if it was already a member.
+///
+/// Installs a `monitor(pid)` whose one-shot `Down` drives eviction: the
+/// registration races `finalize_actor` under the slot's cold lock exactly as
+/// every other monitor does, so no death slips between the join and the
+/// registration. A redundant (idempotent) join tears its extra monitor back
+/// down.
 ///
 /// Panics if called outside `Runtime::run()`.
-//
-// Phase 2 will install `monitor(pid)` on first join (registered under the
-// shared mutex, racing `finalize_actor` exactly as every other monitor does)
-// and route its one-shot `Down` to `remove_where(|m| m.pid == pid)`.
 pub fn join(group: impl Into<String>, pid: Pid) -> bool {
     let group = group.into();
-    with_runtime(|inner| {
-        let member = member_for(inner, pid);
-        inner.process_groups.lock().join(&group, member)
-    })
+    // Install the monitor BEFORE taking the group lock: monitor() acquires the
+    // target's cold lock (Leaf), and two Leaf locks are never held at once.
+    let mon = monitor(pid);
+
+    let (rejected, reaped) = with_runtime(|inner| {
+        let ms = Membership { member: member_for(inner, pid), monitor: mon };
+        let mut pg = inner.process_groups.lock();
+        let reaped = pg.reap_group(&group);
+        let rejected = pg.join(&group, ms);
+        (rejected, reaped)
+    });
+
+    // Outside the group lock: drop the reaped (dead) monitors, and if this join
+    // was redundant, demonitor + drop the extra monitor we just installed.
+    drop(reaped);
+    match rejected {
+        Some(dup) => {
+            demonitor(&dup.monitor);
+            false
+        }
+        None => true,
+    }
 }
 
 /// Drop `pid`'s membership of `group`. Returns whether a membership was
-/// removed.
+/// removed. The membership's monitor is demonitored and dropped.
 ///
 /// Panics if called outside `Runtime::run()`.
 pub fn leave(group: &str, pid: Pid) -> bool {
-    with_runtime(|inner| {
+    let (removed, reaped) = with_runtime(|inner| {
         let member = member_for(inner, pid);
-        inner.process_groups.lock().leave(group, member)
-    })
+        let mut pg = inner.process_groups.lock();
+        let reaped = pg.reap_group(group);
+        let removed = pg.leave(group, member);
+        (removed, reaped)
+    });
+
+    drop(reaped);
+    match removed {
+        Some(ms) => {
+            demonitor(&ms.monitor);
+            true
+        }
+        None => false,
+    }
 }
 
-/// Fan-out read: every member of `group`.
+/// Fan-out read: every live member of `group`.
 ///
-/// The set is self-pruning once the death hook lands (Phase 2), so liveness is
-/// maintained by eviction rather than filtered on read; Phase 3 adds a
-/// generation-checked liveness read as the belt-and-braces backstop.
+/// The touched group is reaped first, so dead members are evicted before the
+/// read rather than filtered out of it. Phase 3 adds a generation-checked
+/// liveness read as a belt-and-braces backstop.
 ///
 /// Panics if called outside `Runtime::run()`.
 pub fn members(group: &str) -> Vec<Pid> {
-    with_runtime(|inner| {
-        inner.process_groups.lock().members_of(group).into_iter().map(|m| m.pid).collect()
-    })
+    let (pids, reaped) = with_runtime(|inner| {
+        let mut pg = inner.process_groups.lock();
+        let reaped = pg.reap_group(group);
+        let pids: Vec<Pid> = pg.members_of(group).into_iter().map(|m| m.pid).collect();
+        (pids, reaped)
+    });
+    drop(reaped);
+    pids
 }
 
-/// Pool/discovery read: one member of `group`, or `None` if empty. Stateless
-/// first-live selection (the first-*live* refinement lands in Phase 3; smarter
-/// routing is an explicitly later, clustered concern).
+/// Pool/discovery read: one live member of `group`, or `None` if empty.
+/// Stateless first-live selection (the touched group is reaped first, so the
+/// first surviving member is live; smarter routing is an explicitly later,
+/// clustered concern per RFC 010).
 ///
 /// Panics if called outside `Runtime::run()`.
 pub fn pick(group: &str) -> Option<Pid> {
-    with_runtime(|inner| {
-        inner.process_groups.lock().members_of(group).into_iter().next().map(|m| m.pid)
-    })
+    let (picked, reaped) = with_runtime(|inner| {
+        let mut pg = inner.process_groups.lock();
+        let reaped = pg.reap_group(group);
+        let picked = pg.first_member(group).map(|m| m.pid);
+        (picked, reaped)
+    });
+    drop(reaped);
+    picked
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::channel::{channel, Sender};
+    use crate::monitor::{Down, DownReason, MonitorId};
 
-    fn m(index: u32, generation: u32) -> Member {
+    fn member(index: u32, generation: u32) -> Member {
         Member {
             node: DEFAULT_NODE_ID,
             incarnation: DEFAULT_INCARNATION,
@@ -247,69 +362,137 @@ mod tests {
         }
     }
 
+    /// A synthetic membership with a real (but slot-less) monitor channel. The
+    /// returned `Sender` stands in for the slot's `Down` sender: hold it to
+    /// keep the member "alive" (`try_recv` → `Ok(None)`), `send` a `Down` to
+    /// simulate death, or `drop` it to simulate a drained/closed channel.
+    fn synth(index: u32, generation: u32) -> (Membership, Sender<Down>) {
+        let pid = Pid::new(index, generation);
+        let (tx, rx) = channel::<Down>();
+        let ms = Membership {
+            member: member(index, generation),
+            monitor: Monitor { id: MonitorId(0), target: pid, rx },
+        };
+        (ms, tx)
+    }
+
     #[test]
     fn join_is_idempotent_within_a_group() {
         let mut pg = ProcessGroups::new();
-        assert!(pg.join("workers", m(1, 0)), "first join is new");
-        assert!(!pg.join("workers", m(1, 0)), "second identical join is a no-op");
-        assert_eq!(pg.members_of("workers"), vec![m(1, 0)]);
+        let (a, _ta) = synth(1, 0);
+        let (b, _tb) = synth(1, 0);
+        assert!(pg.join("workers", a).is_none(), "first join inserts");
+        assert!(pg.join("workers", b).is_some(), "second identical join is handed back");
+        assert_eq!(pg.members_of("workers"), vec![member(1, 0)]);
     }
 
     #[test]
     fn same_pid_in_many_groups_is_independent() {
         let mut pg = ProcessGroups::new();
-        pg.join("a", m(1, 0));
-        pg.join("b", m(1, 0));
-        pg.join("b", m(2, 0));
-        assert_eq!(pg.members_of("a"), vec![m(1, 0)]);
-        assert_eq!(pg.members_of("b"), vec![m(1, 0), m(2, 0)]);
+        let (a, _ta) = synth(1, 0);
+        let (b, _tb) = synth(1, 0);
+        let (c, _tc) = synth(2, 0);
+        pg.join("a", a);
+        pg.join("b", b);
+        pg.join("b", c);
+        assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
+        assert_eq!(pg.members_of("b"), vec![member(1, 0), member(2, 0)]);
     }
 
     #[test]
     fn distinct_generations_are_distinct_members() {
         // ABA guard: same slot index, different generation = different actor.
         let mut pg = ProcessGroups::new();
-        assert!(pg.join("g", m(1, 0)));
-        assert!(pg.join("g", m(1, 1)), "different generation is a distinct member");
-        assert_eq!(pg.members_of("g"), vec![m(1, 0), m(1, 1)]);
+        let (a, _ta) = synth(1, 0);
+        let (b, _tb) = synth(1, 1);
+        assert!(pg.join("g", a).is_none());
+        assert!(pg.join("g", b).is_none(), "different generation is a distinct member");
+        assert_eq!(pg.members_of("g"), vec![member(1, 0), member(1, 1)]);
     }
 
     #[test]
     fn leave_removes_one_membership_and_prunes_empty_groups() {
         let mut pg = ProcessGroups::new();
-        pg.join("g", m(1, 0));
-        pg.join("g", m(2, 0));
-        assert!(pg.leave("g", m(1, 0)));
-        assert_eq!(pg.members_of("g"), vec![m(2, 0)]);
-        assert!(!pg.leave("g", m(1, 0)), "second leave finds nothing");
-        assert!(pg.leave("g", m(2, 0)));
+        let (a, _ta) = synth(1, 0);
+        let (b, _tb) = synth(2, 0);
+        pg.join("g", a);
+        pg.join("g", b);
+        assert!(pg.leave("g", member(1, 0)).is_some());
+        assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
+        assert!(pg.leave("g", member(1, 0)).is_none(), "second leave finds nothing");
+        assert!(pg.leave("g", member(2, 0)).is_some());
         assert!(pg.members_of("g").is_empty(), "group is now empty");
-        // Leaving a never-joined group is a clean no-op.
-        assert!(!pg.leave("never", m(9, 0)));
+        assert!(pg.leave("never", member(9, 0)).is_none(), "leaving an unknown group is a no-op");
     }
 
     #[test]
     fn remove_where_sweeps_every_group() {
         let mut pg = ProcessGroups::new();
-        pg.join("a", m(1, 0));
-        pg.join("a", m(2, 0));
-        pg.join("b", m(1, 0));
-        pg.join("c", m(3, 0));
+        for (g, (m, _t)) in [("a", synth(1, 0)), ("a", synth(2, 0)), ("b", synth(1, 0)), ("c", synth(3, 0))] {
+            pg.join(g, m);
+        }
         // Death of pid index 1 (any generation) evicts it everywhere.
-        pg.remove_where(|mem| mem.pid.index() == 1);
-        assert_eq!(pg.members_of("a"), vec![m(2, 0)]);
+        let evicted = pg.remove_where(|mem| mem.pid.index() == 1);
+        assert_eq!(evicted.len(), 2, "pid 1 was in a and b");
+        assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
         assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned");
-        assert_eq!(pg.members_of("c"), vec![m(3, 0)]);
+        assert_eq!(pg.members_of("c"), vec![member(3, 0)]);
     }
 
     #[test]
     fn remove_where_can_match_an_incarnation_sweep() {
         // Shape check for the later evict_incarnation(node, inc) caller.
         let mut pg = ProcessGroups::new();
-        let dead = Member { incarnation: Incarnation::new(7), ..m(1, 0) };
+        let pid = Pid::new(1, 0);
+        let (tx, rx) = channel::<Down>();
+        let dead = Membership {
+            member: Member { node: DEFAULT_NODE_ID, incarnation: Incarnation::new(7), pid },
+            monitor: Monitor { id: MonitorId(0), target: pid, rx },
+        };
+        let _keep = tx;
+        let (live, _tl) = synth(2, 0);
         pg.join("g", dead);
-        pg.join("g", m(2, 0));
-        pg.remove_where(|mem| mem.incarnation == Incarnation::new(7));
-        assert_eq!(pg.members_of("g"), vec![m(2, 0)]);
+        pg.join("g", live);
+        let evicted = pg.remove_where(|mem| mem.incarnation == Incarnation::new(7));
+        assert_eq!(evicted.len(), 1);
+        assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
+    }
+
+    #[test]
+    fn reap_keeps_live_members() {
+        let mut pg = ProcessGroups::new();
+        let (a, _ta) = synth(1, 0); // sender held: member stays alive
+        pg.join("a", a);
+        assert!(pg.reap_group("a").is_empty(), "no deaths");
+        assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
+    }
+
+    #[test]
+    fn reap_evicts_a_dead_member_and_sweeps_all_its_groups() {
+        let mut pg = ProcessGroups::new();
+        let (a1, ta1) = synth(1, 0); // pid 1 in group a
+        let (a2, _ta2) = synth(2, 0); // pid 2 in group a (stays alive)
+        let (b1, _tb1) = synth(1, 0); // pid 1 in group b
+        pg.join("a", a1);
+        pg.join("a", a2);
+        pg.join("b", b1);
+        // pid 1 dies: its group-a monitor receives a Down. Its group-b monitor
+        // has not — reap must still sweep pid 1 out of b by the pid predicate.
+        ta1.send(Down { pid: Pid::new(1, 0), reason: DownReason::Exit }).unwrap();
+        let evicted = pg.reap_group("a");
+        assert_eq!(evicted.len(), 2, "pid 1's memberships in both a and b are evicted");
+        assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
+        assert!(pg.members_of("b").is_empty(), "swept from b too; pruned");
+    }
+
+    #[test]
+    fn reap_treats_a_closed_channel_as_dead() {
+        let mut pg = ProcessGroups::new();
+        let (a, ta) = synth(1, 0);
+        pg.join("a", a);
+        drop(ta); // sender gone, queue empty → try_recv = Err(RecvError) = dead
+        let evicted = pg.reap_group("a");
+        assert_eq!(evicted.len(), 1);
+        assert!(pg.members_of("a").is_empty());
     }
 }
diff --git a/tests/pg.rs b/tests/pg.rs
new file mode 100644
index 0000000..eb46ca0
--- /dev/null
+++ b/tests/pg.rs
@@ -0,0 +1,131 @@
+//! Process-group tests that run under the scheduler: `join` installs a real
+//! monitor on a live actor, and a real death drives eviction on next contact.
+//! (Pure structural invariants live in the `pg` unit tests.)
+
+use smarm::{channel, members, pick, run, spawn};
+use smarm::{join, leave};
+
+#[test]
+fn join_then_members_lists_a_live_member() {
+    run(|| {
+        let (tx, rx) = channel::<()>();
+        let h = spawn(move || {
+            rx.recv().unwrap();
+        });
+        let pid = h.pid();
+        assert!(join("workers", pid), "first join is new");
+        assert!(!join("workers", pid), "second join is idempotent");
+        assert_eq!(members("workers"), vec![pid]);
+        assert_eq!(pick("workers"), Some(pid));
+        tx.send(()).unwrap();
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn a_dead_actor_vanishes_from_every_group_it_joined() {
+    run(|| {
+        let (tx, rx) = channel::<()>();
+        let h = spawn(move || {
+            rx.recv().unwrap();
+        });
+        let pid = h.pid();
+        join("g1", pid);
+        join("g2", pid);
+        assert_eq!(members("g1"), vec![pid]);
+        assert_eq!(members("g2"), vec![pid]);
+
+        // Release and reap the actor. finalize_actor queues the Down to our
+        // monitors before unparking joiners, so by the time join() returns the
+        // Down is already waiting in the membership channel.
+        tx.send(()).unwrap();
+        h.join().unwrap();
+
+        // Drain-on-contact: touching g1 detects the death and sweeps the pid
+        // out of every group (g2 included), not just g1.
+        assert!(members("g1").is_empty(), "evicted from the touched group");
+        assert!(members("g2").is_empty(), "and swept from the untouched group");
+        assert_eq!(pick("g1"), None);
+    });
+}
+
+#[test]
+fn pick_returns_none_when_the_only_member_is_dead() {
+    run(|| {
+        let (tx, rx) = channel::<()>();
+        let h = spawn(move || {
+            rx.recv().unwrap();
+        });
+        let pid = h.pid();
+        join("pool", pid);
+        tx.send(()).unwrap();
+        h.join().unwrap();
+        assert_eq!(pick("pool"), None);
+        assert!(members("pool").is_empty());
+    });
+}
+
+#[test]
+fn live_members_survive_a_peers_death() {
+    run(|| {
+        let (tx_a, rx_a) = channel::<()>();
+        let (tx_b, rx_b) = channel::<()>();
+        let a = spawn(move || {
+            rx_a.recv().unwrap();
+        });
+        let b = spawn(move || {
+            rx_b.recv().unwrap();
+        });
+        join("svc", a.pid());
+        join("svc", b.pid());
+
+        // Kill a; b is still parked on its channel.
+        tx_a.send(()).unwrap();
+        a.join().unwrap();
+
+        assert_eq!(members("svc"), vec![b.pid()], "only the dead peer is reaped");
+        assert_eq!(pick("svc"), Some(b.pid()));
+
+        tx_b.send(()).unwrap();
+        b.join().unwrap();
+    });
+}
+
+#[test]
+fn leave_drops_a_membership_without_affecting_others() {
+    run(|| {
+        let (tx_a, rx_a) = channel::<()>();
+        let (tx_b, rx_b) = channel::<()>();
+        let a = spawn(move || {
+            rx_a.recv().unwrap();
+        });
+        let b = spawn(move || {
+            rx_b.recv().unwrap();
+        });
+        join("g", a.pid());
+        join("g", b.pid());
+        assert!(leave("g", a.pid()), "a was a member");
+        assert!(!leave("g", a.pid()), "leaving twice finds nothing");
+        assert_eq!(members("g"), vec![b.pid()]);
+
+        tx_a.send(()).unwrap();
+        tx_b.send(()).unwrap();
+        a.join().unwrap();
+        b.join().unwrap();
+    });
+}
+
+#[test]
+fn joining_an_already_dead_pid_is_evicted_on_next_contact() {
+    run(|| {
+        let h = spawn(|| {});
+        let pid = h.pid();
+        h.join().unwrap(); // actor is finalized before we join it to anything
+
+        // monitor() on a gone pid queues a NoProc Down immediately, so the
+        // membership is reaped the next time the group is touched.
+        join("late", pid);
+        assert!(members("late").is_empty(), "dead-at-join member is reaped on read");
+        assert_eq!(pick("late"), None);
+    });
+}