feat(scheduler,gen_server,gen_statem,introspect): SpawnOpts — per-actor stack shape on every spawn surface (RFC 019 §1)

SpawnOpts { stack_reserve, guard_size } with Option<usize> fields, None
resolving to the Config defaults at spawn time — a deliberate deviation
from the RFC's plain-usize struct so struct-update syntax works without
a runtime handle in scope. Threaded across the five surfaces:
spawn_with, spawn_under_with, spawn_addr_with,
GenServerBuilder::stack_opts (mirrored on NamedGenServerBuilder), and
gen_statem::spawn_with (gen_statem has no builder, so the opts ride a
_with variant — Claude-solo surface call, flagged for review). Existing
spawns forward defaults; no call-site churn.

introspect::stack_shape(pid) pulled forward (agreed) as the first slice
of the RFC 019 introspection surface, giving tests an observable.

Tests (tests/spawn_opts.rs): override/partial-override/rounding on each
surface; obligation 4 from the outside — a dead custom stack is never
handed to the next default spawn (LIFO pool would expose it), and the
reverse (default stacks ARE recycled); 8 MiB reserve behaviorally
permits ~1 MiB recursion. Also: silence unused-Result in the c1
runtime test (join now unwrapped).

tests/spawn_opts.rs

@@ -0,0 +1,212 @@
+//! RFC 019 commit 2 — the `SpawnOpts` surface.
+//!
+//! Covers: per-spawn stack shape overrides on every spawn surface, the
+//! `None ⇒ Config default` resolution, the pool rule from the outside
+//! (obligation 4: a custom-shaped stack never enters the pool), and that a
+//! big reserve behaviorally takes effect (deep recursion completes).
+
+use smarm::runtime::{Config, DEFAULT_STACK_GUARD, DEFAULT_STACK_RESERVE};
+use smarm::{
+    self_pid, spawn, spawn_under_with, spawn_with, GenServerBuilder, SpawnOpts,
+};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+fn rt1() -> smarm::runtime::Runtime {
+    smarm::runtime::init(Config::exact(1))
+}
+
+#[test]
+fn default_spawn_has_default_shape() {
+    rt1().run(|| {
+        let h = spawn(|| {
+            let shape = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(shape, (DEFAULT_STACK_RESERVE, DEFAULT_STACK_GUARD));
+        });
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn spawn_with_overrides_reserve_and_guard() {
+    rt1().run(|| {
+        let opts = SpawnOpts {
+            stack_reserve: Some(1024 * 1024),
+            guard_size: Some(256 * 1024),
+        };
+        let h = spawn_with(opts, || {
+            let shape = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(shape, (1024 * 1024, 256 * 1024));
+        });
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn spawn_with_partial_override_keeps_config_default_for_the_rest() {
+    rt1().run(|| {
+        let opts = SpawnOpts { stack_reserve: Some(1024 * 1024), ..SpawnOpts::default() };
+        let h = spawn_with(opts, || {
+            let shape = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(shape, (1024 * 1024, DEFAULT_STACK_GUARD));
+        });
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn spawn_with_rounds_to_pages() {
+    rt1().run(|| {
+        let opts = SpawnOpts { stack_reserve: Some(64 * 1024 + 1), guard_size: Some(4097) };
+        let h = spawn_with(opts, || {
+            let (reserve, guard) = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(reserve % 4096, 0);
+            assert_eq!(guard % 4096, 0);
+            assert!(reserve >= 64 * 1024 + 1);
+            assert!(guard >= 4097);
+        });
+        h.join().unwrap();
+    });
+}
+
+#[test]
+fn spawn_under_with_takes_opts() {
+    rt1().run(|| {
+        let me = self_pid();
+        let opts = SpawnOpts { stack_reserve: Some(128 * 1024), ..SpawnOpts::default() };
+        let h = spawn_under_with(me, opts, || {
+            let (reserve, _) = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(reserve, 128 * 1024);
+        });
+        h.join().unwrap();
+    });
+}
+
+/// Obligation 4, from the outside: a dead custom stack must not be handed to
+/// the next default spawn. The pool is LIFO, so if the custom stack had been
+/// (wrongly) pushed at death, the very next default-shaped spawn on this
+/// single-threaded runtime would pop it and report a custom shape.
+#[test]
+fn custom_stack_never_enters_the_pool() {
+    rt1().run(|| {
+        spawn_with(
+            SpawnOpts { stack_reserve: Some(512 * 1024), guard_size: Some(128 * 1024) },
+            || {},
+        )
+        .join()
+        .unwrap();
+        let h = spawn(|| {
+            let shape = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(shape, (DEFAULT_STACK_RESERVE, DEFAULT_STACK_GUARD));
+        });
+        h.join().unwrap();
+    });
+}
+
+/// The reverse direction of the pool rule: a default-shaped stack IS pooled
+/// and reused (cap = threads × 4 ≥ 1 here, pool empty at start).
+#[test]
+fn default_stack_is_recycled() {
+    rt1().run(|| {
+        spawn(|| {}).join().unwrap();
+        let h = spawn(|| {
+            let shape = smarm::introspect::stack_shape(self_pid()).unwrap();
+            assert_eq!(shape, (DEFAULT_STACK_RESERVE, DEFAULT_STACK_GUARD));
+        });
+        h.join().unwrap();
+    });
+}
+
+/// Burn ~`frames` × 4 KiB of stack (see tests/runtime.rs twin).
+#[inline(never)]
+fn burn_stack(frames: usize) -> u64 {
+    let mut local = [0u8; 4096];
+    local[0] = frames as u8;
+    let below = if frames == 0 { 0 } else { burn_stack(frames - 1) };
+    std::hint::black_box(&mut local);
+    below.wrapping_add(local[0] as u64)
+}
+
+#[test]
+fn big_reserve_behaviorally_takes_effect() {
+    // ~1 MiB deep on an 8 MiB per-spawn reserve, runtime default untouched.
+    rt1().run(|| {
+        let done = Arc::new(AtomicBool::new(false));
+        let done2 = done.clone();
+        spawn_with(
+            SpawnOpts { stack_reserve: Some(8 * 1024 * 1024), ..SpawnOpts::default() },
+            move || {
+                std::hint::black_box(burn_stack(256));
+                done2.store(true, Ordering::SeqCst);
+            },
+        )
+        .join()
+        .unwrap();
+        assert!(done.load(Ordering::SeqCst));
+    });
+}
+
+// ---------------------------------------------------------------------------
+// Builder surfaces
+// ---------------------------------------------------------------------------
+
+struct Echo;
+impl smarm::GenServer for Echo {
+    type Call = ();
+    type Reply = (usize, usize);
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+    fn handle_call(&mut self, _c: ()) -> (usize, usize) {
+        smarm::introspect::stack_shape(self_pid()).unwrap()
+    }
+    fn handle_cast(&mut self, _c: ()) {}
+}
+
+#[test]
+fn gen_server_builder_stack_opts() {
+    rt1().run(|| {
+        let server = GenServerBuilder::new(Echo)
+            .stack_opts(SpawnOpts { stack_reserve: Some(256 * 1024), ..SpawnOpts::default() })
+            .start();
+        let (reserve, guard) = server.call(()).unwrap();
+        assert_eq!(reserve, 256 * 1024);
+        assert_eq!(guard, DEFAULT_STACK_GUARD);
+        server.shutdown();
+    });
+}
+
+struct Probe;
+impl smarm::Machine for Probe {
+    type Ev = smarm::channel::Sender<(usize, usize)>;
+    fn state_timeout_ev() -> Self::Ev {
+        unreachable!("no timers in this test")
+    }
+    fn timeout_ev(_name: &'static str) -> Self::Ev {
+        unreachable!("no timers in this test")
+    }
+    fn on_start(&mut self, _cx: &mut smarm::Cx<Self::Ev>) {}
+    fn handle(
+        &mut self,
+        ev: Self::Ev,
+        _cx: &mut smarm::Cx<Self::Ev>,
+    ) -> smarm::gen_statem::Step<Self::Ev> {
+        let _ = ev.send(smarm::introspect::stack_shape(self_pid()).unwrap());
+        smarm::gen_statem::Step::Stayed
+    }
+}
+
+#[test]
+fn gen_statem_spawn_with_stack_opts() {
+    rt1().run(|| {
+        let m = smarm::gen_statem::spawn_with(
+            SpawnOpts { stack_reserve: Some(256 * 1024), ..SpawnOpts::default() },
+            Probe,
+        );
+        let (tx, rx) = smarm::channel::channel();
+        m.send(tx).unwrap();
+        let (reserve, guard) = rx.recv().unwrap();
+        assert_eq!(reserve, 256 * 1024);
+        assert_eq!(guard, DEFAULT_STACK_GUARD);
+    });
+}