benches/worker_benchmarks.rs

@@ -3,32 +3,6 @@ use std::collections::VecDeque;
 use criterion::{
     criterion_group, criterion_main, BenchmarkId, Criterion, Throughput,
 };
-use swactor::worker::drain_count;
-
-// ---------------------------------------------------------------------------
-// drain_count O(1) verification
-// ---------------------------------------------------------------------------
-
-fn bench_drain_count(c: &mut Criterion) {
-    let mut group = c.benchmark_group("drain_count");
-
-    // Below waterlevel
-    group.bench_function("below", |b| {
-        b.iter(|| std::hint::black_box(drain_count(50, 100)));
-    });
-
-    // At waterlevel
-    group.bench_function("at", |b| {
-        b.iter(|| std::hint::black_box(drain_count(100, 100)));
-    });
-
-    // Above waterlevel
-    group.bench_function("above", |b| {
-        b.iter(|| std::hint::black_box(drain_count(500, 100)));
-    });
-
-    group.finish();
-}
 
 // ---------------------------------------------------------------------------
 // VecDeque push throughput (mirrors old mailbox_push)
@@ -79,43 +53,9 @@ fn vecdeque_pop(c: &mut Criterion) {
     group.finish();
 }
 
-// ---------------------------------------------------------------------------
-// Simulated actor tick: drain_count + pop N from VecDeque
-// ---------------------------------------------------------------------------
-
-fn simulated_actor_tick(c: &mut Criterion) {
-    let mut group = c.benchmark_group("simulated_actor_tick");
-
-    for (wl, fill) in [(10, 5), (10, 10), (10, 50), (100, 200)] {
-        let param = format!("wl={wl},fill={fill}");
-        group.bench_function(BenchmarkId::from_parameter(&param), |b| {
-            b.iter_batched(
-                || {
-                    let mut q: VecDeque<u64> = VecDeque::new();
-                    for i in 0..fill {
-                        q.push_back(i as u64);
-                    }
-                    q
-                },
-                |mut q| {
-                    let n = drain_count(q.len(), wl);
-                    for _ in 0..n {
-                        std::hint::black_box(q.pop_front());
-                    }
-                },
-                criterion::BatchSize::SmallInput,
-            );
-        });
-    }
-
-    group.finish();
-}
-
 criterion_group!(
     benches,
-    bench_drain_count,
     vecdeque_push,
     vecdeque_pop,
-    simulated_actor_tick,
 );
 criterion_main!(benches);

crates/swactor-python/src/lib.rs

@@ -227,8 +227,6 @@ pub struct PyRuntimeConfig {
     #[pyo3(get, set)]
     actor_max_messages: usize,
     #[pyo3(get, set)]
-    mailbox_waterlevel: usize,
-    #[pyo3(get, set)]
     spin_threshold: u32,
     #[pyo3(get, set)]
     yield_threshold: u32,
@@ -246,7 +244,6 @@ impl PyRuntimeConfig {
         num_threads = 1,
         max_actors = 1_000,
         actor_max_messages = 1_000,
-        mailbox_waterlevel = 10,
         spin_threshold = 64,
         yield_threshold = 256,
         sleep_increment_us = 50,
@@ -256,7 +253,6 @@ impl PyRuntimeConfig {
         num_threads: usize,
         max_actors: usize,
         actor_max_messages: usize,
-        mailbox_waterlevel: usize,
         spin_threshold: u32,
         yield_threshold: u32,
         sleep_increment_us: u64,
@@ -266,7 +262,6 @@ impl PyRuntimeConfig {
             num_threads,
             max_actors,
             actor_max_messages,
-            mailbox_waterlevel,
             spin_threshold,
             yield_threshold,
             sleep_increment_us,
@@ -281,7 +276,6 @@ impl From<PyRuntimeConfig> for RuntimeConfig {
             num_threads: py.num_threads,
             max_actors: py.max_actors,
             actor_max_messages: py.actor_max_messages,
-            mailbox_waterlevel: py.mailbox_waterlevel,
             backoff_policy: BackoffPolicy {
                 spin_threshold: py.spin_threshold,
                 yield_threshold: py.yield_threshold,

fuzz/fuzz_targets/fuzz_runtime.rs

@@ -210,7 +210,6 @@ enum RawAction {
 #[derive(Debug, Arbitrary)]
 struct FuzzInput {
     max_actors: u8,
-    mailbox_waterlevel: u8,
     scenarios: Vec<Scenario>,
 }
 
@@ -773,7 +772,6 @@ impl fmt::Debug for FuzzState {
 
 fuzz_target!(|input: FuzzInput| {
     let max_actors = (input.max_actors as usize).max(1).min(200);
-    let mailbox_waterlevel = (input.mailbox_waterlevel as usize).max(1).min(50);
 
     let interval = log_interval();
     let run = if interval > 0 {
@@ -783,7 +781,6 @@ fuzz_target!(|input: FuzzInput| {
 
     let config = RuntimeConfig {
         max_actors,
-        mailbox_waterlevel,
         num_threads: 1,
         ..Default::default()
     };
@@ -806,7 +803,7 @@ fuzz_target!(|input: FuzzInput| {
         let depth: usize = stats.workers.iter().map(|w| w.mailbox_depth).sum();
         let processed: u64 = stats.workers.iter().map(|w| w.messages_processed).sum();
         eprintln!("\
-\n=== Run #{run} | cap={max_actors} waterlevel={mailbox_waterlevel} ===
+\n=== Run #{run} | cap={max_actors} ===
 {trace}\
 --- {spawned} spawned, {sent} sent, {recv} received, \
 {alive} alive, {depth} queued, {processed} processed ---\n",

src/actor.rs

@@ -64,7 +64,6 @@ where
 pub trait ContextInner {
     fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
     fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>) -> Result<(), Error>;
-    fn mailbox_waterlevel(&self) -> usize;
 }
 
 /// Actor syscall interface — passed to `ActorInterface::handle()`.

src/channel.rs

@@ -17,6 +17,10 @@ impl<T> HybridChannel<T> {
     }
 
     pub fn push(&self, value: T) -> Result<(), T> {
+        if !self.overflow.is_empty() {
+            self.overflow.push(value);
+            return Ok(());
+        }
         match self.ring.push(value) {
             Ok(()) => Ok(()),
             Err(v) => {

src/config.rs

@@ -28,7 +28,6 @@ pub struct RuntimeConfig {
     pub max_actors: usize,
     pub actor_max_messages: usize,
     pub num_threads: usize,
-    pub mailbox_waterlevel: usize,
     pub backoff_policy: BackoffPolicy,
 }
 
@@ -40,15 +39,12 @@ const DEFAULT_MAX_ACTORS: usize = 1_000;
 /// 1_000 * 16kB = 16MB
 const DEFAULT_ACTOR_MAX_MESSAGES: usize = 1_000;
 
-const DEFAULT_MAILBOX_WATERLEVEL: usize = 10;
-
 impl Default for RuntimeConfig {
     fn default() -> Self {
         Self {
             max_actors: DEFAULT_MAX_ACTORS,
             actor_max_messages: DEFAULT_ACTOR_MAX_MESSAGES,
             num_threads: 1,
-            mailbox_waterlevel: DEFAULT_MAILBOX_WATERLEVEL,
             backoff_policy: BackoffPolicy::default(),
         }
     }

src/crate_test/mod.rs

@@ -253,13 +253,8 @@ fn wrong_type_silently_dropped() {
 }
 
 #[test]
-fn backpressure_drains_half() {
-    let config = RuntimeConfig {
-        mailbox_waterlevel: 4,
-        ..Default::default()
-    };
-
-    run_with(config, &[
+fn all_messages_drain_in_one_tick() {
+    run(&[
         Step::Spawn(1),
         Step::Tick,
 
@@ -268,15 +263,7 @@ fn backpressure_drains_half() {
         Step::Send(1, 4), Step::Send(1, 5), Step::Send(1, 6), Step::Send(1, 7),
         Step::Send(1, 8), Step::Send(1, 9),
 
-        // Tick 1: 10 in mailbox, drain_count(10, 4) = 5
-        Step::Tick,
-        Step::Expect { pool_len: 1, depth: 5, processed: 5 },
-
-        // Tick 2: 5 remaining, drain_count(5, 4) = 2
-        Step::Tick,
-        Step::Expect { pool_len: 1, depth: 3, processed: 7 },
-
-        // Tick 3: 3 remaining, drain_count(3, 4) = 3 (below waterlevel → all)
+        // All 10 processed in a single tick
         Step::Tick,
         Step::Expect { pool_len: 1, depth: 0, processed: 10 },
         Step::ExpectHandled(1, 10),

src/runtime.rs

@@ -66,15 +66,12 @@ pub struct Runtime {
     placement: Placement,
     is_running: AtomicBool,
     worker_stats: Vec<Arc<WorkerStats>>,
-    /// Single-threaded mode: worker stored inline
-    single_worker: Option<RefCell<Worker>>,
-    /// Multi-threaded mode: workers waiting to be assigned to threads by run()
-    pending_workers: Option<Vec<Worker>>,
+    /// Workers available for tick(). run() drains this and moves workers to threads.
+    tick_workers: RefCell<Vec<Worker>>,
 }
 
-// Safety: RefCell<Worker> is only accessed from the thread that owns the Runtime
-// in single-threaded mode. In multi-threaded mode, single_worker is None and
-// pending_workers is consumed by run() before Arc sharing.
+// Safety: RefCell<Vec<Worker>> is only accessed from the owning thread via tick().
+// After run() the RefCell is empty and not accessed by worker threads.
 unsafe impl Sync for Runtime {}
 
 impl Runtime {
@@ -111,35 +108,16 @@ impl Runtime {
             workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats));
         }
 
-        if config.num_threads < 2 {
-            // Single-threaded: store one worker inline
-            let worker = workers.remove(0);
-            Self {
-                config,
-                address_map,
-                inbox_registry,
-                transfer_txs,
-                spawn_txs,
-                placement,
-                is_running: AtomicBool::new(false),
-                worker_stats,
-                single_worker: Some(RefCell::new(worker)),
-                pending_workers: None,
-            }
-        } else {
-            // Multi-threaded: stash workers for run()
-            Self {
-                config,
-                address_map,
-                inbox_registry,
-                transfer_txs,
-                spawn_txs,
-                placement,
-                is_running: AtomicBool::new(false),
-                worker_stats,
-                single_worker: None,
-                pending_workers: Some(workers),
-            }
+        Self {
+            config,
+            address_map,
+            inbox_registry,
+            transfer_txs,
+            spawn_txs,
+            placement,
+            is_running: AtomicBool::new(false),
+            worker_stats,
+            tick_workers: RefCell::new(workers),
         }
     }
 
@@ -179,17 +157,23 @@ impl Runtime {
     }
 
     /// Drive one tick of the single-threaded worker.
+    ///
+    /// Panics if called on a multi-threaded runtime — use `run()` instead.
     pub fn tick(&self) {
-        if let Some(ref worker) = self.single_worker {
-            let tc = TickContext {
-                address_map: &self.address_map,
-                transfer_txs: &self.transfer_txs,
-                spawn_txs: &self.spawn_txs,
-                placement: &self.placement,
-                inbox_registry: &self.inbox_registry,
-                config: &self.config,
-            };
-            worker.borrow_mut().tick_once(&tc);
+        assert!(
+            self.config.num_threads < 2,
+            "tick() is only valid for single-threaded runtimes; use run() for multi-threaded"
+        );
+        let tc = TickContext {
+            address_map: &self.address_map,
+            transfer_txs: &self.transfer_txs,
+            spawn_txs: &self.spawn_txs,
+            placement: &self.placement,
+            inbox_registry: &self.inbox_registry,
+            config: &self.config,
+        };
+        for worker in self.tick_workers.borrow_mut().iter_mut() {
+            worker.tick_once(&tc);
         }
     }
 
@@ -198,17 +182,10 @@ impl Runtime {
     ///
     /// Works in both single-threaded and multi-threaded configurations.
     /// In single-threaded mode, one background thread is spawned.
-    pub fn run(mut self) -> Result<RuntimeHandle, Error> {
+    pub fn run(self) -> Result<RuntimeHandle, Error> {
         self.is_running.store(true, Ordering::Release);
 
-        let mut workers: Vec<Worker> = Vec::new();
-
-        if let Some(w) = self.single_worker.take() {
-            workers.push(w.into_inner());
-        }
-        if let Some(ws) = self.pending_workers.take() {
-            workers.extend(ws);
-        }
+        let workers: Vec<Worker> = self.tick_workers.replace(Vec::new());
 
         let rt = Arc::new(self);
         let mut handles: Vec<JoinHandle<()>> = Vec::with_capacity(workers.len());
@@ -294,8 +271,4 @@ impl ContextInner for Runtime {
             .try_send((addr, actor))
             .map_err(|_| Error::from("Spawn queue full"))
     }
-
-    fn mailbox_waterlevel(&self) -> usize {
-        self.config.mailbox_waterlevel
-    }
 }

src/worker.rs

@@ -71,7 +71,14 @@ impl Worker {
             }
         }
 
-        // 4. Drain pending_local buffer → deliver to local actors
+        // 4. Drain spawn queue again — actors spawned during step 3
+        //    must be in the pool before pending_local delivery.
+        while let Some((addr, actor)) = self.spawn_rx.try_recv() {
+            self.pool.insert(addr, actor);
+            did_work = true;
+        }
+
+        // 5. Drain pending_local buffer → deliver to local actors
         let pending = pending_local.into_inner();
         if !pending.is_empty() {
             did_work = true;
@@ -80,7 +87,7 @@ impl Worker {
             self.pool.deliver(&addr, msg);
         }
 
-        // 5. Publish stats
+        // 6. Publish stats
         self.stats.num_actors.store(self.pool.len(), Ordering::Relaxed);
         self.stats.total_mailbox_depth.store(self.pool.total_mailbox_depth(), Ordering::Relaxed);
         self.stats.messages_processed.fetch_add(processed as u64, Ordering::Relaxed);
@@ -152,20 +159,6 @@ impl ContextInner for WorkerContext<'_> {
             .map_err(|_| Error::from("Spawn queue full"))
     }
 
-    fn mailbox_waterlevel(&self) -> usize {
-        self.tc.config.mailbox_waterlevel
-    }
-}
-
-/// How many messages to process this tick:
-/// - `len < waterlevel` → process all (`len`)
-/// - `len >= waterlevel` → process half (`len >> 1`)
-pub fn drain_count(len: usize, waterlevel: usize) -> usize {
-    if len < waterlevel {
-        len
-    } else {
-        len >> 1
-    }
 }
 
 struct ActorSlot {
@@ -211,16 +204,15 @@ impl ActorPool {
     pub fn tick_all(&mut self, inner: &dyn ContextInner) -> usize {
         let mut count = 0;
         for (&addr, slot) in self.actors.iter_mut() {
-            let len = slot.mailbox.len();
-            let n = drain_count(len, inner.mailbox_waterlevel());
-            if n > 0 {
-                let ctx = Ctx::new(inner, addr);
-                for _ in 0..n {
-                    if let Some(msg) = slot.mailbox.pop_front() {
-                        slot.actor.handle_any(&ctx, msg);
-                        count += 1;
-                    }
+            let ctx = Ctx::new(inner, addr);
+            while let Some(msg) = slot.mailbox.pop_front() {
+                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                    slot.actor.handle_any(&ctx, msg);
+                }));
+                if result.is_err() {
+                    eprintln!("swactor: actor {addr} panicked in handler");
                 }
+                count += 1;
             }
         }
         count

tests/runtime_api.rs

@@ -156,9 +156,314 @@ fn send_to_unknown_address_fails() {
 
 #[test]
 fn actor_spawns_child_and_delegates() {
-    // Uses 2 workers so parent and child land on different workers,
-    // avoiding the single-worker timing issue where pending_local
-    // delivery precedes spawn-queue draining.
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
+
+    let parent = rt.spawn(DelegateActor).expect("spawn parent");
+    let inbox: Inbox<DoubleResponse> = rt.new_inbox().unwrap();
+
+    rt.send_to(
+        parent,
+        DelegateRequest {
+            value: 7,
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
+
+    for _ in 0..10 {
+        rt.tick();
+        if let Some(resp) = inbox.try_recv() {
+            assert_eq!(resp, DoubleResponse(14));
+            return;
+        }
+    }
+
+    panic!("Did not receive DoubleResponse");
+}
+
+#[test]
+fn multiple_actors_independent_mailboxes() {
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    let inbox_a: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+    let inbox_b: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+    let inbox_c: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+
+    let actor_a = rt.spawn(EchoActor).unwrap();
+    let actor_b = rt.spawn(EchoActor).unwrap();
+    let actor_c = rt.spawn(EchoActor).unwrap();
+
+    rt.send_to(actor_a, EchoMessage { payload: 10, reply_to: *inbox_a.addr() }).unwrap();
+    rt.send_to(actor_b, EchoMessage { payload: 20, reply_to: *inbox_b.addr() }).unwrap();
+    rt.send_to(actor_c, EchoMessage { payload: 30, reply_to: *inbox_c.addr() }).unwrap();
+
+    for _ in 0..10 {
+        rt.tick();
+    }
+
+    assert_eq!(inbox_a.try_recv(), Some(EchoResponse(10)));
+    assert_eq!(inbox_b.try_recv(), Some(EchoResponse(20)));
+    assert_eq!(inbox_c.try_recv(), Some(EchoResponse(30)));
+    // No cross-contamination
+    assert_eq!(inbox_a.try_recv(), None);
+    assert_eq!(inbox_b.try_recv(), None);
+    assert_eq!(inbox_c.try_recv(), None);
+}
+
+// ---------------------------------------------------------------------------
+// Additional fixtures for spawn/delegate tests
+// ---------------------------------------------------------------------------
+
+/// Three-deep chain: Parent → Child → Grandchild → reply_to
+struct ChainActor {
+    depth: usize,
+}
+
+#[derive(Clone)]
+struct ChainRequest {
+    remaining: usize,
+    reply_to: ActorAddress,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+struct ChainDone(usize);
+
+impl ActorInterface for ChainActor {
+    type Incoming = ChainRequest;
+    type Response = ChainDone;
+
+    fn handle(&mut self, ctx: &Ctx, msg: ChainRequest) {
+        if msg.remaining == 0 {
+            let _ = ctx.send(msg.reply_to, ChainDone(self.depth));
+        } else {
+            let child = ctx
+                .spawn(ChainActor {
+                    depth: self.depth + 1,
+                })
+                .expect("spawn chain child");
+            let _ = ctx.send(
+                child,
+                ChainRequest {
+                    remaining: msg.remaining - 1,
+                    reply_to: msg.reply_to,
+                },
+            );
+        }
+    }
+}
+
+/// Spawns N children and sends work to each. Each child replies to reply_to.
+struct FanOutActor;
+
+#[derive(Clone)]
+struct FanOutRequest {
+    count: usize,
+    reply_to: ActorAddress,
+}
+
+impl ActorInterface for FanOutActor {
+    type Incoming = FanOutRequest;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: FanOutRequest) {
+        for i in 0..msg.count {
+            let child = ctx.spawn(DoubleActor).expect("spawn fan-out child");
+            let _ = ctx.send(
+                child,
+                DoubleRequest {
+                    value: i + 1,
+                    reply_to: msg.reply_to,
+                },
+            );
+        }
+    }
+}
+
+/// Actor A: spawns a child, sends work to it, and also forwards the child's
+/// address to a "buddy" so the buddy can send to the child too.
+struct SpawnAndBroadcastActor;
+
+#[derive(Clone)]
+struct SpawnAndBroadcastRequest {
+    buddy: ActorAddress,
+    reply_to: ActorAddress,
+}
+
+/// Sent from A to buddy B, carrying the child's address.
+#[derive(Clone)]
+struct ForwardToChild {
+    child: ActorAddress,
+    reply_to: ActorAddress,
+}
+
+impl ActorInterface for SpawnAndBroadcastActor {
+    type Incoming = SpawnAndBroadcastRequest;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: SpawnAndBroadcastRequest) {
+        let child = ctx.spawn(DoubleActor).expect("spawn child");
+        // Send work from self
+        let _ = ctx.send(
+            child,
+            DoubleRequest {
+                value: 10,
+                reply_to: msg.reply_to,
+            },
+        );
+        // Tell buddy about the child
+        let _ = ctx.send(
+            msg.buddy,
+            ForwardToChild {
+                child,
+                reply_to: msg.reply_to,
+            },
+        );
+    }
+}
+
+/// Buddy actor B: receives a child address and sends work to it.
+struct BuddyActor;
+
+impl ActorInterface for BuddyActor {
+    type Incoming = ForwardToChild;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: ForwardToChild) {
+        let _ = ctx.send(
+            msg.child,
+            DoubleRequest {
+                value: 20,
+                reply_to: msg.reply_to,
+            },
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Bug-fix regression tests
+// ---------------------------------------------------------------------------
+
+#[test]
+fn inbox_preserves_fifo_across_overflow() {
+    let rt = Runtime::new(RuntimeConfig {
+        actor_max_messages: 2,
+        ..Default::default()
+    });
+
+    let inbox: Inbox<u64> = rt.new_inbox().unwrap();
+    let addr = *inbox.addr();
+
+    // Push 4 items: ring gets [1,2], overflow gets [3,4].
+    for v in 1..=4u64 {
+        rt.send_to(addr, v).unwrap();
+    }
+
+    // Drain the ring.
+    assert_eq!(inbox.try_recv(), Some(1));
+    assert_eq!(inbox.try_recv(), Some(2));
+
+    // Push a new item — must land behind 3 and 4 in overflow.
+    rt.send_to(addr, 5u64).unwrap();
+
+    let rest: Vec<u64> = std::iter::from_fn(|| inbox.try_recv()).collect();
+    assert_eq!(rest, vec![3, 4, 5]);
+}
+
+#[test]
+fn delegate_works_on_single_worker() {
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
+
+    let parent = rt.spawn(DelegateActor).expect("spawn parent");
+    let inbox: Inbox<DoubleResponse> = rt.new_inbox().unwrap();
+
+    rt.send_to(
+        parent,
+        DelegateRequest {
+            value: 5,
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
+
+    for _ in 0..10 {
+        rt.tick();
+        if let Some(resp) = inbox.try_recv() {
+            assert_eq!(resp, DoubleResponse(10));
+            return;
+        }
+    }
+
+    panic!("Did not receive DoubleResponse on single worker");
+}
+
+#[test]
+fn spawn_chain_three_deep() {
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
+
+    let root = rt.spawn(ChainActor { depth: 0 }).expect("spawn root");
+    let inbox: Inbox<ChainDone> = rt.new_inbox().unwrap();
+
+    rt.send_to(
+        root,
+        ChainRequest {
+            remaining: 2,
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
+
+    for _ in 0..50 {
+        rt.tick();
+        if let Some(resp) = inbox.try_recv() {
+            assert_eq!(resp, ChainDone(2));
+            return;
+        }
+    }
+
+    panic!("Did not receive ChainDone from grandchild");
+}
+
+#[test]
+fn spawn_fan_out() {
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
+
+    let fan = rt.spawn(FanOutActor).expect("spawn fan-out");
+    let inbox: Inbox<DoubleResponse> = rt.new_inbox().unwrap();
+
+    rt.send_to(
+        fan,
+        FanOutRequest {
+            count: 5,
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
+
+    for _ in 0..50 {
+        rt.tick();
+    }
+
+    let mut results: Vec<usize> = std::iter::from_fn(|| inbox.try_recv())
+        .map(|r| r.0)
+        .collect();
+    results.sort();
+    assert_eq!(results, vec![2, 4, 6, 8, 10]);
+}
+
+#[test]
+fn delegate_works_cross_worker() {
     let config = RuntimeConfig {
         num_threads: 2,
         ..Default::default()
@@ -196,34 +501,40 @@ fn actor_spawns_child_and_delegates() {
 }
 
 #[test]
-fn multiple_actors_independent_mailboxes() {
-    let rt = Runtime::new(RuntimeConfig::default());
+fn multiple_senders_to_new_child() {
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
 
-    let inbox_a: Inbox<EchoResponse> = rt.new_inbox().unwrap();
-    let inbox_b: Inbox<EchoResponse> = rt.new_inbox().unwrap();
-    let inbox_c: Inbox<EchoResponse> = rt.new_inbox().unwrap();
+    let buddy = rt.spawn(BuddyActor).expect("spawn buddy");
+    let parent = rt.spawn(SpawnAndBroadcastActor).expect("spawn parent");
+    let inbox: Inbox<DoubleResponse> = rt.new_inbox().unwrap();
 
-    let actor_a = rt.spawn(EchoActor).unwrap();
-    let actor_b = rt.spawn(EchoActor).unwrap();
-    let actor_c = rt.spawn(EchoActor).unwrap();
+    rt.send_to(
+        parent,
+        SpawnAndBroadcastRequest {
+            buddy,
+            reply_to: *inbox.addr(),
+        },
+    )
+    .unwrap();
 
-    rt.send_to(actor_a, EchoMessage { payload: 10, reply_to: *inbox_a.addr() }).unwrap();
-    rt.send_to(actor_b, EchoMessage { payload: 20, reply_to: *inbox_b.addr() }).unwrap();
-    rt.send_to(actor_c, EchoMessage { payload: 30, reply_to: *inbox_c.addr() }).unwrap();
-
-    for _ in 0..10 {
+    for _ in 0..50 {
         rt.tick();
     }
 
-    assert_eq!(inbox_a.try_recv(), Some(EchoResponse(10)));
-    assert_eq!(inbox_b.try_recv(), Some(EchoResponse(20)));
-    assert_eq!(inbox_c.try_recv(), Some(EchoResponse(30)));
-    // No cross-contamination
-    assert_eq!(inbox_a.try_recv(), None);
-    assert_eq!(inbox_b.try_recv(), None);
-    assert_eq!(inbox_c.try_recv(), None);
+    let mut results: Vec<usize> = std::iter::from_fn(|| inbox.try_recv())
+        .map(|r| r.0)
+        .collect();
+    results.sort();
+    assert_eq!(results, vec![20, 40]);
 }
 
+// ---------------------------------------------------------------------------
+// Distribution tests
+// ---------------------------------------------------------------------------
+
 #[test]
 fn round_robin_distributes_across_workers() {
     let config = RuntimeConfig {

tests/test_python.py

@@ -38,7 +38,6 @@ class TestRuntimeConfig(unittest.TestCase):
         self.assertEqual(cfg.num_threads, 1)
         self.assertEqual(cfg.max_actors, 1000)
         self.assertEqual(cfg.actor_max_messages, 1000)
-        self.assertEqual(cfg.mailbox_waterlevel, 10)
         self.assertEqual(cfg.spin_threshold, 64)
         self.assertEqual(cfg.yield_threshold, 256)
         self.assertEqual(cfg.sleep_increment_us, 50)