feat(xtask): provisioning-reconciler-demo with live fleet control

`cargo xtask provisioning-reconciler-demo [--port n] [--nodes n]` boots a lightweight orchestrator for visual, human-checked E2E confirmation of the provisioning reconciler: swactor engine + real ClusterDriver + demo provider, with node children re-exec'ing the same xtask binary in node role and joining the supervisor over real iroh connections. - supervisor actor owns driver/provider/shape on a 250ms wall-clock tick, mirroring the production ClusterReconciler poll semantics; emits prov.reconciler.events/snapshot plus per-node lifecycle/status streams - k8s-styled reconciler view: ready/desired header, node stage cards, commands-out and events-in feeds - dashboard `demo-control` feature: POST /control/{kill,provision,remove} + Fleet Control view; regular builds compile none of it (symbol-verified) - fleet cards fold proc.<node>.lifecycle and node.status heartbeats into per-node pid/state pills that stay live - hardening: exe resolution survives binary replacement by rebuilds, spawn failures feed back as BootstrapFailed so the reconciler retries instead of wedging at SshReady, teardown skips exit waits for never-started children Verified in-browser: boot 3/3 converged with real joins; dashboard kill dips and fully recovers with a replacement; provision +1 → 4/4; remove −2 graceful teardown → 2/2; child process count matches reconciler nodes.
2026-08-15 18:10:52 +04:00 · 2026-08-15 18:10:52 +04:00 · d19dd91324
commit d19dd91324
parent b3dbd7ce11
18 changed files with 2599 additions and 29 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -899,11 +899,11 @@ name = "dashboard"
 version = "0.1.0"
 dependencies = [
 "axum",
 "telemetry",
 "parking_lot",
 "serde",
 "serde_json",
 "swactor",
 "telemetry",
 "tokio",
 "tokio-stream",
 ]
@ -952,23 +952,10 @@ dependencies = [
 name = "data-plane"
 version = "0.1.0"
 dependencies = [
 "libc",
 "serde",
 "swactor",
 "telemetry",
 "libc",
 "serde",
 "swactor",
 ]
 [[package]]
 name = "telemetry"
 version = "0.1.0"
 dependencies = [
 "crossbeam-channel",
 "iroh",
 "libc",
 "serde",
 "serde_json",
 "swactor",
 "swactor-transport",
 ]
 [[package]]
@ -1138,13 +1125,13 @@ dependencies = [
 name = "distribution"
 version = "0.1.0"
 dependencies = [
 "telemetry",
 "libc",
 "proptest",
 "serde",
 "serde_json",
 "swactor",
 "swactor-transport",
 "telemetry",
 "uuid",
 ]
@ -2141,7 +2128,6 @@ name = "iroh-driver"
 version = "0.1.0"
 dependencies = [
 "crossbeam-channel",
 "telemetry",
 "distribution",
 "iroh",
 "iroh-relay",
@ -2151,6 +2137,7 @@ dependencies = [
 "swactor",
 "swactor-engine",
 "swactor-transport",
 "telemetry",
 "tokio",
 ]
@ -2524,7 +2511,6 @@ dependencies = [
 "blake3",
 "dashboard",
 "data-plane",
 "telemetry",
 "distribution",
 "iroh",
 "iroh-driver",
@ -2539,6 +2525,7 @@ dependencies = [
 "swactor-process",
 "swactor-transport",
 "swactor-vastai",
 "telemetry",
 "tokio",
 "toml 0.8.23",
 "ureq",
@ -3362,6 +3349,7 @@ dependencies = [
 name = "provisioning"
 version = "0.1.0"
 dependencies = [
 "parking_lot",
 "serde",
 "serde_json",
 ]
@ -4418,12 +4406,12 @@ name = "swactor-process"
 version = "0.1.0"
 dependencies = [
 "crossbeam-queue",
 "telemetry",
 "libc",
 "serde",
 "serde_json",
 "serde_yaml",
 "swactor",
 "telemetry",
 ]
 [[package]]
@ -4513,6 +4501,19 @@ version = "0.12.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1"
 [[package]]
 name = "telemetry"
 version = "0.1.0"
 dependencies = [
 "crossbeam-channel",
 "iroh",
 "libc",
 "serde",
 "serde_json",
 "swactor",
 "swactor-transport",
 ]
 [[package]]
 name = "tempfile"
 version = "3.27.0"
@ -5837,8 +5838,21 @@ name = "xtask"
 version = "0.1.0"
 dependencies = [
 "blake3",
 "dashboard",
 "distribution",
 "iroh",
 "iroh-driver",
 "libc",
 "parking_lot",
 "provisioning",
 "serde",
 "serde_json",
 "swactor",
 "swactor-engine",
 "swactor-process",
 "swactor-transport",
 "telemetry",
 "tokio",
 ]
 [[package]]
--- a/crates/dashboard/Cargo.toml
+++ b/crates/dashboard/Cargo.toml
@ -13,3 +13,8 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio.workspace = true
 tokio-stream = "0.1"
 [features]
 # Live control actions (kill processes, provision nodes) for the
 # provisioning-reconciler demo. Never enabled in shipping builds.
 demo-control = []
--- a/crates/dashboard/src/control.rs
+++ b/crates/dashboard/src/control.rs
@ -0,0 +1,44 @@
 //! Demo-only control plane: write actions out of the dashboard.
 //!
 //! This module exists only under the `demo-control` feature. The dashboard's
 //! data path stays read-only in every regular build; the provisioning
 //! reconciler demo turns this feature on so a human can kill provisioned
 //! processes and request new ones from the fleet view.
 //!
 //! Routes (only present when the feature is enabled and a control sink is
 //! installed):
 //! - `POST /control/kill`     body `{"node": "<stream node id>"}`
 //! - `POST /control/provision` body `{"count": 1}`
 use std::sync::mpsc::Sender;
 use std::sync::OnceLock;
 use serde::Deserialize;
 /// A control command issued from the dashboard UI.
 #[derive(Clone, Debug, Deserialize)]
 pub enum ControlCommand {
    /// Kill the process backing the fleet card identified by its stream node.
    Kill { node: String },
    /// Ask the reconciler to provision `count` additional nodes.
    Provision { count: u32 },
    /// Lower the desired cluster size by `count` nodes (graceful scale
    /// down: teardown through the reconciler, not a kill).
    Remove { count: u32 },
 }
 static CONTROL_SENDER: OnceLock<Sender<ControlCommand>> = OnceLock::new();
 /// Install the sink that receives dashboard-issued control commands.
 ///
 /// Called once by the embedding demo before the HTTP server starts. Without a
 /// sink the control routes answer `503 Service Unavailable`.
 pub fn set_control_sender(sender: Sender<ControlCommand>) {
    let _ = CONTROL_SENDER.set(sender);
 }
 pub(crate) fn dispatch(command: ControlCommand) -> bool {
    CONTROL_SENDER
        .get()
        .is_some_and(|sender| sender.send(command).is_ok())
 }
--- a/crates/dashboard/src/demo_control.rs
+++ b/crates/dashboard/src/demo_control.rs
@ -0,0 +1,112 @@
 //! Demo-only fleet control view (`demo-control` feature).
 //!
 //! A k8s-style control panel beside the fleet view: one row per provisioned
 //! process (streamed on `proc.<node>.lifecycle`), each with a kill action,
 //! plus a provision action that asks the reconciler for more nodes. Inert in
 //! regular builds — this module compiles only under `demo-control`.
 use std::collections::BTreeMap;
 use std::time::Instant;
 use parking_lot::Mutex;
 use serde::Serialize;
 use serde_json::Value;
 use telemetry::frame::{Frame, StreamId};
 use crate::view::DashboardView;
 use crate::FrameEvent;
 #[derive(Clone, Serialize)]
 struct ProcessEntry {
    node: String,
    pid: Option<u32>,
    state: String,
    seen_ms_ago: u64,
 }
 #[derive(Serialize)]
 struct ControlSnapshot {
    nodes: Vec<ProcessEntry>,
 }
 #[derive(Default)]
 struct ProcessState {
    pid: Option<u32>,
    state: String,
    seen: Option<Instant>,
 }
 #[derive(Default)]
 pub struct DemoControlView {
    processes: Mutex<BTreeMap<String, ProcessState>>,
 }
 fn is_lifecycle_channel(channel: &str) -> bool {
    channel.starts_with("proc.") && channel.ends_with(".lifecycle")
 }
 impl DashboardView for DemoControlView {
    fn id(&self) -> &'static str {
        "demo-control"
    }
    fn title(&self) -> &'static str {
        "Fleet Control"
    }
    fn path(&self) -> &'static str {
        "demo-control"
    }
    fn channels(&self) -> &'static [&'static str] {
        &[]
    }
    fn ingest(&self, _stream: &StreamId, _frame: &Frame, event: &FrameEvent) {
        if !is_lifecycle_channel(&event.channel) && event.channel != "node.status" {
            return;
        }
        let Ok(payload) = serde_json::from_slice::<Value>(&event.payload) else {
            return;
        };
        let node = event.stream.node.clone();
        let mut processes = self.processes.lock();
        let entry = processes.entry(node).or_default();
        entry.seen = Some(Instant::now());
        if let Some(pid) = payload.get("pid").and_then(Value::as_u64) {
            entry.pid = Some(pid as u32);
        }
        if let Some(kind) = payload.get("event").and_then(Value::as_str) {
            entry.state = match kind {
                "started" => "running".to_owned(),
                "exited" => "exited".to_owned(),
                "spawn_failed" | "error" => "failed".to_owned(),
                other => other.to_owned(),
            };
        }
    }
    fn snapshot_json(&self) -> Value {
        let now = Instant::now();
        let nodes: Vec<ProcessEntry> = self
            .processes
            .lock()
            .iter()
            .map(|(node, state)| ProcessEntry {
                node: node.clone(),
                pid: state.pid,
                state: state.state.clone(),
                seen_ms_ago: state
                    .seen
                    .map(|seen| now.duration_since(seen).as_millis() as u64)
                    .unwrap_or(u64::MAX),
            })
            .collect();
        serde_json::to_value(ControlSnapshot { nodes })
            .unwrap_or_else(|_| serde_json::json!({ "nodes": [] }))
    }
    fn html(&self) -> Option<&'static str> {
        Some(include_str!("demo_control_page.html"))
    }
 }
--- a/crates/dashboard/src/demo_control_page.html
+++ b/crates/dashboard/src/demo_control_page.html
@ -0,0 +1,148 @@
 <!doctype html>
 <html lang="en">
 <head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Fleet Control</title>
  <style>
    :root {
      color-scheme: dark;
      font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
      background: #0f172a;
      color: #e2e8f0;
    }
    * { box-sizing: border-box; }
    body { margin: 0; padding: 20px; }
    .nav { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 18px; padding: 8px; background: #111827; border: 1px solid #334155; border-radius: 12px; }
    .nav a { padding: 7px 10px; color: #cbd5e1; border: 1px solid transparent; border-radius: 8px; text-decoration: none; }
    .nav a:hover { color: #f8fafc; background: #1e293b; }
    header { display: flex; align-items: center; justify-content: space-between; flex-wrap: wrap; gap: 14px; margin-bottom: 18px; }
    h1 { font-size: 24px; margin: 0; }
    .muted { color: #94a3b8; }
    .panel { background: #1e293b; border: 1px solid #334155; border-radius: 14px; padding: 16px; margin-bottom: 18px; }
    table { width: 100%; border-collapse: collapse; }
    th, td { text-align: left; padding: 8px 10px; border-bottom: 1px solid #334155; }
    th { color: #94a3b8; font-weight: 600; font-size: 13px; text-transform: uppercase; letter-spacing: 0.04em; }
    .pill { display: inline-flex; padding: 3px 8px; border-radius: 999px; border: 1px solid #334155; font-size: 12px; }
    .running { color: #34d399; }
    .exited, .failed { color: #f87171; }
    button { font: inherit; border-radius: 8px; border: 1px solid #475569; background: #334155; color: #e2e8f0; padding: 6px 12px; cursor: pointer; }
    button.danger { background: #7f1d1d; border-color: #b91c1c; color: #fee2e2; }
    button:disabled { opacity: 0.5; cursor: not-allowed; }
    #status { margin-left: 12px; font-size: 13px; }
  </style>
 </head>
 <body>
  <nav class="nav">
    <a href="/">dashboard</a>
    <a href="/view/fleet">fleet</a>
    <a href="/view/demo-control" data-active="true">fleet control</a>
  </nav>
  <header>
    <h1>Fleet Control</h1>
    <div>
      <label class="muted" for="provision-count">nodes</label>
      <input id="provision-count" type="number" min="1" max="8" value="1" style="width: 3.5em; background:#0f172a; color:#e2e8f0; border:1px solid #334155; border-radius:6px; padding:5px 8px;">
      <button id="provision">+ provision</button>
      <button id="remove">− remove</button>
      <span id="status" class="muted"></span>
    </div>
  </header>
  <section class="panel">
    <table>
      <thead>
        <tr><th>node</th><th>pid</th><th>state</th><th>seen</th><th></th></tr>
      </thead>
      <tbody id="rows"></tbody>
    </table>
    <p class="muted" id="empty" hidden>No provisioned processes observed yet.</p>
  </section>
  <script>
    'use strict';
    const API_URL = '/api/view/demo-control';
    const rows = document.getElementById('rows');
    const empty = document.getElementById('empty');
    const status = document.getElementById('status');
    function setStatus(text, isError) {
      status.textContent = text;
      status.className = isError ? 'failed' : 'muted';
      if (text) setTimeout(() => { if (status.textContent === text) status.textContent = ''; }, 4000);
    }
    function escapeHtml(value) {
      return String(value).replace(/[&<>"']/g, (ch) => ({
        '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;'
      }[ch]));
    }
    function fmtSeen(msAgo) {
      if (msAgo < 1000) return 'just now';
      if (msAgo < 60_000) return Math.round(msAgo / 1000) + 's ago';
      return Math.round(msAgo / 60_000) + 'm ago';
    }
    function render(snapshot) {
      const nodes = (snapshot && snapshot.nodes) || [];
      empty.hidden = nodes.length > 0;
      rows.innerHTML = nodes.map((entry) => {
        const stateClass = entry.state === 'running' ? 'running' : entry.state;
        return `<tr>
          <td>${escapeHtml(entry.node)}</td>
          <td>${entry.pid == null ? '—' : escapeHtml(entry.pid)}</td>
          <td><span class="pill ${stateClass}">${escapeHtml(entry.state)}</span></td>
          <td class="muted">${escapeHtml(fmtSeen(entry.seen_ms_ago))}</td>
          <td><button class="danger" data-node="${escapeHtml(entry.node)}" ${entry.state !== 'running' ? 'disabled' : ''}>kill</button></td>
        </tr>`;
      }).join('');
      for (const button of rows.querySelectorAll('button[data-node]')) {
        button.addEventListener('click', () => {
          button.disabled = true;
          fetch('/control/kill', {
            method: 'POST',
            headers: { 'content-type': 'application/json' },
            body: JSON.stringify({ Kill: { node: button.dataset.node } })
          }).then((response) => {
            setStatus(response.ok ? 'kill issued' : 'kill failed: HTTP ' + response.status, !response.ok);
          }).catch((error) => setStatus('kill failed: ' + error, true));
        });
      }
    }
    function nodeCount() {
      return Math.max(1, Math.min(8, Number(document.getElementById('provision-count').value) || 1));
    }
    document.getElementById('provision').addEventListener('click', () => {
      const count = nodeCount();
      fetch('/control/provision', {
        method: 'POST',
        headers: { 'content-type': 'application/json' },
        body: JSON.stringify({ Provision: { count } })
      }).then((response) => {
        setStatus(response.ok ? 'provision issued' : 'provision failed: HTTP ' + response.status, !response.ok);
      }).catch((error) => setStatus('provision failed: ' + error, true));
    });
    document.getElementById('remove').addEventListener('click', () => {
      const count = nodeCount();
      fetch('/control/remove', {
        method: 'POST',
        headers: { 'content-type': 'application/json' },
        body: JSON.stringify({ Remove: { count } })
      }).then((response) => {
        setStatus(response.ok ? 'remove issued' : 'remove failed: HTTP ' + response.status, !response.ok);
      }).catch((error) => setStatus('remove failed: ' + error, true));
    });
    async function poll() {
      try {
        const response = await fetch(API_URL);
        if (response.ok) render(await response.json());
      } catch { /* transient */ }
    }
    poll();
    setInterval(poll, 1000);
  </script>
 </body>
 </html>
--- a/crates/dashboard/src/hardware_page.html
+++ b/crates/dashboard/src/hardware_page.html
@ -664,6 +664,7 @@
        </div>
        <div class="fleet-card-meta">
          <span class="muted">Seen ${escapeHtml(fmtDuration(node.last_seen_ms_ago))} ago</span>
          ${node.process ? `<span class="pill ${node.process.state === 'running' ? 'ok' : 'warn'}">pid ${escapeHtml(node.process.pid ?? '—')} · ${escapeHtml(node.process.state)}</span>` : ''}
          <span class="${errors.length ? 'bad' : 'ok'}">${escapeHtml(fmtNumber(errors.length))} error${errors.length === 1 ? '' : 's'}</span>
        </div>
        <div class="fleet-activity">
--- a/crates/dashboard/src/hardware_view.rs
+++ b/crates/dashboard/src/hardware_view.rs
@ -40,9 +40,17 @@ struct NodeHardwareState {
    cpu: Option<HostCpuSample>,
    gpu: Option<HostGpuSample>,
    net: Option<NetSnapshot>,
    process: Option<ProcessSnapshot>,
    history: VecDeque<HardwareHistoryState>,
 }
 /// Managed-process state folded from `proc.<label>.lifecycle` frames.
 #[derive(Clone, Serialize)]
 struct ProcessSnapshot {
    pid: Option<u32>,
    state: String,
 }
 impl NodeHardwareState {
    fn new(stream: StreamEvent, now: Instant) -> Self {
        Self {
@ -52,6 +60,7 @@ impl NodeHardwareState {
            cpu: None,
            gpu: None,
            net: None,
            process: None,
            history: VecDeque::with_capacity(HISTORY_CAP),
        }
    }
@ -83,7 +92,19 @@ impl NodeHardwareState {
                }
                Err(error) => self.store_decode_error(HOST_NET_CHANNEL, error),
            },
-            _ => {}
+            _ => {
                if channel.starts_with("proc.") && channel.ends_with(".lifecycle") {
                    self.process = decode_process_snapshot(payload);
                    // Lifecycle frames also prove liveness.
                    self.last_seen = now;
                } else if channel == "node.status" {
                    // Liveness heartbeat from the supervisor.
                    self.last_seen = now;
                    if let Some(status) = decode_process_snapshot(payload) {
                        self.process.get_or_insert(status);
                    }
                }
            }
        }
    }
@ -108,13 +129,12 @@ impl NodeHardwareState {
                    gpu_max_percent =
                        Some(gpu_max_percent.map_or(percent, |current: u64| current.max(percent)));
                }
-                gpu_memory_used_mib =
+                gpu_memory_used_mib = gpu_memory_used_mib
-                    gpu_memory_used_mib.saturating_add(device.memory_used_mib.unwrap_or_default());
+                    .saturating_add(device.memory_used_mib.unwrap_or_default());
                gpu_memory_total_mib = gpu_memory_total_mib
                    .saturating_add(device.memory_total_mib.unwrap_or_default());
            }
        }
        let (net_rx_bps, net_tx_bps) = self.net.as_ref().map_or((0.0, 0.0), |net| {
            net.interfaces.iter().fold((0.0, 0.0), |totals, interface| {
                (
@ -187,6 +207,28 @@ impl NodeHardwareState {
    }
 }
 /// Decode a `swactor_process.lifecycle.v1` payload into fleet-card state.
 fn decode_process_snapshot(payload: &[u8]) -> Option<ProcessSnapshot> {
    let value: serde_json::Value = serde_json::from_slice(payload).ok()?;
    let pid = value
        .get("pid")
        .and_then(serde_json::Value::as_u64)
        .map(|pid| pid as u32);
    let state = match value.get("event").and_then(serde_json::Value::as_str) {
        Some("started") => "running",
        Some("exited") => "exited",
        Some("spawn_failed") | Some("error") => "failed",
        Some(other) => other,
        None => match value.get("alive").and_then(serde_json::Value::as_bool) {
            Some(true) => "running",
            Some(false) => "exited",
            None => return None,
        },
    }
    .to_owned();
    Some(ProcessSnapshot { pid, state })
 }
 #[derive(Clone, Serialize)]
 struct NetSnapshot {
    seq: u64,
@ -340,6 +382,7 @@ struct NodeHardwareSnapshot {
    cpu: Option<CpuSnapshot>,
    gpu: Option<GpuSnapshot>,
    net: Option<NetSnapshot>,
    process: Option<ProcessSnapshot>,
    history: Vec<HardwareHistorySnapshot>,
 }
@ -486,6 +529,7 @@ fn node_snapshot(node: &NodeHardwareState, now: Instant) -> NodeHardwareSnapshot
        cpu: node.cpu.as_ref().map(CpuSnapshot::from),
        gpu: node.gpu.as_ref().map(GpuSnapshot::from),
        net: node.net.clone(),
        process: node.process.clone(),
        history: node
            .history
            .iter()
--- a/crates/dashboard/src/lib.rs
+++ b/crates/dashboard/src/lib.rs
@ -1,3 +1,7 @@
 #[cfg(feature = "demo-control")]
 pub mod control;
 #[cfg(feature = "demo-control")]
 mod demo_control;
 mod hardware_view;
 mod live_explorer;
 mod server;
@ -7,8 +11,8 @@ pub mod view;
 use std::sync::Arc;
 use telemetry::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
 use serde::Serialize;
 use telemetry::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
 use tokio::sync::broadcast;
 use crate::store::DashboardStore;
@ -79,6 +83,7 @@ impl FrameEvent {
 /// The handle's data path publishes observed telemetry frames to HTTP clients
 /// and registered views. It does not send signals back to producers or mutate
 /// runtime state.
 #[derive(Clone)]
 pub struct DashboardHandle {
    port: u16,
    frames: broadcast::Sender<FrameEvent>,
@ -96,6 +101,8 @@ impl DashboardHandle {
        let views = Arc::new(ViewRegistry::new());
        views.register(Arc::new(live_explorer::LiveTelemetryExplorer::default()));
        views.register(Arc::new(hardware_view::HardwareDashboardView::default()));
        #[cfg(feature = "demo-control")]
        views.register(Arc::new(demo_control::DemoControlView::default()));
        views.register(swactor::worker_view());
        views.register(swactor::actor_overview_view());
        views.register(swactor::actor_dossier_view());
--- a/crates/dashboard/src/server.rs
+++ b/crates/dashboard/src/server.rs
@ -35,14 +35,67 @@ pub(crate) async fn run_server(state: AppState, port: u16) {
 }
 fn router(state: AppState) -> Router {
-    Router::new()
+    let router = Router::new()
        .route("/", get(root_page))
        .route("/events", get(frame_stream))
        .route("/api/frames", get(recent_frames))
        .route("/api/views", get(views_json))
        .route("/api/view/{*path}", get(view_snapshot))
-        .route("/view/{*path}", get(view_page))
+        .route("/view/{*path}", get(view_page));
-        .with_state(state)
+    #[cfg(feature = "demo-control")]
    let router = router
        .route("/control/kill", axum::routing::post(control_kill))
        .route("/control/provision", axum::routing::post(control_provision))
        .route("/control/remove", axum::routing::post(control_remove));
    router.with_state(state)
 }
 #[cfg(feature = "demo-control")]
 async fn control_kill(
    Json(command): Json<crate::control::ControlCommand>,
 ) -> impl IntoResponse {
    match command {
        crate::control::ControlCommand::Kill { .. } => {
            if crate::control::dispatch(command) {
                StatusCode::ACCEPTED
            } else {
                StatusCode::SERVICE_UNAVAILABLE
            }
        }
        _ => StatusCode::UNPROCESSABLE_ENTITY,
    }
 }
 #[cfg(feature = "demo-control")]
 async fn control_provision(
    Json(command): Json<crate::control::ControlCommand>,
 ) -> impl IntoResponse {
    match command {
        crate::control::ControlCommand::Provision { .. } => {
            if crate::control::dispatch(command) {
                StatusCode::ACCEPTED
            } else {
                StatusCode::SERVICE_UNAVAILABLE
            }
        }
        _ => StatusCode::UNPROCESSABLE_ENTITY,
    }
 }
 #[cfg(feature = "demo-control")]
 async fn control_remove(
    Json(command): Json<crate::control::ControlCommand>,
 ) -> impl IntoResponse {
    match command {
        crate::control::ControlCommand::Remove { .. } => {
            if crate::control::dispatch(command) {
                StatusCode::ACCEPTED
            } else {
                StatusCode::SERVICE_UNAVAILABLE
            }
        }
        _ => StatusCode::UNPROCESSABLE_ENTITY,
    }
 }
 async fn root_page(State(state): State<AppState>) -> Html<String> {
--- a/xtask/Cargo.toml
+++ b/xtask/Cargo.toml
@ -7,6 +7,19 @@ license = "AGPL-3.0-only"
 [dependencies]
 serde_json = "1"
 blake3 = "1"
 swactor = { path = "..", features = ["serde"] }
 swactor-engine = { path = "../crates/engine" }
 swactor-transport = { path = "../crates/transport" }
 swactor-process = { path = "../crates/process" }
 provisioning = { path = "../crates/provisioning" }
 telemetry = { path = "../crates/telemetry" }
 dashboard = { path = "../crates/dashboard", features = ["demo-control"] }
 iroh-driver = { path = "../crates/iroh-driver" }
 distribution = { path = "../crates/distribution" }
 iroh = "0.98"
 tokio.workspace = true
 serde = { version = "1", features = ["derive"] }
 parking_lot = "0.12"
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"
--- a/xtask/src/main.rs
+++ b/xtask/src/main.rs
@ -266,6 +266,10 @@ COMMANDS:
                     Run real cargo myelin-chat acceptance check and write benchmark artifacts.
  myelin-chat-compare <baseline-summary.json> <candidate-summary.json>
                     Compare two benchmark summaries and report comparable deltas.
  provisioning-reconciler-demo [--port n] [--nodes n]
                      Run the visual E2E provisioning reconciler sanity demo
                      (supervisor + dashboard on localhost, node children
                      join over iroh). Ctrl-C tears down.
  check-telemetry-isolation  Verify no frame types appear in control-plane modules.
  test                Run the basic non-binding test barrier: root crate plus each
                      non-binding repository package with `cargo test -p`."
@ -6933,6 +6937,8 @@ fn collect_rs_files(dir: &str, out: &mut Vec<String>) {
    }
 }
 mod provisioning_demo;
 fn main() -> ExitCode {
    let mut args = std::env::args().skip(1);
    match args.next().as_deref() {
@ -6941,6 +6947,9 @@ fn main() -> ExitCode {
        Some("myelin-chat-check") => run_myelin_chat_check(args.collect()),
        Some("myelin-chat-compare") => run_myelin_chat_compare(args.collect()),
        Some("myelin-chat") => run_myelin_chat(args.collect()),
        Some("provisioning-reconciler-demo") => {
            provisioning_demo::run(&args.collect::<Vec<String>>())
        }
        Some("help" | "--help" | "-h") | None => {
            print_usage();
            ExitCode::SUCCESS
--- a/xtask/src/provisioning_demo/control.rs
+++ b/xtask/src/provisioning_demo/control.rs
@ -0,0 +1,36 @@
 //! Control plumbing: dashboard control commands → supervisor actor.
 //!
 //! The dashboard (under `demo-control`) dispatches into a std mpsc channel;
 //! an engine task forwards each command as a `SupervisorMsg::Control` message
 //! to the supervisor actor.
 use std::sync::mpsc as std_mpsc;
 use swactor::runtime::ExternalSender;
 use swactor_engine::EngineHandle;
 use crate::provisioning_demo::feed::SupervisorMsg;
 /// Wire the dashboard control channel to the supervisor actor.
 pub fn install(
    engine: &EngineHandle,
    sender: ExternalSender,
    supervisor: std::sync::Arc<std::sync::OnceLock<swactor::actor::ActorAddress>>,
 ) {
    let (control_tx, control_rx) = std_mpsc::channel::<dashboard::control::ControlCommand>();
    dashboard::control::set_control_sender(control_tx);
    let engine = engine.clone();
    engine.spawn(async move {
        loop {
            match control_rx.recv() {
                Ok(command) => {
                    if let Some(addr) = supervisor.get() {
                        let _ = sender.send_to(addr.clone(), SupervisorMsg::Control(command));
                    }
                }
                Err(_) => return,
            }
        }
    });
 }
--- a/xtask/src/provisioning_demo/feed.rs
+++ b/xtask/src/provisioning_demo/feed.rs
@ -0,0 +1,811 @@
 //! The provisioning supervisor: an actor that owns the `ClusterDriver`, the
 //! demo provider, the desired shape, and the 250ms tick.
 //!
 //! Each tick mirrors the production `ClusterReconciler` poll semantics:
 //! drain executor results (closing bootstrap sessions after convergence),
 //! classify due operations, requeue, drive until blocked — then feeds the
 //! real world back in (key-file observations, iroh join checks, child exits),
 //! emits `prov.reconciler.*` telemetry, and drains every telemetry endpoint
 //! into the dashboard.
 use std::collections::BTreeMap;
 use std::time::{Duration, SystemTime};
 use serde_json::json;
 use swactor::actor::{ActorInterface, Ctx};
 use swactor_engine::EngineHandle;
 use swactor_process::{spawn_local_process, ProcessOutputConfig, ProcessSpec};
 use provisioning::executor::{
    BlockingEffectSpawner, BlockingEffectWork, ExecutorOperationStatus,
    IdempotentEffectExecutor,
 };
 use provisioning::node::{
    BootstrapObservation, BootstrapStage, NodeGroupId,
    NodeStage, RoleId, RunId, SwactorId,
 };
 use provisioning::reconciler::{
    ClusterShape, NodeObservation, RetryPolicy, PlannedEffect, OperationOutcome,
 };
 use provisioning::reconciler::{ClusterDriver, EffectExecutor};
 use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer};
 use crate::provisioning_demo::node::read_key_report;
 use crate::provisioning_demo::provider::{
    register_node_channels, unix_ms, DemoBackend, NodeManager, NodeRelayActor,
    NodeTelemetry,
 };
 /// Supervisor telemetry: channels + name resolution for the dashboard path.
 pub struct SupervisorTelemetry {
    pub endpoint: TelemetryEndpoint,
    pub producer: TelemetryProducer,
    names: BTreeMap<telemetry::ChannelId, String>,
 }
 impl SupervisorTelemetry {
    pub fn new(node: &str) -> Self {
        let stream = telemetry::frame::StreamId::new(
            telemetry::frame::NodeId::new(node),
            telemetry::frame::Lifetime(1),
        );
        let endpoint = TelemetryEndpoint::with_descriptor(
            StreamDescriptor {
                stream,
                label: Some("provisioning supervisor".to_owned()),
                origin: telemetry::frame::StreamOrigin::Orchestrator,
            },
            512,
            16,
        );
        let producer = endpoint.producer();
        Self {
            endpoint,
            producer,
            names: BTreeMap::new(),
        }
    }
    pub fn register(&mut self, name: &str) -> telemetry::ChannelId {
        let id = self.endpoint.register_channel(
            name,
            ChannelContent::JsonRecord {
                schema: Some("demo.prov.v1".to_owned()),
            },
        );
        self.names.insert(id, name.to_owned());
        id
    }
 }
 /// Engine-backed spawner for executor blocking work.
 #[derive(Clone)]
 pub struct EngineSpawner {
    engine: EngineHandle,
 }
 impl EngineSpawner {
    pub fn new(engine: EngineHandle) -> Self {
        Self { engine }
    }
 }
 impl BlockingEffectSpawner for EngineSpawner {
    type SpawnError = std::convert::Infallible;
    fn spawn_blocking(&self, work: BlockingEffectWork) -> Result<(), Self::SpawnError> {
        self.engine.spawn_blocking(move || work());
        Ok(())
    }
 }
 /// Wrapper that records every dispatched effect for the command feed.
 struct FeedExecutor<'a> {
    inner: &'a mut IdempotentEffectExecutor<DemoBackend, EngineSpawner>,
    dispatched: Vec<String>,
 }
 impl EffectExecutor for FeedExecutor<'_> {
    type SubmitError = provisioning::executor::ExecutorSubmitError;
    fn submit(&mut self, effect: &PlannedEffect) -> Result<(), Self::SubmitError> {
        use provisioning::node::NodeManagerCommand;
        let command = match &effect.command {
            NodeManagerCommand::CreateLease(_) => "CreateLease".to_owned(),
            NodeManagerCommand::LookupEndpoint(_) => "LookupEndpoint".to_owned(),
            NodeManagerCommand::StartBootstrap(_) => "StartBootstrap".to_owned(),
            NodeManagerCommand::BootstrapConvergenceObserved { .. } => {
                "BootstrapConvergenceObserved".to_owned()
            }
            NodeManagerCommand::CancelBootstrap { .. } => "CancelBootstrap".to_owned(),
            NodeManagerCommand::DestroyLease(_) => "DestroyLease".to_owned(),
        };
        self.dispatched
            .push(format!("{} → {}", effect.node.0, command));
        self.inner.submit(effect)
    }
 }
 #[derive(Clone)]
 pub enum SupervisorMsg {
    Tick,
    Control(dashboard::control::ControlCommand),
    Spawn(crate::provisioning_demo::provider::SpawnNodeRequest),
    /// Drain the cluster: desired → empty, stop every child, flag when done.
    Shutdown {
        drained: std::sync::Arc<std::sync::atomic::AtomicBool>,
    },
 }
 /// Per-node telemetry handle kept while the attempt is live.
 struct NodeStreams {
    telemetry: NodeTelemetry,
    status_channel: telemetry::ChannelId,
 }
 /// The provisioning supervisor actor.
 pub struct SupervisorActor {
    pub driver: ClusterDriver,
    pub executor: IdempotentEffectExecutor<DemoBackend, EngineSpawner>,
    pub manager: NodeManager,
    pub driver_handle: std::sync::Arc<crate::provisioning_demo::DemoDriverHandle>,
    pub telemetry: SupervisorTelemetry,
    pub events_channel: telemetry::ChannelId,
    pub snapshot_channel: telemetry::ChannelId,
    pub sender: swactor::runtime::ExternalSender,
    /// Desired shape slots: singleton groups, one per logical node.
    slots: Vec<String>,
    slot_seq: u64,
    run_id: RunId,
    nodes: BTreeMap<u64, NodeStreams>,
    node_life: u64,
    last_stages: BTreeMap<String, (NodeStage, Option<BootstrapStage>)>,
    pub dashboard: dashboard::DashboardHandle,
    status_tick: u64,
    exe: std::path::PathBuf,
 }
 impl SupervisorActor {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        driver: ClusterDriver,
        executor: IdempotentEffectExecutor<DemoBackend, EngineSpawner>,
        manager: NodeManager,
        driver_handle: std::sync::Arc<crate::provisioning_demo::DemoDriverHandle>,
        mut telemetry: SupervisorTelemetry,
        dashboard: dashboard::DashboardHandle,
        sender: swactor::runtime::ExternalSender,
        initial_slots: Vec<String>,
        run_id: RunId,
        exe: std::path::PathBuf,
    ) -> Self {
        let events_channel = telemetry.register("prov.reconciler.events");
        let snapshot_channel = telemetry.register("prov.reconciler.snapshot");
        Self {
            driver,
            executor,
            manager,
            driver_handle,
            telemetry,
            events_channel,
            snapshot_channel,
            sender,
            nodes: BTreeMap::new(),
            run_id,
            slot_seq: initial_slots.len() as u64,
            slots: initial_slots,
            node_life: 0,
            last_stages: BTreeMap::new(),
            dashboard,
            status_tick: 0,
            exe,
        }
    }
    fn desired_shape(&self, generation: u64) -> ClusterShape {
        ClusterShape {
            run_id: self.run_id.clone(),
            generation,
            groups: self.slots.iter().map(|slot| slot_group(slot)).collect(),
        }
    }
    fn emit_event(&mut self, kind: &str, node: &str, detail: String) {
        let payload = json!({
            "at_ms": unix_ms(SystemTime::now()),
            "kind": kind,
            "node": node,
            "detail": detail,
        });
        let bytes = serde_json::to_vec(&payload).expect("event serializes");
        self.telemetry.producer.submit_bytes(self.events_channel, bytes);
    }
    /// Handle a spawn request from the provider (runs in actor context).
    fn spawn_node(&mut self, ctx: &Ctx, request: crate::provisioning_demo::provider::SpawnNodeRequest) {
        let attempt = request.attempt;
        let relay = match ctx.spawn(NodeRelayActor::new(self.manager.clone(), attempt)) {
            Ok(addr) => addr,
            Err(error) => {
                let _ = request.reply.send(Err(format!("spawn relay actor: {error}")));
                return;
            }
        };
        let spec = ProcessSpec {
            command: self.exe.to_string_lossy().to_string(),
            args: vec![
                "provisioning-reconciler-demo".to_owned(),
                "--demo-node".to_owned(),
                self.driver_handle.supervisor_addr_json.clone(),
            ],
            env: [
                (
                    "DEMO_NODE_KEY_FILE".to_owned(),
                    request.key_file.to_string_lossy().to_string(),
                ),
                ("DEMO_NODE_ID".to_owned(), request.logical_node.clone()),
            ]
            .into_iter()
            .collect(),
            working_dir: None,
            label: Some(request.logical_node.clone()),
        };
        self.node_life += 1;
        let telemetry = NodeTelemetry::new(&request.logical_node, self.node_life);
        let status_channel = register_node_channels(&telemetry.producer);
        // The lifecycle channel is registered by the process crate with the
        // sanitized label; mirror it for name resolution during drain.
        let output =
            ProcessOutputConfig::telemetry_mirror(relay, telemetry.producer.clone());
        match spawn_local_process(ctx, &self.sender, spec, output) {
            Ok(process_actor) => {
                self.nodes.insert(
                    attempt,
                    NodeStreams {
                        telemetry,
                        status_channel,
                    },
                );
                self.manager.register(
                    crate::provisioning_demo::provider::NodeRuntime {
                        attempt,
                        logical_node: request.logical_node.clone(),
                        process_actor,
                        key_file: request.key_file.clone(),
                        pid: None,
                        exited: None,
                        spawn_failed: None,
                    },
                );
                let _ = request.reply.send(Ok(
                    crate::provisioning_demo::provider::NodeRuntime {
                        attempt,
                        logical_node: request.logical_node,
                        process_actor,
                        key_file: request.key_file,
                        pid: None,
                        exited: None,
                        spawn_failed: None,
                    },
                ));
            }
            Err(error) => {
                let _ = request.reply.send(Err(format!("spawn process actor: {error}")));
            }
        }
    }
    /// Feed real-world observations into the driver.
    fn observe_world(&mut self, now: SystemTime) {
        let node_ids: Vec<String> = self
            .driver
            .state()
            .nodes
            .keys()
            .map(|id| id.0.clone())
            .collect();
        for node_id in node_ids {
            let Some(managed) = self.driver.state().nodes.get(&provisioning::node::LogicalNodeId(node_id.clone())) else {
                continue;
            };
            let attempt = managed.attempt;
            let stage = managed.record.stage;
            let active_bootstrap = managed.active_bootstrap;
            let runtime = match self.manager.get(attempt.0) {
                Some(runtime) => runtime,
                None => continue,
            };
            // Child exit or spawn failure: fail the attempt while it is
            // still bootstrapping so the reconciler retries with a fresh
            // lease instead of wedging at SshReady forever.
            if runtime.exited.is_some() || runtime.spawn_failed.is_some() {
                if stage != NodeStage::Failed && active_bootstrap.is_some() {
                    let reason = if let Some(status) = &runtime.exited {
                        format!("node process exited: {status:?}")
                    } else {
                        format!(
                            "node process failed to spawn: {}",
                            runtime.spawn_failed.as_deref().unwrap_or("unknown")
                        )
                    };
                    self.emit_event("observation", &node_id, reason.clone());
                    self.driver.apply_observation(
                        &provisioning::node::LogicalNodeId(node_id.clone()),
                        attempt,
                        NodeObservation::BootstrapFailed {
                            session_id: active_bootstrap.expect("checked above"),
                            reason,
                        },
                        now,
                    );
                }
                continue;
            }
            // Bootstrap progression from the key file and iroh join state.
            if let Some(session_id) = active_bootstrap {
                let report = read_key_report(&runtime.key_file);
                let mut stage_seen = BootstrapStage::SshReady;
                if let Some(report) = &report {
                    let connected = report
                        .node_hex
                        .parse_key()
                        .is_some_and(|key| self.driver_handle.has_active_connection(key));
                    if connected {
                        let heartbeat_age = unix_ms(now).saturating_sub(report.last_seen_ms);
                        self.emit_event(
                            "observation",
                            &node_id,
                            format!(
                                "swactor join confirmed (key {}…, heartbeat {}ms old)",
                                &report.node_hex[..8.min(report.node_hex.len())],
                                heartbeat_age
                            ),
                        );
                        let swactor_id = SwactorId(report.node_hex.clone());
                        self.driver.apply_observation(
                            &provisioning::node::LogicalNodeId(node_id.clone()),
                            attempt,
                            NodeObservation::SwactorJoined {
                                session_id,
                                swactor_id,
                            },
                            now,
                        );
                        continue;
                    }
                    stage_seen = BootstrapStage::WaitingForSwactorJoin;
                }
                self.driver.apply_observation(
                    &provisioning::node::LogicalNodeId(node_id.clone()),
                    attempt,
                    NodeObservation::BootstrapObserved {
                        session_id,
                        observation: BootstrapObservation::stage(stage_seen),
                    },
                    now,
                );
            }
        }
    }
    /// Replace ready nodes whose child has died (shape-native replacement:
    /// retire the dead singleton slot and add a fresh one).
    fn replace_dead_ready_nodes(&mut self, now: SystemTime) {
        let state = self.driver.state().clone();
        let mut replacements: Vec<(String, String)> = Vec::new();
        for (id, managed) in &state.nodes {
            if managed.record.ready && managed.intent == provisioning::reconciler::NodeIntent::Active
            {
                let Some(runtime) = self.manager.get(managed.attempt.0) else {
                    continue;
                };
                if runtime.exited.is_some() {
                    self.slot_seq += 1;
                    let fresh = format!("node-{}", self.slot_seq);
                    replacements.push((id.0.clone(), fresh));
                }
            }
        }
        if replacements.is_empty() {
            return;
        }
        for (dead, fresh) in &replacements {
            self.emit_event("control", dead, format!("runtime death; replacing as {fresh}"));
            self.slots.retain(|slot| slot_group_id(slot) != *dead);
            self.slots.push(fresh.clone());
        }
        let generation = self.driver.desired().generation.saturating_add(1);
        if let Err(error) = self.driver.update_desired(self.desired_shape(generation)) {
            eprintln!("demo: replace update_desired failed: {error}");
        }
        let _ = now;
    }
    /// One poll pass mirroring the production reconciler loop.
    fn poll(&mut self, now: SystemTime) {
        // 1. Drain executor results; close bootstrap after convergence.
        for result in self.executor.drain_results() {
            let close = matches!(
                result.result,
                Ok(OperationOutcome::BootstrapConvergenceAccepted)
            )
            .then_some((result.node.clone(), result.operation.attempt));
            let detail = match &result.result {
                Ok(outcome) => format!("{outcome:?}"),
                Err(error) => format!("failed: {}", error.reason),
            };
            let node = result.node.0.clone();
            let applied = self.driver.apply_executor_result(result, now);
            self.emit_event(
                "result",
                &node,
                format!("{detail}{}", if applied { "" } else { " (stale)" }),
            );
            if applied && let Some((node, attempt)) = close {
                self.driver.apply_observation(
                    &node.clone(),
                    attempt,
                    NodeObservation::BootstrapClosed {
                        session_id: provisioning::node::BootstrapSessionId(attempt.0),
                    },
                    now,
                );
            }
        }
        // 2. Classify due operations.
        for operation in self.driver.pending_operations_due(now) {
            match self.executor.operation_status(operation.operation) {
                ExecutorOperationStatus::Unknown => {
                    self.driver.operation_timed_out(
                        &operation,
                        "executor lost pending operation",
                        now,
                    );
                }
                ExecutorOperationStatus::InFlight => {
                    self.executor.expire(
                        operation.operation,
                        "executor operation timed out with an ambiguous outcome",
                    );
                }
                ExecutorOperationStatus::Completed => {}
            }
        }
        // 3. Drive the state machine, recording dispatched commands.
        self.driver.trigger_if_due(now);
        let mut feed = FeedExecutor {
            inner: &mut self.executor,
            dispatched: Vec::new(),
        };
        match self.driver.drive_until_blocked(now, &mut feed) {
            Ok(_) => {
                for line in feed.dispatched {
                    self.emit_event("command", "", line);
                }
            }
            Err(error) => {
                self.emit_event("error", "", format!("{error}"));
            }
        }
    }
    /// Per-node liveness heartbeat on the `node.status` channel so fleet
    /// cards and the control view stay live between lifecycle transitions.
    fn emit_node_status(&mut self, now: SystemTime) {
        self.status_tick = self.status_tick.wrapping_add(1);
        if self.status_tick % 4 != 0 {
            return; // 250ms ticks → heartbeat every second
        }
        let attempts: Vec<u64> = self.nodes.keys().copied().collect();
        for attempt in attempts {
            let Some(streams) = self.nodes.get(&attempt) else {
                continue;
            };
            let Some(runtime) = self.manager.get(attempt) else {
                continue;
            };
            let report = read_key_report(&runtime.key_file);
            let heartbeat_ms_ago = report
                .as_ref()
                .map(|r| unix_ms(now).saturating_sub(r.last_seen_ms))
                .unwrap_or(u64::MAX);
            let payload = json!({
                "at_ms": unix_ms(now),
                "node": runtime.logical_node,
                "alive": runtime.exited.is_none(),
                "pid": runtime.pid,
                "heartbeat_ms_ago": heartbeat_ms_ago,
            });
            let bytes = serde_json::to_vec(&payload).expect("status serializes");
            streams
                .telemetry
                .producer
                .submit_bytes(streams.status_channel, bytes);
        }
    }
    /// Emit stage transitions and the snapshot.
    fn emit_feed(&mut self, now: SystemTime) {
        let state = self.driver.state().clone();
        let mut nodes_json = Vec::new();
        let mut ready_count = 0_u64;
        for (id, managed) in &state.nodes {
            let bootstrap = managed
                .record
                .bootstrap
                .as_ref()
                .map(|facts| facts.last_stage);
            let current = (managed.record.stage, bootstrap);
            let fmt_boot = |stage: Option<BootstrapStage>| {
                stage
                    .map(|stage| format!("{stage:?}"))
                    .unwrap_or_else(|| "-".to_owned())
            };
            if let Some(previous) = self.last_stages.get(&id.0) {
                if previous.0 != current.0 || previous.1 != current.1 {
                    self.emit_event(
                        "transition",
                        &id.0,
                        format!(
                            "{:?} ({}) → {:?} ({})",
                            previous.0,
                            fmt_boot(previous.1),
                            current.0,
                            fmt_boot(current.1),
                        ),
                    );
                }
            } else if managed.record.stage != NodeStage::New {
                self.emit_event(
                    "transition",
                    &id.0,
                    format!("New → {:?} ({})", current.0, fmt_boot(current.1)),
                );
            }
            self.last_stages.insert(id.0.clone(), current);
            if managed.record.ready {
                ready_count += 1;
            }
            let runtime_pid = self.manager.get(managed.attempt.0).and_then(|r| r.pid);
            nodes_json.push(json!({
                "id": id.0,
                "intent": format!("{:?}", managed.intent),
                "stage": format!("{:?}", managed.record.stage),
                "bootstrap": fmt_boot(current.1),
                "attempt": managed.attempt.0,
                "ready": managed.record.ready,
                "pid": runtime_pid,
                "failure": managed.record.failed_reason,
            }));
        }
        let snapshot = json!({
            "at_ms": unix_ms(now),
            "desired": self.slots.len(),
            "ready": ready_count,
            "generation": self.driver.desired().generation,
            "converged": self.driver.is_converged(),
            "nodes": nodes_json,
        });
        let bytes = serde_json::to_vec(&snapshot).expect("snapshot serializes");
        self.telemetry
            .producer
            .submit_bytes(self.snapshot_channel, bytes);
    }
    /// Drain every telemetry endpoint into the dashboard.
    fn flush_telemetry(&mut self) {
        let supervisor_stream = self.telemetry.endpoint.stream_id().clone();
        for frame in self.telemetry.endpoint.mux().drain() {
            let channel = self
                .telemetry
                .names
                .get(&frame.channel)
                .cloned()
                .unwrap_or_else(|| format!("channel#{}", frame.channel.0));
            publish_frame(&self.dashboard, &supervisor_stream, &channel, &frame);
        }
        let attempts: Vec<u64> = self.nodes.keys().copied().collect();
        for attempt in attempts {
            let Some(streams) = self.nodes.get(&attempt) else {
                continue;
            };
            let stream = streams.telemetry.endpoint.stream_id().clone();
            let catalog = streams.telemetry.endpoint.catalog_snapshot();
            for frame in streams.telemetry.endpoint.mux().drain() {
                let channel = catalog
                    .channels
                    .get(&telemetry::frame::ChannelRef {
                        stream: stream.clone(),
                        channel: frame.channel,
                    })
                    .map(|descriptor| descriptor.name.clone())
                    .unwrap_or_else(|| format!("channel#{}", frame.channel.0));
                publish_frame(&self.dashboard, &stream, &channel, &frame);
            }
        }
    }
 }
 fn publish_frame(
    dashboard: &dashboard::DashboardHandle,
    stream: &telemetry::frame::StreamId,
    channel: &str,
    frame: &telemetry::frame::Frame,
 ) {
    dashboard.publish(dashboard::FrameEvent {
        stream: dashboard::StreamEvent {
            node: stream.node.as_str().to_string(),
            life: stream.life.0,
        },
        channel: channel.to_owned(),
        position: frame.position.0,
        payload: frame.payload.clone(),
    });
 }
 impl ActorInterface for SupervisorActor {
    type Incoming = SupervisorMsg;
    type Response = ();
    fn handle(&mut self, ctx: &Ctx, msg: SupervisorMsg) {
        match msg {
            SupervisorMsg::Tick => {
                let now = SystemTime::now();
                self.poll(now);
                self.observe_world(now);
                self.replace_dead_ready_nodes(now);
                self.poll(now);
                self.emit_node_status(now);
                self.emit_feed(now);
                self.flush_telemetry();
            }
            SupervisorMsg::Control(command) => self.handle_control(command),
            SupervisorMsg::Spawn(request) => self.spawn_node(ctx, request),
            SupervisorMsg::Shutdown { drained } => {
                self.slots.clear();
                let generation = self.driver.desired().generation.saturating_add(1);
                if let Err(error) = self.driver.update_desired(self.desired_shape(generation)) {
                    eprintln!("demo: shutdown update_desired failed: {error}");
                }
                self.emit_event("control", "", "shutdown: desired → empty".to_owned());
                drained.store(true, std::sync::atomic::Ordering::SeqCst);
            }
        }
    }
 }
 impl SupervisorActor {
    fn handle_control(&mut self, command: dashboard::control::ControlCommand) {
        match command {
            dashboard::control::ControlCommand::Kill { node } => {
                match self.manager.find_by_stream_node(&node) {
                    Some(runtime) => {
                        self.emit_event(
                            "control",
                            &node,
                            format!("kill requested (pid {:?})", runtime.pid),
                        );
                        let _ = swactor_process::send_process_command(
                            &self.sender,
                            runtime.process_actor,
                            swactor_process::ProcessCommand::Stop {
                                kill_after: Some(Duration::ZERO),
                            },
                        );
                    }
                    None => self.emit_event("control", &node, "kill: unknown node".to_owned()),
                }
            }
            dashboard::control::ControlCommand::Remove { count } => {
                let removed = self.slots.len().min(count as usize);
                if removed == 0 {
                    self.emit_event("control", "", "remove: nothing to remove".to_owned());
                    return;
                }
                for _ in 0..removed {
                    self.slots.pop();
                }
                self.emit_event("control", "", format!("remove -{removed}"));
                let generation = self.driver.desired().generation.saturating_add(1);
                if let Err(error) = self.driver.update_desired(self.desired_shape(generation)) {
                    eprintln!("demo: remove update_desired failed: {error}");
                }
            }
            dashboard::control::ControlCommand::Provision { count } => {
                if count == 0 {
                    return;
                }
                self.slot_seq += count as u64;
                let start = self.slot_seq - count as u64 + 1;
                for seq in start..=self.slot_seq {
                    self.slots.push(format!("node-{seq}"));
                }
                self.emit_event("control", "", format!("provision +{count}"));
                let generation = self.driver.desired().generation.saturating_add(1);
                if let Err(error) = self.driver.update_desired(self.desired_shape(generation)) {
                    eprintln!("demo: provision update_desired failed: {error}");
                }
            }
        }
    }
 }
 /// Parse a hex node key into a transport NodeId.
 trait ParseKey {
    fn parse_key(&self) -> Option<swactor_transport::NodeId>;
 }
 impl ParseKey for String {
    fn parse_key(&self) -> Option<swactor_transport::NodeId> {
        let bytes = swactor_transport::hex_decode(self)?;
        let array: [u8; 32] = bytes.try_into().ok()?;
        Some(swactor_transport::NodeId(array))
    }
 }
 fn slot_group(slot: &str) -> provisioning::node::RunNodeGroupSpec {
    let mut group = demo_group(slot, 1);
    group.group_id = NodeGroupId(slot.to_owned());
    group
 }
 fn slot_group_id(slot: &str) -> String {
    format!("{slot}-0")
 }
 pub fn demo_group(id: &str, count: u32) -> provisioning::node::RunNodeGroupSpec {
    provisioning::node::RunNodeGroupSpec {
        run_id: RunId(1),
        group_id: NodeGroupId(id.to_owned()),
        role: RoleId("worker".to_owned()),
        count,
        provider: provisioning::node::ProviderKind::new("demo"),
        shape: provisioning::node::DesiredNodeShape {
            image: "demo-node".to_owned(),
            disk_gb: 1,
            gpu_name: None,
            min_gpu_ram_mb: None,
            min_down_mbps: None,
            min_up_mbps: None,
            min_reliability: None,
            require_verified: false,
            provider_labels: BTreeMap::new(),
        },
        boot: provisioning::node::BootSpec {
            ssh_user: "demo".to_owned(),
            verify_commands: vec!["true".to_owned()],
            start_swactor_command: "xtask provisioning-reconciler-demo".to_owned(),
            stdout_sources: Vec::new(),
            stderr_sources: Vec::new(),
            env: Vec::new(),
            args: Vec::new(),
            mounts: Vec::new(),
        },
        swarm_join: provisioning::node::SwarmJoinTemplate {
            orch_swactor_addr: "127.0.0.1:1".to_owned(),
            join_token_ref: "demo".to_owned(),
        },
    }
 }
 /// Marker helpers used by tests and the module glue.
 pub fn initial_slots(count: u64) -> Vec<String> {
    (0..count).map(|index| format!("node-{index}")).collect()
 }
 pub fn demo_retry_policy() -> RetryPolicy {
    RetryPolicy {
        initial_delay: Duration::from_millis(500),
        max_delay: Duration::from_secs(2),
        jitter: Duration::ZERO,
        operation_timeout: Duration::from_secs(10),
        endpoint_probe_interval: Duration::from_secs(1),
    }
 }
--- a/xtask/src/provisioning_demo/mod.rs
+++ b/xtask/src/provisioning_demo/mod.rs
@ -0,0 +1,359 @@
 //! `cargo xtask provisioning-reconciler-demo` — a visual, human-checked E2E
 //! sanity scenario for the provisioning reconciler.
 //!
 //! Supervisor role (default): a lightweight orchestrator — swactor engine +
 //! real `ClusterDriver` + demo provider (real node-role children over iroh) +
 //! telemetry + dashboard. Node role (`--demo-node <supervisor-addr-json>`):
 //! re-exec of this binary as a real swactor runtime that joins the
 //! supervisor's iroh endpoint.
 //!
 //! Humans watch `/view/reconciler` (k8s-style current-vs-desired, node
 //! stages, command/result feeds) and the fleet cards (per-node PID/state),
 //! kill nodes from the Fleet Control view or a shell, and watch the
 //! reconciler replace them for real.
 pub mod control;
 pub mod feed;
 pub mod node;
 pub mod provider;
 pub mod view;
 use std::process::ExitCode;
 use std::sync::Arc;
 use std::time::Duration;
 use iroh::RelayMode;
 use swactor::config::RuntimeConfig;
 use swactor::runtime::RuntimeParts;
 use swactor_engine::{Engine, TokioBackend, TokioConfig};
 use distribution::node::DistributedNodeConfig;
 use iroh_driver::{IrohDriver, IrohDriverConfig};
 use provisioning::executor::IdempotentEffectExecutor;
 use provisioning::{ClusterShape, RunId};
 use provisioning::reconciler::ClusterDriver;
 use feed::{
    demo_retry_policy, initial_slots, EngineSpawner, SupervisorActor, SupervisorMsg,
    SupervisorTelemetry,
 };
 use provider::{DemoBackend, DemoProvider, NodeManager};
 /// Supervisor tick period.
 pub const TICK: Duration = Duration::from_millis(250);
 /// Node heartbeat period (node role writes a key-file line this often).
 pub const HEARTBEAT_PERIOD: Duration = Duration::from_secs(2);
 /// Default dashboard port.
 pub const DEFAULT_PORT: u16 = 9871;
 /// Default initial cluster size.
 pub const DEFAULT_NODES: u64 = 3;
 /// Resolve the executable path for node-role re-execs. `current_exe()`
 /// returns a "(deleted)" path or fails once the binary file has been
 /// replaced by a rebuild while this process runs, so fall back through
 /// argv[0] and PATH.
 fn resolve_exe() -> std::path::PathBuf {
    if let Ok(path) = std::env::current_exe() {
        if !path.to_string_lossy().ends_with(" (deleted)") {
            return path;
        }
    }
    if let Some(arg0) = std::env::args_os().next() {
        let candidate = std::path::PathBuf::from(&arg0);
        if candidate.is_absolute() {
            return candidate;
        }
        if let Ok(cwd) = std::env::current_dir() {
            let joined = cwd.join(&candidate);
            if joined.exists() {
                return joined;
            }
        }
        if let Ok(path_var) = std::env::var("PATH") {
            for dir in path_var.split(':') {
                let joined = std::path::PathBuf::from(dir).join(&candidate);
                if joined.exists() {
                    return joined;
                }
            }
        }
    }
    std::path::PathBuf::from("xtask")
 }
 /// Shared handle the supervisor actor uses to check iroh connections.
 pub struct DemoDriverHandle {
    pub supervisor_addr_json: String,
    driver: IrohDriver,
 }
 impl DemoDriverHandle {
    pub fn has_active_connection(&self, node: swactor_transport::NodeId) -> bool {
        self.driver.has_active_connection(&node)
    }
 }
 /// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N]` for the
 /// supervisor, or `--demo-node <supervisor-addr-json>` for node children.
 pub fn run(args: &[String]) -> ExitCode {
    if let Some(index) = args.iter().position(|arg| arg == "--demo-node") {
        let addr = args
            .get(index + 1)
            .map(String::as_str)
            .unwrap_or_default();
        return match node::run_node_role(addr) {
            Ok(()) => ExitCode::SUCCESS,
            Err(error) => {
                eprintln!("demo node: {error}");
                ExitCode::FAILURE
            }
        };
    }
    match run_supervisor(args) {
        Ok(()) => ExitCode::SUCCESS,
        Err(error) => {
            eprintln!("provisioning-reconciler-demo: {error}");
            ExitCode::FAILURE
        }
    }
 }
 fn arg_value(args: &[String], name: &str) -> Option<String> {
    args.iter()
        .position(|arg| arg == name)
        .and_then(|index| args.get(index + 1).cloned())
 }
 fn run_supervisor(args: &[String]) -> Result<(), String> {
    let port: u16 = arg_value(args, "--port")
        .and_then(|value| value.parse().ok())
        .unwrap_or(DEFAULT_PORT);
    let nodes: u64 = arg_value(args, "--nodes")
        .and_then(|value| value.parse().ok())
        .unwrap_or(DEFAULT_NODES);
    // Telemetry endpoint with the runtime stats hook attached before the
    // engine takes the parts.
    let mut telemetry = SupervisorTelemetry::new("supervisor");
    let actors_channel = telemetry.register("runtime.actors");
    let stats_hook = telemetry.producer.stats_hook_on(actors_channel);
    let mut parts = RuntimeParts::new(RuntimeConfig::default());
    parts = parts.with_stats_hook(stats_hook);
    let runtime = parts.runtime().clone();
    let sender = runtime.create_sender();
    let engine = Engine::new(
        parts,
        TokioBackend::new(TokioConfig::default()).map_err(|e| format!("backend: {e}"))?,
    )
    .map_err(|e| format!("engine: {e}"))?;
    // iroh: accept joins from node-role children. The minimal actor bridge
    // exists so the production adapter pump folds accepted connections into
    // the driver cache — `has_active_connection` is our real join signal.
    // Frames that decode against the (empty) codec registry drop harmlessly.
    let mut driver = IrohDriver::with_engine(
        engine.handle(),
        IrohDriverConfig {
            secret_key: None,
            relay_mode: RelayMode::Disabled,
            node: DistributedNodeConfig::default(),
            peer_auth: None,
            additional_alpns: vec![],
        },
    )
    .map_err(|e| format!("iroh driver: {e}"))?;
    let supervisor_addr_json =
        serde_json::to_string(&driver.endpoint_addr()).map_err(|e| format!("addr: {e}"))?;
    {
        use distribution::transport_bridge::{Outbox, RelayMirror, RouteView};
        use swactor_transport::CodecRegistry;
        struct NoopActor;
        impl swactor::actor::ActorInterface for NoopActor {
            type Incoming = ();
            type Response = ();
            fn handle(&mut self, _ctx: &swactor::actor::Ctx, _msg: ()) {}
        }
        let noop = runtime
            .spawn(NoopActor)
            .map_err(|e| format!("spawn noop actor: {e}"))?;
        let relay_mirror: RelayMirror = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
        let route_view: RouteView = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
        let outbox: Outbox = Arc::new(std::sync::Mutex::new(Vec::new()));
        driver.enable_actor_bridge(
            runtime.clone(),
            Arc::new(CodecRegistry::new()),
            std::collections::HashMap::new(),
            noop,
            relay_mirror,
            route_view,
            outbox,
        );
        driver.install_actor_bridge_pump(Duration::from_millis(250));
    }
    let driver_handle = Arc::new(DemoDriverHandle {
        supervisor_addr_json: supervisor_addr_json.clone(),
        driver,
    });
    // Dashboard.
    let dashboard = dashboard::DashboardHandle::new(dashboard::DashboardConfig {
        port,
        ..dashboard::DashboardConfig::default()
    });
    dashboard.register_view(Arc::new(view::ReconcilerDashboardView::default()));
    engine.handle().spawn(dashboard.http_server());
    // Provisioning: driver + plugin + executor.
    let keys_dir = std::env::temp_dir().join(format!(
        "provisioning-demo-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&keys_dir).map_err(|e| format!("keys dir: {e}"))?;
    let manager = NodeManager::new();
    let (spawn_tx, spawn_rx) = std::sync::mpsc::channel::<provider::SpawnNodeRequest>();
    manager.set_spawn_channel(spawn_tx);
    let slots = initial_slots(nodes);
    let shape = ClusterShape {
        run_id: RunId(1),
        generation: 1,
        groups: slots.iter().map(|slot| feed::demo_group(slot, 1)).collect(),
    };
    let cluster_driver =
        ClusterDriver::new(shape, demo_retry_policy()).map_err(|e| format!("driver: {e}"))?;
    let plugin = DemoProvider::new(manager.clone(), keys_dir);
    let spawner = EngineSpawner::new(engine.handle());
    let executor = IdempotentEffectExecutor::new(
        DemoBackend {
            plugin: Arc::new(std::sync::Mutex::new(plugin)),
            manager: manager.clone(),
            sender: sender.clone(),
        },
        spawner,
    );
    let supervisor = SupervisorActor::new(
        cluster_driver,
        executor,
        manager,
        driver_handle,
        telemetry,
        dashboard.clone(),
        sender.clone(),
        slots,
        RunId(1),
        resolve_exe(),
    );
    // Long-lived engine tasks are installed BEFORE the actor spawn: the
    // supervisor's address travels through a shared slot, and the tasks wait
    // for it lazily. (Spawning engine tasks after the actor spawn proved
    // flaky at startup.)
    let supervisor_slot: Arc<std::sync::OnceLock<swactor::actor::ActorAddress>> =
        Arc::new(std::sync::OnceLock::new());
    // Control plane: dashboard → supervisor.
    control::install(&engine.handle(), sender.clone(), supervisor_slot.clone());
    // Spawn-request pump: provider blocking threads → supervisor actor.
    let pump_sender = sender.clone();
    let pump_slot = supervisor_slot.clone();
    engine.handle().spawn_blocking(move || loop {
        match spawn_rx.recv() {
            Ok(request) => {
                if let Some(addr) = pump_slot.get() {
                    let _ = pump_sender.send_to(addr.clone(), SupervisorMsg::Spawn(request));
                }
            }
            Err(_) => return,
        }
    });
    // Tick: engine interval → supervisor actor.
    let tick_sender = sender.clone();
    let tick_engine = engine.handle();
    let interval_engine = tick_engine.clone();
    let tick_slot = supervisor_slot.clone();
    tick_engine.spawn(async move {
        let mut interval = interval_engine.interval(TICK);
        loop {
            (&mut interval).await;
            if let Some(addr) = tick_slot.get() {
                let _ = tick_sender.send_to(addr.clone(), SupervisorMsg::Tick);
            }
        }
    });
    let supervisor_addr = runtime
        .spawn(supervisor)
        .map_err(|e| format!("spawn supervisor actor: {e}"))?;
    supervisor_slot
        .set(supervisor_addr.clone())
        .expect("supervisor address slot set once");
    println!("provisioning-reconciler-demo: dashboard on http://localhost:{port}");
    println!("  /view/reconciler   — reconciler state machine, current vs desired");
    println!("  /view/fleet        — per-node cards (pid, lifecycle)");
    println!("  /view/demo-control — kill / provision controls");
    println!("  Ctrl-C to tear down.");
    // Block until Ctrl-C (synchronous signal flag — the wait must not depend
    // on engine task progression).
    install_sigint_flag();
    while !sigint_requested() {
        std::thread::sleep(Duration::from_millis(100));
    }
    // Teardown: drain the cluster through the real destroy path (desired →
    // empty → BeginDelete → DestroyLease → process actors stop children),
    // pumping ticks until the driver quiesces.
    let drained = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let _ = sender.send_to(
        supervisor_addr.clone(),
        SupervisorMsg::Shutdown {
            drained: drained.clone(),
        },
    );
    let mut settle_ticks = 0_u32;
    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    while std::time::Instant::now() < deadline {
        if drained.load(std::sync::atomic::Ordering::SeqCst) {
            // Give DestroyLease results a few extra ticks to land.
            settle_ticks += 1;
            if settle_ticks >= 8 {
                break;
            }
        }
        std::thread::sleep(TICK);
    }
    // Best-effort drain window closed: children still alive (if any) are
    // killed by the process-tree teardown of the launching console, so exit
    // deterministically rather than risking a wedged teardown path.
    std::process::exit(0);
 }
 static SIGINT_REQUESTED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
 fn sigint_requested() -> bool {
    SIGINT_REQUESTED.load(std::sync::atomic::Ordering::SeqCst)
 }
 #[cfg(target_os = "linux")]
 fn install_sigint_flag() {
    unsafe {
        let handler: extern "C" fn(libc::c_int) = sigint_handler;
        libc::signal(libc::SIGINT, handler as usize);
    }
 }
 #[cfg(target_os = "linux")]
 extern "C" fn sigint_handler(_signal: libc::c_int) {
    SIGINT_REQUESTED.store(true, std::sync::atomic::Ordering::SeqCst);
 }
 #[cfg(not(target_os = "linux"))]
 fn install_sigint_flag() {
    // Non-Linux builds wait for SIGTERM's default disposition instead.
 }
--- a/xtask/src/provisioning_demo/node.rs
+++ b/xtask/src/provisioning_demo/node.rs
@ -0,0 +1,119 @@
 //! Node role: a real swactor runtime that joins the supervisor over iroh.
 //!
 //! Re-exec'd from the xtask binary by the demo provider. Builds an engine +
 //! `IrohDriver` (relay disabled), joins the supervisor's endpoint, and
 //! reports its swactor node key + heartbeats to the key file given in
 //! `DEMO_NODE_KEY_FILE` (the process actor supervises children with stdio
 //! null, so stdout is not observable).
 use std::io::Write;
 use iroh::RelayMode;
 use swactor::config::RuntimeConfig;
 use swactor::runtime::RuntimeParts;
 use swactor_engine::{Engine, TokioBackend, TokioConfig};
 use distribution::node::DistributedNodeConfig;
 use iroh_driver::{IrohDriver, IrohDriverConfig};
 use crate::provisioning_demo::HEARTBEAT_PERIOD;
 /// Key-file line marking the node key (first line): `<ms> KEY <hex>`.
 pub const KEY_LINE_KIND: &str = "KEY";
 /// Key-file heartbeat line: `<ms> ALIVE`.
 pub const ALIVE_LINE_KIND: &str = "ALIVE";
 /// Run the node role. `supervisor_addr_json` is a serde-serialized
 /// `iroh::EndpointAddr` of the supervisor's iroh endpoint.
 pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> {
    let supervisor_addr: iroh::EndpointAddr = serde_json::from_str(supervisor_addr_json)
        .map_err(|error| format!("invalid supervisor endpoint address: {error}"))?;
    let key_file = std::env::var("DEMO_NODE_KEY_FILE")
        .map_err(|_| "DEMO_NODE_KEY_FILE not set".to_owned())?;
    let parts = RuntimeParts::new(RuntimeConfig::default());
    let engine = Engine::new(
        parts,
        TokioBackend::new(TokioConfig::default()).map_err(|error| format!("backend: {error}"))?,
    )
    .map_err(|error| format!("engine: {error}"))?;
    // Bind the driver, then keep it alive for the process lifetime: dropping
    // it closes the endpoint.
    let driver = IrohDriver::with_engine(
        engine.handle(),
        IrohDriverConfig {
            secret_key: None,
            relay_mode: RelayMode::Disabled,
            node: DistributedNodeConfig::default(),
            peer_auth: None,
            additional_alpns: vec![],
        },
    )
    .map_err(|error| format!("iroh driver: {error}"))?;
    let node_hex = swactor_transport::hex_encode(&driver.node_id().0);
    driver.join(&[supervisor_addr]);
    append_key_line(&key_file, KEY_LINE_KIND, &node_hex);
    let heartbeat_file = key_file.clone();
    let interval_handle = engine.handle();
    interval_handle.clone().spawn(async move {
        let mut interval = interval_handle.interval(HEARTBEAT_PERIOD);
        loop {
            (&mut interval).await;
            append_key_line(&heartbeat_file, ALIVE_LINE_KIND, "");
        }
    });
    // The engine owns progression; park this thread until killed. `driver`
    // stays alive until process exit.
    let _keep_driver = driver;
    loop {
        std::thread::park();
    }
 }
 fn append_key_line(path: &str, kind: &str, value: &str) {
    let Ok(mut file) = std::fs::OpenOptions::new().create(true).append(true).open(path) else {
        return;
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0);
    let _ = writeln!(file, "{now} {kind} {value}");
 }
 /// Parsed key-file contents: the child's node key and its last heartbeat.
 pub struct NodeKeyReport {
    pub node_hex: String,
    pub last_seen_ms: u64,
 }
 /// Read a node's key file: its node key and last heartbeat.
 pub fn read_key_report(path: &std::path::Path) -> Option<NodeKeyReport> {
    let contents = std::fs::read_to_string(path).ok()?;
    let mut node_hex: Option<String> = None;
    let mut last_seen_ms = 0_u64;
    for line in contents.lines() {
        let mut parts = line.split_whitespace();
        let Some(stamp) = parts.next().and_then(|value| value.parse::<u64>().ok()) else {
            continue;
        };
        let Some(kind) = parts.next() else {
            continue;
        };
        if kind == KEY_LINE_KIND {
            node_hex = parts.next().map(str::to_owned);
            last_seen_ms = last_seen_ms.max(stamp);
        } else if kind == ALIVE_LINE_KIND {
            last_seen_ms = last_seen_ms.max(stamp);
        }
    }
    Some(NodeKeyReport {
        node_hex: node_hex?,
        last_seen_ms,
    })
 }
--- a/xtask/src/provisioning_demo/provider.rs
+++ b/xtask/src/provisioning_demo/provider.rs
@ -0,0 +1,469 @@
 //! The demo provider seam: real node-role children behind the reconciler's
 //! effect executor.
 //!
 //! `DemoProvider` is the demo's `ProvisionPlugin` — create spawns a
 //! `swactor-process` actor whose child re-execs this binary in node role
 //! (`--demo-node`), stop goes through the real process-actor command path.
 //! `DemoBackend` adapts the plugin to `NodeManagerCommand` effects, mirroring
 //! the conformance kit's adapter semantics (attempt-keyed identity, adoption
 //! on retry).
 //!
 //! Actor spawning needs an actor context, so spawn requests travel over a
 //! channel to the supervisor actor; the plugin call blocks for the reply on
 //! an executor blocking thread.
 //!
 //! Because the process actor supervises children with stdio null, each child
 //! publishes its swactor node key and heartbeats to a per-attempt key file
 //! (`DEMO_NODE_KEY_FILE`); the supervisor reads it to learn the child's iroh
 //! identity and liveness.
 use std::collections::BTreeMap;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
 use swactor::actor::{ActorAddress, ActorInterface, Ctx};
 use swactor::runtime::ExternalSender;
 use swactor_process::{ExitStatus, ProcessOutput};
 use provisioning::executor::{EffectBackend, EffectError};
 use provisioning::node::{
    BootstrapSessionId, CreateLeaseResult, DestroyHandle, LeaseFacts, NodeManagerCommand,
    ProviderKind, ProviderLeaseId, SshEndpoint,
 };
 use provisioning::reconciler::{OperationId, OperationOutcome, PlannedEffect};
 use provisioning::plugin::{NodeProvisionSpec, PluginNodeHandle, PluginSink, ProvisionPlugin};
 use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer};
 /// How long a blocking plugin call waits for the supervisor actor.
 pub const BACKEND_WAIT: Duration = Duration::from_secs(20);
 /// One provisioned node's runtime facts.
 #[derive(Clone)]
 pub struct NodeRuntime {
    pub attempt: u64,
    pub logical_node: String,
    pub process_actor: ActorAddress,
    pub key_file: PathBuf,
    pub pid: Option<u32>,
    pub exited: Option<ExitStatus>,
    pub spawn_failed: Option<String>,
 }
 /// Spawn request from the plugin (blocking thread) to the supervisor actor.
 #[derive(Clone)]
 pub struct SpawnNodeRequest {
    pub attempt: u64,
    pub logical_node: String,
    pub key_file: PathBuf,
    pub reply: std::sync::mpsc::Sender<Result<NodeRuntime, String>>,
 }
 /// Shared node registry + spawn queue: hub between executor blocking threads,
 /// relay actors, and the supervisor actor.
 #[derive(Clone, Default)]
 pub struct NodeManager {
    inner: Arc<Mutex<NodeManagerInner>>,
 }
 #[derive(Default)]
 struct NodeManagerInner {
    nodes: BTreeMap<u64, NodeRuntime>,
    spawn_tx: Option<std::sync::mpsc::Sender<SpawnNodeRequest>>,
 }
 impl NodeManager {
    pub fn new() -> Self {
        Self::default()
    }
    pub fn set_spawn_channel(&self, sender: std::sync::mpsc::Sender<SpawnNodeRequest>) {
        self.inner.lock().expect("node manager").spawn_tx = Some(sender);
    }
    pub fn request_spawn(
        &self,
        attempt: u64,
        logical_node: String,
        key_file: PathBuf,
    ) -> Result<NodeRuntime, String> {
        let (reply_tx, reply_rx) = std::sync::mpsc::channel();
        let request = SpawnNodeRequest {
            attempt,
            logical_node,
            key_file,
            reply: reply_tx,
        };
        {
            let inner = self.inner.lock().expect("node manager");
            let sender = inner
                .spawn_tx
                .as_ref()
                .ok_or_else(|| "supervisor spawn channel not installed".to_owned())?;
            sender
                .send(request)
                .map_err(|_| "supervisor actor gone".to_owned())?;
        }
        reply_rx
            .recv_timeout(BACKEND_WAIT)
            .map_err(|_| "timed out waiting for node spawn".to_owned())?
    }
    pub fn register(&self, runtime: NodeRuntime) {
        self.inner
            .lock()
            .expect("node manager")
            .nodes
            .insert(runtime.attempt, runtime);
    }
    pub fn get(&self, attempt: u64) -> Option<NodeRuntime> {
        self.inner
            .lock()
            .expect("node manager")
            .nodes
            .get(&attempt)
            .cloned()
    }
    pub fn find_by_stream_node(&self, node: &str) -> Option<NodeRuntime> {
        self.inner
            .lock()
            .expect("node manager")
            .nodes
            .values()
            .find(|runtime| runtime.logical_node == node)
            .cloned()
    }
    fn update(&self, attempt: u64, mutate: impl FnOnce(&mut NodeRuntime)) {
        if let Some(runtime) = self
            .inner
            .lock()
            .expect("node manager")
            .nodes
            .get_mut(&attempt)
        {
            mutate(runtime);
        }
    }
    pub fn set_pid(&self, attempt: u64, pid: u32) {
        self.update(attempt, |runtime| runtime.pid = Some(pid));
    }
    pub fn set_exited(&self, attempt: u64, status: ExitStatus) {
        self.update(attempt, |runtime| runtime.exited = Some(status));
    }
    pub fn set_spawn_failed(&self, attempt: u64, reason: String) {
        self.update(attempt, |runtime| runtime.spawn_failed = Some(reason));
    }
    pub fn remove(&self, attempt: u64) {
        self.inner.lock().expect("node manager").nodes.remove(&attempt);
    }
 }
 /// The demo `ProvisionPlugin`: resources are node-role child processes.
 pub struct DemoProvider {
    manager: NodeManager,
    keys_dir: PathBuf,
 }
 impl DemoProvider {
    pub fn new(manager: NodeManager, keys_dir: PathBuf) -> Self {
        Self { manager, keys_dir }
    }
 }
 impl ProvisionPlugin for DemoProvider {
    fn create_node(
        &mut self,
        spec: NodeProvisionSpec,
        _sink: PluginSink,
    ) -> Result<PluginNodeHandle, String> {
        let attempt = spec.attempt_id;
        if let Some(runtime) = self.manager.get(attempt) {
            // Adoption: the child for this attempt already exists.
            return Ok(PluginNodeHandle {
                id: attempt,
                provider_process_id: runtime.pid,
            });
        }
        let logical_node = spec
            .env
            .iter()
            .find(|(key, _)| key == "DEMO_LOGICAL_NODE")
            .map(|(_, value)| value.clone())
            .ok_or_else(|| "spec missing DEMO_LOGICAL_NODE".to_owned())?;
        let key_file = self.keys_dir.join(format!("node-{attempt}.key"));
        let runtime = self
            .manager
            .request_spawn(attempt, logical_node, key_file)?;
        Ok(PluginNodeHandle {
            id: attempt,
            provider_process_id: runtime.pid,
        })
    }
    fn start_bootstrap(&mut self, _handle: &PluginNodeHandle) -> Result<(), String> {
        Ok(())
    }
    fn cancel_bootstrap(&mut self, _handle: &PluginNodeHandle) -> Result<(), String> {
        Ok(())
    }
    fn complete_bootstrap(&mut self, _handle: &PluginNodeHandle) -> Result<(), String> {
        Ok(())
    }
    fn stop_node(&mut self, handle: &PluginNodeHandle) -> Result<(), String> {
        // The backend path sends the process Stop command before calling
        // here; direct plugin-level stop just drops the registration.
        self.manager.remove(handle.id);
        Ok(())
    }
 }
 /// Per-node telemetry: one endpoint/producer per provisioned node so each
 /// lands on its own dashboard stream (one fleet card per node).
 pub struct NodeTelemetry {
    pub endpoint: TelemetryEndpoint,
    pub producer: TelemetryProducer,
 }
 impl NodeTelemetry {
    pub fn new(logical_node: &str, life: u64) -> Self {
        let stream = telemetry::frame::StreamId::new(
            telemetry::frame::NodeId::new(logical_node),
            telemetry::frame::Lifetime(life),
        );
        let endpoint = TelemetryEndpoint::with_descriptor(
            StreamDescriptor {
                stream,
                label: Some(format!("demo node {logical_node}")),
                origin: telemetry::frame::StreamOrigin::RemoteNode,
            },
            256,
            16,
        );
        let producer = endpoint.producer();
        Self { endpoint, producer }
    }
 }
 /// Relay actor: folds process-actor lifecycle reports into the registry.
 pub struct NodeRelayActor {
    manager: NodeManager,
    attempt: u64,
 }
 impl NodeRelayActor {
    pub fn new(manager: NodeManager, attempt: u64) -> Self {
        Self { manager, attempt }
    }
 }
 impl ActorInterface for NodeRelayActor {
    type Incoming = ProcessOutput;
    type Response = ();
    fn handle(&mut self, _ctx: &Ctx, output: ProcessOutput) {
        match output {
            ProcessOutput::Started { pid } => self.manager.set_pid(self.attempt, pid),
            ProcessOutput::Exited { status } => self.manager.set_exited(self.attempt, status),
            ProcessOutput::SpawnFailed { error } => {
                eprintln!("demo node {}: spawn failed: {error}", self.attempt);
                self.manager
                    .set_spawn_failed(self.attempt, error.to_string());
            }
            ProcessOutput::Error { error } => {
                eprintln!("demo node {}: process actor error: {error}", self.attempt);
            }
        }
    }
 }
 /// Demo lease/session identity (attempt-encoded, kit conventions).
 pub fn demo_lease(attempt: u64) -> LeaseFacts {
    let provider = ProviderKind::new("demo");
    let lease_id = ProviderLeaseId(format!("demo-lease-{attempt}"));
    LeaseFacts {
        destroy_handle: DestroyHandle {
            provider: provider.clone(),
            lease_id: lease_id.clone(),
            provider_contract_id: format!("demo-contract-{attempt}"),
        },
        provider,
        lease_id,
        provider_contract_id: format!("demo-contract-{attempt}"),
        offer_id: None,
        provider_metadata: BTreeMap::new(),
    }
 }
 pub fn demo_endpoint() -> SshEndpoint {
    SshEndpoint {
        host: "127.0.0.1".to_owned(),
        port: 0,
        user: "demo".to_owned(),
        auth_ref: "demo-key".to_owned(),
    }
 }
 /// Session identity: one bootstrap session per attempt (production
 /// convention — the supervisor's close path addresses it the same way).
 pub fn session_id_for(operation: OperationId) -> BootstrapSessionId {
    BootstrapSessionId(operation.attempt.0)
 }
 fn runtime_exited(manager: &NodeManager, attempt: u64) -> bool {
    manager.get(attempt).is_some_and(|runtime| runtime.exited.is_some())
 }
 /// The demo `EffectBackend`: routes effects through the plugin.
 pub struct DemoBackend {
    /// The plugin lives behind a mutex: `EffectBackend::execute` is `&self`
    /// while `ProvisionPlugin` methods are `&mut self` (kit adapter shape).
    pub plugin: Arc<Mutex<DemoProvider>>,
    pub manager: NodeManager,
    pub sender: ExternalSender,
 }
 impl DemoBackend {
    fn handle_for(&self, attempt: u64) -> PluginNodeHandle {
        let pid = self.manager.get(attempt).and_then(|r| r.pid);
        PluginNodeHandle {
            id: attempt,
            provider_process_id: pid,
        }
    }
 }
 impl EffectBackend for DemoBackend {
    fn execute(&self, effect: &PlannedEffect) -> Result<OperationOutcome, EffectError> {
        let attempt = effect.operation.attempt.0;
        match &effect.command {
            NodeManagerCommand::CreateLease(request) => {
                let spec = NodeProvisionSpec {
                    run_id: request.spec.run_id.0,
                    node_id: attempt,
                    attempt_id: attempt,
                    stage_index: None,
                    image: request.spec.shape.image.clone(),
                    env: vec![(
                        "DEMO_LOGICAL_NODE".to_owned(),
                        request.spec.logical_node_id.0.clone(),
                    )],
                    args: Vec::new(),
                    mounts: Vec::new(),
                };
                self.plugin
                    .lock()
                    .expect("demo provider")
                    .create_node(spec, null_sink())
                    .map_err(EffectError::definite)?;
                Ok(OperationOutcome::LeaseCreated(CreateLeaseResult {
                    lease: demo_lease(attempt),
                    endpoint: Some(demo_endpoint()),
                }))
            }
            NodeManagerCommand::LookupEndpoint(_) => {
                Ok(OperationOutcome::EndpointLookup(Some(demo_endpoint())))
            }
            NodeManagerCommand::StartBootstrap(_) => {
                let handle = self.handle_for(attempt);
                self.plugin
                    .lock()
                    .expect("demo provider")
                    .start_bootstrap(&handle)
                    .map_err(EffectError::definite)?;
                Ok(OperationOutcome::BootstrapStarted {
                    session_id: session_id_for(effect.operation),
                })
            }
            NodeManagerCommand::BootstrapConvergenceObserved { .. } => {
                let handle = self.handle_for(attempt);
                self.plugin
                    .lock()
                    .expect("demo provider")
                    .complete_bootstrap(&handle)
                    .map_err(EffectError::definite)?;
                Ok(OperationOutcome::BootstrapConvergenceAccepted)
            }
            NodeManagerCommand::CancelBootstrap { .. } => {
                let handle = self.handle_for(attempt);
                self.plugin
                    .lock()
                    .expect("demo provider")
                    .cancel_bootstrap(&handle)
                    .map_err(EffectError::definite)?;
                Ok(OperationOutcome::BootstrapCancelled)
            }
            NodeManagerCommand::DestroyLease(_) => {
                let handle = self.handle_for(attempt);
                // stop_node needs the sender; route it explicitly.
                stop_node_with_sender(&self.plugin, &self.manager, &handle, &self.sender)
                    .map_err(EffectError::definite)?;
                Ok(OperationOutcome::LeaseDestroyed)
            }
        }
    }
 }
 fn stop_node_with_sender(
    plugin: &Arc<Mutex<DemoProvider>>,
    manager: &NodeManager,
    handle: &PluginNodeHandle,
    sender: &ExternalSender,
 ) -> Result<(), String> {
    if let Some(runtime) = manager.get(handle.id) {
        if runtime.pid.is_some() && runtime.exited.is_none() {
            let _ = swactor_process::send_process_command(
                sender,
                runtime.process_actor,
                swactor_process::ProcessCommand::Stop {
                    kill_after: Some(Duration::from_secs(1)),
                },
            );
        }
        let deadline = std::time::Instant::now() + BACKEND_WAIT;
        while !runtime_exited(manager, handle.id) {
            if std::time::Instant::now() > deadline {
                break;
            }
            std::thread::sleep(Duration::from_millis(50));
        }
    }
    plugin.lock().expect("demo provider").stop_node(handle)
 }
 fn null_sink() -> PluginSink {
    struct NullSink;
    impl provisioning::plugin::PluginObservationSink for NullSink {
        fn observe(&self, _observation: provisioning::plugin::PluginObservation) {}
    }
    PluginSink::new(Arc::new(NullSink))
 }
 /// Register per-node supplementary channels; returns the `node.status`
 /// channel id for the supervisor's liveness heartbeats.
 pub fn register_node_channels(producer: &TelemetryProducer) -> telemetry::ChannelId {
    producer.register_channel(
        "node.status",
        ChannelContent::JsonRecord {
            schema: Some("demo.node.status.v1".to_owned()),
        },
    )
 }
 pub fn unix_ms(now: SystemTime) -> u64 {
    now.duration_since(SystemTime::UNIX_EPOCH)
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0)
 }
--- a/xtask/src/provisioning_demo/reconciler_page.html
+++ b/xtask/src/provisioning_demo/reconciler_page.html
@ -0,0 +1,156 @@
 <!doctype html>
 <html lang="en">
 <head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Provisioning Reconciler</title>
  <style>
    :root {
      color-scheme: dark;
      font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
      background: #0f172a;
      color: #e2e8f0;
    }
    * { box-sizing: border-box; }
    body { margin: 0; padding: 20px; }
    .nav { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 18px; padding: 8px; background: #111827; border: 1px solid #334155; border-radius: 12px; }
    .nav a { padding: 7px 10px; color: #cbd5e1; border: 1px solid transparent; border-radius: 8px; text-decoration: none; }
    .nav a:hover { color: #f8fafc; background: #1e293b; }
    header { display: flex; align-items: baseline; gap: 18px; flex-wrap: wrap; margin-bottom: 16px; }
    h1 { font-size: 24px; margin: 0; }
    .counter { font-size: 22px; font-weight: 700; }
    .counter .ready { color: #34d399; }
    .muted { color: #94a3b8; }
    .grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(240px, 1fr)); gap: 12px; margin-bottom: 18px; }
    .node-card { background: #1e293b; border: 1px solid #334155; border-radius: 12px; padding: 12px 14px; }
    .node-card h3 { margin: 0 0 6px; font-size: 15px; overflow-wrap: anywhere; }
    .badge { display: inline-block; padding: 2px 8px; border-radius: 999px; font-size: 12px; font-weight: 600; border: 1px solid; }
    .b-ready { color: #34d399; border-color: #065f46; background: #064e3b33; }
    .b-progress { color: #60a5fa; border-color: #1e40af; background: #1e3a8a33; }
    .b-failed { color: #f87171; border-color: #7f1d1d; background: #7f1d1d33; }
    .b-deleting { color: #fbbf24; border-color: #78350f; background: #78350f33; }
    .b-destroyed { color: #94a3b8; border-color: #334155; background: #1e293b66; }
    .meta { display: flex; flex-wrap: wrap; gap: 6px 12px; margin-top: 8px; font-size: 12px; }
    .feeds { display: grid; grid-template-columns: 1fr 1fr; gap: 12px; }
    @media (max-width: 900px) { .feeds { grid-template-columns: 1fr; } }
    .feed { background: #1e293b; border: 1px solid #334155; border-radius: 12px; padding: 10px 12px; }
    .feed h2 { margin: 0 0 8px; font-size: 14px; text-transform: uppercase; letter-spacing: 0.05em; color: #94a3b8; }
    .feed ul { list-style: none; margin: 0; padding: 0; font-family: ui-monospace, "SF Mono", Menlo, monospace; font-size: 12px; max-height: 420px; overflow-y: auto; }
    .feed li { padding: 3px 0; border-bottom: 1px solid #1e293b; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
    .feed li time { color: #64748b; margin-right: 8px; }
    .k-command { color: #93c5fd; }
    .k-result { color: #86efac; }
    .k-transition { color: #fbbf24; }
    .k-control { color: #f0abfc; }
    .k-observation { color: #fca5a5; }
    .k-error { color: #f87171; font-weight: 700; }
  </style>
 </head>
 <body>
  <nav class="nav">
    <a href="/">dashboard</a>
    <a href="/view/reconciler" data-active="true">reconciler</a>
    <a href="/view/fleet">fleet</a>
    <a href="/view/demo-control">fleet control</a>
  </nav>
  <header>
    <h1>Provisioning Reconciler</h1>
    <div class="counter"><span class="ready" id="ready">0</span> / <span id="desired">0</span> <span class="muted">Ready</span></div>
    <div class="muted">gen <span id="gen">0</span> · <span id="converged">—</span> · snapshot <span id="age">—</span> old</div>
  </header>
  <section class="grid" id="nodes"></section>
  <section class="feeds">
    <div class="feed">
      <h2>Commands out</h2>
      <ul id="commands"></ul>
    </div>
    <div class="feed">
      <h2>Events in · transitions · control</h2>
      <ul id="events"></ul>
    </div>
  </section>
  <script>
    'use strict';
    const API_URL = '/api/view/reconciler';
    const STAGE_CLASS = {
      New: 'b-destroyed', LeaseRequested: 'b-progress', LeaseCreated: 'b-progress',
      EndpointKnown: 'b-progress', BootstrapRunning: 'b-progress',
      SwactorJoined: 'b-progress', HandedOff: 'b-ready', Dormant: 'b-ready',
      Failed: 'b-failed', Destroyed: 'b-destroyed'
    };
    const INTENT_DELETING = new Set(['Deleting']);
    function escapeHtml(value) {
      return String(value).replace(/[&<>"']/g, (ch) => ({
        '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;'
      }[ch]));
    }
    function fmtTime(ms) {
      if (!ms) return '--:--:--';
      const date = new Date(ms);
      return [date.getHours(), date.getMinutes(), date.getSeconds()]
        .map((part) => String(part).padStart(2, '0')).join(':');
    }
    function fmtAge(ms) {
      if (ms < 1500) return ms + 'ms';
      if (ms < 60_000) return Math.round(ms / 1000) + 's';
      return Math.round(ms / 60_000) + 'm';
    }
    function renderNode(node) {
      const stage = node.stage || 'New';
      const deleting = INTENT_DELETING.has(node.intent);
      const badgeClass = deleting ? 'b-deleting' : (STAGE_CLASS[stage] || 'b-progress');
      const sub = node.bootstrap ? ` · ${escapeHtml(node.bootstrap)}` : '';
      const failure = node.failure ? `<div class="k-error">${escapeHtml(node.failure)}</div>` : '';
      return `<article class="node-card">
        <h3>${escapeHtml(node.id)}</h3>
        <span class="badge ${badgeClass}">${escapeHtml(stage)}${sub}</span>
        <div class="meta">
          <span class="muted">attempt ${escapeHtml(node.attempt)}</span>
          ${node.pid != null ? `<span class="muted">pid ${escapeHtml(node.pid)}</span>` : ''}
          ${node.ready ? '<span class="b-ready badge">ready</span>' : ''}
        </div>
        ${failure}
      </article>`;
    }
    function renderFeedLine(line, filter) {
      if (filter && line.kind !== filter) return '';
      return `<li class="k-${escapeHtml(line.kind)}"><time>${fmtTime(line.at_ms)}</time>` +
        `${line.node ? '[' + escapeHtml(line.node) + '] ' : ''}${escapeHtml(line.detail)}</li>`;
    }
    function render(snapshot) {
      if (!snapshot) return;
      document.getElementById('ready').textContent = snapshot.ready;
      document.getElementById('desired').textContent = snapshot.desired;
      document.getElementById('gen').textContent = snapshot.generation;
      document.getElementById('converged').textContent = snapshot.converged ? 'converged' : 'converging…';
      document.getElementById('age').textContent = fmtAge(snapshot.age_ms);
      document.getElementById('nodes').innerHTML =
        (snapshot.nodes || []).map(renderNode).join('') ||
        '<p class="muted">No nodes yet.</p>';
      const feed = (snapshot.feed || []).slice().reverse();
      document.getElementById('commands').innerHTML =
        feed.map((line) => renderFeedLine(line, 'command')).join('') || '<li class="muted">none</li>';
      const eventKinds = new Set(['command']);
      document.getElementById('events').innerHTML =
        feed.filter((line) => !eventKinds.has(line.kind))
            .map((line) => renderFeedLine(line))
            .join('') || '<li class="muted">none</li>';
    }
    async function poll() {
      try {
        const response = await fetch(API_URL);
        if (response.ok) render(await response.json());
      } catch { /* transient */ }
    }
    poll();
    setInterval(poll, 1000);
  </script>
 </body>
 </html>
--- a/xtask/src/provisioning_demo/view.rs
+++ b/xtask/src/provisioning_demo/view.rs
@ -0,0 +1,170 @@
 //! Reconciler dashboard view — k8s workload semantics.
 //!
 //! Header: current-ready vs desired (Deployment-style `2 / 3`).
 //! Node cards: stage badge, bootstrap sub-stage, attempt, PID — pod status.
 //! Feeds: commands out / results + transitions in, `kubectl describe` events
 //! style, one line each with timestamps.
 use std::collections::VecDeque;
 use std::time::{SystemTime, UNIX_EPOCH};
 use parking_lot::Mutex;
 use serde::Serialize;
 use serde_json::Value;
 use telemetry::frame::{Frame, StreamId};
 use dashboard::view::DashboardView;
 use dashboard::FrameEvent;
 const EVENTS_CHANNEL: &str = "prov.reconciler.events";
 const SNAPSHOT_CHANNEL: &str = "prov.reconciler.snapshot";
 const FEED_CAP: usize = 250;
 #[derive(Clone, Serialize)]
 struct FeedLine {
    at_ms: u64,
    kind: String,
    node: String,
    detail: String,
 }
 #[derive(Serialize)]
 struct ReconcilerSnapshot {
    ready: u64,
    desired: u64,
    generation: u64,
    converged: bool,
    age_ms: u64,
    nodes: Vec<Value>,
    feed: Vec<FeedLine>,
 }
 struct ViewState {
    header: Option<Value>,
    nodes: Vec<Value>,
    feed: VecDeque<FeedLine>,
    snapshot_at_ms: u64,
 }
 impl ViewState {
    fn push_feed(&mut self, line: FeedLine) {
        self.feed.push_back(line);
        while self.feed.len() > FEED_CAP {
            self.feed.pop_front();
        }
    }
 }
 pub struct ReconcilerDashboardView {
    state: Mutex<ViewState>,
 }
 impl Default for ReconcilerDashboardView {
    fn default() -> Self {
        Self {
            state: Mutex::new(ViewState {
                header: None,
                nodes: Vec::new(),
                feed: VecDeque::new(),
                snapshot_at_ms: 0,
            }),
        }
    }
 }
 fn unix_ms(now: SystemTime) -> u64 {
    now.duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_millis() as u64)
        .unwrap_or(0)
 }
 impl DashboardView for ReconcilerDashboardView {
    fn id(&self) -> &'static str {
        "reconciler"
    }
    fn title(&self) -> &'static str {
        "Provisioning Reconciler"
    }
    fn path(&self) -> &'static str {
        "reconciler"
    }
    fn channels(&self) -> &'static [&'static str] {
        &[EVENTS_CHANNEL, SNAPSHOT_CHANNEL]
    }
    fn ingest(&self, _stream: &StreamId, _frame: &Frame, event: &FrameEvent) {
        let Ok(payload) = serde_json::from_slice::<Value>(&event.payload) else {
            return;
        };
        let mut state = self.state.lock();
        match event.channel.as_str() {
            SNAPSHOT_CHANNEL => {
                state.nodes = payload
                    .get("nodes")
                    .and_then(Value::as_array)
                    .cloned()
                    .unwrap_or_default();
                state.snapshot_at_ms = payload
                    .get("at_ms")
                    .and_then(Value::as_u64)
                    .unwrap_or_else(|| unix_ms(SystemTime::now()));
                state.header = Some(payload);
            }
            EVENTS_CHANNEL => {
                let line = FeedLine {
                    at_ms: payload
                        .get("at_ms")
                        .and_then(Value::as_u64)
                        .unwrap_or_default(),
                    kind: payload
                        .get("kind")
                        .and_then(Value::as_str)
                        .unwrap_or("event")
                        .to_owned(),
                    node: payload
                        .get("node")
                        .and_then(Value::as_str)
                        .unwrap_or("")
                        .to_owned(),
                    detail: payload
                        .get("detail")
                        .and_then(Value::as_str)
                        .unwrap_or("")
                        .to_owned(),
                };
                state.push_feed(line);
            }
            _ => {}
        }
    }
    fn snapshot_json(&self) -> Value {
        let state = self.state.lock();
        let header = state.header.clone().unwrap_or(Value::Null);
        let now = unix_ms(SystemTime::now());
        serde_json::to_value(ReconcilerSnapshot {
            ready: header.get("ready").and_then(Value::as_u64).unwrap_or(0),
            desired: header.get("desired").and_then(Value::as_u64).unwrap_or(0),
            generation: header
                .get("generation")
                .and_then(Value::as_u64)
                .unwrap_or(0),
            converged: header
                .get("converged")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            age_ms: now.saturating_sub(state.snapshot_at_ms),
            nodes: state.nodes.clone(),
            feed: state.feed.iter().cloned().collect(),
        })
        .unwrap_or_else(|_| serde_json::json!({}))
    }
    fn html(&self) -> Option<&'static str> {
        Some(include_str!("reconciler_page.html"))
    }
 }