diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs
index 03c2238..609e4cd 100644
--- a/apps/myelin/src/orchestration/app.rs
+++ b/apps/myelin/src/orchestration/app.rs
@@ -2353,6 +2353,9 @@ fn parse_control_node_id(value: &str) -> Result<u64, String> {
         .ok_or_else(|| format!("control node id {value:?} contains no numeric logical id"))
 }
 
+fn destroys_provider_resources_on_exit(provider: &str, explicitly_requested: bool) -> bool {
+    explicitly_requested || provider == "process"
+}
 impl ServeCluster<'_> {
     fn ack_context(&mut self) -> RuntimeReadyAckLoop<'_> {
         RuntimeReadyAckLoop {
@@ -2427,6 +2430,11 @@ impl ServeCluster<'_> {
                                     "snapshot node {node_id} telemetry endpoint is invalid: {error}"
                                 )
                             })?;
+                        // The daemon identity survives restart but its direct
+                        // socket addresses do not. Dial the worker from the
+                        // fresh endpoint so SWIM and actor routing can
+                        // converge over the new connection.
+                        self.driver.join(std::slice::from_ref(&endpoint));
                         self.collector.subscribe_node(
                             &self.engine,
                             self.driver.endpoint(),
@@ -2491,6 +2499,18 @@ impl ServeCluster<'_> {
             logical_node_id,
             0,
         )?;
+        // Persist the complete provider intent before creating anything. If
+        // the daemon dies during bootstrap, restart can recover the labeled
+        // resource by run/node identity even before its attempt is known.
+        self.snapshot.upsert_node(daemon::SnapshotNode {
+            logical_node_id,
+            spec: Some(spec.clone()),
+            provider_ref: Some(self.provisioner.provider_ref_for(&spec)),
+            status: daemon::NodeStatus::Running,
+            runtime: None,
+            last_seen_unix_ms: daemon::unix_ms_now(),
+        });
+        self.save_snapshot()?;
         self.emit_command_event(
             logical_node_id,
             ProvisionEventKind::ProvisionStart,
@@ -2505,19 +2525,45 @@ impl ServeCluster<'_> {
             self.sink.clone(),
         )?;
         let readies =
-            wait_for_runtime_readies(self.ack_context(), &[logical_node_id], &mut cluster)?;
+            match wait_for_runtime_readies(self.ack_context(), &[logical_node_id], &mut cluster) {
+                Ok(readies) => readies,
+                Err(error) => {
+                    if stop_requested(self.stop_signal)
+                        && !destroys_provider_resources_on_exit(
+                            self.provider.as_str(),
+                            self.destroy_on_exit,
+                        )
+                    {
+                        cluster.detach();
+                    }
+                    return Err(error);
+                }
+            };
         let ready = readies
             .get(&logical_node_id)
             .cloned()
             .ok_or_else(|| format!("node {logical_node_id} did not announce runtime ready"))?;
-        let acknowledged = wait_for_runtime_ready_acks(
+        let acknowledged = match wait_for_runtime_ready_acks(
             self.ack_context(),
             &[RuntimeReadyAckTarget {
                 node_id: logical_node_id,
                 ready: ready.clone(),
             }],
             &mut cluster,
-        )?;
+        ) {
+            Ok(acknowledged) => acknowledged,
+            Err(error) => {
+                if stop_requested(self.stop_signal)
+                    && !destroys_provider_resources_on_exit(
+                        self.provider.as_str(),
+                        self.destroy_on_exit,
+                    )
+                {
+                    cluster.detach();
+                }
+                return Err(error);
+            }
+        };
         if !acknowledged {
             return Err(format!(
                 "node {logical_node_id} changed attempt before runtime-ready acknowledgement"
@@ -2725,9 +2771,89 @@ impl ServeCluster<'_> {
         }
     }
 
+    fn recover_runtime_from_bootstrap(
+        &mut self,
+        observation: &PluginObservation,
+    ) -> Result<bool, String> {
+        let PluginObservation::TelemetryFrame {
+            run_id,
+            node_id,
+            channel,
+            payload,
+        } = observation
+        else {
+            return Ok(false);
+        };
+        if *run_id != self.run_id
+            || channel != "myelin.node.bootstrap"
+            || self
+                .snapshot
+                .node(*node_id)
+                .is_none_or(|node| node.runtime.is_some())
+        {
+            return Ok(false);
+        }
+        let event: Value = serde_json::from_str(payload)
+            .map_err(|error| format!("parse node {node_id} bootstrap telemetry: {error}"))?;
+        if event.get("phase").and_then(Value::as_str) != Some("runtime_ready_local")
+            || event.get("status").and_then(Value::as_str) != Some("ready")
+        {
+            return Ok(false);
+        }
+        let detail = event
+            .get("detail")
+            .ok_or_else(|| format!("node {node_id} runtime-ready event has no detail"))?;
+        let endpoint: EndpointAddr = serde_json::from_value(
+            detail
+                .get("endpoint")
+                .cloned()
+                .ok_or_else(|| format!("node {node_id} runtime-ready event has no endpoint"))?,
+        )
+        .map_err(|error| format!("parse node {node_id} runtime-ready endpoint: {error}"))?;
+        let node_actor = serde_json::from_value(
+            detail
+                .get("node_actor")
+                .cloned()
+                .ok_or_else(|| format!("node {node_id} runtime-ready event has no actor"))?,
+        )
+        .map_err(|error| format!("parse node {node_id} runtime-ready actor: {error}"))?;
+        let stage_index = detail
+            .get("stage_index")
+            .and_then(Value::as_u64)
+            .and_then(|value| u32::try_from(value).ok())
+            .ok_or_else(|| format!("node {node_id} runtime-ready event has no stage index"))?;
+        let readiness_id = detail
+            .get("readiness_id")
+            .and_then(Value::as_u64)
+            .ok_or_else(|| format!("node {node_id} runtime-ready event has no readiness id"))?;
+
+        self.driver.join(std::slice::from_ref(&endpoint));
+        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)
+                    .map_err(|error| format!("serialize node {node_id} endpoint: {error}"))?,
+                node_actor,
+                swim_node_id: DistNodeId(*endpoint.id.as_bytes()),
+                stage_index,
+                readiness_id,
+            });
+            node.last_seen_unix_ms = daemon::unix_ms_now();
+        }
+        Ok(true)
+    }
+
     fn drain_observations(&mut self) -> Result<(), String> {
         let mut dirty = false;
         while let Ok(observation) = self.obs_rx.try_recv() {
+            dirty |= self.recover_runtime_from_bootstrap(&observation)?;
             emit_plugin_observation(
                 self.orch_telemetry,
                 self.dashboard,
@@ -2753,10 +2879,9 @@ impl ServeCluster<'_> {
     }
 
     fn teardown_if_requested(&mut self) -> Result<(), String> {
-        // 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") {
+        // Process-provider children cannot be adopted. Durable provider
+        // resources survive unless teardown was explicitly requested.
+        if destroys_provider_resources_on_exit(self.provider.as_str(), self.destroy_on_exit) {
             for (_, mut cluster) in std::mem::take(&mut self.live_clusters) {
                 cluster.stop()?;
             }
@@ -3220,3 +3345,16 @@ where
         .parse::<T>()
         .map_err(|e| format!("invalid {name}={value:?}: {e}"))
 }
+
+#[cfg(test)]
+mod lifecycle_policy_tests {
+    use super::destroys_provider_resources_on_exit;
+
+    #[test]
+    fn durable_providers_survive_unrequested_daemon_shutdown() {
+        assert!(!destroys_provider_resources_on_exit("docker", false));
+        assert!(!destroys_provider_resources_on_exit("vastai", false));
+        assert!(destroys_provider_resources_on_exit("process", false));
+        assert!(destroys_provider_resources_on_exit("docker", true));
+    }
+}
diff --git a/apps/myelin/src/orchestration/provisioning.rs b/apps/myelin/src/orchestration/provisioning.rs
index 3278f5f..a36243e 100644
--- a/apps/myelin/src/orchestration/provisioning.rs
+++ b/apps/myelin/src/orchestration/provisioning.rs
@@ -34,7 +34,6 @@ struct LocalDockerNode {
     spec: NodeProvisionSpec,
     sink: PluginSink,
     container_name: String,
-    stdin: Option<ChildStdin>,
 }
 
 pub(crate) struct LocalProcessPlugin {
@@ -89,6 +88,7 @@ fn docker_container_labels(prefix: &str, spec: &NodeProvisionSpec) -> Vec<String
         format!("myelin.daemon={prefix}"),
         format!("myelin.run={}", spec.run_id),
         format!("myelin.node={}", spec.node_id),
+        format!("myelin.attempt={}", spec.attempt_id),
     ]
 }
 
@@ -165,6 +165,58 @@ fn docker_labeled_containers(prefix: &str) -> Result<Vec<String>, String> {
         .collect())
 }
 
+fn docker_containers_for_spec(
+    prefix: &str,
+    spec: &NodeProvisionSpec,
+) -> Result<Vec<String>, String> {
+    let output = Command::new("docker")
+        .args(["ps", "-a"])
+        .arg("--filter")
+        .arg(format!("label=myelin.daemon={prefix}"))
+        .arg("--filter")
+        .arg(format!("label=myelin.run={}", spec.run_id))
+        .arg("--filter")
+        .arg(format!("label=myelin.node={}", spec.node_id))
+        .args(["--format", "{{.Names}}"])
+        .output()
+        .map_err(|error| format!("list Docker containers for node {}: {error}", spec.node_id))?;
+    if !output.status.success() {
+        return Err(format!(
+            "list Docker containers for node {} exited with {}: {}",
+            spec.node_id,
+            output.status,
+            String::from_utf8_lossy(&output.stderr).trim()
+        ));
+    }
+    Ok(String::from_utf8_lossy(&output.stdout)
+        .lines()
+        .map(str::trim)
+        .filter(|line| !line.is_empty())
+        .map(str::to_owned)
+        .collect())
+}
+
+fn docker_container_for_spec(
+    prefix: &str,
+    spec: &NodeProvisionSpec,
+) -> Result<Option<String>, String> {
+    let expected = docker_container_name(prefix, spec);
+    if !docker_container_is_absent(&expected)? {
+        return Ok(Some(expected));
+    }
+    let matches = docker_containers_for_spec(prefix, spec)?;
+    match matches.as_slice() {
+        [] => Ok(None),
+        [container] => Ok(Some(container.clone())),
+        _ => Err(format!(
+            "multiple Docker containers match run {} node {}: {}",
+            spec.run_id,
+            spec.node_id,
+            matches.join(", ")
+        )),
+    }
+}
+
 fn docker_mount_arg(mount: &ProviderMount) -> String {
     let mut arg = format!(
         "type=bind,src={},dst={}",
@@ -587,7 +639,6 @@ impl ProvisionPlugin for LocalDockerPlugin {
                 container_name: docker_container_name(&self.container_name_prefix, &spec),
                 spec,
                 sink,
-                stdin: None,
             },
         );
         Ok(handle)
@@ -598,10 +649,13 @@ impl ProvisionPlugin for LocalDockerPlugin {
     fn start_bootstrap(&mut self, handle: &PluginNodeHandle) -> Result<(), String> {
         let node = self
             .nodes
-            .get_mut(&handle.id)
+            .get(&handle.id)
             .ok_or_else(|| format!("Docker node handle {} is absent", handle.id))?;
-        if node.stdin.is_some() {
-            return Ok(());
+        if !docker_container_is_absent(&node.container_name)? {
+            return Err(format!(
+                "Docker container {} already exists before bootstrap",
+                node.container_name
+            ));
         }
         let spec = node.spec.clone();
         let sink = node.sink.clone();
@@ -609,7 +663,7 @@ impl ProvisionPlugin for LocalDockerPlugin {
         let mut command = Command::new("docker");
         command
             .arg("run")
-            .arg("--rm")
+            .arg("-d")
             .arg("--add-host")
             .arg("host.docker.internal:host-gateway")
             .arg("--name")
@@ -617,7 +671,6 @@ impl ProvisionPlugin for LocalDockerPlugin {
         for label in docker_container_labels(&self.container_name_prefix, &spec) {
             command.arg("--label").arg(label);
         }
-        command.arg("-i");
         let docker_gpus = spec
             .env
             .iter()
@@ -652,46 +705,18 @@ impl ProvisionPlugin for LocalDockerPlugin {
         for arg in &spec.args {
             command.arg(arg);
         }
-        let mut child = command
-            .stdin(Stdio::piped())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-            .map_err(|e| format!("spawn Docker node {}: {e}", spec.node_id))?;
-
-        let (Some(stdin), Some(stdout), Some(stderr)) =
-            (child.stdin.take(), child.stdout.take(), child.stderr.take())
-        else {
-            let _ = child.kill();
-            let _ = child.wait();
-            let _ = Command::new("docker")
-                .arg("rm")
-                .arg("-f")
-                .arg(&container_name)
-                .stdout(Stdio::null())
-                .stderr(Stdio::null())
-                .status();
+        let output = command
+            .output()
+            .map_err(|error| format!("run Docker node {}: {error}", spec.node_id))?;
+        if !output.status.success() {
             return Err(format!(
-                "Docker node {} did not expose piped stdio",
-                spec.node_id
+                "run Docker node {} exited with {}: {}",
+                spec.node_id,
+                output.status,
+                String::from_utf8_lossy(&output.stderr).trim()
             ));
-        };
-        node.stdin = Some(stdin);
-        spawn_stdout_reader(spec.clone(), sink.clone(), stdout);
-        spawn_stderr_reader(spec.clone(), sink.clone(), stderr);
-        thread::spawn(move || match child.wait() {
-            Ok(status) => sink.observe(PluginObservation::Exited {
-                run_id: spec.run_id,
-                node_id: spec.node_id,
-                status: status.code(),
-            }),
-            Err(error) => sink.observe(PluginObservation::Failed {
-                run_id: spec.run_id,
-                node_id: spec.node_id,
-                reason: format!("wait Docker node: {error}"),
-            }),
-        });
-        Ok(())
+        }
+        observe_docker_container(spec, sink, container_name, "all")
     }
 
     fn complete_bootstrap(&mut self, _handle: &PluginNodeHandle) -> Result<(), String> {
@@ -699,44 +724,10 @@ impl ProvisionPlugin for LocalDockerPlugin {
     }
 
     fn stop_node(&mut self, handle: &PluginNodeHandle) -> Result<(), String> {
-        let Some(mut node) = self.nodes.remove(&handle.id) else {
+        let Some(node) = self.nodes.remove(&handle.id) else {
             return Ok(());
         };
-        let Some(stdin) = node.stdin.as_mut() else {
-            return Ok(());
-        };
-        let _ = writeln!(stdin, "shutdown");
-        let _ = stdin.flush();
-        let result = match Command::new("docker")
-            .arg("stop")
-            .arg("-t")
-            .arg("2")
-            .arg(&node.container_name)
-            .stdout(Stdio::null())
-            .stderr(Stdio::null())
-            .status()
-        {
-            Ok(status) if status.success() => Ok(()),
-            Ok(status) => match docker_container_is_absent(&node.container_name) {
-                Ok(true) => Ok(()),
-                Ok(false) => Err(format!(
-                    "docker stop {} exited with {status}",
-                    node.container_name
-                )),
-                Err(inspect_error) => Err(format!(
-                    "docker stop {} exited with {status}; {inspect_error}",
-                    node.container_name
-                )),
-            },
-            Err(error) => match docker_container_is_absent(&node.container_name) {
-                Ok(true) => Ok(()),
-                Ok(false) => Err(format!("docker stop {}: {error}", node.container_name)),
-                Err(inspect_error) => Err(format!(
-                    "docker stop {}: {error}; {inspect_error}",
-                    node.container_name
-                )),
-            },
-        };
+        let result = remove_docker_container(&node.container_name);
         if result.is_err() {
             self.nodes.insert(handle.id, node);
         }
@@ -748,10 +739,10 @@ impl ProvisionPlugin for LocalDockerPlugin {
         spec: &NodeProvisionSpec,
         sink: PluginSink,
     ) -> Result<Option<AdoptedNode>, String> {
-        let container_name = docker_container_name(&self.container_name_prefix, spec);
-        if docker_container_is_absent(&container_name)? {
+        let Some(container_name) = docker_container_for_spec(&self.container_name_prefix, spec)?
+        else {
             return Ok(None);
-        }
+        };
         let running = docker_container_is_running(&container_name)?;
         let handle = PluginNodeHandle {
             id: self.next_handle_id,
@@ -759,52 +750,18 @@ impl ProvisionPlugin for LocalDockerPlugin {
         };
         self.next_handle_id = self.next_handle_id.wrapping_add(1).max(1);
         if running {
-            // Follow the adopted container's logs and wait for its exit; no
-            // lifecycle action is taken — adoption is observation-only.
-            for (_stream, stdout_flag) in [("stdout", true), ("stderr", false)] {
-                let mut logs = Command::new("docker");
-                logs.arg("logs")
-                    .arg("-f")
-                    .arg("--tail")
-                    .arg("0")
-                    .arg(if stdout_flag { "--stdout" } else { "--stderr" })
-                    .arg(&container_name);
-                if let Ok(child) = logs.stdout(Stdio::piped()).stderr(Stdio::null()).spawn() {
-                    if let Some(out) = child.stdout {
-                        if stdout_flag {
-                            spawn_stdout_reader(spec.clone(), sink.clone(), out);
-                        } else {
-                            spawn_stderr_reader(spec.clone(), sink.clone(), out);
-                        }
-                    }
-                }
-            }
-            let wait_spec = spec.clone();
-            let wait_sink = sink.clone();
-            let wait_name = container_name.clone();
-            thread::spawn(move || {
-                let status = Command::new("docker").arg("wait").arg(&wait_name).output();
-                let code = status.ok().and_then(|output| {
-                    String::from_utf8_lossy(&output.stdout)
-                        .trim()
-                        .parse::<i32>()
-                        .ok()
-                });
-                wait_sink.observe(PluginObservation::Exited {
-                    run_id: wait_spec.run_id,
-                    node_id: wait_spec.node_id,
-                    status: code,
-                });
-            });
+            // Replay this container's bootstrap log into the fresh daemon,
+            // then follow new output. The replay supplies runtime facts when
+            // the prior daemon died before persisting readiness.
+            observe_docker_container(spec.clone(), sink.clone(), container_name.clone(), "all")?;
         }
-        let adopted_name = docker_container_name(&self.container_name_prefix, spec);
+        let adopted_name = container_name.clone();
         self.nodes.insert(
             handle.id,
             LocalDockerNode {
                 spec: spec.clone(),
                 sink: sink.clone(),
                 container_name,
-                stdin: None,
             },
         );
         sink.observe(PluginObservation::TelemetryFrame {
@@ -834,20 +791,11 @@ impl ProvisionPlugin for LocalDockerPlugin {
     }
 
     fn stop_by_spec(&mut self, spec: &NodeProvisionSpec, sink: PluginSink) -> Result<bool, String> {
-        let container_name = docker_container_name(&self.container_name_prefix, spec);
-        if docker_container_is_absent(&container_name)? {
+        let Some(container_name) = docker_container_for_spec(&self.container_name_prefix, spec)?
+        else {
             return Ok(false);
-        }
-        let status = Command::new("docker")
-            .arg("rm")
-            .arg("-f")
-            .arg(&container_name)
-            .stdout(Stdio::null())
-            .stderr(Stdio::null())
-            .status()
-            .map_err(|error| format!("docker rm {container_name}: {error}"))?;
-        let removed =
-            status.success() || matches!(docker_container_is_absent(&container_name), Ok(true));
+        };
+        remove_docker_container(&container_name)?;
         sink.observe(PluginObservation::TelemetryFrame {
             run_id: spec.run_id,
             node_id: spec.node_id,
@@ -856,16 +804,12 @@ impl ProvisionPlugin for LocalDockerPlugin {
                 "type":"DockerContainerRemoved",
                 "provider":"Docker",
                 "container":container_name,
-                "removed":removed,
-                "exit_ok":status.success(),
+                "removed":true,
+                "exit_ok":true,
             })
             .to_string(),
         });
-        if removed {
-            Ok(true)
-        } else {
-            Err(format!("docker rm {container_name} exited with {status}"))
-        }
+        Ok(true)
     }
 
     fn detach_all(&mut self) {
@@ -873,6 +817,71 @@ impl ProvisionPlugin for LocalDockerPlugin {
     }
 }
 
+fn remove_docker_container(container_name: &str) -> Result<(), String> {
+    let status = Command::new("docker")
+        .arg("rm")
+        .arg("-f")
+        .arg(container_name)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .status()
+        .map_err(|error| format!("docker rm {container_name}: {error}"))?;
+    if status.success() || matches!(docker_container_is_absent(container_name), Ok(true)) {
+        Ok(())
+    } else {
+        Err(format!("docker rm {container_name} exited with {status}"))
+    }
+}
+
+fn observe_docker_container(
+    spec: NodeProvisionSpec,
+    sink: PluginSink,
+    container_name: String,
+    tail: &str,
+) -> Result<(), String> {
+    let mut logs = Command::new("docker");
+    let mut logs = logs
+        .arg("logs")
+        .arg("--follow")
+        .arg("--tail")
+        .arg(tail)
+        .arg(&container_name)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|error| format!("follow Docker logs for {container_name}: {error}"))?;
+    let stdout = logs
+        .stdout
+        .take()
+        .ok_or_else(|| format!("Docker logs for {container_name} has no stdout"))?;
+    let stderr = logs
+        .stderr
+        .take()
+        .ok_or_else(|| format!("Docker logs for {container_name} has no stderr"))?;
+    spawn_stdout_reader(spec.clone(), sink.clone(), stdout);
+    spawn_stderr_reader(spec.clone(), sink.clone(), stderr);
+    thread::spawn(move || {
+        let _ = logs.wait();
+    });
+
+    let wait_name = container_name;
+    thread::spawn(move || {
+        let status = Command::new("docker").arg("wait").arg(&wait_name).output();
+        let code = status.ok().and_then(|output| {
+            String::from_utf8_lossy(&output.stdout)
+                .trim()
+                .parse::<i32>()
+                .ok()
+        });
+        sink.observe(PluginObservation::Exited {
+            run_id: spec.run_id,
+            node_id: spec.node_id,
+            status: code,
+        });
+    });
+    Ok(())
+}
+
 fn spawn_stdout_reader(
     spec: NodeProvisionSpec,
     sink: PluginSink,