fix(myelin): bootstrap pull telemetry and clean docker shutdown

Subscribe to each worker telemetry pull endpoint during runtime readiness, retry interrupted streams, and remove the obsolete telemetry actor-address path.

Make shutdown state sticky, clean local Docker resources on Ctrl+C, validate fleet controls, and keep dashboard node state honest under failure. Wait for bidirectional iroh routes before job submission and render the orchestrator role as a card header.

apps/myelin/src/codecs.rs

@@ -2,12 +2,10 @@
 //!
 //! The runtime's [`CodecRegistry`](swactor_transport::CodecRegistry) needs an
 //! encoder/decoder entry for each inter-node message type. This is the single
-//! aggregator that wires up the node, orchestrator, prompt, and telemetry
-//! publisher codecs.
+//! aggregator that wires up the node, orchestrator, and prompt codecs.
 
 pub(crate) fn register_myelin_actor_codecs(registry: &mut swactor_transport::CodecRegistry) {
     crate::node_actor::register_codecs(registry);
     crate::orchestration::actor::register_codecs(registry);
-    telemetry::register_telemetry_publisher_codec(registry);
     swactor_job_runner::register_job_codecs(registry);
 }

apps/myelin/src/node/actor.rs

@@ -137,7 +137,6 @@ pub(crate) enum NodeAgentMsg {
         stage_index: u32,
         endpoint: EndpointAddr,
         node_actor: ActorAddress,
-        telemetry_publisher: ActorAddress,
         readiness_id: u64,
     },
     RuntimeReadyAck {
@@ -450,7 +449,6 @@ impl NodeAgentActor {
                 stage_index,
                 endpoint,
                 node_actor,
-                telemetry_publisher,
                 readiness_id,
             } => {
                 self.core.observe(stage::StageEvent::WorkerReady);
@@ -462,7 +460,6 @@ impl NodeAgentActor {
                         stage_index,
                         endpoint,
                         node_actor,
-                        telemetry_publisher,
                         readiness_id,
                     },
                 );

apps/myelin/src/node/prompt_wire.rs

@@ -26,20 +26,6 @@ pub(crate) enum PromptEvent {
     },
 }
 
-impl PromptEvent {
-    pub(crate) fn request_id(&self) -> u64 {
-        match self {
-            Self::TextDelta { request_id, .. }
-            | Self::Done { request_id, .. }
-            | Self::Fault { request_id, .. } => *request_id,
-        }
-    }
-
-    pub(crate) fn is_terminal(&self) -> bool {
-        matches!(self, Self::Done { .. } | Self::Fault { .. })
-    }
-}
-
 impl NetworkMessage for PromptEvent {
     fn type_tag() -> &'static str {
         "myelin::PromptEvent"

apps/myelin/src/node/worker_node_runtime.rs

@@ -19,8 +19,7 @@ use std::time::{Duration, Instant};
 use telemetry::frame::TelemetryEvent;
 use telemetry::{
     ChannelContent, ChannelId, Lifetime, NodeId, Record, StreamDescriptor, StreamId, StreamOrigin,
-    TELEMETRY_PUBLISHER_NAME, TelemetryEndpoint, TelemetryProducer, TelemetryPublisherActor,
-    TelemetrySubscribe, TelemetrySubscription,
+    TelemetryEndpoint, TelemetryProducer, TelemetrySubscription,
 };
 
 use crate::codecs::register_myelin_actor_codecs;
@@ -43,10 +42,7 @@ use distribution::node::DistributedNodeConfig;
 use distribution::telemetry::{MembershipTransition, SwimProbeEvent};
 use distribution::types::{MemberState, NodeId as DistNodeId};
 use iroh::EndpointAddr;
-use iroh_driver::{
-    EDGE_ALPN, IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, TelemetryPublishHandle,
-    TelemetryQuicHeader, spawn_pull_server,
-};
+use iroh_driver::{EDGE_ALPN, IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, spawn_pull_server};
 use iroh_driver::{EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint};
 use parking_lot::Mutex;
 use serde_json::{Value, json};
@@ -1593,23 +1589,8 @@ fn run() -> Result<(), String> {
     let node_shutdown = |ds: &mut NodeTelemetry, phase: &str, status: &str, detail: Value| {
         emit_node_event(ds, &config, NODE_SHUTDOWN_CHANNEL, phase, status, detail)
     };
-    let telemetry_transport = driver.telemetry_publish_handle();
-    let telemetry_publisher = match stack
-        .runtime
-        .spawn(telemetry.publisher_actor(telemetry_transport))
-    {
-        Ok(actor) => actor,
-        Err(error) => {
-            node_boot(
-                &mut telemetry,
-                "telemetry_publisher",
-                "failed",
-                json!({"error":error.to_string()}),
-            );
-            return Err(format!("spawn telemetry publisher: {error}"));
-        }
-    };
-    stack.register_local_actor(driver.register_actor(telemetry_publisher, 1));
+    // Telemetry leaves this node exclusively through pull subscriptions served
+    // by `serve_telemetry_pulls` on `TELEMETRY_ALPN`; no publisher actor.
     let sampler_health_channel = telemetry.channel_by_name(NODE_SAMPLER_CHANNEL);
     let sampler_health_context = SamplerHealthContext::from_config(&config);
     spawn_host_gpu_sampler(
@@ -1632,12 +1613,6 @@ fn run() -> Result<(), String> {
         telemetry.channels.arena,
         Arc::clone(&arena_manager),
     );
-    node_boot(
-        &mut telemetry,
-        "telemetry_publisher",
-        "ready",
-        json!({"actor":telemetry_publisher,"name":TELEMETRY_PUBLISHER_NAME,"subscription_transport":"iroh"}),
-    );
     let worker_synthetic_id = format!(
         "myelin-worker-{}-{}-telemetry-preflight",
         config.logical_node_id, config.stage_index
@@ -1657,7 +1632,7 @@ fn run() -> Result<(), String> {
                 "producer_class":"rust-worker-node",
                 "synthetic_id":worker_synthetic_id,
                 "telemetry_endpoint":{
-                    "role":"worker-node-iroh-publisher",
+                    "role":"worker-node-iroh-pull-server",
                     "transport":"iroh-telemetry",
                     "endpoint_addr_mask":config.endpoint_addr_mask.as_str(),
                     "relay_mode":format!("{:?}", config.relay_mode),
@@ -1756,18 +1731,13 @@ fn run() -> Result<(), String> {
             "ready",
             json!({"framework":"none","workloads":"external_jobs"}),
         )?;
-        let mut pending_runtime_ready = PendingRuntimeReady::new(
-            &config,
-            advertised_self_endpoint.clone(),
-            node_actor,
-            telemetry_publisher,
-        );
+        let mut pending_runtime_ready =
+            PendingRuntimeReady::new(&config, advertised_self_endpoint.clone(), node_actor);
         let ready = json!({
             "type":"ready",
             "role":"node",
             "endpoint":advertised_self_endpoint.clone(),
             "node_actor":node_actor,
-            "telemetry_publisher":telemetry_publisher,
             "logical_node_id":config.logical_node_id,
             "stage_index":config.stage_index,
         });
@@ -1907,19 +1877,14 @@ fn run() -> Result<(), String> {
         }
     }
     let mut edge_runtime = WorkerEdgeRuntime::new(config.logical_node_id);
-    let mut pending_runtime_ready = PendingRuntimeReady::new(
-        &config,
-        advertised_self_endpoint.clone(),
-        node_actor,
-        telemetry_publisher,
-    );
+    let mut pending_runtime_ready =
+        PendingRuntimeReady::new(&config, advertised_self_endpoint.clone(), node_actor);
 
     let ready = json!({
         "type":"ready",
         "role":"node",
         "endpoint": advertised_self_endpoint.clone(),
         "node_actor": node_actor,
-        "telemetry_publisher": telemetry_publisher,
         "logical_node_id": config.logical_node_id,
         "stage_index": config.stage_index,
     });
@@ -2262,28 +2227,8 @@ impl NodeTelemetry {
             archive.drain(&self.by_id);
         }
     }
-
-    fn publisher_actor(&self, transport: TelemetryPublishHandle) -> TelemetryPublisherActor {
-        TelemetryPublisherActor::new(
-            Arc::clone(&self.endpoint),
-            move |subscribe: TelemetrySubscribe, subscription: TelemetrySubscription| {
-                let Ok(header) = TelemetryQuicHeader::from_snapshot(
-                    subscribe.flow_id,
-                    subscribe.token,
-                    subscription.snapshot(),
-                ) else {
-                    return;
-                };
-                transport.publish_subscription(
-                    subscribe.collector,
-                    header,
-                    subscription,
-                    Duration::from_millis(10),
-                );
-            },
-        )
-    }
 }
+
 fn serve_telemetry_pulls(
     driver: &IrohDriver,
     engine: &EngineHandle,
@@ -2378,7 +2323,6 @@ struct PendingRuntimeReady {
     stage_index: u32,
     endpoint: EndpointAddr,
     node_actor: ActorAddress,
-    telemetry_publisher: ActorAddress,
     coordinator: Option<DistNodeId>,
     readiness_id: u64,
     attempts: u32,
@@ -2389,19 +2333,13 @@ struct PendingRuntimeReady {
 }
 
 impl PendingRuntimeReady {
-    fn new(
-        config: &DeploymentConfig,
-        endpoint: EndpointAddr,
-        node_actor: ActorAddress,
-        telemetry_publisher: ActorAddress,
-    ) -> Self {
+    fn new(config: &DeploymentConfig, endpoint: EndpointAddr, node_actor: ActorAddress) -> Self {
         Self {
             run_id: config.run_id,
             node_id: config.logical_node_id,
             stage_index: config.stage_index,
             endpoint,
             node_actor,
-            telemetry_publisher,
             coordinator: config
                 .coordinator_endpoint
                 .as_ref()
@@ -2463,7 +2401,6 @@ impl PendingRuntimeReady {
                     stage_index: self.stage_index,
                     endpoint: self.endpoint.clone(),
                     node_actor: self.node_actor,
-                    telemetry_publisher: self.telemetry_publisher,
                     readiness_id: self.readiness_id,
                 },
             )

apps/myelin/src/orchestration/actor.rs

@@ -33,7 +33,6 @@ pub(crate) enum OrchestratorMsg {
         stage_index: u32,
         endpoint: EndpointAddr,
         node_actor: ActorAddress,
-        telemetry_publisher: ActorAddress,
         readiness_id: u64,
     },
     ObserveNodeRuntimeReadyAck {
@@ -154,7 +153,6 @@ pub(crate) enum OrchestratorReport {
         stage_index: u32,
         endpoint: EndpointAddr,
         node_actor: ActorAddress,
-        telemetry_publisher: ActorAddress,
         readiness_id: u64,
     },
     NodeRuntimeReadyAck {
@@ -326,7 +324,6 @@ impl ActorInterface for OrchestratorActor {
                 stage_index,
                 endpoint,
                 node_actor,
-                telemetry_publisher,
                 readiness_id,
             } => {
                 if let Some(report_to) = self.report_to {
@@ -338,7 +335,6 @@ impl ActorInterface for OrchestratorActor {
                             stage_index,
                             endpoint,
                             node_actor,
-                            telemetry_publisher,
                             readiness_id,
                         },
                     );

apps/myelin/src/orchestration/app.rs

@@ -5,6 +5,7 @@ use std::io::{BufRead, BufReader, Write};
 use std::os::fd::FromRawFd;
 use std::path::{Path, PathBuf};
 use std::process::Command;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, mpsc};
 use std::thread;
 use std::time::{Duration, Instant, SystemTime};
@@ -391,7 +392,7 @@ where
         json!({"actor":orchestrator_actor}),
     );
 
-    let stop_rx = stop_rx.unwrap_or_else(spawn_stop_listener);
+    let stop_signal = spawn_stop_listener(stop_rx);
 
     let provisioner = config.build_provisioner(stack.runtime.clone())?;
     bootstrap(
@@ -437,7 +438,7 @@ where
         obs_rx: &obs_rx,
         collector: &collector,
         orchestrator_reports: &orchestrator_reports,
-        stop_rx: &stop_rx,
+        stop_signal: stop_signal.as_ref(),
         dashboard: dashboard.as_ref(),
         orch_telemetry: &mut orch_telemetry,
         orch_stdio_rx: orch_stdio_rx.as_ref(),
@@ -1757,7 +1758,6 @@ impl Config {
 struct RuntimeReady {
     endpoint: EndpointAddr,
     node_actor: ActorAddress,
-    telemetry_publisher: ActorAddress,
     stage_index: u32,
     readiness_id: u64,
     swim_node_id: DistNodeId,
@@ -1800,14 +1800,14 @@ struct RuntimeReadyAckLoop<'a> {
     obs_rx: &'a mpsc::Receiver<PluginObservation>,
     collector: &'a FrameCollector,
     orchestrator_reports: &'a swactor::runtime::Inbox<OrchestratorReport>,
-    stop_rx: &'a mpsc::Receiver<()>,
+    stop_signal: &'a AtomicBool,
     dashboard: Option<&'a DashboardSupport>,
     orch_telemetry: &'a mut OrchTelemetry,
     orch_stdio_rx: Option<&'a mpsc::Receiver<OrchStdioLine>>,
     run_id: u64,
     orchestrator_node_id: u64,
     provider: &'a ProviderKind,
-    orchestrator_actor: ActorAddress,
+    engine: EngineHandle,
 }
 
 // synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
@@ -1823,7 +1823,7 @@ fn wait_for_runtime_ready_acks(
         obs_rx,
         collector,
         orchestrator_reports,
-        stop_rx,
+        stop_signal,
         dashboard,
         orch_telemetry,
         orch_stdio_rx,
@@ -1877,7 +1877,7 @@ fn wait_for_runtime_ready_acks(
             run_id,
             orchestrator_node_id,
         );
-        if stop_requested(stop_rx) {
+        if stop_requested(stop_signal) {
             return Err(
                 "shutdown requested while waiting for runtime-ready acknowledgements".to_owned(),
             );
@@ -2060,11 +2060,12 @@ fn wait_for_runtime_readies(
         obs_rx,
         collector,
         orchestrator_reports,
-        stop_rx,
+        stop_signal,
         dashboard,
         orch_telemetry,
         orch_stdio_rx,
         run_id,
+        engine,
         provider,
         ..
     } = ctx;
@@ -2100,7 +2101,7 @@ fn wait_for_runtime_readies(
             run_id,
             expected_node_ids.first().copied().unwrap_or(0),
         );
-        if stop_requested(stop_rx) {
+        if stop_requested(stop_signal) {
             return Err("shutdown requested while waiting for pipeline nodes ready".to_owned());
         }
         while let Ok(observation) = obs_rx.try_recv() {
@@ -2121,7 +2122,6 @@ fn wait_for_runtime_readies(
                 stage_index,
                 endpoint,
                 node_actor,
-                telemetry_publisher,
                 readiness_id,
             } = report
                 && report_run_id == run_id
@@ -2129,12 +2129,24 @@ fn wait_for_runtime_readies(
                 && cluster.current_attempt(node_id)
                     == Some(::provisioning::NodeAttemptId(readiness_id))
             {
+                // Bootstrap-owned telemetry: the first runtime-ready report
+                // for a node dials its pull server and retains the live
+                // subscription for the node's lifetime — the same contract
+                // as the xtask demo's NodeTelemetryCollector.
+                if !pending.contains_key(&node_id) {
+                    collector.subscribe_node(
+                        &engine,
+                        driver.endpoint(),
+                        endpoint.clone(),
+                        run_id,
+                        node_id,
+                    );
+                }
                 pending.insert(
                     node_id,
                     RuntimeReady {
                         endpoint: endpoint.clone(),
                         node_actor,
-                        telemetry_publisher,
                         stage_index,
                         readiness_id,
                         swim_node_id: DistNodeId(*endpoint.id.as_bytes()),
@@ -2254,33 +2266,37 @@ impl PluginObservationSink for ChannelObservationSink {
     }
 }
 
-fn stop_requested(stop_rx: &mpsc::Receiver<()>) -> bool {
-    stop_rx.try_recv().is_ok()
+fn stop_requested(stop_signal: &AtomicBool) -> bool {
+    stop_signal.load(Ordering::Acquire)
 }
 
 // top-level OS signal handling is process control, out of scope (ENGINE_SPEC.md §2)
 #[allow(clippy::disallowed_methods)]
-fn spawn_stop_listener() -> mpsc::Receiver<()> {
-    let (tx, rx) = mpsc::channel();
-    #[cfg(target_os = "linux")]
-    {
+fn spawn_stop_listener(external: Option<mpsc::Receiver<()>>) -> Arc<AtomicBool> {
+    let requested = Arc::new(AtomicBool::new(false));
+    let listener_requested = Arc::clone(&requested);
+    if let Some(external) = external {
         thread::spawn(move || {
-            let Ok(mut signals) = signal_hook::iterator::Signals::new([
-                signal_hook::consts::signal::SIGINT,
-                signal_hook::consts::signal::SIGTERM,
-            ]) else {
-                return;
-            };
-            if signals.forever().next().is_some() {
-                let _ = tx.send(());
+            if external.recv().is_ok() {
+                listener_requested.store(true, Ordering::Release);
             }
         });
+        return requested;
     }
-    #[cfg(not(target_os = "linux"))]
-    {
-        drop(tx);
-    }
-    rx
+
+    #[cfg(target_os = "linux")]
+    thread::spawn(move || {
+        let Ok(mut signals) = signal_hook::iterator::Signals::new([
+            signal_hook::consts::signal::SIGINT,
+            signal_hook::consts::signal::SIGTERM,
+        ]) else {
+            return;
+        };
+        if signals.forever().next().is_some() {
+            listener_requested.store(true, Ordering::Release);
+        }
+    });
+    requested
 }
 
 struct ServeCluster<'a> {
@@ -2289,7 +2305,7 @@ struct ServeCluster<'a> {
     obs_rx: &'a mpsc::Receiver<PluginObservation>,
     collector: &'a FrameCollector,
     orchestrator_reports: &'a swactor::runtime::Inbox<OrchestratorReport>,
-    stop_rx: &'a mpsc::Receiver<()>,
+    stop_signal: &'a AtomicBool,
     dashboard: Option<&'a DashboardSupport>,
     orch_telemetry: &'a mut OrchTelemetry,
     orch_stdio_rx: Option<&'a mpsc::Receiver<OrchStdioLine>>,
@@ -2345,14 +2361,14 @@ impl ServeCluster<'_> {
             obs_rx: self.obs_rx,
             collector: self.collector,
             orchestrator_reports: self.orchestrator_reports,
-            stop_rx: self.stop_rx,
+            stop_signal: self.stop_signal,
             dashboard: self.dashboard,
             orch_telemetry: self.orch_telemetry,
             orch_stdio_rx: self.orch_stdio_rx,
             run_id: self.run_id,
             orchestrator_node_id: self.orchestrator_node_id,
             provider: self.provider,
-            orchestrator_actor: self.orchestrator_actor,
+            engine: self.engine.clone(),
         }
     }
 
@@ -2494,13 +2510,6 @@ impl ServeCluster<'_> {
             .get(&logical_node_id)
             .cloned()
             .ok_or_else(|| format!("node {logical_node_id} did not announce runtime ready"))?;
-        self.collector.subscribe_node(
-            &self.engine,
-            self.driver.endpoint(),
-            ready.endpoint.clone(),
-            self.run_id,
-            logical_node_id,
-        );
         let acknowledged = wait_for_runtime_ready_acks(
             self.ack_context(),
             &[RuntimeReadyAckTarget {
@@ -2541,7 +2550,6 @@ impl ServeCluster<'_> {
                 endpoint: serde_json::to_string(&ready.endpoint)
                     .map_err(|error| format!("serialize node endpoint: {error}"))?,
                 node_actor: ready.node_actor,
-                telemetry_publisher: ready.telemetry_publisher,
                 swim_node_id: ready.swim_node_id,
                 stage_index: ready.stage_index,
                 readiness_id: ready.readiness_id,
@@ -2630,11 +2638,25 @@ impl ServeCluster<'_> {
             return;
         }
         let result = match command {
+            ControlCommand::Provision { count: 0, .. } => {
+                Err("provision count must be at least 1".to_owned())
+            }
+            ControlCommand::Provision { count, .. } if count > 8 => {
+                Err(format!("provision count {count} exceeds maximum 8"))
+            }
             ControlCommand::Provision { count, .. } => {
                 (0..count).try_for_each(|_| self.add_node().map(|_| ()))
             }
-            ControlCommand::Kill { node, .. } => {
-                parse_control_node_id(&node).and_then(|node_id| self.kill_node(node_id).map(|_| ()))
+            ControlCommand::Kill { node, .. } => parse_control_node_id(&node).and_then(|node_id| {
+                self.kill_node(node_id)?.then_some(()).ok_or_else(|| {
+                    format!("node {node_id} is not tracked or has no provision intent")
+                })
+            }),
+            ControlCommand::Remove { count: 0, .. } => {
+                Err("remove count must be at least 1".to_owned())
+            }
+            ControlCommand::Remove { count, .. } if count > 8 => {
+                Err(format!("remove count {count} exceeds maximum 8"))
             }
             ControlCommand::Remove { count, .. } => {
                 let mut ids = self
@@ -2669,17 +2691,31 @@ impl ServeCluster<'_> {
             stage_index,
             endpoint,
             node_actor,
-            telemetry_publisher,
             readiness_id,
         } = report
             && run_id == self.run_id
         {
+            // Bootstrap-owned telemetry: a new runtime generation announces a
+            // fresh pull server, so re-dial and retain the live subscription.
+            let known_readiness = self
+                .snapshot
+                .node(node_id)
+                .and_then(|node| node.runtime.as_ref())
+                .map(|runtime| runtime.readiness_id);
+            if known_readiness != Some(readiness_id) {
+                self.collector.subscribe_node(
+                    &self.engine,
+                    self.driver.endpoint(),
+                    endpoint.clone(),
+                    run_id,
+                    node_id,
+                );
+            }
             if let Some(node) = self.snapshot.node_mut(node_id) {
                 node.status = daemon::NodeStatus::Running;
                 node.runtime = Some(daemon::RuntimeFacts {
                     endpoint: serde_json::to_string(&endpoint).unwrap_or_default(),
                     node_actor,
-                    telemetry_publisher,
                     swim_node_id: DistNodeId(*endpoint.id.as_bytes()),
                     stage_index,
                     readiness_id,
@@ -2717,9 +2753,10 @@ impl ServeCluster<'_> {
     }
 
     fn teardown_if_requested(&mut self) -> Result<(), String> {
-        // Local process children cannot be adopted, so detaching them on an
-        // interactive shutdown only creates invisible orphan processes.
-        if self.destroy_on_exit || self.provider.as_str() == "process" {
+        // Local process and Docker nodes are run-scoped development resources:
+        // Ctrl+C must not leave invisible processes or containers behind.
+        // Remote providers remain adoptable unless explicitly destroyed.
+        if self.destroy_on_exit || matches!(self.provider.as_str(), "process" | "docker") {
             for (_, mut cluster) in std::mem::take(&mut self.live_clusters) {
                 cluster.stop()?;
             }
@@ -2798,7 +2835,7 @@ fn serve_cluster(mut ctx: ServeCluster<'_>) -> Result<(), String> {
         while let Ok(command) = ctx.control_rx.try_recv() {
             ctx.handle_dashboard_command(command);
         }
-        if stop_requested(ctx.stop_rx) {
+        if stop_requested(ctx.stop_signal) {
             break;
         }
         thread::sleep(PUMP_INTERVAL);

apps/myelin/src/orchestration/daemon.rs

@@ -7,7 +7,9 @@
 //! intent and facts so a restarted daemon can adopt what still exists and
 //! never silently re-provisions.
 
-use std::collections::{BTreeMap, BTreeSet};
+#[cfg(test)]
+use std::collections::BTreeMap;
+use std::collections::BTreeSet;
 use std::fs;
 use std::path::{Path, PathBuf};
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -39,7 +41,6 @@ pub(crate) enum NodeStatus {
 pub(crate) struct RuntimeFacts {
     pub endpoint: String,
     pub node_actor: ActorAddress,
-    pub telemetry_publisher: ActorAddress,
     pub swim_node_id: DistNodeId,
     pub stage_index: u32,
     pub readiness_id: u64,

apps/myelin/src/tests/job_runner_iroh.rs

@@ -9,8 +9,6 @@
 use std::collections::BTreeMap;
 use std::time::{Duration, Instant};
 
-use swactor::actor::Message;
-use swactor::runtime::Inbox;
 use swactor_engine::{Engine, TokioBackend, TokioConfig};
 use swactor_job_runner::{
     Job, JobDone, JobState, NodeJobActor, OrchestratorJobActor, OrchestratorJobMsg, Workspace,
@@ -79,10 +77,6 @@ fn job_runs_across_two_nodes_over_real_iroh() {
     // route to it through the converged directory.
     let (engine_b, driver_b, stack_b) = build_composition();
     let sender_b = stack_b.runtime.create_sender();
-    let done_b_echo = stack_b
-        .runtime
-        .new_inbox::<JobDone>()
-        .expect("worker echo inbox");
     // Placeholder orchestrator address: the real one is on A; the node only
     // needs it once the orchestrator submits. We point the node at A's
     // orchestrator after it exists (address is fixed below), but the node actor
@@ -112,17 +106,18 @@ fn job_runs_across_two_nodes_over_real_iroh() {
         .expect("spawn node job actor on B");
     stack_b.register_local_actor(driver_b.register_actor(job_actor, 1));
 
-    // Orchestrator (A) joins the worker (B) over iroh and waits for the directory
-    // to converge: A's route view must learn job_actor → B.
+    // Both control directions must be routable before Submit. A learning the
+    // worker route does not imply B has already learned the orchestrator route;
+    // submitting at that one-way boundary loses the first node event.
     driver_a_join(&stack_a, &_driver_a, &driver_b);
     stack_a.register_local_actor(_driver_a.register_actor(orch, 1));
 
     let converged = wait_until(CONVERGE_DEADLINE, || {
-        stack_a.route_view.read().unwrap().contains_key(&job_actor)
+        stack_a.route_owner(job_actor).is_some() && stack_b.route_owner(orch).is_some()
     });
     assert!(
         converged,
-        "directory did not converge: A never learned the worker's NodeJobActor"
+        "directory did not converge bidirectionally for orchestrator and worker actors"
     );
 
     let job = Job {

apps/myelin/src/tests/node_guarantees.rs

@@ -21,7 +21,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() {
         .expect("orchestrator inbox");
     let orchestrator = *orchestrator_inbox.addr();
     let node_actor = ActorAddress::new_random();
-    let telemetry_publisher = ActorAddress::new_random();
     let endpoint = EndpointAddr::new(SecretKey::from_bytes(&[9; 32]).public());
     let actor = runtime
         .spawn(NodeAgentActor::new(stage::NodeId(11), orchestrator, None))
@@ -36,7 +35,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() {
                 stage_index: 3,
                 endpoint: endpoint.clone(),
                 node_actor,
-                telemetry_publisher,
                 readiness_id: 99,
             },
         )
@@ -51,7 +49,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() {
             stage_index: 3,
             endpoint,
             node_actor,
-            telemetry_publisher,
             readiness_id: 99,
         })
     );

apps/myelin/src/tests/prompt_guarantees.rs

@@ -2,51 +2,24 @@
 
 use myelin::node::prompt_wire::PromptEvent;
 
-#[test]
-fn event_terminal_state_is_explicit() {
-    assert!(
-        !PromptEvent::TextDelta {
-            request_id: 1,
-            text: "a".to_owned(),
-        }
-        .is_terminal()
-    );
-    assert!(
-        PromptEvent::Done {
-            request_id: 1,
-            final_text: "a".to_owned(),
-            tokens_generated: 1,
-            elapsed_ms: 2,
-        }
-        .is_terminal()
-    );
-    assert!(
-        PromptEvent::Fault {
-            request_id: 1,
-            error: "boom".to_owned(),
-        }
-        .is_terminal()
-    );
+fn is_terminal(event: &PromptEvent) -> bool {
+    matches!(event, PromptEvent::Done { .. } | PromptEvent::Fault { .. })
 }
 
 #[test]
-fn event_request_ids_are_always_available() {
-    let events = [
-        PromptEvent::TextDelta {
-            request_id: 4,
-            text: String::new(),
-        },
-        PromptEvent::Done {
-            request_id: 5,
-            final_text: String::new(),
-            tokens_generated: 0,
-            elapsed_ms: 0,
-        },
-        PromptEvent::Fault {
-            request_id: 6,
-            error: String::new(),
-        },
-    ];
-    let ids: Vec<u64> = events.iter().map(PromptEvent::request_id).collect();
-    assert_eq!(ids, vec![4, 5, 6]);
+fn event_terminal_state_is_explicit() {
+    assert!(!is_terminal(&PromptEvent::TextDelta {
+        request_id: 1,
+        text: "a".to_owned(),
+    }));
+    assert!(is_terminal(&PromptEvent::Done {
+        request_id: 1,
+        final_text: "a".to_owned(),
+        tokens_generated: 1,
+        elapsed_ms: 2,
+    }));
+    assert!(is_terminal(&PromptEvent::Fault {
+        request_id: 1,
+        error: "boom".to_owned(),
+    }));
 }

crates/dashboard/src/control_plane_page.html

@@ -59,7 +59,7 @@
     #status { font-family: var(--mono); font-size: 13px; color: var(--amber); }
     a.node-card { color: inherit; text-decoration: none; display: block; }
     .node-card.orchestrator { grid-column: 1 / -1; border-color: var(--amber); background: var(--card-hover); }
-    .node-card .role { display: inline-flex; margin-left: 8px; padding: 2px 6px; border: 1px solid var(--amber); border-radius: var(--r); color: var(--amber); font: 600 10px/1.4 var(--mono); text-transform: uppercase; letter-spacing: .07em; vertical-align: 2px; }
+    .node-card.orchestrator .role { margin-bottom: 6px; color: var(--muted); font: 600 9px/1.2 var(--mono); text-transform: uppercase; letter-spacing: .12em; }
     :is(a, button, input, summary):focus-visible, tr[data-addr]:focus-visible { outline: 2px solid var(--cyan); outline-offset: 2px; }
     .grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(300px, 1fr)); gap: 12px; }
     .node-card { background: var(--panel); border: 1px solid var(--border); border-radius: var(--r); padding: 14px; cursor: pointer; transition: border-color var(--t), background var(--t); }
@@ -262,7 +262,8 @@ function nodeCard(node) {
   // supervisors; it renders as a pass-filled span so the card markup — and
   // with it the whole grid — stays byte-identical between polls.
   return `<a class="node-card${isOrchestrator ? ' orchestrator' : ''}" data-stream="${esc(node.stream.key)}" data-live="${node.live ? 'true' : 'false'}" href="?stream=${encodeURIComponent(node.stream.key)}">
-    <h3>${esc(node.stream.node)}${role} <span class="muted" style="font-size:12px">life ${fmt(node.stream.life)}</span></h3>
+    ${role}
+    <h3>${esc(node.stream.node)} <span class="muted" style="font-size:12px">life ${fmt(node.stream.life)}</span></h3>
     <div class="meta">${proc}seen <span data-seen></span> ago · <span data-stats></span>${label}</div>
     ${summary.poisoned ? `<div class="notice err">${fmt(summary.poisoned)} poisoned actor(s)</div>` : ''}
   </a>`;

crates/dashboard/src/demo_control_page.html

@@ -62,6 +62,7 @@
     .seg .seg-v { position: relative; }
     .seg-sm { font-size: 15px; }
     .blink { animation: seg-blink 1.1s steps(2, start) infinite; }
+    .counter[hidden] { display: none; }
     @keyframes seg-blink { to { visibility: hidden; } }
     .controls { display: flex; align-items: center; flex-wrap: wrap; gap: 8px; }
     .controls input {

crates/iroh-driver/src/telemetry_transport.rs

@@ -104,11 +104,11 @@ pub fn spawn_pull_server(
     });
 }
 
-/// 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.
+/// Supervisor side: retain a pull subscription to a node on `TELEMETRY_ALPN`.
+///
+/// A transport interruption reconnects with bounded backoff. Returning after
+/// the first EOF leaves a healthy node permanently stale, which is especially
+/// easy to trigger while several freshly-bootstrapped nodes answer at once.
 pub fn spawn_pull_collector(
     engine: &EngineHandle,
     endpoint: Endpoint,
@@ -119,35 +119,74 @@ pub fn spawn_pull_collector(
     fanout: std::sync::Arc<telemetry::DeliveryFanout>,
     on_header: std::sync::mpsc::Sender<TelemetryQuicHeader>,
 ) {
+    let engine_handle = engine.clone();
     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);
+        let mut retry_delay = Duration::from_millis(250);
+        loop {
+            match collect_pull_once(
+                &endpoint, &peer, flow_id, &token, &request, &fanout, &on_header,
+            )
+            .await
+            {
+                Ok(()) => return,
+                Err(error) => {
+                    eprintln!(
+                        "telemetry-pull: {peer_id}: {error}; retrying in {} ms",
+                        retry_delay.as_millis()
+                    );
+                }
+            }
+            engine_handle.timer(retry_delay).await;
+            retry_delay = retry_delay
+                .checked_mul(2)
+                .unwrap_or(Duration::from_secs(5))
+                .min(Duration::from_secs(5));
         }
     });
 }
+
+async fn collect_pull_once(
+    endpoint: &Endpoint,
+    peer: &EndpointAddr,
+    flow_id: [u8; 16],
+    token: &[u8],
+    request: &telemetry::SubscriptionRequest,
+    fanout: &telemetry::DeliveryFanout,
+    on_header: &std::sync::mpsc::Sender<TelemetryQuicHeader>,
+) -> Result<(), String> {
+    let conn = endpoint
+        .connect(peer.clone(), TELEMETRY_ALPN)
+        .await
+        .map_err(|error| format!("connect failed: {error}"))?;
+    let mut req = conn
+        .open_uni()
+        .await
+        .map_err(|error| format!("open request stream failed: {error}"))?;
+    write_pull_request(&mut req, flow_id, token, request)
+        .await
+        .map_err(|error| format!("write request failed: {error}"))?;
+    let mut recv = conn
+        .accept_uni()
+        .await
+        .map_err(|error| format!("no answer stream: {error}"))?;
+    let header = read_header(&mut recv)
+        .await
+        .map_err(|error| format!("answer header unreadable: {error}"))?;
+    if on_header.send(header.clone()).is_err() {
+        return Ok(());
+    }
+    let stream = header.stream;
+    loop {
+        match read_next_event(&mut recv, &stream).await {
+            Ok(Some(event)) => {
+                fanout.publish(event);
+            }
+            Ok(None) => return Err("answer stream closed".to_owned()),
+            Err(error) => return Err(format!("read answer stream failed: {error}")),
+        }
+    }
+}
 const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024;
 
 type BoxError = Box<dyn Error + Send + Sync + 'static>;