feat: Router actor for pooled message distribution (Cycle 19)

Add Router<M> actor that manages a pool of identical workers and
distributes incoming messages via configurable routing strategies:
RoundRobin, Random, and Broadcast. Workers are monitored and
auto-replaced on failure with meltdown protection.

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 18 COMPLETE
+### Status: Cycle 19 COMPLETE
 
 ## Plan Overview
 1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
@@ -158,6 +158,38 @@
   - `supervisor_on_stop_kills_children` — supervisor shutdown cascades to children
 - **Result**: 138 tests pass (130 behavioral + 7 proptest + 1 doctest), zero warnings, full workspace compiles
 
+### Cycle 19: Router (Actor Pool with Message Routing)
+- **Research**: Cross-framework analysis of actor pool/router patterns:
+  - Erlang: poolboy (checkout/checkin), wpool (transparent forwarding, 6 strategies + custom)
+  - Akka: Router actors (Pool vs Group), 8 strategies (RoundRobin, Random, SmallestMailbox,
+    Balancing, Broadcast, ScatterGather, TailChopping, ConsistentHashing), Resizer for dynamic sizing
+  - Ractor: No built-in router (process groups only)
+  - Actix: SyncArbiter (shared queue, implicit work-stealing)
+  - Kameo: ActorPool (least-connections, auto-replace dead workers)
+  - Key finding: Router-as-actor with transparent forwarding (wpool/Akka style) is the best fit
+  - Decision: user-space actor like Supervisor, reusing monitor + handle_down for worker replacement
+- **Implementation**: `Router<M>` actor in `src/actor.rs`
+  - `RoutingStrategy::RoundRobin` — sequential circular distribution
+  - `RoutingStrategy::Random` — random worker selection via `get_random()`
+  - `RoutingStrategy::Broadcast` — clone message to all live workers
+  - Generic over `M: Message` (same Incoming type as workers) — transparent forwarding
+  - Workers spawned in `on_start`, monitored via `ctx.monitor()`, auto-replaced via `handle_down`
+  - Meltdown protection: `total_restarts > max_restarts` → `ctx.stop_self()`
+  - Cascading shutdown: `on_stop` sends stop signals to all workers
+  - Reuses `ActiveChild` struct from Supervisor (addr + monitor_ref)
+  - Factory: `Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>`
+  - SmallestMailbox deferred: requires runtime stats access not available in user-space
+  - ConsistentHashing deferred: requires hash_fn parameter, can add later as builder method
+- **Tests**: 7 new behavioral tests
+  - `router_round_robin_distributes_across_workers` — 6 msgs to 3 workers, each gets 2
+  - `router_broadcast_sends_to_all_workers` — 1 msg, all 3 workers receive
+  - `router_random_delivers_to_some_worker` — 30 msgs across 3 workers, at least 2 used
+  - `router_replaces_dead_worker` — panicked worker auto-replaced, pool size maintained
+  - `router_meltdown_after_max_restarts` — 3 deaths with max_restarts=2 → router stops
+  - `router_on_stop_kills_workers` — stopping router cascades to all workers
+  - `router_broadcast_multiple_messages_all_received` — 5 msgs × 3 workers = 15 received
+- **Result**: 148 tests pass (140 behavioral + 7 proptest + 1 doctest), zero warnings
+
 ### Cycle 18: OneForAll + RestForOne Supervisor Strategies
 - **Research**: Investigated SmallBox/InlineAny optimization (44% queue throughput improvement)
   but deferred due to unsafe code risk and 32+ call-site changes violating structural constraints.

docs/runtime.md

@@ -239,6 +239,31 @@ restarts the full set in spec order. Already-dead children are handled immediate
 Meltdown: supervisor stops itself when total restarts exceed `max_restarts`.
 Cascading: supervisor stops all children in `on_stop`.
 
+## Router — Actor Pool with Message Routing
+
+The `Router<M>` actor manages a pool of identical workers and distributes
+incoming messages across them. Callers send messages to the router's address,
+and the router forwards them according to the configured strategy.
+
+```
+  let router = Router::new(
+      RoutingStrategy::RoundRobin,
+      5,                                // pool size
+      |ctx| ctx.spawn(MyWorker::new()), // worker factory
+      10,                               // max restarts before meltdown
+  );
+  let router_addr = rt.spawn(router)?;
+  rt.send_to(router_addr, WorkerMsg::DoWork(42))?;
+```
+
+Routing strategies:
+- `RoundRobin`: sequential circular distribution
+- `Random`: random worker selection
+- `Broadcast`: clone message to all workers
+
+Workers are monitored and automatically replaced on failure. Meltdown
+protection stops the router when total restarts exceed `max_restarts`.
+
 ### handle_down Callback
 
 Any actor can override `handle_down` to react to monitored actor deaths

src/actor.rs

@@ -1,4 +1,5 @@
 use std::any::Any;
+use std::marker::PhantomData;
 use std::sync::Arc;
 
 use crate::Error;
@@ -707,3 +708,168 @@ impl ActorInterface for Supervisor {
         }
     }
 }
+
+// ---------------------------------------------------------------------------
+// Router — pool of identical workers with configurable routing strategy
+// ---------------------------------------------------------------------------
+
+/// Strategy for distributing messages across pool workers.
+#[derive(Debug, Clone)]
+pub enum RoutingStrategy {
+    /// Sequential round-robin distribution.
+    RoundRobin,
+    /// Random worker selection.
+    Random,
+    /// Send to all workers (message is cloned to each).
+    Broadcast,
+}
+
+/// A router actor that manages a pool of identical workers and distributes
+/// incoming messages across them according to a [`RoutingStrategy`].
+///
+/// Workers are spawned during `on_start`, monitored for failures, and
+/// automatically replaced to maintain the target pool size. Meltdown
+/// protection stops the router when total restarts exceed `max_restarts`.
+///
+/// # Example
+///
+/// ```ignore
+/// let router = Router::new(
+///     RoutingStrategy::RoundRobin,
+///     5,
+///     |ctx| ctx.spawn(MyWorker::new()),
+///     10,
+/// );
+/// let router_addr = rt.spawn(router)?;
+/// rt.send_to(router_addr, WorkerMessage::DoWork(42))?;
+/// ```
+pub struct Router<M: Message> {
+    strategy: RoutingStrategy,
+    pool_size: usize,
+    factory: Arc<dyn Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync>,
+    workers: Vec<Option<ActiveChild>>,
+    rr_index: usize,
+    total_restarts: u32,
+    max_restarts: u32,
+    _marker: PhantomData<M>,
+}
+
+impl<M: Message> Router<M> {
+    pub fn new(
+        strategy: RoutingStrategy,
+        pool_size: usize,
+        factory: impl Fn(&Ctx) -> Result<ActorAddress, Error> + Send + Sync + 'static,
+        max_restarts: u32,
+    ) -> Self {
+        Self {
+            strategy,
+            pool_size,
+            factory: Arc::new(factory),
+            workers: (0..pool_size).map(|_| None).collect(),
+            rr_index: 0,
+            total_restarts: 0,
+            max_restarts,
+            _marker: PhantomData,
+        }
+    }
+
+    fn start_worker(&mut self, ctx: &Ctx, idx: usize) -> Result<(), Error> {
+        let addr = (self.factory)(ctx)?;
+        let mref = ctx.monitor(addr);
+        self.workers[idx] = Some(ActiveChild {
+            addr,
+            _monitor_ref: mref,
+        });
+        Ok(())
+    }
+
+    fn find_worker_idx(&self, addr: ActorAddress) -> Option<usize> {
+        self.workers
+            .iter()
+            .position(|w| w.as_ref().map_or(false, |ac| ac.addr == addr))
+    }
+
+    fn live_workers(&self) -> Vec<ActorAddress> {
+        self.workers
+            .iter()
+            .filter_map(|w| w.as_ref().map(|ac| ac.addr))
+            .collect()
+    }
+
+    fn select_one(&mut self) -> Option<ActorAddress> {
+        let live = self.live_workers();
+        if live.is_empty() {
+            return None;
+        }
+        match self.strategy {
+            RoutingStrategy::RoundRobin => {
+                let idx = self.rr_index % live.len();
+                self.rr_index = self.rr_index.wrapping_add(1);
+                Some(live[idx])
+            }
+            RoutingStrategy::Random => {
+                let mut buf = [0u8; 8];
+                crate::get_random(&mut buf);
+                let r = u64::from_ne_bytes(buf) as usize;
+                Some(live[r % live.len()])
+            }
+            RoutingStrategy::Broadcast => None, // handled separately
+        }
+    }
+}
+
+impl<M: Message> ActorInterface for Router<M> {
+    type Incoming = M;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: M) {
+        match self.strategy {
+            RoutingStrategy::Broadcast => {
+                let live = self.live_workers();
+                for addr in live {
+                    let _ = ctx.send(addr, msg.clone());
+                }
+            }
+            _ => {
+                if let Some(addr) = self.select_one() {
+                    let _ = ctx.send(addr, msg);
+                }
+            }
+        }
+    }
+
+    fn on_start(&mut self, ctx: &Ctx) {
+        for idx in 0..self.pool_size {
+            if let Err(e) = self.start_worker(ctx, idx) {
+                eprintln!("swactor: router failed to start worker {idx}: {e}");
+            }
+        }
+    }
+
+    fn on_stop(&mut self, ctx: &Ctx) {
+        for child in self.workers.iter().flatten() {
+            let _ = ctx.stop_actor(child.addr);
+        }
+    }
+
+    fn handle_down(&mut self, ctx: &Ctx, down: Down) {
+        let Some(idx) = self.find_worker_idx(down.addr) else {
+            return;
+        };
+        self.workers[idx] = None;
+
+        self.total_restarts += 1;
+        if self.total_restarts > self.max_restarts {
+            eprintln!(
+                "swactor: router reached max restarts ({}), shutting down",
+                self.max_restarts
+            );
+            ctx.stop_self();
+            return;
+        }
+
+        if let Err(e) = self.start_worker(ctx, idx) {
+            eprintln!("swactor: router failed to restart worker {idx}: {e}");
+        }
+    }
+}

tests/runtime_api.rs

@@ -2,8 +2,8 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use swactor::actor::{
-    ActorAddress, ActorInterface, ChildSpec, Down, MonitorRef, RestartPolicy, StopReason,
+    ActorAddress, ActorInterface, ChildSpec, Down, MonitorRef, RestartPolicy, Router,
-    Supervisor, SupervisorStrategy,
+    RoutingStrategy, StopReason, Supervisor, SupervisorStrategy,
 };
 use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};
 
@@ -4230,3 +4230,302 @@ fn supervisor_on_stop_kills_children() {
 
     assert_eq!(rt.stats().workers[0].num_actors, 0);
 }
+
+// ── Router tests ─────────────────────────────────────────────────────────────
+
+#[test]
+fn router_round_robin_distributes_across_workers() {
+    // Given a round-robin router with 3 workers
+    // When we send 6 messages
+    // Then each worker should receive exactly 2 messages
+    let rt = Runtime::new(RuntimeConfig::default());
+    let collected = Arc::new(std::sync::Mutex::new(Vec::new()));
+
+    struct Collector(Arc<std::sync::Mutex<Vec<(ActorAddress, usize)>>>);
+    #[derive(Clone)]
+    struct Work(usize);
+    impl ActorInterface for Collector {
+        type Incoming = Work;
+        type Response = ();
+        fn handle(&mut self, ctx: &Ctx, msg: Work) {
+            self.0.lock().unwrap().push((ctx.self_addr(), msg.0));
+        }
+    }
+
+    let c = collected.clone();
+    let router = Router::<Work>::new(
+        RoutingStrategy::RoundRobin,
+        3,
+        move |ctx| ctx.spawn(Collector(c.clone())),
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick(); // on_start spawns 3 workers
+
+    for i in 0..6 {
+        rt.send_to(router_addr, Work(i)).unwrap();
+    }
+    rt.tick(); // router receives 6 Work messages, forwards to workers
+    rt.tick(); // workers process their messages
+
+    let data = collected.lock().unwrap();
+    assert_eq!(data.len(), 6);
+
+    // Count how many unique workers received messages
+    let mut per_worker = std::collections::HashMap::new();
+    for (addr, _) in data.iter() {
+        *per_worker.entry(*addr).or_insert(0usize) += 1;
+    }
+    // All 3 workers should have received exactly 2 messages each
+    assert_eq!(per_worker.len(), 3);
+    for count in per_worker.values() {
+        assert_eq!(*count, 2);
+    }
+}
+
+#[test]
+fn router_broadcast_sends_to_all_workers() {
+    // Given a broadcast router with 3 workers
+    // When we send 1 message
+    // Then all 3 workers should receive it
+    let rt = Runtime::new(RuntimeConfig::default());
+    let count = Arc::new(AtomicUsize::new(0));
+
+    struct Counter(Arc<AtomicUsize>);
+    #[derive(Clone)]
+    struct Ping;
+    impl ActorInterface for Counter {
+        type Incoming = Ping;
+        type Response = ();
+        fn handle(&mut self, _ctx: &Ctx, _msg: Ping) {
+            self.0.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    let c = count.clone();
+    let router = Router::<Ping>::new(
+        RoutingStrategy::Broadcast,
+        3,
+        move |ctx| ctx.spawn(Counter(c.clone())),
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick(); // on_start spawns workers
+
+    rt.send_to(router_addr, Ping).unwrap();
+    rt.tick(); // router broadcasts
+    rt.tick(); // workers process
+
+    assert_eq!(count.load(Ordering::Relaxed), 3);
+}
+
+#[test]
+fn router_random_delivers_to_some_worker() {
+    // Given a random router with 3 workers
+    // When we send 30 messages
+    // Then at least 2 different workers should have received messages
+    let rt = Runtime::new(RuntimeConfig::default());
+    let collected = Arc::new(std::sync::Mutex::new(Vec::new()));
+
+    struct Collector(Arc<std::sync::Mutex<Vec<ActorAddress>>>);
+    #[derive(Clone)]
+    struct Work;
+    impl ActorInterface for Collector {
+        type Incoming = Work;
+        type Response = ();
+        fn handle(&mut self, ctx: &Ctx, _msg: Work) {
+            self.0.lock().unwrap().push(ctx.self_addr());
+        }
+    }
+
+    let c = collected.clone();
+    let router = Router::<Work>::new(
+        RoutingStrategy::Random,
+        3,
+        move |ctx| ctx.spawn(Collector(c.clone())),
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick();
+
+    for _ in 0..30 {
+        rt.send_to(router_addr, Work).unwrap();
+    }
+    rt.tick();
+    rt.tick();
+
+    let data = collected.lock().unwrap();
+    assert_eq!(data.len(), 30);
+
+    let unique: std::collections::HashSet<_> = data.iter().collect();
+    // With 30 messages across 3 workers, probability of all going to 1 is vanishingly small
+    assert!(unique.len() >= 2, "expected at least 2 workers used, got {}", unique.len());
+}
+
+#[test]
+fn router_replaces_dead_worker() {
+    // Given a router with 3 workers
+    // When one worker panics
+    // Then the router should spawn a replacement and messages continue to be delivered
+    let rt = Runtime::new(RuntimeConfig::default());
+    let spawn_count = Arc::new(AtomicUsize::new(0));
+
+    struct PanicOnFirst {
+        first: bool,
+    }
+    #[derive(Clone)]
+    struct Work;
+    impl ActorInterface for PanicOnFirst {
+        type Incoming = Work;
+        type Response = ();
+        fn handle(&mut self, _ctx: &Ctx, _msg: Work) {
+            if self.first {
+                self.first = false;
+                panic!("first message panic");
+            }
+        }
+    }
+
+    let sc = spawn_count.clone();
+    let router = Router::<Work>::new(
+        RoutingStrategy::RoundRobin,
+        3,
+        move |ctx| {
+            let n = sc.fetch_add(1, Ordering::Relaxed);
+            // Only the first worker panics on its first message
+            ctx.spawn(PanicOnFirst { first: n == 0 })
+        },
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick(); // spawn workers (3 spawned)
+    assert_eq!(spawn_count.load(Ordering::Relaxed), 3);
+
+    // Send a message that will hit worker 0 (round-robin starts at 0)
+    rt.send_to(router_addr, Work).unwrap();
+    rt.tick(); // router forwards to worker 0
+    rt.tick(); // worker 0 panics
+    rt.tick(); // cleanup + Down delivered to router
+    rt.tick(); // router spawns replacement
+    rt.tick(); // replacement starts
+
+    // Should have spawned 4 total (3 original + 1 replacement)
+    assert_eq!(spawn_count.load(Ordering::Relaxed), 4);
+
+    // Verify all 3 slots are live — stats should show router + 3 workers
+    assert_eq!(rt.stats().workers[0].num_actors, 4);
+}
+
+#[test]
+fn router_meltdown_after_max_restarts() {
+    // Given a router with max_restarts=2
+    // When 3 workers die in succession
+    // Then the router should stop itself
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    struct AlwaysPanics;
+    #[derive(Clone)]
+    struct Work;
+    impl ActorInterface for AlwaysPanics {
+        type Incoming = Work;
+        type Response = ();
+        fn handle(&mut self, _ctx: &Ctx, _msg: Work) {
+            panic!("always");
+        }
+    }
+
+    let router = Router::<Work>::new(
+        RoutingStrategy::RoundRobin,
+        1,
+        |ctx| ctx.spawn(AlwaysPanics),
+        2, // max 2 restarts
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick(); // on_start
+
+    // Kill the worker 3 times (> max_restarts=2)
+    for _ in 0..3 {
+        rt.send_to(router_addr, Work).unwrap();
+        for _ in 0..5 {
+            rt.tick();
+        }
+    }
+
+    // After 3 restarts, router should have shut down
+    for _ in 0..5 {
+        rt.tick();
+    }
+    assert_eq!(rt.stats().workers[0].num_actors, 0);
+}
+
+#[test]
+fn router_on_stop_kills_workers() {
+    // Given a running router with 3 workers
+    // When the router is stopped
+    // Then all workers should also be stopped
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    struct Dummy;
+    #[derive(Clone)]
+    struct Work;
+    impl ActorInterface for Dummy {
+        type Incoming = Work;
+        type Response = ();
+        fn handle(&mut self, _ctx: &Ctx, _msg: Work) {}
+    }
+
+    let router = Router::<Work>::new(
+        RoutingStrategy::RoundRobin,
+        3,
+        |ctx| ctx.spawn(Dummy),
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick(); // on_start
+    assert_eq!(rt.stats().workers[0].num_actors, 4); // router + 3 workers
+
+    rt.stop_actor(router_addr).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    assert_eq!(rt.stats().workers[0].num_actors, 0);
+}
+
+#[test]
+fn router_broadcast_multiple_messages_all_received() {
+    // Given a broadcast router
+    // When we send 5 messages to 3 workers
+    // Then total received = 5 * 3 = 15
+    let rt = Runtime::new(RuntimeConfig::default());
+    let total = Arc::new(AtomicUsize::new(0));
+
+    struct Sink(Arc<AtomicUsize>);
+    #[derive(Clone)]
+    struct Tick;
+    impl ActorInterface for Sink {
+        type Incoming = Tick;
+        type Response = ();
+        fn handle(&mut self, _ctx: &Ctx, _msg: Tick) {
+            self.0.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    let t = total.clone();
+    let router = Router::<Tick>::new(
+        RoutingStrategy::Broadcast,
+        3,
+        move |ctx| ctx.spawn(Sink(t.clone())),
+        10,
+    );
+    let router_addr = rt.spawn(router).unwrap();
+    rt.tick();
+
+    for _ in 0..5 {
+        rt.send_to(router_addr, Tick).unwrap();
+    }
+    rt.tick(); // router broadcasts
+    rt.tick(); // workers process
+
+    assert_eq!(total.load(Ordering::Relaxed), 15);
+}