stash

apps/pipeline-parallel-inference/src/bin/pp_orchestrator.rs

@@ -54,7 +54,7 @@ use dashboard::collector::StatsCollector;
 use dashboard::datastream_source::{fleet_cache_plugin, FleetView};
 use dashboard::{start_dashboard, DashboardConfig};
 
-use pipeline_parallel_inference::dist_plugin::{DistDashPlugin, MsgCounts, SharedSnapshot};
+use pipeline_parallel_inference::dist_plugin::{DistDashPlugin, SharedSnapshot};
 use pipeline_parallel_inference::netmap_plugin::{spawn_conn_poller, ConnTracker, NetmapPlugin};
 use pipeline_parallel_inference::iroh_transport::{
     ActorMessagePump, IrohActorTransport, ACTOR_ALPN,
@@ -500,12 +500,6 @@ fn run_seed(args: &Args) -> i32 {
     // its snapshot cell exists whenever the in-process dashboard is on.
     let want_dist = std::env::var_os("PP_DASHBOARD").is_some();
 
-    // Orchestrator dashboard message tallies. The per-message wire tally used
-    // to decorate the driver's diagnostics emitter; that emitter was removed
-    // from the engine, so the counts stay empty (the distribution page still
-    // renders membership/routing, just without live message counters).
-    let msg_counts = Arc::new(MsgCounts::default());
-
     let my_id = cluster.node_id();
     let my_hex: String = my_id.0.iter().map(|b| format!("{:02x}", b)).collect();
     let direct: Vec<SocketAddr> = cluster.driver.direct_addresses().to_vec();
@@ -545,19 +539,15 @@ fn run_seed(args: &Args) -> i32 {
         // plugins ("Distribution"/"Netmap" nav tabs) against the shared cell.
         if let (Some((handle, collector, port)), Some(cached)) = (&dashboard, &dist_cached) {
             handle.set_runtime(Arc::clone(&rt), Arc::clone(collector));
-            handle.register_plugin(Arc::new(DistDashPlugin::new(
+            handle.register_plugin(Arc::new(DistDashPlugin::new(Arc::clone(cached))));
-                Arc::clone(cached),
-                Arc::clone(&msg_counts),
-            )));
             // Net map plugin: a live connection/bandwidth graph. Shares the
-            // cached snapshot and message tallies; a background poller keeps its
+            // cached snapshot; a background poller keeps its transport map fresh
-            // transport map fresh by querying the iroh endpoint directly. The
+            // by querying the iroh endpoint directly. The poller's stop flag
-            // poller's stop flag rides the process lifetime (the driver's tokio
+            // rides the process lifetime (the driver's tokio runtime is torn
-            // runtime is torn down at end of run, aborting the task).
+            // down at end of run, aborting the task).
             let conn_tracker = Arc::new(ConnTracker::default());
             handle.register_plugin(Arc::new(NetmapPlugin::new(
                 Arc::clone(cached),
-                Arc::clone(&msg_counts),
                 Arc::clone(&conn_tracker),
             )));
             let poll_stop = Arc::new(AtomicBool::new(false));
@@ -1499,12 +1489,6 @@ fn run_vastai(args: &Args) -> i32 {
     // snapshot cell exists whenever the in-process dashboard is on.
     let want_dist = std::env::var_os("PP_DASHBOARD").is_some();
 
-    // Orchestrator dashboard message tallies. The per-message wire tally used
-    // to decorate the driver's diagnostics emitter; that emitter was removed
-    // from the engine, so the counts stay empty (the distribution page still
-    // renders membership/routing, just without live message counters).
-    let msg_counts = Arc::new(MsgCounts::default());
-
     // Per-cluster drive counter, emitted on pp_drive_start / pp_drive_end so
     // the bundle reader can slice the interleaved event stream by attempt.
     // One-shot and --hold each drive exactly once per process, so this is 1.
@@ -1581,19 +1565,15 @@ fn run_vastai(args: &Args) -> i32 {
     // plugins ("Distribution"/"Netmap" nav tabs) against the shared cell.
     if let (Some((handle, collector, port)), Some(cached)) = (&dashboard, &dist_cached) {
         handle.set_runtime(Arc::clone(&rt), Arc::clone(collector));
-        handle.register_plugin(Arc::new(DistDashPlugin::new(
+        handle.register_plugin(Arc::new(DistDashPlugin::new(Arc::clone(cached))));
-            Arc::clone(cached),
+        // Net map plugin: a live connection/bandwidth graph. Shares the cached
-            Arc::clone(&msg_counts),
+        // snapshot; a background poller keeps its transport map fresh by querying
-        )));
+        // the iroh endpoint directly. The poller's stop flag rides the process
-        // Net map plugin: a live connection/bandwidth graph. Shares the
+        // lifetime (the driver's tokio runtime is torn down at end of run,
-        // cached snapshot and message tallies; a background poller keeps its
+        // aborting the task).
-        // transport map fresh by querying the iroh endpoint directly. The
-        // poller's stop flag rides the process lifetime (the driver's tokio
-        // runtime is torn down at end of run, aborting the task).
         let conn_tracker = Arc::new(ConnTracker::default());
         handle.register_plugin(Arc::new(NetmapPlugin::new(
             Arc::clone(cached),
-            Arc::clone(&msg_counts),
             Arc::clone(&conn_tracker),
         )));
         let poll_stop = Arc::new(AtomicBool::new(false));

apps/pipeline-parallel-inference/src/bin/pp_worker.rs

@@ -992,7 +992,26 @@ fn main_pump(
             let members = fleet::members_to_pairs(&cluster.members_raw());
             let runtime = fleet::runtime_stats(&cluster.rt);
             let relay_connected = cluster.driver.home_relay_url().is_some();
-            fleet.tick(&members, runtime, relay_connected, 0);
+
+            // Real membership transitions (with cause) from the SWIM observer —
+            // submitted before tick() so they drain this cycle (replaces the
+            // emitter's reason-less member-list diff).
+            for t in cluster.drain_swim_transitions() {
+                fleet.submit_membership(&fleet::membership_transition(&t));
+            }
+            // Consolidated distribution state: registry + location cache + recent
+            // probe targets + directory route count.
+            let dist = fleet::build_dist_state(
+                &cluster.registry_snapshot(),
+                &cluster.location_cache_entries(),
+                &cluster.swim_recent_targets(),
+                cluster.driver.directory_route_count() as u32,
+            );
+            fleet.submit_dist_state(&dist);
+            // Worker-runtime counters (the deep slice behind runtime.stats).
+            fleet.submit_worker_counters(&fleet::worker_counters(&cluster.rt));
+
+            fleet.tick(&members, runtime, relay_connected, 0, cluster.swim_rtt_p50());
         }
 
         if let Some(status) = status_inbox.try_recv() {

apps/pipeline-parallel-inference/src/cluster.rs

@@ -39,10 +39,14 @@ use distribution::directory_actor::{DirectoryActor, DirectoryIn};
 use distribution::iroh_driver::{IrohDriver, IrohDriverConfig};
 use distribution::node::DistributedNodeConfig;
 use distribution::node_metadata_actor::{MetadataActor, MetadataIn, RelayInfo};
-use distribution::registry_actor::{NameResolved, RegistryActor, RegistryIn};
+use distribution::registry::RegistrySnapshot;
-use distribution::snapshot::{DistributionNodeSnapshot, MemberInfo};
+use distribution::registry_actor::{NameResolved, RegistryActor, RegistryIn, RegistryView};
+use distribution::snapshot::{
+    CacheEntryInfo, DistributionNodeSnapshot, MemberInfo, RegistryEntryInfo,
+};
 use distribution::swim::actor::{MembershipChanged, SwimActor, SwimIn};
 use distribution::swim::member_list::MemberList;
+use distribution::swim::telemetry::{ObservedTransition, SwimTelemetry};
 use distribution::transport_bridge::{
     IrohPeerDirectory, IrohRouteBinder, Outbox, RelayMirror, RouteView, RouteViewTransport,
 };
@@ -89,6 +93,11 @@ pub struct ClusterNode {
     membership_mirror: Arc<Mutex<MemberList>>,
     relay_mirror: RelayMirror,
     _route_view: RouteView,
+    /// Read-mirror of the cluster registry, for the `dist.state` telemetry.
+    registry_view: RegistryView,
+    /// Production SWIM observer: probe RTT, recent probe targets, and membership
+    /// transitions (with cause), for `transport.internals` / `membership` / `dist.state`.
+    swim_telemetry: Arc<SwimTelemetry>,
     /// Reused per `resolve_name` call to avoid leaking inbox addresses in the
     /// runtime's inbox registry.
     resolve_inbox: Inbox<NameResolved>,
@@ -168,21 +177,24 @@ impl ClusterNode {
             Arc::clone(&outbox),
         ));
 
+        // Telemetry mirrors observed by the fleet emitter (same wiring as the
+        // standalone node): the SWIM observer (probe RTT, recent targets,
+        // transitions-with-cause) and the registry read-mirror.
+        let swim_telemetry = SwimTelemetry::new();
+        let registry_view: RegistryView = Arc::new(RwLock::new(RegistrySnapshot::default()));
+
         // The four protocol actors.
         let swim_addr = rt
-            .spawn(SwimActor::new(
+            .spawn(
-                node_id,
+                SwimActor::new(node_id, swim_config, Instant::now(), peer_directory.clone())
-                swim_config,
+                    .with_observer(Box::new(Arc::clone(&swim_telemetry))),
-                Instant::now(),
+            )
-                peer_directory.clone(),
-            ))
             .expect("spawn SwimActor");
         let registry_addr = rt
-            .spawn(RegistryActor::new(
+            .spawn(
-                node_id,
+                RegistryActor::new(node_id, registry_config, peer_directory.clone())
-                registry_config,
+                    .with_view(Arc::clone(&registry_view)),
-                peer_directory.clone(),
+            )
-            ))
             .expect("spawn RegistryActor");
         let metadata_addr = rt
             .spawn(MetadataActor::new(
@@ -266,11 +278,59 @@ impl ClusterNode {
             membership_mirror,
             relay_mirror,
             _route_view: route_view,
+            registry_view,
+            swim_telemetry,
             resolve_inbox,
             relay_inbox,
         }
     }
 
+    // ── Telemetry accessors (for the fleet emitter) ─────────────────────────
+
+    /// A consistent snapshot of the cluster registry (size / tombstones / entries).
+    pub fn registry_snapshot(&self) -> RegistrySnapshot {
+        self.registry_view.read().expect("registry view poisoned").clone()
+    }
+
+    /// Median SWIM probe round-trip time (ms); `0` until a probe completes.
+    pub fn swim_rtt_p50(&self) -> u32 {
+        self.swim_telemetry.rtt_ms_p50()
+    }
+
+    /// Recent SWIM probe targets (most recent last).
+    pub fn swim_recent_targets(&self) -> Vec<NodeId> {
+        self.swim_telemetry.recent_targets()
+    }
+
+    /// Drain the membership transitions captured since the last call (each carries
+    /// a real cause string).
+    pub fn drain_swim_transitions(&self) -> Vec<ObservedTransition> {
+        self.swim_telemetry.drain_transitions()
+    }
+
+    /// This node's location cache: the remote `(actor, host)` pairs it knows, from
+    /// the directory route-view **and** the registry's live bindings (the demo
+    /// resolves neighbours by name, so the registry is its location directory).
+    /// Self-hosted and tombstoned entries are excluded; deduplicated by address.
+    pub fn location_cache_entries(&self) -> Vec<(ActorAddress, NodeId)> {
+        let self_id = self.driver.node_id();
+        let mut seen = std::collections::HashSet::new();
+        let mut out: Vec<(ActorAddress, NodeId)> = Vec::new();
+        for (addr, host) in self.driver.location_cache_entries() {
+            if seen.insert(addr) {
+                out.push((addr, host));
+            }
+        }
+        let registry = self.registry_view.read().expect("registry view poisoned");
+        for e in &registry.entries {
+            if !e.tombstone && e.node_id != self_id && seen.insert(e.actor_addr) {
+                out.push((e.actor_addr, e.node_id));
+            }
+        }
+        out.sort_by(|a, b| a.0.0.cmp(&b.0.0));
+        out
+    }
+
     // ── Driver passthroughs ─────────────────────────────────────────────────
 
     pub fn node_id(&self) -> NodeId {
@@ -449,9 +509,12 @@ impl ClusterNode {
     pub fn snapshot(&self) -> DistributionNodeSnapshot {
         let mut snap = self.driver.snapshot();
 
-        // Members + counts from the SWIM mirror.
+        // Members + counts from the SWIM mirror, each labelled with the cause of
+        // its most recent liveness transition from the production SWIM observer
+        // (a non-draining read — the fleet emitter still owns the drain).
         let mirror = self.membership_mirror.lock().unwrap();
         let relays = self.relay_mirror.read().ok();
+        let reasons = self.swim_telemetry.last_reasons();
         let mut members: Vec<MemberInfo> = Vec::with_capacity(mirror.len());
         let (mut alive, mut suspect, mut dead) = (0usize, 0usize, 0usize);
         for entry in mirror.all_members() {
@@ -482,6 +545,7 @@ impl ClusterNode {
                 label: None,
                 relay_url,
                 node_name: None,
+                reason: reasons.get(&entry.node_id).map(|r| r.to_string()),
             });
         }
         snap.members = members;
@@ -499,6 +563,39 @@ impl ClusterNode {
             }
         }
 
+        // Registry (name directory), location cache, and recent probe targets,
+        // from this node's telemetry mirrors — the `dist.state` fields the
+        // Distribution page renders. (`driver.snapshot()` fills only the
+        // directory route count.)
+        let hex = |bytes: &[u8]| -> String { bytes.iter().map(|b| format!("{b:02x}")).collect() };
+        let registry = self.registry_snapshot();
+        snap.registry_size = registry.size;
+        snap.registry_tombstones = registry.tombstones;
+        snap.registry_entries = registry
+            .entries
+            .iter()
+            .map(|e| RegistryEntryInfo {
+                name: e.name.clone(),
+                actor_addr: hex(&e.actor_addr.0),
+                node_id: hex(&e.node_id.0),
+                tombstone: e.tombstone,
+            })
+            .collect();
+        let cache = self.location_cache_entries();
+        snap.cache_size = cache.len();
+        snap.cache_entries = cache
+            .iter()
+            .map(|(addr, host)| CacheEntryInfo {
+                actor_addr: hex(&addr.0),
+                node_id: hex(&host.0),
+            })
+            .collect();
+        snap.recent_probe_targets = self
+            .swim_recent_targets()
+            .iter()
+            .map(|t| hex(&t.0))
+            .collect();
+
         snap
     }
 }

apps/pipeline-parallel-inference/src/dist_page.html

@@ -64,16 +64,6 @@
       </div>
     </div>
 
-    <div class="card">
-      <h2>Messages (this node, wire)</h2>
-      <div class="stats">
-        <div class="stat"><span id="m-sent" class="v sent">0</span><span class="l">Sent</span></div>
-        <div class="stat"><span id="m-recv" class="v recv">0</span><span class="l">Received</span></div>
-        <div class="stat"><span id="m-sentb" class="v sent">0</span><span class="l">Sent bytes</span></div>
-        <div class="stat"><span id="m-recvb" class="v recv">0</span><span class="l">Recv bytes</span></div>
-      </div>
-    </div>
-
     <div class="card full">
       <h2>SWIM membership graph</h2>
       <canvas id="graph"></canvas>
@@ -83,26 +73,11 @@
     <div class="card">
       <h2>Members</h2>
       <table>
-        <thead><tr><th>State</th><th>Name / Node</th><th>Addr</th><th class="num">Inc</th></tr></thead>
+        <thead><tr><th>State</th><th>Name / Node</th><th>Addr</th><th class="num">Inc</th><th>Reason</th></tr></thead>
-        <tbody id="members"><tr><td colspan="4" class="muted">waiting for snapshot…</td></tr></tbody>
+        <tbody id="members"><tr><td colspan="5" class="muted">waiting for snapshot…</td></tr></tbody>
       </table>
     </div>
 
-    <div class="card">
-      <h2>Messages by kind</h2>
-      <table>
-        <thead><tr><th>Kind</th><th class="num">Sent</th><th class="num">Recv</th></tr></thead>
-        <tbody id="bykind"><tr><td colspan="3" class="muted">no traffic yet</td></tr></tbody>
-      </table>
-    </div>
-
-    <div class="card full">
-      <h2>Messages by peer</h2>
-      <table>
-        <thead><tr><th>Peer</th><th>Name</th><th class="num">Sent</th><th class="num">Recv</th></tr></thead>
-        <tbody id="bypeer"><tr><td colspan="4" class="muted">no traffic yet</td></tr></tbody>
-      </table>
-    </div>
   </div>
 
 <script>
@@ -184,32 +159,10 @@
       return '<tr><td><span class="dot" style="background:'+stateColor(m.state)+'"></span>'+m.state+'</td>'
         + '<td>'+(nm? nm+' <span class="mono">'+short(m.node_id)+'</span>' : '<span class="mono">'+short(m.node_id)+'</span>')+'</td>'
         + '<td class="mono">'+(m.addr||'—')+'</td>'
-        + '<td class="num">'+m.incarnation+'</td></tr>';
+        + '<td class="num">'+m.incarnation+'</td>'
+        + '<td class="muted">'+(m.reason||'—')+'</td></tr>';
     }).join('');
-    document.getElementById('members').innerHTML = mb || '<tr><td colspan="4" class="muted">no members</td></tr>';
+    document.getElementById('members').innerHTML = mb || '<tr><td colspan="5" class="muted">no members</td></tr>';
-
-    // messages
-    var mc = d.msg_counts || {};
-    document.getElementById('m-sent').textContent = mc.sent_total || 0;
-    document.getElementById('m-recv').textContent = mc.recv_total || 0;
-    document.getElementById('m-sentb').textContent = (mc.sent_bytes || 0).toLocaleString();
-    document.getElementById('m-recvb').textContent = (mc.recv_bytes || 0).toLocaleString();
-
-    // by kind (union of sent+recv kinds)
-    var ks = mc.by_kind_sent || {}, kr = mc.by_kind_recv || {};
-    var kinds = Object.keys(ks).concat(Object.keys(kr)).filter(function(v,i,a){return a.indexOf(v)===i;}).sort();
-    var kt = kinds.map(function(k){
-      return '<tr><td class="mono">'+k+'</td><td class="num sent">'+(ks[k]||0)+'</td><td class="num recv">'+(kr[k]||0)+'</td></tr>';
-    }).join('');
-    document.getElementById('bykind').innerHTML = kt || '<tr><td colspan="3" class="muted">no traffic yet</td></tr>';
-
-    // by peer
-    var bp = mc.by_peer || {};
-    var pt = Object.keys(bp).sort().map(function(p){
-      var nm = (names[p] && names[p].join(', ')) || '';
-      return '<tr><td class="mono">'+short(p)+'</td><td>'+nm+'</td><td class="num sent">'+(bp[p].sent||0)+'</td><td class="num recv">'+(bp[p].recv||0)+'</td></tr>';
-    }).join('');
-    document.getElementById('bypeer').innerHTML = pt || '<tr><td colspan="4" class="muted">no traffic yet</td></tr>';
 
     drawGraph(d);
   }

apps/pipeline-parallel-inference/src/dist_plugin.rs

@@ -1,23 +1,12 @@
 //! Orchestrator distribution dashboard plugin.
 //!
-//! Two cooperating pieces give the orchestrator dashboard its "Distribution"
+//! [`DistDashPlugin`] — a read-only [`DashboardPlugin`] named `"distribution"`,
-//! tab (which is otherwise dead — the dashboard ships the nav link but no
+//! which otherwise has no backing (the dashboard ships the nav link but nothing
-//! plugin is registered to back `/plugin/distribution`):
+//! serves `/plugin/distribution`). It serves a cached
-//!
+//! [`DistributionNodeSnapshot`] (SWIM members, routing table, location cache,
-//!   * [`MsgCounts`] — a per-kind / per-peer message + byte tally injected into
+//! name registry) plus a lean HTML page that draws a SWIM membership graph.
-//!     the snapshot JSON under `msg_counts`. The engine no longer carries a
-//!     message-counting emitter, so the tallies currently stay at zero; the
-//!     panel renders the shape and is ready to repopulate if a datastream
-//!     message-count channel is wired.
-//!
-//!   * [`DistDashPlugin`] — a read-only [`DashboardPlugin`] named
-//!     `"distribution"`. It serves a cached [`DistributionNodeSnapshot`] (SWIM
-//!     members, routing table, location cache, name registry) with the message
-//!     tallies injected under `msg_counts`, plus a lean HTML page that draws a
-//!     SWIM membership graph and a messages panel.
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use dashboard::plugin::{DashboardPlugin, PluginResponse};
@@ -26,89 +15,27 @@ use distribution::snapshot::DistributionNodeSnapshot;
 /// Shared snapshot cell the orchestrator refreshes each tick.
 pub type SharedSnapshot = Arc<Mutex<Option<DistributionNodeSnapshot>>>;
 
-/// Cumulative wire-message tallies for this node.
+/// Render a distribution snapshot as the JSON the distribution UI consumes,
-#[derive(Default)]
+/// served by [`DistDashPlugin`].
-pub struct MsgCounts {
+pub fn render_dist_json(snap: &DistributionNodeSnapshot) -> Option<String> {
-    sent_total: AtomicU64,
+    serde_json::to_string(snap).ok()
-    recv_total: AtomicU64,
-    sent_bytes: AtomicU64,
-    recv_bytes: AtomicU64,
-    by_kind_sent: Mutex<HashMap<String, u64>>,
-    by_kind_recv: Mutex<HashMap<String, u64>>,
-    /// peer hex → cumulative per-peer message/byte tallies
-    by_peer: Mutex<HashMap<String, PeerTally>>,
-}
-
-/// Cumulative per-peer tally: message counts and payload bytes, each split by
-/// direction. Byte counters are monotonic, so a UI can derive a rate from the
-/// delta between two snapshots.
-#[derive(Default, Clone, Copy)]
-pub struct PeerTally {
-    pub sent: u64,
-    pub recv: u64,
-    pub sent_bytes: u64,
-    pub recv_bytes: u64,
-}
-
-impl MsgCounts {
-    /// Render the tallies as the JSON object injected under `msg_counts`.
-    pub fn to_json(&self) -> serde_json::Value {
-        let by_peer: serde_json::Map<String, serde_json::Value> = self
-            .by_peer
-            .lock()
-            .unwrap()
-            .iter()
-            .map(|(k, t)| {
-                (
-                    k.clone(),
-                    serde_json::json!({
-                        "sent": t.sent,
-                        "recv": t.recv,
-                        "sent_bytes": t.sent_bytes,
-                        "recv_bytes": t.recv_bytes,
-                    }),
-                )
-            })
-            .collect();
-        serde_json::json!({
-            "sent_total": self.sent_total.load(Ordering::Relaxed),
-            "recv_total": self.recv_total.load(Ordering::Relaxed),
-            "sent_bytes": self.sent_bytes.load(Ordering::Relaxed),
-            "recv_bytes": self.recv_bytes.load(Ordering::Relaxed),
-            "by_kind_sent": self.by_kind_sent.lock().unwrap().clone(),
-            "by_kind_recv": self.by_kind_recv.lock().unwrap().clone(),
-            "by_peer": by_peer,
-        })
-    }
-}
-
-/// Render a distribution snapshot with the message tallies injected under
-/// `msg_counts` — the exact JSON shape the distribution UI consumes, served by
-/// [`DistDashPlugin`].
-pub fn render_dist_json(snap: &DistributionNodeSnapshot, counts: &MsgCounts) -> Option<String> {
-    let mut v = serde_json::to_value(snap).ok()?;
-    if let serde_json::Value::Object(ref mut m) = v {
-        m.insert("msg_counts".to_string(), counts.to_json());
-    }
-    serde_json::to_string(&v).ok()
 }
 
 /// Read-only dashboard plugin backing `/plugin/distribution`.
 pub struct DistDashPlugin {
     cached: SharedSnapshot,
-    counts: Arc<MsgCounts>,
 }
 
 impl DistDashPlugin {
-    pub fn new(cached: SharedSnapshot, counts: Arc<MsgCounts>) -> Self {
+    pub fn new(cached: SharedSnapshot) -> Self {
-        Self { cached, counts }
+        Self { cached }
     }
 
-    /// Serialize the cached snapshot with the message tallies injected.
+    /// Serialize the cached snapshot.
     fn rendered_json(&self) -> Option<String> {
         let guard = self.cached.lock().unwrap();
         let snap = guard.as_ref()?;
-        render_dist_json(snap, &self.counts)
+        render_dist_json(snap)
     }
 }
 

apps/pipeline-parallel-inference/src/fleet.rs

@@ -14,12 +14,17 @@
 use std::sync::{Arc, OnceLock};
 use std::time::{Duration, Instant};
 
-use datastream::catalog::{IdentityRecord, RuntimeStats as DsRuntimeStats};
+use datastream::catalog::{
+    CacheEntryRec, DistributionState, IdentityRecord, MembershipTransition, RegistryEntryRec,
+    RuntimeStats as DsRuntimeStats, WorkerCounters,
+};
 use datastream::emit::{
     ClusterFrameSink, DatastreamEmitter, EmitterConfig, TickInput,
 };
+use distribution::registry::RegistrySnapshot;
 use distribution::swim::member_list::MemberEntry;
-use distribution::types::MemberState;
+use distribution::swim::telemetry::ObservedTransition;
+use distribution::types::{MemberState, NodeId};
 use swactor::actor::ActorAddress;
 use swactor::runtime::Runtime;
 
@@ -51,7 +56,7 @@ impl FleetEmitter {
         listen_addr: &str,
     ) -> Self {
         let sink = ClusterFrameSink::new(rt, sink_slot);
-        let emitter = DatastreamEmitter::new(
+        let mut emitter = DatastreamEmitter::new(
             EmitterConfig {
                 node_hex: node_hex.to_string(),
                 life,
@@ -68,18 +73,24 @@ impl FleetEmitter {
             listen_addr: listen_addr.to_string(),
             ..Default::default()
         });
+        // Membership is driven from the SWIM observer (real transitions, with a
+        // cause), so the emitter's reason-less member-list diff is turned off; the
+        // caller drains transitions via [`submit_membership`](Self::submit_membership).
+        emitter.use_external_membership();
         Self { emitter }
     }
 
-    /// Ship one periodic sample: host resource + runtime + transport, plus any
+    /// Ship one periodic sample: host resource + runtime + transport (with the
-    /// membership transitions since the last tick. Call on the
+    /// stage's real SWIM `rtt_ms_p50`) + datastream health. Membership, dist.state,
-    /// [`FLEET_TICK_INTERVAL`] cadence from the node's main loop.
+    /// and worker counters are submitted separately before this drains the mux.
+    /// Call on the [`FLEET_TICK_INTERVAL`] cadence from the node's main loop.
     pub fn tick(
         &mut self,
         members: &[(String, String)],
         runtime: DsRuntimeStats,
         relay_connected: bool,
         relay_peers: u32,
+        rtt_ms_p50: u32,
     ) {
         self.emitter.tick(
             TickInput {
@@ -87,10 +98,119 @@ impl FleetEmitter {
                 runtime,
                 relay_connected,
                 relay_peers,
+                rtt_ms_p50,
             },
             true,
         );
     }
+
+    /// Emit one membership transition from the SWIM observer (carries a real
+    /// `reason`). Submit before [`tick`](Self::tick) so it drains this cycle.
+    pub fn submit_membership(&self, transition: &MembershipTransition) {
+        self.emitter.submit_membership(transition);
+    }
+
+    /// Emit the consolidated distribution-subsystem state (registry / location
+    /// cache / probe targets). Submit before [`tick`](Self::tick).
+    pub fn submit_dist_state(&self, state: &DistributionState) {
+        self.emitter.submit_dist_state(state);
+    }
+
+    /// Emit the aggregated worker-runtime counters. Submit before [`tick`](Self::tick).
+    pub fn submit_worker_counters(&self, counters: &WorkerCounters) {
+        self.emitter.submit_worker_counters(counters);
+    }
+}
+
+/// Hex-encode raw id bytes the way every other id on the stream is encoded.
+fn hex(bytes: &[u8]) -> String {
+    bytes.iter().map(|b| format!("{b:02x}")).collect()
+}
+
+fn member_state_str(state: MemberState) -> &'static str {
+    match state {
+        MemberState::Alive => "alive",
+        MemberState::Suspect => "suspect",
+        MemberState::Dead => "dead",
+    }
+}
+
+/// Convert a SWIM-observer transition into the wire record (with its real cause).
+pub fn membership_transition(t: &ObservedTransition) -> MembershipTransition {
+    MembershipTransition {
+        peer: hex(&t.peer.0),
+        from: t.from.map(member_state_str).unwrap_or("unknown").to_string(),
+        to: member_state_str(t.to).to_string(),
+        reason: t.reason.to_string(),
+    }
+}
+
+/// Build the consolidated `dist.state` record from a stage's live subsystem
+/// mirrors — the same shape the standalone node assembles. The demo runs open
+/// peer-auth, so `peer_auth_mode` is `"open"` and `authorized_peer_count` is 0.
+pub fn build_dist_state(
+    registry: &RegistrySnapshot,
+    cache: &[(ActorAddress, NodeId)],
+    recent_targets: &[NodeId],
+    directory_route_count: u32,
+) -> DistributionState {
+    DistributionState {
+        cache_size: cache.len() as u32,
+        cache_entries: cache
+            .iter()
+            .map(|(addr, host)| CacheEntryRec {
+                actor_addr: hex(&addr.0),
+                node_id: hex(&host.0),
+            })
+            .collect(),
+        directory_route_count,
+        registry_size: registry.size as u32,
+        registry_tombstones: registry.tombstones as u32,
+        registry_entries: registry
+            .entries
+            .iter()
+            .map(|e| RegistryEntryRec {
+                name: e.name.clone(),
+                actor_addr: hex(&e.actor_addr.0),
+                node_id: hex(&e.node_id.0),
+                tombstone: e.tombstone,
+            })
+            .collect(),
+        recent_probe_targets: recent_targets.iter().map(|t| hex(&t.0)).collect(),
+        peer_auth_mode: "open".to_string(),
+        authorized_peer_count: 0,
+    }
+}
+
+/// Aggregate the runtime's per-worker counters + tick timing into the
+/// `runtime.workers` record (the deep slice behind the thin `runtime.stats`).
+pub fn worker_counters(rt: &Runtime) -> WorkerCounters {
+    let rs = rt.stats();
+    let mut wc = WorkerCounters {
+        num_workers: rs.workers.len() as u32,
+        ..Default::default()
+    };
+    for w in &rs.workers {
+        wc.scheduled_tasks += w.num_actors as u32;
+        wc.local_sends += w.local_sends;
+        wc.cross_sends += w.cross_sends;
+        wc.inbox_sends += w.inbox_sends;
+        wc.type_mismatches += w.type_mismatches;
+        wc.panics += w.panics;
+        wc.messages_dropped += w.messages_dropped;
+        wc.restarts += w.restarts;
+        wc.stops += w.stops;
+        wc.messages_processed += w.messages_processed;
+    }
+    let mut tick_us: Vec<u64> = rs
+        .tick_timings
+        .iter()
+        .flatten()
+        .map(|t| t.phase_us.iter().sum())
+        .collect();
+    tick_us.sort_unstable();
+    wc.tick_p50_us = tick_us.get(tick_us.len() / 2).copied().unwrap_or(0);
+    wc
 }
 
 /// A cheap "has the fleet cadence elapsed" gate, so a 20 ms main loop only

apps/pipeline-parallel-inference/src/netmap_page.html

@@ -205,7 +205,7 @@
   }
 
   function computeRates(d) {
-    var bp = (d.msg_counts && d.msg_counts.by_peer) || {};
+    var bp = {};
     var now = d.conn_ts_ms || Date.now();
     var rates = {};
     Object.keys(bp).forEach(function(p) {
@@ -229,7 +229,7 @@
     document.getElementById('node').textContent = 'node ' + short(d.node_id);
 
     var conn = d.conn || {};
-    var bp = (d.msg_counts && d.msg_counts.by_peer) || {};
+    var bp = {};
     var members = d.members || [];
     var rows = members.map(function(m) {
       var hex = m.node_id, c = conn[hex] || {}, rt = lastRates[hex] || {}, t = bp[hex] || {};
@@ -257,7 +257,7 @@
     var hit = nodeHits.find(function(h) { return Math.hypot(mx - h.x, my - h.y) <= h.r + 4; });
     if (!hit) { tip.style.display = 'none'; return; }
     var d = last, c = (d.conn || {})[hit.hex] || {}, rt = lastRates[hit.hex] || {};
-    var t = ((d.msg_counts && d.msg_counts.by_peer) || {})[hit.hex] || {};
+    var t = ({})[hit.hex] || {};
     tip.innerHTML = '<b>' + labelFor(d, hit.hex, hit.mem) + '</b> <span class="mono">' + short(hit.hex) + '</span><br>'
       + 'transport: <b>' + (c.type || 'None') + '</b>'
       + (c.relay_url ? '<br>relay: <span class="mono">' + c.relay_url + '</span>' : '')

apps/pipeline-parallel-inference/src/netmap_plugin.rs

@@ -8,12 +8,10 @@
 //! by live bandwidth** (bytes/sec, derived client-side from the delta of the
 //! cumulative per-peer byte counters between successive snapshots).
 //!
-//! Two pieces feed it, both **shared** with the distribution plugin:
+//! Two pieces feed it:
 //!
-//!   * [`SharedSnapshot`] + [`MsgCounts`](crate::dist_plugin::MsgCounts) — the
+//!   * [`SharedSnapshot`] — the cached cluster snapshot (members, names),
-//!     cached cluster snapshot (members, names) and the per-peer message/byte
+//!     **shared** with the distribution plugin.
-//!     tallies. The engine no longer carries a message-counting emitter, so the
-//!     tallies stay at zero; the graph still renders the live transport mesh.
 //!   * [`ConnTracker`] — a per-peer transport map kept fresh by
 //!     [`spawn_conn_poller`], which queries `endpoint.remote_info` directly —
 //!     a self-contained poll, independent of any telemetry pipeline.
@@ -27,7 +25,7 @@ use dashboard::plugin::{DashboardPlugin, PluginResponse};
 use distribution::iroh_driver::{conn_type_of, ConnType, IrohDriver};
 use iroh::PublicKey;
 
-use crate::dist_plugin::{MsgCounts, SharedSnapshot};
+use crate::dist_plugin::SharedSnapshot;
 
 /// Milliseconds since the Unix epoch.
 fn wall_ms_now() -> u64 {
@@ -91,27 +89,21 @@ const NETMAP_HTML: &str = include_str!("netmap_page.html");
 /// Read-only dashboard plugin backing `/plugin/netmap`.
 pub struct NetmapPlugin {
     cached: SharedSnapshot,
-    counts: Arc<MsgCounts>,
     conn: Arc<ConnTracker>,
 }
 
 impl NetmapPlugin {
-    pub fn new(cached: SharedSnapshot, counts: Arc<MsgCounts>, conn: Arc<ConnTracker>) -> Self {
+    pub fn new(cached: SharedSnapshot, conn: Arc<ConnTracker>) -> Self {
-        Self {
+        Self { cached, conn }
-            cached,
-            counts,
-            conn,
-        }
     }
 
-    /// Serialize the cached snapshot with message tallies, the transport map, and
+    /// Serialize the cached snapshot with the transport map and a server
-    /// a server timestamp the frontend uses as the time base for bytes/sec.
+    /// timestamp the frontend uses as the time base for bytes/sec.
     fn rendered_json(&self) -> Option<String> {
         let guard = self.cached.lock().unwrap();
         let snap = guard.as_ref()?;
         let mut v = serde_json::to_value(snap).ok()?;
         if let serde_json::Value::Object(ref mut m) = v {
-            m.insert("msg_counts".to_string(), self.counts.to_json());
             m.insert("conn".to_string(), self.conn.to_json());
             m.insert(
                 "conn_ts_ms".to_string(),

crates/dashboard/src/datastream_source.rs

@@ -25,9 +25,9 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use datastream::catalog::{
-    self, ActorRuntimeDetail, DatastoreState, DistributionState, IdentityRecord, LifecycleCost,
+    self, ActorRuntimeDetail, DatastoreState, DatastreamHealth, DistributionState, IdentityRecord,
     MembershipTransition, Record, ResourceSample, RuntimeStats as DsRuntimeStats,
-    TransportInternals,
+    TransportInternals, WorkerCounters,
 };
 use datastream::frame::{Frame, StreamId};
 
@@ -54,18 +54,24 @@ struct DatastreamModel {
     resource: Option<ResourceSample>,
     runtime: Option<DsRuntimeStats>,
     transport: Option<TransportInternals>,
-    lifecycle: Option<LifecycleCost>,
     /// Consolidated distribution-subsystem state (cache/registry/directory/...).
     dist_state: Option<DistributionState>,
     /// Datastore steady metrics.
     datastore_state: Option<DatastoreState>,
     /// Per-actor runtime detail (the real actor table).
     actor_detail: Option<ActorRuntimeDetail>,
+    /// Aggregated worker-runtime counters (routing/error tallies + tick timing).
+    worker_counters: Option<WorkerCounters>,
+    /// Datastream self-health (mux assigned/dropped + loss rate).
+    datastream_health: Option<DatastreamHealth>,
     /// Recent datastore op events (newest last), capped at [`DATASTORE_EVENT_CAP`].
     datastore_events: VecDeque<serde_json::Value>,
     /// peer node-id → latest liveness state.
     membership: HashMap<String, String>,
-    /// last membership transition, formatted for display.
+    /// peer node-id → cause of its most recent liveness transition (the SWIM
+    /// observer's reason string; the value-add of W3/W4 over the state-diff).
+    membership_reason: HashMap<String, String>,
+    /// last membership transition, formatted for display (with its cause).
     last_transition: Option<String>,
     /// proc label → (line count, last line).
     procs: HashMap<String, (u64, String)>,
@@ -109,11 +115,6 @@ impl DatastreamModel {
                     self.transport = Some(r);
                 }
             }
-            catalog::LIFECYCLE_COST => {
-                if let Ok(r) = LifecycleCost::decode(payload) {
-                    self.lifecycle = Some(r);
-                }
-            }
             catalog::DIST_STATE => {
                 if let Ok(r) = DistributionState::decode(payload) {
                     self.dist_state = Some(r);
@@ -129,6 +130,16 @@ impl DatastreamModel {
                     self.actor_detail = Some(r);
                 }
             }
+            catalog::RUNTIME_WORKERS => {
+                if let Ok(r) = WorkerCounters::decode(payload) {
+                    self.worker_counters = Some(r);
+                }
+            }
+            catalog::DATASTREAM_HEALTH => {
+                if let Ok(r) = DatastreamHealth::decode(payload) {
+                    self.datastream_health = Some(r);
+                }
+            }
             // Datastore op events: structured text lines, tailed into a capped
             // ring so the datastore page can show a recent-operations timeline.
             catalog::DATASTORE_EVENTS => {
@@ -145,9 +156,19 @@ impl DatastreamModel {
             catalog::MEMBERSHIP => {
                 if let Ok(t) = MembershipTransition::decode(payload) {
                     // Display the short id; key the membership map by the full
-                    // id so the views can resolve it to a friendly label.
+                    // id so the views can resolve it to a friendly label. Carry
-                    let line = format!("{}: {} → {}", short_id(&t.peer), t.from, t.to);
+                    // the observer's cause string through to the display — it is
+                    // the whole value-add of the production observer over the old
+                    // state-diff (which only ever knew *that* a peer changed).
+                    let line = if t.reason.is_empty() {
+                        format!("{}: {} → {}", short_id(&t.peer), t.from, t.to)
+                    } else {
+                        format!("{}: {} → {} ({})", short_id(&t.peer), t.from, t.to, t.reason)
+                    };
                     self.membership.insert(t.peer.clone(), t.to.clone());
+                    if !t.reason.is_empty() {
+                        self.membership_reason.insert(t.peer.clone(), t.reason.clone());
+                    }
                     self.last_transition = Some(line.clone());
                     events.push(LogEvent::Info(format!("membership {line}")));
                 }
@@ -289,20 +310,27 @@ impl DatastreamModel {
             scheduled_tasks: 0,
         });
 
-        // One synthetic worker = this node.
+        // One synthetic worker = this node. Routing/error counters come from the
+        // `runtime.workers` channel (the deep slice behind the thin runtime.stats);
+        // they stay 0 only until that channel has been folded.
+        let wc = self.worker_counters.clone().unwrap_or_default();
         let worker = WorkerInfo {
             id: 0,
             num_actors: ds_rt.actors_live as usize,
             mailbox_depth: ds_rt.mailbox_depth as usize,
-            messages_processed: self.total_proc_lines(),
+            messages_processed: if wc.messages_processed > 0 {
-            local_sends: 0,
+                wc.messages_processed
-            cross_sends: 0,
+            } else {
-            inbox_sends: 0,
+                self.total_proc_lines()
-            type_mismatches: 0,
+            },
-            panics: 0,
+            local_sends: wc.local_sends,
-            messages_dropped: 0,
+            cross_sends: wc.cross_sends,
-            restarts: 0,
+            inbox_sends: wc.inbox_sends,
-            stops: 0,
+            type_mismatches: wc.type_mismatches,
+            panics: wc.panics,
+            messages_dropped: wc.messages_dropped,
+            restarts: wc.restarts,
+            stops: wc.stops,
         };
 
         let actor_details = self.actor_rows();
@@ -312,15 +340,12 @@ impl DatastreamModel {
             .map(|a| (a.address, a.worker_id))
             .collect();
 
+        // Uptime since this node's first frame was folded (the legacy
+        // provider.lifecycle uptime had no producer and was removed).
         let uptime_ms = self
-            .lifecycle
+            .first_seen
-            .as_ref()
+            .map(|t| t.elapsed().as_millis() as u64)
-            .map(|l| l.uptime_s * 1000)
+            .unwrap_or(0);
-            .unwrap_or_else(|| {
-                self.first_seen
-                    .map(|t| t.elapsed().as_millis() as u64)
-                    .unwrap_or(0)
-            });
 
         RuntimeStats {
             num_workers: 1,
@@ -375,6 +400,9 @@ impl DatastreamModel {
                     "state": state,
                     "incarnation": 0,
                     "node_name": peer_labels.get(peer).cloned().unwrap_or_else(|| short_id(peer)),
+                    // Cause of this peer's last transition (observer reason),
+                    // null until one has been seen.
+                    "reason": self.membership_reason.get(peer).cloned(),
                 })
             })
             .collect();
@@ -504,6 +532,8 @@ impl DatastreamModel {
         let r = self.resource.as_ref();
         let t = self.transport.as_ref();
         let rt = self.runtime.as_ref();
+        let wc = self.worker_counters.as_ref();
+        let dh = self.datastream_health.as_ref();
         let last_proc = self
             .procs
             .values()
@@ -523,8 +553,28 @@ impl DatastreamModel {
             "mem_used_mb": r.map(|r| r.mem_used_mb).unwrap_or(0),
             "mem_total_mb": r.map(|r| r.mem_total_mb).unwrap_or(0),
             "gpu_pct": r.map(|r| r.gpu_pct.round() as u32).unwrap_or(0),
+            "disk_used_gb": r.map(|r| r.disk_used_gb).unwrap_or(0),
+            "net_rx_kbps": r.map(|r| r.net_rx_kbps).unwrap_or(0),
+            "net_tx_kbps": r.map(|r| r.net_tx_kbps).unwrap_or(0),
             "actors_live": rt.map(|r| r.actors_live).unwrap_or(0),
             "mailbox_depth": rt.map(|r| r.mailbox_depth).unwrap_or(0),
+            // Real polled value (was a hardcoded 0 in the consumer).
+            "scheduled_tasks": rt.map(|r| r.scheduled_tasks).unwrap_or(0),
+            // Worker-runtime counters (runtime.workers channel): routing/error
+            // tallies + tick timing, the deep slice behind runtime.stats.
+            "worker_counters": {
+                "num_workers": wc.map(|w| w.num_workers).unwrap_or(0),
+                "local_sends": wc.map(|w| w.local_sends).unwrap_or(0),
+                "cross_sends": wc.map(|w| w.cross_sends).unwrap_or(0),
+                "inbox_sends": wc.map(|w| w.inbox_sends).unwrap_or(0),
+                "type_mismatches": wc.map(|w| w.type_mismatches).unwrap_or(0),
+                "panics": wc.map(|w| w.panics).unwrap_or(0),
+                "messages_dropped": wc.map(|w| w.messages_dropped).unwrap_or(0),
+                "restarts": wc.map(|w| w.restarts).unwrap_or(0),
+                "stops": wc.map(|w| w.stops).unwrap_or(0),
+                "messages_processed": wc.map(|w| w.messages_processed).unwrap_or(0),
+                "tick_p50_us": wc.map(|w| w.tick_p50_us).unwrap_or(0),
+            },
             "relay_connected": t.map(|t| t.relay_connected).unwrap_or(false),
             "direct_peers": t.map(|t| t.direct_peers).unwrap_or(0),
             "relay_peers": t.map(|t| t.relay_peers).unwrap_or(0),
@@ -533,6 +583,18 @@ impl DatastreamModel {
             "suspect": suspect,
             "dead": dead,
             "converged": converged,
+            // Most recent SWIM liveness transition this node observed, carrying
+            // the observer's cause string (the `reason` was always "" before the
+            // migration installed the production observer). `null` until a
+            // transition has been folded.
+            "last_transition": self.last_transition.clone(),
+            // Datastream self-health (datastream.health channel): the pipe
+            // reporting its own integrity. loss_rate_ppm = dropped/assigned × 1e6.
+            "datastream": {
+                "assigned": dh.map(|d| d.assigned).unwrap_or(0),
+                "dropped": dh.map(|d| d.dropped).unwrap_or(0),
+                "loss_rate_ppm": dh.map(|d| d.loss_rate_ppm).unwrap_or(0),
+            },
             "proc_lines": self.total_proc_lines(),
             "last_proc": last_proc,
         })
@@ -924,8 +986,9 @@ const FLEET_HTML: &str = r#"<!doctype html>
   <table>
     <thead><tr>
       <th>node</th><th>state</th><th>region</th><th>role</th><th>CPU</th><th>mem</th>
+      <th class="num">disk</th><th class="num">net ↓/↑</th>
       <th class="num">actors</th><th class="num">mbox</th><th>transport</th>
-      <th class="num">peers</th><th class="num">proc</th><th>last line</th>
+      <th class="num">peers</th><th class="num">proc</th><th class="num">pipe</th><th>last line</th>
     </tr></thead>
     <tbody id="rows"></tbody>
   </table>
@@ -956,11 +1019,17 @@ const FLEET_HTML: &str = r#"<!doctype html>
         + '<td><span class="tag '+esc(n.role)+'">'+esc(n.role||"?")+'</span></td>'
         + '<td>'+bar(n.cpu_pct)+'</td>'
         + '<td>'+n.mem_used_mb+'/'+n.mem_total_mb+'MB</td>'
+        + '<td class="num">'+(n.disk_used_gb||0)+'G</td>'
+        + '<td class="num">'+(n.net_rx_kbps||0)+'/'+(n.net_tx_kbps||0)+'</td>'
         + '<td class="num">'+n.actors_live+'</td>'
         + '<td class="num">'+n.mailbox_depth+'</td>'
         + '<td>'+(n.relay_connected?'relay ':'')+n.direct_peers+'d/'+n.relay_peers+'r</td>'
         + '<td class="num">'+peers+'</td>'
         + '<td class="num">'+n.proc_lines+'</td>'
+        + '<td class="num" title="frames assigned / dropped">'
+          +((n.datastream&&n.datastream.assigned)||0)
+          +((n.datastream&&n.datastream.dropped)?(' <span class="st dead">-'+n.datastream.dropped+'</span>'):'')
+          +'</td>'
         + '<td class="last">'+esc(n.last_proc)+'</td>'
         + '</tr>';
     }).join("");

crates/datastore/README.md

@@ -1,4 +1,5 @@
-# swactor-datastore
+# swactor-datastore **CURRENTLY OBSELETE**
+
 
 Distributed content-addressed datastore built on [swactor](../../README.md). Objects are split into fixed-size chunks, identified by their blake3 hash, and replicated across a peer-to-peer network via epidemic gossip.
 

crates/datastream/src/catalog.rs

@@ -35,9 +35,6 @@ pub const TRANSPORT_INTERNALS: &str = "transport.internals";
 pub const MEMBERSHIP: &str = "membership";
 /// Runtime stats — the node's own actor-runtime metrics. Periodic.
 pub const RUNTIME_STATS: &str = "runtime.stats";
-/// Provider / lifecycle / cost — coarse lifecycle and cost facts about the
-/// node as a rented resource.
-pub const LIFECYCLE_COST: &str = "provider.lifecycle";
 /// Distribution-subsystem state — location cache, directory, registry, gossip
 /// probes, and peer-auth. Periodic, consolidated; the parts of a node's
 /// distribution view the `membership`/`identity` channels do not already carry.
@@ -48,6 +45,14 @@ pub const DATASTORE_STATE: &str = "datastore.state";
 /// Per-actor runtime detail — one row per live actor, behind the aggregate
 /// [`RUNTIME_STATS`]. Periodic.
 pub const RUNTIME_ACTORS: &str = "runtime.actors";
+/// Worker-runtime counters — the routing/error tallies and tick timing the
+/// runtime keeps per worker, aggregated. The deep slice behind the thin
+/// [`RUNTIME_STATS`] heartbeat. Periodic.
+pub const RUNTIME_WORKERS: &str = "runtime.workers";
+/// Datastream self-health — the pipe reporting on its own integrity: positions
+/// assigned vs. frames dropped on mux overflow, and the resulting loss rate.
+/// Periodic.
+pub const DATASTREAM_HEALTH: &str = "datastream.health";
 /// Datastore operation events — a raw-text channel carrying one line per
 /// recorded op (`<kind> <hash> <size>`), the `proc.*` model applied to ops.
 pub const DATASTORE_EVENTS: &str = "datastore.events";
@@ -102,7 +107,8 @@ pub fn classify(channel: &ChannelId) -> ChannelKind {
     let id = channel.as_str();
     match id {
         IDENTITY | HOST_RESOURCE | TRANSPORT_INTERNALS | MEMBERSHIP | RUNTIME_STATS
-        | LIFECYCLE_COST | DIST_STATE | DATASTORE_STATE | RUNTIME_ACTORS => ChannelKind::Typed,
+        | DIST_STATE | DATASTORE_STATE | RUNTIME_ACTORS | RUNTIME_WORKERS
+        | DATASTREAM_HEALTH => ChannelKind::Typed,
         DATASTORE_EVENTS => ChannelKind::Text,
         _ if id.starts_with("proc.") => ChannelKind::Text,
         _ => ChannelKind::Opaque,
@@ -220,16 +226,6 @@ pub struct RuntimeStats {
     pub scheduled_tasks: u32,
 }
 
-/// Provider / lifecycle / cost record (spec §6.1).
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct LifecycleCost {
-    pub phase: String,
-    #[serde(default)]
-    pub cost_usd_per_hr: f32,
-    #[serde(default)]
-    pub uptime_s: u64,
-}
-
 /// Distribution-subsystem state record. A periodic, consolidated view of the
 /// cache / directory / registry / gossip / peer-auth state a node observes.
 /// Members and their liveness ride the `membership` channel and node identity
@@ -323,6 +319,59 @@ pub struct TransferRec {
     pub chunks_total: u64,
 }
 
+/// Worker-runtime counters — the routing/error tallies the runtime keeps per
+/// worker on its hot path, aggregated across workers, plus a tick-timing summary.
+/// These are live in-process (the dashboard's `WorkerStats`) but the thin
+/// [`RuntimeStats`] heartbeat carries none of them; this record puts them on the
+/// pipe. All-integer so it derives `Eq`.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WorkerCounters {
+    /// Number of runtime workers folded into these totals.
+    #[serde(default)]
+    pub num_workers: u32,
+    /// Scheduled actors across all workers (the deep value behind
+    /// [`RuntimeStats::scheduled_tasks`]).
+    #[serde(default)]
+    pub scheduled_tasks: u32,
+    #[serde(default)]
+    pub local_sends: u64,
+    #[serde(default)]
+    pub cross_sends: u64,
+    #[serde(default)]
+    pub inbox_sends: u64,
+    #[serde(default)]
+    pub type_mismatches: u64,
+    #[serde(default)]
+    pub panics: u64,
+    #[serde(default)]
+    pub messages_dropped: u64,
+    #[serde(default)]
+    pub restarts: u64,
+    #[serde(default)]
+    pub stops: u64,
+    /// Total messages processed across all workers.
+    #[serde(default)]
+    pub messages_processed: u64,
+    /// Median recent per-tick duration (sum of the 6 phase timings), microseconds.
+    #[serde(default)]
+    pub tick_p50_us: u64,
+}
+
+/// Datastream self-health — the mux's own integrity counters. `assigned` is the
+/// gap-free high-water mark (every position handed out); `dropped` is the frames
+/// lost to mux overflow. A consumer computes the loss rate as `dropped/assigned`;
+/// `loss_rate_ppm` carries it pre-computed (parts-per-million) for convenience.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct DatastreamHealth {
+    #[serde(default)]
+    pub assigned: u64,
+    #[serde(default)]
+    pub dropped: u64,
+    /// `dropped / assigned × 1_000_000`; `0` when nothing has been assigned yet.
+    #[serde(default)]
+    pub loss_rate_ppm: u32,
+}
+
 /// Per-actor runtime detail — the rows behind the aggregate [`RuntimeStats`].
 /// Periodic, on its own channel so a consumer that only wants the summary never
 /// pays to decode the full table.
@@ -366,9 +415,6 @@ impl Record for MembershipTransition {
 impl Record for RuntimeStats {
     const CHANNEL: &'static str = RUNTIME_STATS;
 }
-impl Record for LifecycleCost {
-    const CHANNEL: &'static str = LIFECYCLE_COST;
-}
 impl Record for DistributionState {
     const CHANNEL: &'static str = DIST_STATE;
 }
@@ -378,3 +424,9 @@ impl Record for DatastoreState {
 impl Record for ActorRuntimeDetail {
     const CHANNEL: &'static str = RUNTIME_ACTORS;
 }
+impl Record for WorkerCounters {
+    const CHANNEL: &'static str = RUNTIME_WORKERS;
+}
+impl Record for DatastreamHealth {
+    const CHANNEL: &'static str = DATASTREAM_HEALTH;
+}

crates/datastream/src/emit.rs

@@ -20,12 +20,12 @@ use swactor::process_observer::ProcessOutputObserver;
 use swactor::runtime::Runtime;
 
 use super::catalog::{
-    self, ActorRuntimeDetail, DatastoreState, DistributionState, IdentityRecord, ProcStream,
+    self, ActorRuntimeDetail, DatastoreState, DatastreamHealth, DistributionState, IdentityRecord,
-    Record, RuntimeStats, TransportInternals,
+    MembershipTransition, ProcStream, Record, RuntimeStats, TransportInternals, WorkerCounters,
 };
 use super::frame::{Frame, Lifetime, NodeId, StreamId};
 use super::mux::Mux;
-use super::source::{self, CpuSampler, MembershipTracker};
+use super::source::{self, HostSampler, MembershipTracker};
 use super::wire::{encode_delivery, DatastreamFrame};
 
 /// Where assembled frames go once the mux has ordered them. A sink is the only
@@ -51,6 +51,9 @@ pub struct TickInput<'a> {
     pub runtime: RuntimeStats,
     pub relay_connected: bool,
     pub relay_peers: u32,
+    /// Median SWIM probe round-trip time (ms) this tick; `0` when no probe has
+    /// completed (e.g. a lone node, or before a real SWIM observer is installed).
+    pub rtt_ms_p50: u32,
 }
 
 /// Forwards managed-process output into a node's mux as `proc.<label>.*` text
@@ -77,9 +80,14 @@ impl ProcessOutputObserver for MuxProcObserver {
 pub struct DatastreamEmitter {
     stream_id: StreamId,
     mux: Arc<Mux>,
-    cpu: CpuSampler,
+    host: HostSampler,
     membership: MembershipTracker,
     sink: Box<dyn FrameSink>,
+    /// When set, [`tick`](Self::tick) does **not** synthesize membership from the
+    /// member-list diff (M4) — the caller drives `membership` from a real event
+    /// source via [`submit_membership`](Self::submit_membership), so the diff
+    /// would only duplicate it with empty reasons.
+    external_membership: bool,
 }
 
 impl DatastreamEmitter {
@@ -95,12 +103,20 @@ impl DatastreamEmitter {
         Self {
             stream_id,
             mux,
-            cpu: CpuSampler::new(),
+            host: HostSampler::new(),
             membership: MembershipTracker::new(),
             sink,
+            external_membership: false,
         }
     }
 
+    /// Switch `membership` to an external event source: [`tick`](Self::tick) stops
+    /// diffing the member list (M4), and the caller emits transitions via
+    /// [`submit_membership`](Self::submit_membership) with a real `reason`.
+    pub fn use_external_membership(&mut self) {
+        self.external_membership = true;
+    }
+
     /// The node's mux, for producers (e.g. a raw demo) that submit directly.
     pub fn mux(&self) -> &Arc<Mux> {
         &self.mux
@@ -121,7 +137,7 @@ impl DatastreamEmitter {
         if sample_periodic {
             self.mux.submit(
                 catalog::HOST_RESOURCE,
-                source::read_host_resource(&mut self.cpu).encode(),
+                source::read_host_resource(&mut self.host).encode(),
             );
             self.mux.submit(catalog::RUNTIME_STATS, input.runtime.encode());
             let transport = TransportInternals {
@@ -132,14 +148,39 @@ impl DatastreamEmitter {
                     .filter(|(_, s)| s == "alive")
                     .count() as u32,
                 relay_peers: input.relay_peers,
-                rtt_ms_p50: 0,
+                rtt_ms_p50: input.rtt_ms_p50,
             };
             self.mux
                 .submit(catalog::TRANSPORT_INTERNALS, transport.encode());
+
+            // The pipe reporting on its own integrity: positions assigned vs.
+            // frames dropped on mux overflow. Read before submitting this frame,
+            // so the figures exclude the health frame itself. Rides every tick, so
+            // it ships on any node that ticks (the demo's FleetEmitter included).
+            let assigned = self.mux.assigned();
+            let dropped = self.mux.dropped();
+            let loss_rate_ppm = if assigned > 0 {
+                ((dropped as u128 * 1_000_000) / assigned as u128).min(u32::MAX as u128) as u32
+            } else {
+                0
+            };
+            self.mux.submit(
+                catalog::DATASTREAM_HEALTH,
+                DatastreamHealth {
+                    assigned,
+                    dropped,
+                    loss_rate_ppm,
+                }
+                .encode(),
+            );
         }
 
-        for transition in self.membership.diff(input.members) {
+        // M4 fallback: synthesize membership from the member-list diff, unless the
-            self.mux.submit(catalog::MEMBERSHIP, transition.encode());
+        // caller drives it from a real event source (`use_external_membership`).
+        if !self.external_membership {
+            for transition in self.membership.diff(input.members) {
+                self.mux.submit(catalog::MEMBERSHIP, transition.encode());
+            }
         }
 
         for frame in self.mux.drain() {
@@ -164,6 +205,21 @@ impl DatastreamEmitter {
         self.mux.submit(catalog::RUNTIME_ACTORS, detail.encode());
     }
 
+    /// Submit the aggregated worker-runtime counters (routing/error tallies +
+    /// tick timing) onto the pipe — the deep slice the thin `runtime.stats`
+    /// heartbeat omits. Periodic.
+    pub fn submit_worker_counters(&self, counters: &WorkerCounters) {
+        self.mux.submit(catalog::RUNTIME_WORKERS, counters.encode());
+    }
+
+    /// Submit one membership transition from a real event source (the SWIM
+    /// observer), carrying a non-empty `reason`. Event-driven; pairs with
+    /// [`use_external_membership`](Self::use_external_membership), which turns off
+    /// the M4 diff so this is the sole `membership` source.
+    pub fn submit_membership(&self, transition: &MembershipTransition) {
+        self.mux.submit(catalog::MEMBERSHIP, transition.encode());
+    }
+
     /// Re-emit the identity record once late-bound fields (name, listen addr,
     /// relay URL, version) are known. "Latest wins" on the consumer, so this
     /// supersedes the minimal boot identity emitted in [`new`](Self::new).

crates/datastream/src/source.rs

@@ -70,21 +70,90 @@ impl CpuSampler {
     }
 }
 
-/// Read a host-resource sample: CPU busy percent (via `sampler`) and memory
+/// Samples host network throughput across calls. Like [`CpuSampler`], the rate
-/// from `/proc/meminfo`. Fields we have no source for stay at their default
+/// needs two observations: the first call seeds the byte baseline and reports 0.
-/// (honest zeros), and on a non-Linux host the whole sample is the default.
+#[derive(Default)]
-pub fn read_host_resource(sampler: &mut CpuSampler) -> ResourceSample {
+pub struct NetSampler {
-    let cpu_pct = sampler.sample();
+    /// `(rx_bytes_total, tx_bytes_total, observed_at)` from the previous call.
+    prev: Option<(u64, u64, Instant)>,
+}
+
+impl NetSampler {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// `(rx_kbps, tx_kbps)` since the previous call — **kilobits per second**
+    /// summed across every non-loopback interface. Returns `(0, 0)` on the first
+    /// call (no baseline) and on any platform without `/proc/net/dev`.
+    pub fn sample(&mut self) -> (u32, u32) {
+        let now = Instant::now();
+        let (rx, tx) = match read_net_bytes() {
+            Some(v) => v,
+            None => return (0, 0),
+        };
+        let out = match self.prev {
+            Some((prev_rx, prev_tx, prev_at)) => {
+                let elapsed = now.duration_since(prev_at).as_secs_f64();
+                if elapsed <= 0.0 {
+                    (0, 0)
+                } else {
+                    let to_kbps = |delta: u64| {
+                        ((delta as f64) * 8.0 / 1000.0 / elapsed).clamp(0.0, u32::MAX as f64) as u32
+                    };
+                    (
+                        to_kbps(rx.saturating_sub(prev_rx)),
+                        to_kbps(tx.saturating_sub(prev_tx)),
+                    )
+                }
+            }
+            None => (0, 0),
+        };
+        self.prev = Some((rx, tx, now));
+        out
+    }
+}
+
+/// All the host-resource sampler state a node carries between ticks: the CPU
+/// rate baseline and the network byte baseline. Disk and GPU are point reads, so
+/// they need no state.
+pub struct HostSampler {
+    cpu: CpuSampler,
+    net: NetSampler,
+}
+
+impl HostSampler {
+    pub fn new() -> Self {
+        Self {
+            cpu: CpuSampler::new(),
+            net: NetSampler::new(),
+        }
+    }
+}
+
+impl Default for HostSampler {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Read a host-resource sample: CPU busy percent and network throughput (rates,
+/// via `sampler`'s baselines), memory from `/proc/meminfo`, disk usage from
+/// `statvfs`, and GPU utilization best-effort (NVML/`nvidia-smi`). Every field
+/// has a real host source; on a CPU-only box `gpu_pct` is an honest `0` (no GPU
+/// present — never a fabricated load), and off-Linux the whole sample defaults.
+pub fn read_host_resource(sampler: &mut HostSampler) -> ResourceSample {
+    let cpu_pct = sampler.cpu.sample();
     let (mem_total_mb, mem_used_mb) = read_meminfo_mb().unwrap_or((0, 0));
-    // Fields we have no host source for stay at honest zeros.
+    let (net_rx_kbps, net_tx_kbps) = sampler.net.sample();
     ResourceSample {
         cpu_pct,
         mem_used_mb,
         mem_total_mb,
-        gpu_pct: 0.0,
+        gpu_pct: read_gpu_pct(),
-        disk_used_gb: 0,
+        disk_used_gb: read_disk_used_gb(),
-        net_rx_kbps: 0,
+        net_rx_kbps,
-        net_tx_kbps: 0,
+        net_tx_kbps,
     }
 }
 
@@ -187,6 +256,100 @@ fn read_proc_stat_busy_jiffies() -> Option<u64> {
     Some(total.saturating_sub(idle))
 }
 
+/// Used disk space in whole gigabytes (base-10) of the root filesystem, via
+/// `statvfs("/")`. In a container the root overlay reports its backing store, so
+/// this is a real reading. `0` on failure or off-Linux.
+fn read_disk_used_gb() -> u32 {
+    #[cfg(target_os = "linux")]
+    {
+        // SAFETY: `statvfs` only reads into the zeroed-out struct we provide, and
+        // `"/"` is always a valid, NUL-terminated C path.
+        unsafe {
+            let mut stat: libc::statvfs = std::mem::zeroed();
+            if libc::statvfs(b"/\0".as_ptr().cast(), &mut stat) == 0 {
+                let frsize = stat.f_frsize as u64;
+                let used_blocks = (stat.f_blocks as u64).saturating_sub(stat.f_bfree as u64);
+                let used_bytes = used_blocks.saturating_mul(frsize);
+                return (used_bytes / 1_000_000_000) as u32;
+            }
+        }
+    }
+    0
+}
+
+/// `(rx_bytes_total, tx_bytes_total)` summed across every interface except
+/// loopback, from `/proc/net/dev`. `None` off-Linux or if the file is unreadable.
+fn read_net_bytes() -> Option<(u64, u64)> {
+    #[cfg(target_os = "linux")]
+    {
+        if let Ok(text) = std::fs::read_to_string("/proc/net/dev") {
+            let mut rx_total = 0u64;
+            let mut tx_total = 0u64;
+            // Two header lines, then one `iface: rx_bytes ... tx_bytes ...` per nic.
+            for line in text.lines().skip(2) {
+                let Some((iface, stats)) = line.split_once(':') else {
+                    continue;
+                };
+                if iface.trim() == "lo" {
+                    continue; // loopback is local traffic, not network throughput
+                }
+                let cols: Vec<u64> = stats
+                    .split_whitespace()
+                    .filter_map(|c| c.parse::<u64>().ok())
+                    .collect();
+                // Receive bytes is column 0; transmit bytes is column 8.
+                if cols.len() >= 9 {
+                    rx_total = rx_total.saturating_add(cols[0]);
+                    tx_total = tx_total.saturating_add(cols[8]);
+                }
+            }
+            return Some((rx_total, tx_total));
+        }
+    }
+    None
+}
+
+/// GPU utilization percent, best-effort. Only probes when an NVIDIA GPU is
+/// actually present (the driver dir is populated); on a CPU-only host — the demo
+/// default — this returns an honest `0`, never a fabricated load. Averaged across
+/// GPUs when several are present.
+fn read_gpu_pct() -> f32 {
+    #[cfg(target_os = "linux")]
+    {
+        let present = std::fs::read_dir("/proc/driver/nvidia/gpus")
+            .map(|mut entries| entries.next().is_some())
+            .unwrap_or(false);
+        if present
+            && let Some(pct) = query_nvidia_smi_util()
+        {
+            return pct;
+        }
+    }
+    0.0
+}
+
+/// Average `utilization.gpu` across GPUs via `nvidia-smi`. Only called once a GPU
+/// is known present, so this never runs on the CPU-only demo. `None` on any error.
+#[cfg(target_os = "linux")]
+fn query_nvidia_smi_util() -> Option<f32> {
+    let out = std::process::Command::new("nvidia-smi")
+        .args(["--query-gpu=utilization.gpu", "--format=csv,noheader,nounits"])
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    let text = String::from_utf8_lossy(&out.stdout);
+    let samples: Vec<f32> = text
+        .lines()
+        .filter_map(|l| l.trim().parse::<f32>().ok())
+        .collect();
+    if samples.is_empty() {
+        return None;
+    }
+    Some(samples.iter().sum::<f32>() / samples.len() as f32)
+}
+
 /// `(MemTotal, MemTotal - MemAvailable)` in MiB from `/proc/meminfo`.
 fn read_meminfo_mb() -> Option<(u32, u32)> {
     let text = std::fs::read_to_string("/proc/meminfo").ok()?;
@@ -214,13 +377,15 @@ mod tests {
     fn host_resource_reports_real_memory_on_linux() {
         // On a real Linux host the machine has some memory; a sample that
         // reported zero total would mean we never read the host at all.
-        let mut sampler = CpuSampler::new();
+        let mut sampler = HostSampler::new();
         let sample = read_host_resource(&mut sampler);
         assert!(sample.mem_total_mb > 0, "expected to read MemTotal from the host");
         assert!(
             sample.mem_used_mb <= sample.mem_total_mb,
             "used memory cannot exceed total"
         );
+        // The root filesystem always has some space in use on a real host.
+        assert!(sample.disk_used_gb > 0, "expected statvfs to report used disk");
     }
 
     #[test]

crates/datastream/tests/datastream_support/mod.rs

@@ -18,9 +18,9 @@
 #![allow(dead_code)]
 
 use datastream::catalog::{
-    self, ActorRec, ActorRuntimeDetail, CacheEntryRec, DatastoreState, DistributionState,
+    self, ActorRec, ActorRuntimeDetail, CacheEntryRec, DatastoreState, DatastreamHealth,
-    IdentityRecord, LifecycleCost, MembershipTransition, ObjectRec, ProcStream, Record,
+    DistributionState, IdentityRecord, MembershipTransition, ObjectRec, ProcStream, Record,
-    RegistryEntryRec, ResourceSample, RuntimeStats, TransferRec, TransportInternals,
+    RegistryEntryRec, ResourceSample, RuntimeStats, TransferRec, TransportInternals, WorkerCounters,
 };
 use datastream::frame::{ChannelId, Frame, Position, StreamId};
 use datastream::mux::Mux;
@@ -221,9 +221,28 @@ pub mod payloads {
         }
     }
 
-    /// A provider/lifecycle/cost record.
+    /// Aggregated worker-runtime counters (the `runtime.workers` channel).
-    pub fn lifecycle(phase: &str, uptime_s: u64) -> LifecycleCost {
+    pub fn worker_counters(tick: u64) -> WorkerCounters {
-        LifecycleCost { phase: phase.to_string(), cost_usd_per_hr: 1.27, uptime_s }
+        WorkerCounters {
+            num_workers: 4,
+            scheduled_tasks: 4 + (tick % 3) as u32,
+            local_sends: 100 + tick,
+            cross_sends: 20 + tick,
+            inbox_sends: tick,
+            messages_processed: 1000 + tick * 7,
+            tick_p50_us: 50 + tick,
+            ..Default::default()
+        }
+    }
+
+    /// Datastream self-health (the `datastream.health` channel). `assigned`
+    /// tracks the seed directly so a scenario can pick a frame out by its value.
+    pub fn datastream_health(tick: u64) -> DatastreamHealth {
+        DatastreamHealth {
+            assigned: tick,
+            dropped: tick % 4,
+            loss_rate_ppm: (tick % 4) as u32,
+        }
     }
 
     /// A realistic line of process output (without trailing newline).

crates/datastream/tests/t_datastream.rs

@@ -13,8 +13,8 @@ mod support;
 use std::sync::Arc;
 
 use datastream::catalog::{
-    self, ActorRuntimeDetail, ChannelKind, DatastoreState, DistributionState, IdentityRecord,
+    self, ActorRuntimeDetail, ChannelKind, DatastoreState, DatastreamHealth, DistributionState,
-    LifecycleCost, ProcStream, Record, ResourceSample,
+    IdentityRecord, ProcStream, Record, ResourceSample,
 };
 use datastream::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
 use datastream::ingest::Consumer;
@@ -36,7 +36,7 @@ fn surviving(sent: &[Frame], dropped: &[u64]) -> Vec<Frame> {
 
 /// A node that has emitted a realistic spread of channels: identity, two
 /// resource samples, a transport snapshot, a membership transition, runtime
-/// stats, two lines of process output, and a lifecycle record.
+/// stats, two lines of process output, and a worker-counters record.
 fn busy_node(stream: &StreamId) -> Vec<Frame> {
     let node = Node::new(stream.clone());
     node.emit(&payloads::identity(stream.node.as_str(), stream.life.0)); // 0
@@ -47,7 +47,7 @@ fn busy_node(stream: &StreamId) -> Vec<Frame> {
     node.emit(&payloads::resource(1)); // 5
     node.emit(&payloads::runtime(2)); // 6
     node.emit_text("trainer", ProcStream::Stderr, "WARN cuda oom, retrying"); // 7
-    node.emit(&payloads::lifecycle("running", 1800)); // 8
+    node.emit(&payloads::worker_counters(8)); // 8
     node.emit(&payloads::resource(2)); // 9
     node.sent()
 }
@@ -95,10 +95,11 @@ fn codec_round_trips_every_typed_channel() {
     assert_round_trip(&payloads::transport(4));
     assert_round_trip(&payloads::membership("node-beta", "alive", "suspect"));
     assert_round_trip(&payloads::runtime(5));
-    assert_round_trip(&payloads::lifecycle("running", 3600));
     assert_round_trip(&payloads::dist_state(6));
     assert_round_trip(&payloads::datastore_state(7));
     assert_round_trip(&payloads::actor_detail(8));
+    assert_round_trip(&payloads::worker_counters(9));
+    assert_round_trip(&payloads::datastream_health(10));
 }
 
 fn assert_round_trip<R: Record + PartialEq + std::fmt::Debug>(record: &R) {
@@ -970,7 +971,9 @@ fn kind_iv_deployment_simulation() {
         at_tick(&mut a_ticks, t, a.emit(&payloads::resource(t)));
     }
     at_tick(&mut a_ticks, 7, a.emit(&payloads::resource(7))); // resumes after the outage
-    at_tick(&mut a_ticks, 8, a.emit(&payloads::lifecycle("finalized", 7200))); // finalize
+    // Finalize frame: a datastream.health record whose `assigned` carries the seed,
+    // so the surviving last frame is identifiable by value below.
+    at_tick(&mut a_ticks, 8, a.emit(&payloads::datastream_health(7200))); // finalize
     let sent_a = a.sent();
 
     // ── Node B: lives, and also emits membership transitions ───────────
@@ -1053,8 +1056,8 @@ fn kind_iv_deployment_simulation() {
     assert_eq!(a_stored.gap_spans(), vec![GapSpan { start: 3, end: 5 }], "A: one surfaced gap");
     let finalize = a_stored.frames().last().unwrap();
     assert_eq!(
-        LifecycleCost::decode(&finalize.payload).unwrap().phase,
+        DatastreamHealth::decode(&finalize.payload).unwrap().assigned,
-        "finalized",
+        7200,
         "A's run ends with the finalize frame, delivered after the outage"
     );
 

crates/distribution/src/bin/swactor-iroh-relay.rs

@@ -166,6 +166,8 @@ async fn main() -> ExitCode {
                         runtime: RuntimeStats::default(),
                         relay_connected: true,
                         relay_peers: 0,
+                        // The relay runs no SWIM probe loop, so no real RTT.
+                        rtt_ms_p50: 0,
                     },
                     true,
                 );

crates/distribution/src/iroh_driver.rs

@@ -582,6 +582,32 @@ impl IrohDriver {
             .unwrap_or(0)
     }
 
+    /// This node's location cache: every directory route whose host is a *remote*
+    /// peer — the `(actor, host)` locations the node has learned in order to route
+    /// across the network. Read off the same directory `RouteView` mirror as
+    /// [`directory_route_count`](Self::directory_route_count), but filtered to
+    /// peer-hosted actors: a node never needs to "cache" the location of an actor
+    /// it hosts itself, so self-hosted routes are excluded. This is the honest
+    /// `dist.state.cache_*` source, distinct from the all-routes count above.
+    pub fn location_cache_entries(&self) -> Vec<(ActorAddress, NodeId)> {
+        let self_id = self.node_id();
+        self.actor_bridge
+            .as_ref()
+            .and_then(|b| {
+                b.route_view.read().ok().map(|view| {
+                    let mut entries: Vec<(ActorAddress, NodeId)> = view
+                        .iter()
+                        .filter(|(_, host)| **host != self_id)
+                        .map(|(actor, host)| (*actor, *host))
+                        .collect();
+                    // Stable order so the dashboard table doesn't reshuffle each tick.
+                    entries.sort_by(|a, b| a.0.0.cmp(&b.0.0));
+                    entries
+                })
+            })
+            .unwrap_or_default()
+    }
+
     /// Get a snapshot of all join statuses.
     pub fn join_statuses(&self) -> HashMap<NodeId, JoinStatus> {
         self.join_statuses.lock().unwrap().clone()

crates/distribution/src/registry.rs

@@ -53,6 +53,30 @@ pub struct RegistryEntry {
     pub tombstone: bool,
 }
 
+// ─── Observability snapshot ──────────────────────────────────────────────────
+
+/// A point-in-time, read-only projection of the registry, decoupled from the
+/// CRDT internals so a consumer (the node's telemetry tick) can read live
+/// registry figures off a shared mirror without touching the actor.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct RegistrySnapshot {
+    /// Total entries, including tombstones.
+    pub size: usize,
+    /// How many of those entries are tombstones.
+    pub tombstones: usize,
+    /// Every entry, sorted by name for stable rendering.
+    pub entries: Vec<RegistryEntrySnapshot>,
+}
+
+/// One entry in a [`RegistrySnapshot`] — a named actor location, possibly a tombstone.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RegistryEntrySnapshot {
+    pub name: String,
+    pub actor_addr: ActorAddress,
+    pub node_id: NodeId,
+    pub tombstone: bool,
+}
+
 // ─── Events ─────────────────────────────────────────────────────────────────
 
 /// Events emitted when the registry changes.
@@ -283,6 +307,32 @@ impl ClusterRegistry {
         self.entries.values()
     }
 
+    /// A read-only projection of the whole registry for observability — the
+    /// size, tombstone count, and every entry in one consistent pass. The node
+    /// publishes this to a [`RegistryView`](crate::registry_actor::RegistryView)
+    /// mirror so telemetry can read live registry figures without `ask`-ing the
+    /// actor on its hot path.
+    pub fn snapshot(&self) -> RegistrySnapshot {
+        let mut entries: Vec<RegistryEntrySnapshot> = self
+            .entries
+            .values()
+            .map(|e| RegistryEntrySnapshot {
+                name: e.name.clone(),
+                actor_addr: e.actor_addr,
+                node_id: e.node_id,
+                tombstone: e.tombstone,
+            })
+            .collect();
+        // Stable order so the dashboard table doesn't reshuffle each tick.
+        entries.sort_by(|a, b| a.name.cmp(&b.name));
+        let tombstones = entries.iter().filter(|e| e.tombstone).count();
+        RegistrySnapshot {
+            size: entries.len(),
+            tombstones,
+            entries,
+        }
+    }
+
     // ─── Internal ───────────────────────────────────────────────────────
 
     fn next_generation(&self, name: &str) -> u64 {

crates/distribution/src/registry_actor.rs

@@ -10,16 +10,23 @@
 //! node returns.
 
 use std::collections::BTreeSet;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use swactor::actor::{ActorAddress, ActorInterface};
 use swactor::runtime::Ctx;
 
 use crate::messages::RegistryGossip;
-use crate::registry::{ClusterRegistry, RegistryConfig};
+use crate::registry::{ClusterRegistry, RegistryConfig, RegistrySnapshot};
 use crate::swim::actor::{MembershipChanged, PeerDirectory};
 use crate::types::{MemberState, NodeId};
 
+/// A single-writer read-mirror of the registry's observable state, published by
+/// the [`RegistryActor`] after each change and read by the node's telemetry tick
+/// (the same discipline as the directory's
+/// [`RouteView`](crate::transport_bridge::RouteView)). Installed via
+/// [`RegistryActor::with_view`]; absent in tests/examples that don't observe it.
+pub type RegistryView = Arc<RwLock<RegistrySnapshot>>;
+
 /// Everything the `RegistryActor` receives, as one enum (only `Gossip` crosses
 /// the wire; the rest are local control — see [`crate::messages::actor_codec_registry`]).
 #[derive(Clone)]
@@ -55,6 +62,9 @@ pub struct RegistryActor {
     /// Round-robins the gossip target across alive peers, one per `Tick` — the
     /// standalone analog of "piggyback on the next probe".
     fanout_cursor: usize,
+    /// Optional read-mirror the node's telemetry tick observes. Republished
+    /// after each registry change. `None` when no one is observing.
+    view: Option<RegistryView>,
 }
 
 impl RegistryActor {
@@ -69,6 +79,23 @@ impl RegistryActor {
             peer_directory,
             alive: BTreeSet::new(),
             fanout_cursor: 0,
+            view: None,
+        }
+    }
+
+    /// Install a read-mirror that this actor republishes after each change, so a
+    /// telemetry consumer can read live registry figures without `ask`-ing it.
+    /// Seeds the mirror with the current (empty) snapshot immediately.
+    pub fn with_view(mut self, view: RegistryView) -> Self {
+        *view.write().expect("registry view poisoned") = self.registry.snapshot();
+        self.view = Some(view);
+        self
+    }
+
+    /// Republish the registry snapshot to the read-mirror, if one is installed.
+    fn publish(&self) {
+        if let Some(view) = &self.view {
+            *view.write().expect("registry view poisoned") = self.registry.snapshot();
         }
     }
 
@@ -116,16 +143,19 @@ impl ActorInterface for RegistryActor {
                     self.alive.remove(&m.node_id);
                     let size = self.cluster_size();
                     self.registry.tombstone_node(m.node_id, size);
+                    self.publish();
                 }
                 MemberState::Suspect => {}
             },
             RegistryIn::RegisterName { name, actor_addr } => {
                 let size = self.cluster_size();
                 self.registry.register(name, actor_addr, self.self_id, size);
+                self.publish();
             }
             RegistryIn::UnregisterName { name } => {
                 let size = self.cluster_size();
                 self.registry.unregister(&name, self.self_id, size);
+                self.publish();
             }
             RegistryIn::ResolveName { name, reply } => {
                 let binding = self.registry.resolve(&name);
@@ -134,10 +164,13 @@ impl ActorInterface for RegistryActor {
             RegistryIn::Gossip(g) => {
                 let size = self.cluster_size();
                 self.registry.merge_batch(g.entries, size);
+                self.publish();
             }
             RegistryIn::Tick => {
                 self.registry.gc_tick();
                 self.disseminate(ctx);
+                // GC may have reaped tombstones; keep the mirror current.
+                self.publish();
             }
         }
     }

crates/distribution/src/snapshot.rs

@@ -28,6 +28,12 @@ pub struct MemberInfo {
     pub relay_url: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub node_name: Option<String>,
+    /// Cause of this member's most recent SWIM liveness transition (the
+    /// production observer's reason string, e.g. "ping-received",
+    /// "probe-timeout"). `None` until a transition has been observed; the
+    /// old state-diff path could never carry this.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub reason: Option<String>,
 }
 
 /// Snapshot of a single LRU cache entry.

crates/distribution/src/swim/actor.rs

@@ -24,7 +24,7 @@ use crate::crypto::verify_directory_entry;
 use crate::messages::{Ack, IndirectAck, JoinRequest, JoinResponse, Ping, PingReq};
 use crate::types::{DirectoryEntry, MemberState, NodeId};
 
-use super::node::{NodeAction, SwimNode};
+use super::node::{NodeAction, SwimNode, SwimObserver};
 use super::probe::SwimConfig;
 
 // ─── The Binding (§3.2) ───────────────────────────────────────────────────────
@@ -179,6 +179,15 @@ impl SwimActor {
         }
     }
 
+    /// Install a [`SwimObserver`] on the wrapped engine, so probe RTT and
+    /// membership transitions (with their cause) surface to a telemetry sink.
+    /// Non-breaking builder over [`new`](Self::new); callers that don't observe
+    /// SWIM leave it off (production previously always did).
+    pub fn with_observer(mut self, observer: Box<dyn SwimObserver>) -> Self {
+        self.node.set_observer(observer);
+        self
+    }
+
     /// Resolve `to` through the Binding and send `msg`; on a binding miss or a
     /// transport-rejected send, deliver `SendFailed{to}` back to ourselves
     /// (§4.3). This never recurses synchronously — the failure is a mailbox

crates/distribution/src/swim/mod.rs

@@ -4,3 +4,4 @@ pub mod dissemination;
 pub mod node;
 pub mod lifeguard;
 pub mod actor;
+pub mod telemetry;

crates/distribution/src/swim/telemetry.rs

@@ -0,0 +1,160 @@
+//! Production SWIM telemetry — the observer the live node installs.
+//!
+//! Outside `simulation`, SWIM ran with `observer = None`, so three live signals
+//! never reached the datastream: per-probe round-trip time, the recent probe
+//! targets, and the *cause* of each membership transition (the M4 state-diff can
+//! see *that* a peer changed but not *why*). This installs a real
+//! [`SwimObserver`] that reconstructs all three from the observation stream, and
+//! exposes them for the node's telemetry tick to read.
+//!
+//! It is a side-channel diagnostic: it never feeds back into the protocol, takes
+//! `&self` (interior mutability behind one `Mutex`), and is safe to read from the
+//! node's main loop while SWIM fires observations on its own worker thread.
+
+use std::collections::{HashMap, VecDeque};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use crate::types::{MemberState, NodeId};
+
+use super::node::{SwimObservation, SwimObserver};
+
+/// Recent RTT samples kept for the running median.
+const RTT_RING: usize = 64;
+/// Recent probe targets kept (most recent last) — matches the probe engine's own
+/// history depth so `dist.state.recent_probe_targets` looks the same either way.
+const TARGET_RING: usize = 16;
+/// Cap on undrained transitions, so a consumer that stops draining can't grow
+/// this without bound. Oldest are dropped first (a lost transition is a gap, not
+/// a renumber — the same tolerance as the mux).
+const TRANSITION_CAP: usize = 256;
+/// In-flight probes older than this are pruned defensively. The probe state
+/// machine resolves every probe (ack or timeout), so this only guards against a
+/// dropped observation leaking an entry forever.
+const IN_FLIGHT_TTL: Duration = Duration::from_secs(30);
+
+/// One captured membership transition, carrying the real cause string the
+/// state-diff path could never know.
+#[derive(Debug, Clone)]
+pub struct ObservedTransition {
+    pub peer: NodeId,
+    pub from: Option<MemberState>,
+    pub to: MemberState,
+    pub reason: &'static str,
+}
+
+#[derive(Default)]
+struct Inner {
+    /// `(target, sequence)` → when its probe was sent, to time the round-trip.
+    in_flight: HashMap<(NodeId, u64), Instant>,
+    /// Recent round-trip samples (ms), newest last.
+    rtts: VecDeque<u32>,
+    /// Recent probe targets, newest last.
+    targets: VecDeque<NodeId>,
+    /// Transitions awaiting drain by the node's telemetry tick.
+    transitions: VecDeque<ObservedTransition>,
+}
+
+/// The installed SWIM observer plus the readouts the node consumes each tick.
+/// Construct with [`new`](Self::new), install a clone as the observer
+/// (`Arc<SwimTelemetry>` implements [`SwimObserver`]), and read the rest.
+pub struct SwimTelemetry {
+    inner: Mutex<Inner>,
+}
+
+impl SwimTelemetry {
+    /// A fresh, empty telemetry sink behind an `Arc` (shared between the observer
+    /// install and the reading node).
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            inner: Mutex::new(Inner::default()),
+        })
+    }
+
+    /// Median (p50) of the recent round-trip samples in milliseconds, or `0` when
+    /// no probe has completed yet (an honest zero — not a fabricated latency).
+    pub fn rtt_ms_p50(&self) -> u32 {
+        let inner = self.inner.lock().expect("swim telemetry poisoned");
+        if inner.rtts.is_empty() {
+            return 0;
+        }
+        let mut samples: Vec<u32> = inner.rtts.iter().copied().collect();
+        samples.sort_unstable();
+        samples[samples.len() / 2]
+    }
+
+    /// The recent probe targets (most recent last).
+    pub fn recent_targets(&self) -> Vec<NodeId> {
+        self.inner
+            .lock()
+            .expect("swim telemetry poisoned")
+            .targets
+            .iter()
+            .copied()
+            .collect()
+    }
+
+    /// Take the transitions captured since the last call (FIFO, then cleared).
+    pub fn drain_transitions(&self) -> Vec<ObservedTransition> {
+        self.inner
+            .lock()
+            .expect("swim telemetry poisoned")
+            .transitions
+            .drain(..)
+            .collect()
+    }
+
+    /// The most recent transition cause per peer, **without** draining the queue.
+    /// A snapshot reader (e.g. the orchestrator's Distribution view) uses this to
+    /// label each member with *why* it last changed state; draining stays
+    /// reserved for the fleet emitter's `membership` channel. Reads oldest→newest
+    /// so the latest reason per peer wins.
+    pub fn last_reasons(&self) -> HashMap<NodeId, &'static str> {
+        let inner = self.inner.lock().expect("swim telemetry poisoned");
+        let mut out = HashMap::new();
+        for t in &inner.transitions {
+            out.insert(t.peer, t.reason);
+        }
+        out
+    }
+
+    fn record(&self, observation: SwimObservation) {
+        let mut inner = self.inner.lock().expect("swim telemetry poisoned");
+        match observation {
+            SwimObservation::ProbeSent { target, sequence, .. } => {
+                // Drop any leaked in-flight entries before tracking a new probe.
+                inner.in_flight.retain(|_, sent| sent.elapsed() < IN_FLIGHT_TTL);
+                inner.in_flight.insert((target, sequence), Instant::now());
+                if inner.targets.len() >= TARGET_RING {
+                    inner.targets.pop_front();
+                }
+                inner.targets.push_back(target);
+            }
+            SwimObservation::ProbeAcked { target, sequence, .. } => {
+                if let Some(sent) = inner.in_flight.remove(&(target, sequence)) {
+                    let rtt = sent.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                    if inner.rtts.len() >= RTT_RING {
+                        inner.rtts.pop_front();
+                    }
+                    inner.rtts.push_back(rtt);
+                }
+            }
+            SwimObservation::ProbeTimedOut { target, sequence, .. } => {
+                // A timeout is not a round-trip — drop the in-flight entry, no sample.
+                inner.in_flight.remove(&(target, sequence));
+            }
+            SwimObservation::Transition { peer, from, to, reason } => {
+                if inner.transitions.len() >= TRANSITION_CAP {
+                    inner.transitions.pop_front();
+                }
+                inner.transitions.push_back(ObservedTransition { peer, from, to, reason });
+            }
+        }
+    }
+}
+
+impl SwimObserver for Arc<SwimTelemetry> {
+    fn observe(&self, observation: SwimObservation) {
+        self.record(observation);
+    }
+}

crates/node/README.md

@@ -18,6 +18,27 @@ Once connected, **SWIM protocol** handles cluster membership: protocol probes ev
 
 **Peer auth** operates in two modes: open (no `peers_file`) or allow-list (`peers.json`). In allow-list mode, SWIM messages from unknown nodes are dropped at the transport layer. New peers can be added via `swactor join` or the dashboard UI, both of which hot-update the allow-list.
 
+> **Note (deferred, not urgent): gossip opens a fresh QUIC stream per message.**
+> The iroh driver reuses the per-peer *connection* but opens and finishes a new
+> uni-stream for every gossip/SWIM message (`iroh_driver.rs` `send_wire` →
+> `open_uni`/`finish` per send; reader does `accept_uni` + `read_to_end` per
+> message). The stream-per-message shape exists only because the wire frame has
+> no payload-length field and relies on stream-EOF to delimit. No per-message RTT
+> is paid (uni-streams are unilateral), but each send costs a tokio task spawn, a
+> fresh read-side allocation, and a slot against `max_concurrent_uni_streams`.
+>
+> Two cheaper shapes, when it's worth doing:
+> - **QUIC datagrams** for the small probe traffic (Ping/Ack/PingReq). One
+>   datagram = one self-delimiting message, no stream state at all, and
+>   best-effort delivery *matches* SWIM's own loss-tolerance instead of fighting
+>   it with redundant QUIC retransmits. Capped at ~path-MTU, so it doesn't cover
+>   large gossip (e.g. a `JoinResponse` with a big member list).
+> - **One persistent length-prefixed stream per connection** for the larger /
+>   must-arrive messages — the same persistent-stream discipline used for blob
+>   edges.
+>
+> Low priority; tracked here so it isn't lost.
+
 ### Relay
 
 Nodes with a public IP auto-promote to embedded relay servers (port 3340). Candidacy is evaluated at startup: the node checks its outbound IP is non-RFC1918 and the relay port is bindable. Relay URLs are announced via SWIM gossip so other nodes discover them automatically. Nodes behind NAT use relays for indirect connectivity — this is why the probe timeout is 600ms instead of the typical 300ms.

crates/node/src/main.rs

@@ -953,7 +953,7 @@ fn run_iroh(
     // transport bridge (decode inbound → actor mailbox; actor outbound → iroh).
     use distribution::directory_actor::{DirectoryActor, DirectoryIn};
     use distribution::node_metadata_actor::{MetadataActor, MetadataIn};
-    use distribution::registry_actor::{RegistryActor, RegistryIn};
+    use distribution::registry_actor::{RegistryActor, RegistryIn, RegistryView};
     use distribution::swim::actor::{SwimActor, SwimIn};
     use distribution::transport_bridge::{
         IrohPeerDirectory, IrohRouteBinder, Outbox, RelayMirror, RouteView, RouteViewTransport,
@@ -965,25 +965,28 @@ fn run_iroh(
         Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
     let route_view: RouteView =
         Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
+    // Read-mirror of the cluster registry the telemetry tick reads to fill the
+    // `dist.state` registry fields (size / tombstones / entries).
+    let registry_view: RegistryView = Arc::new(std::sync::RwLock::new(Default::default()));
+    // Production SWIM observer: reconstructs probe RTT, recent probe targets, and
+    // membership transitions (with cause) from the SWIM observation stream.
+    let swim_telemetry = distribution::swim::telemetry::SwimTelemetry::new();
     let peer_directory = Arc::new(IrohPeerDirectory::new(
         Arc::clone(&transport_router),
         Arc::clone(&outbox),
     ));
 
     let swim_addr = rt
-        .spawn(SwimActor::new(
+        .spawn(
-            node_id,
+            SwimActor::new(node_id, swim_config, Instant::now(), peer_directory.clone())
-            swim_config,
+                .with_observer(Box::new(Arc::clone(&swim_telemetry))),
-            Instant::now(),
+        )
-            peer_directory.clone(),
-        ))
         .expect("spawn SwimActor");
     let registry_addr = rt
-        .spawn(RegistryActor::new(
+        .spawn(
-            node_id,
+            RegistryActor::new(node_id, registry_config, peer_directory.clone())
-            registry_config,
+                .with_view(Arc::clone(&registry_view)),
-            peer_directory.clone(),
+        )
-        ))
         .expect("spawn RegistryActor");
     let metadata_addr = rt
         .spawn(MetadataActor::new(
@@ -1218,6 +1221,10 @@ fn run_iroh(
         },
         Box::new(sink),
     );
+    // The node drives `membership` from the SWIM observer (real transitions with a
+    // cause), so turn off the emitter's member-list diff to avoid duplicate,
+    // reason-less transitions.
+    emitter.use_external_membership();
     rt.set_process_output_observer(emitter.process_observer());
     // Stream datastore op events onto the node's own datastream.
     if let Some(metrics) = &ds_metrics {
@@ -1373,10 +1380,11 @@ fn run_iroh(
                     (members, relay_peers)
                 };
 
-                // Consolidated distribution-subsystem state. Directory route count
+                // Consolidated distribution-subsystem state. Directory route count,
-                // and peer-auth are live; cache / registry / probe targets live in
+                // peer-auth, the cluster registry (via `registry_view`), and the
-                // the actors and would be surfaced via read-mirrors — currently
+                // location cache (remote routes off the directory's RouteView) are
-                // empty, exactly as the pre-datastream snapshot reported them.
+                // live; probe targets still come from the actors and remain empty
+                // until the SWIM observer mirror lands.
                 {
                     let (mode, count) = {
                         let pa = peer_auth.lock().unwrap();
@@ -1386,10 +1394,40 @@ fn run_iroh(
                             ("allow-list".to_string(), pa.list_peers().len() as u32)
                         }
                     };
+                    let registry = registry_view.read().unwrap();
+                    let registry_entries = registry
+                        .entries
+                        .iter()
+                        .map(|e| catalog::RegistryEntryRec {
+                            name: e.name.clone(),
+                            actor_addr: hex(&e.actor_addr.0),
+                            node_id: hex(&e.node_id.0),
+                            tombstone: e.tombstone,
+                        })
+                        .collect();
+                    let cache = driver.location_cache_entries();
+                    let cache_entries = cache
+                        .iter()
+                        .map(|(addr, host)| catalog::CacheEntryRec {
+                            actor_addr: hex(&addr.0),
+                            node_id: hex(&host.0),
+                        })
+                        .collect();
+                    let recent_probe_targets = swim_telemetry
+                        .recent_targets()
+                        .iter()
+                        .map(|t| hex(&t.0))
+                        .collect();
                     emitter.submit_dist_state(&DistributionState {
                         directory_route_count: driver.directory_route_count() as u32,
                         peer_auth_mode: mode,
                         authorized_peer_count: count,
+                        registry_size: registry.size as u32,
+                        registry_tombstones: registry.tombstones as u32,
+                        registry_entries,
+                        cache_size: cache.len() as u32,
+                        cache_entries,
+                        recent_probe_targets,
                         ..Default::default()
                     });
                 }
@@ -1443,10 +1481,53 @@ fn run_iroh(
                         })
                         .collect();
                     emitter.submit_actor_detail(&ActorRuntimeDetail { actors });
+
+                    // Worker-runtime counters (W7): the routing/error tallies and
+                    // tick timing the runtime keeps per worker — live in-process but
+                    // never on the pipe until now. Aggregated across workers from the
+                    // same snapshot.
+                    let mut wc = catalog::WorkerCounters {
+                        num_workers: rs.workers.len() as u32,
+                        ..Default::default()
+                    };
+                    for w in &rs.workers {
+                        wc.scheduled_tasks += w.num_actors as u32;
+                        wc.local_sends += w.local_sends;
+                        wc.cross_sends += w.cross_sends;
+                        wc.inbox_sends += w.inbox_sends;
+                        wc.type_mismatches += w.type_mismatches;
+                        wc.panics += w.panics;
+                        wc.messages_dropped += w.messages_dropped;
+                        wc.restarts += w.restarts;
+                        wc.stops += w.stops;
+                        wc.messages_processed += w.messages_processed;
+                    }
+                    let mut tick_us: Vec<u64> = rs
+                        .tick_timings
+                        .iter()
+                        .flatten()
+                        .map(|t| t.phase_us.iter().sum())
+                        .collect();
+                    tick_us.sort_unstable();
+                    wc.tick_p50_us = tick_us.get(tick_us.len() / 2).copied().unwrap_or(0);
+                    emitter.submit_worker_counters(&wc);
                 }
 
-                // Periodic host/runtime/transport samples + event-driven membership
+                // Real membership transitions (with cause) from the SWIM observer,
-                // transitions, then drain — the sink renders the node-local dashboard.
+                // submitted before the tick so they drain on this iteration. Replaces
+                // the emitter's reason-less member-list diff (`use_external_membership`).
+                for t in swim_telemetry.drain_transitions() {
+                    emitter.submit_membership(&datastream::catalog::MembershipTransition {
+                        peer: hex(&t.peer.0),
+                        from: t.from.map(member_state_str).unwrap_or("unknown").to_string(),
+                        to: member_state_str(t.to).to_string(),
+                        reason: t.reason.to_string(),
+                    });
+                }
+
+                // Periodic host/runtime/transport samples, then drain — the sink
+                // renders the node-local dashboard. `rtt_ms_p50` is the SWIM
+                // observer's real probe round-trip median (0 until a probe completes).
                 {
                     let rs = rt.stats();
                     emitter.tick(
@@ -1459,6 +1540,7 @@ fn run_iroh(
                             },
                             relay_connected: driver.home_relay_url().is_some(),
                             relay_peers,
+                            rtt_ms_p50: swim_telemetry.rtt_ms_p50(),
                         },
                         true,
                     );
@@ -1594,6 +1676,17 @@ fn spawn_actors(
     addrs
 }
 
+/// The `membership` channel's state strings, matching the member-list diff's
+/// convention so the consumer renders observer-driven and diffed transitions alike.
+fn member_state_str(state: distribution::types::MemberState) -> &'static str {
+    use distribution::types::MemberState;
+    match state {
+        MemberState::Alive => "alive",
+        MemberState::Suspect => "suspect",
+        MemberState::Dead => "dead",
+    }
+}
+
 fn hex(bytes: &[u8]) -> String {
     bytes.iter().map(|b| format!("{b:02x}")).collect()
 }

examples/single-gpu-inference/src/cluster.rs

@@ -482,6 +482,8 @@ impl ClusterNode {
                 label: None,
                 relay_url,
                 node_name: None,
+                // No SWIM observer wired in this single-node example.
+                reason: None,
             });
         }
         snap.members = members;