refactor: better worker thread tests

Show basic information from the runtime, such as number of actors, worker threads, etc.

Cargo.toml

@@ -25,3 +25,7 @@ criterion = { version = "0.5", features = ["html_reports"] }
 [[bench]]
 name = "runtime_benchmarks"
 harness = false
+
+[[bench]]
+name = "worker_benchmarks"
+harness = false

benches/worker_benchmarks.rs

@@ -0,0 +1,153 @@
+use criterion::{
+    criterion_group, criterion_main, BenchmarkId, Criterion, Throughput,
+};
+use swactor::worker::Mailbox;
+
+// ---------------------------------------------------------------------------
+// Push throughput
+// ---------------------------------------------------------------------------
+
+fn mailbox_push(c: &mut Criterion) {
+    let mut group = c.benchmark_group("mailbox_push");
+    for n in [100, 1_000, 10_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter(|| {
+                let mut mb: Mailbox<u64> = Mailbox::new(n);
+                for i in 0..n {
+                    mb.push(i as u64);
+                }
+            });
+        });
+    }
+    group.finish();
+}
+
+// ---------------------------------------------------------------------------
+// Pop throughput
+// ---------------------------------------------------------------------------
+
+fn mailbox_pop(c: &mut Criterion) {
+    let mut group = c.benchmark_group("mailbox_pop");
+    for n in [100, 1_000, 10_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter_batched(
+                || {
+                    let mut mb: Mailbox<u64> = Mailbox::new(n);
+                    for i in 0..n {
+                        mb.push(i as u64);
+                    }
+                    mb
+                },
+                |mut mb| {
+                    for _ in 0..n {
+                        std::hint::black_box(mb.pop());
+                    }
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+    group.finish();
+}
+
+// ---------------------------------------------------------------------------
+// Interleaved push+pop
+// ---------------------------------------------------------------------------
+
+fn mailbox_interleaved(c: &mut Criterion) {
+    let mut group = c.benchmark_group("mailbox_interleaved");
+    for n in [100, 1_000, 10_000] {
+        group.throughput(Throughput::Elements(n as u64 * 2));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter(|| {
+                let mut mb: Mailbox<u64> = Mailbox::new(n);
+                for i in 0..n {
+                    mb.push(i as u64);
+                    std::hint::black_box(mb.pop());
+                }
+            });
+        });
+    }
+    group.finish();
+}
+
+// ---------------------------------------------------------------------------
+// drain_count O(1) verification
+// ---------------------------------------------------------------------------
+
+fn mailbox_drain_count(c: &mut Criterion) {
+    let mut group = c.benchmark_group("mailbox_drain_count");
+
+    // Below waterlevel
+    group.bench_function("below", |b| {
+        let mut mb: Mailbox<u64> = Mailbox::new(100);
+        for i in 0..50 {
+            mb.push(i);
+        }
+        b.iter(|| std::hint::black_box(mb.drain_count()));
+    });
+
+    // At waterlevel
+    group.bench_function("at", |b| {
+        let mut mb: Mailbox<u64> = Mailbox::new(100);
+        for i in 0..100 {
+            mb.push(i);
+        }
+        b.iter(|| std::hint::black_box(mb.drain_count()));
+    });
+
+    // Above waterlevel
+    group.bench_function("above", |b| {
+        let mut mb: Mailbox<u64> = Mailbox::new(100);
+        for i in 0..500 {
+            mb.push(i);
+        }
+        b.iter(|| std::hint::black_box(mb.drain_count()));
+    });
+
+    group.finish();
+}
+
+// ---------------------------------------------------------------------------
+// Simulated actor tick: drain_count + pop N
+// ---------------------------------------------------------------------------
+
+fn mailbox_actor_tick(c: &mut Criterion) {
+    let mut group = c.benchmark_group("mailbox_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 mb: Mailbox<u64> = Mailbox::new(wl);
+                    for i in 0..fill {
+                        mb.push(i as u64);
+                    }
+                    mb
+                },
+                |mut mb| {
+                    let n = mb.drain_count();
+                    for _ in 0..n {
+                        std::hint::black_box(mb.pop());
+                    }
+                },
+                criterion::BatchSize::SmallInput,
+            );
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(
+    benches,
+    mailbox_push,
+    mailbox_pop,
+    mailbox_interleaved,
+    mailbox_drain_count,
+    mailbox_actor_tick,
+);
+criterion_main!(benches);

docs/worker-thread.md

@@ -0,0 +1,550 @@
+# Worker Thread Architecture
+
+## Structure
+
+```
+┌─ Worker ───────────────────────────────────────────────────────────────┐
+│                                                                        │
+│  id: WorkerId                                                          │
+│                                                                        │
+│  ┌─ spawn_rx ─────────────────────┐  ┌─ transfer_rx ──────────────────┐│
+│  │ Receiver<(Addr, Box<AnyActor>)>│  │ Receiver<Envelope>             ││
+│  │                                │  │                                ││
+│  │ from: Runtime.spawn()          │  │ from: other workers, Runtime   ││
+│  │       ctx.spawn()              │  │       ctx.send()               ││
+│  └────────────────────────────────┘  └────────────────────────────────┘│
+│                                                                        │
+│  ┌─ ActorPool ──────────────────────────────────────────────────────┐  │
+│  │                                                                  │  │
+│  │  actors: HashMap<ActorAddress, ActorSlot>                        │  │
+│  │                                                                  │  │
+│  │  ┌─ ActorSlot [addr_0] ──────────────────────────────────────┐   │  │
+│  │  │                                                           │   │  │
+│  │  │  ┌─ mailbox ──────────────────────────────────────────┐   │   │  │
+│  │  │  │ VecDeque<Box<dyn Any + Send>>                      │   │   │  │
+│  │  │  │                                                    │   │   │  │
+│  │  │  │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐                    │   │   │  │
+│  │  │  │ │ msg │ │ msg │ │ msg │ │ ... │  <- push_back      │   │   │  │
+│  │  │  │ └─────┘ └─────┘ └─────┘ └─────┘                    │   │   │  │
+│  │  │  │ pop_front ->                 untyped; Box<Any>     │   │   │  │
+│  │  │  └────────────────────────────────────────────────────┘   │   │  │
+│  │  │                                                           │   │  │
+│  │  │  ┌─ actor ────────────────────────────────────────────┐   │   │  │
+│  │  │  │ Box<dyn AnyActor>                                  │   │   │  │
+│  │  │  │                                                    │   │   │  │
+│  │  │  │ wraps Actor<A>(A) where A: ActorInterface          │   │   │  │
+│  │  │  │                                                    │   │   │  │
+│  │  │  │ handle_any(ctx, msg):                              │   │   │  │
+│  │  │  │   downcast Box<Any> -> A::Incoming                 │   │   │  │
+│  │  │  │   ok  -> A.handle(ctx, typed_msg)                  │   │   │  │
+│  │  │  │   err -> silently drop                             │   │   │  │
+│  │  │  └────────────────────────────────────────────────────┘   │   │  │
+│  │  │                                                           │   │  │
+│  │  └───────────────────────────────────────────────────────────┘   │  │
+│  │                                                                  │  │
+│  │  ┌─ ActorSlot [addr_1] ──────────────────────────────────────┐   │  │
+│  │  │ ...                                                       │   │  │
+│  │  └───────────────────────────────────────────────────────────┘   │  │
+│  │                                                                  │  │
+│  └──────────────────────────────────────────────────────────────────┘  │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+## Shared State (borrowed via TickContext)
+
+Lives on `Arc<Runtime>`, shared read-only across all worker threads.
+
+```
+┌─ TickContext<'a> ──────────────────────────────────────────────────────┐
+│                                                                        │
+│  address_map:    &AddressMap     -- ActorAddress -> WorkerId lookup    │
+│  transfer_txs:   &[Sender]      -- one Sender per worker (cross-send)  │
+│  spawn_txs:      &[Sender]      -- one Sender per worker (spawn reqs)  │
+│  placement:      &Placement     -- round-robin next-worker picker      │
+│  inbox_registry: &InboxRegistry -- external Inbox<M> receivers         │
+│  config:         &RuntimeConfig -- waterlevel, backoff params, etc.    │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+## Run Loop
+
+```
+┌─ Worker::run ──────────────────────────────────────────────────────────┐
+│                                                                        │
+│  ┌────────────────────────────────────┐                                │
+│  │       is_running.load() ?          │                                │
+│  └──────────┬─────────────────────────┘                                │
+│         yes │                                                          │
+│             v                                                          │
+│  ┌────────────────────────────────────┐                                │
+│  │        tick_once(&tc)              │─────────┐                      │
+│  └──────────┬─────────────────────────┘         │                      │
+│             │                                    │                     │
+│        ┌────┴────┐                               │                     │
+│        v         v                               │                     │
+│     did work   no work                           │                     │
+│        │         │                               │                     │
+│        v         v                               │                     │
+│  idle = 0    idle++                              │                     │
+│        │         │                               │                     │
+│        │    ┌────┴────────────────────────┐      │                     │
+│        │    │ idle < spin_thr:  spin      │      │                     │
+│        │    │ idle < yield_thr: yield     │      │                     │
+│        │    │ else: sleep(incr, capped)   │      │                     │
+│        │    └─────────────────────────┬───┘      │                     │
+│        │                              │          │                     │
+│        └──────────┬───────────────────┘          │                     │
+│                   │                              │                     │
+│                   └──── loop back ───────────────┘                     │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+## Tick Once (four phases)
+
+```
+┌─ tick_once ────────────────────────────────────────────────────────────┐
+│                                                                        │
+│  PHASE 1 --- Drain Spawn Queue                                         │
+│  ┌────────────────────────────────────────────────────────────────────┐│
+│  │                                                                    ││
+│  │  spawn_rx --try_recv()--> (addr, Box<dyn AnyActor>)                ││
+│  │                                   │                                ││
+│  │                                   v                                ││
+│  │                          pool.insert(addr, actor)                  ││
+│  │                                   │                                ││
+│  │                                   v                                ││
+│  │                          ActorSlot {                               ││
+│  │                              mailbox: VecDeque::new()              ││
+│  │                              actor: <the new actor>                ││
+│  │                          }                                         ││
+│  │                                                                    ││
+│  └────────────────────────────────────────────────────────────────────┘│
+│      │                                                                 │
+│      v                                                                 │
+│  PHASE 2 --- Drain Transfer Queue                                      │
+│  ┌────────────────────────────────────────────────────────────────────┐│
+│  │                                                                    ││
+│  │  transfer_rx --try_recv()--> Envelope { dest, payload }            ││
+│  │                                         │                          ││
+│  │                                         v                          ││
+│  │                             pool.deliver(&dest, payload)           ││
+│  │                                         │                          ││
+│  │                                         v                          ││
+│  │                             slot.mailbox.push_back(msg)            ││
+│  │                             (untyped; type check at handle time)   ││
+│  │                                                                    ││
+│  └────────────────────────────────────────────────────────────────────┘│
+│      │                                                                 │
+│      v                                                                 │
+│  PHASE 3 --- Tick All Actors                                           │
+│  ┌────────────────────────────────────────────────────────────────────┐│
+│  │                                                                    ││
+│  │  ┌─ WorkerContext (on stack) ─────────────────────────────────┐    ││
+│  │  │  implements ContextInner                                   │    ││
+│  │  │  owns pending_local: RefCell<Vec<(Addr, Box<Any>)>>        │    ││
+│  │  └────────────────────────────────────────────────────────────┘    ││
+│  │                                                                    ││
+│  │  for each (addr, slot) in pool:                                    ││
+│  │                                                                    ││
+│  │    ┌─ drain_count ──────────────────────────────────────────┐      ││
+│  │    │  len = slot.mailbox.len()                              │      ││
+│  │    │  len <  waterlevel --> n = len     (drain all)         │      ││
+│  │    │  len >= waterlevel --> n = len / 2 (backpressure)      │      ││
+│  │    └────────────────────────────────────────────────────────┘      ││
+│  │                                                                    ││
+│  │    ctx = Ctx { inner: &worker_ctx, self_addr: addr }               ││
+│  │                                                                    ││
+│  │    repeat n times:                                                 ││
+│  │      msg = slot.mailbox.pop_front()                                ││
+│  │      slot.actor.handle_any(&ctx, msg)                              ││
+│  │              │                                                     ││
+│  │              │ actor calls ctx.send() or ctx.spawn()               ││
+│  │              v                                                     ││
+│  │                                                                    ││
+│  │    ┌─ WorkerContext routes ─────────────────────────────────────┐  ││
+│  │    │                                                            │  ││
+│  │    │  send_any(addr, msg):                                      │  ││
+│  │    │    ┌──────────────┬──────────────┬─────────────────┐       │  ││
+│  │    │    │ same worker  │ other worker │ unknown addr    │       │  ││
+│  │    │    │              │              │                 │       │  ││
+│  │    │    │ pending_     │ transfer_tx  │ inbox_registry  │       │  ││
+│  │    │    │  local.push()│  [wid].send()│  .try_deliver() │       │  ││
+│  │    │    └──────────────┴──────────────┴─────────────────┘       │  ││
+│  │    │                                                            │  ││
+│  │    │  spawn_any(addr, actor):                                   │  ││
+│  │    │    wid = placement.next_worker()                           │  ││
+│  │    │    address_map.insert(addr, wid)                           │  ││
+│  │    │    spawn_txs[wid].send((addr, actor))                      │  ││
+│  │    │                                                            │  ││
+│  │    └────────────────────────────────────────────────────────────┘  ││
+│  │                                                                    ││
+│  └────────────────────────────────────────────────────────────────────┘│
+│      │                                                                 │
+│      v                                                                 │
+│  PHASE 4 --- Drain Pending Local                                       │
+│  ┌────────────────────────────────────────────────────────────────────┐│
+│  │                                                                    ││
+│  │  for (addr, msg) in pending_local.into_inner():                    ││
+│  │      pool.deliver(&addr, msg)                                      ││
+│  │          --> slot.mailbox.push_back(msg)                           ││
+│  │                                                                    ││
+│  │  these sit in the mailbox until NEXT tick                          ││
+│  │                                                                    ││
+│  └────────────────────────────────────────────────────────────────────┘│
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+## External Interactions
+
+Everything the worker talks to, and everything that talks to it.
+
+### Who writes into the worker's queues
+
+```
+┌─ User Code ────────────────────────────────────────────────────────────┐
+│                                                                        │
+│  let rt = Runtime::new(config);                                        │
+│  let addr = rt.spawn(my_actor)?;     // --┐                            │
+│  rt.send_to(addr, MyMsg(42))?;       // --┤                            │
+│                                       // │                             │
+└──────────────────────────────────┼──┼───────────────────────────────────┘
+                                   │  │
+        ┌───────────────────────────┘  │
+        │                              │
+        v                              v
+┌─ Runtime ──────────────────────────────────────────────────────────────┐
+│                                                                        │
+│  spawn():                                                              │
+│    addr = ActorAddress::new_random()                                   │
+│    wid  = placement.next_worker()          -- round-robin pick         │
+│    address_map.insert(addr, wid)           -- register globally        │
+│    spawn_txs[wid].try_send((addr, boxed))  -- push to worker queue     │
+│         │                                                              │
+│         │       ┌────────────────────────────────────────────┐         │
+│         └──────>│ Worker.spawn_rx  (Receiver side)           │         │
+│                 └────────────────────────────────────────────┘         │
+│                                                                        │
+│  send_to():                                                            │
+│    wid = address_map.lookup(&addr)                                     │
+│    transfer_txs[wid].try_send(Envelope::new(addr, msg))                │
+│         │                                                              │
+│         │       ┌────────────────────────────────────────────┐         │
+│         └──────>│ Worker.transfer_rx  (Receiver side)        │         │
+│                 └────────────────────────────────────────────┘         │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Who the worker talks to during a tick
+
+```
+┌─ Worker (during phase 3: tick_all) ────────────────────────────────────┐
+│                                                                        │
+│  An actor calls ctx.send(addr, msg) or ctx.spawn(new_actor).           │
+│  These go through WorkerContext, which implements ContextInner.        │
+│                                                                        │
+│  ctx.send(addr, msg)                                                   │
+│       │                                                                │
+│       v                                                                │
+│  ┌─ WorkerContext.send_any ─────────────────────────────────────────┐  │
+│  │                                                                  │  │
+│  │  address_map.lookup(addr) --> which worker owns this actor?      │  │
+│  │       │                                                          │  │
+│  │  ┌────┴──────────────┬──────────────────┬──────────────────┐     │  │
+│  │  │                   │                  │                  │     │  │
+│  │  v                   v                  v                  │     │  │
+│  │  SAME WORKER         OTHER WORKER       NOT FOUND          │     │  │
+│  │  │                   │                  │                  │     │  │
+│  │  │ pending_local     │ transfer_txs     │ inbox_registry   │     │  │
+│  │  │   .push(addr,msg) │  [wid].send()    │  .try_deliver()  │     │  │
+│  │  │                   │                  │                  │     │  │
+│  │  │ stays in this     │ crosses to       │ goes to an       │     │  │
+│  │  │ worker; delivered │ another worker   │ external         │     │  │
+│  │  │ in phase 4        │ thread's         │ Inbox<M>         │     │  │
+│  │  │                   │ transfer_rx      │ receiver          │    │  │
+│  │  └───────────────────┴──────────────────┴──────────────────┘     │  │
+│  │                                                                  │  │
+│  └──────────────────────────────────────────────────────────────────┘  │
+│                                                                        │
+│  ctx.spawn(new_actor)                                                  │
+│       │                                                                │
+│       v                                                                │
+│  ┌─ WorkerContext.spawn_any ────────────────────────────────────────┐  │
+│  │                                                                  │  │
+│  │  wid = placement.next_worker()  -- round-robin target            │  │
+│  │  address_map.insert(addr, wid)  -- register in global map        │  │
+│  │  spawn_txs[wid].try_send(...)   -- enqueue for target worker     │  │
+│  │                                                                  │  │
+│  │  may land on THIS worker or a DIFFERENT worker                   │  │
+│  │  target picks it up in phase 1 of its next tick                  │  │
+│  │                                                                  │  │
+│  └──────────────────────────────────────────────────────────────────┘  │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### The channel connecting everything
+
+Each worker has two inbound channels. The channels are lock-free MPSC queues
+backed by `crossbeam::ArrayQueue` with a `SegQueue` overflow.
+
+```
+┌─ HybridChannel<T> ─────────────────────────────────────────────────────┐
+│                                                                        │
+│  ┌─ ring: ArrayQueue<T> ──────────────────────────────────────┐        │
+│  │  fixed capacity, lock-free, bounded                        │        │
+│  │  ┌───┬───┬───┬───┬───┬───┬───┬───┐                         │        │
+│  │  │   │   │   │   │   │   │   │   │  (pre-allocated)        │        │
+│  │  └───┴───┴───┴───┴───┴───┴───┴───┘                         │        │
+│  └────────────────────────────────────────────────────────────┘        │
+│                                                                        │
+│  ┌─ overflow: SegQueue<T> ────────────────────────────────────┐        │
+│  │  unbounded, lock-free, linked nodes                        │        │
+│  │  used only when ring is full                               │        │
+│  └────────────────────────────────────────────────────────────┘        │
+│                                                                        │
+│  push(v):  try ring first, spill to overflow                           │
+│  pop():    drain ring first, then overflow                             │
+│                                                                        │
+│  ┌─ Sender<T> ─────────┐       ┌─ Receiver<T> ─────────┐               │
+│  │ Arc<HybridChannel<T>>│       │ Arc<HybridChannel<T>>  │             │
+│  │ .try_send(v)         │──────>│ .try_recv() -> Option  │             │
+│  │ clonable (new_sender)│ same  │ single consumer        │             │
+│  └──────────────────────┘ Arc   └────────────────────────┘             │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+
+Who holds what:
+
+  transfer channel:
+    Sender   held by: Runtime.transfer_txs[i], workers via TickContext
+    Receiver held by: Worker[i].transfer_rx
+
+  spawn channel:
+    Sender   held by: Runtime.spawn_txs[i], workers via TickContext
+    Receiver held by: Worker[i].spawn_rx
+```
+
+### The AddressMap: global actor directory
+
+```
+┌─ AddressMap ───────────────────────────────────────────────────────────┐
+│                                                                        │
+│  RwLock< HashMap<ActorAddress, WorkerId> >                             │
+│                                                                        │
+│  ┌──────────────────────────────────────────────────────────────────┐  │
+│  │  addr_0 -> WorkerId(0)                                           │  │
+│  │  addr_1 -> WorkerId(2)                                           │  │
+│  │  addr_2 -> WorkerId(0)                                           │  │
+│  │  addr_3 -> WorkerId(1)                                           │  │
+│  │  ...                                                             │  │
+│  └──────────────────────────────────────────────────────────────────┘  │
+│                                                                        │
+│  READERS (concurrent, RwLock read):                                    │
+│    WorkerContext.send_any() -- every message send does a lookup        │
+│    Runtime.send_to()       -- external sends do a lookup               │
+│                                                                        │
+│  WRITERS (rare, exclusive lock):                                       │
+│    Runtime.spawn()          -- registers new actor at spawn time       │
+│    WorkerContext.spawn_any() -- actor spawns another actor             │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### The InboxRegistry: escape hatch to user code
+
+```
+┌─ InboxRegistry ────────────────────────────────────────────────────────┐
+│                                                                        │
+│  RwLock< HashMap<ActorAddress, Arc<dyn SenderT>> >                     │
+│                                                                        │
+│  For addresses belonging to external Inbox<M>, not actors.             │
+│                                                                        │
+│  ┌─ Registration ─────────────────────────────────────────────────┐    │
+│  │                                                                │    │
+│  │  Runtime.new_inbox::<M>()                                      │    │
+│  │    addr = ActorAddress::new_random()                           │    │
+│  │    receiver = Receiver::<M>::new(capacity)                     │    │
+│  │    sender   = receiver.new_sender()                            │    │
+│  │    inbox_registry.register(addr, Arc::new(sender))             │    │
+│  │    returns Inbox { addr, inner: receiver }                     │    │
+│  │                                                                │    │
+│  └────────────────────────────────────────────────────────────────┘    │
+│                                                                        │
+│  ┌─ Delivery (when address_map lookup fails) ─────────────────────┐    │
+│  │                                                                │    │
+│  │  WorkerContext.send_any(addr, msg)                             │    │
+│  │    address_map.lookup(addr) -> None                            │    │
+│  │    inbox_registry.try_deliver(addr, msg)                       │    │
+│  │      senders.read().get(addr).try_send_any(msg)                │    │
+│  │        downcast Box<Any> -> M, push into Receiver<M>           │    │
+│  │                                                                │    │
+│  └────────────────────────────────────────────────────────────────┘    │
+│                                                                        │
+│  ┌─ Consumption (user code) ──────────────────────────────────────┐    │
+│  │                                                                │    │
+│  │  let inbox = rt.new_inbox::<MyMsg>()?;                         │    │
+│  │  // later, from any thread:                                    │    │
+│  │  if let Some(msg) = inbox.try_recv() { ... }                   │    │
+│  │                                                                │    │
+│  └────────────────────────────────────────────────────────────────┘    │
+│                                                                        │
+└────────────────────────────────────────────────────────────────────────┘
+```
+
+### Full system topology
+
+```
+┌─ User Code ────────────────────────────────────────────────────────────┐
+│ rt.spawn()      rt.send_to()      inbox.try_recv()    rt.shutdown()   │
+└────┬──────────────────┬──────────────────┬──────────────────┬──────────┘
+     │                  │                  ^                  │
+     v                  v                  │                  v
+┌─ Arc<Runtime> ────────────────────────────────────────────────────────────┐
+│                                                                           │
+│ ┌───────────┐ ┌────────────┐ ┌─────────────┐ ┌────────────┐              │
+│ │AddressMap │ │ Placement  │ │InboxRegistry│ │ is_running │              │
+│ │ addr->wid │ │ round-robin│ │ addr->Sender│ │ AtomicBool │              │
+│ └─────┬─────┘ └──────┬─────┘ └──────┬──────┘ └──────┬─────┘              │
+│       │               │              │               │                    │
+│ ┌─────┴───────────────┴──────────────┴───────────────┴────────────────┐   │
+│ │                  TickContext (borrows all above)                     │   │
+│ └──────────────────────────┬──────────────────────────────────────────┘   │
+│                            │                                              │
+│ ┌─ transfer_txs[] ────┐   │   ┌─ spawn_txs[] ──────┐                     │
+│ │ [0]: Sender<Envelope│   │   │ [0]: Sender<(A,Box)>│                     │
+│ │ [1]: Sender<Envelope│   │   │ [1]: Sender<(A,Box)>│                     │
+│ │ [2]: Sender<Envelope│   │   │ [2]: Sender<(A,Box)>│                     │
+│ └──┬──────┬──────┬────┘   │   └──┬──────┬──────┬────┘                     │
+│    │      │      │        │      │      │      │                          │
+└────┼──────┼──────┼────────┼──────┼──────┼──────┼──────────────────────────┘
+     │      │      │        │      │      │      │
+     v      v      v        │      v      v      v
+┌────────┐┌────────┐┌───────┴┐┌────────┐┌────────┐┌────────┐
+│xfer    ││xfer    ││xfer    ││ spawn  ││ spawn  ││ spawn  │
+│_rx[0]  ││_rx[1]  ││_rx[2]  ││ _rx[0] ││ _rx[1] ││ _rx[2] │
+└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘└───┬────┘
+    │         │         │         │         │         │
+    v         v         v         v         v         v
+┌─ Worker 0 ──────┐ ┌─ Worker 1 ──────┐ ┌─ Worker 2 ──────┐
+│                  │ │                  │ │                  │
+│ ┌─ pool ──────┐  │ │ ┌─ pool ──────┐  │ │ ┌─ pool ──────┐  │
+│ │ ┌──────────┐│  │ │ │ ┌──────────┐│  │ │ │ ┌──────────┐│  │
+│ │ │ slot:    ││  │ │ │ │ slot:    ││  │ │ │ │ slot:    ││  │
+│ │ │  mailbox ││  │ │ │ │  mailbox ││  │ │ │ │  mailbox ││  │
+│ │ │  actor   ││  │ │ │ │  actor   ││  │ │ │ │  actor   ││  │
+│ │ └──────────┘│  │ │ │ └──────────┘│  │ │ │ └──────────┘│  │
+│ │ ┌──────────┐│  │ │ │ ┌──────────┐│  │ │ │             │  │
+│ │ │ slot:    ││  │ │ │ │ slot:    ││  │ │ └─────────────┘  │
+│ │ │  mailbox ││  │ │ │ │  mailbox ││  │ │                  │
+│ │ │  actor   ││  │ │ │ │  actor   ││  │ │ thread 2         │
+│ │ └──────────┘│  │ │ │ └──────────┘│  │ └──────────────────┘
+│ └─────────────┘  │ │ └─────────────┘  │
+│                  │ │                  │
+│ thread 0         │ │ thread 1         │
+└──────────────────┘ └──────────────────┘
+
+Workers also send to EACH OTHER during phase 3:
+  WorkerContext.send_any()  -> transfer_txs[other_wid].try_send()
+  WorkerContext.spawn_any() -> spawn_txs[target_wid].try_send()
+```
+
+## Message Lifecycle
+
+```
+                       PRODUCERS
+       ┌──────────────────┬──────────────────────┐
+       │                  │                       │
+       v                  v                       v
+┌─────────────┐  ┌──────────────┐  ┌──────────────────────┐
+│ Runtime     │  │ ctx.send()   │  │ ctx.send()           │
+│  .send_to() │  │ same worker  │  │ other worker         │
+└──────┬──────┘  └──────┬───────┘  └──────────┬───────────┘
+       │                │                      │
+       v                v                      v
+┌─────────────┐  ┌─────────────┐  ┌──────────────────────┐
+│ transfer_tx │  │ pending_    │  │ transfer_tx          │
+│ [wid].send()│  │ local.push()│  │ [wid].send()         │
+└──────┬──────┘  └──────┬──────┘  └──────────┬───────────┘
+       │                │                     │
+       │          (end of phase 3)            │
+       │                │                     │
+       │          phase 4:                    │
+       │                │                     │
+       v                v                     v
+┌────────────────────────────────────────────────────┐
+│                                                    │
+│           pool.deliver(&addr, msg)                 │
+│                    │                               │
+│                    v                               │
+│      slot.mailbox.push_back(msg)                   │
+│                                                    │
+└───────────────────────┬────────────────────────────┘
+                        │
+                  next tick_once
+                    phase 3
+                        │
+                        v
+┌────────────────────────────────────────────────────┐
+│                                                    │
+│  msg = slot.mailbox.pop_front()                    │
+│                 │                                  │
+│                 v                                  │
+│  slot.actor.handle_any(&ctx, msg)                  │
+│                 │                                  │
+│                 v                                  │
+│  ┌──────────────────────────────────────────┐      │
+│  │ downcast Box<dyn Any> to A::Incoming     │      │
+│  │                                          │      │
+│  │ ok:  A.handle(ctx, typed_msg)            │      │
+│  │ err: silently dropped                    │      │
+│  └──────────────────────────────────────────┘      │
+│                                                    │
+└────────────────────────────────────────────────────┘
+```
+
+## Shutdown Flow
+
+```
+┌─ User Code ──────┐
+│                   │
+│  rt.shutdown()    │
+│       │           │
+└───────┼───────────┘
+        │
+        v
+┌─ Runtime ──────────────────────────────────────┐
+│                                                 │
+│  is_running.store(false, Release)               │
+│                                                 │
+└────────────────────────┬────────────────────────┘
+                         │
+       ┌─────────────────┼─────────────────┐
+       │                 │                 │
+       v                 v                 v
+┌─ Worker 0 ────┐ ┌─ Worker 1 ────┐ ┌─ Worker 2 ────┐
+│                │ │                │ │                │
+│ is_running     │ │ is_running     │ │ is_running     │
+│  .load(Acquire)│ │  .load(Acquire)│ │  .load(Acquire)│
+│   -> false     │ │   -> false     │ │   -> false     │
+│                │ │                │ │                │
+│ run() returns  │ │ run() returns  │ │ run() returns  │
+│ thread exits   │ │ thread exits   │ │ thread exits   │
+└────────────────┘ └────────────────┘ └────────────────┘
+       │                 │                 │
+       └─────────────────┼─────────────────┘
+                         │
+                         v
+               ┌─ RuntimeHandle ──┐
+               │                   │
+               │  .join()          │
+               │  waits for all    │
+               │  JoinHandles      │
+               │                   │
+               └───────────────────┘
+```
+

examples/python/.ipynb_checkpoints/getting_started-checkpoint.ipynb

@@ -0,0 +1,98 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Swactor — Getting Started\n",
+    "\n",
+    "This notebook walks through the basics of the swactor actor runtime from Python."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from swactor import Runtime"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Single-threaded: spawn, send, tick\n",
+    "\n",
+    "The simplest way to use swactor is with `tick()` — manually stepping the runtime one tick at a time."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def echo(ctx, msg):\n",
+    "    ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
+    "\n",
+    "rt = Runtime()\n",
+    "addr = rt.spawn(echo)\n",
+    "inbox = rt.inbox()\n",
+    "\n",
+    "rt.send(addr, {\"name\": \"world\", \"reply_to\": inbox.addr})\n",
+    "rt.tick()\n",
+    "\n",
+    "print(inbox.try_recv())"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Multi-threaded: background runtime\n",
+    "\n",
+    "For real workloads you can run the runtime on background threads with `RuntimeConfig`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import time\n",
+    "from swactor import Runtime, RuntimeConfig\n",
+    "\n",
+    "def echo(ctx, msg):\n",
+    "    ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
+    "\n",
+    "rt = Runtime(RuntimeConfig(num_threads=2))\n",
+    "addr = rt.spawn(echo)\n",
+    "inbox = rt.inbox()\n",
+    "handle = rt.run()\n",
+    "\n",
+    "for name in [\"alice\", \"bob\", \"charlie\"]:\n",
+    "    handle.send(addr, {\"name\": name, \"reply_to\": inbox.addr})\n",
+    "    time.sleep(0.05)  # give the runtime a moment\n",
+    "    print(inbox.try_recv())\n",
+    "\n",
+    "handle.shutdown()\n",
+    "handle.join()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.9.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}

examples/python/getting_started.ipynb

@@ -0,0 +1,98 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Swactor — Getting Started\n",
+    "\n",
+    "This notebook walks through the basics of the swactor actor runtime from Python."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from swactor import Runtime"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Single-threaded: spawn, send, tick\n",
+    "\n",
+    "The simplest way to use swactor is with `tick()` — manually stepping the runtime one tick at a time."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def echo(ctx, msg):\n",
+    "    ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
+    "\n",
+    "rt = Runtime()\n",
+    "addr = rt.spawn(echo)\n",
+    "inbox = rt.inbox()\n",
+    "\n",
+    "rt.send(addr, {\"name\": \"world\", \"reply_to\": inbox.addr})\n",
+    "rt.tick()\n",
+    "\n",
+    "print(inbox.try_recv())"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Multi-threaded: background runtime\n",
+    "\n",
+    "For real workloads you can run the runtime on background threads with `RuntimeConfig`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import time\n",
+    "from swactor import Runtime, RuntimeConfig\n",
+    "\n",
+    "def echo(ctx, msg):\n",
+    "    ctx.send(msg[\"reply_to\"], f\"hello, {msg['name']}!\")\n",
+    "\n",
+    "rt = Runtime(RuntimeConfig(num_threads=2))\n",
+    "addr = rt.spawn(echo)\n",
+    "inbox = rt.inbox()\n",
+    "handle = rt.run()\n",
+    "\n",
+    "for name in [\"alice\", \"bob\", \"charlie\"]:\n",
+    "    handle.send(addr, {\"name\": name, \"reply_to\": inbox.addr})\n",
+    "    time.sleep(0.05)  # give the runtime a moment\n",
+    "    print(inbox.try_recv())\n",
+    "\n",
+    "handle.shutdown()\n",
+    "handle.join()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.9.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}

examples/python/hello_async.py

@@ -30,6 +30,8 @@ async def main():
         reply = await recv(inbox)
         print(reply)
 
+     # show us our actors!
+    print(handle.stats())
     handle.shutdown()
     handle.join()
 

pyproject.toml

@@ -7,5 +7,8 @@ name = "swactor"
 version = "0.1.0"
 requires-python = ">=3.9"
 
+[dependency-groups]
+dev = ["jupyter", "ipykernel"]
+
 [tool.maturin]
 features = ["python"]

src/actor.rs

@@ -1,6 +1,6 @@
 use std::any::Any;
 
-use crate::{get_random, runtime::{ContextInner, Ctx}, worker::Mailbox};
+use crate::runtime::Ctx;
 
 /// The primary trait defining data that can be passed to and from actor processes
 pub trait Message: 'static + Sized + Clone + Send + Sync {}
@@ -20,63 +20,32 @@ pub struct ActorAddress(pub [u8; 32]);
 impl ActorAddress {
     pub fn new_random() -> Self {
         let mut bytes = [0u8; 32];
-        get_random(&mut bytes);
+        crate::get_random(&mut bytes);
         Self(bytes)
     }
 }
 
-/// The actor process as represented in the Runtime, with the actor state stored with its mailbox.
-pub(crate) struct Actor<A>
-where
-    A: ActorInterface,
-{
-    addr: ActorAddress,
-    mailbox: Mailbox<A::Incoming>,
-    inner: A,
-}
+/// The actor process as represented in the Runtime — thin wrapper around user state.
+pub(crate) struct Actor<A: ActorInterface>(A);
 
 impl<A: ActorInterface> Actor<A> {
-    pub(crate) fn new(addr: ActorAddress, mailbox: Mailbox<A::Incoming>, inner: A) -> Self {
-        Self {
-            addr,
-            mailbox,
-            inner,
-        }
+    pub(crate) fn new(inner: A) -> Self {
+        Self(inner)
     }
 }
 
-/// Trait for type-erased actors
+/// Trait for type-erased actors — single-message handler.
 pub(crate) trait AnyActor: Send {
-    /// Tick the actor, processing pending messages. Returns `true` if any work was done.
-    fn tick(&mut self, inner: &dyn ContextInner) -> bool;
-    /// Deliver a type-erased message into this actor's mailbox.
-    /// Returns `true` if the downcast succeeded.
-    fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool;
+    fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>);
 }
 
 impl<A> AnyActor for Actor<A>
 where
     A: ActorInterface,
 {
-    fn tick(&mut self, inner: &dyn ContextInner) -> bool {
-        let n = self.mailbox.drain_count();
-        if n > 0 {
-            let ctx = Ctx::new(inner, self.addr);
-            for _ in 0..n {
-                if let Some(msg) = self.mailbox.pop() {
-                    self.inner.handle(&ctx, msg);
-                }
-            }
-        }
-        n > 0
-    }
-
-    fn deliver(&mut self, msg: Box<dyn Any + Send>) -> bool {
+    fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) {
         if let Ok(typed) = msg.downcast::<A::Incoming>() {
-            self.mailbox.push(*typed);
-            true
-        } else {
-            false
+            self.0.handle(ctx, *typed);
         }
     }
 }

src/address_map.rs

@@ -49,6 +49,16 @@ impl AddressMap {
     pub fn len(&self) -> usize {
         self.inner.read().unwrap().len()
     }
+
+    /// Returns a snapshot of all (address, worker) pairs.
+    pub fn snapshot(&self) -> Vec<(ActorAddress, WorkerId)> {
+        self.inner
+            .read()
+            .unwrap()
+            .iter()
+            .map(|(addr, wid)| (*addr, *wid))
+            .collect()
+    }
 }
 
 /// Round-robin actor placement strategy.

src/python.rs

@@ -7,7 +7,6 @@ use pyo3::types::PyModule;
 use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor};
 use crate::config::{BackoffPolicy, RuntimeConfig};
 use crate::runtime::{Ctx, Inbox, Runtime, RuntimeHandle};
-use crate::worker::Mailbox;
 use crate::Error;
 
 // ─── PyMsg newtype ───────────────────────────────────────────────────────────
@@ -176,16 +175,14 @@ impl ActorInterface for PyActor {
                 for effect in effects {
                     match effect {
                         Effect::Send { addr, msg } => {
-                            let _ = ctx.raw_inner().send_via_queue(
+                            let _ = ctx.raw_inner().send_any(
                                 addr,
                                 Box::new(PyMsg(msg)) as Box<dyn Any + Send>,
                             );
                         }
                         Effect::Spawn { addr, handler } => {
-                            let waterlevel = ctx.raw_inner().mailbox_waterlevel();
                             let actor = PyActor::new(handler);
-                            let actor = Actor::new(addr, Mailbox::new(waterlevel), actor);
-                            let boxed: Box<dyn AnyActor> = Box::new(actor);
+                            let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
                             let _ = ctx.raw_inner().spawn_any(addr, boxed);
                         }
                     }
@@ -364,6 +361,14 @@ impl PyRuntime {
         })
     }
 
+    fn stats(&self) -> PyResult<PyRuntimeStats> {
+        let rt = self
+            .inner
+            .as_ref()
+            .ok_or_else(|| pyo3::exceptions::PyRuntimeError::new_err("Runtime consumed by run()"))?;
+        Ok(build_stats(rt))
+    }
+
     fn shutdown(&self) -> PyResult<()> {
         let rt = self
             .inner
@@ -419,6 +424,16 @@ impl PyRuntimeHandle {
         Ok(PyInbox { inner: inbox })
     }
 
+    fn stats(&self) -> PyResult<PyRuntimeStats> {
+        let handle = self
+            .inner
+            .as_ref()
+            .ok_or_else(|| {
+                pyo3::exceptions::PyRuntimeError::new_err("RuntimeHandle consumed by join()")
+            })?;
+        Ok(build_stats(&handle.runtime))
+    }
+
     fn shutdown(&self) -> PyResult<()> {
         let handle = self
             .inner
@@ -442,6 +457,117 @@ impl PyRuntimeHandle {
     }
 }
 
+// ─── ActorInfo / RuntimeStats ────────────────────────────────────────────────
+
+#[pyclass(name = "ActorInfo")]
+#[derive(Clone)]
+pub struct PyActorInfo {
+    #[pyo3(get)]
+    address: PyActorAddress,
+    #[pyo3(get)]
+    worker_id: usize,
+}
+
+#[pymethods]
+impl PyActorInfo {
+    fn __repr__(&self) -> String {
+        let hex = self.address.hex();
+        format!("ActorInfo(address={hex}, worker={})", self.worker_id)
+    }
+}
+
+#[pyclass(name = "WorkerInfo")]
+#[derive(Clone)]
+pub struct PyWorkerInfo {
+    #[pyo3(get)]
+    id: usize,
+    #[pyo3(get)]
+    num_actors: usize,
+    #[pyo3(get)]
+    mailbox_depth: usize,
+    #[pyo3(get)]
+    messages_processed: u64,
+}
+
+#[pymethods]
+impl PyWorkerInfo {
+    fn __repr__(&self) -> String {
+        format!(
+            "WorkerInfo(id={}, actors={}, queued={}, processed={})",
+            self.id, self.num_actors, self.mailbox_depth, self.messages_processed
+        )
+    }
+}
+
+#[pyclass(name = "RuntimeStats")]
+#[derive(Clone)]
+pub struct PyRuntimeStats {
+    #[pyo3(get)]
+    num_actors: usize,
+    #[pyo3(get)]
+    num_workers: usize,
+    #[pyo3(get)]
+    actors: Vec<PyActorInfo>,
+    #[pyo3(get)]
+    workers: Vec<PyWorkerInfo>,
+}
+
+#[pymethods]
+impl PyRuntimeStats {
+    fn __repr__(&self) -> String {
+        let mut out = format!(
+            "RuntimeStats(actors={}, workers={})",
+            self.num_actors, self.num_workers
+        );
+
+        for w in &self.workers {
+            out.push_str(&format!(
+                "\n  Worker {}: {} actors, {} queued, {} processed",
+                w.id, w.num_actors, w.mailbox_depth, w.messages_processed
+            ));
+            for info in &self.actors {
+                if info.worker_id == w.id {
+                    out.push_str(&format!("\n    - {}", info.address.hex()));
+                }
+            }
+        }
+
+        out
+    }
+
+    fn __str__(&self) -> String {
+        self.__repr__()
+    }
+}
+
+fn build_stats(runtime: &Runtime) -> PyRuntimeStats {
+    let stats = runtime.stats();
+    let actors: Vec<PyActorInfo> = stats
+        .actors
+        .into_iter()
+        .map(|(addr, wid)| PyActorInfo {
+            address: PyActorAddress::from(addr),
+            worker_id: wid,
+        })
+        .collect();
+    let workers: Vec<PyWorkerInfo> = stats
+        .workers
+        .into_iter()
+        .map(|w| PyWorkerInfo {
+            id: w.id,
+            num_actors: w.num_actors,
+            mailbox_depth: w.mailbox_depth,
+            messages_processed: w.messages_processed,
+        })
+        .collect();
+    PyRuntimeStats {
+        num_actors: actors.len(),
+        num_workers: stats.num_workers,
+        actors,
+        workers,
+    }
+}
+
 // ─── Module registration ─────────────────────────────────────────────────────
 
 pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
@@ -451,5 +577,8 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PyRuntimeConfig>()?;
     m.add_class::<PyRuntime>()?;
     m.add_class::<PyRuntimeHandle>()?;
+    m.add_class::<PyActorInfo>()?;
+    m.add_class::<PyWorkerInfo>()?;
+    m.add_class::<PyRuntimeStats>()?;
     Ok(())
 }

src/runtime.rs

@@ -10,11 +10,26 @@ use crate::address_map::{AddressMap, Placement, WorkerId};
 use crate::channel::{Receiver, Sender};
 // Re-export config types so existing code using `runtime::RuntimeConfig` still works
 pub use crate::config::{BackoffPolicy, RuntimeConfig};
-use crate::worker::Mailbox;
-use crate::worker::{TickContext, Worker};
+use crate::worker::{TickContext, Worker, WorkerStats};
 use crate::Error;
 
 
+/// Snapshot of per-worker state.
+pub struct WorkerInfo {
+    pub id: usize,
+    pub num_actors: usize,
+    pub mailbox_depth: usize,
+    pub messages_processed: u64,
+}
+
+/// Snapshot of overall runtime state.
+pub struct RuntimeStats {
+    pub num_workers: usize,
+    /// Each entry is (address, worker_id).
+    pub actors: Vec<(ActorAddress, usize)>,
+    pub workers: Vec<WorkerInfo>,
+}
+
 /// Generic message inbox for receiving messages outside of the runtime.
 pub struct Inbox<M: Message> {
     addr: ActorAddress,
@@ -82,9 +97,7 @@ impl<'a> Ctx<'a> {
     /// Spawn a new actor, returning its address.
     pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
         let addr = ActorAddress::new_random();
-        let waterlevel = self.inner.mailbox_waterlevel();
-        let actor = Actor::new(addr, Mailbox::new(waterlevel), actor);
-        let boxed: Box<dyn AnyActor> = Box::new(actor);
+        let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
         self.inner.spawn_any(addr, boxed)?;
         Ok(addr)
     }
@@ -115,6 +128,7 @@ pub struct Runtime {
     spawn_txs: Vec<Sender<(ActorAddress, Box<dyn AnyActor>)>>,
     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()
@@ -142,6 +156,7 @@ impl Runtime {
 
         let mut transfer_txs = Vec::with_capacity(num_workers);
         let mut spawn_txs = Vec::with_capacity(num_workers);
+        let mut worker_stats = Vec::with_capacity(num_workers);
         let mut workers = Vec::with_capacity(num_workers);
 
         for i in 0..num_workers {
@@ -154,7 +169,9 @@ impl Runtime {
             let spawn_tx = spawn_rx.new_sender();
             spawn_txs.push(spawn_tx);
 
-            workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx));
+            let stats = Arc::new(WorkerStats::new());
+            worker_stats.push(stats.clone());
+            workers.push(Worker::new(WorkerId(i), transfer_rx, spawn_rx, stats));
         }
 
         if config.num_threads < 2 {
@@ -168,6 +185,7 @@ impl Runtime {
                 spawn_txs,
                 placement,
                 is_running: AtomicBool::new(false),
+                worker_stats,
                 single_worker: Some(RefCell::new(worker)),
                 pending_workers: None,
             }
@@ -181,6 +199,7 @@ impl Runtime {
                 spawn_txs,
                 placement,
                 is_running: AtomicBool::new(false),
+                worker_stats,
                 single_worker: None,
                 pending_workers: Some(workers),
             }
@@ -192,8 +211,7 @@ impl Runtime {
         let addr = ActorAddress::new_random();
         let worker_id = self.placement.next_worker();
         self.address_map.insert(addr, worker_id);
-        let actor = Actor::new(addr, Mailbox::new(self.config.mailbox_waterlevel), actor);
-        let boxed: Box<dyn AnyActor> = Box::new(actor);
+        let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
         self.spawn_txs[worker_id.as_usize()]
             .try_send((addr, boxed))
             .map_err(|_| Error::from("Runtime error: spawn queue full"))?;
@@ -280,6 +298,37 @@ impl Runtime {
         })
     }
 
+    /// Returns a snapshot of runtime stats: actor placements and per-worker info.
+    pub fn stats(&self) -> RuntimeStats {
+        let num_workers = if self.config.num_threads < 2 {
+            1
+        } else {
+            self.config.num_threads
+        };
+        let workers = self
+            .worker_stats
+            .iter()
+            .enumerate()
+            .map(|(i, ws)| WorkerInfo {
+                id: i,
+                num_actors: ws.num_actors.load(Ordering::Relaxed),
+                mailbox_depth: ws.total_mailbox_depth.load(Ordering::Relaxed),
+                messages_processed: ws.messages_processed.load(Ordering::Relaxed),
+            })
+            .collect();
+        let actors = self
+            .address_map
+            .snapshot()
+            .into_iter()
+            .map(|(addr, wid)| (addr, wid.as_usize()))
+            .collect();
+        RuntimeStats {
+            num_workers,
+            actors,
+            workers,
+        }
+    }
+
     /// Signal all workers to stop
     pub fn shutdown(&self) {
         self.is_running.store(false, Ordering::Release);

src/worker/mod.rs

@@ -1,24 +1,42 @@
 use std::any::Any;
 use std::cell::RefCell;
 use std::collections::{HashMap, VecDeque};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::Arc;
 use std::thread;
 
 use crate::actor::{ActorAddress, AnyActor, Message};
 use crate::address_map::{AddressMap, Placement, WorkerId};
 use crate::channel::{Receiver, Sender};
 use crate::config::{BackoffPolicy, RuntimeConfig};
-use crate::runtime::{ContextInner, Envelope, InboxRegistry};
+use crate::runtime::{ContextInner, Ctx, Envelope, InboxRegistry};
 use crate::Error;
 
+/// Per-worker stats published via atomics. Readable from any thread.
+pub(crate) struct WorkerStats {
+    pub num_actors: AtomicUsize,
+    pub total_mailbox_depth: AtomicUsize,
+    pub messages_processed: AtomicU64,
+}
+
+impl WorkerStats {
+    pub fn new() -> Self {
+        Self {
+            num_actors: AtomicUsize::new(0),
+            total_mailbox_depth: AtomicUsize::new(0),
+            messages_processed: AtomicU64::new(0),
+        }
+    }
+}
+
 /// Shared state passed to tick_once — single thin pointer avoids register spill.
 pub(crate) struct TickContext<'a> {
-    pub address_map: &'a AddressMap,
-    pub transfer_txs: &'a [Sender<Envelope>],
-    pub spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
-    pub placement: &'a Placement,
-    pub inbox_registry: &'a InboxRegistry,
-    pub config: &'a RuntimeConfig,
+    pub(crate) address_map: &'a AddressMap,
+    pub(crate) transfer_txs: &'a [Sender<Envelope>],
+    pub(crate) spawn_txs: &'a [Sender<(ActorAddress, Box<dyn AnyActor>)>],
+    pub(crate) placement: &'a Placement,
+    pub(crate) inbox_registry: &'a InboxRegistry,
+    pub(crate) config: &'a RuntimeConfig,
 }
 
 /// A worker owns a set of actors and runs them in a loop.
@@ -27,24 +45,27 @@ pub(crate) struct Worker {
     pool: ActorPool,
     transfer_rx: Receiver<Envelope>,
     spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
+    stats: Arc<WorkerStats>,
 }
 
 impl Worker {
-    pub fn new(
+    pub(crate) fn new(
         id: WorkerId,
         transfer_rx: Receiver<Envelope>,
         spawn_rx: Receiver<(ActorAddress, Box<dyn AnyActor>)>,
+        stats: Arc<WorkerStats>,
     ) -> Self {
         Self {
             id,
             pool: ActorPool::new(),
             transfer_rx,
             spawn_rx,
+            stats,
         }
     }
 
     /// Run one iteration of the worker loop. Returns `true` if any work was done.
-    pub fn tick_once(&mut self, tc: &TickContext) -> bool {
+    pub(crate) fn tick_once(&mut self, tc: &TickContext) -> bool {
         let mut did_work = false;
 
         // 1. Drain spawn queue → add actors to pool
@@ -65,6 +86,7 @@ impl Worker {
         let pending_local: RefCell<Vec<(ActorAddress, Box<dyn Any + Send>)>> =
             RefCell::new(Vec::new());
 
+        let processed;
         {
             let worker_ctx = WorkerContext {
                 worker_id: self.id,
@@ -76,7 +98,8 @@ impl Worker {
                 config: tc.config,
                 pending_local: &pending_local,
             };
-            if self.pool.tick_all(&worker_ctx) {
+            processed = self.pool.tick_all(&worker_ctx);
+            if processed > 0 {
                 did_work = true;
             }
         }
@@ -90,6 +113,11 @@ impl Worker {
             self.pool.deliver(&addr, msg);
         }
 
+        // 5. 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);
+
         did_work
     }
 
@@ -166,9 +194,25 @@ impl ContextInner for WorkerContext<'_> {
     }
 }
 
-/// Per-worker actor storage.
+/// How many messages to process this tick:
+/// - `len < waterlevel` → process all (`len`)
+/// - `len >= waterlevel` → process half (`len >> 1`)
+pub(crate) fn drain_count(len: usize, waterlevel: usize) -> usize {
+    if len < waterlevel {
+        len
+    } else {
+        len >> 1
+    }
+}
+
+struct ActorSlot {
+    mailbox: VecDeque<Box<dyn Any + Send>>,
+    actor: Box<dyn AnyActor>,
+}
+
+/// Per-worker actor storage. Owns per-actor mailboxes.
 pub(crate) struct ActorPool {
-    actors: HashMap<ActorAddress, Box<dyn AnyActor>>,
+    actors: HashMap<ActorAddress, ActorSlot>,
 }
 
 impl ActorPool {
@@ -179,42 +223,58 @@ impl ActorPool {
     }
 
     pub fn insert(&mut self, addr: ActorAddress, actor: Box<dyn AnyActor>) {
-        self.actors.insert(addr, actor);
+        self.actors.insert(addr, ActorSlot {
+            mailbox: VecDeque::new(),
+            actor,
+        });
     }
 
     pub fn remove(&mut self, addr: &ActorAddress) -> Option<Box<dyn AnyActor>> {
-        self.actors.remove(addr)
+        self.actors.remove(addr).map(|slot| slot.actor)
     }
 
     /// Deliver a type-erased message to the actor at `addr`.
-    /// Returns `true` if the actor was found and the message type matched.
+    /// Returns `true` if the actor exists (message is queued; type check deferred to tick).
     pub fn deliver(&mut self, addr: &ActorAddress, msg: Box<dyn Any + Send>) -> bool {
-        if let Some(actor) = self.actors.get_mut(addr) {
-            actor.deliver(msg)
+        if let Some(slot) = self.actors.get_mut(addr) {
+            slot.mailbox.push_back(msg);
+            true
         } else {
             false
         }
     }
 
-    /// Tick all actors in the pool. Returns `true` if any actor processed messages.
-    pub fn tick_all(&mut self, inner: &dyn ContextInner) -> bool {
-        let mut did_work = false;
-        for actor in self.actors.values_mut() {
-            if actor.tick(inner) {
-                did_work = true;
+    /// Tick all actors in the pool. Returns the number of messages processed.
+    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;
+                    }
+                }
             }
         }
-        did_work
+        count
     }
 
     pub fn len(&self) -> usize {
         self.actors.len()
     }
+
+    pub fn total_mailbox_depth(&self) -> usize {
+        self.actors.values().map(|slot| slot.mailbox.len()).sum()
+    }
 }
 
 
 
-pub struct Mailbox<M: Message> {
+pub(crate) struct Mailbox<M: Message> {
     queue: VecDeque<M>,
     waterlevel: usize,
 }
@@ -247,12 +307,9 @@ impl<M: Message> Mailbox<M> {
     /// - `len < waterlevel` → process all (`len`)
     /// - `len >= waterlevel` → process half (`len >> 1`)
     pub fn drain_count(&self) -> usize {
-        let len = self.queue.len();
-        if len < self.waterlevel {
-            len
-        } else {
-            len >> 1
-        }
+        drain_count(self.queue.len(), self.waterlevel)
     }
 }
 
+#[cfg(test)]
+mod tests;

src/worker/tests.rs

@@ -0,0 +1,420 @@
+use std::any::Any;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::thread;
+
+use crate::actor::{ActorAddress, AnyActor};
+use crate::address_map::{AddressMap, Placement, WorkerId};
+use crate::channel::Receiver;
+use crate::config::{BackoffPolicy, RuntimeConfig};
+use crate::runtime::{Ctx, Envelope, InboxRegistry};
+
+use super::{TickContext, Worker, WorkerStats};
+
+// ── Actors ─────────────────────────────────────────────────────────
+
+/// Counts how many u64 messages it successfully handled.
+struct CounterActor(Arc<AtomicUsize>);
+
+impl AnyActor for CounterActor {
+    fn handle_any(&mut self, _ctx: &Ctx, msg: Box<dyn Any + Send>) {
+        if msg.downcast::<u64>().is_ok() {
+            self.0.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+}
+
+// ── Harness ────────────────────────────────────────────────────────
+
+fn addr(id: u8) -> ActorAddress {
+    let mut bytes = [0u8; 32];
+    bytes[0] = id;
+    ActorAddress(bytes)
+}
+
+/// Self-contained single-worker test environment.
+///
+/// Holds the worker, its channels, shared state for TickContext, and
+/// per-actor handle counters — everything needed to drive scenarios.
+struct Env {
+    worker: Worker,
+    stats: Arc<WorkerStats>,
+    feed_transfer: crate::channel::Sender<Envelope>,
+    feed_spawn: crate::channel::Sender<(ActorAddress, Box<dyn AnyActor>)>,
+    address_map: AddressMap,
+    placement: Placement,
+    inbox_registry: InboxRegistry,
+    config: RuntimeConfig,
+    tc_transfer_txs: Vec<crate::channel::Sender<Envelope>>,
+    tc_spawn_txs: Vec<crate::channel::Sender<(ActorAddress, Box<dyn AnyActor>)>>,
+    counters: HashMap<u8, Arc<AtomicUsize>>,
+}
+
+impl Env {
+    fn new() -> Self {
+        Self::with_config(RuntimeConfig::default())
+    }
+
+    fn with_config(config: RuntimeConfig) -> Self {
+        let transfer_rx = Receiver::<Envelope>::new(256);
+        let spawn_rx = Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(256);
+
+        let feed_transfer = transfer_rx.new_sender();
+        let feed_spawn = spawn_rx.new_sender();
+        let tc_transfer = transfer_rx.new_sender();
+        let tc_spawn = spawn_rx.new_sender();
+
+        let stats = Arc::new(WorkerStats::new());
+        let worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats.clone());
+
+        Self {
+            worker,
+            stats,
+            feed_transfer,
+            feed_spawn,
+            address_map: AddressMap::new(),
+            placement: Placement::new(1),
+            inbox_registry: InboxRegistry::new(),
+            config,
+            tc_transfer_txs: vec![tc_transfer],
+            tc_spawn_txs: vec![tc_spawn],
+            counters: HashMap::new(),
+        }
+    }
+
+    /// Enqueue an actor spawn (drained on next tick, phase 1).
+    fn spawn(&mut self, id: u8) {
+        let counter = Arc::new(AtomicUsize::new(0));
+        let actor: Box<dyn AnyActor> = Box::new(CounterActor(counter.clone()));
+        self.feed_spawn.try_send((addr(id), actor)).ok().unwrap();
+        self.counters.insert(id, counter);
+    }
+
+    /// Enqueue a u64 message (drained on next tick, phase 2).
+    fn send(&self, id: u8, val: u64) {
+        self.feed_transfer
+            .try_send(Envelope::new(addr(id), Box::new(val)))
+            .ok()
+            .unwrap();
+    }
+
+    /// Enqueue a wrong-typed message (String instead of u64).
+    fn send_bad(&self, id: u8) {
+        self.feed_transfer
+            .try_send(Envelope::new(addr(id), Box::new("bad".to_string())))
+            .ok()
+            .unwrap();
+    }
+
+    /// Remove an actor from the pool (immediate, no tick needed).
+    fn remove(&mut self, id: u8) {
+        self.worker.pool.remove(&addr(id));
+    }
+
+    /// Run one tick of the worker loop.
+    fn tick(&mut self) {
+        let tc = TickContext {
+            address_map: &self.address_map,
+            transfer_txs: &self.tc_transfer_txs,
+            spawn_txs: &self.tc_spawn_txs,
+            placement: &self.placement,
+            inbox_registry: &self.inbox_registry,
+            config: &self.config,
+        };
+        self.worker.tick_once(&tc);
+    }
+
+    // ── Readouts ───────────────────────────────────────────────────
+
+    fn handled(&self, id: u8) -> usize {
+        self.counters[&id].load(Ordering::Relaxed)
+    }
+
+    fn pool_len(&self) -> usize {
+        self.worker.pool.len()
+    }
+
+    fn depth(&self) -> usize {
+        self.stats.total_mailbox_depth.load(Ordering::Relaxed)
+    }
+
+    fn processed(&self) -> u64 {
+        self.stats.messages_processed.load(Ordering::Relaxed)
+    }
+
+    fn num_actors_stat(&self) -> usize {
+        self.stats.num_actors.load(Ordering::Relaxed)
+    }
+}
+
+// ── Step-driven runner ─────────────────────────────────────────────
+
+enum Step {
+    Spawn(u8),
+    Send(u8, u64),
+    SendBad(u8),
+    Remove(u8),
+    Tick,
+    Expect { pool_len: usize, depth: usize, processed: u64 },
+    ExpectHandled(u8, usize),
+}
+
+fn run(steps: &[Step]) {
+    run_with(RuntimeConfig::default(), steps);
+}
+
+fn run_with(config: RuntimeConfig, steps: &[Step]) {
+    let mut env = Env::with_config(config);
+    for (i, step) in steps.iter().enumerate() {
+        match step {
+            Step::Spawn(id) => env.spawn(*id),
+            Step::Send(id, val) => env.send(*id, *val),
+            Step::SendBad(id) => env.send_bad(*id),
+            Step::Remove(id) => env.remove(*id),
+            Step::Tick => env.tick(),
+            Step::Expect { pool_len, depth, processed } => {
+                assert_eq!(env.pool_len(), *pool_len, "step {i}: pool_len");
+                assert_eq!(env.depth(), *depth, "step {i}: depth");
+                assert_eq!(env.processed(), *processed, "step {i}: processed");
+            }
+            Step::ExpectHandled(id, n) => {
+                assert_eq!(env.handled(*id), *n, "step {i}: handled({})", id);
+            }
+        }
+    }
+}
+
+// ── Tests ──────────────────────────────────────────────────────────
+
+#[test]
+fn spawn_send_process() {
+    run(&[
+        Step::Spawn(1),
+        Step::Spawn(2),
+        Step::Tick,
+        Step::Expect { pool_len: 2, depth: 0, processed: 0 },
+
+        Step::Send(1, 10),
+        Step::Send(1, 20),
+        Step::Send(2, 30),
+        Step::Tick,
+        Step::Expect { pool_len: 2, depth: 0, processed: 3 },
+        Step::ExpectHandled(1, 2),
+        Step::ExpectHandled(2, 1),
+    ]);
+}
+
+#[test]
+fn remove_drops_future_messages() {
+    run(&[
+        Step::Spawn(1),
+        Step::Spawn(2),
+        Step::Tick,
+
+        // Remove actor 1 directly from pool
+        Step::Remove(1),
+        Step::Expect { pool_len: 1, depth: 0, processed: 0 },
+
+        // Messages to actor 1 are drained from the transfer queue
+        // but pool.deliver finds no slot — silently dropped
+        Step::Send(1, 42),
+        Step::Send(2, 99),
+        Step::Tick,
+
+        Step::Expect { pool_len: 1, depth: 0, processed: 1 },
+        Step::ExpectHandled(1, 0),
+        Step::ExpectHandled(2, 1),
+    ]);
+}
+
+#[test]
+fn wrong_type_silently_dropped() {
+    run(&[
+        Step::Spawn(1),
+        Step::Tick,
+
+        // Mix correct (u64) and incorrect (String) types
+        Step::Send(1, 1),
+        Step::SendBad(1),
+        Step::Send(1, 2),
+        Step::SendBad(1),
+        Step::SendBad(1),
+        Step::Send(1, 3),
+        Step::Tick,
+
+        // All 6 popped from mailbox ("processed" by the pool),
+        // but only the 3 u64 messages were handled by the actor
+        Step::Expect { pool_len: 1, depth: 0, processed: 6 },
+        Step::ExpectHandled(1, 3),
+    ]);
+}
+
+#[test]
+fn backpressure_drains_half() {
+    let config = RuntimeConfig {
+        mailbox_waterlevel: 4,
+        ..Default::default()
+    };
+
+    run_with(config, &[
+        Step::Spawn(1),
+        Step::Tick,
+
+        // Send 10 messages
+        Step::Send(1, 0), Step::Send(1, 1), Step::Send(1, 2), Step::Send(1, 3),
+        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)
+        Step::Tick,
+        Step::Expect { pool_len: 1, depth: 0, processed: 10 },
+        Step::ExpectHandled(1, 10),
+    ]);
+}
+
+#[test]
+fn spawn_and_send_same_tick() {
+    // Spawn is phase 1, transfer is phase 2, processing is phase 3.
+    // All three happen within a single tick_once call.
+    run(&[
+        Step::Spawn(1),
+        Step::Send(1, 42),
+        Step::Tick,
+        Step::Expect { pool_len: 1, depth: 0, processed: 1 },
+        Step::ExpectHandled(1, 1),
+    ]);
+}
+
+#[test]
+fn stats_track_pool_mutations() {
+    let mut env = Env::new();
+
+    // Before any tick, stats are zeroed
+    assert_eq!(env.num_actors_stat(), 0);
+    assert_eq!(env.depth(), 0);
+    assert_eq!(env.processed(), 0);
+
+    // Spawn 3 + tick → stats reflect 3 actors
+    env.spawn(1);
+    env.spawn(2);
+    env.spawn(3);
+    env.tick();
+    assert_eq!(env.num_actors_stat(), 3);
+
+    // Send 5 to actor 1 + tick → processed increases
+    for i in 0..5 {
+        env.send(1, i);
+    }
+    env.tick();
+    assert_eq!(env.processed(), 5);
+    assert_eq!(env.depth(), 0);
+    assert_eq!(env.num_actors_stat(), 3);
+
+    // Remove actor 2 + tick → stats update
+    env.remove(2);
+    env.tick();
+    assert_eq!(env.num_actors_stat(), 2);
+    assert_eq!(env.pool_len(), 2);
+}
+
+/// A long mixed-action sequence: spawns, sends, removes, wrong types,
+/// and backpressure — all in one run.
+#[test]
+fn interleaved_lifecycle() {
+    run(&[
+        // ── Phase 1: build the pool ────────────────────────────────
+        Step::Spawn(1),
+        Step::Spawn(2),
+        Step::Spawn(3),
+        Step::Tick,
+        Step::Expect { pool_len: 3, depth: 0, processed: 0 },
+
+        // ── Phase 2: normal message flow ───────────────────────────
+        Step::Send(1, 100),
+        Step::Send(2, 200),
+        Step::Send(3, 300),
+        Step::Tick,
+        Step::ExpectHandled(1, 1),
+        Step::ExpectHandled(2, 1),
+        Step::ExpectHandled(3, 1),
+        Step::Expect { pool_len: 3, depth: 0, processed: 3 },
+
+        // ── Phase 3: remove actor 2, send to all 3 ────────────────
+        Step::Remove(2),
+        Step::Send(1, 101),
+        Step::Send(2, 201), // actor 2 gone — dropped at deliver
+        Step::Send(3, 301),
+        Step::Tick,
+        Step::Expect { pool_len: 2, depth: 0, processed: 5 },
+        Step::ExpectHandled(1, 2),
+        Step::ExpectHandled(2, 1), // unchanged since removal
+        Step::ExpectHandled(3, 2),
+
+        // ── Phase 4: late spawn + immediate send ───────────────────
+        Step::Spawn(4),
+        Step::Send(4, 400),
+        Step::Tick,
+        Step::Expect { pool_len: 3, depth: 0, processed: 6 },
+        Step::ExpectHandled(4, 1),
+
+        // ── Phase 5: bad types mixed with good ─────────────────────
+        Step::SendBad(1),
+        Step::SendBad(1),
+        Step::SendBad(1),
+        Step::Send(1, 999),
+        Step::Tick,
+        // 4 popped (3 bad + 1 good), only 1 handled by actor
+        Step::Expect { pool_len: 3, depth: 0, processed: 10 },
+        Step::ExpectHandled(1, 3), // 2 from prior phases + 1 good
+
+        // ── Phase 6: remove all, send to ghosts ────────────────────
+        Step::Remove(1),
+        Step::Remove(3),
+        Step::Remove(4),
+        Step::Expect { pool_len: 0, depth: 0, processed: 10 },
+        Step::Send(1, 0),
+        Step::Send(3, 0),
+        Step::Tick,
+        Step::Expect { pool_len: 0, depth: 0, processed: 10 },
+    ]);
+}
+
+#[test]
+fn run_loop_stops_on_shutdown() {
+    let transfer_rx = Receiver::<Envelope>::new(64);
+    let spawn_rx = Receiver::<(ActorAddress, Box<dyn AnyActor>)>::new(64);
+    let transfer_tx = transfer_rx.new_sender();
+    let spawn_tx = spawn_rx.new_sender();
+
+    let stats = Arc::new(WorkerStats::new());
+    let mut worker = Worker::new(WorkerId(0), transfer_rx, spawn_rx, stats);
+
+    let is_running = AtomicBool::new(false);
+    let backoff = BackoffPolicy::default();
+    let address_map = AddressMap::new();
+    let placement = Placement::new(1);
+    let inbox_registry = InboxRegistry::new();
+    let config = RuntimeConfig::default();
+
+    let tc = TickContext {
+        address_map: &address_map,
+        transfer_txs: &[transfer_tx],
+        spawn_txs: &[spawn_tx],
+        placement: &placement,
+        inbox_registry: &inbox_registry,
+        config: &config,
+    };
+
+    thread::scope(|s| {
+        s.spawn(|| worker.run(&tc, &is_running, &backoff));
+    });
+}

tests/stats_demo.rs

@@ -0,0 +1,136 @@
+use swactor::{
+    actor::{ActorAddress, ActorInterface},
+    runtime::{Ctx, Runtime, RuntimeConfig},
+};
+
+#[derive(Clone)]
+struct Ping {
+    reply_to: ActorAddress,
+}
+
+#[derive(Clone)]
+struct Pong;
+
+struct PingActor;
+
+impl ActorInterface for PingActor {
+    type Incoming = Ping;
+    type Response = Pong;
+
+    fn handle(&mut self, ctx: &Ctx, msg: Ping) {
+        let _ = ctx.send(msg.reply_to, Pong);
+    }
+}
+
+/// Counter that just counts messages.
+struct Counter(u64);
+
+impl ActorInterface for Counter {
+    type Incoming = u64;
+    type Response = ();
+
+    fn handle(&mut self, _ctx: &Ctx, _msg: u64) {
+        self.0 += 1;
+    }
+}
+
+#[test]
+fn stats_demo_single_thread() {
+    let rt = Runtime::new(RuntimeConfig::default());
+
+    // Spawn a few actors
+    let ping1 = rt.spawn(PingActor).unwrap();
+    let ping2 = rt.spawn(PingActor).unwrap();
+    let counter = rt.spawn(Counter(0)).unwrap();
+
+    // Send some messages (they queue up before we tick)
+    for i in 0..20u64 {
+        rt.send_to(counter, i).unwrap();
+    }
+
+    // Stats BEFORE ticking — messages are in the transfer queue, not yet in mailboxes
+    let s = rt.stats();
+    println!("=== Before any ticks ===");
+    print_stats(&s);
+
+    // Tick once — drains transfer queue into mailboxes, then processes messages
+    rt.tick();
+
+    let s = rt.stats();
+    println!("\n=== After 1 tick ===");
+    print_stats(&s);
+
+    // Tick a few more times to drain remaining messages
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    let s = rt.stats();
+    println!("\n=== After 6 ticks total ===");
+    print_stats(&s);
+
+    assert_eq!(s.num_workers, 1);
+    assert_eq!(s.actors.len(), 3);
+    assert_eq!(s.workers[0].num_actors, 3);
+    // All 20 messages should be processed by now
+    assert_eq!(s.workers[0].mailbox_depth, 0);
+    assert!(s.workers[0].messages_processed >= 20);
+}
+
+#[test]
+fn stats_demo_multi_thread() {
+    let config = RuntimeConfig {
+        num_threads: 3,
+        ..Default::default()
+    };
+    let rt = Runtime::new(config);
+
+    // Spawn actors — round-robin will spread them across 3 workers
+    let mut addrs = Vec::new();
+    for _ in 0..6 {
+        addrs.push(rt.spawn(Counter(0)).unwrap());
+    }
+
+    // Send messages to each actor
+    for &addr in &addrs {
+        for i in 0..10u64 {
+            rt.send_to(addr, i).unwrap();
+        }
+    }
+
+    let handle = rt.run().unwrap();
+
+    // Let it process
+    std::thread::sleep(std::time::Duration::from_millis(50));
+
+    let s = handle.runtime.stats();
+    println!("\n=== Multi-threaded (3 workers, 6 actors, 60 messages) ===");
+    print_stats(&s);
+
+    handle.shutdown();
+    handle.join();
+
+    assert_eq!(s.num_workers, 3);
+    assert_eq!(s.actors.len(), 6);
+    let total_processed: u64 = s.workers.iter().map(|w| w.messages_processed).sum();
+    assert_eq!(total_processed, 60);
+}
+
+fn print_stats(s: &swactor::runtime::RuntimeStats) {
+    println!(
+        "RuntimeStats(actors={}, workers={})",
+        s.actors.len(),
+        s.num_workers
+    );
+    for w in &s.workers {
+        println!(
+            "  Worker {}: {} actors, {} queued, {} processed",
+            w.id, w.num_actors, w.mailbox_depth, w.messages_processed
+        );
+        for (addr, wid) in &s.actors {
+            if *wid == w.id {
+                println!("    - {:x?}...", &addr.0[..4]);
+            }
+        }
+    }
+}