diff --git a/Cargo.lock b/Cargo.lock
index c7bbf979..67f8d545 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4214,9 +4214,9 @@ dependencies = [
 
 [[package]]
 name = "tinyvec"
-version = "1.13.0"
+version = "1.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ba2077be5cd93c7408c849e45a4ab261b59f923f4bcdee2a724b8ed5a175a77"
+checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b"
 dependencies = [
  "tinyvec_macros",
 ]
diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs
index e61d3fd9..c3f977cb 100644
--- a/apps/neshub/src/ipc_listener.rs
+++ b/apps/neshub/src/ipc_listener.rs
@@ -292,7 +292,24 @@ pub async fn run_stats_ipc_listener(
     tracing::info!("stats IPC listener exited");
 }
 
-pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::NestriTicket) {
+/// Serve the address a client needs, rebuilt on every read.
+///
+/// **The ticket is built per connection and not once at startup.** An endpoint
+/// does not know all of its own addresses when it binds: a direct one is there
+/// immediately, and a relayed or hole-punched one becomes known seconds later.
+/// A ticket captured once therefore carries only the address that was available
+/// first, which is the one that works on the same network and fails from
+/// anywhere else — and whoever reads this polls precisely so that a better
+/// answer can replace it. Serving a snapshot made that polling pointless: every
+/// read returned the same local-only address forever.
+///
+/// The stream name is generated once and kept, because it identifies this
+/// session rather than describing how to reach it. Only the addresses change.
+pub async fn run_ticket_ipc_listener(
+    socket_path: PathBuf,
+    endpoint: iroh::Endpoint,
+    stream_name: String,
+) {
     if socket_path.exists() {
         let _ = std::fs::remove_file(&socket_path);
     }
@@ -319,6 +336,9 @@ pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::Nestri
     loop {
         match listener.accept().await {
             Ok((mut stream, _)) => {
+                // Asked of the endpoint now, so an address it has learned since
+                // the last read is in this answer.
+                let ticket = crate::NestriTicket::new(endpoint.addr(), stream_name.clone());
                 if let Err(e) = stream.write_all(format!("{ticket}\n").as_bytes()).await {
                     tracing::warn!("could not write ticket to IPC: {e}");
                 }
diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs
index 27636cd1..9cb96bc9 100644
--- a/apps/neshub/src/main.rs
+++ b/apps/neshub/src/main.rs
@@ -207,7 +207,10 @@ async fn main() -> Result<()> {
 
     // ── Accept mode: generate ticket, wait for desktop-app to connect ─────
     let stream_name = ticket::generate_stream_name();
-    let ticket = NestriTicket::new(endpoint_addr, stream_name);
+    // For the log line below only. What a reader of the socket gets is built
+    // per read from the endpoint itself, because the addresses this can be
+    // reached at are not all known yet.
+    let ticket = NestriTicket::new(endpoint_addr, stream_name.clone());
 
     tracing::info!("╔═══════════════╗");
     tracing::info!("║ NESTRI TICKET ║");
@@ -245,7 +248,12 @@ async fn main() -> Result<()> {
 
     let ticket_ipc = args.ticket_ipc.clone();
     tokio::spawn({
-        async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, ticket).await }
+        // The endpoint rather than a ticket made from it: the addresses it can
+        // be reached at are not all known yet, and whoever reads this socket
+        // re-reads it so that a better one can replace the first.
+        let endpoint = endpoint.clone();
+        let stream_name = stream_name.clone();
+        async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, endpoint, stream_name).await }
     });
 
     // Accept loop
diff --git a/apps/nesinit/src/filesystems.rs b/apps/nesinit/src/filesystems.rs
new file mode 100644
index 00000000..04d03d5e
--- /dev/null
+++ b/apps/nesinit/src/filesystems.rs
@@ -0,0 +1,276 @@
+// The filesystems PID 1 has to establish before anything asks for them.
+//
+// The root arrives read-only — the host attaches it that way and nothing in
+// the guest may write to it — and there is no init system behind this process
+// to make up the difference. So a guest gets `/proc`, and it gets somewhere to
+// put a socket, only if this mounts them.
+//
+// Without that the failure is not an error anyone sees. Everything that needs
+// a writable path fails one layer down, separately, as `EROFS` on a socket:
+// the payload relay never binds, whatever serves the session's address never
+// binds either, and the session is reported as a workload that ran and
+// published nothing. Three unrelated-looking symptoms, one missing mount.
+
+use std::ffi::CString;
+use std::path::Path;
+
+/// A filesystem this process mounts, and why it has to exist.
+struct Early {
+    /// What appears in `/proc/mounts` as the source. Conventionally the type.
+    source: &'static str,
+    target: &'static str,
+    fstype: &'static str,
+    flags: libc::c_ulong,
+    /// Mount options, or empty for none.
+    data: &'static str,
+    /// Said when it could not be mounted, in terms of what stops working.
+    cost: &'static str,
+}
+
+/// `nosuid` and `nodev` on everything: none of these carry an image's files,
+/// so a device node or a setuid bit appearing in one did not come from us.
+const NOSUID_NODEV: libc::c_ulong = libc::MS_NOSUID | libc::MS_NODEV;
+
+const EARLY: &[Early] = &[
+    Early {
+        source: "proc",
+        target: "/proc",
+        fstype: "proc",
+        flags: NOSUID_NODEV | libc::MS_NOEXEC,
+        data: "",
+        cost: "this process cannot make itself ineligible for the OOM killer, \
+               and nothing in the guest can read its own state",
+    },
+    Early {
+        source: "sysfs",
+        target: "/sys",
+        fstype: "sysfs",
+        flags: NOSUID_NODEV | libc::MS_NOEXEC,
+        data: "",
+        cost: "a workload that looks up a device finds nothing",
+    },
+    // Writable, and the reason any of this is here. Both sockets in this
+    // component live on a tmpfs because the root they would otherwise sit on
+    // is read-only.
+    Early {
+        source: "tmpfs",
+        target: "/tmp",
+        fstype: "tmpfs",
+        flags: NOSUID_NODEV,
+        // The sticky bit, because the workload does not run as this process
+        // does and what it binds here is its own.
+        data: "mode=1777",
+        cost: "whatever serves this session's address cannot bind its socket, \
+               so the session never gets one",
+    },
+    // `/run` before anything under it, for the same reason `/proc` comes first:
+    // a directory cannot be created inside a mount that is not there, and the
+    // root it would otherwise land on is read-only.
+    Early {
+        source: "tmpfs",
+        target: "/run",
+        fstype: "tmpfs",
+        flags: NOSUID_NODEV,
+        // Octal, and without a leading zero on purpose: the kernel parses a
+        // tmpfs mode as octal either way, and this is the spelling `mount`
+        // itself documents.
+        data: "mode=755",
+        cost: "there is nowhere for a runtime socket to live, so neither the \
+               payload relay nor this session's address can be served",
+    },
+    // The relay's own directory, and it is deliberately **not** in the tree the
+    // session's shares live in.
+    //
+    // It was, and that was wrong in a way no test here would have caught: a
+    // fresh tmpfs over the share tree hides every directory the image prepared
+    // underneath it — the install, the user state, the work directory, and the
+    // mount point the log share is attached to from `fstab`. The box then has a
+    // socket and none of the places its workload expects to find its files, and
+    // the exact-path check below cannot notice, because what `fstab` mounts is
+    // a directory *inside* that tree rather than the tree itself.
+    //
+    // Owned by this process and writable by nothing else, which is what makes
+    // the socket in it unreplaceable. The workload reaches it because the
+    // directory is traversable and the socket itself is not restricted; see
+    // `payload::serve`.
+    Early {
+        source: "tmpfs",
+        target: crate::payload::DIRECTORY,
+        fstype: "tmpfs",
+        flags: NOSUID_NODEV | libc::MS_NOEXEC,
+        data: "mode=755",
+        cost: "the payload relay cannot bind, so nothing reaches the workload \
+               over the channel",
+    },
+];
+
+/// Mount what the rest of this component assumes is already there.
+///
+/// Best effort, one line per failure. Refusing to boot over any of these would
+/// replace a session that fails with a reason with a guest that never dialled
+/// out at all, and the second is strictly harder to diagnose from the host.
+pub fn establish() {
+    // `/proc` first and unconditionally: it is the only way to find out what is
+    // already mounted, so everything after it can be skipped when an image has
+    // done it already, and it cannot itself be checked that way.
+    mount(&EARLY[0]);
+
+    let existing = std::fs::read_to_string("/proc/self/mountinfo").unwrap_or_default();
+    for early in &EARLY[1..] {
+        if mounted_at(&existing, early.target) {
+            tracing::debug!(target = early.target, "already mounted by the image");
+            continue;
+        }
+        mount(early);
+    }
+}
+
+/// Whether `mountinfo` already has a mount at this exact path.
+///
+/// The mount point is the fifth field and it is the one that has to match:
+/// a prefix test would read `/tmpfoo` as `/tmp`, and a substring test would
+/// find the path in the options of something else entirely.
+fn mounted_at(mountinfo: &str, target: &str) -> bool {
+    mountinfo
+        .lines()
+        .filter_map(|line| line.split_whitespace().nth(4))
+        .any(|point| point == target)
+}
+
+fn mount(early: &Early) {
+    // A mount point that is not in the image cannot be created on a read-only
+    // root, so this is allowed to fail and the mount below reports it.
+    if !Path::new(early.target).exists() {
+        let _ = std::fs::create_dir_all(early.target);
+    }
+
+    let (Ok(source), Ok(target), Ok(fstype), Ok(data)) = (
+        CString::new(early.source),
+        CString::new(early.target),
+        CString::new(early.fstype),
+        CString::new(early.data),
+    ) else {
+        // Every one of these is a literal in this file, so this is
+        // unreachable rather than a case to handle.
+        tracing::error!(target = early.target, "a mount table entry has a nul byte");
+        return;
+    };
+    let data = if early.data.is_empty() {
+        std::ptr::null()
+    } else {
+        data.as_ptr().cast()
+    };
+
+    // SAFETY: four pointers that outlive the call, and a flag word.
+    let mounted = unsafe {
+        libc::mount(
+            source.as_ptr(),
+            target.as_ptr(),
+            fstype.as_ptr(),
+            early.flags,
+            data,
+        )
+    };
+    if mounted != 0 {
+        tracing::warn!(
+            target = early.target,
+            error = %std::io::Error::last_os_error(),
+            cost = early.cost,
+            "could not mount"
+        );
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The whole point of reading `mountinfo` is to not mount twice over
+    /// something an image already did.
+    #[test]
+    fn a_mount_point_that_is_present_is_recognised() {
+        let info = "\
+23 1 0:5 / /proc rw,nosuid,nodev,noexec - proc proc rw
+24 1 0:6 / /tmp rw,nosuid,nodev - tmpfs tmpfs rw,mode=1777";
+        assert!(mounted_at(info, "/proc"));
+        assert!(mounted_at(info, "/tmp"));
+    }
+
+    /// A prefix is not a mount point, and neither is a path that only appears
+    /// in another line's options.
+    #[test]
+    fn something_else_is_not_mistaken_for_a_mount_point() {
+        let info = "\
+23 1 0:5 / /tmpfoo rw - tmpfs tmpfs rw
+24 1 0:6 / /var rw - ext4 /dev/vda rw,journal_path=/nestri";
+        assert!(!mounted_at(info, "/tmp"));
+        assert!(!mounted_at(info, "/nestri"));
+    }
+
+    /// Every entry has to be nul-free, because `establish` treats a nul as
+    /// unreachable rather than handling it.
+    #[test]
+    fn the_mount_table_can_be_carried_out() {
+        for early in EARLY {
+            assert!(CString::new(early.source).is_ok(), "{}", early.target);
+            assert!(CString::new(early.target).is_ok(), "{}", early.target);
+            assert!(CString::new(early.fstype).is_ok(), "{}", early.target);
+            assert!(CString::new(early.data).is_ok(), "{}", early.target);
+            assert!(!early.cost.is_empty(), "{} has no cost", early.target);
+        }
+    }
+
+    /// `/proc` is mounted before `mountinfo` is read, so it has to be first.
+    #[test]
+    fn proc_is_the_first_entry() {
+        assert_eq!(EARLY[0].target, "/proc");
+    }
+
+    /// A mount has to come after whatever it lives inside, or it is a
+    /// directory created on a read-only root and the mount fails.
+    #[test]
+    fn nothing_is_mounted_before_the_mount_it_lives_inside() {
+        for (i, early) in EARLY.iter().enumerate() {
+            for other in &EARLY[i + 1..] {
+                assert!(
+                    !early.target.starts_with(&format!("{}/", other.target)),
+                    "{} is mounted before {}, which contains it",
+                    early.target,
+                    other.target
+                );
+            }
+        }
+    }
+
+    /// **Nothing here may be mounted over the tree the session's shares live
+    /// in.** A fresh tmpfs there hides every directory the image prepared
+    /// underneath — the install, the user state, the work directory, and the
+    /// mount point the log share attaches to — and the exact-path check cannot
+    /// notice, because what is mounted from `fstab` is a directory inside that
+    /// tree rather than the tree itself. So a box would come up with a socket
+    /// and without any of the places its workload looks for its files.
+    #[test]
+    fn the_share_tree_is_never_mounted_over() {
+        for early in EARLY {
+            assert_ne!(
+                early.target, "/nestri",
+                "this hides the directories the image prepared for a session"
+            );
+        }
+    }
+
+    /// The relay's directory is the one this cannot hardcode: it belongs to
+    /// `payload`, and a rename there that missed this file would take the
+    /// relay down again in exactly the way this exists to prevent.
+    #[test]
+    fn the_relay_directory_is_the_one_the_relay_uses() {
+        let entry = EARLY
+            .iter()
+            .find(|e| e.target == crate::payload::DIRECTORY)
+            .expect("the relay's directory is mounted");
+        assert!(
+            crate::payload::SOCKET.starts_with(entry.target),
+            "the relay's socket is not under the directory that is mounted for it"
+        );
+    }
+}
diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs
index 8ce6a6fc..32df7edd 100644
--- a/apps/nesinit/src/lib.rs
+++ b/apps/nesinit/src/lib.rs
@@ -8,8 +8,10 @@
 // and what an exit means, and it carries that out; a field that only makes
 // sense for one kind of workload cannot reach it. ref(d-0033)
 
+pub mod filesystems;
 pub mod payload;
 pub mod reap;
 pub mod session;
 pub mod shutdown;
+pub mod ticket;
 pub mod workload;
diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs
index 2952761f..cb44cbed 100644
--- a/apps/nesinit/src/main.rs
+++ b/apps/nesinit/src/main.rs
@@ -4,13 +4,14 @@
 // the workload the channel describes, and turn the end of either into an
 // ordered shutdown.
 
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use nesinit::payload::{self, Ports};
 use nesinit::reap::{self, Waiters};
 use nesinit::session::{self, Outcome};
 use nesinit::shutdown::{self, Machine};
+use nesinit::ticket;
 use nesinit::workload::{Process, Workload};
 use nesprotocol::lifecycle::CONTROL_PORT;
 use tokio::signal::unix::{SignalKind, signal};
@@ -25,6 +26,13 @@ const GRACE: Duration = Duration::from_secs(10);
 /// deep queue holds stale copies of it rather than protecting anything.
 const RELAY_DEPTH: usize = 8;
 
+/// How many addresses may be waiting to be forwarded.
+///
+/// Two, because only the newest one matters: an address is superseded by the
+/// next one rather than added to, so a deeper queue holds stale copies of it
+/// and delays the one that is current.
+const ADDRESS_DEPTH: usize = 2;
+
 fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt()
         .with_env_filter(
@@ -32,6 +40,11 @@ fn main() -> anyhow::Result<()> {
         )
         .init();
 
+    // Before everything, including the two below: the root is read-only and
+    // nothing else in this guest is an init system, so until this runs there
+    // is no `/proc` to score this process in and nowhere to put a socket.
+    nesinit::filesystems::establish();
+
     // Both before anything is started, so nothing can be orphaned or scored
     // in the window where neither is true yet.
     if let Err(error) = reap::become_subreaper() {
@@ -97,8 +110,22 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result<Outc
         from_workload: up_rx,
     };
 
+    // Started before the workload, like the relay, and for the same reason:
+    // whatever serves the address may bind the moment it comes up, and nothing
+    // here should be the reason a session waits to be reachable.
+    let (found_tx, mut found_rx) = tokio::sync::mpsc::channel(ADDRESS_DEPTH);
+    // Which user the carrier must not accept an address from. Empty until the
+    // descriptor names it, which is also when the workload that could abuse it
+    // is started -- so there is nothing to refuse before it is filled in.
+    let untrusted = ticket::Untrusted::unknown();
+    tokio::spawn(ticket::carry(
+        PathBuf::from(ticket::SOCKET),
+        found_tx,
+        untrusted.clone(),
+    ));
+
     let outcome = tokio::select! {
-        outcome = session::run(channel, workload, &mut ports) => outcome?,
+        outcome = session::run(channel, workload, &mut ports, &mut found_rx, &untrusted) => outcome?,
         signal = asked_to_stop() => {
             signal?;
             tracing::info!("asked to stop");
diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs
index 96b3a4ad..cbfebe00 100644
--- a/apps/nesinit/src/payload.rs
+++ b/apps/nesinit/src/payload.rs
@@ -9,6 +9,7 @@
 // anything above it.
 
 use std::io;
+use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 
 use nesprotocol::lifecycle::{Payload, from_line, to_line};
@@ -16,8 +17,24 @@ use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::sync::mpsc::{Receiver, Sender};
 
+/// The directory the relay's socket lives in.
+///
+/// Named separately because it is mounted before it is used: the guest's root
+/// is read-only, so this is a tmpfs that `filesystems` puts there, and a
+/// rename here that did not reach the mount table would take the relay down
+/// with an `EROFS` that looks like nothing to do with a path.
+///
+/// **Under `/run` rather than in the tree the session's shares live in**, and
+/// that was a correction. Putting a fresh tmpfs on the share tree hid every
+/// directory the image had prepared underneath it — the install, the user
+/// state, the work directory and the log share's own mount point — so a box
+/// gained a socket and lost the places its workload was supposed to find its
+/// files. `/run` is where a runtime socket belongs, it is a tmpfs already, and
+/// nothing else is mounted inside it.
+pub const DIRECTORY: &str = "/run/nestri";
+
 /// Where the workload finds the relay.
-pub const SOCKET: &str = "/nestri/payload.sock";
+pub const SOCKET: &str = "/run/nestri/payload.sock";
 
 /// The longest envelope this will assemble before giving up on the connection.
 ///
@@ -55,6 +72,17 @@ pub async fn serve(
     let _ = std::fs::remove_file(path);
     let listener = UnixListener::bind(path)?;
 
+    // The workload does not run as this process does, and it has to be able to
+    // connect. It reaches the socket through a directory this process owns and
+    // nothing else may write to, so the permission that matters is on the
+    // socket rather than on the path: the directory is what stops anybody
+    // replacing this listener, and this is what lets the workload talk to it.
+    //
+    // Without it the workload gets `EACCES` on connect and the payload layer
+    // is dead in both directions, silently, because nothing in the guest is
+    // waiting to be told about a relay it cannot reach.
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))?;
+
     loop {
         let stream = tokio::select! {
             accepted = listener.accept() => accepted?.0,
diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs
index 78e0e28d..a94e0b33 100644
--- a/apps/nesinit/src/session.rs
+++ b/apps/nesinit/src/session.rs
@@ -14,6 +14,7 @@ use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader
 
 use crate::payload::Ports;
 use crate::workload::{Exited, Failure, Workload};
+use tokio::sync::mpsc::Receiver;
 
 /// How a session ended.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -39,12 +40,14 @@ pub async fn run<C, W>(
     channel: C,
     workload: &mut W,
     payload: &mut Ports,
+    addresses: &mut Receiver<String>,
+    untrusted: &crate::ticket::Untrusted,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
     W: Workload,
 {
-    match converse(channel, workload, payload).await {
+    match converse(channel, workload, payload, addresses, untrusted).await {
         Err(error) if channel_gone(&error) => {
             // A caller that has stopped reading has also stopped being able to
             // tell us to stop, which is the same situation as the channel
@@ -70,6 +73,8 @@ async fn converse<C, W>(
     channel: C,
     workload: &mut W,
     payload: &mut Ports,
+    addresses: &mut Receiver<String>,
+    untrusted: &crate::ticket::Untrusted,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
@@ -91,6 +96,7 @@ where
 
     let mut running: Option<Exited> = None;
     let mut relay_open = true;
+    let mut carrier_open = true;
 
     loop {
         let event = match running.as_mut() {
@@ -98,10 +104,12 @@ where
                 ended = exited => Event::Ended(ended?),
                 line = lines.next_line() => Event::Line(line?),
                 up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up),
+                found = addresses.recv(), if carrier_open => Event::Address(found),
             },
             None => tokio::select! {
                 line = lines.next_line() => Event::Line(line?),
                 up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up),
+                found = addresses.recv(), if carrier_open => Event::Address(found),
             },
         };
 
@@ -115,6 +123,20 @@ where
                 send(&mut writer, &GuestToHost::Payload { payload }).await?;
                 continue;
             }
+            Event::Address(Some(ticket)) => {
+                // Sent whenever a better one is found, not only the first time:
+                // a caller that keeps the first address it is given works on a
+                // local network and fails from anywhere else.
+                send(&mut writer, &GuestToHost::Ticket { ticket }).await?;
+                continue;
+            }
+            Event::Address(None) => {
+                // Nothing will look for an address again. Not an ending: the
+                // session has whatever it was already told, and the workload
+                // still has to be stopped and its exit reported.
+                carrier_open = false;
+                continue;
+            }
             Event::FromWorkload(None) => {
                 // The relay is gone. The session is not: the workload can
                 // still be stopped, and its exit still has to be reported.
@@ -166,6 +188,11 @@ where
                     }
                 }
 
+                // Before the workload exists, so there is no window in which
+                // it is running and something else would still be trusted to
+                // serve this session's address.
+                untrusted.is(descriptor.exec.uid);
+
                 match workload.start(&descriptor.exec) {
                     Ok(exited) => {
                         send(&mut writer, &GuestToHost::Started).await?;
@@ -195,6 +222,7 @@ enum Event {
     Line(Option<String>),
     Ended(Exit),
     FromWorkload(Option<Payload>),
+    Address(Option<String>),
 }
 
 /// Hand an envelope to the relay, and treat a relay that is not there as the
@@ -276,6 +304,17 @@ mod tests {
         )
     }
 
+    /// A carrier that never finds an address, for the tests that are not
+    /// about one. Held open rather than closed: a closed channel is itself a
+    /// case, and it is tested on purpose below.
+    fn nowhere() -> mpsc::Receiver<String> {
+        let (tx, rx) = mpsc::channel(1);
+        // Kept alive for the process, so `recv` pends rather than resolving
+        // `None` and taking a branch these tests are not exercising.
+        Box::leak(Box::new(tx));
+        rx
+    }
+
     /// The other end of the channel, as a caller would drive it.
     struct Caller {
         lines: tokio::io::Lines<BufReader<DuplexStream>>,
@@ -322,7 +361,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -339,6 +386,99 @@ mod tests {
         assert_eq!(outcome, Outcome::Shutdown);
     }
 
+    /// The address goes up the channel as it is found. This is the only way
+    /// out: standard output here is a log file inside a VM and the person who
+    /// needs the address is outside it.
+    #[tokio::test]
+    async fn an_address_that_is_found_is_reported_to_the_caller() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (found_tx, mut found_rx) = mpsc::channel(4);
+
+        let session = tokio::spawn(async move {
+            let (mut ports, _to_workload, _from_workload) = ports();
+            let mut workload = Double::exits_when_stopped(Exit::code(0));
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut found_rx,
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
+        });
+
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ready {
+                protocol_version: 2
+            }
+        );
+
+        found_tx
+            .send("nestri:local-only".to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ticket {
+                ticket: "nestri:local-only".into()
+            }
+        );
+
+        // And a better one replaces it rather than being the caller's problem
+        // to have missed.
+        found_tx
+            .send("nestri:with-relays".to_string())
+            .await
+            .unwrap();
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ticket {
+                ticket: "nestri:with-relays".into()
+            }
+        );
+
+        caller.say(&HostToGuest::Shutdown).await;
+        assert_eq!(session.await.unwrap(), Outcome::Shutdown);
+    }
+
+    /// Nothing looking for an address any more is not an ending. The session
+    /// keeps whatever it was already told and still has to report an exit.
+    #[tokio::test]
+    async fn a_carrier_that_stops_does_not_end_the_session() {
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let (found_tx, mut found_rx) = mpsc::channel(4);
+
+        let session = tokio::spawn(async move {
+            let (mut ports, _to_workload, _from_workload) = ports();
+            let mut workload = Double::exits_when_stopped(Exit::code(0));
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut found_rx,
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
+        });
+
+        assert_eq!(
+            caller.expect().await,
+            GuestToHost::Ready {
+                protocol_version: 2
+            }
+        );
+        drop(found_tx);
+
+        // Still answering, which is the whole assertion.
+        caller.say(&HostToGuest::Shutdown).await;
+        assert_eq!(session.await.unwrap(), Outcome::Shutdown);
+    }
+
     #[tokio::test]
     async fn the_descriptor_mounts_and_starts_what_it_names() {
         let (guest, host) = tokio::io::duplex(4096);
@@ -347,7 +487,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -373,7 +521,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(3));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -409,7 +565,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::signal(9));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -440,7 +604,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -466,7 +638,15 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.mount_failure = Some(Failure::new("EACCES: /mnt/user"));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -501,7 +681,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -524,7 +712,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -556,7 +752,15 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.start_failure = Some(Failure::new("ENOENT: /usr/bin/workload"));
-            let outcome = run(guest, &mut workload, &mut ports).await.unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -598,7 +802,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
         let mut to_relay = down_rx;
 
@@ -647,7 +859,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -693,7 +913,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports).await.unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs
new file mode 100644
index 00000000..82eebf18
--- /dev/null
+++ b/apps/nesinit/src/ticket.rs
@@ -0,0 +1,455 @@
+// Carrying the address a client needs from inside the guest to outside it.
+//
+// Whatever serves media in the guest knows how it can be reached, and the
+// person who needs to know is not in the guest. Standard output here is a log
+// file inside a VM, so the control channel is the delivery path rather than a
+// convenience — which is why this is init's job and not a detail of whichever
+// component happens to bind the port.
+//
+// # Why it is polled rather than read once
+//
+// An address is not a value, it is the best answer so far. An endpoint
+// discovers more ways to reach it after it binds — a local one immediately, a
+// relayed or hole-punched one some seconds later — so the first answer is the
+// one that works on a local network and fails from anywhere else. Re-reading
+// and forwarding only what changed keeps that from being decided by whoever
+// asked first.
+//
+// # Why it dials rather than listens
+//
+// The opposite of the payload relay next door, and deliberately: there the
+// guest listens because the workload starts later, and here the server is the
+// long-lived one. Dialling also means a server that has not bound yet is an
+// error this retries, rather than a connection that has to be waited for
+// without knowing whether it is coming.
+//
+// # Who is allowed to answer
+//
+// Dialling a path means trusting whoever is behind it, and an address is the
+// capability to reach this session — so the wrong answer here does not break a
+// session, it hands one to somebody else. The workload runs arbitrary code, and
+// on a writable directory it can unlink whatever bound the socket and bind a
+// replacement; every read after that returns an address of its choosing, and
+// the client outside connects there instead.
+//
+// So the peer's credentials are checked and an answer from the workload's own
+// user is refused. That check is only as good as the workload having a user of
+// its own: run it as the same user as the process serving the address and
+// nothing can tell the two apart, which is [`Peer::indistinguishable`] and is
+// said out loud at boot rather than discovered later.
+
+use std::io;
+use std::os::fd::AsRawFd;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
+use tokio::net::UnixStream;
+use tokio::sync::mpsc::Sender;
+
+/// Where the address is read from.
+pub const SOCKET: &str = "/tmp/nestri-ticket.sock";
+
+/// How often to look for a better address.
+///
+/// Short, because the window this closes is the first few seconds of a session
+/// and a player waiting to connect is waiting on exactly this. It costs one
+/// connection to a unix socket per interval.
+const EVERY: Duration = Duration::from_secs(2);
+
+/// How long one look is given before it is abandoned.
+///
+/// A cap on time, next to the cap on size below and for the same reason: the
+/// far end can accept a connection and then write nothing at all, and a read
+/// with no deadline turns that into a poll loop that never runs again. The
+/// address it already forwarded stays correct; the better one that arrives
+/// later never would.
+const PATIENCE: Duration = Duration::from_secs(5);
+
+/// The longest address this will read.
+///
+/// One line of text. A cap rather than a preference: the process on the other
+/// end can write without ever sending a newline, and the process holding the
+/// buffer is the one the kernel has been told not to kill.
+const LONGEST: u64 = 8 * 1024;
+
+/// Who may not serve this session's address.
+///
+/// The workload's user, and nothing else is excluded — this is not an allow
+/// list of trusted uids, because init does not know which user an image happens
+/// to run its media components as, and inventing one here would be a second
+/// place for that to be configured wrongly.
+///
+/// Shared and filled in later, because the carrier starts before the descriptor
+/// that names the user arrives. That leaves no gap: a workload cannot serve
+/// anything before it is started, and it is started from the same descriptor,
+/// which sets this first.
+#[derive(Clone, Debug, Default)]
+pub struct Untrusted(std::sync::Arc<std::sync::atomic::AtomicU32>);
+
+/// No workload has been started, so there is no untrusted user yet.
+const NOBODY: u32 = u32::MAX;
+
+impl Untrusted {
+    pub fn unknown() -> Self {
+        Self(std::sync::Arc::new(std::sync::atomic::AtomicU32::new(
+            NOBODY,
+        )))
+    }
+
+    /// Name the user the workload runs as. Called before it is started.
+    pub fn is(&self, uid: u32) {
+        self.0.store(uid, std::sync::atomic::Ordering::Release);
+        if uid == 0 {
+            tracing::warn!(
+                "the workload runs as root, so an address it serves cannot be \
+                 told apart from a real one. Give it a user of its own."
+            );
+        }
+    }
+
+    /// The uid to refuse, or `None` when refusing anything would be wrong.
+    ///
+    /// `None` covers two cases that want the same answer for different reasons:
+    /// nothing has been started yet, so no peer can be the workload; and the
+    /// workload runs as `root`, which is every user at once — refusing root
+    /// would refuse whatever legitimately serves the address as well. The uid
+    /// being shared is the thing an operator has to fix, and `is` says so.
+    fn refuse(&self) -> Option<u32> {
+        match self.0.load(std::sync::atomic::Ordering::Acquire) {
+            NOBODY | 0 => None,
+            uid => Some(uid),
+        }
+    }
+}
+
+/// Forward every new address for as long as the session lasts.
+///
+/// Never returns on its own. A server that is not there yet, or has gone away,
+/// is retried at the next interval — there is nothing here worth ending a
+/// running session over, and an address that stops being re-offered does not
+/// stop being correct.
+pub async fn carry(path: PathBuf, out: Sender<String>, untrusted: Untrusted) {
+    let mut sent: Option<String> = None;
+    loop {
+        match look(&path, &untrusted).await {
+            Ok(current) if Some(&current) != sent.as_ref() => {
+                // The address itself is not logged. It is a capability to reach
+                // this session, and a log inside the guest is the one place it
+                // has no reason to be.
+                tracing::info!(
+                    first = sent.is_none(),
+                    "forwarding an address for this session"
+                );
+                if out.send(current.clone()).await.is_err() {
+                    // The session is over. Nothing else reads this.
+                    return;
+                }
+                sent = Some(current);
+            }
+            Ok(_) => {}
+            // Not the same as no address yet, and it must not be logged as
+            // though it were: this says something *is* serving an address and
+            // it is the one thing that may not. A session with no address at
+            // all is a legible failure; a session pointed somewhere else is
+            // not, so this is the line that has to be found afterwards.
+            Err(error) if error.kind() == io::ErrorKind::PermissionDenied => {
+                tracing::error!(%error, "refusing an address for this session");
+            }
+            Err(error) => {
+                // Expected until whatever serves the address has bound, so it
+                // is not a warning the first several times. It stays at this
+                // level afterwards too: a session that already has an address
+                // is not harmed by failing to look for a better one.
+                tracing::debug!(%error, "no address available yet");
+            }
+        }
+        tokio::time::sleep(EVERY).await;
+    }
+}
+
+/// One look at the socket, abandoned if it takes longer than [`PATIENCE`].
+async fn look(path: &Path, untrusted: &Untrusted) -> io::Result<String> {
+    match tokio::time::timeout(PATIENCE, read(path, untrusted)).await {
+        Ok(result) => result,
+        Err(_) => Err(io::Error::new(
+            io::ErrorKind::TimedOut,
+            "the socket accepted a connection and did not answer",
+        )),
+    }
+}
+
+/// One line from the socket, which is the whole protocol.
+async fn read(path: &Path, untrusted: &Untrusted) -> io::Result<String> {
+    let stream = UnixStream::connect(path).await?;
+
+    // Before a byte is read. The kernel answers this about the socket's peer
+    // rather than about the path, so it cannot be spoofed by whoever holds the
+    // path — which is the whole reason the check is worth anything.
+    if let Some(refuse) = untrusted.refuse()
+        && peer_uid(&stream)? == refuse
+    {
+        return Err(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            "the workload is serving this session's address, which would point \
+             a client at whatever it chose",
+        ));
+    }
+
+    let mut line = String::new();
+    BufReader::new(stream.take(LONGEST))
+        .read_line(&mut line)
+        .await?;
+    let line = line.trim().to_string();
+    if line.is_empty() {
+        return Err(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "the socket answered with nothing",
+        ));
+    }
+    Ok(line)
+}
+
+/// The uid of the process on the other end of a connected unix socket.
+///
+/// From the kernel, at connect time, and not from anything the peer says about
+/// itself. `SO_PEERCRED` records who held the other end when it connected, so a
+/// process cannot claim a uid it does not have.
+fn peer_uid(stream: &UnixStream) -> io::Result<u32> {
+    let mut credentials = libc::ucred {
+        pid: 0,
+        uid: u32::MAX,
+        gid: u32::MAX,
+    };
+    let mut length = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
+    // SAFETY: a connected socket this function borrows, and an out-parameter of
+    // exactly the length being passed.
+    let rc = unsafe {
+        libc::getsockopt(
+            stream.as_raw_fd(),
+            libc::SOL_SOCKET,
+            libc::SO_PEERCRED,
+            (&raw mut credentials).cast::<libc::c_void>(),
+            &raw mut length,
+        )
+    };
+    if rc != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(credentials.uid)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::AsyncWriteExt;
+    use tokio::net::UnixListener;
+    use tokio::sync::mpsc;
+
+    /// Serve a sequence of addresses, one per connection, the way a real one
+    /// does: it answers every dial with what it currently knows.
+    fn serve(path: PathBuf, answers: Vec<Option<String>>) {
+        let listener = UnixListener::bind(&path).unwrap();
+        tokio::spawn(async move {
+            let mut answers = answers.into_iter().cycle();
+            loop {
+                let Ok((mut stream, _)) = listener.accept().await else {
+                    return;
+                };
+                if let Some(answer) = answers.next().flatten() {
+                    let _ = stream.write_all(format!("{answer}\n").as_bytes()).await;
+                }
+            }
+        });
+    }
+
+    fn scratch(name: &str) -> PathBuf {
+        let dir =
+            std::env::temp_dir().join(format!("nesinit-ticket-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        dir.join("ticket.sock")
+    }
+
+    #[tokio::test]
+    async fn an_address_reaches_the_channel() {
+        let path = scratch("one");
+        serve(path.clone(), vec![Some("nestri:abc".into())]);
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("no address arrived")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:abc");
+
+        // And it is not sent again. An address that has not changed is the same
+        // address, and re-sending it is a write per interval for nothing.
+        assert!(
+            tokio::time::timeout(EVERY * 3, rx.recv()).await.is_err(),
+            "an unchanged address was forwarded again"
+        );
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// The case that decides whether this works from anywhere but a local
+    /// network: a better address arrives after the first one has been sent.
+    #[tokio::test]
+    async fn a_better_address_replaces_the_one_before_it() {
+        let path = scratch("better");
+        serve(
+            path.clone(),
+            vec![
+                Some("nestri:local-only".into()),
+                Some("nestri:with-relays".into()),
+            ],
+        );
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
+
+        let mut seen = Vec::new();
+        while seen.len() < 2 {
+            let next = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+                .await
+                .expect("the second address never arrived")
+                .expect("the channel closed");
+            seen.push(next);
+        }
+        assert_eq!(seen, ["nestri:local-only", "nestri:with-relays"]);
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// The peer's user decides whether an address is trusted, and this process
+    /// is the peer in a test — so naming *it* as the workload is a real refusal
+    /// of a real connection, not a stubbed one.
+    ///
+    /// The attack this closes: the workload unlinks whatever bound the socket,
+    /// binds its own, and every read afterwards hands the client an address of
+    /// the workload's choosing.
+    #[tokio::test]
+    async fn an_address_served_by_the_workload_is_refused() {
+        let path = scratch("hostile");
+        serve(path.clone(), vec![Some("nestri:attacker".into())]);
+
+        // SAFETY: reading this process's own uid cannot fail.
+        let ours = unsafe { libc::getuid() };
+        let untrusted = Untrusted::unknown();
+        untrusted.is(ours);
+
+        let error = look(&path, &untrusted)
+            .await
+            .expect_err("an address from the workload's own user was accepted");
+        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
+
+        // And the same socket is read happily once the workload is somebody
+        // else, which is what shows the refusal is about the peer and not about
+        // the socket.
+        let elsewhere = Untrusted::unknown();
+        elsewhere.is(ours + 1);
+        assert_eq!(look(&path, &elsewhere).await.unwrap(), "nestri:attacker");
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// Before a workload is started there is nothing to refuse, and refusing
+    /// anything then would mean no session ever got an address.
+    #[tokio::test]
+    async fn nothing_is_refused_before_a_workload_exists() {
+        let path = scratch("early");
+        serve(path.clone(), vec![Some("nestri:real".into())]);
+        let untrusted = Untrusted::unknown();
+        assert_eq!(untrusted.refuse(), None);
+        assert_eq!(look(&path, &untrusted).await.unwrap(), "nestri:real");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// A workload running as root is every user at once, so refusing root would
+    /// refuse whatever legitimately serves the address too. The check stands
+    /// down and `is` warns instead — the uid being shared is the operator's to
+    /// fix and this is not the place to fail closed over it.
+    #[tokio::test]
+    async fn a_root_workload_leaves_nothing_to_tell_apart() {
+        let untrusted = Untrusted::unknown();
+        untrusted.is(0);
+        assert_eq!(untrusted.refuse(), None);
+    }
+
+    /// Nothing serving the socket yet is the ordinary case at boot, not a
+    /// failure: this starts before whatever binds it.
+    #[tokio::test]
+    async fn a_socket_that_is_not_there_yet_is_waited_out_rather_than_failed() {
+        let path = scratch("late");
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
+
+        tokio::time::sleep(EVERY * 2).await;
+        serve(path.clone(), vec![Some("nestri:late".into())]);
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("an address that arrived late was never picked up")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:late");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// A server that accepts and then says nothing must not take the poll loop
+    /// with it. Without a deadline on the read this hangs forever, and the
+    /// address that arrives afterwards is never seen.
+    #[tokio::test]
+    async fn a_server_that_answers_nothing_does_not_stop_the_search() {
+        let path = scratch("mute");
+        let listener = UnixListener::bind(&path).unwrap();
+        let held = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new()));
+        {
+            let held = held.clone();
+            tokio::spawn(async move {
+                // Accepted and kept open, deliberately unanswered, which is
+                // what a wedged producer looks like from here.
+                let mut answered = false;
+                loop {
+                    let Ok((mut stream, _)) = listener.accept().await else {
+                        return;
+                    };
+                    if answered {
+                        let _ = stream.write_all(b"nestri:eventually\n").await;
+                    } else {
+                        answered = true;
+                        held.lock().await.push(stream);
+                    }
+                }
+            });
+        }
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
+
+        let first = tokio::time::timeout(PATIENCE + EVERY * 4, rx.recv())
+            .await
+            .expect("the carrier never got past a server that would not answer")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:eventually");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// An answer with nothing in it is not an address. Forwarding one would
+    /// publish an empty string as somewhere to connect.
+    #[tokio::test]
+    async fn an_empty_answer_is_not_an_address() {
+        let path = scratch("empty");
+        serve(path.clone(), vec![None, Some("nestri:real".into())]);
+
+        let (tx, mut rx) = mpsc::channel(4);
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
+
+        let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("no address arrived")
+            .expect("the channel closed");
+        assert_eq!(first, "nestri:real");
+        let _ = std::fs::remove_file(&path);
+    }
+}
diff --git a/build/Dockerfile b/build/Dockerfile
index 29abc082..0c9be9e3 100644
--- a/build/Dockerfile
+++ b/build/Dockerfile
@@ -145,9 +145,23 @@ RUN pacman -Syu --noconfirm --needed \
 
 # groupadd -f so this is idempotent whether or not udev's rules already
 # created these.
+#
+# **Two users, and they must stay two.** `nestri` runs the services that come
+# with this image; `nesplay` is who a workload runs as. Sharing one user between
+# them is what lets workload code impersonate a service — it can replace the
+# socket a service listens on and answer in its place, and the answer that
+# matters is the address a client is told to connect to. Init refuses an address
+# served by the workload's own user, so a single shared user does not merely
+# weaken that check, it makes every session fail it.
+#
+# The uid a workload actually runs as is chosen by whoever asks for the box, not
+# here; this account exists so that the number has a home, a shell and a name in
+# `ps`, and so the separation has somewhere to be written down.
 RUN groupadd -f audio && groupadd -f video && groupadd -f input && groupadd -f render && \
     useradd -m -u 1000 -s /bin/bash nestri && \
-    for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done
+    for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done && \
+    useradd -m -u 1001 -s /bin/bash nesplay && \
+    for g in audio video input render; do gpasswd -a nesplay "$g" >/dev/null; done
 
 
 # ───────────────────────────────────────────────────────────