RFC 016 Chunk 4: observer gen_server (feature-gated)

A thin GenServer consumer of the Chunk-1 read primitive — the live
observer process (D12). ObserverRequest/ObserverReply are the wire
contract (D11); the version rides along on the snapshot/tree payloads,
which already carry SNAPSHOT_FORMAT_VERSION (D1). Behind the new
`observer` Cargo feature, off by default (D10): the primitive stays
always-on, only the transport is gated. Cast is Infallible, so the
server takes no async traffic and handle_cast is statically unreachable.

Gated integration test proves each verb relays exactly what the
corresponding primitive returns (snapshot/tree/actor_info), incl. a
forged-pid None and a live Parked classification.

tests/observer.rs

@@ -0,0 +1,121 @@
+//! RFC 016 Chunk 4 — the observer gen_server. The whole file is gated on the
+//! `observer` feature (run with `cargo test --features observer`); without it
+//! the module does not exist and there is nothing to compile.
+//!
+//! The point these prove: the observer is *transport over the same reads*. Each
+//! verb returns exactly what the corresponding Chunk-1 primitive would, just
+//! marshalled over the gen_server call channel — so a known spawned actor that
+//! `snapshot()` / `actor_info()` would see is equally visible through the
+//! observer.
+#![cfg(feature = "observer")]
+
+use smarm::observer::{self, ObserverReply, ObserverRequest};
+use smarm::{channel, run, ActorState, SNAPSHOT_FORMAT_VERSION};
+
+#[test]
+fn observer_relays_snapshot_tree_and_actor_info() {
+    run(|| {
+        // A worker parked on an empty gate: a known, stable actor for the
+        // observer to find across all three verbs.
+        let (ready_tx, ready_rx) = channel::<()>();
+        let (gate_tx, gate_rx) = channel::<()>();
+        let worker = smarm::spawn(move || {
+            ready_tx.send(()).unwrap();
+            gate_rx.recv().unwrap();
+        });
+        ready_rx.recv().unwrap();
+
+        let obs = observer::start();
+
+        // Snapshot: the worker is present, and so is the observer itself — it
+        // is a scheduled actor like any other.
+        let ObserverReply::Snapshot(snap) = obs.call(ObserverRequest::Snapshot).unwrap() else {
+            panic!("Snapshot verb must reply Snapshot");
+        };
+        assert_eq!(snap.format_version, SNAPSHOT_FORMAT_VERSION);
+        assert!(
+            snap.actors.iter().any(|a| a.pid == worker.pid()),
+            "observer's snapshot should contain the spawned worker"
+        );
+        assert!(
+            snap.actors.iter().any(|a| a.pid == obs.pid()),
+            "observer should appear in the snapshot it produced"
+        );
+
+        // Tree: same data folded into the parentage forest, same version.
+        let ObserverReply::Tree(t) = obs.call(ObserverRequest::Tree).unwrap() else {
+            panic!("Tree verb must reply Tree");
+        };
+        assert_eq!(t.format_version, SNAPSHOT_FORMAT_VERSION);
+        fn contains(nodes: &[smarm::TreeNode], pid: smarm::Pid) -> bool {
+            nodes
+                .iter()
+                .any(|n| n.info.pid == pid || contains(&n.children, pid))
+        }
+        assert!(
+            contains(&t.roots, worker.pid()),
+            "worker should appear somewhere in the observer's tree"
+        );
+
+        // ActorInfo: coherent single-actor view, matching a direct read.
+        let ObserverReply::ActorInfo(Some(info)) =
+            obs.call(ObserverRequest::ActorInfo(worker.pid())).unwrap()
+        else {
+            panic!("ActorInfo verb must reply ActorInfo(Some) for a live worker");
+        };
+        assert_eq!(info.pid, worker.pid());
+
+        gate_tx.send(()).unwrap();
+        worker.join().unwrap();
+    });
+}
+
+#[test]
+fn observer_reports_none_for_a_forged_pid() {
+    run(|| {
+        let obs = observer::start();
+        // An index that is not in the slab at all — the verb relays the
+        // primitive's `None` faithfully.
+        let forged = smarm::Pid::new(u32::MAX - 1, 0);
+        let ObserverReply::ActorInfo(none) =
+            obs.call(ObserverRequest::ActorInfo(forged)).unwrap()
+        else {
+            panic!("ActorInfo verb must reply ActorInfo");
+        };
+        assert!(none.is_none(), "a forged pid should relay as None");
+    });
+}
+
+#[test]
+fn observer_sees_a_parked_actor_as_parked() {
+    run(|| {
+        let (ready_tx, ready_rx) = channel::<()>();
+        let (gate_tx, gate_rx) = channel::<()>();
+        let worker = smarm::spawn(move || {
+            ready_tx.send(()).unwrap();
+            gate_rx.recv().unwrap(); // nothing to do but park on the gate
+        });
+        ready_rx.recv().unwrap();
+
+        let obs = observer::start();
+        // Bounded poll through the observer until the worker reaches Parked —
+        // proving the live state classification rides the call channel intact.
+        let mut parked = false;
+        for _ in 0..100_000 {
+            let ObserverReply::ActorInfo(info) =
+                obs.call(ObserverRequest::ActorInfo(worker.pid())).unwrap()
+            else {
+                panic!("ActorInfo verb must reply ActorInfo");
+            };
+            if matches!(info, Some(i) if i.state == ActorState::Parked) {
+                parked = true;
+                break;
+            }
+            smarm::yield_now();
+        }
+        assert!(parked, "observer should eventually report the worker as Parked");
+
+        gate_tx.send(()).unwrap();
+        worker.join().unwrap();
+    });
+}