diff --git a/nvx-rs/src/guest/init.rs b/nvx-rs/src/guest/init.rs
index 0956f7942..ec2e871c2 100644
--- a/nvx-rs/src/guest/init.rs
+++ b/nvx-rs/src/guest/init.rs
@@ -1,5 +1,5 @@
-//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1 for layered roots, which runs ahead of the
-//! workloads.
+//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1, which initializes a layered root and serves
+//! its managed control session through the same dispatcher as blockless VMs.
 //!
 //! A layered root's early boot, which the shell `/init` runs, mounts the kernel's file systems,
 //! runs `nvx-time boot`, whose child becomes the time daemon, configures the network, and, when
@@ -30,8 +30,8 @@
 //!   nothing from it;
 //! - ends the VM itself, without another guest program: it writes every file system's modified
 //!   data to its storage, with `sync`, and then stops the VM through OpenVMM's control port with a
-//!   status: 0 after `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after a failure, which
-//!   an `NVX-INIT-ERROR:` line on the console names first.
+//!   status: 0 after a host `STOP` or `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after
+//!   a failure, which an `NVX-INIT-ERROR:` line on the console names first.
 //!
 //! Outside PID 1, `nvx-init` changes nothing: it fails with status 2 and a message.
 
@@ -64,8 +64,6 @@ const NICE: i32 = -10;
 /// The signals that PID 1 reads from its signal descriptor: the end of a child, and the request
 /// to end the VM.
 const SIGNALS: [Signal; 2] = [Signal::SIGCHLD, Signal::SIGTERM];
-/// Pause before PID 1 polls its signal descriptor again after `poll` failed.
-const POLL_RETRY: Duration = Duration::from_secs(1);
 
 /// The helpers that the shell `/init` hands over: the only processes whose end ends the VM.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -769,8 +767,6 @@ trait System {
     /// Collects the status of one child that ended, without waiting, or returns `None` when none
     /// did.
     fn reap(&mut self) -> Option<(ProcessId, ChildStatus, i32)>;
-    /// Waits for one of [`SIGNALS`], and returns it.
-    fn wait_signal(&mut self) -> Option<Signal>;
     /// Writes the diagnostic `line` to the console.
     fn report(&mut self, line: &[u8]);
     /// Writes every file system's modified data to its storage, and then stops the VM with
@@ -811,19 +807,6 @@ impl System for RealSystem {
         sys::reap_any_nonblocking().map(|child| (child.process, child.status, child.wait_status))
     }
 
-    fn wait_signal(&mut self) -> Option<Signal> {
-        let signals = self.signals.as_ref()?;
-        loop {
-            if let Some(event) = signals.read() {
-                return event.signal;
-            }
-            match sys::poll([Pollable::Signals(signals)], None) {
-                Ok(_) | Err(Errno::EINTR) => {}
-                Err(_) => std::thread::sleep(POLL_RETRY),
-            }
-        }
-    }
-
     fn report(&mut self, line: &[u8]) {
         write_console(kernel::CONSOLE_DEVICE, line);
     }
@@ -853,10 +836,6 @@ impl Runtime {
     }
 
     /// Poll alongside control input so a blocked peer cannot hide a helper's end.
-    #[cfg_attr(
-        not(test),
-        expect(dead_code, reason = "SL7.6b polls the init runtime with control input")
-    )]
     pub(crate) fn wake(&self) -> Pollable<'_> {
         Pollable::Signals(self.system.signals.as_ref().expect("listening"))
     }
@@ -907,12 +886,13 @@ fn serve_layered(runtime: &mut Runtime) -> Result<(), String> {
         )
         .as_bytes(),
     );
-    loop {
-        if runtime.service(|_, _| {})? {
-            return Ok(());
-        }
-        runtime.terminating |= runtime.system.wait_signal() == Some(Signal::SIGTERM);
-    }
+    super::managed_agent::AgentConfig::layered(&boot.command_line, boot.user, boot.home)
+        .map_err(|error| format!("invalid layered identity: {}", error.desc()))?
+        .serve_layered(
+            runtime,
+            boot.control,
+            (boot.table, boot.mapper, boot.export),
+        )
 }
 
 /// Keeps diagnostics, sync and direct power-off with the original PID-1 owner.
diff --git a/nvx-rs/src/guest/init/tests.rs b/nvx-rs/src/guest/init/tests.rs
index 5632fcea1..3a017f74a 100644
--- a/nvx-rs/src/guest/init/tests.rs
+++ b/nvx-rs/src/guest/init/tests.rs
@@ -23,12 +23,12 @@ use crate::guest::sys::{ForkResult, wait};
 
 /// Drives the unchanged helper/signal scripts through production preparation, collection and
 /// shutdown; layered boot itself is exercised by the boot tests.
-fn init(system: &mut impl System, arguments: &[CString]) {
+fn init(system: &mut impl Waiting, arguments: &[CString]) {
     let result = serve(system, arguments);
     finish(system, result);
 }
 
-fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String> {
+fn serve(system: &mut impl Waiting, arguments: &[CString]) -> Result<(), String> {
     let helpers = prepare(system, arguments)?;
     let mut terminating = false;
     loop {
@@ -40,6 +40,25 @@ fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String>
     }
 }
 
+trait Waiting: System {
+    fn wait_signal(&mut self) -> Option<Signal>;
+}
+
+impl Waiting for RealSystem {
+    fn wait_signal(&mut self) -> Option<Signal> {
+        let signals = self.signals.as_ref()?;
+        loop {
+            if let Some(event) = signals.read() {
+                return event.signal;
+            }
+            match sys::poll([Pollable::Signals(signals)], None) {
+                Ok(_) | Err(Errno::EINTR) => {}
+                Err(_) => std::thread::sleep(Duration::from_secs(1)),
+            }
+        }
+    }
+}
+
 /// One operation that PID 1 asked its system for.
 #[derive(Clone, Debug, PartialEq, Eq)]
 enum Call {
@@ -51,7 +70,7 @@ enum Call {
     Adopt(i32),
     /// [`System::reap`].
     Reap,
-    /// [`System::wait_signal`].
+    /// [`Waiting::wait_signal`].
     Wait,
     /// [`System::report`] of this line.
     Report(String),
@@ -73,7 +92,7 @@ struct Fake {
     /// The answers of [`System::reap`], in order: a child that ended, or `None` when none did.
     /// Once they run out, no child ended.
     reaped: VecDeque<Option<(i32, ChildStatus)>>,
-    /// The signals that [`System::wait_signal`] returns, in order. The check fails if PID 1 waits
+    /// The signals that [`Waiting::wait_signal`] returns, in order. The check fails if PID 1 waits
     /// for another.
     signals: VecDeque<Signal>,
 }
@@ -112,15 +131,6 @@ impl System for Fake {
         Some((pid(process), status, raw))
     }
 
-    fn wait_signal(&mut self) -> Option<Signal> {
-        self.calls.push(Call::Wait);
-        Some(
-            self.signals
-                .pop_front()
-                .expect("PID 1 waits for no signal that the check sends"),
-        )
-    }
-
     fn report(&mut self, line: &[u8]) {
         self.calls
             .push(Call::Report(String::from_utf8(line.to_vec()).unwrap()));
@@ -131,6 +141,17 @@ impl System for Fake {
     }
 }
 
+impl Waiting for Fake {
+    fn wait_signal(&mut self) -> Option<Signal> {
+        self.calls.push(Call::Wait);
+        Some(
+            self.signals
+                .pop_front()
+                .expect("PID 1 waits for no signal that the check sends"),
+        )
+    }
+}
+
 #[test]
 fn the_shared_collector_forwards_workload_status_and_keeps_helper_failure_fatal() {
     let helpers = Helpers {
@@ -261,6 +282,16 @@ impl System for Subreaper {
         Some(child)
     }
 
+    fn report(&mut self, line: &[u8]) {
+        self.reports.push_str(std::str::from_utf8(line).unwrap());
+    }
+
+    fn shut_down(&mut self, status: u8) {
+        self.status = Some(status);
+    }
+}
+
+impl Waiting for Subreaper {
     fn wait_signal(&mut self) -> Option<Signal> {
         if let Some(before_wait) = self.before_wait.take() {
             before_wait();
@@ -269,14 +300,6 @@ impl System for Subreaper {
         self.signals.extend(signal);
         signal
     }
-
-    fn report(&mut self, line: &[u8]) {
-        self.reports.push_str(std::str::from_utf8(line).unwrap());
-    }
-
-    fn shut_down(&mut self, status: u8) {
-        self.status = Some(status);
-    }
 }
 
 /// A system whose priorities, children, and statuses are the real ones, which records the
@@ -312,6 +335,16 @@ impl System for Observer {
         self.real.reap()
     }
 
+    fn report(&mut self, line: &[u8]) {
+        self.reports.push_str(std::str::from_utf8(line).unwrap());
+    }
+
+    fn shut_down(&mut self, status: u8) {
+        self.status = Some(status);
+    }
+}
+
+impl Waiting for Observer {
     fn wait_signal(&mut self) -> Option<Signal> {
         let processes = std::iter::once(sys::current_process()).chain(self.helpers);
         self.observed = processes
@@ -323,14 +356,6 @@ impl System for Observer {
             .collect();
         Some(Signal::SIGTERM)
     }
-
-    fn report(&mut self, line: &[u8]) {
-        self.reports.push_str(std::str::from_utf8(line).unwrap());
-    }
-
-    fn shut_down(&mut self, status: u8) {
-        self.status = Some(status);
-    }
 }
 
 /// Returns the process ID `raw`.
diff --git a/nvx-rs/src/guest/managed_agent/dispatch.rs b/nvx-rs/src/guest/managed_agent/dispatch.rs
index f46a6e641..b94c61b97 100644
--- a/nvx-rs/src/guest/managed_agent/dispatch.rs
+++ b/nvx-rs/src/guest/managed_agent/dispatch.rs
@@ -1,4 +1,8 @@
-use crate::guest::sys::Errno;
+use std::ffi::CString;
+
+use crate::guest::init::Runtime;
+use crate::guest::mappings::{Mapper, MappingTable, SourceRoot};
+use crate::guest::sys::{self, Errno, FileDescriptor};
 use crate::protocol::{
     self, APP_CANCEL, APP_EXEC, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, APP_STOP, APP_STOPPED,
     GuestFeatures, INVALID_REQUEST, MAPPINGS_INCOMPLETE, UNSUPPORTED_OPERATION,
@@ -6,14 +10,67 @@ use crate::protocol::{
 
 use super::diagnostics;
 use super::mappings::{HostMappings, MappingError};
-use super::seam::System;
+use super::seam::{Paths, RealSystem, System};
 use super::session::{Event, ReceiveFailure, Session};
-use super::{AgentConfig, exec, tty};
+use super::{AgentConfig, exec, launch, tty};
+
+impl AgentConfig {
+    /// Takes a layered cold boot's CLI identity and its already-resolved account defaults.
+    pub(crate) fn layered(
+        command_line: &[u8],
+        user: CString,
+        home: CString,
+    ) -> Result<Self, Errno> {
+        let mut identities = crate::protocol::cmdline::init_tokens(command_line)
+            .filter(|token| token.name == b"nvx_identity")
+            .filter_map(|token| token.value);
+        let identity = identities.next().ok_or(Errno::EINVAL)?;
+        let mut ids = identity.split(|&byte| byte == b':');
+        let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else {
+            return Err(Errno::EINVAL);
+        };
+        if identities.next().is_some() {
+            return Err(Errno::EINVAL);
+        }
+        launch::parse_id(uid)?;
+        launch::parse_id(gid)?;
+        Ok(Self {
+            rootfs: c"-".to_owned(),
+            hostname: c"nvx-managed".to_owned(),
+            uid: sys::c_string(uid)?,
+            gid: sys::c_string(gid)?,
+            user,
+            home,
+            direct: true,
+            entry: None,
+        })
+    }
+
+    /// Serves an initialized layered root, returning shutdown to its init owner.
+    pub(crate) fn serve_layered(
+        &self,
+        runtime: &mut Runtime,
+        control: FileDescriptor,
+        (table, mapper, export): (MappingTable, Mapper, Option<SourceRoot>),
+    ) -> Result<(), String> {
+        let mut system = RealSystem::new(Paths::default());
+        let mut session = Session::with_runtime(control, runtime)?;
+        let mut mappings = HostMappings::layered(table, mapper, export)
+            .map_err(|error| format!("invalid layered mappings: {}", error.desc()))?;
+        run(&mut system, &mut session, self, &mut mappings)
+            .map(|Outcome::Stopped| ())
+            .map_err(|error| {
+                session
+                    .runtime_failure()
+                    .unwrap_or_else(|| format!("control session failed: {}", error.desc()))
+            })
+    }
+}
 
 /// Terminal outcome of the long-lived dispatch loop.
 #[derive(Debug)]
 pub(crate) enum Outcome {
-    /// The host requested and acknowledged an orderly stop.
+    /// The host requested STOP, or init accepted an orderly termination.
     Stopped,
 }
 
@@ -24,16 +81,30 @@ pub(crate) fn run(
     config: &AgentConfig,
     mappings: &mut HostMappings,
 ) -> Result<Outcome, Errno> {
-    session.send_attach()?;
+    if session.stopping() {
+        return Ok(Outcome::Stopped);
+    }
+    if let Err(error) = session.send_attach() {
+        return if session.stopping() {
+            Ok(Outcome::Stopped)
+        } else {
+            Err(error)
+        };
+    }
     loop {
         let payload = match session.next_event() {
             Ok(Event::Reset) => {
-                session.acknowledge().inspect_err(|error| {
-                    diagnostics::error(system, "reset-ack", *error as i32);
-                })?;
+                if let Err(error) = session.acknowledge() {
+                    if session.stopping() {
+                        return Ok(Outcome::Stopped);
+                    }
+                    diagnostics::error(system, "reset-ack", error as i32);
+                    return Err(error);
+                }
                 continue;
             }
             Ok(Event::Data(payload)) => payload,
+            Err(_) if session.stopping() => return Ok(Outcome::Stopped),
             Err(failure) => return Err(report_failure(system, failure)),
         };
         // As in the C agent, a DATA record that fails also reports the `data` stage, after any
@@ -43,6 +114,7 @@ pub(crate) fn run(
         match handle_data(system, session, config, mappings, &payload) {
             Ok(Some(outcome)) => return Ok(outcome),
             Ok(None) => {}
+            Err(_) if session.stopping() => return Ok(Outcome::Stopped),
             Err(error) => {
                 diagnostics::error(system, "data", error as i32);
                 return Err(error);
diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs
index 4422aff6b..b16ead8d7 100644
--- a/nvx-rs/src/guest/managed_agent/dispatch/tests.rs
+++ b/nvx-rs/src/guest/managed_agent/dispatch/tests.rs
@@ -67,13 +67,17 @@ impl Harness {
 
     /// Starts a direct agent, or a sandbox agent whose assembled root is `rootfs`, which
     /// attaches and waits for the broker's first RESET.
-    fn spawn_at(rootfs: Option<&Path>, mut system: TestSystem, expected_mappings: u32) -> Self {
+    fn spawn_at(rootfs: Option<&Path>, system: TestSystem, expected_mappings: u32) -> Self {
+        let config = AgentConfig::for_test(rootfs);
+        let mappings = HostMappings::announced_for(&config, expected_mappings);
+        Self::spawn_with(system, config, mappings)
+    }
+
+    fn spawn_with(mut system: TestSystem, config: AgentConfig, mut mappings: HostMappings) -> Self {
         let (host, guest) = UnixStream::pair().unwrap();
         let descriptor = FileDescriptor::from_owned(guest.into());
-        let config = AgentConfig::for_test(rootfs);
         let thread = thread::spawn(move || {
             let mut session = Session::new(descriptor);
-            let mut mappings = HostMappings::announced_for(&config, expected_mappings);
             let outcome = run(&mut system, &mut session, &config, &mut mappings);
             (outcome, system)
         });
@@ -208,15 +212,21 @@ impl Harness {
     }
 
     fn read_outer(&mut self) -> (OuterHeader, Vec<u8>) {
-        let mut header = [0u8; OUTER_HEADER_LEN];
-        self.host.read_exact(&mut header).unwrap();
-        let header = decode_outer_header(&header).unwrap();
-        let mut payload = vec![0; header.length];
-        self.host.read_exact(&mut payload).unwrap();
-        (header, payload)
+        read_outer(&mut self.host)
     }
 }
 
+fn read_outer(host: &mut UnixStream) -> (OuterHeader, Vec<u8>) {
+    let mut header = [0u8; OUTER_HEADER_LEN];
+    host.read_exact(&mut header).unwrap();
+    let header = decode_outer_header(&header).unwrap();
+    let mut payload = vec![0; header.length];
+    host.read_exact(&mut payload).unwrap();
+    (header, payload)
+}
+
+mod runtime;
+
 fn response(kind: u8, request_id: u64, status: i32, payload: &[u8]) -> Response {
     Response {
         kind,
@@ -473,6 +483,124 @@ fn sandbox_root() -> (tempfile::TempDir, TestSystem, std::path::PathBuf) {
     (temporary, system, rootfs)
 }
 
+#[test]
+fn prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher() -> TestResult {
+    if super::super::tests::isolate_process_test(
+        "NVX_LAYERED_DISPATCH_CHILD",
+        "guest::managed_agent::dispatch::tests::prepared_layered_mappings_and_cli_identity_use_the_existing_dispatcher",
+    ) {
+        return Ok(());
+    }
+    use crate::guest::mappings::{Mapper, MappingTable, Reserved, SourceRoot, TargetRoot};
+
+    let (_temporary, mut system, rootfs) = sandbox_root();
+    system.direct_starter = Some(|request| {
+        assert_eq!((request.uid, request.gid), (1234, 2345));
+        super::super::seam::tests::start_as_privileged(request)
+    });
+    let export = system.paths.hostfs_dir.join("root").canonicalize()?;
+    let source = SourceRoot::open(&export)?;
+    let target = TargetRoot::open_with(&mut system.kernel, &rootfs)?;
+    let mut mapper = Mapper::new(target, Reserved::container_root());
+    let mut table = MappingTable::new(2).unwrap();
+    table
+        .apply(
+            &maps_payload(
+                0,
+                &[(b"0/src", b"/workspace", protocol::MAP_READ_ONLY)],
+                None,
+            ),
+            |batch| {
+                mapper
+                    .apply_with(&mut system.kernel, &source, batch)
+                    .map_err(|failure| (failure.entry(), failure.error().raw_os_error()))
+            },
+        )
+        .unwrap();
+    let mappings = HostMappings::layered(table, mapper, Some(source))?;
+    system.paths.hostfs_dir = rootfs.join("not-the-retained-export");
+    let config = AgentConfig::layered(
+        b"nvx_identity=1234:2345",
+        c"workload".to_owned(),
+        c"/home/workload".to_owned(),
+    )?;
+    let mut harness = Harness::spawn_with(system, config, mappings);
+    harness.reset(1, 7);
+    assert_eq!(harness.request(APP_PING, &[]).kind, APP_READY);
+    assert_eq!(
+        harness.request(APP_EXEC, &exec_payload(None)),
+        response(APP_ERROR, 42, Errno::EAGAIN as i32, b"mappings-incomplete")
+    );
+    harness.reset(2, 0);
+    assert_eq!(
+        harness.request(APP_MAPS, &maps_payload(0, &[(b"1", b"/out", 0)], None)),
+        response(APP_ERROR, 42, Errno::EINVAL as i32, b"invalid-request")
+    );
+    assert_eq!(
+        harness
+            .request(APP_MAPS, &maps_payload(1, &[(b"1", b"/out", 0)], None))
+            .kind,
+        APP_READY
+    );
+    assert_eq!(
+        stdout_from(
+            harness.request_all(
+                9,
+                APP_EXEC,
+                &workload_payload(&["/bin/sh", "-c", "printf layered; exit 7"], Some("/"))?,
+            ),
+            9,
+            7
+        ),
+        b"layered"
+    );
+    let system = harness.stop();
+    let expected: Vec<_> = [
+        (
+            export.join("0/src"),
+            rootfs.join("workspace"),
+            Access::ReadOnly,
+        ),
+        (export.join("1"), rootfs.join("out"), Access::ReadWrite),
+    ]
+    .into_iter()
+    .flat_map(|(from, to, access)| attach_calls(&from, &to, access))
+    .collect();
+    assert_eq!(system.kernel.calls(), expected);
+    assert_eq!(system.spawn_calls, 1);
+    assert_eq!(system.entries_prepared, 0);
+    Ok(())
+}
+
+#[test]
+fn a_layered_root_without_an_export_refuses_maps_but_keeps_its_complete_table() {
+    use crate::guest::mappings::{Mapper, MappingTable, Reserved, TargetRoot};
+
+    let (_temporary, mut system, rootfs) = sandbox_root();
+    let mapper = Mapper::new(
+        TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap(),
+        Reserved::container_root(),
+    );
+    let mappings = HostMappings::layered(MappingTable::new(0).unwrap(), mapper, None).unwrap();
+    let config =
+        AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap();
+    let mut harness = Harness::spawn_with(system, config, mappings);
+    harness.reset(1, 0);
+    for payload in [
+        maps_payload(0, &[], None),
+        maps_payload(0, &[(b"0", b"/workspace", 0)], None),
+    ] {
+        assert_eq!(
+            harness.request(APP_MAPS, &payload),
+            response(APP_ERROR, 42, Errno::EINVAL as i32, b"invalid-request")
+        );
+    }
+    harness.reset(2, 0);
+    assert_eq!(harness.request(APP_PING, &[]).kind, APP_READY);
+    let system = harness.stop();
+    assert!(system.kernel.calls().is_empty());
+}
+
 #[test]
 fn sandbox_mappings_mount_in_order_beneath_its_root_and_unblock_workloads() {
     let (_temporary, system, rootfs) = sandbox_root();
diff --git a/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs
new file mode 100644
index 000000000..e9bfd863a
--- /dev/null
+++ b/nvx-rs/src/guest/managed_agent/dispatch/tests/runtime.rs
@@ -0,0 +1,784 @@
+use std::time::{Duration, Instant};
+
+use crate::guest::mappings::{Mapper, MappingTable, Reserved, SourceRoot, TargetRoot};
+use crate::guest::sys::tests::fork_isolated_test;
+use crate::guest::sys::{self, ChildStatus, ForkResult, ProcessId, Signal};
+
+use super::*;
+
+struct OwnedChild {
+    process: ProcessId,
+    descriptor: FileDescriptor,
+}
+
+impl OwnedChild {
+    fn new(process: ProcessId) -> Self {
+        let descriptor = sys::open_pidfd(process).unwrap_or_else(|error| {
+            sys::kill_process(process).unwrap();
+            let _ = sys::wait(process);
+            panic!("cannot retain the test child's identity: {error}");
+        });
+        Self {
+            process,
+            descriptor,
+        }
+    }
+}
+
+impl Drop for OwnedChild {
+    fn drop(&mut self) {
+        assert!(matches!(
+            sys::kill_through_pidfd(self.descriptor.borrowed(), false),
+            Ok(()) | Err(Errno::ESRCH)
+        ));
+        // The Runtime may already have collected this owned child.
+        let _ = sys::wait(self.process);
+    }
+}
+
+fn isolated(marker: &str, name: &str, check: impl FnOnce() + std::panic::UnwindSafe) {
+    if super::super::super::tests::isolate_process_test(marker, name) {
+        return;
+    }
+    match fork_isolated_test().unwrap() {
+        ForkResult::Child => {
+            sys::reset_child_signal();
+            let result = std::panic::catch_unwind(check);
+            sys::exit_immediately(u8::from(result.is_err()));
+        }
+        ForkResult::Parent(child) => {
+            assert_eq!(sys::wait(child), Some(ChildStatus::Exited(0)));
+        }
+    }
+}
+
+fn helper() -> OwnedChild {
+    match sys::fork().unwrap() {
+        ForkResult::Child => {
+            std::thread::sleep(Duration::from_secs(30));
+            sys::exit_immediately(0);
+        }
+        ForkResult::Parent(process) => OwnedChild::new(process),
+    }
+}
+
+fn peer(
+    check: impl FnOnce(UnixStream, UnixStream) + std::panic::UnwindSafe,
+) -> (FileDescriptor, OwnedChild, UnixStream) {
+    let (host, guest) = UnixStream::pair().unwrap();
+    let (proof, mut completed) = UnixStream::pair().unwrap();
+    host.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+    proof
+        .set_read_timeout(Some(Duration::from_secs(5)))
+        .unwrap();
+    match sys::fork().unwrap() {
+        ForkResult::Child => {
+            drop(proof);
+            let probe = guest.try_clone().unwrap();
+            drop(guest);
+            let result = std::panic::catch_unwind(|| check(host, probe));
+            if result.is_ok() {
+                completed.write_all(&[1]).unwrap();
+            }
+            sys::exit_immediately(u8::from(result.is_err()));
+        }
+        ForkResult::Parent(process) => {
+            drop(host);
+            drop(completed);
+            (
+                FileDescriptor::from_owned(guest.into()),
+                OwnedChild::new(process),
+                proof,
+            )
+        }
+    }
+}
+
+fn completed(mut proof: UnixStream) {
+    let mut success = [0];
+    proof.read_exact(&mut success).unwrap();
+    assert_eq!(success, [1]);
+}
+
+fn until(mut ready: impl FnMut() -> bool) {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while !ready() {
+        assert!(Instant::now() < deadline, "the owned peer made no progress");
+        std::thread::sleep(Duration::from_millis(1));
+    }
+}
+
+fn fragment(host: &mut UnixStream, probe: UnixStream, record: &[u8]) {
+    use std::os::fd::AsFd;
+
+    host.write_all(&record[..1]).unwrap();
+    // Only the guest reads this endpoint: no readable byte proves it consumed the fragment.
+    until(|| !sys::readable_or_hung_up(&[probe.as_fd()], 0).unwrap()[0]);
+    drop(probe);
+}
+
+fn prepared() -> (
+    tempfile::TempDir,
+    (MappingTable, Mapper, Option<SourceRoot>),
+) {
+    let (temporary, mut system, rootfs) = sandbox_root();
+    let target = TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap();
+    (
+        temporary,
+        (
+            MappingTable::new(0).unwrap(),
+            Mapper::new(target, Reserved::container_root()),
+            None,
+        ),
+    )
+}
+
+fn termination(helper_ends: bool) {
+    let helper = helper();
+    let parent = sys::open_pidfd(sys::current_process()).unwrap();
+    let (descriptor, _peer, proof) = peer(|mut host, probe| {
+        assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH);
+        let reset = encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap();
+        fragment(&mut host, probe, &reset);
+        if helper_ends {
+            sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap();
+            assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]);
+        }
+        sys::signal_through_pidfd(parent.borrowed(), Signal::SIGTERM as i32, false).unwrap();
+        let mut traffic = Vec::new();
+        host.read_to_end(&mut traffic).unwrap();
+        assert!(
+            traffic.is_empty(),
+            "SIGTERM must not synthesize NVXC frames"
+        );
+    });
+    let mut runtime =
+        Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+    let (_temporary, mappings) = prepared();
+    let config =
+        AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap();
+    let result = if helper_ends {
+        let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap();
+        let mut system = TestSystem::new(Paths::default());
+        let (table, mapper, export) = mappings;
+        let mut mappings = HostMappings::layered(table, mapper, export).unwrap();
+        assert_eq!(
+            run(&mut system, &mut session, &config, &mut mappings).err(),
+            Some(Errno::ECHILD)
+        );
+        Err(session.runtime_failure().unwrap())
+    } else {
+        config.serve_layered(&mut runtime, descriptor, mappings)
+    };
+    drop(runtime);
+    if helper_ends {
+        let failure = result.unwrap_err();
+        assert!(failure.contains("the time daemon"), "{failure}");
+        assert!(failure.contains("signal 9"), "{failure}");
+    } else {
+        result.unwrap();
+    }
+    completed(proof);
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_stops_during_a_fragment_without_protocol_traffic() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_STOP_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_during_a_fragment_without_protocol_traffic",
+        || termination(false),
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_helper_failure_wins_over_pending_sigterm() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_HELPER_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_helper_failure_wins_over_pending_sigterm",
+        || termination(true),
+    );
+}
+
+fn receive(host: &mut UnixStream) -> Response {
+    let (header, payload) = read_outer(host);
+    assert_eq!(header.record_type, OUTER_DATA);
+    let frame = decode_app(&payload).unwrap();
+    response(frame.kind, frame.request_id, frame.status, frame.payload)
+}
+
+fn prelaunch_termination(helper_ends: bool) {
+    let helper = helper();
+    let mut runtime =
+        Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+    let (mut host, guest) = UnixStream::pair().unwrap();
+    let mut session =
+        Session::with_runtime(FileDescriptor::from_owned(guest.into()), &mut runtime).unwrap();
+    session.send_attach().unwrap();
+    assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH);
+    host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap())
+        .unwrap();
+    assert!(matches!(session.next_event(), Ok(Event::Reset)));
+    session.acknowledge().unwrap();
+    assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK);
+    let execution = workload_payload(&["/bin/true"], Some("/")).unwrap();
+    host.write_all(
+        &encode_outer(
+            OUTER_DATA,
+            &INSTANCE,
+            1,
+            1,
+            &encode_app(APP_EXEC, 9, 0, &execution),
+        )
+        .unwrap(),
+    )
+    .unwrap();
+    let Ok(Event::Data(payload)) = session.next_event() else {
+        panic!("the complete EXEC record was not received");
+    };
+    let Ok(request) = session.accept_request(&payload) else {
+        panic!("the complete EXEC request was not accepted");
+    };
+    assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+    let execution = crate::protocol::decode_guest_exec_payload(request.payload).unwrap();
+    if helper_ends {
+        sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap();
+        assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]);
+    } else {
+        let process = sys::open_pidfd(sys::current_process()).unwrap();
+        sys::signal_through_pidfd(process.borrowed(), Signal::SIGTERM as i32, false).unwrap();
+    }
+    let mut system = TestSystem::new(Paths::default());
+    let config =
+        AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into()).unwrap();
+    let error = super::super::super::exec::run(
+        &mut system,
+        &mut session,
+        &config,
+        request.request_id,
+        &execution,
+    )
+    .err();
+    drop(session);
+    drop(runtime);
+    let mut traffic = Vec::new();
+    host.read_to_end(&mut traffic).unwrap();
+    assert_eq!(
+        error,
+        Some(if helper_ends {
+            Errno::ECHILD
+        } else {
+            Errno::ECANCELED
+        })
+    );
+    assert!(
+        traffic.is_empty(),
+        "prelaunch shutdown must not send an application error"
+    );
+    assert_eq!(system.spawn_calls, 0);
+    assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics);
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_prelaunch_sigterm_sends_no_application_error() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_PRELAUNCH_STOP_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_prelaunch_sigterm_sends_no_application_error",
+        || prelaunch_termination(false),
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_prelaunch_helper_failure_sends_no_application_error() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_PRELAUNCH_HELPER_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_prelaunch_helper_failure_sends_no_application_error",
+        || prelaunch_termination(true),
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver_events() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_STATUS_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_delivers_a_fragment_collected_status_before_driver_events",
+        || {
+            let temporary = tempfile::tempdir().unwrap();
+            let pid_path = temporary.path().join("leader");
+            let gate = temporary.path().join("gate");
+            crate::guest::sys::tests::create_fifo(
+                &sys::path_c_string(&gate).unwrap(),
+                sys::Mode::from_bits_retain(0o600),
+            )
+            .unwrap();
+            let payload = workload_payload(
+                &[
+                    "/bin/sh",
+                    "-c",
+                    "printf '%s' \"$$\" > \"$1\"; read -r go < \"$2\"; printf observed; exit 7",
+                    "sh",
+                    pid_path.to_str().unwrap(),
+                    gate.to_str().unwrap(),
+                ],
+                Some("/"),
+            )
+            .unwrap();
+            let helper = helper();
+            let (descriptor, _peer, proof) = peer(|mut host, probe| {
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH);
+                host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap())
+                    .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK);
+                let execution = encode_app(APP_EXEC, 9, 0, &payload);
+                host.write_all(&encode_outer(OUTER_DATA, &INSTANCE, 1, 1, &execution).unwrap())
+                    .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                until(|| pid_path.exists() && std::fs::metadata(&pid_path).unwrap().len() != 0);
+                let id: i64 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap();
+                let leader = sys::open_pidfd(ProcessId::from_positive(id).unwrap()).unwrap();
+                let ping = encode_outer(
+                    OUTER_DATA,
+                    &INSTANCE,
+                    1,
+                    2,
+                    &encode_app(APP_PING, 98, 0, &[]),
+                )
+                .unwrap();
+                fragment(&mut host, probe, &ping);
+                std::fs::OpenOptions::new()
+                    .write(true)
+                    .open(&gate)
+                    .unwrap()
+                    .write_all(b"go\n")
+                    .unwrap();
+                assert!(sys::readable_or_hung_up(&[leader.borrowed()], 5000).unwrap()[0]);
+                // The driver never reaps: disappearance here proves Runtime collected while
+                // the same dispatcher remained inside the incomplete control record.
+                until(|| !Path::new(&format!("/proc/{id}")).exists());
+                host.write_all(&ping[1..]).unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                assert_eq!(
+                    receive(&mut host),
+                    response(APP_ERROR, 98, Errno::EBUSY as i32, b"busy")
+                );
+                let mut output = Vec::new();
+                loop {
+                    let frame = receive(&mut host);
+                    if frame.kind == APP_EXIT {
+                        assert_eq!(frame, response(APP_EXIT, 9, 7, b"exit"));
+                        break;
+                    }
+                    assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9));
+                    output.extend(frame.payload);
+                }
+                assert_eq!(output, b"observed");
+                host.write_all(
+                    &encode_outer(
+                        OUTER_DATA,
+                        &INSTANCE,
+                        1,
+                        3,
+                        &encode_app(APP_STOP, 99, 0, &[]),
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                assert_eq!(receive(&mut host), response(APP_STOPPED, 99, 0, &[]));
+            });
+            let mut runtime =
+                Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+            let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap();
+            let mut system = TestSystem::new(Paths::default());
+            system.direct_starter = Some(super::super::super::seam::tests::start_as_privileged);
+            let config =
+                AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into())
+                    .unwrap();
+            let mut mappings = HostMappings::announced_for(&config, 0);
+            assert!(matches!(
+                run(&mut system, &mut session, &config, &mut mappings),
+                Ok(Outcome::Stopped)
+            ));
+            assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics);
+            drop(session);
+            drop(runtime);
+            completed(proof);
+        },
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_keeps_helper_policy_and_first_error_during_failed_start() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_FAILED_START_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_keeps_helper_policy_and_first_error_during_failed_start",
+        || {
+            let temporary = tempfile::tempdir().unwrap();
+            let marker = temporary.path().join("failed-child");
+            let gate = temporary.path().join("gate");
+            crate::guest::sys::tests::create_fifo(
+                &sys::path_c_string(&gate).unwrap(),
+                sys::Mode::from_bits_retain(0o600),
+            )
+            .unwrap();
+            let payload = workload_payload(
+                &[
+                    "/bin/sh",
+                    "-c",
+                    "read -r go < \"$1\"; exit 19",
+                    "_",
+                    gate.to_str().unwrap(),
+                    marker.to_str().unwrap(),
+                ],
+                Some("/"),
+            )
+            .unwrap();
+            let helper = helper();
+            let (descriptor, _peer, proof) = peer(|mut host, probe| {
+                drop(probe);
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_GUEST_ATTACH);
+                host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap())
+                    .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_ACK);
+                host.write_all(
+                    &encode_outer(
+                        OUTER_DATA,
+                        &INSTANCE,
+                        1,
+                        1,
+                        &encode_app(APP_EXEC, 9, 0, &payload),
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                until(|| marker.exists() && std::fs::metadata(&marker).unwrap().len() != 0);
+                let id = std::fs::read_to_string(&marker)
+                    .unwrap()
+                    .trim()
+                    .parse()
+                    .unwrap();
+                let failed = OwnedChild::new(ProcessId::from_positive(id).unwrap());
+                assert!(!sys::readable_or_hung_up(&[failed.descriptor.borrowed()], 0).unwrap()[0]);
+                sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap();
+                let mut traffic = Vec::new();
+                host.read_to_end(&mut traffic).unwrap();
+                assert!(
+                    traffic.is_empty(),
+                    "helper failure must end the control loop"
+                );
+                drop(failed);
+            });
+            let mut runtime =
+                Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+            let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap();
+            let mut system = TestSystem::new(Paths::default());
+            system.direct_starter = Some(start_with_failed_pidfd_and_cleanup);
+            let config =
+                AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into())
+                    .unwrap();
+            let mut mappings = HostMappings::announced_for(&config, 0);
+            assert_eq!(
+                run(&mut system, &mut session, &config, &mut mappings).err(),
+                Some(Errno::ECHILD)
+            );
+            let failure = session.runtime_failure().unwrap();
+            assert!(failure.contains("the time daemon"), "{failure}");
+            let diagnostics = String::from_utf8(system.diagnostics).unwrap();
+            assert!(diagnostics.starts_with(&format!(
+                "NVX-MANAGED-ERROR: stage=exec-launch status={}\n\
+                 NVX-MANAGED-ERROR: stage=exec-kill status={}\n",
+                Errno::EPERM as i32,
+                Errno::EPERM as i32
+            )));
+            assert!(!diagnostics.contains("stage=exec-wait"), "{diagnostics}");
+            drop(session);
+            drop(runtime);
+            completed(proof);
+            let id = std::fs::read_to_string(&marker)
+                .unwrap()
+                .trim()
+                .parse()
+                .unwrap();
+            assert_eq!(
+                sys::wait(ProcessId::from_positive(id).unwrap()),
+                Some(ChildStatus::Signaled(Signal::SIGKILL as i32))
+            );
+        },
+    );
+}
+
+struct KillParentOnPanic<'a>(&'a FileDescriptor);
+
+impl Drop for KillParentOnPanic<'_> {
+    fn drop(&mut self) {
+        if std::thread::panicking() {
+            assert!(matches!(
+                sys::kill_through_pidfd(self.0.borrowed(), false),
+                Ok(()) | Err(Errno::ESRCH)
+            ));
+        }
+    }
+}
+
+fn narrow_output(probe: &UnixStream) {
+    use nix::sys::socket::{getsockopt, setsockopt, sockopt::SndBuf};
+
+    setsockopt(probe, SndBuf, &1024).unwrap();
+    assert!(getsockopt(probe, SndBuf).unwrap() < 6000);
+}
+
+fn output_is_blocked(probe: &UnixStream) {
+    use std::os::fd::AsFd;
+
+    use nix::poll::{PollFd, PollFlags};
+
+    until(|| {
+        let mut descriptors = [PollFd::new(probe.as_fd(), PollFlags::POLLOUT)];
+        nix::poll::poll(&mut descriptors, 0u16).unwrap();
+        !descriptors[0]
+            .revents()
+            .unwrap()
+            .contains(PollFlags::POLLOUT)
+    });
+}
+
+fn reset_peer(host: &mut UnixStream) {
+    assert_eq!(read_outer(host).0.record_type, OUTER_GUEST_ATTACH);
+    host.write_all(&encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap())
+        .unwrap();
+    assert_eq!(read_outer(host).0.record_type, OUTER_ACK);
+}
+
+fn only_stdout_prefix(mut traffic: &[u8]) {
+    assert!(!traffic.is_empty());
+    while traffic.len() >= OUTER_HEADER_LEN {
+        let header = decode_outer_header(traffic[..OUTER_HEADER_LEN].try_into().unwrap()).unwrap();
+        assert_eq!(header.record_type, OUTER_DATA);
+        let length = OUTER_HEADER_LEN + header.length;
+        if traffic.len() < length {
+            return;
+        }
+        let frame = decode_app(&traffic[OUTER_HEADER_LEN..length]).unwrap();
+        assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9));
+        traffic = &traffic[length..];
+    }
+}
+
+fn backpressured_termination(helper_ends: bool) {
+    let helper = helper();
+    let parent = sys::open_pidfd(sys::current_process()).unwrap();
+    let (descriptor, _peer, proof) = peer(|mut host, probe| {
+        let _guard = KillParentOnPanic(&parent);
+        narrow_output(&probe);
+        reset_peer(&mut host);
+        output_is_blocked(&probe);
+        drop(probe);
+        if helper_ends {
+            sys::kill_through_pidfd(helper.descriptor.borrowed(), false).unwrap();
+            assert!(sys::readable_or_hung_up(&[helper.descriptor.borrowed()], 5000).unwrap()[0]);
+        }
+        sys::signal_through_pidfd(parent.borrowed(), Signal::SIGTERM as i32, false).unwrap();
+        let mut traffic = Vec::new();
+        host.read_to_end(&mut traffic).unwrap();
+        only_stdout_prefix(&traffic);
+    });
+    let mut runtime =
+        Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+    let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap();
+    session.send_attach().unwrap();
+    assert!(matches!(session.next_event(), Ok(Event::Reset)));
+    session.acknowledge().unwrap();
+    assert_eq!(
+        session.send_app(APP_STDOUT, 9, 0, &[0; 32768]),
+        Err(if helper_ends {
+            Errno::ECHILD
+        } else {
+            Errno::ECANCELED
+        })
+    );
+    if helper_ends {
+        let failure = session.runtime_failure().unwrap();
+        assert!(failure.contains("the time daemon"), "{failure}");
+        assert!(failure.contains("signal 9"), "{failure}");
+    } else {
+        assert!(session.stopping());
+        assert!(session.runtime_failure().is_none());
+    }
+    drop(session);
+    drop(runtime);
+    completed(proof);
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_stops_with_backpressured_control_output() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_OUTPUT_STOP_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_with_backpressured_control_output",
+        || backpressured_termination(false),
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_helper_failure_wins_with_backpressured_control_output() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_OUTPUT_HELPER_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_helper_failure_wins_with_backpressured_control_output",
+        || backpressured_termination(true),
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_stops_before_attach_on_pending_sigterm() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_ATTACH_STOP_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_stops_before_attach_on_pending_sigterm",
+        || {
+            let helper = helper();
+            let mut runtime =
+                Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+            let (mut host, guest) = UnixStream::pair().unwrap();
+            let mut session =
+                Session::with_runtime(FileDescriptor::from_owned(guest.into()), &mut runtime)
+                    .unwrap();
+            let process = sys::open_pidfd(sys::current_process()).unwrap();
+            sys::signal_through_pidfd(process.borrowed(), Signal::SIGTERM as i32, false).unwrap();
+            let mut system = TestSystem::new(Paths::default());
+            let config =
+                AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into())
+                    .unwrap();
+            let mut mappings = HostMappings::announced_for(&config, 0);
+            assert!(matches!(
+                run(&mut system, &mut session, &config, &mut mappings),
+                Ok(Outcome::Stopped)
+            ));
+            assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics);
+            drop(session);
+            drop(runtime);
+            let mut traffic = Vec::new();
+            host.read_to_end(&mut traffic).unwrap();
+            assert!(
+                traffic.is_empty(),
+                "pending termination must not send ATTACH"
+            );
+        },
+    );
+}
+
+#[test]
+#[ignore = "requires root with CAP_SYS_NICE and CAP_SYS_RESOURCE"]
+fn privileged_layered_runtime_delivers_a_write_collected_status_before_driver_events() {
+    isolated(
+        "NVX_LAYERED_RUNTIME_OUTPUT_STATUS_CHILD",
+        "guest::managed_agent::dispatch::tests::runtime::privileged_layered_runtime_delivers_a_write_collected_status_before_driver_events",
+        || {
+            let temporary = tempfile::tempdir().unwrap();
+            let pid_path = temporary.path().join("leader");
+            let gate = sys::path_c_string(&temporary.path().join("gate")).unwrap();
+            crate::guest::sys::tests::create_fifo(&gate, sys::Mode::from_bits_retain(0o600))
+                .unwrap();
+            let payload = workload_payload(
+                &[
+                    "/bin/sh",
+                    "-c",
+                    "printf '%s' \"$$\" > \"$1\"; head -c 6000 /dev/zero; read -r go < \"$2\"; exit 7",
+                    "sh",
+                    pid_path.to_str().unwrap(),
+                    gate.to_str().unwrap(),
+                ],
+                Some("/"),
+            )
+            .unwrap();
+            let helper = helper();
+            let parent = sys::open_pidfd(sys::current_process()).unwrap();
+            let (descriptor, _peer, proof) = peer(|mut host, probe| {
+                let _guard = KillParentOnPanic(&parent);
+                narrow_output(&probe);
+                reset_peer(&mut host);
+                host.write_all(
+                    &encode_outer(
+                        OUTER_DATA,
+                        &INSTANCE,
+                        1,
+                        1,
+                        &encode_app(APP_EXEC, 9, 0, &payload),
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                output_is_blocked(&probe);
+                drop(probe);
+                let id: i64 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap();
+                let leader = sys::open_pidfd(ProcessId::from_positive(id).unwrap()).unwrap();
+                let mut release = None;
+                until(|| {
+                    match sys::open(
+                        &gate,
+                        sys::OpenFlags::O_WRONLY | sys::OpenFlags::O_NONBLOCK,
+                        sys::Mode::empty(),
+                    ) {
+                        Ok(writer) => {
+                            release = Some(writer);
+                            true
+                        }
+                        Err(Errno::ENXIO) => false,
+                        Err(error) => panic!("cannot release the owned leader: {error}"),
+                    }
+                });
+                release.unwrap().write_all(b"go\n").unwrap();
+                assert!(sys::readable_or_hung_up(&[leader.borrowed()], 5000).unwrap()[0]);
+                until(|| !Path::new(&format!("/proc/{id}")).exists());
+                let mut output = Vec::new();
+                loop {
+                    let frame = receive(&mut host);
+                    if frame.kind == APP_EXIT {
+                        assert_eq!(frame, response(APP_EXIT, 9, 7, b"exit"));
+                        break;
+                    }
+                    assert_eq!((frame.kind, frame.request_id), (APP_STDOUT, 9));
+                    output.extend(frame.payload);
+                }
+                assert_eq!(output, vec![0; 6000]);
+                host.write_all(
+                    &encode_outer(
+                        OUTER_DATA,
+                        &INSTANCE,
+                        1,
+                        2,
+                        &encode_app(APP_STOP, 99, 0, &[]),
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+                assert_eq!(read_outer(&mut host).0.record_type, OUTER_CREDIT);
+                assert_eq!(receive(&mut host), response(APP_STOPPED, 99, 0, &[]));
+            });
+            let mut runtime =
+                Runtime::new(&[CString::new(helper.process.raw().to_string()).unwrap()]).unwrap();
+            let mut session = Session::with_runtime(descriptor, &mut runtime).unwrap();
+            let mut system = TestSystem::new(Paths::default());
+            system.direct_starter = Some(super::super::super::seam::tests::start_as_privileged);
+            let config =
+                AgentConfig::layered(b"nvx_identity=1000:1000", c"workload".into(), c"/".into())
+                    .unwrap();
+            let mut mappings = HostMappings::announced_for(&config, 0);
+            assert!(matches!(
+                run(&mut system, &mut session, &config, &mut mappings),
+                Ok(Outcome::Stopped)
+            ));
+            assert!(system.diagnostics.is_empty(), "{:?}", system.diagnostics);
+            drop(session);
+            drop(runtime);
+            completed(proof);
+        },
+    );
+}
diff --git a/nvx-rs/src/guest/managed_agent/exec.rs b/nvx-rs/src/guest/managed_agent/exec.rs
index d0e18698e..69be440e1 100644
--- a/nvx-rs/src/guest/managed_agent/exec.rs
+++ b/nvx-rs/src/guest/managed_agent/exec.rs
@@ -54,7 +54,10 @@ pub(crate) fn run(
             }
         }
     }
-    let mut workload = match launch(system, config, execution) {
+    if config.direct {
+        session.collect(|_, _| {})?;
+    }
+    let mut workload = match launch(system, session, config, execution) {
         Ok(launched) => launched,
         Err(LaunchError::Entry(error)) => {
             let diagnostic =
@@ -63,7 +66,7 @@ pub(crate) fn run(
             return session.send_error(request_id, 125, LAUNCH_FAILED);
         }
         Err(LaunchError::Start(failure)) => {
-            settle_start(system, &failure)?;
+            settle_start(system, session, &failure)?;
             return session.send_error(request_id, 125, LAUNCH_FAILED);
         }
         Err(LaunchError::Other) => return session.send_error(request_id, 125, LAUNCH_FAILED),
@@ -99,8 +102,11 @@ pub(crate) fn run(
                 }
                 Err(error) => {
                     workload.terminate();
+                    if session.stopping() {
+                        return Err(error);
+                    }
                     if !(config.direct && state.status_lost)
-                        && let Err(failure) = workload.wait()
+                        && let Err(failure) = workload.wait(session)
                     {
                         report_wait_failure(system, &failure);
                     }
@@ -115,7 +121,7 @@ pub(crate) fn run(
             &mut buffer[..]
         };
         state.forward(system, session, request_id, &mut workload, output)?;
-        workload.collect();
+        workload.collect(session)?;
         if state.running(&workload)
             && execution.timeout_ms != 0
             && started.elapsed() >= Duration::from_millis(u64::from(execution.timeout_ms))
@@ -128,7 +134,7 @@ pub(crate) fn run(
         if state.ending()
             && state.running(&workload)
             && !config.direct
-            && let Err(error) = workload.wait()
+            && let Err(error) = workload.wait(session)
         {
             state.lose_status(system, &error);
         }
@@ -333,15 +339,15 @@ impl Workload {
         }
     }
 
-    fn collect(&mut self) {
-        collect(|id, status| match self {
-            Self::Direct(workload) => {
-                workload.observe(id, status);
-            }
-            Self::Sandbox(workload) => {
-                workload.observe(id, status);
-            }
-        });
+    fn observe(&mut self, id: u32, status: std::process::ExitStatus) {
+        match self {
+            Self::Direct(workload) => workload.observe(id, status),
+            Self::Sandbox(workload) => workload.observe(id, status),
+        };
+    }
+
+    fn collect(&mut self, session: &Session) -> Result<(), Errno> {
+        session.collect(|id, status| self.observe(id, status))
     }
 
     fn terminate(&mut self) {
@@ -379,12 +385,12 @@ impl Workload {
         }
     }
 
-    fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
+    fn wait(&mut self, session: &Session) -> std::io::Result<std::process::ExitStatus> {
         if let Self::Sandbox(workload) = self {
             return workload.wait();
         }
         loop {
-            self.collect();
+            self.collect(session)?;
             if let Some(status) = self.status() {
                 return Ok(status);
             }
@@ -417,7 +423,7 @@ impl Workload {
 }
 
 /// Feeds the agent's sole status collector to its active or failed-start owner.
-fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) {
+pub(super) fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) {
     while let Some(reaped) = sys::reap_any_nonblocking() {
         observe(
             reaped.process.raw().unsigned_abs(),
@@ -427,18 +433,23 @@ fn collect(mut observe: impl FnMut(u32, std::process::ExitStatus)) {
 }
 
 /// Keeps a failed native launch owned until its child is collected or explicitly lost.
-fn settle_start(system: &mut impl System, failure: &init::StartFailure) -> Result<(), Errno> {
+fn settle_start(
+    system: &mut impl System,
+    session: &Session,
+    failure: &init::StartFailure,
+) -> Result<(), Errno> {
     super::diagnostics::error(system, "exec-launch", sys::errno_of(&failure.error) as i32);
     if let Some(error) = failure.cleanup {
         super::diagnostics::error(system, "exec-kill", error as i32);
     }
     if let Some(process) = failure.process {
+        session.watch(process.raw().unsigned_abs());
         loop {
             let mut settled = false;
-            collect(|id, status| {
+            session.collect(|id, status| {
                 settled |= id == process.raw().unsigned_abs()
                     && (status.code().is_some() || status.signal().is_some());
-            });
+            })?;
             if settled {
                 break;
             }
@@ -505,6 +516,9 @@ impl RunState {
         buffer: &mut [u8],
     ) -> Result<(), Errno> {
         loop {
+            if let Some((id, status)) = session.take_status() {
+                workload.observe(id, status);
+            }
             let event = match workload.next_event(buffer) {
                 Ok(Some(event)) => event,
                 Ok(None) => return Ok(()),
@@ -638,15 +652,16 @@ impl From<Errno> for LaunchError {
 /// Starts a native direct workload, or the unchanged sandbox entry and placement sequence.
 fn launch(
     system: &mut impl System,
+    session: &Session,
     config: &AgentConfig,
     execution: &GuestExecPayload,
 ) -> Result<Workload, LaunchError> {
     if config.direct {
-        collect(|_, _| {});
-        return system
+        let workload = system
             .start_direct(launch::direct_request(config, execution)?)
-            .map(Workload::Direct)
-            .map_err(LaunchError::Start);
+            .map_err(LaunchError::Start)?;
+        session.watch(workload.id());
+        return Ok(Workload::Direct(workload));
     }
     let (pipes, ends) = Pipes::new(false).map_err(|error| sys::errno_of(&error))?;
     let request = LaunchRequest {
@@ -681,7 +696,7 @@ fn launch(
     {
         super::diagnostics::error(system, "exec-launch", error as i32);
         workload.terminate();
-        if let Err(error) = workload.wait() {
+        if let Err(error) = workload.wait(session) {
             report_wait_failure(system, &error);
         }
         return Err(LaunchError::Other);
@@ -810,6 +825,7 @@ fn handle_control(
             return Ok(ExecControl::SessionLost);
         }
         Ok(SessionEvent::Data(payload)) => payload,
+        Err(_) if session.stopping() => return Err(Errno::ECANCELED),
         Err(failure) => return Err(report_control_failure(system, failure)),
     };
     let request = match session.accept_request(&payload) {
diff --git a/nvx-rs/src/guest/managed_agent/mappings.rs b/nvx-rs/src/guest/managed_agent/mappings.rs
index c63590da4..0fae6cbd4 100644
--- a/nvx-rs/src/guest/managed_agent/mappings.rs
+++ b/nvx-rs/src/guest/managed_agent/mappings.rs
@@ -38,6 +38,8 @@ enum TableRoot {
     Direct,
     /// The sandbox agent's `ROOTFS`, the container's assembled root.
     Sandbox(PathBuf),
+    /// A layered root's retained mapper and export, never reopened by path.
+    Layered,
 }
 
 /// The mapper of the host mapping table beneath the table's root, and the root of the host's
@@ -46,10 +48,26 @@ struct TableMapper {
     /// The mapper, which tracks the table's mappings, and latches at the first that fails.
     mapper: Mapper,
     /// The host export's mount, `hostfs_dir/root`.
-    export: SourceRoot,
+    export: Option<SourceRoot>,
 }
 
 impl HostMappings {
+    /// Adopts a layered root's table and prepared mapping handles without resetting either.
+    pub(crate) fn layered(
+        table: MappingTable,
+        mapper: Mapper,
+        export: Option<SourceRoot>,
+    ) -> Result<Self, Errno> {
+        if table.expected() != 0 && export.is_none() {
+            return Err(Errno::ENOENT);
+        }
+        Ok(Self {
+            table,
+            root: TableRoot::Layered,
+            mapper: Some(TableMapper { mapper, export }),
+        })
+    }
+
     /// Returns the state of the table of the agent that `config` configures, before
     /// [`prepare`] reads its announcement: the direct agent maps beneath `/`, and the sandbox
     /// agent beneath `ROOTFS`.
@@ -138,9 +156,10 @@ pub(crate) fn apply_request(
                 None => open_table(system, root).map_err(|error| (0, error as i32))?,
             };
             let opened = mapper.insert(opened);
+            let export = opened.export.as_ref().ok_or((0, Errno::ENOENT as i32))?;
             opened
                 .mapper
-                .apply_with(system.kernel(), &opened.export, batch)
+                .apply_with(system.kernel(), export, batch)
                 .map_err(|failure| (failure.entry(), failure.error().raw_os_error()))
         })
         .map_err(|error| match error {
@@ -172,11 +191,12 @@ fn open_table(system: &mut impl System, root: &TableRoot) -> Result<TableMapper,
             TargetRoot::open_with(system.kernel(), rootfs).map_err(errno)?,
             Reserved::container_root(),
         ),
+        TableRoot::Layered => return Err(Errno::EINVAL),
     };
     let export = SourceRoot::open(&system.paths().hostfs_dir.join("root")).map_err(errno)?;
     Ok(TableMapper {
         mapper: Mapper::new(root, reserved),
-        export,
+        export: Some(export),
     })
 }
 
diff --git a/nvx-rs/src/guest/managed_agent/mappings/tests.rs b/nvx-rs/src/guest/managed_agent/mappings/tests.rs
index 54c617d94..913d194cb 100644
--- a/nvx-rs/src/guest/managed_agent/mappings/tests.rs
+++ b/nvx-rs/src/guest/managed_agent/mappings/tests.rs
@@ -133,6 +133,20 @@ fn a_table_without_mappings_has_nothing_to_tear_down() {
     assert!(system.kernel.calls().is_empty());
 }
 
+#[test]
+fn a_layered_table_that_announces_mappings_needs_its_retained_export() {
+    let (_temporary, mut system, _mappings, rootfs) = sandbox(1);
+    let mapper = Mapper::new(
+        TargetRoot::open_with(&mut system.kernel, &rootfs).unwrap(),
+        Reserved::container_root(),
+    );
+    assert_eq!(
+        HostMappings::layered(MappingTable::new(1).unwrap(), mapper, None).err(),
+        Some(Errno::ENOENT)
+    );
+    assert!(system.kernel.calls().is_empty());
+}
+
 fn payload(first: u32, entries: &[(&[u8], &[u8], u16)], count: Option<u32>) -> Vec<u8> {
     let mut payload = Vec::new();
     payload.extend_from_slice(&first.to_le_bytes());
diff --git a/nvx-rs/src/guest/managed_agent/session.rs b/nvx-rs/src/guest/managed_agent/session.rs
index bbf3f834c..99111a654 100644
--- a/nvx-rs/src/guest/managed_agent/session.rs
+++ b/nvx-rs/src/guest/managed_agent/session.rs
@@ -1,20 +1,127 @@
-use std::io;
+use std::cell::{Cell, RefCell};
+use std::io::{self, Read, Write};
+use std::process::ExitStatus;
 
+use crate::guest::init::Runtime;
 use crate::guest::nvxs::{GuestConnection, GuestConnectionError};
-use crate::guest::sys::{Errno, FileDescriptor};
+use crate::guest::sys::{self, Errno, FileDescriptor, Pollable};
 use crate::protocol::nvxs::{GuestEvent, RecordError, RecordType, SessionError};
 use crate::protocol::{self, APP_ERROR, OUTER_MAX_PAYLOAD, OUTER_RECEIVE_WINDOW};
 
 /// The agent's control session: the guest's role of the shared NVXS session on the control
 /// console, and the NVXC frames that its DATA records carry.
-pub(crate) struct Session {
+pub(crate) struct Session<'r> {
     /// The guest's NVXS connection on the owned control-console descriptor.
-    connection: GuestConnection<FileDescriptor>,
+    connection: GuestConnection<Control<'r>>,
     /// Error of the connection's latest failure, which a connection that a failure ended reports
     /// again.
     last_error: Option<Errno>,
 }
 
+/// Byte transport only: the existing connection still owns all record framing.
+struct Control<'r> {
+    descriptor: FileDescriptor,
+    runtime: Option<RefCell<&'r mut Runtime>>,
+    leader: Cell<u32>,
+    status: Cell<Option<(u32, ExitStatus)>>,
+    outcome: RefCell<Result<bool, String>>,
+}
+
+impl Control<'_> {
+    fn service_leader(&self) -> Result<(), Errno> {
+        if self.service(|id, status| {
+            // Keep the first leader status across partial I/O, not unrelated orphans.
+            if id == self.leader.get() && self.status.get().is_none() {
+                self.status.set(Some((id, status)));
+            }
+        })? {
+            return Err(Errno::ECANCELED);
+        }
+        Ok(())
+    }
+
+    fn service(&self, observed: impl FnMut(u32, ExitStatus)) -> Result<bool, Errno> {
+        if self.outcome.borrow().is_err() {
+            return Err(Errno::ECHILD);
+        }
+        if let Some(runtime) = &self.runtime {
+            *self.outcome.borrow_mut() = runtime.borrow_mut().service(observed);
+        } else {
+            super::exec::collect(observed);
+        }
+        self.outcome
+            .borrow()
+            .as_ref()
+            .copied()
+            .map_err(|_| Errno::ECHILD)
+    }
+}
+
+impl Read for Control<'_> {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        if bytes.is_empty() {
+            return Read::read(&mut self.descriptor, bytes);
+        }
+        let Some(runtime) = &self.runtime else {
+            return Read::read(&mut self.descriptor, bytes);
+        };
+        loop {
+            self.service_leader()?;
+            let signals = sys::poll(
+                [Pollable::File(&self.descriptor), runtime.borrow().wake()],
+                None,
+            )?[1];
+            if signals.error {
+                return Err(Errno::EIO.into());
+            }
+            if !signals.readable {
+                // A control-only wake includes HUP, which PollEvents does not expose.
+                match self.descriptor.read(bytes) {
+                    Err(Errno::EAGAIN) => {}
+                    result => return result.map_err(io::Error::from),
+                }
+            }
+        }
+    }
+}
+
+impl Write for Control<'_> {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        let Some(runtime) = &self.runtime else {
+            return Write::write(&mut self.descriptor, bytes);
+        };
+        if bytes.is_empty() {
+            return Write::write(&mut self.descriptor, bytes);
+        }
+        loop {
+            self.service_leader()?;
+            match self.descriptor.write(bytes) {
+                Err(Errno::EAGAIN) => loop {
+                    let events = sys::poll(
+                        [
+                            Pollable::Writable(&self.descriptor),
+                            runtime.borrow().wake(),
+                        ],
+                        None,
+                    )?;
+                    if events[1].error {
+                        return Err(Errno::EIO.into());
+                    }
+                    self.service_leader()?;
+                    if events[0].writable || !events[1].readable {
+                        break;
+                    }
+                },
+                result => return result.map_err(io::Error::from),
+            }
+        }
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Write::flush(&mut self.descriptor)
+    }
+}
+
 /// Record from the broker that the session accepted.
 pub(crate) enum Event {
     /// The broker reset the session. The agent abandons the earlier epoch's work, then
@@ -82,20 +189,88 @@ impl DataFailure {
     }
 }
 
-impl Session {
+impl<'r> Session<'r> {
     /// Starts unattached session state on an owned control descriptor.
     pub(crate) fn new(descriptor: FileDescriptor) -> Self {
+        Self::open(descriptor, None)
+    }
+
+    fn open(descriptor: FileDescriptor, runtime: Option<&'r mut Runtime>) -> Self {
         Self {
-            connection: GuestConnection::new(descriptor),
+            connection: GuestConnection::new(Control {
+                descriptor,
+                runtime: runtime.map(RefCell::new),
+                leader: Cell::new(0),
+                status: Cell::new(None),
+                outcome: RefCell::new(Ok(false)),
+            }),
             last_error: None,
         }
     }
 
+    /// Borrows init's sole collector for the lifetime of this control session.
+    pub(crate) fn with_runtime(
+        descriptor: FileDescriptor,
+        runtime: &'r mut Runtime,
+    ) -> Result<Self, String> {
+        let session = Self::open(descriptor, Some(runtime));
+        session
+            .connection
+            .get_ref()
+            .service(|_, _| {})
+            .map_err(|_| {
+                session
+                    .runtime_failure()
+                    .expect("the collector retained its failure")
+            })?;
+        session
+            .descriptor()
+            .set_nonblocking()
+            .map_err(|error| format!("cannot configure layered control: {}", error.desc()))?;
+        Ok(session)
+    }
+
+    /// Registers the newly launched leader before any control I/O can collect it.
+    pub(crate) fn watch(&self, id: u32) {
+        self.connection.get_ref().leader.set(id);
+        self.connection.get_ref().status.set(None);
+    }
+
+    /// Takes a status collected while a control record was being read or written.
+    pub(crate) fn take_status(&self) -> Option<(u32, ExitStatus)> {
+        self.connection.get_ref().status.take()
+    }
+
+    /// Collects in the caller's existing position, after handing over any earlier status.
+    pub(crate) fn collect(&self, mut observed: impl FnMut(u32, ExitStatus)) -> Result<(), Errno> {
+        if let Some((id, status)) = self.take_status() {
+            observed(id, status);
+        }
+        if self.connection.get_ref().service(observed)? {
+            return Err(Errno::ECANCELED);
+        }
+        Ok(())
+    }
+
+    /// A helper failure overrides even an already-latched normal termination.
+    pub(crate) fn stopping(&self) -> bool {
+        matches!(*self.connection.get_ref().outcome.borrow(), Ok(true))
+    }
+
+    /// Returns the original init diagnostic to its shutdown owner.
+    pub(crate) fn runtime_failure(&self) -> Option<String> {
+        std::mem::replace(
+            &mut *self.connection.get_ref().outcome.borrow_mut(),
+            Ok(false),
+        )
+        .err()
+    }
+
     /// Borrows the control descriptor for polling and TTY operations. The connection never reads
     /// past the record in progress and holds no partial record between calls, so polling sees
     /// every record that the session has not returned.
     pub(crate) fn descriptor(&self) -> &FileDescriptor {
-        self.connection.get_ref()
+        &self.connection.get_ref().descriptor
     }
 
     /// Sends the initial guest-attach record.
diff --git a/nvx-rs/src/guest/managed_agent/tests.rs b/nvx-rs/src/guest/managed_agent/tests.rs
index 1b92c42f8..52f07529c 100644
--- a/nvx-rs/src/guest/managed_agent/tests.rs
+++ b/nvx-rs/src/guest/managed_agent/tests.rs
@@ -50,6 +50,53 @@ fn sandbox_config(hostname: &[u8], uid: &[u8]) -> AgentConfig {
     }
 }
 
+#[test]
+fn a_layered_cold_boot_takes_only_the_clis_workload_identity() {
+    for command_line in [
+        b"nvx_workload_uid=9 nvx_workload_gid=10 nvx_identity=1000:1001".as_slice(),
+        b"x\x0cnvx_identity=1001:1001 nvx_identity=1000:1001",
+    ] {
+        let config = AgentConfig::layered(
+            command_line,
+            c"workload".to_owned(),
+            c"/home/workload".to_owned(),
+        )
+        .unwrap();
+        assert_eq!(config.uid.as_bytes(), b"1000");
+        assert_eq!(config.gid.as_bytes(), b"1001");
+        assert_eq!(config.user.as_c_str(), c"workload");
+        assert_eq!(config.home.as_c_str(), c"/home/workload");
+        assert!(config.direct);
+        assert!(config.entry.is_none());
+    }
+}
+
+#[test]
+fn a_layered_cold_boot_refuses_missing_repeated_or_invalid_cli_identities() {
+    for command_line in [
+        b"nvx_workload_uid=1000 nvx_workload_gid=1000".as_slice(),
+        b"nvx_identity=1000:1000 nvx_identity=1001:1001",
+        b"nvx_identity=",
+        b"nvx_identity=1000",
+        b"nvx_identity=:1000",
+        b"nvx_identity=1000:",
+        b"nvx_identity=1000:1000:1000",
+        b"nvx_identity=0:1000",
+        b"nvx_identity=1000:000",
+        b"nvx_identity=4294967295:1000",
+        b"nvx_identity=1000:4294967296",
+        b"nvx_identity=+1000:1000",
+        b"x\rnvx_identity=1000:1000",
+        b"x\x0cnvx_identity=1000:1000",
+    ] {
+        assert_eq!(
+            AgentConfig::layered(command_line, c"workload".to_owned(), c"/tmp".to_owned()).err(),
+            Some(Errno::EINVAL),
+            "{command_line:?}"
+        );
+    }
+}
+
 #[test]
 fn the_sandbox_entry_takes_an_existing_account_and_one_machine_id_per_boot() {
     let (_temporary, system) = TestSystem::with_container_cgroup();
diff --git a/nvx-rs/src/guest/sys/poll.rs b/nvx-rs/src/guest/sys/poll.rs
index 65078733d..a6f9cfc0e 100644
--- a/nvx-rs/src/guest/sys/poll.rs
+++ b/nvx-rs/src/guest/sys/poll.rs
@@ -37,11 +37,13 @@ pub(crate) fn hung_up(descriptor: BorrowedFd<'_>) -> Result<bool, Errno> {
         .is_some_and(|events| events.contains(PollFlags::POLLHUP)))
 }
 
-/// A descriptor owner that [`poll`] waits on for input.
+/// A descriptor owner and the readiness that [`poll`] waits for.
 #[derive(Clone, Copy, Debug)]
 pub(crate) enum Pollable<'a> {
     /// A file, device, pipe, or socket.
     File(&'a FileDescriptor),
+    /// A file, device, pipe, or socket with space to write.
+    Writable(&'a FileDescriptor),
     /// A timer, which polls readable once it expires.
     Timer(&'a Timer),
     /// A signal descriptor, which polls readable while one of its signals is pending.
@@ -54,7 +56,7 @@ impl Pollable<'_> {
     /// Borrows the owner's descriptor.
     fn borrowed(&self) -> BorrowedFd<'_> {
         match self {
-            Self::File(descriptor) => descriptor.borrowed(),
+            Self::File(descriptor) | Self::Writable(descriptor) => descriptor.borrowed(),
             Self::Timer(timer) => timer.borrowed(),
             Self::Signals(signals) => signals.borrowed(),
             Self::ChildSignal(wait) => wait.borrowed(),
@@ -67,12 +69,14 @@ impl Pollable<'_> {
 pub(crate) struct PollEvents {
     /// The descriptor has input to read.
     pub(crate) readable: bool,
-    /// The descriptor has an error condition, which a read returns.
+    /// The descriptor has space to write.
+    pub(crate) writable: bool,
+    /// The descriptor has an error condition, which its next I/O returns.
     pub(crate) error: bool,
 }
 
-/// Waits at most `timeout`, or without a limit for `None`, until one of `descriptors` has input
-/// or an error, with one `poll` call and without allocating, and returns each descriptor's
+/// Waits at most `timeout`, or without a limit for `None`, until one of `descriptors` has its
+/// requested readiness or an error, with one `poll` call and without allocating, and returns their
 /// events in the caller's order. An interrupted call is `EINTR`, not a retry.
 ///
 /// A timeout of more than `i32::MAX` milliseconds waits for `i32::MAX` milliseconds, about 24
@@ -84,14 +88,19 @@ pub(crate) fn poll<const N: usize>(
     let timeout = timeout.map_or(PollTimeout::NONE, |timeout| {
         PollTimeout::try_from(timeout.as_millis()).unwrap_or(PollTimeout::MAX)
     });
-    let mut raw = descriptors
-        .each_ref()
-        .map(|descriptor| PollFd::new(descriptor.borrowed(), PollFlags::POLLIN));
+    let mut raw = descriptors.each_ref().map(|descriptor| {
+        let interest = match descriptor {
+            Pollable::Writable(_) => PollFlags::POLLOUT,
+            _ => PollFlags::POLLIN,
+        };
+        PollFd::new(descriptor.borrowed(), interest)
+    });
     nix::poll::poll(&mut raw, timeout)?;
     Ok(raw.map(|descriptor| {
         let returned = descriptor.revents().unwrap_or(PollFlags::empty());
         PollEvents {
             readable: returned.contains(PollFlags::POLLIN),
+            writable: returned.contains(PollFlags::POLLOUT),
             error: returned.contains(PollFlags::POLLERR),
         }
     }))
diff --git a/nvx-rs/src/guest/sys/poll/tests.rs b/nvx-rs/src/guest/sys/poll/tests.rs
index dfe8e631e..6054c63e0 100644
--- a/nvx-rs/src/guest/sys/poll/tests.rs
+++ b/nvx-rs/src/guest/sys/poll/tests.rs
@@ -19,8 +19,35 @@ fn poll_reports_input_in_the_callers_order() {
             PollEvents::default(),
             PollEvents {
                 readable: true,
+                writable: false,
                 error: false
             }
         ]
     );
 }
+
+#[test]
+fn poll_distinguishes_backpressure_from_a_readable_wake() {
+    let (reader, writer) = create_pipe().unwrap();
+    writer.set_nonblocking().unwrap();
+    assert!(poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable);
+    let bytes = [0; 4096];
+    loop {
+        match writer.write(&bytes) {
+            Ok(count) => assert!(count != 0),
+            Err(Errno::EAGAIN) => break,
+            Err(error) => panic!("cannot fill the owned pipe: {error}"),
+        }
+    }
+    let (wake, signal) = create_pipe().unwrap();
+    signal.write_all(b"x").unwrap();
+    let events = poll(
+        [Pollable::Writable(&writer), Pollable::File(&wake)],
+        Some(Duration::ZERO),
+    )
+    .unwrap();
+    assert_eq!(events[0], PollEvents::default());
+    assert!(events[1].readable && !events[1].writable);
+    reader.read_exact(&mut [0; 4096]).unwrap();
+    assert!(poll([Pollable::Writable(&writer)], Some(Duration::ZERO)).unwrap()[0].writable);
+}
diff --git a/nvx-rs/src/guest/sys/process/tests.rs b/nvx-rs/src/guest/sys/process/tests.rs
index b75477f61..ca1512d01 100644
--- a/nvx-rs/src/guest/sys/process/tests.rs
+++ b/nvx-rs/src/guest/sys/process/tests.rs
@@ -17,7 +17,7 @@ pub(crate) fn isolate_process_test(marker: &str, test: &str) -> bool {
     let status = std::process::Command::new(
         std::env::current_exe().expect("the test executable has a path"),
     )
-    .args(["--exact", test, "--nocapture"])
+    .args(["--exact", test, "--nocapture", "--include-ignored"])
     .env(marker, "1")
     .status()
     .expect("the isolated test process starts");
diff --git a/nvx-rs/tests/managed_lifecycle.rs b/nvx-rs/tests/managed_lifecycle.rs
index 57b9aa7dc..3b7a895f8 100644
--- a/nvx-rs/tests/managed_lifecycle.rs
+++ b/nvx-rs/tests/managed_lifecycle.rs
@@ -15,6 +15,9 @@
 //! - `NVX_MANAGED_LIFECYCLE_TIMEOUT`: the longest wait for each operation, in seconds.
 //! - `NVX_MANAGED_LIFECYCLE_OUTPUT_DIR`: receives OpenVMM's output, `managed-lifecycle.log`, and
 //!   the guest console, `managed-lifecycle-guest.log`, which the scenario then scans.
+//! - `NVX_LAYERED_INIT_MAPS`, only for the dedicated real layered-init fixture: `0` for a live
+//!   share without an export, or `2` for the retained host export and an ordered mapping table.
+//!   That fixture boots the actual `nvx-init` as PID 1 with CLI identity `1000:1001`.
 //!
 //! The test also records OpenVMM's process ID in `NVX_ADVERSARIAL_OPENVMM_PID_JOURNAL`, if set,
 //! as the microVM scenarios do for the adversarial campaigns.
@@ -36,7 +39,10 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use nvx::host::instance::generate_capability;
 use nvx::host::platform::process_start_time;
 use nvx::host::session::{self, ControlSession, ExecEvent};
-use nvx::protocol::{CAPABILITY_LEN, CWD_FAILED, ExitCategory, Workload, WorkloadEnvironment};
+use nvx::protocol::{
+    self, APP_ERROR, APP_MAPS, APP_READY, CAPABILITY_LEN, CWD_FAILED, ExitCategory, HostMap,
+    Workload, WorkloadEnvironment,
+};
 
 /// An entry whose value has spaces, an equals sign, and a character beyond ASCII.
 const COMPLEX: &str = "COMPLEX=space = \u{2603}";
@@ -363,10 +369,136 @@ fn the_process_id_journal_has_the_entries_that_the_adversarial_executor_reads()
     assert_eq!(pids, [4321, u64::from(u32::MAX)]);
 }
 
+fn layered_mapping_session(
+    guest: &Guest,
+    mut session: ControlSession,
+    count: u32,
+) -> ControlSession {
+    assert!(
+        matches!(count, 0 | 2),
+        "the layered fixture has zero or two mappings"
+    );
+    let maps = ["/workspace", "/run/task-state"].map(|target| HostMap {
+        source: "0".to_owned(),
+        target: target.to_owned(),
+        read_only: true,
+    });
+    let parts = protocol::encode_maps_payloads(&maps, 40).unwrap();
+    assert_eq!(parts.len(), 2);
+    let send = |session: &mut ControlSession, payload: &[u8], accepted| {
+        let deadline = Instant::now() + guest.timeout;
+        session
+            .send_request(APP_MAPS, 900, payload, deadline)
+            .unwrap();
+        let expected = if accepted {
+            (APP_READY, 900, 0, Vec::new())
+        } else {
+            (
+                APP_ERROR,
+                900,
+                22,
+                protocol::INVALID_REQUEST.as_bytes().to_vec(),
+            )
+        };
+        assert_eq!(session.read_response(Some(deadline)).unwrap(), expected);
+    };
+    if count == 0 {
+        send(&mut session, &parts[0], false);
+    } else {
+        let incomplete = |session: &mut ControlSession| {
+            let outcome = guest.exec(
+                session,
+                &["/bin/true"],
+                5_000,
+                None,
+                WorkloadEnvironment::Default,
+            );
+            assert_eq!(
+                outcome.refusal,
+                Some((11, protocol::MAPPINGS_INCOMPLETE.to_owned()))
+            );
+            assert!(outcome.exit.is_none());
+        };
+        incomplete(&mut session);
+        send(&mut session, &parts[1], false);
+        send(&mut session, &parts[0], true);
+        drop(session);
+        session = guest.connect();
+        incomplete(&mut session);
+        send(&mut session, &parts[1], true);
+    }
+    drop(session);
+    session = guest.connect();
+    session.ping(Instant::now() + guest.timeout).unwrap();
+    send(&mut session, &parts[0], false);
+    let defaults = guest.exec(
+        &mut session,
+        &[
+            "/bin/sh",
+            "-c",
+            "test \"$(cat /proc/1/comm)\" = nvx-init && \
+             test \"$(cat /proc/sys/kernel/hostname)\" = sl76b-real-root && \
+             test -s /run/nvx/time/state && test \"$(wc -c </etc/machine-id)\" -eq 33 && \
+             test ! -e /run/nvx/hostfs/root && \
+             printf '%s:%s|%s|%s|%s' \"$(id -u)\" \"$(id -g)\" \"$USER\" \"$HOME\" \"$PWD\"",
+        ],
+        5_000,
+        None,
+        WorkloadEnvironment::Default,
+    );
+    assert!(defaults.printed("1000:1001|nvx|/tmp|/"), "{defaults:?}");
+    let targets: &[&str] = if count == 0 {
+        &["/run/app/data"]
+    } else {
+        &["/workspace/data", "/run/task-state/data"]
+    };
+    for path in targets {
+        let mapped = guest.exec(
+            &mut session,
+            &["/bin/cat", path],
+            5_000,
+            None,
+            WorkloadEnvironment::Default,
+        );
+        assert!(mapped.printed("sl76b-real-export\n"), "{path}: {mapped:?}");
+    }
+    let request = session
+        .start_exec(
+            &Workload {
+                argv: vec!["/bin/sleep".to_owned(), "30".to_owned()],
+                timeout_ms: 0,
+                cwd: None,
+                environment: WorkloadEnvironment::Default,
+            },
+            Instant::now() + guest.timeout,
+        )
+        .unwrap();
+    session
+        .cancel_exec(request, Instant::now() + guest.timeout)
+        .unwrap();
+    assert!(matches!(
+        session
+            .next_exec_event(request, Some(Instant::now() + guest.timeout))
+            .unwrap(),
+        ExecEvent::Exit {
+            category: ExitCategory::Cancelled,
+            status: 137
+        }
+    ));
+    session
+}
+
 #[test]
 #[ignore = "requires a hypervisor and NVX guest artifacts; run scripts/nvx.py test-microvm \
             --scenario managed-lifecycle"]
 fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() {
+    let layered_maps = env::var_os("NVX_LAYERED_INIT_MAPS").map(|value| {
+        value
+            .into_string()
+            .expect("NVX_LAYERED_INIT_MAPS is ASCII")
+            .parse::<u32>()
+            .expect("NVX_LAYERED_INIT_MAPS is 0 or 2")
+    });
     let command: Vec<String> = serde_json::from_str(&variable("NVX_MANAGED_LIFECYCLE_COMMAND"))
         .expect("NVX_MANAGED_LIFECYCLE_COMMAND is a JSON array of strings");
     let timeout = Duration::from_secs_f64(
@@ -415,6 +547,9 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() {
     session
         .ping(Instant::now() + timeout)
         .expect("the guest agent is ready");
+    if let Some(count) = layered_maps {
+        session = layered_mapping_session(&guest, session, count);
+    }
     let exact = vec!["EMPTY=".to_owned(), COMPLEX.to_owned()];
     let first = guest.exec(
         &mut session,
@@ -476,7 +611,11 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() {
         WorkloadEnvironment::Default,
     );
     assert!(
-        identity.printed(WORKLOAD_IDENTITY),
+        identity.printed(if layered_maps.is_some() {
+            "1000:1001"
+        } else {
+            WORKLOAD_IDENTITY
+        }),
         "the workload identity is {identity:?}"
     );
     // State persists across requests, while working directories and environments do not, and a
@@ -603,4 +742,22 @@ fn a_managed_guest_runs_request_scoped_workloads_until_it_stops() {
         status.success(),
         "the managed OpenVMM process exited with {status}"
     );
+    if let Some(count) = layered_maps {
+        let output = [
+            fs::read(output_dir.join("managed-lifecycle-guest.log")).unwrap(),
+            fs::read(output_dir.join("managed-lifecycle.log")).unwrap(),
+        ]
+        .concat();
+        let marker = format!("NVX-INIT-ROOT: switched mappings={count}");
+        assert!(
+            output
+                .windows(marker.len())
+                .any(|bytes| bytes == marker.as_bytes())
+        );
+        assert!(
+            !output
+                .windows(b"NVX-INIT-ERROR:".len())
+                .any(|bytes| bytes == b"NVX-INIT-ERROR:")
+        );
+    }
 }
diff --git a/scripts/test-layout-allowlist.json b/scripts/test-layout-allowlist.json
index ca8ea1991..5b7e6aa10 100644
--- a/scripts/test-layout-allowlist.json
+++ b/scripts/test-layout-allowlist.json
@@ -5,7 +5,7 @@
     "nvx-rs/src/guest/cgroup.rs": 2,
     "nvx-rs/src/guest/entry.rs": 1,
     "nvx-rs/src/guest/exec.rs": 4,
-    "nvx-rs/src/guest/init.rs": 8,
+    "nvx-rs/src/guest/init.rs": 7,
     "nvx-rs/src/guest/init/boot.rs": 1,
     "nvx-rs/src/guest/reseed.rs": 1,
     "nvx-rs/src/guest/sys/entry.rs": 3,
diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md
index 5c16c3525..d700bda5f 100644
--- a/tests/runtime-adapter/README.md
+++ b/tests/runtime-adapter/README.md
@@ -2559,6 +2559,11 @@ configuration, on Linux and Windows:
   in the initramfs too, and missing install metadata, which fails as NVX
   reports it.
 
+`tests/listener_client.rs` checks, in the host-only configuration, the real TCP
+greeting and answer of the VM tests' listener client on Linux and Windows.
+On Linux it also checks that a self-connection leaves the guest's publication
+port bindable and that a connection reset before inspection is retried.
+
 `tests/staged_images.rs` checks, in the host-only configuration, on Linux and
 Windows:
 
@@ -3217,8 +3222,8 @@ layered roots, captures of templates, and the starts and runs of their clones.
 
 The VM tests, `tests/vm_restart.rs`, `tests/vm_steps.rs`,
 `tests/vm_layered.rs`, `tests/vm_template.rs`, and `tests/vm_clone.rs`, share
-their sandboxes and
-the fixtures of their layered roots through `tests/common`. The sandboxes start
+their sandboxes, the fixtures of their layered roots, and their host listener
+client through `tests/common`. The sandboxes start
 the host program's commands with `nvx::host::supervisor::spawn_command`, so
 that on Windows a command cannot inherit the handles of a VM that a test
 launches in its own process meanwhile, such as its log and its capability's
diff --git a/tests/runtime-adapter/tests/common/mod.rs b/tests/runtime-adapter/tests/common/mod.rs
index f27089851..3c3540923 100644
--- a/tests/runtime-adapter/tests/common/mod.rs
+++ b/tests/runtime-adapter/tests/common/mod.rs
@@ -20,9 +20,11 @@ use std::env;
 use std::error::Error;
 use std::ffi::{OsStr, OsString};
 use std::fs::{self, File};
-use std::io;
+use std::io::{self, Read, Write};
+use std::net::{Shutdown, SocketAddr, TcpStream};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::thread;
 use std::time::{Duration, Instant};
 
@@ -468,6 +470,78 @@ impl Drop for Sandbox {
     }
 }
 
+/// Exchanges a line each way with the guest until it succeeds or `stop` is set.
+/// A connect runs to completion: cancelling a Windows loopback connect can leave a late
+/// connection that consumes the guest's one-shot listener without exchanging its greeting.
+pub(crate) fn reach_guest(
+    address: SocketAddr,
+    stop: &AtomicBool,
+    greeting: &str,
+    answer: &str,
+) -> bool {
+    reach_guest_with(address, stop, greeting, answer, TcpStream::connect)
+}
+
+/// Runs the listener client with a connector that can supply a real socket of a known kind.
+pub(crate) fn reach_guest_with(
+    address: SocketAddr,
+    stop: &AtomicBool,
+    greeting: &str,
+    answer: &str,
+    mut connect: impl FnMut(SocketAddr) -> io::Result<TcpStream>,
+) -> bool {
+    while !stop.load(Ordering::SeqCst) {
+        let connected = connect(address);
+        #[cfg(target_os = "linux")]
+        let connected = connected.and_then(|stream| {
+            let peer = match stream.peer_addr() {
+                Ok(peer) => peer,
+                Err(error) if error.kind() == io::ErrorKind::NotConnected => return Err(error),
+                Err(error) => panic!("inspect the listener client's peer: {error}"),
+            };
+            if stream.local_addr().unwrap() == peer {
+                // Linux can assign the unpublished destination as this client's source port.
+                // Reset that self-connection so TIME_WAIT cannot block the later publication.
+                nix::sys::socket::setsockopt(
+                    &stream,
+                    nix::sys::socket::sockopt::Linger,
+                    &nix::libc::linger {
+                        l_onoff: 1,
+                        l_linger: 0,
+                    },
+                )
+                .expect("reset the client's self-connection before publishing the guest's port");
+                return Err(io::ErrorKind::ConnectionAborted.into());
+            }
+            Ok(stream)
+        });
+        if let Ok(mut stream) = connected {
+            stream
+                .set_read_timeout(Some(Duration::from_secs(5)))
+                .unwrap();
+            let mut received = Vec::new();
+            let mut byte = [0];
+            while !received.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) {
+                received.push(byte[0]);
+            }
+            if received == format!("{greeting}\n").as_bytes() {
+                stream.write_all(format!("{answer}\n").as_bytes()).unwrap();
+                stream.shutdown(Shutdown::Write).unwrap();
+                let mut rest = Vec::new();
+                stream.read_to_end(&mut rest).unwrap();
+                assert!(rest.is_empty(), "the guest's listener then sent {rest:?}");
+                return true;
+            }
+            assert!(
+                received.is_empty(),
+                "the guest's listener greeted the host with {received:?}"
+            );
+        }
+        thread::sleep(Duration::from_millis(100));
+    }
+    false
+}
+
 /// Describes `error`, with its causes.
 pub(crate) fn describe(error: &dyn Error) -> String {
     let mut description = error.to_string();
diff --git a/tests/runtime-adapter/tests/listener_client.rs b/tests/runtime-adapter/tests/listener_client.rs
new file mode 100644
index 000000000..f1a3c9f64
--- /dev/null
+++ b/tests/runtime-adapter/tests/listener_client.rs
@@ -0,0 +1,146 @@
+//! Real TCP regressions for the hardware scenarios' listener client.
+#![cfg(feature = "host")]
+
+mod common;
+
+use std::io::{Read, Write};
+use std::net::TcpListener;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
+
+#[test]
+fn a_published_listener_exchanges_its_greeting() {
+    let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
+    let address = listener.local_addr().unwrap();
+    let stop = AtomicBool::new(false);
+    let (finished, completion) = mpsc::channel();
+    thread::scope(|scope| {
+        let stop_ref = &stop;
+        scope.spawn(move || {
+            if completion.recv_timeout(Duration::from_secs(5)).is_err() {
+                stop_ref.store(true, Ordering::SeqCst);
+            }
+        });
+        let server = scope.spawn(|| {
+            let (mut stream, _) = listener.accept().unwrap();
+            stream.write_all(b"guest\n").unwrap();
+            let mut answer = Vec::new();
+            stream.read_to_end(&mut answer).unwrap();
+            assert_eq!(answer, b"host\n");
+        });
+        let reached = common::reach_guest(address, &stop, "guest", "host");
+        finished
+            .send(())
+            .expect("the listener exchange must finish before its deadline");
+        assert!(reached);
+        server.join().unwrap();
+    });
+}
+
+#[test]
+#[cfg(target_os = "linux")]
+fn a_connection_reset_before_inspection_is_retried() {
+    use std::net::TcpStream;
+
+    use nix::sys::socket::{setsockopt, sockopt};
+
+    let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
+    let address = listener.local_addr().unwrap();
+    let mut stream = TcpStream::connect(address).unwrap();
+    let (server, _) = listener.accept().unwrap();
+    setsockopt(
+        &server,
+        sockopt::Linger,
+        &nix::libc::linger {
+            l_onoff: 1,
+            l_linger: 0,
+        },
+    )
+    .unwrap();
+    drop(server);
+    stream
+        .set_read_timeout(Some(Duration::from_secs(1)))
+        .unwrap();
+    assert_eq!(
+        stream.read(&mut [0]).unwrap_err().kind(),
+        std::io::ErrorKind::ConnectionReset
+    );
+
+    let mut supplied = Some(stream);
+    let mut retried = false;
+    let stop = AtomicBool::new(false);
+    assert!(!common::reach_guest_with(
+        address,
+        &stop,
+        "guest",
+        "host",
+        |_| {
+            if let Some(stream) = supplied.take() {
+                Ok(stream)
+            } else {
+                retried = true;
+                stop.store(true, Ordering::SeqCst);
+                Err(std::io::ErrorKind::ConnectionRefused.into())
+            }
+        }
+    ));
+    assert!(
+        retried,
+        "a reset connection must retry instead of panicking"
+    );
+}
+
+#[test]
+#[cfg(target_os = "linux")]
+fn a_self_connection_does_not_hold_the_publication_port() {
+    use std::net::TcpStream;
+    use std::os::fd::AsRawFd;
+
+    use nix::sys::socket::{
+        AddressFamily, SockFlag, SockProtocol, SockType, SockaddrIn, bind, connect, getsockname,
+        setsockopt, socket, sockopt,
+    };
+
+    let client = socket(
+        AddressFamily::Inet,
+        SockType::Stream,
+        SockFlag::SOCK_CLOEXEC,
+        SockProtocol::Tcp,
+    )
+    .unwrap();
+    bind(client.as_raw_fd(), &SockaddrIn::new(127, 0, 0, 1, 0)).unwrap();
+    let address: SockaddrIn = getsockname(client.as_raw_fd()).unwrap();
+    connect(client.as_raw_fd(), &address).unwrap();
+    let stream = TcpStream::from(client);
+    assert_eq!(stream.local_addr().unwrap(), stream.peer_addr().unwrap());
+    let peer = stream.peer_addr().unwrap();
+    let mut supplied = Some(stream);
+    let stop = AtomicBool::new(false);
+    assert!(!common::reach_guest_with(
+        peer,
+        &stop,
+        "guest",
+        "host",
+        |_| {
+            if let Some(stream) = supplied.take() {
+                Ok(stream)
+            } else {
+                stop.store(true, Ordering::SeqCst);
+                Err(std::io::ErrorKind::ConnectionRefused.into())
+            }
+        }
+    ));
+
+    let publisher = socket(
+        AddressFamily::Inet,
+        SockType::Stream,
+        SockFlag::SOCK_CLOEXEC,
+        SockProtocol::Tcp,
+    )
+    .unwrap();
+    setsockopt(&publisher, sockopt::ReuseAddr, &true).unwrap();
+    bind(publisher.as_raw_fd(), &address)
+        .expect("the client must not leave the guest's publication port in TIME_WAIT");
+}
diff --git a/tests/runtime-adapter/tests/vm_clone.rs b/tests/runtime-adapter/tests/vm_clone.rs
index 21b4414ce..654fad486 100644
--- a/tests/runtime-adapter/tests/vm_clone.rs
+++ b/tests/runtime-adapter/tests/vm_clone.rs
@@ -208,11 +208,11 @@ use std::env;
 use std::ffi::{OsStr, OsString};
 use std::fs::{self, File};
 use std::hash::{BuildHasher, RandomState};
-use std::io::{self, Read, Write};
+use std::io;
+#[cfg(target_os = "linux")]
+use std::io::Read;
 use std::mem;
-use std::net::{
-    IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket,
-};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, UdpSocket};
 use std::num::NonZeroU64;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -4786,32 +4786,7 @@ fn a_clone_whose_mapping_fails_ends_before_its_sandbox_starts() {
 /// timeout gave up on could still open once OpenVMM publishes the listener, which would then take
 /// that connection, already closed, for its one client.
 fn reach_guest(address: SocketAddr, stop: &AtomicBool) -> bool {
-    while !stop.load(Ordering::SeqCst) {
-        if let Ok(mut stream) = TcpStream::connect(address) {
-            stream
-                .set_read_timeout(Some(Duration::from_secs(5)))
-                .unwrap();
-            let mut greeting = Vec::new();
-            let mut byte = [0];
-            while !greeting.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) {
-                greeting.push(byte[0]);
-            }
-            if greeting == format!("{GREETING}\n").as_bytes() {
-                stream.write_all(format!("{ANSWER}\n").as_bytes()).unwrap();
-                stream.shutdown(Shutdown::Write).unwrap();
-                let mut rest = Vec::new();
-                stream.read_to_end(&mut rest).unwrap();
-                assert!(rest.is_empty(), "the guest's listener then sent {rest:?}");
-                return true;
-            }
-            assert!(
-                greeting.is_empty(),
-                "the guest's listener greeted the host with {greeting:?}"
-            );
-        }
-        thread::sleep(Duration::from_millis(100));
-    }
-    false
+    common::reach_guest(address, stop, GREETING, ANSWER)
 }
 
 /// Sets its flag once it drops, so that a thread that waits for the flag ends even when the check
diff --git a/tests/runtime-adapter/tests/vm_layered.rs b/tests/runtime-adapter/tests/vm_layered.rs
index 00307e614..5284504fa 100644
--- a/tests/runtime-adapter/tests/vm_layered.rs
+++ b/tests/runtime-adapter/tests/vm_layered.rs
@@ -43,9 +43,7 @@ use std::ffi::{OsStr, OsString};
 use std::fs::{self, File};
 use std::hash::{BuildHasher, RandomState};
 use std::io::{self, Read, Seek, SeekFrom, Write};
-use std::net::{
-    IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket,
-};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, TcpStream, UdpSocket};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Mutex, PoisonError};
@@ -3017,32 +3015,7 @@ fn check_guest_listener(vm: &mut Vm<'_>, address: SocketAddr, reached: bool) {
 /// OpenVMM publishes the listener, which would then take that connection, already closed, for its
 /// one client.
 fn reach_guest(address: SocketAddr, stop: &AtomicBool) -> bool {
-    while !stop.load(Ordering::SeqCst) {
-        if let Ok(mut stream) = TcpStream::connect(address) {
-            stream
-                .set_read_timeout(Some(Duration::from_secs(5)))
-                .unwrap();
-            let mut greeting = Vec::new();
-            let mut byte = [0];
-            while !greeting.ends_with(b"\n") && matches!(stream.read(&mut byte), Ok(1)) {
-                greeting.push(byte[0]);
-            }
-            if greeting == format!("{GREETING}\n").as_bytes() {
-                stream.write_all(format!("{ANSWER}\n").as_bytes()).unwrap();
-                stream.shutdown(Shutdown::Write).unwrap();
-                let mut rest = Vec::new();
-                stream.read_to_end(&mut rest).unwrap();
-                assert!(rest.is_empty(), "the guest's listener then sent {rest:?}");
-                return true;
-            }
-            assert!(
-                greeting.is_empty(),
-                "the guest's listener greeted the host with {greeting:?}"
-            );
-        }
-        thread::sleep(Duration::from_millis(100));
-    }
-    false
+    common::reach_guest(address, stop, GREETING, ANSWER)
 }
 
 /// Serves the test proxy's `listener` until `stop`: answers each request with [`PROXY_RESPONSE`],