diff --git a/.gitignore b/.gitignore
index 8ea342e..a0e90fe 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,4 +27,7 @@ deploy.toml
# Local dev node state
.dev-node/
.dev-cluster/
-.sim-cluster/
\ No newline at end of file
+.sim-cluster/
+
+# vast.ai run logs/artifacts
+.vastai-logs/
\ No newline at end of file
diff --git a/Cargo.lock b/Cargo.lock
index 30408f6..3c3728d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1100,11 +1100,13 @@ dependencies = [
name = "dashboard"
version = "0.1.0"
dependencies = [
+ "anyhow",
"axum",
"clap",
"crossbeam-queue",
"crossterm",
"ctrlc",
+ "distribution",
"ratatui",
"serde",
"serde_json",
diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml
index 4601dea..5d8db61 100644
--- a/crates/dashboard/Cargo.toml
+++ b/crates/dashboard/Cargo.toml
@@ -17,12 +17,23 @@ ratatui = { version = "0.29", optional = true, default-features = false, feature
crossterm = { version = "0.28", optional = true }
clap = { version = "4", features = ["derive"], optional = true }
ctrlc = "3"
+anyhow = { version = "1", optional = true }
+distribution = { path = "../distribution", optional = true }
[features]
default = []
tui = ["dep:ratatui", "dep:crossterm"]
+# Offline replay viewer example: loads a finalized diagnostics bundle
+# (tarball or spool dir) and serves a localhost web UI for scrubbing
+# through the event timeline.
+replay-viewer = ["dep:anyhow", "dep:distribution", "distribution/collector"]
[[bin]]
name = "swactor-tui"
path = "src/bin/tui.rs"
required-features = ["tui"]
+
+[[example]]
+name = "replay_viewer"
+path = "examples/replay_viewer.rs"
+required-features = ["replay-viewer"]
diff --git a/crates/dashboard/examples/replay_viewer.html b/crates/dashboard/examples/replay_viewer.html
new file mode 100644
index 0000000..c58d40e
--- /dev/null
+++ b/crates/dashboard/examples/replay_viewer.html
@@ -0,0 +1,657 @@
+
+
+
+
+Replay Viewer
+
+
+
+
+
+
+
+
+ 0.0 s
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/crates/dashboard/examples/replay_viewer.rs b/crates/dashboard/examples/replay_viewer.rs
new file mode 100644
index 0000000..ac330b1
--- /dev/null
+++ b/crates/dashboard/examples/replay_viewer.rs
@@ -0,0 +1,608 @@
+//! Visual replay viewer for diagnostics bundles.
+//!
+//! Loads a finalized deployment bundle (`vastai-N3-*.tar.gz`), an
+//! uncompressed collector spool dir (`vastai-N3-*/`), or a simulation
+//! bundle dir (`manifest.json` + `events.ndjson` + `snapshots/...`),
+//! normalizes both formats into a single in-memory event timeline, and
+//! serves a one-page HTML viewer on localhost.
+//!
+//! Run:
+//! cargo run --example replay_viewer -p dashboard \
+//! --features replay-viewer --
+
+use std::collections::BTreeMap;
+use std::fs;
+use std::io::{BufRead, BufReader};
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use anyhow::{Context, Result, anyhow, bail};
+use axum::Router;
+use axum::extract::State;
+use axum::http::header;
+use axum::response::{Html, IntoResponse};
+use axum::routing::get;
+use serde::Serialize;
+use serde_json::Value;
+
+use distribution::diagnostics::postproc::Bundle;
+
+// ─── wire types ────────────────────────────────────────────────────────────
+
+#[derive(Clone, Copy, Serialize)]
+#[serde(rename_all = "lowercase")]
+enum SourceKind {
+ Deployment,
+ Sim,
+}
+
+#[derive(Clone, Copy, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+enum Severity {
+ Info,
+ Notable,
+ Error,
+}
+
+#[derive(Serialize)]
+struct UnifiedEvent {
+ t_ms: f64,
+ node_label: String,
+ kind: String,
+ severity: Severity,
+ fields: Value,
+}
+
+#[derive(Serialize)]
+struct NodeInfo {
+ label: String,
+ role: Option,
+ color: String,
+}
+
+#[derive(Serialize)]
+struct KindInfo {
+ kind: String,
+ count: u64,
+ color: String,
+}
+
+#[derive(Serialize)]
+struct SnapshotInfo {
+ t_ms: f64,
+ node_label: String,
+ summary: String,
+}
+
+#[derive(Serialize)]
+struct BundleView {
+ run_id: String,
+ source_kind: SourceKind,
+ t_start_ms: f64,
+ t_end_ms: f64,
+ nodes: Vec,
+ event_kinds: Vec,
+ events: Vec,
+ snapshots: Vec,
+}
+
+// ─── format detection ──────────────────────────────────────────────────────
+
+enum DetectedFormat {
+ DeploymentTar(PathBuf),
+ DeploymentDir(PathBuf),
+ Sim(PathBuf),
+}
+
+fn detect_format(path: &Path) -> Result {
+ let md = fs::metadata(path)
+ .with_context(|| format!("stat {}", path.display()))?;
+ if md.is_file() {
+ let name = path.file_name().and_then(|s| s.to_str()).unwrap_or("");
+ if name.ends_with(".tar.gz") || name.ends_with(".tgz") {
+ return Ok(DetectedFormat::DeploymentTar(path.to_path_buf()));
+ }
+ bail!("unrecognized file: {} (expected .tar.gz)", path.display());
+ }
+ if path.join("MANIFEST.json").is_file() {
+ return Ok(DetectedFormat::DeploymentDir(path.to_path_buf()));
+ }
+ if path.join("manifest.json").is_file() && path.join("events.ndjson").is_file() {
+ return Ok(DetectedFormat::Sim(path.to_path_buf()));
+ }
+ bail!(
+ "could not classify {} — expected .tar.gz, dir with MANIFEST.json, or sim dir with manifest.json+events.ndjson",
+ path.display()
+ )
+}
+
+// ─── deployment loaders ────────────────────────────────────────────────────
+
+fn load_deployment_tar(path: &Path) -> Result {
+ let bundle = Bundle::parse_path(path)
+ .map_err(|e| anyhow!("parse {}: {e}", path.display()))?;
+ let nodes_meta: Vec<(String, Option)> = bundle
+ .manifest
+ .nodes
+ .iter()
+ .map(|n| (n.label.clone(), n.role.clone()))
+ .collect();
+
+ let mut events: Vec = Vec::new();
+ let mut snapshots: Vec = Vec::new();
+ let mut t0: u64 = bundle.manifest.run_start_collector_ms.unwrap_or(u64::MAX);
+ for node in bundle.nodes.values() {
+ for ev in &node.events {
+ if ev.wall_ms < t0 {
+ t0 = ev.wall_ms;
+ }
+ }
+ for snap in &node.snapshots {
+ if snap.wall_ms < t0 {
+ t0 = snap.wall_ms;
+ }
+ }
+ }
+ if t0 == u64::MAX {
+ t0 = 0;
+ }
+
+ let mut t_end: f64 = 0.0;
+ for node in bundle.nodes.values() {
+ for ev in &node.events {
+ let value = serde_json::to_value(&ev.event).unwrap_or(Value::Null);
+ let kind = unified_kind_from_value(&value);
+ let severity = severity_for(&kind, &value);
+ let t_ms = (ev.wall_ms.saturating_sub(t0)) as f64;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ events.push(UnifiedEvent {
+ t_ms,
+ node_label: node.label.clone(),
+ kind,
+ severity,
+ fields: value,
+ });
+ }
+ for snap in &node.snapshots {
+ let t_ms = (snap.wall_ms.saturating_sub(t0)) as f64;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ let body = serde_json::to_value(&snap.body).unwrap_or(Value::Null);
+ snapshots.push(SnapshotInfo {
+ t_ms,
+ node_label: node.label.clone(),
+ summary: summarize_snapshot(&body),
+ });
+ }
+ }
+ events.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+ snapshots.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+
+ Ok(BundleView {
+ run_id: bundle.run_id,
+ source_kind: SourceKind::Deployment,
+ t_start_ms: 0.0,
+ t_end_ms: t_end,
+ nodes: assign_node_colors(nodes_meta),
+ event_kinds: tally_event_kinds(&events),
+ events,
+ snapshots,
+ })
+}
+
+#[derive(serde::Deserialize)]
+struct DirManifestNode {
+ node_id_hex: String,
+ label: String,
+ #[serde(default)]
+ role: Option,
+}
+
+#[derive(serde::Deserialize)]
+struct DirManifest {
+ run_id: String,
+ #[serde(default)]
+ run_start_collector_ms: Option,
+ #[serde(default)]
+ nodes: Vec,
+}
+
+fn load_deployment_dir(root: &Path) -> Result {
+ let manifest_bytes = fs::read(root.join("MANIFEST.json"))
+ .with_context(|| format!("read MANIFEST.json under {}", root.display()))?;
+ let manifest: DirManifest = serde_json::from_slice(&manifest_bytes)
+ .context("parse MANIFEST.json")?;
+
+ let hex_to_label: BTreeMap = manifest
+ .nodes
+ .iter()
+ .map(|n| (n.node_id_hex.to_lowercase(), n.label.clone()))
+ .collect();
+ let nodes_meta: Vec<(String, Option)> = manifest
+ .nodes
+ .iter()
+ .map(|n| (n.label.clone(), n.role.clone()))
+ .collect();
+
+ let mut events: Vec = Vec::new();
+ let mut snapshots: Vec = Vec::new();
+ let mut t0: u64 = manifest.run_start_collector_ms.unwrap_or(u64::MAX);
+ let mut raw_events: Vec<(String, Value)> = Vec::new();
+ let mut raw_snapshots: Vec<(String, Value)> = Vec::new();
+
+ for entry in fs::read_dir(root).with_context(|| format!("readdir {}", root.display()))? {
+ let entry = entry?;
+ if !entry.file_type()?.is_dir() {
+ continue;
+ }
+ let dir_name = entry.file_name().to_string_lossy().into_owned();
+ let label = hex_to_label
+ .get(&dir_name.to_lowercase())
+ .cloned()
+ .unwrap_or_else(|| {
+ if dir_name.len() >= 8 {
+ format!("node-{}", &dir_name[..8])
+ } else {
+ dir_name.clone()
+ }
+ });
+ for f in fs::read_dir(entry.path())? {
+ let f = f?;
+ let fname = f.file_name().to_string_lossy().into_owned();
+ if fname.starts_with("events-") && fname.ends_with(".json") {
+ let bytes = fs::read(f.path())
+ .with_context(|| format!("read {}", f.path().display()))?;
+ let batch: Value = serde_json::from_slice(&bytes)
+ .with_context(|| format!("parse {}", f.path().display()))?;
+ if let Some(arr) = batch.as_array() {
+ for ev in arr {
+ if let Some(wall) = ev.get("wall_ms").and_then(Value::as_u64) {
+ if wall < t0 {
+ t0 = wall;
+ }
+ }
+ raw_events.push((label.clone(), ev.clone()));
+ }
+ }
+ } else if fname.starts_with("snapshot-") && fname.ends_with(".json") {
+ let bytes = fs::read(f.path())?;
+ let snap: Value = serde_json::from_slice(&bytes)
+ .with_context(|| format!("parse {}", f.path().display()))?;
+ if let Some(wall) = snap.get("wall_ms").and_then(Value::as_u64) {
+ if wall < t0 {
+ t0 = wall;
+ }
+ }
+ raw_snapshots.push((label.clone(), snap));
+ }
+ }
+ }
+ if t0 == u64::MAX {
+ t0 = 0;
+ }
+
+ let mut t_end: f64 = 0.0;
+ for (label, ev) in raw_events {
+ let wall = ev.get("wall_ms").and_then(Value::as_u64).unwrap_or(t0);
+ let t_ms = wall.saturating_sub(t0) as f64;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ let kind = unified_kind_from_value(&ev);
+ let severity = severity_for(&kind, &ev);
+ events.push(UnifiedEvent {
+ t_ms,
+ node_label: label,
+ kind,
+ severity,
+ fields: ev,
+ });
+ }
+ for (label, snap) in raw_snapshots {
+ let wall = snap.get("wall_ms").and_then(Value::as_u64).unwrap_or(t0);
+ let t_ms = wall.saturating_sub(t0) as f64;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ let body = snap.get("body").cloned().unwrap_or(Value::Null);
+ snapshots.push(SnapshotInfo {
+ t_ms,
+ node_label: label,
+ summary: summarize_snapshot(&body),
+ });
+ }
+ events.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+ snapshots.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+
+ Ok(BundleView {
+ run_id: manifest.run_id,
+ source_kind: SourceKind::Deployment,
+ t_start_ms: 0.0,
+ t_end_ms: t_end,
+ nodes: assign_node_colors(nodes_meta),
+ event_kinds: tally_event_kinds(&events),
+ events,
+ snapshots,
+ })
+}
+
+// ─── sim loader ────────────────────────────────────────────────────────────
+
+fn load_sim(dir: &Path) -> Result {
+ let manifest_bytes = fs::read(dir.join("manifest.json"))
+ .with_context(|| format!("read manifest.json under {}", dir.display()))?;
+ let manifest: Value = serde_json::from_slice(&manifest_bytes).context("parse manifest.json")?;
+ let run_id = manifest
+ .get("scenario_name")
+ .and_then(Value::as_str)
+ .or_else(|| manifest.get("name").and_then(Value::as_str))
+ .map(|s| s.to_string())
+ .unwrap_or_else(|| dir.file_name().and_then(|s| s.to_str()).unwrap_or("sim").to_string());
+
+ let f = fs::File::open(dir.join("events.ndjson"))
+ .with_context(|| format!("open events.ndjson under {}", dir.display()))?;
+ let reader = BufReader::new(f);
+
+ let mut hosts: BTreeMap = BTreeMap::new();
+ let mut events: Vec = Vec::new();
+ let mut t_end: f64 = 0.0;
+
+ for line in reader.lines() {
+ let line = line?;
+ if line.trim().is_empty() {
+ continue;
+ }
+ let v: Value = match serde_json::from_str(&line) {
+ Ok(v) => v,
+ Err(_) => continue,
+ };
+ let t_ns = v.get("virtual_time_ns").and_then(Value::as_u64).unwrap_or(0);
+ let host = v
+ .get("host_id")
+ .and_then(Value::as_str)
+ .unwrap_or("unknown")
+ .to_string();
+ let kind = v
+ .get("kind_tag")
+ .and_then(Value::as_str)
+ .unwrap_or("event")
+ .to_string();
+ let payload = v.get("event").cloned().unwrap_or_else(|| v.clone());
+ let severity = severity_for(&kind, &payload);
+ let t_ms = (t_ns as f64) / 1.0e6;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ hosts.entry(host.clone()).or_insert(());
+ events.push(UnifiedEvent {
+ t_ms,
+ node_label: host,
+ kind,
+ severity,
+ fields: payload,
+ });
+ }
+ events.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+
+ let mut snapshots: Vec = Vec::new();
+ let snap_root = dir.join("snapshots");
+ if snap_root.is_dir() {
+ for host_entry in fs::read_dir(&snap_root)? {
+ let host_entry = host_entry?;
+ if !host_entry.file_type()?.is_dir() {
+ continue;
+ }
+ let host = host_entry.file_name().to_string_lossy().into_owned();
+ for f in fs::read_dir(host_entry.path())? {
+ let f = f?;
+ let bytes = fs::read(f.path())?;
+ let snap: Value = match serde_json::from_slice(&bytes) {
+ Ok(v) => v,
+ Err(_) => continue,
+ };
+ let t_ns = snap.get("virtual_time_ns").and_then(Value::as_u64).unwrap_or(0);
+ let t_ms = (t_ns as f64) / 1.0e6;
+ if t_ms > t_end {
+ t_end = t_ms;
+ }
+ snapshots.push(SnapshotInfo {
+ t_ms,
+ node_label: host.clone(),
+ summary: summarize_snapshot(&snap),
+ });
+ }
+ }
+ snapshots.sort_by(|a, b| a.t_ms.partial_cmp(&b.t_ms).unwrap_or(std::cmp::Ordering::Equal));
+ }
+
+ let nodes_meta: Vec<(String, Option)> =
+ hosts.into_keys().map(|h| (h, None)).collect();
+
+ Ok(BundleView {
+ run_id,
+ source_kind: SourceKind::Sim,
+ t_start_ms: 0.0,
+ t_end_ms: t_end,
+ nodes: assign_node_colors(nodes_meta),
+ event_kinds: tally_event_kinds(&events),
+ events,
+ snapshots,
+ })
+}
+
+// ─── helpers ───────────────────────────────────────────────────────────────
+
+fn unified_kind_from_value(v: &Value) -> String {
+ // Deployment event records use serde tag `type`; collector spool
+ // files use a flat `kind` string. The `Custom` variant carries an
+ // inner `kind` field we want to surface as `Custom:`.
+ if let Some(tag) = v.get("type").and_then(Value::as_str) {
+ if tag == "Custom" {
+ if let Some(inner) = v.get("kind").and_then(Value::as_str) {
+ return format!("Custom:{inner}");
+ }
+ }
+ return tag.to_string();
+ }
+ if let Some(k) = v.get("kind").and_then(Value::as_str) {
+ return k.to_string();
+ }
+ "event".to_string()
+}
+
+fn severity_for(kind: &str, fields: &Value) -> Severity {
+ if kind == "Error" || kind.starts_with("error") {
+ return Severity::Error;
+ }
+ if kind == "DialOutcome" {
+ if let Some(outcome) = fields.get("outcome") {
+ let ok = outcome
+ .as_str()
+ .map(|s| s == "Success")
+ .or_else(|| {
+ outcome
+ .as_object()
+ .map(|m| m.keys().next().map(|k| k == "Success").unwrap_or(false))
+ })
+ .unwrap_or(false);
+ return if ok { Severity::Info } else { Severity::Error };
+ }
+ }
+ if kind == "SwimTransition" {
+ if fields.get("to").and_then(Value::as_str) == Some("Dead") {
+ return Severity::Error;
+ }
+ return Severity::Notable;
+ }
+ if kind == "ConnectionCacheInvalidated"
+ || kind == "RelayChanged"
+ || kind == "IrohConnTypeChanged"
+ {
+ return Severity::Notable;
+ }
+ if kind.to_ascii_lowercase().contains("drop") {
+ return Severity::Notable;
+ }
+ Severity::Info
+}
+
+fn summarize_snapshot(body: &Value) -> String {
+ let reach = body
+ .get("reachability")
+ .and_then(Value::as_array)
+ .map(|a| a.len())
+ .unwrap_or(0);
+ let tail = body
+ .get("events")
+ .and_then(Value::as_array)
+ .map(|a| a.len())
+ .unwrap_or(0);
+ let mut parts = vec![format!("peers={reach}"), format!("tail={tail}")];
+ if body.get("iroh").is_some() && !body.get("iroh").unwrap().is_null() {
+ parts.push("iroh".into());
+ }
+ if body.get("swim").is_some() && !body.get("swim").unwrap().is_null() {
+ parts.push("swim".into());
+ }
+ if body.get("host").is_some() && !body.get("host").unwrap().is_null() {
+ parts.push("host".into());
+ }
+ parts.join(" ")
+}
+
+const NODE_PALETTE: &[&str] = &[
+ "#4a90e2", "#f06292", "#81c784", "#ffb74d", "#ba68c8", "#4dd0e1", "#aed581", "#ff8a65",
+];
+
+fn assign_node_colors(meta: Vec<(String, Option)>) -> Vec {
+ meta.into_iter()
+ .enumerate()
+ .map(|(i, (label, role))| NodeInfo {
+ label,
+ role,
+ color: NODE_PALETTE[i % NODE_PALETTE.len()].to_string(),
+ })
+ .collect()
+}
+
+const KIND_PALETTE: &[&str] = &[
+ "#4a90e2", "#ffb74d", "#81c784", "#f06292", "#ba68c8", "#4dd0e1", "#aed581", "#ff8a65",
+ "#e57373", "#9575cd", "#64b5f6", "#ffd54f",
+];
+
+fn tally_event_kinds(events: &[UnifiedEvent]) -> Vec {
+ let mut counts: BTreeMap = BTreeMap::new();
+ for ev in events {
+ *counts.entry(ev.kind.clone()).or_insert(0) += 1;
+ }
+ let mut pairs: Vec<(String, u64)> = counts.into_iter().collect();
+ pairs.sort_by(|a, b| b.1.cmp(&a.1));
+ pairs
+ .into_iter()
+ .enumerate()
+ .map(|(i, (kind, count))| KindInfo {
+ kind,
+ count,
+ color: KIND_PALETTE[i % KIND_PALETTE.len()].to_string(),
+ })
+ .collect()
+}
+
+// ─── routes ────────────────────────────────────────────────────────────────
+
+async fn get_index() -> Html<&'static str> {
+ Html(INDEX_HTML)
+}
+
+async fn get_bundle(State(view): State>) -> impl IntoResponse {
+ (
+ [(header::CONTENT_TYPE, "application/json")],
+ view.as_str().to_owned(),
+ )
+}
+
+const INDEX_HTML: &str = include_str!("replay_viewer.html");
+
+// ─── main ──────────────────────────────────────────────────────────────────
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() -> Result<()> {
+ let path = std::env::args().nth(1).ok_or_else(|| {
+ anyhow!("usage: replay_viewer ")
+ })?;
+ let path = PathBuf::from(path);
+
+ let view = match detect_format(&path)? {
+ DetectedFormat::DeploymentTar(p) => load_deployment_tar(&p)?,
+ DetectedFormat::DeploymentDir(p) => load_deployment_dir(&p)?,
+ DetectedFormat::Sim(p) => load_sim(&p)?,
+ };
+
+ let n_events = view.events.len();
+ let n_nodes = view.nodes.len();
+ let span_s = view.t_end_ms / 1000.0;
+ let run_id = view.run_id.clone();
+ let payload = Arc::new(serde_json::to_string(&view).context("serialize bundle view")?);
+
+ let app = Router::new()
+ .route("/", get(get_index))
+ .route("/api/bundle", get(get_bundle))
+ .with_state(payload);
+
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+ .await
+ .context("bind 127.0.0.1:0")?;
+ let addr: SocketAddr = listener.local_addr()?;
+
+ println!(
+ "replay-viewer: run={run_id} nodes={n_nodes} events={n_events} span={span_s:.1}s"
+ );
+ println!(" open: http://{addr}");
+
+ axum::serve(listener, app).await.context("axum serve")?;
+ Ok(())
+}
diff --git a/crates/distribution/Cargo.toml b/crates/distribution/Cargo.toml
index 4518bae..b377869 100644
--- a/crates/distribution/Cargo.toml
+++ b/crates/distribution/Cargo.toml
@@ -8,6 +8,7 @@ default = []
iroh = ["dep:iroh", "dep:tokio", "dep:iroh-metrics"]
relay = [
"iroh",
+ "collector",
"dep:iroh-relay",
"tokio/macros",
"tokio/signal",
diff --git a/crates/distribution/build.rs b/crates/distribution/build.rs
new file mode 100644
index 0000000..dd2211c
--- /dev/null
+++ b/crates/distribution/build.rs
@@ -0,0 +1,130 @@
+//! Build-time discovery of dependency versions that the runtime needs to
+//! report honestly in the diagnostics bundle.
+//!
+//! Today we only emit the `iroh` version (gap 6 in
+//! `examples/pipeline-parallel-inference/N3_OBSERVABILITY_UPGRADE_SPEC.md`),
+//! but the same parser handles any other crate the diagnostics layer
+//! reports about — add another `emit_version` call and a const in
+//! `diagnostics::dep_versions` when one comes up.
+//!
+//! Versions come from the workspace `Cargo.lock`, located by walking
+//! upward from `OUT_DIR`'s ancestors until a sibling file named
+//! `Cargo.lock` is found. We never fall back to a hardcoded literal —
+//! the whole point of this is to keep the bundle honest about what was
+//! linked, so a missing lockfile is a build failure, not a silent zero.
+
+use std::env;
+use std::fs;
+use std::path::{Path, PathBuf};
+
+fn main() {
+ let lock_path = find_cargo_lock().expect(
+ "build.rs could not locate Cargo.lock — diagnostics requires it for honest version \
+ reporting. Run from inside the workspace.",
+ );
+ println!("cargo:rerun-if-changed={}", lock_path.display());
+ let body = fs::read_to_string(&lock_path)
+ .unwrap_or_else(|e| panic!("read {}: {e}", lock_path.display()));
+ emit_version(&body, "iroh", "DISTRIBUTION_IROH_VERSION");
+ emit_build_git_sha();
+}
+
+/// Best-effort `git rev-parse HEAD` capture. If the repo is unavailable
+/// or the call fails, the env var is left unset and the runtime
+/// constant resolves to `None`. The point is to keep the bundle honest
+/// — never fabricate a placeholder — while letting builds outside a
+/// git checkout still succeed.
+fn emit_build_git_sha() {
+ println!("cargo:rerun-if-env-changed=DISTRIBUTION_GIT_SHA_OVERRIDE");
+ if let Ok(override_sha) = env::var("DISTRIBUTION_GIT_SHA_OVERRIDE") {
+ let trimmed = override_sha.trim();
+ if !trimmed.is_empty() {
+ println!("cargo:rustc-env=DISTRIBUTION_GIT_SHA={trimmed}");
+ return;
+ }
+ }
+ let manifest_dir = env::var_os("CARGO_MANIFEST_DIR").map(PathBuf::from);
+ if let Some(dir) = manifest_dir {
+ if let Ok(out) = std::process::Command::new("git")
+ .args(["rev-parse", "HEAD"])
+ .current_dir(&dir)
+ .output()
+ {
+ if out.status.success() {
+ let sha = String::from_utf8_lossy(&out.stdout).trim().to_string();
+ if !sha.is_empty() {
+ println!("cargo:rustc-env=DISTRIBUTION_GIT_SHA={sha}");
+ // Re-run when the head commit changes so a dirty
+ // rebuild reports the right SHA.
+ let head = locate_git_head(&dir);
+ if let Some(head) = head {
+ println!("cargo:rerun-if-changed={}", head.display());
+ }
+ }
+ }
+ }
+ }
+}
+
+fn locate_git_head(start: &Path) -> Option {
+ let mut dir = start;
+ loop {
+ let candidate = dir.join(".git").join("HEAD");
+ if candidate.is_file() {
+ return Some(candidate);
+ }
+ dir = dir.parent()?;
+ }
+}
+
+fn emit_version(lockfile: &str, package: &str, env_var: &str) {
+ let version = lockfile_version(lockfile, package).unwrap_or_else(|| {
+ panic!(
+ "Cargo.lock has no entry for `{package}`. The diagnostics layer reports its version \
+ and refuses to make one up."
+ )
+ });
+ println!("cargo:rustc-env={env_var}={version}");
+}
+
+/// Lookup the `version = "..."` line of the `[[package]]` block named
+/// `package`. Naive but adequate — Cargo.lock is well-formed TOML with
+/// predictable layout. We deliberately avoid a TOML dependency in
+/// build.rs so this stays a zero-cost build script.
+fn lockfile_version(body: &str, package: &str) -> Option {
+ let needle = format!("name = \"{package}\"");
+ let mut lines = body.lines();
+ while let Some(line) = lines.next() {
+ if line.trim() != needle {
+ continue;
+ }
+ for next in lines.by_ref() {
+ let t = next.trim();
+ if t.starts_with("[[package]]") {
+ // Reached the next package without a version line.
+ return None;
+ }
+ if let Some(rest) = t.strip_prefix("version = \"") {
+ if let Some(end) = rest.find('"') {
+ return Some(rest[..end].to_string());
+ }
+ }
+ }
+ }
+ None
+}
+
+fn find_cargo_lock() -> Option {
+ // CARGO_MANIFEST_DIR points at the crate root. Walk up looking for
+ // a sibling Cargo.lock — both the workspace root and standalone
+ // crates have one.
+ let start = env::var_os("CARGO_MANIFEST_DIR").map(PathBuf::from)?;
+ let mut dir: &Path = &start;
+ loop {
+ let candidate = dir.join("Cargo.lock");
+ if candidate.is_file() {
+ return Some(candidate);
+ }
+ dir = dir.parent()?;
+ }
+}
diff --git a/crates/distribution/src/bin/swactor-iroh-relay.rs b/crates/distribution/src/bin/swactor-iroh-relay.rs
index 5d347f1..4c08280 100644
--- a/crates/distribution/src/bin/swactor-iroh-relay.rs
+++ b/crates/distribution/src/bin/swactor-iroh-relay.rs
@@ -8,11 +8,41 @@
//!
//! Defaults to plain HTTP on `0.0.0.0:7843`. No TLS — meant for diagnostic
//! / experimental deployments behind a firewall the operator controls.
+//!
+//! ## Observability (spec §1, gap 1)
+//!
+//! When `SWACTOR_DIAG_COLLECTOR_URL` is set this binary boots its own
+//! diagnostics aggregator with `Role::custom("relay")` and installs a
+//! [`distribution::diagnostics::RelayObservability`] helper on it. The
+//! aggregator reports into the same collector / bundle as the cluster's
+//! nodes, so the post-processor's `## Relay sessions` section can
+//! correlate relay-reported close reasons against node-side
+//! `connection_cache[peer].last_failure_reason`. Per-session lifecycle
+//! events are emitted via [`RelayObservability::note_session_opened`]
+//! / `note_session_closed` — wired today as a skeleton (iroh-relay's
+//! native server does not expose session hooks); when the upstream
+//! relay grows them, the call sites slot in here and the bundle
+//! starts answering "who closed and why" automatically.
use std::net::SocketAddr;
use std::process::ExitCode;
+use std::sync::Arc;
+
+use distribution::diagnostics::aggregator::{spawn_periodic_snapshots, PeriodicConfig};
+use distribution::diagnostics::{
+ wall_ms_now, Aggregator, HttpSink, Identity, RelayObservability, RelayServerIntrospector,
+ Role, SinkConfig, SnapshotSignal, GIT_SHA, IROH_VERSION,
+};
+use distribution::diagnostics::sink::{DynEmitter, EventEmitter};
+use distribution::types::NodeId;
const DEFAULT_BIND: &str = "0.0.0.0:7843";
+const ENV_COLLECTOR_URL: &str = "SWACTOR_DIAG_COLLECTOR_URL";
+const ENV_RUN_ID: &str = "SWACTOR_DIAG_RUN_ID";
+const ENV_SPOOL_DIR: &str = "SWACTOR_DIAG_SPOOL_DIR";
+const ENV_RELAY_LABEL: &str = "SWACTOR_DIAG_RELAY_LABEL";
+const DEFAULT_RUN_ID: &str = "pp-run";
+const DEFAULT_SPOOL_DIR: &str = "/tmp/swactor-diag-relay";
fn print_help() {
eprintln!(
@@ -101,6 +131,13 @@ async fn main() -> ExitCode {
let url = format!("http://{}:{}/", url_host, addr.port());
eprintln!("swactor-iroh-relay: listening on {bind} (advertised URL: {url})");
+ // Spec §1: when a collector is configured, this relay reports
+ // into the same bundle as the cluster nodes under its own
+ // identity. Holding `_diag` keeps the aggregator + spawned tasks
+ // alive for the lifetime of the binary; dropping it at shutdown
+ // flushes the sink.
+ let _diag = install_relay_diagnostics(&url);
+
if let Err(e) = tokio::signal::ctrl_c().await {
eprintln!("swactor-iroh-relay: signal listen failed: {e}");
return ExitCode::from(1);
@@ -108,3 +145,111 @@ async fn main() -> ExitCode {
eprintln!("swactor-iroh-relay: shutdown signal received");
ExitCode::SUCCESS
}
+
+/// Holder for the relay's diagnostics state. `RelayObservability` is
+/// exposed so a future call site that hooks iroh-relay's session
+/// lifecycle can record opens/closes through it.
+struct RelayDiag {
+ _agg: Arc>,
+ _observability: Arc,
+}
+
+fn install_relay_diagnostics(advertised_url: &str) -> Option {
+ let collector_url = std::env::var(ENV_COLLECTOR_URL).ok()?;
+ let collector_url = collector_url.trim().to_string();
+ if collector_url.is_empty() {
+ return None;
+ }
+ let run_id = env_string(ENV_RUN_ID).unwrap_or_else(|| DEFAULT_RUN_ID.to_string());
+ let spool_dir = std::path::PathBuf::from(
+ env_string(ENV_SPOOL_DIR).unwrap_or_else(|| DEFAULT_SPOOL_DIR.to_string()),
+ );
+
+ // The relay has no `iroh::Endpoint` and therefore no `NodeId`. We
+ // synthesize a deterministic-per-process id from the advertised
+ // URL so the bundle's manifest keeps a stable handle on this
+ // relay across reboots within a run.
+ let node_id = synthesize_node_id(advertised_url);
+ let node_id_hex: String = node_id.0.iter().map(|b| format!("{:02x}", b)).collect();
+
+ let mut identity = Identity::new(node_id, Role::custom("relay"), run_id.clone())
+ .with_process_start(wall_ms_now());
+ identity = identity.with_host_context(
+ distribution::diagnostics::HostContext::from_env()
+ .with_iroh_version(IROH_VERSION)
+ .with_git_sha(GIT_SHA.map(|s| s.to_string()))
+ .with_binary_version(option_env!("CARGO_PKG_VERSION").map(|s| s.to_string()))
+ .with_home_relay_url(Some(advertised_url.to_string())),
+ );
+ if let Some(label) = env_string(ENV_RELAY_LABEL) {
+ // Caller can override the friendly hostname carried in the
+ // host context so the bundle reader recognises the relay by
+ // its operational name rather than just its synthetic node id.
+ identity.hostname = Some(label);
+ }
+
+ let signal = SnapshotSignal::new();
+ let sink_config = SinkConfig::new(
+ collector_url.clone(),
+ run_id.clone(),
+ node_id_hex,
+ spool_dir,
+ )
+ .with_snapshot_signal(signal.clone());
+ let sink = match HttpSink::new(sink_config) {
+ Ok(s) => s,
+ Err(e) => {
+ eprintln!(
+ "swactor-iroh-relay: HttpSink::new failed ({e}); continuing without diagnostics"
+ );
+ return None;
+ }
+ };
+ let aggregator = Arc::new(Aggregator::new(identity, sink));
+
+ let observability = Arc::new(RelayObservability::new());
+ let emitter: DynEmitter = aggregator.clone() as Arc;
+ observability.set_emitter(emitter);
+ aggregator.set_relay_server_introspector(
+ observability.clone() as Arc,
+ );
+
+ // Periodic snapshots: same cadence as nodes so the bundle reader
+ // can line snapshots up by wall_ms.
+ let _ = spawn_periodic_snapshots(aggregator.clone(), PeriodicConfig::default(), signal);
+
+ eprintln!(
+ "swactor-iroh-relay: diagnostics installed (collector={collector_url} run_id={run_id} \
+ role=relay url={advertised_url})"
+ );
+ Some(RelayDiag {
+ _agg: aggregator,
+ _observability: observability,
+ })
+}
+
+fn env_string(var: &str) -> Option {
+ std::env::var(var)
+ .ok()
+ .map(|s| s.trim().to_string())
+ .filter(|s| !s.is_empty())
+}
+
+/// FNV-1a 64-bit folded across the URL bytes, repeated to fill 32
+/// bytes. Deterministic per-URL so the relay's identity is stable
+/// across restarts within a run, without taking on a key dependency.
+fn synthesize_node_id(seed: &str) -> NodeId {
+ const FNV_OFFSET: u64 = 0xcbf29ce484222325;
+ const FNV_PRIME: u64 = 0x100000001b3;
+ let mut hash: u64 = FNV_OFFSET;
+ for b in seed.bytes() {
+ hash ^= b as u64;
+ hash = hash.wrapping_mul(FNV_PRIME);
+ }
+ let mut out = [0u8; 32];
+ for (i, chunk) in out.chunks_mut(8).enumerate() {
+ let seeded = hash.wrapping_add(i as u64);
+ chunk.copy_from_slice(&seeded.to_be_bytes());
+ }
+ NodeId(out)
+}
diff --git a/crates/distribution/src/diagnostics/aggregator.rs b/crates/distribution/src/diagnostics/aggregator.rs
index 88430ec..eca1409 100644
--- a/crates/distribution/src/diagnostics/aggregator.rs
+++ b/crates/distribution/src/diagnostics/aggregator.rs
@@ -19,8 +19,9 @@ use crate::diagnostics::identity::Identity;
use crate::diagnostics::reachability::{PeerReachability, StateTransition, node_id_hex};
use crate::diagnostics::sink::{EventEmitter, Sink};
use crate::diagnostics::snapshot::{
- HostIntrospector, IrohIntrospector, ProbeIntrospector, ProcessIntrospector, Snapshot,
- SnapshotBody, SnapshotTrigger, SwimIntrospector, VastaiIntrospector,
+ HostIntrospector, IrohIntrospector, ProbeIntrospector, ProcessIntrospector,
+ RegistryIntrospector, RelayServerIntrospector, Snapshot, SnapshotBody, SnapshotTrigger,
+ SubprocessIntrospector, SwimIntrospector, VastaiIntrospector,
};
use crate::types::NodeId;
@@ -43,6 +44,9 @@ pub struct Aggregator {
probe_introspector: Mutex