provisioning-reconciler-demo: wire-announce readiness + --docker node kind

Replace the per-attempt key-file side channel (DEMO_NODE_KEY_FILE,
read_key_report, JoinCheck) with a control-plane announce: node roles
send a tagged gossip frame {attempt, logical_node, key_hex,
endpoint_addr_json} after joining and every heartbeat thereafter.

- iroh-driver: IrohDriver::send_tagged_gossip — fire-and-forget
  tag-routed gossip egress for bridge-less clients (reuses cached/join
  connections, dials with backoff).
- provisioning: BootstrapMsg::Announce — first delivery while
  bootstrapping completes the attempt (collector + exactly-once
  Bootstrapped report); duplicates, misrouted attempts, and
  terminal-phase announces drop. Unit-tested.
- xtask demo: AnnounceActor decodes the tag-routed frame and forwards
  by attempt to the owning bootstrap actor; last_announce_ms is the
  wire heartbeat. LocalProcessLogic keeps only process lifecycle.
- --docker: DockerProcessLogic (kind "docker") — attached
  "docker run --rm" child on a per-run labeled bridge network
  (foreign-node masking: per-container IPs, gateway-dialed
  supervisor). Standalone scratch image from the static-musl xtask
  binary (37MB), staged one-file build context. The container is
  force-removed on every terminal path so a SIGKILLed docker CLI
  cannot orphan a running container.
- Cleanup: no volumes/mounts; label-filtered exit sweep (SIGINT and
  SIGTERM both drain first) + startup sweep of stale demo resources;
  images persist per run token.

Verified live: process kind (kill -> replacement in 3.4s, provision/
remove/kill waves, zero orphans) and docker kind (8-node abuse waves
across docker kill, mid-provision control kills, CLI SIGKILL orphans
force-removed, SIGKILL-crash leftovers swept on restart, clean exits
leave zero containers/networks/CLIs). provisioning 22 + iroh-driver
13 tests pass.

crates/iroh-driver/src/iroh_driver.rs

@@ -31,14 +31,14 @@ use distribution::swim::actor::SwimIn;
 use distribution::transport_bridge::{OutFrame, Outbox, RelayMirror, RouteView, peer_addr};
 use distribution::types::NodeId;
 
-use crate::telemetry_transport::{
-    TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, read_events_from_stream,
-    spawn_subscription_writer,
-};
 use crate::edge_transport::{
     EDGE_ALPN, EdgeSendHandle, EdgeTransportEvent, spawn_edge_recv_pump,
     spawn_edge_send_pump as spawn_edge_sender_task,
 };
+use crate::telemetry_transport::{
+    TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, read_events_from_stream,
+    spawn_subscription_writer,
+};
 use swactor::actor::ActorAddress;
 use swactor::runtime::Runtime;
 use swactor_transport::CodecRegistry;
@@ -402,15 +402,15 @@ impl IrohDriver {
 
             // Only relax relay-cert verification for a custom relay; Default /
             // Staging relays keep full WebPKI verification.
-        if custom_relay {
+            if custom_relay {
-            builder = builder.ca_roots_config(iroh::tls::CaRootsConfig::insecure_skip_verify());
+                builder = builder.ca_roots_config(iroh::tls::CaRootsConfig::insecure_skip_verify());
-            // Relay-only: drop direct IP transports so the endpoint neither
+                // Relay-only: drop direct IP transports so the endpoint neither
-            // advertises nor chases direct addresses. Without this, iroh learns
+                // advertises nor chases direct addresses. Without this, iroh learns
-            // a peer's NAT-obscured/container-local direct addr via discovery
+                // a peer's NAT-obscured/container-local direct addr via discovery
-            // and prefers it over the relay, black-holing all data. Applied to
+                // and prefers it over the relay, black-holing all data. Applied to
-            // every endpoint so neither side has a direct addr to publish.
+                // every endpoint so neither side has a direct addr to publish.
-            builder = builder.clear_ip_transports();
+                builder = builder.clear_ip_transports();
-        }
+            }
 
             if let Some(key) = secret_key {
                 builder = builder.secret_key(key);
@@ -444,8 +444,7 @@ impl IrohDriver {
             Arc::new(Mutex::new(Vec::new()));
         let other_accepted_conns: Arc<Mutex<Vec<(NodeId, Vec<u8>, Connection)>>> =
             Arc::new(Mutex::new(Vec::new()));
-        let telemetry_reads: Arc<Mutex<Vec<TelemetryQuicRead>>> =
+        let telemetry_reads: Arc<Mutex<Vec<TelemetryQuicRead>>> = Arc::new(Mutex::new(Vec::new()));
-            Arc::new(Mutex::new(Vec::new()));
         let edge_events: Arc<Mutex<Vec<EdgeTransportEvent>>> = Arc::new(Mutex::new(Vec::new()));
         {
             let ep = endpoint.clone();
@@ -509,7 +508,6 @@ impl IrohDriver {
         })
     }
 
-
     /// Clone the iroh endpoint for creating outbound connections.
     pub fn endpoint(&self) -> Endpoint {
         self.endpoint.clone()
@@ -851,11 +849,12 @@ impl IrohDriver {
                     );
                 }
 
-                let connect_result = engine.timeout(
+                let connect_result = engine
-                    per_attempt_timeout,
+                    .timeout(
-                    endpoint.connect(seed_addr.clone(), ALPN),
+                        per_attempt_timeout,
-                )
+                        endpoint.connect(seed_addr.clone(), ALPN),
-                .await;
+                    )
+                    .await;
 
                 match connect_result {
                     Ok(Ok(conn)) => {
@@ -949,7 +948,117 @@ impl IrohDriver {
             }
         });
     }
-
+    /// Fire-and-forget a single tagged gossip frame to the peer behind
+    /// `addr`.
+    ///
+    /// The frame format matches the actor-bridge ingress (`write_message`)
+    /// with the destination set to the peer's peer-mailbox, so a receiving
+    /// bridge routes it by `type_tag` through its `routes` table. This is
+    /// lightweight egress for clients that never install the actor bridge —
+    /// e.g. a bootstrap node announcing its identity to a supervisor. The
+    /// send reuses an open cached connection (folding completed
+    /// [`Self::join`] dials first), connects fresh otherwise, and retries
+    /// with backoff; callers that need delivery guarantees re-send at their
+    /// own cadence.
+    pub fn send_tagged_gossip(&self, addr: EndpointAddr, tag: &[u8], payload: Vec<u8>) {
+        let tag = tag.to_vec();
+        let peer = NodeId(*addr.id.as_bytes());
+        let endpoint = self.endpoint.clone();
+        let engine = self.engine.clone();
+        let conns = Arc::clone(&self.conns);
+        let pending_joins = Arc::clone(&self.pending_joins);
+        self.engine.spawn(async move {
+            let dest = peer_addr(peer);
+            let tag_len = (tag.len() as u32).to_be_bytes();
+            let mut delay = Duration::from_millis(250);
+            let max_delay = Duration::from_secs(2);
+            for _ in 0..4 {
+                // Reuse an open cached connection, or fold a completed (but
+                // uncached) join dial into the cache and take it. Sync-only,
+                // so no guard crosses an await.
+                let reused = {
+                    let mut cache = conns.lock();
+                    if let Some(cached) = cache.connections.get(&peer) {
+                        if cached.conn.close_reason().is_none() {
+                            Some(cached.conn.clone())
+                        } else {
+                            cache.connections.remove(&peer);
+                            None
+                        }
+                    } else {
+                        None
+                    }
+                }
+                .or_else(|| {
+                    let mut pending = pending_joins.lock();
+                    let index = pending.iter().position(|result| result.node_id == peer)?;
+                    let result = pending.remove(index);
+                    let mut cache = conns.lock();
+                    let generation = cache.next_generation;
+                    cache.next_generation = cache.next_generation.wrapping_add(1).max(1);
+                    cache.connections.insert(
+                        peer,
+                        CachedConnection {
+                            generation,
+                            conn: result.conn.clone(),
+                        },
+                    );
+                    Some(result.conn)
+                });
+                let conn = match reused {
+                    Some(conn) => conn,
+                    None => {
+                        // No connection yet: dial one (no locks held). The
+                        // fresh connection is cached so later sends reuse it.
+                        let connect = engine
+                            .timeout(
+                                Duration::from_secs(10),
+                                endpoint.connect(addr.clone(), ALPN),
+                            )
+                            .await;
+                        match connect {
+                            Ok(Ok(conn)) => {
+                                let mut cache = conns.lock();
+                                let generation = cache.next_generation;
+                                cache.next_generation =
+                                    cache.next_generation.wrapping_add(1).max(1);
+                                cache.connections.insert(
+                                    peer,
+                                    CachedConnection {
+                                        generation,
+                                        conn: conn.clone(),
+                                    },
+                                );
+                                conn
+                            }
+                            _ => {
+                                engine.timer(delay).await;
+                                delay = (delay * 2).min(max_delay);
+                                continue;
+                            }
+                        }
+                    }
+                };
+                let result: Result<(), Box<dyn std::error::Error + Send + Sync>> = async {
+                    let mut send = conn.open_uni().await?;
+                    send.write_all(&dest.0).await?;
+                    send.write_all(&tag_len).await?;
+                    send.write_all(&tag).await?;
+                    send.write_all(&payload).await?;
+                    send.finish()?;
+                    Ok(())
+                }
+                .await;
+                if result.is_ok() {
+                    return;
+                }
+                // Write failed: evict so the next attempt re-dials.
+                conns.lock().connections.remove(&peer);
+                engine.timer(delay).await;
+                delay = (delay * 2).min(max_delay);
+            }
+        });
+    }
     // ─── Actor bridge: iroh ⇄ swactor runtime ─────────────────────────
 
     /// Install the actor-bridge wiring so the driver shuttles frames between iroh
@@ -1129,9 +1238,13 @@ impl AdapterPump {
             let mut cache = self.conns.lock();
             let generation = cache.next_generation;
             cache.next_generation = cache.next_generation.wrapping_add(1).max(1);
-            cache
+            cache.connections.insert(
-                .connections
+                node_id,
-                .insert(node_id, CachedConnection { generation, conn: conn.clone() });
+                CachedConnection {
+                    generation,
+                    conn: conn.clone(),
+                },
+            );
             generation
         };
         self.spawn_reader(node_id, generation, conn);
@@ -1310,11 +1423,12 @@ impl AdapterPump {
             const ATTEMPTS: u32 = 3;
             let per_attempt_timeout = Duration::from_secs(10);
             for attempt in 1..=ATTEMPTS {
-                let result = engine.timeout(
+                let result = engine
-                    per_attempt_timeout,
+                    .timeout(
-                    endpoint.connect(dial_addr.clone(), ALPN),
+                        per_attempt_timeout,
-                )
+                        endpoint.connect(dial_addr.clone(), ALPN),
-                .await;
+                    )
+                    .await;
                 if let Ok(Ok(conn)) = result {
                     pending.lock().push(JoinResult { node_id, conn });
                     break;

crates/iroh-driver/src/lib.rs

@@ -9,17 +9,17 @@
 // work goes through `EngineHandle`.
 #![deny(clippy::disallowed_methods)]
 
-pub mod telemetry_transport;
 pub mod driver_pumps;
 pub mod edge_transport;
 pub mod endpoint_advertisement;
 pub mod iroh_driver;
+pub mod telemetry_transport;
 
 pub use endpoint_advertisement::{
     EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint,
 };
 pub use iroh_driver::{
-    ConnType, TelemetryPublishHandle, IrohDriver, IrohDriverConfig, JoinPhase, JoinStatus,
+    ConnType, IrohDriver, IrohDriverConfig, JoinPhase, JoinStatus, TelemetryPublishHandle,
     conn_type_of, discover_lan_ips,
 };
 
@@ -27,7 +27,8 @@ pub use edge_transport::{EDGE_ALPN, EdgeSendHandle, EdgeTransportEvent, EdgeTran
 
 pub use telemetry_transport::{
     TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, TelemetryQuicWriteStats,
-    read_events_from_stream, read_next_event, read_next_uni_from_connection, read_stream_header,
+    read_events_from_stream, read_next_event, read_next_uni_from_connection, read_pull_request,
-    read_stream_into_fanout, spawn_connection_reader, spawn_subscription_writer,
+    read_stream_header, read_stream_into_fanout, spawn_connection_reader, spawn_pull_collector,
-    write_available_subscription, write_event, write_subscription_until_closed,
+    spawn_pull_server, spawn_subscription_writer, write_available_subscription, write_event,
+    write_pull_request, write_subscription_until_closed,
 };

crates/iroh-driver/src/telemetry_transport.rs

@@ -5,12 +5,14 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use crossbeam_channel::TryRecvError;
-use telemetry::{TelemetrySnapshot, TelemetrySubscription};
-use telemetry::frame::{ChannelDescriptor, ChannelId, ChannelRef, TelemetryEvent, FrameDelivery, Position,
-StreamDescriptor,};
 use iroh::endpoint::{Connection, RecvStream, SendStream};
 use iroh::{Endpoint, EndpointAddr};
 use swactor_engine::EngineHandle;
+use telemetry::frame::{
+    ChannelDescriptor, ChannelId, ChannelRef, FrameDelivery, Position, StreamDescriptor,
+    TelemetryEvent,
+};
+use telemetry::{TelemetrySnapshot, TelemetrySubscription};
 
 pub const TELEMETRY_ALPN: &[u8] = b"swactor/telemetry/0";
 
@@ -18,6 +20,134 @@ const MAGIC: &[u8; 4] = b"DSQ1";
 const TAG_CHANNEL_DECLARED: u8 = 0x01;
 const TAG_FRAME: u8 = 0x02;
 const TAG_STREAM_ENDED: u8 = 0x03;
+
+// ─── Pull model: collector-initiated subscriptions ───────────────────────────
+//
+// A supervising node dials a freshly-bootstrapped node on `TELEMETRY_ALPN`,
+// sends one subscription request on the first uni stream, and the node answers
+// by writing the existing header+events stream shape on a uni stream of the
+// same connection. Subscription lifetime = connection lifetime.
+
+/// Write a pull request (magic + flow id + token + `SubscriptionRequest`) and
+/// finish the stream so the serving side's read completes.
+const REQUEST_MAGIC: &[u8; 4] = b"DSQR";
+pub async fn write_pull_request(
+    send: &mut SendStream,
+    flow_id: [u8; 16],
+    token: &[u8],
+    request: &telemetry::SubscriptionRequest,
+) -> Result<(), BoxError> {
+    if token.len() > u16::MAX as usize {
+        return Err("telemetry pull token exceeds u16 length prefix".into());
+    }
+    send.write_all(REQUEST_MAGIC).await?;
+    send.write_all(&flow_id).await?;
+    send.write_all(&(token.len() as u16).to_le_bytes()).await?;
+    send.write_all(token).await?;
+    write_json(send, request).await?;
+    send.finish()?;
+    Ok(())
+}
+
+/// Read a pull request written by [`write_pull_request`].
+pub async fn read_pull_request(
+    recv: &mut RecvStream,
+) -> Result<([u8; 16], Vec<u8>, telemetry::SubscriptionRequest), BoxError> {
+    let mut magic = [0u8; 4];
+    recv.read_exact(&mut magic).await?;
+    if &magic != REQUEST_MAGIC {
+        return Err("invalid telemetry pull request magic".into());
+    }
+    let mut flow_id = [0u8; 16];
+    recv.read_exact(&mut flow_id).await?;
+    let mut token_len = [0u8; 2];
+    recv.read_exact(&mut token_len).await?;
+    let token_len = u16::from_le_bytes(token_len) as usize;
+    let mut token = vec![0u8; token_len];
+    recv.read_exact(&mut token).await?;
+    let request = read_json(recv).await?;
+    Ok((flow_id, token, request))
+}
+
+/// Node side: serve one accepted `TELEMETRY_ALPN` connection. Reads the pull
+/// request from the first uni stream, subscribes the local endpoint, and
+/// writes the answering subscription stream (header + events, until the
+/// subscription ends or the connection drops) on a uni stream of the same
+/// connection. One request per connection; the task exits when the writer
+/// ends or the connection fails.
+pub fn spawn_pull_server(
+    engine: &EngineHandle,
+    conn: Connection,
+    endpoint: std::sync::Arc<telemetry::TelemetryEndpoint>,
+    idle_sleep: Duration,
+) {
+    let engine_handle = engine.clone();
+    engine.spawn(async move {
+        let Ok(mut recv) = conn.accept_uni().await else {
+            return;
+        };
+        let Ok((_flow_id, token, request)) = read_pull_request(&mut recv).await else {
+            return;
+        };
+        let subscription = endpoint.subscribe("supervisor-pull", request);
+        let header =
+            match TelemetryQuicHeader::from_snapshot(_flow_id, token, subscription.snapshot()) {
+                Ok(header) => header,
+                Err(_) => return,
+            };
+        let Ok(send) = conn.open_uni().await else {
+            return;
+        };
+        let _ =
+            write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep)
+                .await;
+    });
+}
+
+/// Supervisor side: dial a node on `TELEMETRY_ALPN`, send the pull request,
+/// and stream answering events into `fanout` as they arrive (incrementally,
+/// not buffered until stream end). The header is reported through
+/// `on_header` first so the caller can register stream/channel metadata
+/// before any frame lands.
+pub fn spawn_pull_collector(
+    engine: &EngineHandle,
+    endpoint: Endpoint,
+    peer: EndpointAddr,
+    flow_id: [u8; 16],
+    token: Vec<u8>,
+    request: telemetry::SubscriptionRequest,
+    fanout: std::sync::Arc<telemetry::DeliveryFanout>,
+    on_header: std::sync::mpsc::Sender<TelemetryQuicHeader>,
+) {
+    engine.spawn(async move {
+        let peer_id = peer.id.to_string();
+        let Ok(conn) = endpoint.connect(peer, TELEMETRY_ALPN).await else {
+            eprintln!("telemetry-pull: connect to {peer_id} failed");
+            return;
+        };
+        let Ok(mut req) = conn.open_uni().await else {
+            eprintln!("telemetry-pull: open request stream to {peer_id} failed");
+            return;
+        };
+        if let Err(error) = write_pull_request(&mut req, flow_id, &token, &request).await {
+            eprintln!("telemetry-pull: write request to {peer_id} failed: {error}");
+            return;
+        }
+        let Ok(mut recv) = conn.accept_uni().await else {
+            eprintln!("telemetry-pull: no answer stream from {peer_id}");
+            return;
+        };
+        let Ok(header) = read_header(&mut recv).await else {
+            eprintln!("telemetry-pull: answer header from {peer_id} unreadable");
+            return;
+        };
+        let _ = on_header.send(header.clone());
+        let stream = header.stream.clone();
+        while let Ok(Some(event)) = read_next_event(&mut recv, &stream).await {
+            fanout.publish(event);
+        }
+    });
+}
 const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024;
 
 type BoxError = Box<dyn Error + Send + Sync + 'static>;
@@ -87,7 +217,9 @@ pub fn spawn_subscription_writer(
         let Ok(send) = conn.open_uni().await else {
             return;
         };
-        let _ = write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep).await;
+        let _ =
+            write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep)
+                .await;
     });
 }
 
@@ -157,10 +289,7 @@ async fn write_subscription_inner(
     Ok(stats)
 }
 
-pub async fn write_event(
+pub async fn write_event(send: &mut SendStream, event: &TelemetryEvent) -> Result<usize, BoxError> {
-    send: &mut SendStream,
-    event: &TelemetryEvent,
-) -> Result<usize, BoxError> {
     let mut bytes = Vec::new();
     match event {
         TelemetryEvent::StreamDeclared(_) => return Ok(0),
@@ -232,10 +361,7 @@ pub fn spawn_connection_reader(
     });
 }
 
-async fn write_header(
+async fn write_header(send: &mut SendStream, header: &TelemetryQuicHeader) -> Result<(), BoxError> {
-    send: &mut SendStream,
-    header: &TelemetryQuicHeader,
-) -> Result<(), BoxError> {
     if header.token.len() > u16::MAX as usize {
         return Err("telemetry token exceeds u16 length prefix".into());
     }

crates/provisioning/Cargo.toml

@@ -7,6 +7,8 @@ publish = false
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
+swactor = { path = "../.." }
+swactor-engine = { path = "../engine" }
 
 [dev-dependencies]
 parking_lot = "0.12"

crates/provisioning/src/bootstrap.rs

@@ -0,0 +1,536 @@
+//! Actor-controlled node bootstrap.
+//!
+//! One [`BootstrapActor`] owns one node provision attempt end-to-end. The
+//! supervision (reconciler loop) talks to every bootstrap actor through the
+//! same standard interface — [`BootstrapMsg`] in, [`BootstrapEvent`] out — and
+//! never learns *how* a node is launched. Target-specific behavior (local OS
+//! process, docker container over ssh, leased remote host over ssh, …) lives
+//! behind the private [`BootstrapLogic`] extension point, resolved from a
+//! [`BootstrapRegistry`] by the spec's `kind`.
+//!
+//! The generic actor owns the universal lifecycle state machine
+//! (`Bootstrapping → Ready → Stopping → Dead`), self-drives its probes on the
+//! engine, starts telemetry collection once a node reports bootstrapped, and
+//! guarantees idempotent termination. Logic implementations report raw
+//! observations; the actor alone decides what the supervision sees.
+//!
+//! Telemetry collection is injected as a [`NodeTelemetryCollector`] hook so
+//! this crate stays independent of any transport; the demo's collector dials
+//! the node on `TELEMETRY_ALPN` and pulls its subscription
+//! (`iroh_driver::spawn_pull_collector`).
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime};
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+use swactor::runtime::ExternalSender;
+use swactor_engine::EngineHandle;
+
+/// Default probe period for the self-driven state machine.
+pub const DEFAULT_PROBE_PERIOD: Duration = Duration::from_millis(250);
+
+/// What the bootstrap actor learned about a node attempt.
+#[derive(Clone, Debug)]
+pub struct NodeIdentity {
+    pub attempt: u64,
+    pub logical_node: String,
+    /// Transport-facing node key (e.g. iroh node id, hex); opaque here.
+    pub key_hex: String,
+    /// Advertised transport address of the node (format owned by the
+    /// collector implementation); opaque here.
+    pub transport_addr: String,
+}
+
+/// Standard events every bootstrap actor reports upward. These are the
+/// supervision-facing interface: identical for every bootstrap kind.
+#[derive(Clone, Debug)]
+pub enum BootstrapEvent {
+    /// The node process came up and joined: identity known, telemetry
+    /// collectible. Reported at most once per attempt.
+    Bootstrapped(NodeIdentity),
+    /// The attempt failed before becoming ready (spawn failure, early exit,
+    /// terminal transport error). The reconciler retries with a fresh lease.
+    Failed { attempt: u64, reason: String },
+    /// A previously-bootstrapped node process exited (death, not bootstrap
+    /// failure). The reconciler replaces it.
+    Exited { attempt: u64, reason: String },
+}
+
+/// How a report leaves the actor: an app-supplied closure that routes the
+/// event into the supervision's message enum.
+pub type BootstrapReporter = Arc<dyn Fn(BootstrapEvent) + Send + Sync>;
+
+/// Standard commands the supervision sends to any bootstrap actor.
+#[derive(Clone, Debug)]
+pub enum BootstrapMsg {
+    /// Begin the attempt: launch the node process.
+    Start,
+    /// Terminate the node process. Idempotent; `kill_after` escalates.
+    Stop { kill_after: Option<Duration> },
+    /// Internal: self-driven state probe. Sending it externally is harmless.
+    Probe,
+    /// The node announced itself over the control plane: its transport
+    /// identity facts arrived from the wire (key + advertised address),
+    /// routed here by the app's announce plumbing. First delivery while
+    /// bootstrapping completes the attempt; later deliveries are heartbeat
+    /// duplicates and are ignored.
+    Announce(NodeIdentity),
+}
+
+#[derive(Clone, Debug)]
+/// Neutral description of one node launch, independent of bootstrap kind.
+pub struct NodeLaunchSpec {
+    /// Bootstrap kind; selects the logic from the registry.
+    pub kind: String,
+    pub attempt: u64,
+    pub logical_node: String,
+    /// Semantic argv the *node* runs (before any transport encoding).
+    pub argv: Vec<String>,
+    pub env: Vec<(String, String)>,
+    pub workdir: Option<std::path::PathBuf>,
+    pub label: Option<String>,
+}
+
+/// One raw probe observation from a logic implementation. The actor applies
+/// state-machine rules (once-only reporting, phase filtering); logics stay
+/// dumb readers of their foreign process.
+pub enum LogicProbe {
+    /// Still coming up; nothing to report.
+    Pending,
+    /// Node process is up and joined.
+    Bootstrapped(NodeIdentity),
+    /// Attempt failed while bootstrapping.
+    Failed(String),
+    /// Node process exited after having reported bootstrapped.
+    Exited(String),
+}
+
+/// Target-specific bootstrap behavior. Implementations live in application
+/// crates and may capture app state (registries, join checks, key files).
+pub trait BootstrapLogic: Send {
+    /// Launch the node process. Called once, in actor context, with the
+    /// owning actor's address (`owner`) so the logic can register itself in
+    /// app-side registries. An error is a terminal attempt failure.
+    fn start(
+        &mut self,
+        ctx: &Ctx,
+        owner: ActorAddress,
+        sender: &ExternalSender,
+    ) -> Result<(), String>;
+    /// Non-blocking state probe (key-file reads, registry lookups, exit
+    /// polls). Never blocks; called every probe period.
+    fn probe(&mut self, now: SystemTime) -> LogicProbe;
+    /// Terminate the node process (best effort, idempotent).
+    fn terminate(&mut self, sender: &ExternalSender, kill_after: Option<Duration>);
+}
+
+/// Telemetry collection hook the actor fires once per attempt, right after
+/// `Bootstrapped` is reported. Implementations dial the node and pull.
+pub trait NodeTelemetryCollector: Send + Sync {
+    fn collect(&self, identity: &NodeIdentity);
+}
+
+/// Creates a [`BootstrapLogic`] for one launch spec.
+pub type BootstrapFactory =
+    Arc<dyn Fn(&NodeLaunchSpec) -> Result<Box<dyn BootstrapLogic>, String> + Send + Sync>;
+
+/// Kind-keyed logic registry: adding bootstrap type #N is a new factory
+/// registration; the supervision never changes.
+#[derive(Default)]
+pub struct BootstrapRegistry {
+    factories: BTreeMap<String, BootstrapFactory>,
+}
+
+impl BootstrapRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn register(&mut self, kind: &str, factory: BootstrapFactory) {
+        self.factories.insert(kind.to_owned(), factory);
+    }
+
+    /// Resolve the logic named by `spec.kind`.
+    pub fn create(&self, spec: &NodeLaunchSpec) -> Result<Box<dyn BootstrapLogic>, String> {
+        self.factories
+            .get(&spec.kind)
+            .ok_or_else(|| format!("no bootstrap logic registered for kind '{}'", spec.kind))?
+            .clone()(spec)
+    }
+}
+
+/// Universal per-attempt lifecycle state.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Phase {
+    /// Launched, no readiness signal yet.
+    Bootstrapping,
+    /// Node joined and reported bootstrapped.
+    Ready,
+    /// Termination requested, waiting for the exit observation.
+    Stopping,
+    /// Terminal: exit observed (or start failed). No further reports.
+    Dead,
+}
+
+/// Configuration for one spawned bootstrap actor.
+pub struct BootstrapConfig {
+    pub spec: NodeLaunchSpec,
+    /// Routes events into the supervision's message enum.
+    pub reporter: BootstrapReporter,
+    /// Sender the logic uses for process commands.
+    pub sender: ExternalSender,
+    /// Optional telemetry collection hook, fired once after `Bootstrapped`.
+    pub collector: Option<Arc<dyn NodeTelemetryCollector>>,
+    pub probe_period: Duration,
+}
+
+/// The generic, transport-agnostic bootstrap actor. See the module docs.
+pub struct BootstrapActor {
+    attempt: u64,
+    logic: Box<dyn BootstrapLogic>,
+    reporter: BootstrapReporter,
+    sender: ExternalSender,
+    collector: Option<Arc<dyn NodeTelemetryCollector>>,
+    phase: Phase,
+    /// `Bootstrapped` reported (and collector fired) exactly once.
+    announced: bool,
+    /// Terminal report (Failed/Exited) emitted exactly once.
+    closed: bool,
+}
+
+impl BootstrapActor {
+    pub fn new(logic: Box<dyn BootstrapLogic>, config: BootstrapConfig) -> Self {
+        Self {
+            attempt: config.spec.attempt,
+            logic,
+            reporter: config.reporter,
+            sender: config.sender,
+            collector: config.collector,
+            phase: Phase::Bootstrapping,
+            announced: false,
+            closed: false,
+        }
+    }
+
+    fn report(&self, event: BootstrapEvent) {
+        (self.reporter)(event);
+    }
+
+    fn fail(&mut self, reason: String) {
+        if self.closed {
+            return;
+        }
+        self.closed = true;
+        self.phase = Phase::Dead;
+        self.report(BootstrapEvent::Failed {
+            attempt: self.attempt,
+            reason,
+        });
+    }
+
+    fn exit(&mut self, reason: String) {
+        if self.closed {
+            return;
+        }
+        self.closed = true;
+        self.phase = Phase::Dead;
+        self.report(BootstrapEvent::Exited {
+            attempt: self.attempt,
+            reason,
+        });
+    }
+
+    fn probe(&mut self, now: SystemTime) {
+        match self.phase {
+            Phase::Dead => {}
+            Phase::Bootstrapping => match self.logic.probe(now) {
+                LogicProbe::Pending => {}
+                LogicProbe::Bootstrapped(identity) => self.become_ready(identity),
+                LogicProbe::Failed(reason) => self.fail(reason),
+                // An exit before the join signal is a failed attempt.
+                LogicProbe::Exited(reason) => self.fail(format!("node process exited: {reason}")),
+            },
+            Phase::Ready | Phase::Stopping => match self.logic.probe(now) {
+                LogicProbe::Exited(reason) => {
+                    if self.phase == Phase::Stopping && !self.announced {
+                        // Stopped before it ever joined: failed attempt.
+                        self.fail(format!("stopped before join: {reason}"));
+                    } else {
+                        self.exit(reason);
+                    }
+                }
+                _ => {}
+            },
+        }
+    }
+
+    /// First readiness signal for the attempt (probe or wire announce):
+    /// fire the collector and report `Bootstrapped` exactly once.
+    fn become_ready(&mut self, identity: NodeIdentity) {
+        if self.announced {
+            return;
+        }
+        self.announced = true;
+        self.phase = Phase::Ready;
+        if let Some(collector) = &self.collector {
+            collector.collect(&identity);
+        }
+        self.report(BootstrapEvent::Bootstrapped(identity));
+    }
+
+    /// Apply a wire announce: first delivery for this attempt completes the
+    /// bootstrap; anything else (mismatched attempt, heartbeat duplicates,
+    /// terminal phases) is a drop.
+    fn apply_announce(&mut self, identity: NodeIdentity) {
+        if identity.attempt != self.attempt {
+            return; // misrouted: drop rather than misreport
+        }
+        if self.phase == Phase::Bootstrapping {
+            self.become_ready(identity);
+        }
+    }
+}
+
+impl ActorInterface for BootstrapActor {
+    type Incoming = BootstrapMsg;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: BootstrapMsg) {
+        match msg {
+            BootstrapMsg::Start => {
+                if self.phase != Phase::Bootstrapping {
+                    return; // Restart of a started/stopped attempt: ignore.
+                }
+                if let Err(reason) = self.logic.start(ctx, ctx.self_addr(), &self.sender) {
+                    self.fail(format!("launch failed: {reason}"));
+                }
+            }
+            BootstrapMsg::Stop { kill_after } => {
+                if self.phase == Phase::Dead {
+                    return;
+                }
+                self.phase = Phase::Stopping;
+                self.logic.terminate(&self.sender, kill_after);
+            }
+            BootstrapMsg::Probe => {
+                self.probe(SystemTime::now());
+                if self.phase == Phase::Dead {
+                    ctx.stop_self();
+                }
+            }
+            BootstrapMsg::Announce(identity) => self.apply_announce(identity),
+        }
+    }
+}
+
+/// Spawn one bootstrap actor with a self-driven probe interval: the actor,
+/// not the supervision tick, owns its progression. `Start` is sent after the
+/// interval is installed so the logic never races its own probes.
+pub fn spawn_bootstrap_actor(
+    ctx: &Ctx,
+    engine: &EngineHandle,
+    logic: Box<dyn BootstrapLogic>,
+    config: BootstrapConfig,
+) -> Result<ActorAddress, String> {
+    let sender = config.sender.clone();
+    let start_sender = sender.clone();
+    let period = config.probe_period;
+    let actor = ctx
+        .spawn(BootstrapActor::new(logic, config))
+        .map_err(|error| format!("spawn bootstrap actor: {error}"))?;
+    let probe_engine = engine.clone();
+    engine.spawn(async move {
+        let mut interval = probe_engine.interval(period);
+        loop {
+            (&mut interval).await;
+            if sender.send_to(actor, BootstrapMsg::Probe).is_err() {
+                return;
+            }
+        }
+    });
+    let _ = start_sender.send_to(actor, BootstrapMsg::Start);
+    Ok(actor)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use parking_lot::Mutex;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    struct ScriptedLogic {
+        probes: Mutex<Vec<LogicProbe>>,
+        started: AtomicUsize,
+        terminated: AtomicUsize,
+    }
+
+    impl ScriptedLogic {
+        fn new(probes: Vec<LogicProbe>) -> Self {
+            Self {
+                probes: Mutex::new(probes),
+                started: AtomicUsize::new(0),
+                terminated: AtomicUsize::new(0),
+            }
+        }
+    }
+
+    impl BootstrapLogic for ScriptedLogic {
+        fn start(
+            &mut self,
+            _ctx: &Ctx,
+            _owner: ActorAddress,
+            _sender: &ExternalSender,
+        ) -> Result<(), String> {
+            self.started.fetch_add(1, Ordering::SeqCst);
+            Ok(())
+        }
+        fn probe(&mut self, _now: SystemTime) -> LogicProbe {
+            self.probes.lock().remove(0)
+        }
+        fn terminate(&mut self, _sender: &ExternalSender, _kill_after: Option<Duration>) {
+            self.terminated.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    fn identity(attempt: u64) -> NodeIdentity {
+        NodeIdentity {
+            attempt,
+            logical_node: format!("node-{attempt}"),
+            key_hex: format!("key{attempt}"),
+            transport_addr: format!("addr{attempt}"),
+        }
+    }
+
+    #[test]
+    fn registry_resolves_by_kind() {
+        let mut registry = BootstrapRegistry::new();
+        registry.register(
+            "process",
+            Arc::new(|_spec| Ok(Box::new(ScriptedLogic::new(vec![])) as Box<dyn BootstrapLogic>)),
+        );
+        let spec = NodeLaunchSpec {
+            kind: "process".to_owned(),
+            attempt: 1,
+            logical_node: "node-1".to_owned(),
+            argv: vec![],
+            env: vec![],
+            workdir: None,
+            label: None,
+        };
+        assert!(registry.create(&spec).is_ok());
+        let unknown = NodeLaunchSpec {
+            kind: "nope".to_owned(),
+            ..spec
+        };
+        assert!(registry.create(&unknown).is_err());
+    }
+    #[test]
+    fn reporter_contract_carries_identity() {
+        let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+        let reporter: BootstrapReporter = {
+            let events = events.clone();
+            Arc::new(move |event| {
+                let name = match event {
+                    BootstrapEvent::Bootstrapped(_) => "bootstrapped",
+                    BootstrapEvent::Failed { .. } => "failed",
+                    BootstrapEvent::Exited { .. } => "exited",
+                };
+                events.lock().push(name.to_owned());
+            })
+        };
+        reporter(BootstrapEvent::Exited {
+            attempt: 7,
+            reason: "test".to_owned(),
+        });
+        assert_eq!(events.lock().as_slice(), ["exited"]);
+        let id = identity(7);
+        assert_eq!(id.logical_node, "node-7");
+    }
+
+    #[test]
+    fn announce_completes_attempt_exactly_once() {
+        let events: Arc<Mutex<Vec<BootstrapEvent>>> = Arc::new(Mutex::new(Vec::new()));
+        let mut actor = announce_actor(7, events.clone(), vec![]);
+
+        actor.apply_announce(identity(7));
+        assert_eq!(actor.phase, Phase::Ready);
+        // Heartbeat duplicates: ignored.
+        actor.apply_announce(identity(7));
+        actor.apply_announce(identity(7));
+        let logged = events.lock();
+        assert_eq!(logged.len(), 1);
+        assert!(matches!(&logged[0], BootstrapEvent::Bootstrapped(id) if id.key_hex == "key7"));
+    }
+
+    #[test]
+    fn announce_with_mismatched_attempt_is_dropped() {
+        let events: Arc<Mutex<Vec<BootstrapEvent>>> = Arc::new(Mutex::new(Vec::new()));
+        let mut actor = announce_actor(7, events.clone(), vec![]);
+
+        actor.apply_announce(identity(8));
+        assert_eq!(actor.phase, Phase::Bootstrapping);
+        assert!(events.lock().is_empty());
+    }
+
+    #[test]
+    fn announce_after_terminal_phase_is_dropped() {
+        let events: Arc<Mutex<Vec<BootstrapEvent>>> = Arc::new(Mutex::new(Vec::new()));
+        let mut actor = announce_actor(7, events.clone(), vec![]);
+
+        actor.apply_announce(identity(7));
+        actor.exit("killed".to_owned());
+        // A stale container from the dead attempt still announcing.
+        actor.apply_announce(identity(7));
+        let logged = events.lock();
+        assert_eq!(logged.len(), 2);
+        assert!(matches!(&logged[1], BootstrapEvent::Exited { attempt, .. } if *attempt == 7));
+    }
+
+    #[test]
+    fn announce_after_probe_bootstrapped_does_not_double_report() {
+        let events: Arc<Mutex<Vec<BootstrapEvent>>> = Arc::new(Mutex::new(Vec::new()));
+        // Probe path reports Bootstrapped first…
+        let mut actor = announce_actor(
+            9,
+            events.clone(),
+            vec![LogicProbe::Bootstrapped(identity(9))],
+        );
+        actor.probe(SystemTime::now());
+        // …then the announce for the same attempt arrives late.
+        actor.apply_announce(identity(9));
+        let logged = events.lock();
+        assert_eq!(logged.len(), 1);
+    }
+
+    /// An actor over a scripted logic whose events land in `events`.
+    fn announce_actor(
+        attempt: u64,
+        events: Arc<Mutex<Vec<BootstrapEvent>>>,
+        probes: Vec<LogicProbe>,
+    ) -> BootstrapActor {
+        let reporter: BootstrapReporter = {
+            let events = events.clone();
+            Arc::new(move |event| events.lock().push(event))
+        };
+        let sender = swactor::runtime::RuntimeParts::new(swactor::config::RuntimeConfig::default())
+            .runtime()
+            .create_sender();
+        let config = BootstrapConfig {
+            spec: NodeLaunchSpec {
+                kind: "process".to_owned(),
+                attempt,
+                logical_node: format!("node-{attempt}"),
+                argv: vec![],
+                env: vec![],
+                workdir: None,
+                label: None,
+            },
+            reporter,
+            sender,
+            collector: None,
+            probe_period: DEFAULT_PROBE_PERIOD,
+        };
+        BootstrapActor::new(Box::new(ScriptedLogic::new(probes)), config)
+    }
+}

crates/provisioning/src/lib.rs

@@ -5,10 +5,12 @@
 //! provider adapters live in application crates and implement the executor
 //! backend contract.
 
+pub mod bootstrap;
 pub mod executor;
 pub mod node;
 pub mod plugin;
 pub mod reconciler;
+pub use bootstrap::*;
 
 pub use executor::*;
 pub use node::*;

xtask/src/main.rs

@@ -266,10 +266,12 @@ 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]
+  provisioning-reconciler-demo [--port n] [--nodes n] [--docker]
                       Run the visual E2E provisioning reconciler sanity demo
                       (supervisor + dashboard on localhost, node children
-                      join over iroh). Ctrl-C tears down.
+                      join over iroh; --docker launches nodes as scratch
+                      containers on a per-run bridge network). Ctrl-C tears
+                      down and sweeps.
   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`."
@@ -364,9 +366,12 @@ fn run_myelin_chat(args: Vec<String>) -> ExitCode {
                 "args": MYELIN_CHAT_CARGO_RUN_ARGS,
             }),
         );
-        if let Err(error) =
+        if let Err(error) = append_synthetic_benchmark_frame(
-            append_synthetic_benchmark_frame(path, "xtask-myelin-chat", "myelin.xtask.benchmark", event)
+            path,
-        {
+            "xtask-myelin-chat",
+            "myelin.xtask.benchmark",
+            event,
+        ) {
             eprintln!("Failed to write myelin-chat benchmark frame: {error}");
             return ExitCode::from(1);
         }
@@ -381,9 +386,12 @@ fn run_myelin_chat(args: Vec<String>) -> ExitCode {
             Err(error) => ("failed", json!({"error": error.to_string()})),
         };
         let event = xtask_myelin_chat_benchmark_event(run_id, status, detail);
-        if let Err(error) =
+        if let Err(error) = append_synthetic_benchmark_frame(
-            append_synthetic_benchmark_frame(path, "xtask-myelin-chat", "myelin.xtask.benchmark", event)
+            path,
-        {
+            "xtask-myelin-chat",
+            "myelin.xtask.benchmark",
+            event,
+        ) {
             eprintln!("Failed to write myelin-chat benchmark frame: {error}");
             return ExitCode::from(1);
         }
@@ -1098,7 +1106,9 @@ fn terminate_myelin_chat_child(child: &mut Child) -> Result<ExitStatus, String>
                 Ok(Some(status)) => return Ok(status),
                 Ok(None) => thread::sleep(Duration::from_millis(MYELIN_CHAT_CHECK_POLL_MS)),
                 Err(error) => {
-                    return Err(format!("myelin-chat-check: poll child after SIGTERM: {error}"));
+                    return Err(format!(
+                        "myelin-chat-check: poll child after SIGTERM: {error}"
+                    ));
                 }
             }
         }
@@ -1347,7 +1357,9 @@ fn find_stdout_marker(
     stdout[start..]
         .find(marker)
         .map(|offset| start + offset)
-        .ok_or_else(|| format!("myelin-chat-check: missing {label} marker for prompt cycle {cycle}"))
+        .ok_or_else(|| {
+            format!("myelin-chat-check: missing {label} marker for prompt cycle {cycle}")
+        })
 }
 
 #[derive(Clone)]
@@ -2196,12 +2208,22 @@ fn validate_benchmark_observability(
                     validation.edges_with_consumer.insert(edge_id);
                 }
             }
-            ("myelin.chat.prompt", Some("ChatProgress"), Some("prompt_submitted"), Some("ready")) => {
+            (
+                "myelin.chat.prompt",
+                Some("ChatProgress"),
+                Some("prompt_submitted"),
+                Some("ready"),
+            ) => {
                 if let Some(request_id) = benchmark_request_id(&record.event) {
                     validation.requests_started.insert(request_id);
                 }
             }
-            ("myelin.chat.prompt", Some("ChatProgress"), Some("request_completed"), Some("ready")) => {
+            (
+                "myelin.chat.prompt",
+                Some("ChatProgress"),
+                Some("request_completed"),
+                Some("ready"),
+            ) => {
                 if let Some(request_id) = benchmark_request_id(&record.event) {
                     validation.requests_completed.insert(request_id);
                 }
@@ -3945,7 +3967,10 @@ fn pipeline_stage_index(event: &Value) -> Option<u64> {
         .or_else(|| event_u64(event, "role_id").and_then(|role| role.checked_sub(1)))
 }
 
-fn benchmark_invariants_json(facts: &DumpLogFacts, scenario: MyelinChatCheckScenario) -> Vec<Value> {
+fn benchmark_invariants_json(
+    facts: &DumpLogFacts,
+    scenario: MyelinChatCheckScenario,
+) -> Vec<Value> {
     let mut invariants = vec![
         invariant_json("chat_config_ready", facts.chat_config_ready),
         invariant_json("prepare_runtime_ready", facts.prepare_runtime_ready),
@@ -4494,7 +4519,9 @@ fn record_gpu_dump_log_event(channel: &str, event: &Value, facts: &mut DumpLogFa
         {
             facts.gpu_probe_ready = true;
         }
-        ("myelin.worker.initialize", Some("WorkerReady")) if worker_ready_backend_is_cuda(event) => {
+        ("myelin.worker.initialize", Some("WorkerReady"))
+            if worker_ready_backend_is_cuda(event) =>
+        {
             facts.gpu_worker_ready = true;
         }
         ("myelin.worker.prompt", Some("DecodeStarted"))
@@ -4862,7 +4889,10 @@ mod tests {
 
         let baseline =
             MyelinChatCheckInvocation::parse_args(Vec::new()).expect("default scenario parses");
-        assert_eq!(baseline.scenario(), MyelinChatCheckScenario::ProcessBaseline);
+        assert_eq!(
+            baseline.scenario(),
+            MyelinChatCheckScenario::ProcessBaseline
+        );
         assert_eq!(
             baseline.myelin_chat_args(42, dump_log),
             strings(&[
@@ -4904,8 +4934,9 @@ mod tests {
             ])
         );
 
-        let multinode_docker = MyelinChatCheckInvocation::parse_args(strings(&["--multinode-docker"]))
+        let multinode_docker =
-            .expect("multinode docker parses");
+            MyelinChatCheckInvocation::parse_args(strings(&["--multinode-docker"]))
+                .expect("multinode docker parses");
         assert_eq!(
             multinode_docker.scenario(),
             MyelinChatCheckScenario::MultinodeDocker
@@ -4959,9 +4990,12 @@ mod tests {
                 "--dump-logs=/tmp/myelin-chat-check.ndjson",
             ])
         );
-        let vastai_parallel =
+        let vastai_parallel = MyelinChatCheckInvocation::parse_args(strings(&[
-            MyelinChatCheckInvocation::parse_args(strings(&["--vastai", "--pipeline-parallel", "4"]))
+            "--vastai",
-                .expect("vastai pipeline-parallel parses");
+            "--pipeline-parallel",
+            "4",
+        ]))
+        .expect("vastai pipeline-parallel parses");
         assert_eq!(
             vastai_parallel.myelin_chat_args(42, dump_log),
             strings(&[
@@ -5597,7 +5631,10 @@ mod tests {
         include_cpu_fallback: bool,
     ) -> Vec<(&'static str, Value)> {
         let mut events = vec![
-            ("myelin.chat.lifecycle", chat_span("config", "ready", 1_000, 0)),
+            (
+                "myelin.chat.lifecycle",
+                chat_span("config", "ready", 1_000, 0),
+            ),
             (
                 "myelin.chat.benchmark",
                 stamped(
@@ -6421,8 +6458,9 @@ mod tests {
     fn benchmark_observability_report_requires_granular_decode_events() {
         let events = parse_synthetic_events("missing-first-token", benchmark_report_events(false));
 
-        let error = build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline)
+        let error =
-            .expect_err("missing first token should fail");
+            build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline)
+                .expect_err("missing first token should fail");
 
         assert!(
             error.starts_with("myelin-chat-check: missing benchmark event "),
@@ -6439,8 +6477,9 @@ mod tests {
         let events =
             parse_synthetic_events("pipeline-report", pipeline_benchmark_report_events(true));
 
-        let report = build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline)
+        let report =
-            .expect("pipeline report builds");
+            build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline)
+                .expect("pipeline report builds");
 
         assert!(
             report

xtask/src/provisioning_demo/bootstrap.rs

@@ -0,0 +1,112 @@
+//! Demo bootstrap logic #1: local OS foreign process.
+//!
+//! Implements [`BootstrapLogic`] for `"process"`-kind specs: the node is a
+//! re-exec'd child of this binary supervised by a `swactor-process` actor
+//! (stdio null). Readiness arrives over the control plane: the child's node
+//! role joins the supervisor's iroh endpoint and announces itself, and the
+//! supervisor's announce relay delivers [`BootstrapMsg::Announce`] to the
+//! owning bootstrap actor. This logic only owns process lifecycle (spawn,
+//! exit observation, termination). The docker variant sits beside it in the
+//! registry — same interface, different foreign process.
+
+use std::time::SystemTime;
+
+use swactor::actor::{ActorAddress, Ctx};
+use swactor::runtime::ExternalSender;
+use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process};
+
+use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec};
+
+use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime};
+
+/// Local-process bootstrap logic. The spawned process actor's lifecycle
+/// reports fold into the shared [`NodeManager`] registry via
+/// [`NodeRelayActor`]; probes read that registry only — readiness is the
+/// wire announce, not a probe observation.
+pub struct LocalProcessLogic {
+    spec: NodeLaunchSpec,
+    manager: NodeManager,
+    process_actor: Option<ActorAddress>,
+}
+
+impl LocalProcessLogic {
+    pub fn new(spec: NodeLaunchSpec, manager: NodeManager) -> Self {
+        Self {
+            spec,
+            manager,
+            process_actor: None,
+        }
+    }
+}
+
+impl BootstrapLogic for LocalProcessLogic {
+    fn start(
+        &mut self,
+        ctx: &Ctx,
+        owner: ActorAddress,
+        sender: &ExternalSender,
+    ) -> Result<(), String> {
+        let attempt = self.spec.attempt;
+        let relay = ctx
+            .spawn(NodeRelayActor::new(self.manager.clone(), attempt))
+            .map_err(|error| format!("spawn relay actor: {error}"))?;
+
+        let (command, args) = self
+            .spec
+            .argv
+            .split_first()
+            .map(|(head, tail)| (head.clone(), tail.to_vec()))
+            .ok_or_else(|| "process spec missing argv".to_owned())?;
+        let spec = ProcessSpec {
+            command,
+            args,
+            env: self.spec.env.clone().into_iter().collect(),
+            working_dir: self.spec.workdir.clone(),
+            label: self.spec.label.clone(),
+        };
+        let output = ProcessOutputConfig::disabled(relay);
+        let process_actor = spawn_local_process(ctx, sender, spec, output)
+            .map_err(|error| format!("spawn process actor: {error}"))?;
+        self.process_actor = Some(process_actor);
+
+        self.manager.register(NodeRuntime {
+            attempt,
+            logical_node: self.spec.logical_node.clone(),
+            bootstrap: owner,
+            pid: None,
+            exited: None,
+            spawn_failed: None,
+            last_announce_ms: None,
+        });
+        Ok(())
+    }
+
+    fn probe(&mut self, _now: SystemTime) -> LogicProbe {
+        let Some(runtime) = self.manager.get(self.spec.attempt) else {
+            // Deregistered (lease destroyed / supervisor teardown): the
+            // node is gone by definition. The relay's exit observation can
+            // lose the race against deregistration, so this is the
+            // reliable terminal report. The actor's phase machine turns it
+            // into Failed (before join) or Exited (after).
+            return LogicProbe::Exited("deregistered (lease destroyed)".to_owned());
+        };
+        if let Some(error) = &runtime.spawn_failed {
+            return LogicProbe::Failed(format!("node process failed to spawn: {error}"));
+        }
+        if let Some(status) = &runtime.exited {
+            return LogicProbe::Exited(format!("{status:?}"));
+        }
+        // Still coming up; readiness is the wire announce.
+        LogicProbe::Pending
+    }
+
+    fn terminate(&mut self, sender: &ExternalSender, kill_after: Option<std::time::Duration>) {
+        if let Some(process_actor) = self.process_actor {
+            let _ = swactor_process::send_process_command(
+                sender,
+                process_actor,
+                swactor_process::ProcessCommand::Stop { kill_after },
+            );
+        }
+    }
+}

xtask/src/provisioning_demo/docker.rs

@@ -0,0 +1,428 @@
+//! Demo bootstrap logic #2: docker container as foreign node.
+//!
+//! Implements [`BootstrapLogic`] for `"docker"`-kind specs: the node is a
+//! `scratch` container running the statically-linked xtask binary in node
+//! role, launched attached (`docker run --rm`) so the supervised `docker`
+//! CLI child's lifetime tracks the container's — its exit *is* the node
+//! exit. Readiness is the same control-plane announce as the process kind.
+//!
+//! Foreign-node masking: every container runs on a per-run user-defined
+//! bridge network, so each node gets its own bridge IP and the supervisor is
+//! reached through the bridge gateway — no localhost shortcuts, UDP
+//! hole-punching over a real (if virtual) network.
+//!
+//! Termination always goes through `docker rm -f` (force-remove): signals to
+//! the attached CLI are unreliable proxies, and a force-remove both kills
+//! the container and satisfies `--rm` cleanup. Zero wastage by construction:
+//! no volumes, no mounts, `--rm` containers, and label-filtered startup +
+//! exit sweeps that remove anything a SIGKILLed supervisor left behind.
+//! Images persist across runs (rebuilds are content-addressed by run token).
+
+use std::path::Path;
+use std::time::SystemTime;
+
+use swactor::actor::{ActorAddress, Ctx};
+use swactor::runtime::ExternalSender;
+use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process};
+
+use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec};
+
+use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime};
+
+/// Generic label present on every demo container/network (sweep key).
+pub const SWEEP_LABEL: &str = "swactor-demo";
+/// Per-run label value: only this run's resources.
+pub const RUN_LABEL: &str = "swactor-demo-run";
+/// Image repository (tagged per run token).
+pub const IMAGE_REPO: &str = "swactor-demo-node";
+/// Static-musl target the node image is built from.
+pub const IMAGE_TARGET: &str = "x86_64-unknown-linux-musl";
+
+const DOCKERFILE: &str = include_str!("docker/Dockerfile");
+
+/// Everything the docker launch style needs, produced by [`preflight`].
+#[derive(Clone)]
+pub struct DockerLaunch {
+    /// Fully-qualified image ref (`swactor-demo-node:<token>`).
+    pub image: String,
+    /// Per-run user-defined bridge network name.
+    pub network: String,
+    /// Per-run label value used by the exit sweep.
+    pub run_token: String,
+    /// Serde `iroh::EndpointAddr` of the supervisor, rebuilt to advertise
+    /// the bridge gateway address (containers cannot use the host's
+    /// localhost or LAN addrs meaningfully).
+    pub supervisor_addr_json: String,
+}
+
+/// Container name for one provision attempt (unique per attempt).
+pub fn container_name(attempt: u64) -> String {
+    format!("{SWEEP_LABEL}-node-{attempt}")
+}
+
+/// Docker bootstrap logic. Identical lifecycle shape to the local process
+/// kind — the supervised child is the attached `docker run` CLI; the
+/// container is force-removed on every terminal path (terminate, exit,
+/// spawn failure, deregistration) so it can never outlive the attempt.
+pub struct DockerProcessLogic {
+    spec: NodeLaunchSpec,
+    manager: NodeManager,
+    process_actor: Option<ActorAddress>,
+    /// The attempt's container has been force-removed (or removal was
+    /// spawned); guards idempotency across probe/terminate paths.
+    removed: bool,
+}
+
+impl DockerProcessLogic {
+    pub fn new(spec: NodeLaunchSpec, manager: NodeManager) -> Self {
+        Self {
+            spec,
+            manager,
+            process_actor: None,
+            removed: false,
+        }
+    }
+
+    /// Idempotent force-remove of this attempt's container (kills it if
+    /// running, satisfies --rm, releases the name; no-op if already gone).
+    fn force_remove(&mut self) {
+        if self.removed {
+            return;
+        }
+        self.removed = true;
+        let name = container_name(self.spec.attempt);
+        std::thread::spawn(move || {
+            let status = std::process::Command::new("docker")
+                .args(["rm", "-f", "--", &name])
+                .stdin(std::process::Stdio::null())
+                .stdout(std::process::Stdio::null())
+                .stderr(std::process::Stdio::null())
+                .status();
+            if let Err(error) = status {
+                eprintln!("demo: docker rm -f {name} failed: {error}");
+            }
+        });
+    }
+}
+
+impl BootstrapLogic for DockerProcessLogic {
+    fn start(
+        &mut self,
+        ctx: &Ctx,
+        owner: ActorAddress,
+        sender: &ExternalSender,
+    ) -> Result<(), String> {
+        let attempt = self.spec.attempt;
+        let relay = ctx
+            .spawn(NodeRelayActor::new(self.manager.clone(), attempt))
+            .map_err(|error| format!("spawn relay actor: {error}"))?;
+
+        let (command, args) = self
+            .spec
+            .argv
+            .split_first()
+            .map(|(head, tail)| (head.clone(), tail.to_vec()))
+            .ok_or_else(|| "docker spec missing argv".to_owned())?;
+        let spec = ProcessSpec {
+            command,
+            args,
+            env: self.spec.env.clone().into_iter().collect(),
+            working_dir: self.spec.workdir.clone(),
+            label: self.spec.label.clone(),
+        };
+        let output = ProcessOutputConfig::disabled(relay);
+        let process_actor = spawn_local_process(ctx, sender, spec, output)
+            .map_err(|error| format!("spawn docker process actor: {error}"))?;
+        self.process_actor = Some(process_actor);
+
+        self.manager.register(NodeRuntime {
+            attempt,
+            logical_node: self.spec.logical_node.clone(),
+            bootstrap: owner,
+            pid: None,
+            exited: None,
+            spawn_failed: None,
+            last_announce_ms: None,
+        });
+        Ok(())
+    }
+
+    fn probe(&mut self, _now: SystemTime) -> LogicProbe {
+        let Some(runtime) = self.manager.get(self.spec.attempt) else {
+            self.force_remove();
+            return LogicProbe::Exited("deregistered (lease destroyed)".to_owned());
+        };
+        if let Some(error) = &runtime.spawn_failed {
+            self.force_remove();
+            return LogicProbe::Failed(format!("docker run failed to spawn: {error}"));
+        }
+        if let Some(status) = &runtime.exited {
+            // The attached CLI's exit mirrors the container's (or is the
+            // launch failure itself); the actor's phase machine classifies
+            // Failed-vs-Exited against the announce. Either way the
+            // container must not outlive the attempt: a SIGKILLed CLI
+            // orphans a *running* container (signal-proxy never fired), so
+            // the exit path force-removes too — not just terminate().
+            self.force_remove();
+            return LogicProbe::Exited(format!("docker run {status:?}"));
+        }
+        LogicProbe::Pending
+    }
+
+    fn terminate(&mut self, sender: &ExternalSender, kill_after: Option<std::time::Duration>) {
+        // Force-remove the container: kills it regardless of signal-proxy
+        // semantics, satisfies --rm, and releases the name. The attached
+        // CLI child then exits on its own; the explicit Stop below is
+        // actor-tree hygiene (the kill_after escalation applies to the
+        // CLI, not the container).
+        self.force_remove();
+        if let Some(process_actor) = self.process_actor {
+            let _ = swactor_process::send_process_command(
+                sender,
+                process_actor,
+                swactor_process::ProcessCommand::Stop { kill_after },
+            );
+        }
+    }
+}
+
+// ─── Preflight: image + network ─────────────────────────────────────────────
+
+/// Build the node image and the per-run network: sweep stale demo resources,
+/// compile the static-musl xtask binary, `docker build` it from a staging
+/// dir, create the labeled bridge network, and resolve the gateway address
+/// the containers will dial the supervisor on.
+pub fn preflight(
+    root: &Path,
+    supervisor_pubkey: iroh::PublicKey,
+    port: u16,
+) -> Result<DockerLaunch, String> {
+    docker_version()?;
+    let run_token = format!("{}-{}", std::process::id(), unix_ms());
+    sweep_stale();
+    let image = format!("{IMAGE_REPO}:{run_token}");
+    build_image(root, &image)?;
+    let network = format!("{SWEEP_LABEL}-net-{run_token}");
+    docker_ok(
+        &[
+            "network",
+            "create",
+            "--label",
+            &format!("{SWEEP_LABEL}=1"),
+            "--label",
+            &format!("{RUN_LABEL}={run_token}"),
+            "--",
+            &network,
+        ],
+        "create demo network",
+    )?;
+    let gateway = docker_output(
+        &[
+            "network",
+            "inspect",
+            "--format",
+            "{{(index .IPAM.Config 0).Gateway}}",
+            "--",
+            &network,
+        ],
+        "read demo network gateway",
+    )?;
+    let gateway_ip: std::net::IpAddr = gateway
+        .trim()
+        .parse()
+        .map_err(|error| format!("network gateway {gateway:?}: {error}"))?;
+    let supervisor_addr = iroh::EndpointAddr::new(supervisor_pubkey)
+        .with_ip_addr(std::net::SocketAddr::new(gateway_ip, port));
+    let supervisor_addr_json = serde_json::to_string(&supervisor_addr)
+        .map_err(|error| format!("serialize supervisor addr: {error}"))?;
+    Ok(DockerLaunch {
+        image,
+        network,
+        run_token,
+        supervisor_addr_json,
+    })
+}
+
+/// Remove every container and network carrying the generic demo label. All
+/// such resources found here are stale: this runs before the current run
+/// creates anything, so anything matched is a leftover (e.g. of a SIGKILLed
+/// supervisor). One demo instance at a time.
+fn sweep_stale() {
+    let containers = docker_output(
+        &["ps", "-aq", "--filter", &format!("label={SWEEP_LABEL}=1")],
+        "list stale demo containers",
+    )
+    .unwrap_or_default();
+    let ids: Vec<String> = containers
+        .lines()
+        .map(str::trim)
+        .filter(|line| !line.is_empty())
+        .map(str::to_owned)
+        .collect();
+    if !ids.is_empty() {
+        let mut args = vec!["rm".to_owned(), "-f".to_owned()];
+        args.extend(ids.iter().map(|id| id.to_string()));
+        let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+        let _ = docker_ok(&refs, "remove stale demo containers");
+    }
+    let networks = docker_output(
+        &[
+            "network",
+            "ls",
+            "-q",
+            "--filter",
+            &format!("label={SWEEP_LABEL}=1"),
+        ],
+        "list stale demo networks",
+    )
+    .unwrap_or_default();
+    let nets: Vec<String> = networks
+        .lines()
+        .map(str::trim)
+        .filter(|line| !line.is_empty())
+        .map(str::to_owned)
+        .collect();
+    if !nets.is_empty() {
+        let mut args = vec!["network".to_owned(), "rm".to_owned()];
+        args.extend(nets.iter().map(|id| id.to_string()));
+        let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+        let _ = docker_ok(&refs, "remove stale demo networks");
+    }
+}
+
+/// Exit-path sweep: remove this run's containers (by run token) and its
+/// network. The image persists (per-run tokens make old images trivially
+/// identifiable; they are tiny scratch images).
+pub fn sweep_run(launch: &DockerLaunch) {
+    let containers = docker_output(
+        &[
+            "ps",
+            "-aq",
+            "--filter",
+            &format!("label={RUN_LABEL}={}", launch.run_token),
+        ],
+        "list run containers",
+    )
+    .unwrap_or_default();
+    let ids: Vec<&str> = containers
+        .lines()
+        .map(str::trim)
+        .filter(|l| !l.is_empty())
+        .collect();
+    if !ids.is_empty() {
+        let mut args = vec!["rm".to_owned(), "-f".to_owned()];
+        args.extend(ids.iter().map(|id| id.to_string()));
+        let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+        let _ = docker_ok(&refs, "remove run containers");
+    }
+    let _ = docker_ok(
+        &["network", "rm", "--", &launch.network],
+        "remove run network",
+    );
+}
+
+/// Build the scratch image: stage the static binary + Dockerfile in a temp
+/// dir (keeps the build context to one file), then `docker build`.
+fn build_image(root: &Path, image: &str) -> Result<(), String> {
+    println!("provisioning-reconciler-demo --docker: building static node binary…");
+    let bin = root
+        .join("target")
+        .join(IMAGE_TARGET)
+        .join("release")
+        .join("xtask");
+    // Build when the static binary is missing; DEMO_DOCKER_REBUILD=1 forces
+    // a rebuild (e.g. after source changes).
+    let force_rebuild = std::env::var("DEMO_DOCKER_REBUILD").as_deref() == Ok("1");
+    if force_rebuild || !bin.exists() {
+        run_cargo_build(root)?;
+    }
+    if !bin.exists() {
+        return Err(format!(
+            "node binary missing after build: {}",
+            bin.display()
+        ));
+    }
+    let staging = std::env::temp_dir().join(format!("{SWEEP_LABEL}-image-{}", unix_ms()));
+    std::fs::create_dir_all(&staging).map_err(|e| format!("staging dir: {e}"))?;
+    let result = (|| {
+        std::fs::copy(&bin, staging.join("xtask")).map_err(|e| format!("stage binary: {e}"))?;
+        std::fs::write(staging.join("Dockerfile"), DOCKERFILE)
+            .map_err(|e| format!("stage Dockerfile: {e}"))?;
+        docker_ok(
+            &["build", "-q", "-t", image, "--", &staging.to_string_lossy()],
+            "build demo node image",
+        )
+    })();
+    let _ = std::fs::remove_dir_all(&staging);
+    result.map(|_| {
+        println!("provisioning-reconciler-demo --docker: image {image} ready");
+    })
+}
+
+fn run_cargo_build(root: &Path) -> Result<(), String> {
+    let output = std::process::Command::new("cargo")
+        .args([
+            "build",
+            "--release",
+            "--target",
+            IMAGE_TARGET,
+            "--package",
+            "xtask",
+        ])
+        .current_dir(root)
+        .stdin(std::process::Stdio::null())
+        .output()
+        .map_err(|e| format!("run cargo: {e}"))?;
+    if !output.status.success() {
+        let mut stderr = String::from_utf8_lossy(&output.stderr).into_owned();
+        if stderr.len() > 4000 {
+            stderr.truncate(4000);
+        }
+        return Err(format!("cargo build for {IMAGE_TARGET} failed:\n{stderr}"));
+    }
+    Ok(())
+}
+
+/// Docker CLI + daemon reachable?
+fn docker_version() -> Result<(), String> {
+    docker_output(
+        &["version", "--format", "{{.Server.Version}}"],
+        "docker daemon",
+    )
+    .map(|_| ())
+}
+
+fn docker_ok(args: &[&str], label: &str) -> Result<(), String> {
+    let output = docker_raw(args)?;
+    if output.status.success() {
+        Ok(())
+    } else {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        Err(format!("{label} failed: {stderr}"))
+    }
+}
+
+fn docker_output(args: &[&str], label: &str) -> Result<String, String> {
+    let output = docker_raw(args)?;
+    if output.status.success() {
+        Ok(String::from_utf8_lossy(&output.stdout).into_owned())
+    } else {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        Err(format!("{label} failed: {stderr}"))
+    }
+}
+
+fn docker_raw(args: &[&str]) -> Result<std::process::Output, String> {
+    std::process::Command::new("docker")
+        .args(args)
+        .stdin(std::process::Stdio::null())
+        .output()
+        .map_err(|e| format!("run docker {args:?}: {e}"))
+}
+
+fn unix_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map(|duration| duration.as_millis() as u64)
+        .unwrap_or(0)
+}

xtask/src/provisioning_demo/docker/Dockerfile

@@ -0,0 +1,9 @@
+# Demo node image: the static-musl xtask binary and nothing else.
+# Staged into a one-file build context by
+# xtask/src/provisioning_demo/docker.rs (`build_image`); never built from
+# the repo root. The binary is fully static (musl + ring), so `scratch`
+# needs no libc, no CA bundle (relay disabled; direct iroh only), and no
+# shell — kill signals reach PID 1 = the node role directly.
+FROM scratch
+COPY xtask /xtask
+ENTRYPOINT ["/xtask"]

xtask/src/provisioning_demo/feed.rs

@@ -4,36 +4,30 @@
 //! 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),
+//! real world back in (wire announces, child exits), emits
-//! emits `prov.reconciler.*` telemetry, and drains every telemetry endpoint
+//! `prov.reconciler.*` telemetry, and drains every telemetry endpoint into
-//! into the dashboard.
+//! the dashboard.
 
 use std::collections::BTreeMap;
 use std::time::{Duration, SystemTime};
 
+use provisioning::executor::{
+    BlockingEffectSpawner, BlockingEffectWork, ExecutorOperationStatus, IdempotentEffectExecutor,
+};
+use provisioning::node::{
+    BootstrapObservation, BootstrapStage, NodeGroupId, NodeStage, RoleId, RunId, SwactorId,
+};
+use provisioning::reconciler::{ClusterDriver, EffectExecutor};
+use provisioning::reconciler::{
+    ClusterShape, NodeObservation, OperationOutcome, PlannedEffect, RetryPolicy,
+};
 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,
+    DemoBackend, NodeManager, NodeTelemetry, register_node_channels, unix_ms,
-    NodeTelemetry,
 };
 
 /// Supervisor telemetry: channels + name resolution for the dashboard path.
@@ -136,12 +130,32 @@ pub enum SupervisorMsg {
     Tick,
     Control(dashboard::control::ControlCommand),
     Spawn(crate::provisioning_demo::provider::SpawnNodeRequest),
+    /// Event from a per-node bootstrap actor.
+    Bootstrap(provisioning::BootstrapEvent),
+    /// A remote node's telemetry pull stream registered its header (stream
+    /// descriptor + channel catalog), associated with its logical node and
+    /// provision attempt. Frames are fused onto the logical node's stream.
+    NodeStream {
+        header: iroh_driver::TelemetryQuicHeader,
+        logical_node: String,
+        attempt: u64,
+    },
     /// Drain the cluster: desired → empty, stop every child, flag when done.
     Shutdown {
         drained: std::sync::Arc<std::sync::atomic::AtomicBool>,
     },
 }
 
+/// Metadata for one remote node telemetry pull, keyed by the node's hex
+/// transport key: channel catalog + which logical node stream its frames
+/// are fused onto.
+#[derive(Clone, Default)]
+struct RemoteStreamMeta {
+    logical_node: String,
+    attempt: u64,
+    channels: BTreeMap<telemetry::ChannelId, String>,
+}
+
 /// Per-node telemetry handle kept while the attempt is live.
 struct NodeStreams {
     telemetry: NodeTelemetry,
@@ -158,6 +172,18 @@ pub struct SupervisorActor {
     pub events_channel: telemetry::ChannelId,
     pub snapshot_channel: telemetry::ChannelId,
     pub sender: swactor::runtime::ExternalSender,
+    /// Bootstrap kind registry (spec.kind → logic).
+    registry: provisioning::BootstrapRegistry,
+    /// Pull collector fired by bootstrap actors on Bootstrapped.
+    collector: std::sync::Arc<dyn provisioning::NodeTelemetryCollector>,
+    /// Engine handle for bootstrap actor probe intervals.
+    engine: EngineHandle,
+    /// Subscription draining the remote-stream fanout into the dashboard.
+    remote_sub: telemetry::TelemetrySubscription,
+    remote_streams: BTreeMap<String, RemoteStreamMeta>,
+    /// Bootstrap reports parked until the reconciler opens the attempt's
+    /// bootstrap session (early joins / failures).
+    pending_reports: BTreeMap<u64, provisioning::BootstrapEvent>,
     /// Desired shape slots: singleton groups, one per logical node.
     slots: Vec<String>,
     slot_seq: u64,
@@ -167,7 +193,7 @@ pub struct SupervisorActor {
     last_stages: BTreeMap<String, (NodeStage, Option<BootstrapStage>)>,
     pub dashboard: dashboard::DashboardHandle,
     status_tick: u64,
-    exe: std::path::PathBuf,
+    launch: crate::provisioning_demo::LaunchStyle,
 }
 
 impl SupervisorActor {
@@ -180,9 +206,13 @@ impl SupervisorActor {
         mut telemetry: SupervisorTelemetry,
         dashboard: dashboard::DashboardHandle,
         sender: swactor::runtime::ExternalSender,
+        registry: provisioning::BootstrapRegistry,
+        collector: std::sync::Arc<dyn provisioning::NodeTelemetryCollector>,
+        engine: EngineHandle,
+        remote_sub: telemetry::TelemetrySubscription,
         initial_slots: Vec<String>,
         run_id: RunId,
-        exe: std::path::PathBuf,
+        launch: crate::provisioning_demo::LaunchStyle,
     ) -> Self {
         let events_channel = telemetry.register("prov.reconciler.events");
         let snapshot_channel = telemetry.register("prov.reconciler.snapshot");
@@ -195,6 +225,12 @@ impl SupervisorActor {
             events_channel,
             snapshot_channel,
             sender,
+            registry,
+            collector,
+            engine,
+            remote_sub,
+            remote_streams: BTreeMap::new(),
+            pending_reports: BTreeMap::new(),
             nodes: BTreeMap::new(),
             run_id,
             slot_seq: initial_slots.len() as u64,
@@ -203,7 +239,7 @@ impl SupervisorActor {
             last_stages: BTreeMap::new(),
             dashboard,
             status_tick: 0,
-            exe,
+            launch,
         }
     }
 
@@ -223,49 +259,103 @@ impl SupervisorActor {
             "detail": detail,
         });
         let bytes = serde_json::to_vec(&payload).expect("event serializes");
-        self.telemetry.producer.submit_bytes(self.events_channel, bytes);
+        self.telemetry
+            .producer
+            .submit_bytes(self.events_channel, bytes);
     }
 
-    /// Handle a spawn request from the provider (runs in actor context).
+    /// Handle a spawn request from the provider (runs in actor context):
-    fn spawn_node(&mut self, ctx: &Ctx, request: crate::provisioning_demo::provider::SpawnNodeRequest) {
+    /// spawn the per-attempt bootstrap actor through the registry — the
+    /// supervision never launches nodes directly.
+    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()),
-        };
 
+        // Supervisor-authored lifecycle stream (kept alive independent of
+        // the node: it reports death even when the node cannot).
         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 (kind, argv, mut env) = match &self.launch {
-        let output =
+            crate::provisioning_demo::LaunchStyle::Process { exe } => (
-            ProcessOutputConfig::telemetry_mirror(relay, telemetry.producer.clone());
+                "process",
-        match spawn_local_process(ctx, &self.sender, spec, output) {
+                vec![
-            Ok(process_actor) => {
+                    exe.to_string_lossy().to_string(),
+                    "provisioning-reconciler-demo".to_owned(),
+                    "--demo-node".to_owned(),
+                    self.driver_handle.supervisor_addr_json.clone(),
+                    "--demo-attempt".to_owned(),
+                    attempt.to_string(),
+                ],
+                Vec::new(),
+            ),
+            crate::provisioning_demo::LaunchStyle::Docker(docker) => (
+                "docker",
+                vec![
+                    "docker".to_owned(),
+                    "run".to_owned(),
+                    "--rm".to_owned(),
+                    "--name".to_owned(),
+                    crate::provisioning_demo::docker::container_name(attempt),
+                    "--label".to_owned(),
+                    format!("{}=1", crate::provisioning_demo::docker::SWEEP_LABEL),
+                    "--label".to_owned(),
+                    format!(
+                        "{}={}",
+                        crate::provisioning_demo::docker::RUN_LABEL,
+                        docker.run_token
+                    ),
+                    "--network".to_owned(),
+                    docker.network.clone(),
+                    docker.image.clone(),
+                    "provisioning-reconciler-demo".to_owned(),
+                    "--demo-node".to_owned(),
+                    docker.supervisor_addr_json.clone(),
+                    "--demo-attempt".to_owned(),
+                    attempt.to_string(),
+                ],
+                Vec::new(),
+            ),
+        };
+        env.push(("DEMO_NODE_ID".to_owned(), request.logical_node.clone()));
+
+        let spec = provisioning::NodeLaunchSpec {
+            kind: kind.to_owned(),
+            attempt,
+            logical_node: request.logical_node.clone(),
+            argv,
+            env,
+            workdir: None,
+            label: Some(request.logical_node.clone()),
+        };
+
+        let reporter: provisioning::BootstrapReporter = {
+            let sender = self.sender.clone();
+            let supervisor = ctx.self_addr();
+            std::sync::Arc::new(move |event| {
+                let _ = sender.send_to(supervisor, SupervisorMsg::Bootstrap(event));
+            })
+        };
+        let config = provisioning::BootstrapConfig {
+            reporter,
+            sender: self.sender.clone(),
+            collector: Some(std::sync::Arc::clone(&self.collector)),
+            probe_period: provisioning::DEFAULT_PROBE_PERIOD,
+            spec: spec.clone(),
+        };
+        let logic = match self.registry.create(&spec) {
+            Ok(logic) => logic,
+            Err(error) => {
+                let _ = request.reply.send(Err(format!("bootstrap logic: {error}")));
+                return;
+            }
+        };
+        match provisioning::spawn_bootstrap_actor(ctx, &self.engine, logic, config) {
+            Ok(bootstrap) => {
                 self.nodes.insert(
                     attempt,
                     NodeStreams {
@@ -273,36 +363,27 @@ impl SupervisorActor {
                         status_channel,
                     },
                 );
-                self.manager.register(
+                let runtime = crate::provisioning_demo::provider::NodeRuntime {
-                    crate::provisioning_demo::provider::NodeRuntime {
+                    attempt,
-                        attempt,
+                    logical_node: request.logical_node.clone(),
-                        logical_node: request.logical_node.clone(),
+                    bootstrap,
-                        process_actor,
+                    pid: None,
-                        key_file: request.key_file.clone(),
+                    exited: None,
-                        pid: None,
+                    spawn_failed: None,
-                        exited: None,
+                    last_announce_ms: None,
-                        spawn_failed: None,
+                };
-                    },
+                let _ = request.reply.send(Ok(runtime));
-                );
-                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}")));
+                let _ = request
+                    .reply
+                    .send(Err(format!("spawn bootstrap actor: {error}")));
             }
         }
     }
 
-    /// Feed real-world observations into the driver.
+    /// Feed bootstrap-progress observations into the driver. Join detection
+    /// and exit classification moved into the per-node bootstrap actors
     fn observe_world(&mut self, now: SystemTime) {
         let node_ids: Vec<String> = self
             .driver
@@ -312,87 +393,246 @@ impl SupervisorActor {
             .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 {
+            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 Some(session_id) = managed.active_bootstrap else {
-            let active_bootstrap = managed.active_bootstrap;
+                continue;
-            let runtime = match self.manager.get(attempt.0) {
-                Some(runtime) => runtime,
-                None => continue,
             };
+            let Some(runtime) = self.manager.get(attempt.0) else {
+                continue;
+            };
+            // Stage evidence: the supervised child started (process pid /
+            // docker CLI pid observed) means the node runtime is coming up
+            // and we are waiting for its control-plane announce; before
+            // that the lease exists but the foreign process is not up yet.
+            let stage_seen = if runtime.pid.is_some() {
+                BootstrapStage::WaitingForSwactorJoin
+            } else {
+                BootstrapStage::SshReady
+            };
+            self.driver.apply_observation(
+                &provisioning::node::LogicalNodeId(node_id.clone()),
+                attempt,
+                NodeObservation::BootstrapObserved {
+                    session_id,
+                    observation: BootstrapObservation::stage(stage_seen),
+                },
+                now,
+            );
+        }
+    }
 
-            // Child exit or spawn failure: fail the attempt while it is
+    /// Fold bootstrap-actor events into the reconciler. A report that
-            // still bootstrapping so the reconciler retries with a fresh
+    /// arrives before the reconciler has an active bootstrap session for
-            // lease instead of wedging at SshReady forever.
+    /// the attempt (the actor starts at spawn, `StartBootstrap` comes
-            if runtime.exited.is_some() || runtime.spawn_failed.is_some() {
+    /// later) is parked and re-delivered on the next tick — the join signal
-                if stage != NodeStage::Failed && active_bootstrap.is_some() {
+    /// is exactly-once, so dropping an early one wedges the session.
-                    let reason = if let Some(status) = &runtime.exited {
+    fn handle_bootstrap_event(&mut self, now: SystemTime, event: provisioning::BootstrapEvent) {
-                        format!("node process exited: {status:?}")
+        let attempt = match &event {
-                    } else {
+            provisioning::BootstrapEvent::Bootstrapped(identity) => identity.attempt,
-                        format!(
+            provisioning::BootstrapEvent::Failed { attempt, .. }
-                            "node process failed to spawn: {}",
+            | provisioning::BootstrapEvent::Exited { attempt, .. } => *attempt,
-                            runtime.spawn_failed.as_deref().unwrap_or("unknown")
+        };
-                        )
+        if !self.deliver_bootstrap_report(now, event.clone()) {
-                    };
+            self.pending_reports.insert(attempt, event);
-                    self.emit_event("observation", &node_id, reason.clone());
+        }
+    }
+
+    /// Deliver one bootstrap report to the reconciler driver. Returns false
+    /// when the attempt has no active bootstrap session yet.
+    fn deliver_bootstrap_report(
+        &mut self,
+        now: SystemTime,
+        event: provisioning::BootstrapEvent,
+    ) -> bool {
+        match event {
+            provisioning::BootstrapEvent::Bootstrapped(identity) => {
+                let Some((node_id, session_id)) = self.node_for_attempt(identity.attempt) else {
+                    return false;
+                };
+                let key_prefix = &identity.key_hex[..8.min(identity.key_hex.len())];
+                self.emit_event(
+                    "observation",
+                    &node_id,
+                    format!("swactor join confirmed (key {key_prefix}…)"),
+                );
+                self.driver.apply_observation(
+                    &provisioning::node::LogicalNodeId(node_id),
+                    provisioning::NodeAttemptId(identity.attempt),
+                    NodeObservation::SwactorJoined {
+                        session_id,
+                        swactor_id: SwactorId(identity.key_hex),
+                    },
+                    now,
+                );
+            }
+            provisioning::BootstrapEvent::Failed { attempt, reason } => {
+                let Some((node_id, session_id)) = self.node_for_attempt(attempt) else {
+                    return false;
+                };
+                self.emit_event("observation", &node_id, reason.clone());
+                self.driver.apply_observation(
+                    &provisioning::node::LogicalNodeId(node_id),
+                    provisioning::NodeAttemptId(attempt),
+                    NodeObservation::BootstrapFailed { session_id, reason },
+                    now,
+                );
+            }
+            provisioning::BootstrapEvent::Exited { attempt, reason } => {
+                // An exit while a session is still open is a bootstrap
+                // failure (death before join); otherwise it is a plain
+                // death observation.
+                if let Some((node_id, session_id)) = self.node_for_attempt(attempt) {
+                    self.emit_event("observation", &node_id, format!("node runtime: {reason}"));
                     self.driver.apply_observation(
-                        &provisioning::node::LogicalNodeId(node_id.clone()),
+                        &provisioning::node::LogicalNodeId(node_id),
-                        attempt,
+                        provisioning::NodeAttemptId(attempt),
                         NodeObservation::BootstrapFailed {
-                            session_id: active_bootstrap.expect("checked above"),
+                            session_id,
-                            reason,
+                            reason: format!("node process exited: {reason}"),
                         },
                         now,
                     );
                 }
-                continue;
+                // Publish the terminal status now: once the reconciler
-            }
+                // destroys the lease the runtime is deregistered, and the
-
+                // Fleet Control table would otherwise keep its last
-            // Bootstrap progression from the key file and iroh join state.
+                // "running" state forever. Derive the logical name from the
-            if let Some(session_id) = active_bootstrap {
+                // lifecycle stream itself — the registry entry can already
-                let report = read_key_report(&runtime.key_file);
+                // be gone (lease destroy races the bootstrap probe).
-                let mut stage_seen = BootstrapStage::SshReady;
+                if let Some(streams) = self.nodes.get(&attempt) {
-                if let Some(report) = &report {
+                    let logical = streams
-                    let connected = report
+                        .telemetry
-                        .node_hex
+                        .endpoint
-                        .parse_key()
+                        .stream_id()
-                        .is_some_and(|key| self.driver_handle.has_active_connection(key));
+                        .node
-                    if connected {
+                        .as_str()
-                        let heartbeat_age = unix_ms(now).saturating_sub(report.last_seen_ms);
+                        .to_owned();
-                        self.emit_event(
+                    let pid = self.manager.get(attempt).and_then(|r| r.pid);
-                            "observation",
+                    let payload = json!({
-                            &node_id,
+                        "at_ms": unix_ms(now),
-                            format!(
+                        "node": logical,
-                                "swactor join confirmed (key {}…, heartbeat {}ms old)",
+                        "alive": false,
-                                &report.node_hex[..8.min(report.node_hex.len())],
+                        "pid": pid,
-                                heartbeat_age
+                        "event": "exited",
-                            ),
+                    });
-                        );
+                    if let Ok(bytes) = serde_json::to_vec(&payload) {
-                        let swactor_id = SwactorId(report.node_hex.clone());
+                        streams
-                        self.driver.apply_observation(
+                            .telemetry
-                            &provisioning::node::LogicalNodeId(node_id.clone()),
+                            .producer
-                            attempt,
+                            .submit_bytes(streams.status_channel, bytes);
-                            NodeObservation::SwactorJoined {
-                                session_id,
-                                swactor_id,
-                            },
-                            now,
-                        );
-                        continue;
                     }
-                    stage_seen = BootstrapStage::WaitingForSwactorJoin;
                 }
-                self.driver.apply_observation(
+                // Death-replacement is shape logic: `replace_dead_ready_nodes`
-                    &provisioning::node::LogicalNodeId(node_id.clone()),
+                // reads the exit from the shared registry.
-                    attempt,
+            }
-                    NodeObservation::BootstrapObserved {
+        }
-                        session_id,
+        true
-                        observation: BootstrapObservation::stage(stage_seen),
+    }
-                    },
+
-                    now,
+    /// Re-deliver parked bootstrap reports whose session has appeared.
-                );
+    fn deliver_pending_reports(&mut self, now: SystemTime) {
+        let attempts: Vec<u64> = self.pending_reports.keys().copied().collect();
+        for attempt in attempts {
+            let Some(event) = self.pending_reports.get(&attempt).cloned() else {
+                continue;
+            };
+            if self.deliver_bootstrap_report(now, event) {
+                self.pending_reports.remove(&attempt);
+            }
+        }
+    }
+
+    /// Resolve a driver node (id + active bootstrap session) by attempt.
+    fn node_for_attempt(
+        &self,
+        attempt: u64,
+    ) -> Option<(String, provisioning::node::BootstrapSessionId)> {
+        self.driver
+            .state()
+            .nodes
+            .iter()
+            .find(|(_, managed)| managed.attempt.0 == attempt)
+            .and_then(|(id, managed)| {
+                managed
+                    .active_bootstrap
+                    .map(|session| (id.0.clone(), session))
+            })
+    }
+
+    /// Register a remote node telemetry stream header (pull side).
+    fn register_remote_stream(
+        &mut self,
+        header: iroh_driver::TelemetryQuicHeader,
+        logical_node: String,
+        attempt: u64,
+    ) {
+        let key = header.stream.stream.node.as_str().to_string();
+        let meta = self.remote_streams.entry(key.clone()).or_default();
+        meta.logical_node = logical_node.clone();
+        meta.attempt = attempt;
+        for channel in &header.channels {
+            meta.channels.insert(channel.id, channel.name.clone());
+        }
+        println!(
+            "demo: fusing telemetry of node {logical_node} (stream {key}, {} channels)",
+            meta.channels.len()
+        );
+    }
+    /// Drain remote-node telemetry frames into the dashboard, fusing them
+    /// onto the logical node's stream: one card per node, carrying both the
+    /// supervisor-authored lifecycle channels and the node's real runtime
+    /// channels (runtime.actors, node.beat, node.status).
+    fn flush_remote_streams(&mut self) {
+        for event in self.remote_sub.drain_available() {
+            match event {
+                telemetry::frame::TelemetryEvent::Frame(delivery) => {
+                    let key = delivery.channel.stream.node.as_str().to_string();
+                    let Some(meta) = self.remote_streams.get(&key).cloned() else {
+                        continue;
+                    };
+                    let Some(node_stream) = self.nodes.get(&meta.attempt) else {
+                        continue;
+                    };
+                    // Fuse: republish on the logical node's stream.
+                    let target_stream = node_stream.telemetry.endpoint.stream_id().clone();
+                    let origin = node_stream.telemetry.origin;
+                    let label = node_stream.telemetry.label.clone();
+                    let channel_name = meta
+                        .channels
+                        .get(&delivery.channel.channel)
+                        .cloned()
+                        .unwrap_or_else(|| format!("channel#{}", delivery.channel.channel.0));
+                    let frame = telemetry::frame::Frame {
+                        channel: delivery.channel.channel,
+                        position: delivery.position,
+                        payload: delivery.payload,
+                    };
+                    publish_frame(
+                        &self.dashboard,
+                        &target_stream,
+                        &channel_name,
+                        &frame,
+                        origin,
+                        &label,
+                    );
+                }
+                telemetry::frame::TelemetryEvent::ChannelDeclared(descriptor) => {
+                    let key = descriptor.stream.node.as_str().to_string();
+                    self.remote_streams
+                        .entry(key)
+                        .or_default()
+                        .channels
+                        .insert(descriptor.id, descriptor.name);
+                }
+                _ => {}
             }
         }
     }
@@ -403,7 +643,8 @@ impl SupervisorActor {
         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
+            if managed.record.ready
+                && managed.intent == provisioning::reconciler::NodeIntent::Active
             {
                 let Some(runtime) = self.manager.get(managed.attempt.0) else {
                     continue;
@@ -419,7 +660,11 @@ impl SupervisorActor {
             return;
         }
         for (dead, fresh) in &replacements {
-            self.emit_event("control", dead, format!("runtime death; replacing as {fresh}"));
+            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());
         }
@@ -514,16 +759,20 @@ impl SupervisorActor {
             let Some(runtime) = self.manager.get(attempt) else {
                 continue;
             };
-            let report = read_key_report(&runtime.key_file);
+            // Wire liveness: the node re-announces every heartbeat period,
-            let heartbeat_ms_ago = report
+            // so the age of the last announce is the control-plane
-                .as_ref()
+            // heartbeat. `null` until the first announce.
-                .map(|r| unix_ms(now).saturating_sub(r.last_seen_ms))
+            let heartbeat_ms_ago = runtime
-                .unwrap_or(u64::MAX);
+                .last_announce_ms
+                .map(|ms| unix_ms(now).saturating_sub(ms));
             let payload = json!({
                 "at_ms": unix_ms(now),
                 "node": runtime.logical_node,
                 "alive": runtime.exited.is_none(),
                 "pid": runtime.pid,
+                // Process state for the Fleet Control table (same shape the
+                // process lifecycle mirror used to publish).
+                "event": if runtime.exited.is_some() { "exited" } else { "started" },
                 "heartbeat_ms_ago": heartbeat_ms_ago,
             });
             let bytes = serde_json::to_vec(&payload).expect("status serializes");
@@ -681,6 +930,7 @@ impl ActorInterface for SupervisorActor {
         match msg {
             SupervisorMsg::Tick => {
                 let now = SystemTime::now();
+                self.deliver_pending_reports(now);
                 self.poll(now);
                 self.observe_world(now);
                 self.replace_dead_ready_nodes(now);
@@ -688,9 +938,20 @@ impl ActorInterface for SupervisorActor {
                 self.emit_node_status(now);
                 self.emit_feed(now);
                 self.flush_telemetry();
+                self.flush_remote_streams();
             }
             SupervisorMsg::Control(command) => self.handle_control(command),
             SupervisorMsg::Spawn(request) => self.spawn_node(ctx, request),
+            SupervisorMsg::Bootstrap(event) => {
+                let now = SystemTime::now();
+                self.handle_bootstrap_event(now, event);
+                self.poll(now);
+            }
+            SupervisorMsg::NodeStream {
+                header,
+                logical_node,
+                attempt,
+            } => self.register_remote_stream(header, logical_node, attempt),
             SupervisorMsg::Shutdown { drained } => {
                 self.slots.clear();
                 let generation = self.driver.desired().generation.saturating_add(1);
@@ -715,10 +976,9 @@ impl SupervisorActor {
                             &node,
                             format!("kill requested (pid {:?})", runtime.pid),
                         );
-                        let _ = swactor_process::send_process_command(
+                        let _ = self.sender.send_to(
-                            &self.sender,
+                            runtime.bootstrap,
-                            runtime.process_actor,
+                            provisioning::BootstrapMsg::Stop {
-                            swactor_process::ProcessCommand::Stop {
                                 kill_after: Some(Duration::ZERO),
                             },
                         );
@@ -760,19 +1020,6 @@ impl SupervisorActor {
     }
 }
 
-/// 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());
@@ -832,4 +1079,3 @@ pub fn demo_retry_policy() -> RetryPolicy {
         endpoint_probe_interval: Duration::from_secs(1),
     }
 }
-

xtask/src/provisioning_demo/mod.rs

@@ -12,12 +12,26 @@
 //! (per-node PID/state), kill nodes from Fleet Control or a shell, and watch
 //! the reconciler replace them for real.
 
+pub mod bootstrap;
 pub mod control;
+pub mod docker;
 pub mod feed;
 pub mod node;
 pub mod provider;
 pub mod view;
 
+/// How the supervisor launches node-role children (bootstrap kind + launch
+/// facts). Selected by `--docker`; default is local processes.
+#[derive(Clone)]
+pub enum LaunchStyle {
+    /// Re-exec this binary as a local child process (`kind: "process"`).
+    Process { exe: std::path::PathBuf },
+    /// Run the demo docker image per node on the per-run bridge network
+    /// (`kind: "docker"`) — nodes are foreign: own IPs, gateway-dialed
+    /// supervisor, no shared filesystem.
+    Docker(docker::DockerLaunch),
+}
+
 use std::process::ExitCode;
 use std::sync::Arc;
 use std::time::Duration;
@@ -30,12 +44,12 @@ 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 provisioning::{ClusterShape, RunId};
 
 use feed::{
-    demo_retry_policy, initial_slots, EngineSpawner, SupervisorActor, SupervisorMsg,
+    EngineSpawner, SupervisorActor, SupervisorMsg, SupervisorTelemetry, demo_retry_policy,
-    SupervisorTelemetry,
+    initial_slots,
 };
 use provider::{DemoBackend, DemoProvider, NodeManager};
 
@@ -81,27 +95,92 @@ fn resolve_exe() -> std::path::PathBuf {
     std::path::PathBuf::from("xtask")
 }
 
-/// Shared handle the supervisor actor uses to check iroh connections.
+/// Shared iroh driver handle: the supervisor's endpoint address (handed to
+/// node roles) and the endpoint for outbound telemetry pulls.
 pub struct DemoDriverHandle {
     pub supervisor_addr_json: String,
-    driver: IrohDriver,
+    pub(crate) driver: IrohDriver,
 }
 
 impl DemoDriverHandle {
-    pub fn has_active_connection(&self, node: swactor_transport::NodeId) -> bool {
+    /// Clone the iroh endpoint for outbound telemetry pulls.
-        self.driver.has_active_connection(&node)
+    pub fn endpoint(&self) -> iroh::Endpoint {
+        self.driver.endpoint()
     }
 }
 
-/// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N]` for the
+/// Pull collector fired by bootstrap actors: dials the freshly-bootstrapped
-/// supervisor, or `--demo-node <supervisor-addr-json>` for node children.
+/// node on `TELEMETRY_ALPN`, sends the subscription request, and streams its
+/// telemetry into the supervisor's remote-stream fanout. The header lands in
+/// the supervisor actor so frames can be classified before publishing.
+struct DemoTelemetryCollector {
+    engine: swactor_engine::EngineHandle,
+    endpoint: iroh::Endpoint,
+    fanout: Arc<telemetry::DeliveryFanout>,
+    sender: swactor::runtime::ExternalSender,
+    supervisor_slot: Arc<std::sync::OnceLock<swactor::actor::ActorAddress>>,
+}
+
+impl provisioning::NodeTelemetryCollector for DemoTelemetryCollector {
+    fn collect(&self, identity: &provisioning::NodeIdentity) {
+        let Ok(addr) = serde_json::from_str::<iroh::EndpointAddr>(&identity.transport_addr) else {
+            eprintln!(
+                "demo: node {} advertised unparsable endpoint address",
+                identity.logical_node
+            );
+            return;
+        };
+        println!(
+            "demo: pulling telemetry from node {} (key {}…)",
+            identity.logical_node,
+            &identity.key_hex[..8.min(identity.key_hex.len())]
+        );
+        let mut flow_id = [0u8; 16];
+        flow_id[..8].copy_from_slice(&identity.attempt.to_le_bytes());
+        let (header_tx, header_rx) = std::sync::mpsc::channel();
+        iroh_driver::spawn_pull_collector(
+            &self.engine,
+            self.endpoint.clone(),
+            addr,
+            flow_id,
+            Vec::new(),
+            telemetry::SubscriptionRequest::all(),
+            Arc::clone(&self.fanout),
+            header_tx,
+        );
+        let sender = self.sender.clone();
+        if let Some(supervisor) = self.supervisor_slot.get() {
+            let supervisor = supervisor.clone();
+            let logical_node = identity.logical_node.clone();
+            let attempt = identity.attempt;
+            std::thread::spawn(move || {
+                if let Ok(header) = header_rx.recv() {
+                    let _ = sender.send_to(
+                        supervisor,
+                        feed::SupervisorMsg::NodeStream {
+                            header,
+                            logical_node,
+                            attempt,
+                        },
+                    );
+                }
+            });
+        }
+    }
+}
+
+/// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N]
+/// [--docker]` for the supervisor, or `--demo-node <supervisor-addr-json>
+/// --demo-attempt <n>` for node children.
 pub fn run(args: &[String]) -> ExitCode {
     if let Some(index) = args.iter().position(|arg| arg == "--demo-node") {
-        let addr = args
+        let addr = args.get(index + 1).map(String::as_str).unwrap_or_default();
-            .get(index + 1)
+        let attempt = arg_value(args, "--demo-attempt").and_then(|value| value.parse::<u64>().ok());
-            .map(String::as_str)
+        let Some(attempt) = attempt else {
-            .unwrap_or_default();
+            eprintln!("demo node: --demo-attempt <n> is required");
-        return match node::run_node_role(addr) {
+            return ExitCode::FAILURE;
+        };
+        return match node::run_node_role(addr, attempt) {
             Ok(()) => ExitCode::SUCCESS,
             Err(error) => {
                 eprintln!("demo node: {error}");
@@ -130,6 +209,13 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
     let nodes: u64 = arg_value(args, "--nodes")
         .and_then(|value| value.parse().ok())
         .unwrap_or(DEFAULT_NODES);
+    let docker_mode = args.iter().any(|arg| arg == "--docker");
+
+    // Node registry + spawn channel: needed before the actor bridge so the
+    // announce relay can route wire announces into bootstrap actors.
+    let manager = NodeManager::new();
+    let (spawn_tx, spawn_rx) = std::sync::mpsc::channel::<provider::SpawnNodeRequest>();
+    manager.set_spawn_channel(spawn_tx);
 
     // Telemetry endpoint with the runtime stats hook attached before the
     // engine takes the parts.
@@ -168,29 +254,42 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
         use distribution::transport_bridge::{Outbox, RelayMirror, RouteView};
         use swactor_transport::CodecRegistry;
 
-        struct NoopActor;
+        // The announce relay is the bridge's only decoded ingress: node
-        impl swactor::actor::ActorInterface for NoopActor {
+        // roles announce over the control plane (tag-routed gossip), and
-            type Incoming = ();
+        // the relay correlates by attempt → bootstrap actor. The
-            type Response = ();
+        // `swim_addr` argument only names a SendFailed target, which never
-            fn handle(&mut self, _ctx: &swactor::actor::Ctx, _msg: ()) {}
+        // fires for inbound frames; the relay doubles for it. Undecodable
-        }
+        // frames (e.g. the driver's own JoinRequest) still drop harmlessly.
-        let noop = runtime
+        let announce = runtime
-            .spawn(NoopActor)
+            .spawn(provider::AnnounceActor::new(
-            .map_err(|e| format!("spawn noop actor: {e}"))?;
+                manager.clone(),
-        let relay_mirror: RelayMirror = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
+                sender.clone(),
-        let route_view: RouteView = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));
+            ))
+            .map_err(|e| format!("spawn announce actor: {e}"))?;
+        let mut codec = CodecRegistry::new();
+        codec.register_decoder::<node::NodeAnnounce>(node::ANNOUNCE_TAG, |bytes| {
+            serde_json::from_slice(bytes)
+                .map_err(|e| swactor::Error::from(format!("announce decode: {e}")))
+        });
+        let mut routes = std::collections::HashMap::new();
+        routes.insert(node::ANNOUNCE_TAG.to_owned(), announce);
+        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()),
+            Arc::new(codec),
-            std::collections::HashMap::new(),
+            routes,
-            noop,
+            announce,
             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,
@@ -205,15 +304,23 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
     engine.handle().spawn(dashboard.http_server());
 
     // Provisioning: driver + plugin + executor.
-    let keys_dir = std::env::temp_dir().join(format!(
+    let launch = if docker_mode {
-        "provisioning-demo-{}",
+        let addrs = driver_handle.driver.direct_addresses();
-        std::process::id()
+        let port = addrs
-    ));
+            .iter()
-    std::fs::create_dir_all(&keys_dir).map_err(|e| format!("keys dir: {e}"))?;
+            .find(|addr| addr.is_ipv4())
-
+            .or_else(|| addrs.first())
-    let manager = NodeManager::new();
+            .map(|addr| addr.port())
-    let (spawn_tx, spawn_rx) = std::sync::mpsc::channel::<provider::SpawnNodeRequest>();
+            .ok_or_else(|| "supervisor has no bound direct address".to_owned())?;
-    manager.set_spawn_channel(spawn_tx);
+        let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .expect("xtask manifest dir has a parent")
+            .to_path_buf();
+        let launch = docker::preflight(&root, driver_handle.driver.endpoint().id(), port)?;
+        LaunchStyle::Docker(launch)
+    } else {
+        LaunchStyle::Process { exe: resolve_exe() }
+    };
 
     let slots = initial_slots(nodes);
     let shape = ClusterShape {
@@ -224,7 +331,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
     let cluster_driver =
         ClusterDriver::new(shape, demo_retry_policy()).map_err(|e| format!("driver: {e}"))?;
 
-    let plugin = DemoProvider::new(manager.clone(), keys_dir.clone());
+    let plugin = DemoProvider::new(manager.clone());
     let spawner = EngineSpawner::new(engine.handle());
     let executor = IdempotentEffectExecutor::new(
         DemoBackend {
@@ -235,6 +342,50 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
         spawner,
     );
 
+    // The supervisor's address travels through a shared slot; long-lived
+    // engine tasks installed below 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());
+
+    // Bootstrap machinery: kind registry with the process logic, the
+    // remote-stream fanout, and the pull collector.
+    let fanout = Arc::new(telemetry::DeliveryFanout::new(1024));
+    let remote_sub = fanout.subscribe_all(
+        "dashboard",
+        telemetry::TelemetrySnapshot {
+            streams: Vec::new(),
+            channels: Vec::new(),
+        },
+    );
+    let mut bootstrap_registry = provisioning::BootstrapRegistry::new();
+    bootstrap_registry.register("process", {
+        let manager = manager.clone();
+        Arc::new(move |spec| {
+            Ok(Box::new(bootstrap::LocalProcessLogic::new(
+                spec.clone(),
+                manager.clone(),
+            )) as Box<dyn provisioning::BootstrapLogic>)
+        })
+    });
+    bootstrap_registry.register("docker", {
+        let manager = manager.clone();
+        Arc::new(move |spec| {
+            Ok(Box::new(docker::DockerProcessLogic::new(
+                spec.clone(),
+                manager.clone(),
+            )) as Box<dyn provisioning::BootstrapLogic>)
+        })
+    });
+    let collector: Arc<dyn provisioning::NodeTelemetryCollector> =
+        Arc::new(DemoTelemetryCollector {
+            engine: engine.handle(),
+            endpoint: driver_handle.endpoint(),
+            fanout: Arc::clone(&fanout),
+            sender: sender.clone(),
+            supervisor_slot: supervisor_slot.clone(),
+        });
+
     let supervisor = SupervisorActor::new(
         cluster_driver,
         executor,
@@ -243,16 +394,14 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
         telemetry,
         dashboard.clone(),
         sender.clone(),
+        bootstrap_registry,
+        collector,
+        engine.handle(),
+        remote_sub,
         slots,
         RunId(1),
-        resolve_exe(),
+        launch.clone(),
     );
-    // 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());
@@ -260,14 +409,16 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
     // 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 {
+    engine.handle().spawn_blocking(move || {
-        match spawn_rx.recv() {
+        loop {
-            Ok(request) => {
+            match spawn_rx.recv() {
-                if let Some(addr) = pump_slot.get() {
+                Ok(request) => {
-                    let _ = pump_sender.send_to(addr.clone(), SupervisorMsg::Spawn(request));
+                    if let Some(addr) = pump_slot.get() {
+                        let _ = pump_sender.send_to(addr.clone(), SupervisorMsg::Spawn(request));
+                    }
                 }
+                Err(_) => return,
             }
-            Err(_) => return,
         }
     });
 
@@ -328,8 +479,11 @@ fn run_supervisor(args: &[String]) -> Result<(), String> {
         std::thread::sleep(TICK);
     }
     // Best-effort drain window closed: children still alive (if any) are
-    // killed by the kernel parent-death signal armed in the node role.
+    // killed by the kernel parent-death signal armed in the node role
-    let _ = std::fs::remove_dir_all(&keys_dir);
+    // (process kind) or swept by label below (docker kind).
+    if let LaunchStyle::Docker(docker) = &launch {
+        docker::sweep_run(docker);
+    }
     std::process::exit(0);
 }
 
@@ -344,6 +498,10 @@ fn install_sigint_flag() {
     unsafe {
         let handler: extern "C" fn(libc::c_int) = sigint_handler;
         libc::signal(libc::SIGINT, handler as usize);
+        // Supervisors under process managers (systemd, container runtimes,
+        // harnesses) stop children with SIGTERM; treat it exactly like
+        // Ctrl-C so the drain + sweep path runs instead of a default kill.
+        libc::signal(libc::SIGTERM, handler as usize);
     }
 }
 

xtask/src/provisioning_demo/node.rs

@@ -1,38 +1,85 @@
 //! 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 +
+//! Re-exec'd from the xtask binary by the demo provider (local process kind)
-//! `IrohDriver` (relay disabled), joins the supervisor's endpoint, and
+//! or launched as the entrypoint of the demo docker image. Builds an engine +
-//! reports its swactor node key + heartbeats to the key file given in
+//! `IrohDriver` (relay disabled), joins the supervisor's endpoint, and then
-//! `DEMO_NODE_KEY_FILE` (the process actor supervises children with stdio
+//! **announces itself over the control plane**: a tagged gossip frame
-//! null, so stdout is not observable).
+//! ([`ANNOUNCE_TAG`]) carrying the provision attempt token, its iroh node
+//! key, and its advertised endpoint address. The supervisor's announce relay
+//! routes that frame by attempt to the owning bootstrap actor — the wire
+//! announce *is* the readiness signal (no shared filesystem, no stdout
+//! capture), and it repeats every [`HEARTBEAT_PERIOD`] as a liveness
+//! heartbeat.
+//!
+//! Telemetry: the node owns a real [`TelemetryEndpoint`] whose stream id is
+//! its transport identity (iroh node key). The runtime stats hook feeds
+//! `runtime.actors` rosters, a `node.status` channel carries liveness
+//! records, and beat actors keep the roster visibly alive. The supervisor
+//! pulls this endpoint over `TELEMETRY_ALPN`
+//! ([`iroh_driver::spawn_pull_server`]) — the node never dials out for
+//! telemetry.
 
-use std::io::Write;
+use std::sync::Arc;
+use std::time::Duration;
 
 use iroh::RelayMode;
+use serde_json::json;
+use swactor::actor::{ActorInterface, Ctx};
 use swactor::config::RuntimeConfig;
 use swactor::runtime::RuntimeParts;
 use swactor_engine::{Engine, TokioBackend, TokioConfig};
 
 use distribution::node::DistributedNodeConfig;
-use iroh_driver::{IrohDriver, IrohDriverConfig};
+use iroh_driver::{IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, spawn_pull_server};
+use telemetry::{ChannelContent, TelemetryEndpoint, TelemetryProducer};
 
 use crate::provisioning_demo::HEARTBEAT_PERIOD;
 
-/// Key-file line marking the node key (first line): `<ms> KEY <hex>`.
+/// Wire tag of the announce gossip frame (`CodecRegistry` decode key on the
-pub const KEY_LINE_KIND: &str = "KEY";
+/// supervisor side; raw tag bytes on the node side).
+pub const ANNOUNCE_TAG: &str = "xtask_demo/NodeAnnounce/1";
 
-/// Key-file heartbeat line: `<ms> ALIVE`.
+/// A node's self-introduction on the supervisor's control plane. Sent right
-pub const ALIVE_LINE_KIND: &str = "ALIVE";
+/// after joining and then every heartbeat period; the supervisor routes it
+/// by `attempt` to the owning bootstrap actor (first delivery = readiness,
+/// later deliveries = liveness).
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct NodeAnnounce {
+    /// Provision attempt token (`--demo-attempt`); correlates the announce
+    /// with the supervisor-side bootstrap actor that launched this node.
+    pub attempt: u64,
+    /// Logical node name (slot id) for dashboard labeling.
+    pub logical_node: String,
+    /// The node's transport key (iroh node id, hex).
+    pub key_hex: String,
+    /// Serde-serialized `iroh::EndpointAddr` the supervisor dials for
+    /// telemetry pulls.
+    pub endpoint_addr_json: String,
+    pub at_ms: u64,
+}
+
+/// How often the node polls its driver for accepted telemetry pull
+/// connections.
+const PULL_POLL_PERIOD: Duration = Duration::from_millis(200);
+
+/// Idle sleep for the pull answer writer between subscription batches.
+const WRITER_IDLE: Duration = Duration::from_millis(100);
+
+/// Number of beat actors on the node — real actors whose message flow keeps
+/// the `runtime.actors` roster visibly alive.
+const BEAT_ACTORS: u64 = 2;
 
 /// Run the node role. `supervisor_addr_json` is a serde-serialized
-/// `iroh::EndpointAddr` of the supervisor's iroh endpoint.
+/// `iroh::EndpointAddr` of the supervisor's iroh endpoint; `attempt` is the
-pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> {
+/// provision attempt token echoed back in every announce.
+pub fn run_node_role(supervisor_addr_json: &str, attempt: u64) -> Result<(), String> {
+    install_parent_death_signal()?;
     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")
+    let logical_node = std::env::var("DEMO_NODE_ID").unwrap_or_else(|_| "node".to_owned());
-        .map_err(|_| "DEMO_NODE_KEY_FILE not set".to_owned())?;
 
     let parts = RuntimeParts::new(RuntimeConfig::default());
+    let runtime = parts.runtime().clone();
     let engine = Engine::new(
         parts,
         TokioBackend::new(TokioConfig::default()).map_err(|error| format!("backend: {error}"))?,
@@ -40,30 +87,170 @@ pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> {
     .map_err(|error| format!("engine: {error}"))?;
 
     // Bind the driver, then keep it alive for the process lifetime: dropping
-    // it closes the endpoint.
+    // it closes the endpoint. The endpoint must advertise TELEMETRY_ALPN so
-    let driver = IrohDriver::with_engine(
+    // the supervisor's pull connection can negotiate it.
-        engine.handle(),
+    let driver = Arc::new(
-        IrohDriverConfig {
+        IrohDriver::with_engine(
-            secret_key: None,
+            engine.handle(),
-            relay_mode: RelayMode::Disabled,
+            IrohDriverConfig {
-            node: DistributedNodeConfig::default(),
+                secret_key: None,
-            peer_auth: None,
+                relay_mode: RelayMode::Disabled,
-            additional_alpns: vec![],
+                node: DistributedNodeConfig::default(),
-        },
+                peer_auth: None,
-    )
+                additional_alpns: vec![TELEMETRY_ALPN.to_vec()],
-    .map_err(|error| format!("iroh driver: {error}"))?;
+            },
-
+        )
+        .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();
+    // Telemetry endpoint: stream identity is the transport node key, so
-    let interval_handle = engine.handle();
+    // every provisioned process lands on its own dashboard stream.
-    interval_handle.clone().spawn(async move {
+    let stream = telemetry::frame::StreamId::new(
-        let mut interval = interval_handle.interval(HEARTBEAT_PERIOD);
+        telemetry::frame::NodeId::new(&node_hex),
+        telemetry::frame::Lifetime(1),
+    );
+    let endpoint = TelemetryEndpoint::with_descriptor(
+        telemetry::frame::StreamDescriptor {
+            stream,
+            label: Some(format!("demo node {logical_node} runtime")),
+            origin: telemetry::frame::StreamOrigin::RemoteNode,
+        },
+        256,
+        16,
+    );
+    let producer = endpoint.producer();
+    let actors_channel = endpoint.register_channel(
+        "runtime.actors",
+        ChannelContent::JsonRecord {
+            schema: Some("runtime.actors".to_owned()),
+        },
+    );
+    let status_channel = endpoint.register_channel(
+        "node.status",
+        ChannelContent::JsonRecord {
+            schema: Some("demo.node.status.v1".to_owned()),
+        },
+    );
+    let beats_channel = endpoint.register_channel(
+        "node.beat",
+        ChannelContent::JsonRecord {
+            schema: Some("demo.node.beat.v1".to_owned()),
+        },
+    );
+    runtime.set_stats_hook(producer.stats_hook_on(actors_channel));
+
+    // Join, then announce identity + advertised address to the supervisor's
+    // bootstrap actor over the control plane (readiness + telemetry dial).
+    driver.join(&[supervisor_addr.clone()]);
+    let addr_json =
+        serde_json::to_string(&driver.endpoint_addr()).map_err(|e| format!("addr: {e}"))?;
+    let announce_driver = Arc::clone(&driver);
+    let announce_logical = logical_node.clone();
+    let announce_key = node_hex.clone();
+    let announce_engine = engine.handle();
+    announce_engine.clone().spawn(async move {
+        let mut interval = announce_engine.interval(HEARTBEAT_PERIOD);
         loop {
             (&mut interval).await;
-            append_key_line(&heartbeat_file, ALIVE_LINE_KIND, "");
+            let announce = NodeAnnounce {
+                attempt,
+                logical_node: announce_logical.clone(),
+                key_hex: announce_key.clone(),
+                endpoint_addr_json: addr_json.clone(),
+                at_ms: unix_ms(),
+            };
+            let Ok(bytes) = serde_json::to_vec(&announce) else {
+                continue;
+            };
+            announce_driver.send_tagged_gossip(
+                supervisor_addr.clone(),
+                ANNOUNCE_TAG.as_bytes(),
+                bytes,
+            );
+        }
+    });
+
+    // Beat actors: real actors with real message flow, so the node's
+    // runtime.actors roster (pulled by the supervisor) is visibly alive.
+    let sender = runtime.create_sender();
+    let mut beat_addrs = Vec::new();
+    for index in 0..BEAT_ACTORS {
+        let name = format!("beat-{index}");
+        let producer = producer.clone();
+        let addr = runtime
+            .spawn(BeatActor {
+                name: name.clone(),
+                producer,
+                channel: beats_channel,
+            })
+            .map_err(|error| format!("spawn beat actor: {error}"))?;
+        beat_addrs.push((addr, name));
+    }
+
+    // Heartbeats: a node.status telemetry record (the wire announce above is
+    // the supervisor-facing liveness channel).
+    let heartbeat_producer = producer.clone();
+    let heartbeat_logical = logical_node.clone();
+    let heartbeat_key = node_hex.clone();
+    let interval_engine = engine.handle();
+    interval_engine.clone().spawn(async move {
+        let mut interval = interval_engine.interval(HEARTBEAT_PERIOD);
+        let mut seq: u64 = 0;
+        loop {
+            (&mut interval).await;
+            seq += 1;
+            let payload = json!({
+                "at_ms": unix_ms(),
+                "node": heartbeat_logical,
+                "key": heartbeat_key,
+                "seq": seq,
+                "alive": true,
+                "pid": std::process::id(),
+                "beats": BEAT_ACTORS,
+            });
+            let bytes = serde_json::to_vec(&payload).expect("status serializes");
+            heartbeat_producer.submit_bytes(status_channel, bytes);
+        }
+    });
+
+    // Beat driver: engine intervals poking the beat actors.
+    for (index, (addr, _name)) in beat_addrs.iter().enumerate() {
+        let addr = *addr;
+        let beat_sender = sender.clone();
+        let beat_engine = engine.handle();
+        // Stagger the beat periods so roster entries tick at different rates.
+        let period = Duration::from_millis(1000 + 500 * index as u64);
+        beat_engine.clone().spawn(async move {
+            let mut interval = beat_engine.interval(period);
+            loop {
+                (&mut interval).await;
+                let _ = beat_sender.send_to(addr, BeatMsg::Wake);
+            }
+        });
+    }
+
+    // Serve telemetry pulls: accepted TELEMETRY_ALPN connections answer with
+    // this endpoint's subscription stream.
+    let telemetry_endpoint = Arc::new(endpoint);
+    let serve_engine = engine.handle();
+    let serve_driver = Arc::clone(&driver);
+    serve_engine.clone().spawn(async move {
+        let mut interval = serve_engine.interval(PULL_POLL_PERIOD);
+        loop {
+            (&mut interval).await;
+            // Drain the mux into the endpoint fanout so producer frames
+            // reach the pull subscription; without this tick the mux fills
+            // and nothing is ever streamed.
+            let _ = telemetry_endpoint.tick();
+            for (_node, conn) in serve_driver.drain_accepted_for_alpn(TELEMETRY_ALPN) {
+                spawn_pull_server(
+                    &serve_engine,
+                    conn,
+                    Arc::clone(&telemetry_endpoint),
+                    WRITER_IDLE,
+                );
+            }
         }
     });
 
@@ -75,45 +262,68 @@ pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> {
     }
 }
 
-fn append_key_line(path: &str, kind: &str, value: &str) {
+/// Message that keeps a beat actor's mailbox flowing (and therefore its
-    let Ok(mut file) = std::fs::OpenOptions::new().create(true).append(true).open(path) else {
+/// `runtime.actors` stats current).
-        return;
+#[derive(Clone, Debug)]
-    };
+pub enum BeatMsg {
-    let now = std::time::SystemTime::now()
+    Wake,
+}
+
+/// A deliberately trivial actor whose only job is existing: every Wake it
+/// submits one `node.beat` record, and the runtime stats hook records its
+/// productive tick on `runtime.actors`.
+pub struct BeatActor {
+    name: String,
+    producer: TelemetryProducer,
+    channel: telemetry::ChannelId,
+}
+
+fn unix_ms() -> u64 {
+    std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|duration| duration.as_millis() as u64)
-        .unwrap_or(0);
+        .unwrap_or(0)
-    let _ = writeln!(file, "{now} {kind} {value}");
 }
 
-/// Parsed key-file contents: the child's node key and its last heartbeat.
+impl ActorInterface for BeatActor {
-pub struct NodeKeyReport {
+    type Incoming = BeatMsg;
-    pub node_hex: String,
+    type Response = ();
-    pub last_seen_ms: u64,
+
+    fn handle(&mut self, _ctx: &Ctx, _msg: BeatMsg) {
+        let payload = json!({
+            "at_ms": unix_ms(),
+            "actor": self.name,
+        });
+        let bytes = serde_json::to_vec(&payload).expect("beat serializes");
+        self.producer.submit_bytes(self.channel, bytes);
+    }
 }
 
-/// Read a node's key file: its node key and last heartbeat.
+/// Arm the kernel's parent-death signal so a hard-killed supervisor
-pub fn read_key_report(path: &std::path::Path) -> Option<NodeKeyReport> {
+/// (SIGKILL, crash, wedged teardown) cannot leave this node orphaned and
-    let contents = std::fs::read_to_string(path).ok()?;
+/// parked forever. The `parent_id` check closes the fork/prctl race: if the
-    let mut node_hex: Option<String> = None;
+/// supervisor died before the prctl landed, the signal would never fire.
-    let mut last_seen_ms = 0_u64;
+/// Inside a container the node is PID 1 (parent 0), so the check passes and
-    for line in contents.lines() {
+/// the prctl is a harmless no-op — the docker bootstrap's `docker rm -f`
-        let mut parts = line.split_whitespace();
+/// owns termination there.
-        let Some(stamp) = parts.next().and_then(|value| value.parse::<u64>().ok()) else {
+#[cfg(target_os = "linux")]
-            continue;
+fn install_parent_death_signal() -> Result<(), String> {
-        };
+    unsafe {
-        let Some(kind) = parts.next() else {
+        if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 {
-            continue;
+            return Err(format!(
-        };
+                "prctl(PR_SET_PDEATHSIG): {}",
-        if kind == KEY_LINE_KIND {
+                std::io::Error::last_os_error()
-            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 {
+    if std::os::unix::process::parent_id() == 1 {
-        node_hex: node_hex?,
+        return Err("supervisor exited before node startup".to_owned());
-        last_seen_ms,
+    }
-    })
+    Ok(())
+}
+
+#[cfg(not(target_os = "linux"))]
+fn install_parent_death_signal() -> Result<(), String> {
+    // Non-Linux builds rely on the supervisor's graceful teardown path.
+    Ok(())
 }

xtask/src/provisioning_demo/provider.rs

@@ -12,13 +12,12 @@
 //! 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
+//! Children run with stdio null, so the supervisor learns a node's iroh
-//! publishes its swactor node key and heartbeats to a per-attempt key file
+//! identity and liveness from the wire announce (see `node.rs`): the
-//! (`DEMO_NODE_KEY_FILE`); the supervisor reads it to learn the child's iroh
+//! [`AnnounceActor`] routes it to the owning bootstrap actor and records
-//! identity and liveness.
+//! `last_announce_ms` in the [`NodeManager`] registry.
 
 use std::collections::BTreeMap;
-use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
 
@@ -31,8 +30,8 @@ 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 provisioning::reconciler::{OperationId, OperationOutcome, PlannedEffect};
 use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer};
 
 /// How long a blocking plugin call waits for the supervisor actor.
@@ -43,11 +42,15 @@ pub const BACKEND_WAIT: Duration = Duration::from_secs(20);
 pub struct NodeRuntime {
     pub attempt: u64,
     pub logical_node: String,
-    pub process_actor: ActorAddress,
+    /// The bootstrap actor that owns this attempt's lifecycle.
-    pub key_file: PathBuf,
+    pub bootstrap: ActorAddress,
     pub pid: Option<u32>,
     pub exited: Option<ExitStatus>,
     pub spawn_failed: Option<String>,
+    /// Wall-clock ms of the last wire announce from the node (None until
+    /// the first announce). The node re-announces every heartbeat period,
+    /// so staleness here means the control-plane path is dead.
+    pub last_announce_ms: Option<u64>,
 }
 
 /// Spawn request from the plugin (blocking thread) to the supervisor actor.
@@ -55,7 +58,6 @@ pub struct NodeRuntime {
 pub struct SpawnNodeRequest {
     pub attempt: u64,
     pub logical_node: String,
-    pub key_file: PathBuf,
     pub reply: std::sync::mpsc::Sender<Result<NodeRuntime, String>>,
 }
 
@@ -81,17 +83,11 @@ impl NodeManager {
         self.inner.lock().expect("node manager").spawn_tx = Some(sender);
     }
 
-    pub fn request_spawn(
+    pub fn request_spawn(&self, attempt: u64, logical_node: String) -> Result<NodeRuntime, String> {
-        &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,
         };
         {
@@ -160,23 +156,27 @@ impl NodeManager {
         self.update(attempt, |runtime| runtime.spawn_failed = Some(reason));
     }
 
-    pub fn remove(&self, attempt: u64) {
+    pub fn set_announce(&self, attempt: u64, at_ms: u64) {
-        self.inner.lock().expect("node manager").nodes.remove(&attempt);
+        self.update(attempt, |runtime| runtime.last_announce_ms = Some(at_ms));
+    }
+    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 {
+    pub fn new(manager: NodeManager) -> Self {
-        Self { manager, keys_dir }
+        Self { manager }
     }
-
 }
 
 impl ProvisionPlugin for DemoProvider {
@@ -199,10 +199,7 @@ impl ProvisionPlugin for DemoProvider {
             .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)?;
-        let runtime = self
-            .manager
-            .request_spawn(attempt, logical_node, key_file)?;
         Ok(PluginNodeHandle {
             id: attempt,
             provider_process_id: runtime.pid,
@@ -297,6 +294,51 @@ impl ActorInterface for NodeRelayActor {
     }
 }
 
+/// Supervisor-side announce relay. The iroh actor bridge decodes inbound
+/// [`NodeAnnounce`] gossip frames and routes them here by wire tag; this
+/// actor correlates by attempt token and forwards to the owning bootstrap
+/// actor ([`provisioning::BootstrapMsg::Announce`]) — the handoff from
+/// bootstrap to control plane. Announces for unknown attempts (deregistered
+/// lease, stale container from a dead attempt) are dropped with a log line.
+pub struct AnnounceActor {
+    manager: NodeManager,
+    sender: ExternalSender,
+}
+
+impl AnnounceActor {
+    pub fn new(manager: NodeManager, sender: ExternalSender) -> Self {
+        Self { manager, sender }
+    }
+}
+
+impl ActorInterface for AnnounceActor {
+    type Incoming = crate::provisioning_demo::node::NodeAnnounce;
+    type Response = ();
+
+    fn handle(&mut self, _ctx: &Ctx, announce: crate::provisioning_demo::node::NodeAnnounce) {
+        self.manager.set_announce(announce.attempt, announce.at_ms);
+        let Some(runtime) = self.manager.get(announce.attempt) else {
+            let key = &announce.key_hex;
+            eprintln!(
+                "demo: announce for unknown attempt {} (key {}…) dropped",
+                announce.attempt,
+                &key[..8.min(key.len())]
+            );
+            return;
+        };
+        let identity = provisioning::NodeIdentity {
+            attempt: announce.attempt,
+            logical_node: announce.logical_node,
+            key_hex: announce.key_hex,
+            transport_addr: announce.endpoint_addr_json,
+        };
+        let _ = self.sender.send_to(
+            runtime.bootstrap,
+            provisioning::BootstrapMsg::Announce(identity),
+        );
+    }
+}
+
 /// Demo lease/session identity (attempt-encoded, kit conventions).
 pub fn demo_lease(attempt: u64) -> LeaseFacts {
     let provider = ProviderKind::new("demo");
@@ -331,7 +373,9 @@ pub fn session_id_for(operation: OperationId) -> BootstrapSessionId {
 }
 
 fn runtime_exited(manager: &NodeManager, attempt: u64) -> bool {
-    manager.get(attempt).is_some_and(|runtime| runtime.exited.is_some())
+    manager
+        .get(attempt)
+        .is_some_and(|runtime| runtime.exited.is_some())
 }
 
 /// The demo `EffectBackend`: routes effects through the plugin.
@@ -353,7 +397,6 @@ impl DemoBackend {
     }
 }
 
-
 impl EffectBackend for DemoBackend {
     fn execute(&self, effect: &PlannedEffect) -> Result<OperationOutcome, EffectError> {
         let attempt = effect.operation.attempt.0;
@@ -433,10 +476,9 @@ fn stop_node_with_sender(
 ) -> 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(
+            let _ = sender.send_to(
-                sender,
+                runtime.bootstrap,
-                runtime.process_actor,
+                provisioning::BootstrapMsg::Stop {
-                swactor_process::ProcessCommand::Stop {
                     kill_after: Some(Duration::from_secs(1)),
                 },
             );

xtask/src/provisioning_demo/view.rs

@@ -13,8 +13,8 @@ use serde::Serialize;
 use serde_json::Value;
 use telemetry::frame::{Frame, StreamId};
 
-use dashboard::view::DashboardView;
 use dashboard::FrameEvent;
+use dashboard::view::DashboardView;
 
 const EVENTS_CHANNEL: &str = "prov.reconciler.events";
 const SNAPSHOT_CHANNEL: &str = "prov.reconciler.snapshot";
@@ -171,4 +171,3 @@ impl DashboardView for ReconcilerDashboardView {
         None
     }
 }
-