Block capsule traffic from leaking to private IP ranges or other capsules

Adds host-level iptables rules on host agent startup that stop sandbox
traffic from escaping onto the private network or reaching another
sandbox directly, plus gitignores local tokensave tool state.

.gitignore

@@ -56,3 +56,6 @@ internal/dashboard/static/*
 # Added by code-review-graph
 .code-review-graph/
 .mcp.json
+
+# tokensave
+.tokensave/

cmd/host-agent/main.go

@@ -19,6 +19,7 @@ import (
 	"git.omukk.dev/wrenn/wrenn/internal/hostagent"
 	"git.omukk.dev/wrenn/wrenn/pkg/auth"
 	"git.omukk.dev/wrenn/wrenn/pkg/logging"
+	"git.omukk.dev/wrenn/wrenn/pkg/network"
 	"git.omukk.dev/wrenn/wrenn/pkg/sandbox"
 	"git.omukk.dev/wrenn/wrenn/proto/hostagent/gen/hostagentv1connect"
 )
@@ -54,6 +55,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Install the egress guard: block capsule traffic to private ranges from
+	// leaving the public NIC (which would leak private-destined packets onto the
+	// upstream network and read as a port scan) and deny cross-capsule reach.
+	if err := network.EnsureEgressGuard(); err != nil {
+		slog.Error("failed to install network egress guard", "error", err)
+		os.Exit(1)
+	}
+
 	listenAddr := envOrDefault("WRENN_HOST_LISTEN_ADDR", ":50051")
 	cpURL := os.Getenv("WRENN_CP_URL")
 	credsFile := filepath.Join(rootDir, "host-credentials.json")

envd-rs/Cargo.lock

@@ -529,7 +529,7 @@ dependencies = [
 
 [[package]]
 name = "envd"
-version = "0.5.0"
+version = "0.6.0"
 dependencies = [
  "async-stream",
  "axum",

envd-rs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "envd"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2024"
 rust-version = "1.95"
 

envd-rs/src/http/init.rs

@@ -84,14 +84,8 @@ pub async fn post_init(
             let mut err = state.mem_preload_error.lock().unwrap();
             state.mem_preload_generation.fetch_add(1, Ordering::SeqCst);
             state.mem_preload_cancel.store(false, Ordering::SeqCst);
-            state.mem_preload_done.store(false, Ordering::SeqCst);
             state.mem_preload_started.store(false, Ordering::SeqCst);
-            state.mem_preload_regions.store(0, Ordering::SeqCst);
-            state.mem_preload_pages.store(0, Ordering::SeqCst);
-            state.mem_preload_bytes.store(0, Ordering::SeqCst);
-            state.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
-            state.mem_preload_source.store(0, Ordering::SeqCst);
-            *err = None;
+            state.reset_preload_run(&mut err);
         }
 
         if let Some(ref port_sub) = state.port_subsystem {
@@ -203,11 +197,15 @@ pub async fn post_init(
         futures::future::join_all(futs).await;
     }
 
-    // Set sandbox/template metadata from request body.
+    // Set sandbox/template metadata from request body. Deliberately NOT
+    // written into envd's own process environment: std::env::set_var is
+    // undefined behavior with the multi-threaded runtime live (concurrent
+    // getenv from any thread races the environ rewrite), and nothing needs
+    // it there — spawned processes get these via defaults.env_vars, and
+    // out-of-band consumers (e.g. the `envd ports` subcommand's
+    // read_identity) fall back to the run files.
     if let Some(ref id) = init_req.sandbox_id {
         tracing::debug!(sandbox_id = %id, "setting sandbox ID from init request");
-        // SAFETY: envd is single-threaded at init time; no concurrent env reads.
-        unsafe { std::env::set_var("WRENN_SANDBOX_ID", id) };
         write_run_file(".WRENN_SANDBOX_ID", id);
         state
             .defaults
@@ -216,8 +214,6 @@ pub async fn post_init(
     }
     if let Some(ref id) = init_req.template_id {
         tracing::debug!(template_id = %id, "setting template ID from init request");
-        // SAFETY: envd is single-threaded at init time; no concurrent env reads.
-        unsafe { std::env::set_var("WRENN_TEMPLATE_ID", id) };
         write_run_file(".WRENN_TEMPLATE_ID", id);
         state
             .defaults
@@ -227,8 +223,6 @@ pub async fn post_init(
     if let Some(ref domain) = init_req.proxy_domain {
         if !domain.is_empty() {
             tracing::debug!(proxy_domain = %domain, "setting proxy domain from init request");
-            // SAFETY: envd is single-threaded at init time; no concurrent env reads.
-            unsafe { std::env::set_var("WRENN_PROXY_DOMAIN", domain) };
             write_run_file(".WRENN_PROXY_DOMAIN", domain);
             state
                 .defaults

envd-rs/src/http/memory.rs

@@ -10,8 +10,11 @@
 // handler fills the page from the source memory-ranges file.
 //
 // Wire protocol:
-//   POST /memory/preload         — starts the loader (idempotent) and returns
-//                                  the current status JSON immediately
+//   POST /memory/preload         — starts the loader (idempotent while a run
+//                                  is in flight or completed successfully;
+//                                  restarts a run that ended failed/cancelled)
+//                                  and returns the current status JSON
+//                                  immediately
 //   GET  /memory/preload         — returns the current status JSON
 //   POST /memory/preload/cancel  — signals the loader to stop early
 //
@@ -50,29 +53,40 @@ pub struct PreloadStatus {
 
 pub async fn post_memory_preload(State(state): State<Arc<AppState>>) -> impl IntoResponse {
     // First caller wins the CAS and spawns the loader; subsequent callers
-    // just report the existing status.
-    let we_start = state
-        .mem_preload_started
-        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
-        .is_ok();
-
-    if we_start {
-        // Fresh run: clear leftovers from a previous lifecycle's loader so a
-        // stale done=true (or stale counters) can't masquerade as this run's
-        // result. Done under the error mutex, the same critical section /init
-        // and result publication use.
-        {
-            let mut err = state.mem_preload_error.lock().unwrap();
-            *err = None;
-            state.mem_preload_done.store(false, Ordering::SeqCst);
-            state.mem_preload_regions.store(0, Ordering::SeqCst);
-            state.mem_preload_pages.store(0, Ordering::SeqCst);
-            state.mem_preload_bytes.store(0, Ordering::SeqCst);
-            state.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
-            state.mem_preload_source.store(0, Ordering::SeqCst);
+    // just report the existing status — EXCEPT when the previous run finished
+    // in failed/cancelled state, which may be retried. Without the retry, one
+    // transient failure would block the host's pause/snapshot gate until the
+    // next /init (which never comes while the sandbox stays running).
+    //
+    // Both the fresh-run reset and the retry check run under the error mutex
+    // — the same critical section /init's reset and the loader's result
+    // publication use — so concurrent POSTs cannot both claim a retry and a
+    // frozen /init cannot interleave.
+    // The generation MUST be captured inside the same mutex-held block as the
+    // CAS/reset: loading it after the lock drops lets an interleaving /init
+    // (bump + started=false) tag this loader with the NEW generation, and a
+    // follow-up POST would then spawn a second concurrent full-RAM walk.
+    let start_generation = {
+        let mut err = state.mem_preload_error.lock().unwrap();
+        let fresh = state
+            .mem_preload_started
+            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+            .is_ok();
+        let retry = !fresh
+            && state.mem_preload_done.load(Ordering::SeqCst)
+            && (err.is_some() || state.mem_preload_cancel.load(Ordering::SeqCst));
+        if fresh || retry {
+            // Clear leftovers (previous lifecycle's or failed run's result) so
+            // a stale done=true can't masquerade as this run's completion.
+            state.reset_preload_run(&mut err);
+            state.mem_preload_cancel.store(false, Ordering::SeqCst);
+            Some(state.mem_preload_generation.load(Ordering::SeqCst))
+        } else {
+            None
         }
-        let generation = state.mem_preload_generation.load(Ordering::SeqCst);
+    };
 
+    if let Some(generation) = start_generation {
         let state_clone = Arc::clone(&state);
         // Detached blocking thread — no axum task lifetime ties it to the
         // request, so the connection can close immediately without aborting
@@ -373,45 +387,63 @@ struct RamSegment {
 // Match every System RAM range from /proc/iomem to its KCORE_RAM PT_LOAD
 // segment. RAM segments share a single direct-map base (vaddr = base +
 // phys_start; base is KASLR-randomised, so it must be derived, not assumed).
-// Candidate bases come from exact-size (segment, range) pairs; a base is
-// accepted only if EVERY RAM range has a segment at base + start with the
-// exact range size. This can never select vmalloc/vmemmap segments: their
+// Candidate bases come from size-matched (segment, range) pairs; a base is
+// accepted only if EVERY RAM range has a segment at its expected vaddr with
+// the expected size. This can never select vmalloc/vmemmap segments: their
 // sizes (tens of TB) match no iomem range.
+//
+// Sizes are compared after clamping the iomem range to page boundaries:
+// kcore's segments are PFN-granular (walk_system_ram_range uses
+// PFN_UP/PFN_DOWN) while iomem ranges are byte-granular — the classic
+// 0x1000-0x9fbff low-RAM range is 0x9ec00 bytes in iomem but 0x9e000 in
+// kcore. Exact byte matching would reject every kernel that has such a range.
 fn match_ram_segments(
     ranges: &[(u64, u64)],
     segments: &[KcoreSegment],
 ) -> Result<Vec<RamSegment>, String> {
     const KERNEL_SPACE_MIN: u64 = 0xffff_8000_0000_0000;
 
+    // Page-clamp: [PFN_UP(start), PFN_DOWN(end)). Ranges smaller than one
+    // page after clamping have no kcore segment and nothing to preload.
+    let clamped: Vec<(u64, u64)> = ranges
+        .iter()
+        .map(|(start, end)| (start.next_multiple_of(PAGE_SIZE), end & !(PAGE_SIZE - 1)))
+        .filter(|(start, end)| end > start)
+        .collect();
+    if clamped.is_empty() {
+        return Err("no page-sized System RAM ranges after clamping".into());
+    }
+
     let mut candidates: Vec<u64> = Vec::new();
     for s in segments.iter().filter(|s| s.vaddr >= KERNEL_SPACE_MIN) {
-        for (start, end) in ranges {
-            let len = end - start;
-            if s.file_size == len && s.vaddr.checked_sub(*start).is_some() {
-                candidates.push(s.vaddr - start);
+        for (start, end) in &clamped {
+            if s.file_size == end - start {
+                if let Some(base) = s.vaddr.checked_sub(*start) {
+                    candidates.push(base);
+                }
             }
         }
     }
     candidates.sort_unstable();
     candidates.dedup();
 
+    let try_base = |base: u64| -> Option<Vec<RamSegment>> {
+        clamped
+            .iter()
+            .map(|(start, end)| {
+                segments
+                    .iter()
+                    .find(|s| s.vaddr == base + start && s.file_size == end - start)
+                    .map(|s| RamSegment {
+                        file_offset: s.file_offset,
+                        len: end - start,
+                    })
+            })
+            .collect()
+    };
+
     for base in &candidates {
-        let mut out = Vec::with_capacity(ranges.len());
-        for (start, end) in ranges {
-            let len = end - start;
-            let Some(seg) = segments
-                .iter()
-                .find(|s| s.vaddr == base.wrapping_add(*start) && s.file_size == len)
-            else {
-                out.clear();
-                break;
-            };
-            out.push(RamSegment {
-                file_offset: seg.file_offset,
-                len,
-            });
-        }
-        if out.len() == ranges.len() {
+        if let Some(out) = try_base(*base) {
             return Ok(out);
         }
     }
@@ -419,7 +451,7 @@ fn match_ram_segments(
     Err(format!(
         "no consistent direct-map base for {} RAM ranges across {} PT_LOAD segments \
          ({} candidate bases tried)",
-        ranges.len(),
+        clamped.len(),
         segments.len(),
         candidates.len(),
     ))
@@ -458,8 +490,11 @@ mod tests {
         ]
     }
 
+    // Byte-granular, as /proc/iomem reports them: the low range ends at
+    // 0x9fbff (parse adds 1 → 0x9fc00), NOT page-aligned. kcore's segment for
+    // it is PFN-clamped to 0x9e000 bytes — matching must tolerate that.
     fn typical_ranges() -> Vec<(u64, u64)> {
-        vec![(0x1000, 0x9f000), (0x100000, 2 * GIB)]
+        vec![(0x1000, 0x9fc00), (0x100000, 2 * GIB)]
     }
 
     #[test]
@@ -505,6 +540,32 @@ mod tests {
         assert!(match_ram_segments(&typical_ranges(), &segments).is_err());
     }
 
+    #[test]
+    fn unaligned_range_is_page_clamped() {
+        // Range with unaligned start AND end: kcore clamps to
+        // [PFN_UP(start), PFN_DOWN(end)).
+        let ranges = vec![(0x2400, 0x9fbff + 1)];
+        let segments = vec![
+            seg(0xffff_c900_0000_0000, 32 * TIB, 0x1000),
+            seg(BASE + 0x3000, 0x9c000, 0x10000), // 0x3000..0x9f000
+        ];
+        let out = match_ram_segments(&ranges, &segments).unwrap();
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].file_offset, 0x10000);
+        assert_eq!(out[0].len, 0x9c000);
+    }
+
+    #[test]
+    fn sub_page_range_is_skipped() {
+        // A range smaller than one page after clamping (e.g. 0x9fc00-0x9ffff)
+        // has no kcore segment; it must be dropped, not fail the match.
+        let ranges = vec![(0x9fc00, 0xa0000), (0x100000, 2 * GIB)];
+        let segments = vec![seg(BASE + 0x100000, 2 * GIB - 0x100000, 0x20000)];
+        let out = match_ram_segments(&ranges, &segments).unwrap();
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].file_offset, 0x20000);
+    }
+
     #[test]
     fn ambiguous_same_size_ranges_still_resolve() {
         // Two RAM ranges of identical size: candidate bases from cross pairs

envd-rs/src/http/snapshot.rs

@@ -20,15 +20,11 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
         port_sub.stop();
     }
 
-    // sync(2) flushes the in-memory FS state. Done before drop_caches so the
-    // pages we drop are clean.
-    sync();
-
-    // Drop the VFS page cache + dentries/inodes. Reduces snapshot size by
-    // ensuring CH only persists memory pages that the guest actually needs.
-    if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
-        tracing::warn!(error = %e, "drop_caches (first pass) failed (continuing)");
-    }
+    // Flush in-memory FS state, then drop the VFS page cache +
+    // dentries/inodes. sync first so the pages we drop are clean; dropping
+    // reduces snapshot size by ensuring CH only persists memory pages the
+    // guest actually needs.
+    flush_and_drop_caches("first pass").await;
 
     // Best-effort fstrim on the rootfs so unused blocks are returned to the
     // dm-snapshot, keeping CoW size minimal.
@@ -37,14 +33,10 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
         .output()
         .await;
 
-    // Second drop_caches pass after fstrim: fstrim re-reads superblock /
-    // group descriptor pages that we just evicted, putting them back in the
-    // page cache. A second pass drops those and any other late readers (e.g.
-    // sync flushers).
-    sync();
-    if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
-        tracing::warn!(error = %e, "drop_caches (second pass) failed (continuing)");
-    }
+    // Second pass after fstrim: fstrim re-reads superblock / group descriptor
+    // pages that we just evicted, putting them back in the page cache. This
+    // drops those and any other late readers (e.g. sync flushers).
+    flush_and_drop_caches("second pass").await;
 
     // No balloon settle window here: free-page reporting drains asynchronously,
     // so any pages not yet hole-punched by the host at snapshot time are written
@@ -59,3 +51,17 @@ pub async fn post_snapshot_prepare(State(state): State<Arc<AppState>>) -> impl I
         [(header::CACHE_CONTROL, "no-store")],
     )
 }
+
+// sync(2) can block for seconds flushing dirty pages, and the drop_caches
+// write blocks while the kernel evicts — both run on the blocking pool or
+// they starve every async task sharing the worker thread (health probes,
+// exec streams) for the whole quiesce window.
+async fn flush_and_drop_caches(pass: &'static str) {
+    let _ = tokio::task::spawn_blocking(move || {
+        sync();
+        if let Err(e) = std::fs::write("/proc/sys/vm/drop_caches", "3") {
+            tracing::warn!(error = %e, pass, "drop_caches failed (continuing)");
+        }
+    })
+    .await;
+}

envd-rs/src/permissions/path.rs

@@ -57,10 +57,18 @@ pub fn ensure_dirs(path: &str, uid: Uid, gid: Gid) -> Result<(), String> {
                 }
             }
             Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
-                fs::create_dir(&current)
-                    .map_err(|e| format!("failed to create directory {current_str}: {e}"))?;
-                chown(&current, Some(uid.as_raw()), Some(gid.as_raw()))
-                    .map_err(|e| format!("failed to chown directory {current_str}: {e}"))?;
+                if let Err(ce) = fs::create_dir(&current) {
+                    // Concurrent request may have created it between the stat
+                    // and here — fine as long as it's a directory now. The
+                    // winner did its own chown; don't re-chown their dir.
+                    let now_dir = fs::metadata(&current).map(|m| m.is_dir()).unwrap_or(false);
+                    if !now_dir {
+                        return Err(format!("failed to create directory {current_str}: {ce}"));
+                    }
+                } else {
+                    chown(&current, Some(uid.as_raw()), Some(gid.as_raw()))
+                        .map_err(|e| format!("failed to chown directory {current_str}: {e}"))?;
+                }
             }
             Err(e) => {
                 return Err(format!("failed to stat directory {current_str}: {e}"));

envd-rs/src/rpc/filesystem_service.rs

@@ -159,26 +159,33 @@ impl Filesystem for FilesystemServiceImpl {
 
         let path = self.resolve_path(request.path, &ctx)?;
 
-        let resolved = std::fs::canonicalize(&path).map_err(|e| {
-            if e.kind() == std::io::ErrorKind::NotFound {
-                ConnectError::new(ErrorCode::NotFound, format!("path not found: {e}"))
-            } else {
-                ConnectError::new(ErrorCode::Internal, format!("error resolving path: {e}"))
+        // The recursive walk stats every entry (plus uid/gid lookups) — on a
+        // large tree that is seconds of blocking syscalls, so it runs on the
+        // blocking pool instead of a runtime worker thread.
+        let entries = tokio::task::spawn_blocking(move || {
+            let resolved = std::fs::canonicalize(&path).map_err(|e| {
+                if e.kind() == std::io::ErrorKind::NotFound {
+                    ConnectError::new(ErrorCode::NotFound, format!("path not found: {e}"))
+                } else {
+                    ConnectError::new(ErrorCode::Internal, format!("error resolving path: {e}"))
+                }
+            })?;
+            let resolved_str = resolved.to_string_lossy().to_string();
+
+            let meta = std::fs::metadata(&resolved).map_err(|e| {
+                ConnectError::new(ErrorCode::Internal, format!("error getting file info: {e}"))
+            })?;
+            if !meta.is_dir() {
+                return Err(ConnectError::new(
+                    ErrorCode::InvalidArgument,
+                    format!("path is not a directory: {path}"),
+                ));
             }
-        })?;
-        let resolved_str = resolved.to_string_lossy().to_string();
 
-        let meta = std::fs::metadata(&resolved).map_err(|e| {
-            ConnectError::new(ErrorCode::Internal, format!("error getting file info: {e}"))
-        })?;
-        if !meta.is_dir() {
-            return Err(ConnectError::new(
-                ErrorCode::InvalidArgument,
-                format!("path is not a directory: {path}"),
-            ));
-        }
-
-        let entries = walk_dir(&path, &resolved_str, depth)?;
+            walk_dir(&path, &resolved_str, depth)
+        })
+        .await
+        .map_err(|e| ConnectError::new(ErrorCode::Internal, format!("list_dir task: {e}")))??;
         Ok((
             ListDirResponse {
                 entries,
@@ -195,14 +202,21 @@ impl Filesystem for FilesystemServiceImpl {
     ) -> Result<(RemoveResponse, Context), ConnectError> {
         let path = self.resolve_path(request.path, &ctx)?;
 
-        if let Err(e1) = std::fs::remove_dir_all(&path) {
-            if let Err(e2) = std::fs::remove_file(&path) {
-                return Err(ConnectError::new(
-                    ErrorCode::Internal,
-                    format!("error removing: {e1}; also tried as file: {e2}"),
-                ));
+        // remove_dir_all recurses through the whole tree — blocking pool, not
+        // a runtime worker thread.
+        tokio::task::spawn_blocking(move || {
+            if let Err(e1) = std::fs::remove_dir_all(&path) {
+                if let Err(e2) = std::fs::remove_file(&path) {
+                    return Err(ConnectError::new(
+                        ErrorCode::Internal,
+                        format!("error removing: {e1}; also tried as file: {e2}"),
+                    ));
+                }
             }
-        }
+            Ok(())
+        })
+        .await
+        .map_err(|e| ConnectError::new(ErrorCode::Internal, format!("remove task: {e}")))??;
 
         Ok((
             RemoveResponse {

envd-rs/src/rpc/process_handler.rs

@@ -1,6 +1,7 @@
 use std::collections::VecDeque;
 use std::io::Read;
-use std::os::unix::io::AsRawFd;
+use std::os::fd::{AsFd, OwnedFd};
+use std::os::unix::io::{AsRawFd, RawFd};
 use std::os::unix::process::CommandExt;
 use std::process::Stdio;
 use std::sync::{Arc, Mutex};
@@ -10,6 +11,8 @@ use connectrpc::{ConnectError, ErrorCode};
 use nix::pty::{Winsize, openpty};
 use nix::sys::signal::{self, Signal};
 use nix::unistd::Pid;
+use tokio::io::Interest;
+use tokio::io::unix::AsyncFd;
 use tokio::sync::broadcast;
 
 use crate::rpc::pb::process::*;
@@ -32,6 +35,14 @@ const COALESCE_CAP: usize = 64 * 1024;
 // buffer can never grow without bound for a chatty long-running process.
 const OUTPUT_LOG_CAPACITY: usize = 256 * 1024;
 
+// Bound on how long one input write may wait for the target process to drain
+// its buffer. The stdin pipe / pty master is O_NONBLOCK, so a full buffer
+// parks the writing task (never an OS thread); within this window a
+// briefly-full buffer recovers transparently, past it the write fails with
+// ResourceExhausted instead of hanging forever on a process that stopped
+// reading its input.
+const INPUT_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
+
 #[derive(Clone)]
 pub enum DataEvent {
     Stdout(Vec<u8>),
@@ -136,6 +147,24 @@ where
                 }
                 publish(std::mem::take(&mut acc));
             }
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                // The pty master is O_NONBLOCK for the input write path and
+                // shares its file description with this reader. Wait for
+                // readability and retry; on POLLHUP/POLLERR the retried read
+                // returns 0/EIO and ends the loop.
+                let mut pfd = libc::pollfd {
+                    fd,
+                    events: libc::POLLIN,
+                    revents: 0,
+                };
+                if unsafe { libc::poll(&mut pfd, 1, -1) } < 0 {
+                    let errno = std::io::Error::last_os_error().raw_os_error();
+                    if errno != Some(libc::EINTR) {
+                        break;
+                    }
+                }
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
             Err(_) => break,
         }
     }
@@ -156,6 +185,10 @@ pub struct ProcessHandle {
 
     stdin: Mutex<Option<std::process::ChildStdin>>,
     pty_master: Mutex<Option<std::fs::File>>,
+    // Serializes input writes so concurrent chunks land in order. A tokio
+    // mutex: a writer parked on a full buffer holds it across an await, and
+    // the next writer waits as a suspended task — no OS thread is pinned.
+    input_gate: tokio::sync::Mutex<()>,
 }
 
 impl ProcessHandle {
@@ -201,32 +234,41 @@ impl ProcessHandle {
         })
     }
 
-    pub fn write_stdin(&self, data: &[u8]) -> Result<(), ConnectError> {
-        use std::io::Write;
-        let mut guard = self.stdin.lock().unwrap();
-        match guard.as_mut() {
-            Some(stdin) => stdin.write_all(data).map_err(|e| {
-                ConnectError::new(ErrorCode::Internal, format!("error writing to stdin: {e}"))
-            }),
-            None => Err(ConnectError::new(
-                ErrorCode::FailedPrecondition,
-                "stdin not enabled or closed",
-            )),
-        }
+    pub async fn write_stdin(&self, data: &[u8]) -> Result<(), ConnectError> {
+        let _ordered = self.input_gate.lock().await;
+        // Dup the fd under the std mutex, then release it before awaiting:
+        // close_stdin stays responsive while a write is parked, and the fd
+        // stays valid for this write even if stdin is closed concurrently.
+        let fd = {
+            let guard = self.stdin.lock().unwrap();
+            match guard.as_ref() {
+                Some(stdin) => dup_writer_fd(stdin, "stdin")?,
+                None => {
+                    return Err(ConnectError::new(
+                        ErrorCode::FailedPrecondition,
+                        "stdin not enabled or closed",
+                    ));
+                }
+            }
+        };
+        write_nonblocking(fd, data, "stdin").await
     }
 
-    pub fn write_pty(&self, data: &[u8]) -> Result<(), ConnectError> {
-        use std::io::Write;
-        let mut guard = self.pty_master.lock().unwrap();
-        match guard.as_mut() {
-            Some(master) => master.write_all(data).map_err(|e| {
-                ConnectError::new(ErrorCode::Internal, format!("error writing to pty: {e}"))
-            }),
-            None => Err(ConnectError::new(
-                ErrorCode::FailedPrecondition,
-                "pty not assigned to process",
-            )),
-        }
+    pub async fn write_pty(&self, data: &[u8]) -> Result<(), ConnectError> {
+        let _ordered = self.input_gate.lock().await;
+        let fd = {
+            let guard = self.pty_master.lock().unwrap();
+            match guard.as_ref() {
+                Some(master) => dup_writer_fd(master, "pty")?,
+                None => {
+                    return Err(ConnectError::new(
+                        ErrorCode::FailedPrecondition,
+                        "pty not assigned to process",
+                    ));
+                }
+            }
+        };
+        write_nonblocking(fd, data, "pty").await
     }
 
     pub fn close_stdin(&self) -> Result<(), ConnectError> {
@@ -272,6 +314,79 @@ impl ProcessHandle {
     }
 }
 
+fn dup_writer_fd(f: &impl AsFd, what: &str) -> Result<OwnedFd, ConnectError> {
+    f.as_fd()
+        .try_clone_to_owned()
+        .map_err(|e| ConnectError::new(ErrorCode::Internal, format!("dup {what} fd: {e}")))
+}
+
+/// Write `data` to a non-blocking fd from async context. Small interactive
+/// writes complete inline; a full buffer surfaces as WouldBlock and we await
+/// writability with a deadline, so a process that stopped reading its input
+/// costs a parked task — never a pinned OS thread.
+async fn write_nonblocking(fd: OwnedFd, data: &[u8], what: &str) -> Result<(), ConnectError> {
+    let afd = AsyncFd::with_interest(fd, Interest::WRITABLE)
+        .map_err(|e| ConnectError::new(ErrorCode::Internal, format!("register {what} fd: {e}")))?;
+    let deadline = tokio::time::Instant::now() + INPUT_WRITE_TIMEOUT;
+    let mut written = 0usize;
+    while written < data.len() {
+        let rest = &data[written..];
+        let n = unsafe {
+            libc::write(
+                afd.get_ref().as_raw_fd(),
+                rest.as_ptr() as *const libc::c_void,
+                rest.len(),
+            )
+        };
+        if n >= 0 {
+            written += n as usize;
+            continue;
+        }
+        let err = std::io::Error::last_os_error();
+        match err.kind() {
+            std::io::ErrorKind::WouldBlock => {
+                match tokio::time::timeout_at(deadline, afd.writable()).await {
+                    Ok(Ok(mut ready)) => ready.clear_ready(),
+                    Ok(Err(e)) => {
+                        return Err(ConnectError::new(
+                            ErrorCode::Internal,
+                            format!("error waiting for {what} to become writable: {e}"),
+                        ));
+                    }
+                    Err(_) => {
+                        return Err(ConnectError::new(
+                            ErrorCode::ResourceExhausted,
+                            format!(
+                                "{what} input buffer full ({written} of {} bytes written): process is not reading its input",
+                                data.len()
+                            ),
+                        ));
+                    }
+                }
+            }
+            std::io::ErrorKind::Interrupted => {}
+            _ => {
+                return Err(ConnectError::new(
+                    ErrorCode::Internal,
+                    format!("error writing to {what}: {err}"),
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
+fn set_nonblocking(fd: RawFd) -> std::io::Result<()> {
+    let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
+    if flags < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(())
+}
+
 pub struct SpawnedProcess {
     pub handle: Arc<ProcessHandle>,
     pub data_rx: broadcast::Receiver<DataEvent>,
@@ -436,6 +551,19 @@ exec {nice_prefix}{target}"#
 
         let pid = child.id();
         let master_file: std::fs::File = master_fd.into();
+        // O_NONBLOCK is per file description, so it covers the write path and
+        // the reader clone alike; coalesce_read_loop handles the resulting
+        // WouldBlock by polling. Without it a write would block an OS thread,
+        // so treat failure as a failed spawn.
+        if let Err(e) = set_nonblocking(master_file.as_raw_fd()) {
+            let mut child = child;
+            let _ = signal::kill(Pid::from_raw(-(pid as i32)), Signal::SIGKILL);
+            let _ = child.wait();
+            return Err(ConnectError::new(
+                ErrorCode::Internal,
+                format!("set pty master non-blocking: {e}"),
+            ));
+        }
         let master_clone = master_file.try_clone().unwrap();
 
         let handle = Arc::new(ProcessHandle {
@@ -448,6 +576,7 @@ exec {nice_prefix}{target}"#
             output_log: Mutex::new(OutputLog::default()),
             stdin: Mutex::new(None),
             pty_master: Mutex::new(Some(master_file)),
+            input_gate: tokio::sync::Mutex::new(()),
         });
 
         let data_rx = handle.subscribe_data();
@@ -530,6 +659,21 @@ exec {nice_prefix}{target}"#
         let stdout = child.stdout.take();
         let stderr = child.stderr.take();
 
+        // Non-blocking stdin is what lets write_stdin fail fast instead of
+        // pinning an OS thread when the process stops draining its input.
+        // Only the pipe's write end is affected — the child's read end is a
+        // separate file description.
+        if let Some(s) = stdin.as_ref() {
+            if let Err(e) = set_nonblocking(s.as_raw_fd()) {
+                let _ = signal::kill(Pid::from_raw(-(pid as i32)), Signal::SIGKILL);
+                let _ = child.wait();
+                return Err(ConnectError::new(
+                    ErrorCode::Internal,
+                    format!("set stdin non-blocking: {e}"),
+                ));
+            }
+        }
+
         let handle = Arc::new(ProcessHandle {
             config,
             tag,
@@ -540,6 +684,7 @@ exec {nice_prefix}{target}"#
             output_log: Mutex::new(OutputLog::default()),
             stdin: Mutex::new(stdin),
             pty_master: Mutex::new(None),
+            input_gate: tokio::sync::Mutex::new(()),
         });
 
         let data_rx = handle.subscribe_data();
@@ -613,3 +758,112 @@ fn current_nice() -> i32 {
         prio
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    fn spawn(cmd: &str, enable_stdin: bool, pty: Option<(u16, u16)>) -> SpawnedProcess {
+        let user = nix::unistd::User::from_uid(nix::unistd::getuid())
+            .unwrap()
+            .unwrap();
+        spawn_process(
+            cmd,
+            &[],
+            &HashMap::new(),
+            "/tmp",
+            pty,
+            enable_stdin,
+            None,
+            &user,
+            &dashmap::DashMap::new(),
+        )
+        .unwrap()
+    }
+
+    /// Wait until the accumulated bytes from `recv` contain `needle`.
+    async fn recv_until_contains(
+        rx: &mut broadcast::Receiver<DataEvent>,
+        needle: &[u8],
+        what: &str,
+    ) -> Vec<u8> {
+        let mut acc: Vec<u8> = Vec::new();
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                match rx.recv().await {
+                    Ok(DataEvent::Stdout(d)) | Ok(DataEvent::Pty(d)) => {
+                        acc.extend_from_slice(&d);
+                        if acc.windows(needle.len()).any(|w| w == needle) {
+                            break;
+                        }
+                    }
+                    Ok(DataEvent::Stderr(_)) => {}
+                    Err(broadcast::error::RecvError::Lagged(_)) => {}
+                    Err(e) => panic!("{what}: data stream closed early: {e}"),
+                }
+            }
+        })
+        .await
+        .unwrap_or_else(|_| panic!("{what}: output never contained expected bytes"));
+        acc
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn write_to_stuck_stdin_errors_instead_of_hanging() {
+        // sleep never reads stdin: the 64K pipe buffer fills and the write
+        // must fail with ResourceExhausted after the bounded wait — the old
+        // write_all here blocked a pool thread forever.
+        let spawned = spawn("sleep 60", true, None);
+        let handle = Arc::clone(&spawned.handle);
+        let big = vec![b'x'; 1024 * 1024];
+        let writer = tokio::spawn(async move { handle.write_stdin(&big).await });
+
+        // While that write is parked, the agent must stay fully responsive:
+        // a fresh process spawns, runs, and reports its exit.
+        let mut quick = spawn("true", false, None);
+        let end = tokio::time::timeout(Duration::from_secs(10), quick.end_rx.recv())
+            .await
+            .expect("agent unresponsive while stdin write pending")
+            .expect("end event");
+        assert!(end.exited);
+
+        let err = tokio::time::timeout(INPUT_WRITE_TIMEOUT + Duration::from_secs(10), writer)
+            .await
+            .expect("stdin write hung past its deadline")
+            .expect("writer task panicked")
+            .expect_err("write into a full pipe nobody reads must error");
+        assert!(
+            matches!(err.code, ErrorCode::ResourceExhausted),
+            "expected ResourceExhausted, got {:?}",
+            err.code
+        );
+
+        let _ = spawned.handle.send_signal(Signal::SIGKILL);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn stdin_write_and_eof_still_work() {
+        let mut spawned = spawn("cat", true, None);
+        spawned.handle.write_stdin(b"hello\n").await.unwrap();
+        recv_until_contains(&mut spawned.data_rx, b"hello\n", "cat stdout").await;
+
+        // close_stdin must still deliver EOF with the non-blocking pipe.
+        spawned.handle.close_stdin().unwrap();
+        let end = tokio::time::timeout(Duration::from_secs(10), spawned.end_rx.recv())
+            .await
+            .expect("cat did not exit after stdin EOF")
+            .expect("end event");
+        assert_eq!(end.exit_code, 0);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn pty_write_and_echo_still_work() {
+        // O_NONBLOCK on the pty master is shared with the output reader;
+        // this covers both directions surviving it.
+        let mut spawned = spawn("cat", false, Some((80, 24)));
+        spawned.handle.write_pty(b"hello\r").await.unwrap();
+        recv_until_contains(&mut spawned.data_rx, b"hello", "pty echo").await;
+        let _ = spawned.handle.send_signal(Signal::SIGKILL);
+    }
+}

envd-rs/src/rpc/process_service.rs

@@ -305,7 +305,7 @@ impl Process for ProcessServiceImpl {
                         ConnectError::new(ErrorCode::FailedPrecondition, "no start event received")
                     })?;
                     if let Some(input) = data.input.as_option() {
-                        write_input(h, input)?;
+                        write_input(h, input).await?;
                     }
                 }
                 Some(stream_input_request::EventView::Keepalive(_)) => {}
@@ -332,7 +332,7 @@ impl Process for ProcessServiceImpl {
         let handle = self.get_process_by_selector(selector)?;
 
         if let Some(input) = request.input.as_option() {
-            write_input(&handle, input)?;
+            write_input(&handle, input).await?;
         }
 
         Ok((
@@ -392,10 +392,18 @@ impl Process for ProcessServiceImpl {
     }
 }
 
-fn write_input(handle: &ProcessHandle, input: &ProcessInputView) -> Result<(), ConnectError> {
+// Input writes go straight to the fd, which is O_NONBLOCK (set at spawn):
+// small interactive writes complete inline on the async worker, and a full
+// buffer parks the task awaiting writability with a bounded deadline inside
+// write_stdin / write_pty — no blocking-pool thread is ever pinned by a
+// process that stopped reading its input.
+async fn write_input(
+    handle: &Arc<ProcessHandle>,
+    input: &ProcessInputView<'_>,
+) -> Result<(), ConnectError> {
     match &input.input {
-        Some(process_input::InputView::Pty(d)) => handle.write_pty(d),
-        Some(process_input::InputView::Stdin(d)) => handle.write_stdin(d),
+        Some(process_input::InputView::Pty(d)) => handle.write_pty(d).await,
+        Some(process_input::InputView::Stdin(d)) => handle.write_stdin(d).await,
         None => Ok(()),
     }
 }

envd-rs/src/state.rs

@@ -31,6 +31,9 @@ pub struct AppState {
     pub mem_preload_started: AtomicBool,
     pub mem_preload_done: AtomicBool,
     pub mem_preload_cancel: AtomicBool,
+    /// See `reset_preload_run` for the per-run reset shared by /init and the
+    /// POST /memory/preload start/retry paths.
+    ///
     /// Bumped by every /init lifecycle change. A loader thread captures the
     /// value at spawn and refuses to run — or to publish results — once it no
     /// longer matches, so a thread that survived a pause/resume (the VM can be
@@ -110,6 +113,22 @@ impl AppState {
 
     /// Records a new lifecycle ID, returning true if it changed (i.e. this
     /// is the first /init since a resume). First-ever call returns false:
+    /// Clears the per-run memory-preload result fields (done flag, counters,
+    /// source, error). Shared by /init's lifecycle reset and POST
+    /// /memory/preload's start/retry paths so the two field lists cannot
+    /// drift. The caller MUST hold the `mem_preload_error` mutex and pass its
+    /// guard's contents in — that mutex is the critical section that also
+    /// serializes the loader thread's result publication.
+    pub fn reset_preload_run(&self, err: &mut Option<String>) {
+        *err = None;
+        self.mem_preload_done.store(false, Ordering::SeqCst);
+        self.mem_preload_regions.store(0, Ordering::SeqCst);
+        self.mem_preload_pages.store(0, Ordering::SeqCst);
+        self.mem_preload_bytes.store(0, Ordering::SeqCst);
+        self.mem_preload_elapsed_us.store(0, Ordering::SeqCst);
+        self.mem_preload_source.store(0, Ordering::SeqCst);
+    }
+
     /// boot-time /init doesn't need port-subsystem restart since the
     /// subsystem hasn't been started yet by anything else.
     pub fn bump_lifecycle(&self, new_id: &str) -> bool {

internal/hostagent/server.go

@@ -76,8 +76,11 @@ func (s *Server) CreateSandbox(
 		return nil, err
 	}
 
+	// disk_size_mb in the request is deprecated and never set by the control
+	// plane; passing 0 lets the manager pick the size (DefaultRootfsSizeMB,
+	// floored at the origin image size).
 	sb, diskSizeBytes, err := s.mgr.Create(ctx, msg.SandboxId, teamID, templateID,
-		int(msg.Vcpus), int(msg.MemoryMb), int(msg.TimeoutSec), int(msg.DiskSizeMb),
+		int(msg.Vcpus), int(msg.MemoryMb), int(msg.TimeoutSec), 0,
 		msg.DefaultUser, msg.DefaultEnv)
 	if err != nil {
 		if errors.Is(err, sandbox.ErrDraining) {

pkg/devicemapper/devicemapper.go

@@ -192,8 +192,19 @@ func RestoreSnapshot(ctx context.Context, name, originLoopDev, cowPath string, o
 
 // RemoveSnapshot tears down a dm-snapshot device and its CoW loop device.
 // The CoW file is NOT deleted — the caller decides whether to keep or remove it.
+//
+// The CoW loop is detached even when the dm removal fails (busy device):
+// callers treat a failed removal as fatal teardown and typically delete the
+// CoW backing file next, which would orphan a still-attached loop forever
+// (losetup can no longer find it by file). Detaching first is always safe —
+// while dm still references the loop the kernel just defers the release via
+// autoclear until that reference drops.
 func RemoveSnapshot(ctx context.Context, dev *SnapshotDevice) error {
 	if err := dmsetupRemove(ctx, dev.Name); err != nil {
+		if derr := losetupDetachRetry(dev.CowLoopDev); derr != nil {
+			slog.Warn("cow loop detach after failed dm remove",
+				"device", dev.CowLoopDev, "name", dev.Name, "error", derr)
+		}
 		return fmt.Errorf("dmsetup remove %s: %w", dev.Name, err)
 	}
 

pkg/envdclient/client.go

@@ -41,9 +41,10 @@ func New(hostIP string) *Client {
 // of the standard http://{hostIP}:{envdPort}. Intended for tests (httptest
 // servers) and nonstandard listeners.
 func NewWithBaseURL(base string) *Client {
-	u, err := url.Parse(base)
 	host := base
-	if err == nil {
+	// Hostname() is empty for scheme-less inputs ("127.0.0.1:8080" parses
+	// with "127.0.0.1" as the scheme) — keep the raw base then.
+	if u, err := url.Parse(base); err == nil && u.Hostname() != "" {
 		host = u.Hostname()
 	}
 	return newWithBase(host, base)

pkg/network/setup.go

@@ -545,6 +545,80 @@ func RemoveNetwork(slot *Slot) error {
 	return errors.Join(errs...)
 }
 
+var privateEgressGuardNets = []string{
+	"10.0.0.0/8",
+	"172.16.0.0/12",
+	"192.168.0.0/16",
+	"169.254.0.0/16",
+	"100.64.0.0/10",
+}
+
+var sandboxSuperNets = []string{"10.11.0.0/16", "10.12.0.0/16"}
+
+const vethMatch = "wrenn-veth+"
+
+// Installs host-level protections that stop capsule from (a) leaking traffic
+// destined to private ranges onto the upstream network via the public NIC
+// and (b) reaching another capsule's host-reachable IP.
+func EnsureEgressGuard() error {
+	defaultIface, err := getDefaultInterface()
+	if err != nil {
+		return fmt.Errorf("resolve default interface: %w", err)
+	}
+
+	// Drop forwarded capsule traffic destined to private ranges out the public
+	// NIC. Insert at the head of FORWARD so it precedes the per-sandbox ACCEPT
+	// rules appended when a sandbox is created.
+	for _, n := range privateEgressGuardNets {
+		if err := ensureHostRule(
+			[]string{"-C", "FORWARD", "-o", defaultIface, "-d", n, "-j", "DROP"},
+			[]string{"-I", "FORWARD", "1", "-o", defaultIface, "-d", n, "-j", "DROP"},
+		); err != nil {
+			return fmt.Errorf("install egress guard for %s: %w", n, err)
+		}
+	}
+
+	// Deny capsule-to-capsule forwarding outright: a veth may reach the uplink,
+	// never another sandbox's veth. This closes cross-tenant reach to the
+	// unauthenticated guest agent even for allocated slots.
+	if err := ensureHostRule(
+		[]string{"-C", "FORWARD", "-i", vethMatch, "-o", vethMatch, "-j", "DROP"},
+		[]string{"-I", "FORWARD", "1", "-i", vethMatch, "-o", vethMatch, "-j", "DROP"},
+	); err != nil {
+		return fmt.Errorf("install inter-sandbox guard: %w", err)
+	}
+
+	// Blackhole the sandbox supernets so packets to unallocated slot IPs are
+	// dropped rather than routed to the default gateway.
+	for _, cidr := range sandboxSuperNets {
+		if err := ensureBlackholeRoute(cidr); err != nil {
+			return fmt.Errorf("blackhole %s: %w", cidr, err)
+		}
+	}
+
+	slog.Info("egress guard installed", "default_iface", defaultIface)
+	return nil
+}
+
+// ensureHostRule inserts a host iptables rule only when an identical rule is
+// not already present, keeping EnsureEgressGuard idempotent across restarts.
+func ensureHostRule(checkArgs, insertArgs []string) error {
+	if exec.Command("iptables", checkArgs...).Run() == nil {
+		return nil // already present
+	}
+	return iptablesHost(insertArgs...)
+}
+
+// ensureBlackholeRoute idempotently installs a blackhole route via `ip route
+// replace` (a no-op if the route already matches).
+func ensureBlackholeRoute(cidr string) error {
+	cmd := exec.Command("ip", "route", "replace", "blackhole", cidr)
+	if out, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("ip route replace blackhole %s: %s: %w", cidr, string(out), err)
+	}
+	return nil
+}
+
 // nsExec runs a command inside a network namespace.
 func nsExec(nsName string, command string, args ...string) error {
 	cmdArgs := append([]string{"netns", "exec", nsName, command}, args...)

pkg/sandbox/manager.go

@@ -622,14 +622,11 @@ func (m *Manager) cleanup(ctx context.Context, sb *sandboxState) {
 	m.slots.Release(sb.SlotIndex)
 
 	// Tear down dm-snapshot and release the base image loop device.
+	// RemoveSnapshot detaches the CoW loop even on failure, so deleting the
+	// CoW file below cannot orphan it.
 	if sb.dmDevice != nil {
 		if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
 			slog.Warn("dm-snapshot remove error", "id", sb.ID, "error", err)
-			// RemoveSnapshot bails before detaching the CoW loop when the dm
-			// device is busy. The CoW file is deleted just below — detach the
-			// loop first (autoclear if dm still holds it) or it becomes
-			// permanently unreclaimable once its backing file is gone.
-			devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
 		}
 		os.Remove(sb.dmDevice.CowPath)
 	}
@@ -1285,12 +1282,11 @@ func (m *Manager) cleanupAfterCrash(sb *sandboxState) {
 	}
 	m.slots.Release(sb.SlotIndex)
 
+	// RemoveSnapshot detaches the CoW loop even on failure, so the RemoveAll
+	// below (which deletes the CoW backing file) cannot orphan it.
 	if sb.dmDevice != nil {
 		if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
 			slog.Warn("crash cleanup: dm-snapshot error", "id", sb.ID, "error", err)
-			// os.RemoveAll below deletes the CoW backing file; detach its
-			// loop first or it can never be reclaimed (see DetachLoopsByFile).
-			devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
 		}
 	}
 	if sb.baseImagePath != "" {

pkg/sandbox/pause.go

@@ -216,6 +216,15 @@ func (m *Manager) Pause(ctx context.Context, sandboxID string) error {
 			m.mu.Lock()
 			sb.SlotIndex = 0
 			m.mu.Unlock()
+			// Remove the sandbox dir wholesale, mirroring cleanupAfterCrash.
+			// This drops the running-state file — its recorded slot is freed
+			// and may be re-claimed, so a later restart's sweep must not tear
+			// down the new owner's network — and any previous pause
+			// generation, which is unusable anyway: the CoW has advanced past
+			// the memory image it was captured with.
+			if err := os.RemoveAll(layout.SandboxDir(m.cfg.WrennDir, sandboxID)); err != nil {
+				slog.Warn("pause rollback: sandbox dir remove failed", "id", sandboxID, "error", err)
+			}
 			return fmt.Errorf("pause %s: %s: %w (and resume failed: %v)",
 				sandboxID, stage, cause, rerr)
 		}
@@ -511,14 +520,11 @@ func (m *Manager) releaseRuntime(sb *sandboxState, cow cowDisposition) {
 		m.slots.Release(sb.SlotIndex)
 	}
 
+	// RemoveSnapshot detaches the CoW loop even on failure, so dropCow's file
+	// deletion cannot orphan it and keepCow's Resume cannot double-attach.
 	if sb.dmDevice != nil {
 		if err := devicemapper.RemoveSnapshot(context.Background(), sb.dmDevice); err != nil {
 			slog.Warn("dm-snapshot remove on pause", "id", sb.ID, "error", err)
-			// The CoW loop was not detached (RemoveSnapshot stops on a busy
-			// dm device). Detach it by backing file — dropCow deletes the file
-			// below, and even for keepCow a lingering attachment would make
-			// Resume's losetupCreateRW attach a second loop to the same file.
-			devicemapper.DetachLoopsByFile(sb.dmDevice.CowPath)
 		}
 		if cow == dropCow {
 			os.Remove(sb.dmDevice.CowPath)
@@ -815,15 +821,16 @@ func (m *Manager) ensureMemoryMaterialized(ctx context.Context, sb *sandboxState
 		return fmt.Errorf("memory preload verify: envd client unavailable")
 	}
 
-	status, err := client.GetMemoryPreloadStatus(ctx)
+	// One POST covers every state: "done" reports immediately, "running"
+	// reports the in-flight run, "idle" starts one (agent died between
+	// restore and the loader's POST, or an earlier start failed on the wire),
+	// and a previous "failed"/"cancelled" run is re-armed — so a transient
+	// guest-side failure blocks only this attempt, not every future pause.
+	status, err := client.StartMemoryPreload(ctx)
 	if err != nil {
-		return fmt.Errorf("memory preload verify: %w", err)
+		return fmt.Errorf("memory preload start: %w", err)
 	}
-
-	if status.State == "idle" || status.State == "running" {
-		if _, err := client.StartMemoryPreload(ctx); err != nil {
-			return fmt.Errorf("memory preload start: %w", err)
-		}
+	if status.State != "done" {
 		status, err = client.WaitMemoryPreload(ctx)
 		if err != nil {
 			return fmt.Errorf("memory preload wait: %w", err)

pkg/sandbox/pause_test.go

@@ -70,7 +70,7 @@ func TestEnsureMemoryMaterialized(t *testing.T) {
 			lazy:      true,
 			states:    []string{"done"},
 			wantErr:   "",
-			wantPosts: 0,
+			wantPosts: 1, // single POST reports done, no polling
 		},
 		{
 			name:      "lazy idle self-heals via start",
@@ -80,25 +80,18 @@ func TestEnsureMemoryMaterialized(t *testing.T) {
 			wantPosts: 1,
 		},
 		{
-			name:      "lazy failed is a hard error",
+			name:      "lazy failed after wait is a hard error",
 			lazy:      true,
-			states:    []string{"failed"},
+			states:    []string{"idle", "failed"},
 			wantErr:   "memory preload failed",
-			wantPosts: 0,
-		},
-		{
-			name:      "lazy start ends in failed",
-			lazy:      true,
-			states:    []string{"idle", "idle", "failed"},
-			wantErr:   "memory preload failed",
-			wantPosts: -1,
+			wantPosts: 1,
 		},
 		{
 			name:      "lazy cancelled is a hard error",
 			lazy:      true,
-			states:    []string{"cancelled"},
+			states:    []string{"running", "cancelled"},
 			wantErr:   "memory preload cancelled",
-			wantPosts: 0,
+			wantPosts: 1,
 		},
 		{
 			name:      "lazy without client fails",

pkg/sandbox/restore_paused.go

@@ -78,6 +78,19 @@ func (m *Manager) RestorePausedSandboxes() {
 			continue
 		}
 
+		// A readable running-state file that survived RestoreRunningSandboxes
+		// means the adoption of a verified-LIVE CH process failed (probe
+		// timeout etc.) and its slot/loop were deliberately left held. The
+		// stale pause generation in the same dir must not be registered — and
+		// above all not trashed: trashing renames the whole sandbox dir,
+		// carrying away the live VM's CoW file and the state file the next
+		// restart needs to retry the adoption.
+		if _, err := readRunningState(m.cfg.WrennDir, name); err == nil {
+			slog.Warn("restore: skipping paused candidate — dir owned by live un-adopted VM",
+				"id", name)
+			continue
+		}
+
 		snapDir := layout.PauseSnapshotDir(m.cfg.WrennDir, name)
 		meta, err := readSnapshotMeta(snapDir)
 		if err != nil {