From d19dd913244d06cded895d56d1b1f2ec6ed5a162 Mon Sep 17 00:00:00 2001
From: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>
Date: Sat, 15 Aug 2026 18:10:52 +0400
Subject: [PATCH] feat(xtask): provisioning-reconciler-demo with live fleet
 control
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`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.
---
 Cargo.lock                                    |  56 +-
 crates/dashboard/Cargo.toml                   |   5 +
 crates/dashboard/src/control.rs               |  44 +
 crates/dashboard/src/demo_control.rs          | 112 +++
 crates/dashboard/src/demo_control_page.html   | 148 ++++
 crates/dashboard/src/hardware_page.html       |   1 +
 crates/dashboard/src/hardware_view.rs         |  52 +-
 crates/dashboard/src/lib.rs                   |   9 +-
 crates/dashboard/src/server.rs                |  59 +-
 xtask/Cargo.toml                              |  13 +
 xtask/src/main.rs                             |   9 +
 xtask/src/provisioning_demo/control.rs        |  36 +
 xtask/src/provisioning_demo/feed.rs           | 811 ++++++++++++++++++
 xtask/src/provisioning_demo/mod.rs            | 359 ++++++++
 xtask/src/provisioning_demo/node.rs           | 119 +++
 xtask/src/provisioning_demo/provider.rs       | 469 ++++++++++
 .../provisioning_demo/reconciler_page.html    | 156 ++++
 xtask/src/provisioning_demo/view.rs           | 170 ++++
 18 files changed, 2599 insertions(+), 29 deletions(-)
 create mode 100644 crates/dashboard/src/control.rs
 create mode 100644 crates/dashboard/src/demo_control.rs
 create mode 100644 crates/dashboard/src/demo_control_page.html
 create mode 100644 xtask/src/provisioning_demo/control.rs
 create mode 100644 xtask/src/provisioning_demo/feed.rs
 create mode 100644 xtask/src/provisioning_demo/mod.rs
 create mode 100644 xtask/src/provisioning_demo/node.rs
 create mode 100644 xtask/src/provisioning_demo/provider.rs
 create mode 100644 xtask/src/provisioning_demo/reconciler_page.html
 create mode 100644 xtask/src/provisioning_demo/view.rs

diff --git a/Cargo.lock b/Cargo.lock
index cc66e65..3f34102 100644
--- 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]]
diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml
index c10b09d..3a3ba2a 100644
--- 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 = []
diff --git a/crates/dashboard/src/control.rs b/crates/dashboard/src/control.rs
new file mode 100644
index 0000000..a36b59c
--- /dev/null
+++ 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())
+}
diff --git a/crates/dashboard/src/demo_control.rs b/crates/dashboard/src/demo_control.rs
new file mode 100644
index 0000000..deaee00
--- /dev/null
+++ 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"))
+    }
+}
diff --git a/crates/dashboard/src/demo_control_page.html b/crates/dashboard/src/demo_control_page.html
new file mode 100644
index 0000000..80014d4
--- /dev/null
+++ 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>
diff --git a/crates/dashboard/src/hardware_page.html b/crates/dashboard/src/hardware_page.html
index 32c58ad..5033b9e 100644
--- 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">
diff --git a/crates/dashboard/src/hardware_view.rs b/crates/dashboard/src/hardware_view.rs
index c0974bf..179db10 100644
--- 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.saturating_add(device.memory_used_mib.unwrap_or_default());
+                gpu_memory_used_mib = gpu_memory_used_mib
+                    .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()
diff --git a/crates/dashboard/src/lib.rs b/crates/dashboard/src/lib.rs
index 5b08b21..e282716 100644
--- 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());
diff --git a/crates/dashboard/src/server.rs b/crates/dashboard/src/server.rs
index 933b28d..72eb705 100644
--- 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))
-        .with_state(state)
+        .route("/view/{*path}", get(view_page));
+    #[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> {
diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml
index 7c71cc8..1b1f6f8 100644
--- 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"
diff --git a/xtask/src/main.rs b/xtask/src/main.rs
index c4235bc..72aa5a0 100644
--- 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
diff --git a/xtask/src/provisioning_demo/control.rs b/xtask/src/provisioning_demo/control.rs
new file mode 100644
index 0000000..b6a1ceb
--- /dev/null
+++ 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,
+            }
+        }
+    });
+}
diff --git a/xtask/src/provisioning_demo/feed.rs b/xtask/src/provisioning_demo/feed.rs
new file mode 100644
index 0000000..3e60f33
--- /dev/null
+++ 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),
+    }
+}
+
diff --git a/xtask/src/provisioning_demo/mod.rs b/xtask/src/provisioning_demo/mod.rs
new file mode 100644
index 0000000..47efe88
--- /dev/null
+++ 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.
+}
diff --git a/xtask/src/provisioning_demo/node.rs b/xtask/src/provisioning_demo/node.rs
new file mode 100644
index 0000000..105e367
--- /dev/null
+++ 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,
+    })
+}
diff --git a/xtask/src/provisioning_demo/provider.rs b/xtask/src/provisioning_demo/provider.rs
new file mode 100644
index 0000000..1598dd1
--- /dev/null
+++ 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)
+}
diff --git a/xtask/src/provisioning_demo/reconciler_page.html b/xtask/src/provisioning_demo/reconciler_page.html
new file mode 100644
index 0000000..1da5a2c
--- /dev/null
+++ 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>
diff --git a/xtask/src/provisioning_demo/view.rs b/xtask/src/provisioning_demo/view.rs
new file mode 100644
index 0000000..872ea59
--- /dev/null
+++ 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"))
+    }
+}
+