fix(runtime): wake idle sibling schedulers at termination

An idle scheduler thread blocks in poll_wake on a snapshot of
peek_deadline / io_outstanding. Only the io threads ever write the wake
pipe (on completion push); enqueue does not. Mid-flight that is masked —
the thread that caused an enqueue is awake and processes it — but at
termination the snapshot can go terminally stale, two ways:

- Stale io_outstanding (infinite hang): an actor parked in wait_readable
  is request_stop'ped. Cancellation deregisters the waiter and produces
  NO completion, so the wake pipe is never written. A sibling blocked on
  io_outstanding > 0 with no timers pending sits in poll(-1) forever.

- Orphaned timer deadline (finite stall): an actor cancelled out of a
  long sleep leaves its timer entry behind. Clearing the wheel at
  AllDone doesn't help a sibling that already blocked on that deadline;
  it sleeps it out in full.

In both cases the remaining actors finish on the other scheduler thread,
which reaches the live==0 && io_out==0 verdict and returns without
waking the blocked one. Runtime::run then stalls or hangs in its worker
join. Found via urus's graceful-shutdown tests (~10% flake, one test per
variant); confirmed by gdb dumps of hung processes showing run() in
JoinHandle::join over a sibling in poll(wake_fd, 59750ms) resp. poll(-1).

Fix: Io::wake() writes the wake pipe directly; the AllDone arm calls it
after the timer clear. One byte wakes every poller; each re-runs the
verdict, independently reaches AllDone, and re-wakes — idempotent.

tests/terminal_wake.rs reproduces both variants deterministically: the
root busy-spins (no timer entries, occupies one scheduler thread) so the
sibling settles into the stale idle wait before the stop is issued. Both
hang without the fix and pass in <0.5s with it.

Known residual gap, deliberately unfixed: the timers-pending /
no-io-subsystem idle branch blocks in thread::sleep and has no wake
mechanism at all — the same stall exists for runtimes that never
initialize io. Roadmap candidate alongside cross-thread unpark and
entry-side check_cancelled in park_current (the lossy-QUEUED-stop bug).

src/io.rs

@@ -260,6 +260,15 @@ impl IoThread {
         self.wake_read
     }
 
+    /// Write the wake pipe directly: rouse every scheduler thread blocked in
+    /// its idle `poll_wake`. Used by the terminal (AllDone) path — an idle
+    /// sibling may be blocked on a snapshot that nothing will ever refresh
+    /// (an orphaned timer deadline, or `io_outstanding` from a waiter that
+    /// was stop-cancelled and so never produces a completion).
+    pub fn wake(&self) {
+        wake_scheduler(self.wake_write);
+    }
+
     /// Register interest in `fd` becoming readable/writable; record `pid`
     /// as the parked waiter. The epoll thread will push a `FdReady`
     /// completion when the kernel signals.

src/runtime.rs

@@ -1238,6 +1238,17 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                     // woken by them — e.g. a sleeper cancelled out of its sleep);
                     // they must not keep the runtime alive. Drop them on the way out.
                     inner.timers.lock().unwrap().clear();
+                    // Terminal wake: a sibling scheduler may be blocked in its
+                    // idle wait on a snapshot that is now terminally stale — an
+                    // orphaned long deadline (it would sleep it out in full) or
+                    // a stale `io_outstanding > 0` from a stop-cancelled waiter
+                    // (it would block in poll(-1) forever; cancellation produces
+                    // no completion, so nothing else writes the wake pipe).
+                    // One byte wakes every poller; each re-runs the verdict,
+                    // reaches AllDone itself, and re-wakes — idempotent.
+                    if let Some(io) = inner.io.lock().unwrap().as_ref() {
+                        io.wake();
+                    }
                     return;
                 }
                 Pop::Idle { io_outstanding, wake_fd } => {

tests/terminal_wake.rs

@@ -0,0 +1,115 @@
+//! Terminal wake: a scheduler thread blocked in the idle wait must be woken
+//! when a sibling reaches the all-done verdict, or `rt.run` stalls in its
+//! worker join.
+//!
+//! Mechanism (runtime.rs, `Pop::Idle`): an idle scheduler snapshots
+//! `peek_deadline()` / `io_outstanding` and blocks in `poll_wake` (or
+//! `thread::sleep`) on that snapshot. `enqueue` does not write the wake
+//! pipe — only IO completions do — so the snapshot can go terminally stale:
+//!
+//! - An actor parked in `wait_readable` that is `request_stop`ped produces
+//!   NO completion (cancellation deregisters the waiter); a sibling that
+//!   blocked on `io_outstanding > 0` with no timers is in `poll(-1)` forever.
+//! - An actor cancelled out of a long `sleep` leaves its timer entry
+//!   orphaned; a sibling that blocked on that deadline sleeps it out in full.
+//!
+//! In both cases the remaining work completes on the *other* scheduler
+//! thread, which hits AllDone and returns — and nothing wakes the blocked
+//! one. `Runtime::run` joins it: a permanent hang in the first case, a
+//! full-deadline stall in the second.
+//!
+//! Both tests force the window deterministically: the root busy-spins
+//! (creating no timer entries and occupying one scheduler thread) so the
+//! other thread settles into the stale idle wait before the stop is issued.
+
+use std::sync::mpsc;
+use std::time::{Duration, Instant};
+
+struct PipePair {
+    read: libc::c_int,
+    write: libc::c_int,
+}
+
+impl PipePair {
+    fn new() -> Self {
+        let mut fds: [libc::c_int; 2] = [0; 2];
+        let r = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
+        assert_eq!(r, 0, "pipe2 failed");
+        PipePair { read: fds[0], write: fds[1] }
+    }
+}
+
+impl Drop for PipePair {
+    fn drop(&mut self) {
+        unsafe {
+            libc::close(self.read);
+            libc::close(self.write);
+        }
+    }
+}
+
+/// Occupy the current scheduler thread without creating timer entries or
+/// parking. A plain spin has no observation points, so the root stays
+/// on-CPU and the sibling scheduler is left alone with the idle branch.
+fn spin_for(d: Duration) {
+    let t0 = Instant::now();
+    while t0.elapsed() < d {
+        std::hint::spin_loop();
+    }
+}
+
+/// Run `body` under a 2-scheduler runtime on a watchdog thread; fail if
+/// `Runtime::run` has not returned within `limit`.
+fn run_with_watchdog(limit: Duration, body: impl FnOnce() + Send + 'static) {
+    let (done_tx, done_rx) = mpsc::channel::<()>();
+    std::thread::spawn(move || {
+        let rt = smarm::init(smarm::Config::exact(2));
+        rt.run(body);
+        let _ = done_tx.send(());
+    });
+    done_rx
+        .recv_timeout(limit)
+        .expect("Runtime::run did not return: idle scheduler thread was never woken at termination");
+}
+
+/// Permanent-hang variant: sibling blocked in `poll_wake(wake_fd, None)`
+/// because `io_outstanding > 0` (one actor parked in `wait_readable`) and no
+/// timers are pending. The waiter is then stop-cancelled — no IO completion
+/// ever writes the wake pipe — and everything else finishes on the root's
+/// thread. Without a terminal wake, `rt.run` never returns.
+#[test]
+fn run_returns_after_io_waiter_is_stop_cancelled() {
+    run_with_watchdog(Duration::from_secs(10), || {
+        let pipe = PipePair::new();
+        let rfd = pipe.read;
+        let h = smarm::spawn(move || {
+            // Never-readable fd (write end open, nothing written).
+            let _ = smarm::wait_readable(rfd);
+        });
+        // Let the waiter park and the sibling scheduler settle into the
+        // io_outstanding>0 / no-timers idle wait: poll(wake_fd, -1).
+        spin_for(Duration::from_millis(200));
+        smarm::request_stop(h.pid());
+        let _ = h.join();
+        drop(pipe);
+    });
+}
+
+/// Finite-stall variant: sibling blocked on an orphaned long timer
+/// deadline. An actor cancelled out of `sleep(60s)` leaves its timer entry
+/// behind (documented as harmless at AllDone — but a sibling that already
+/// blocked on that deadline sleeps it out in full, stalling `rt.run` for
+/// the better part of a minute).
+#[test]
+fn run_returns_after_long_sleeper_is_stop_cancelled() {
+    run_with_watchdog(Duration::from_secs(10), || {
+        let h = smarm::spawn(|| {
+            smarm::sleep(Duration::from_secs(60));
+        });
+        // Let the sleeper park and the sibling scheduler block on the 60s
+        // deadline: poll(wake_fd, ~60_000ms).
+        spin_for(Duration::from_millis(200));
+        smarm::request_stop(h.pid());
+        let _ = h.join();
+    });
+}