feat(neshub): open the media hub

The component nescapture, neswire and nescope all talk to, and the only
thing in the guest that speaks to the client. It muxes their frames into
one iroh QUIC endpoint and fans input back.

Renamed from nestri-guest-hub, which named a location rather than a job.

Four files came across unchanged -- session.rs, ipc_listener.rs,
ticket.rs, screenshot.rs. Between them they mention Steam zero times, and
they import only nesprotocol's open modules; the control feature carrying
LaunchIntent and SteamIdentity is used exclusively by the three files that
are staying closed. The two clusters shared a main.rs and nothing else, so
there was no untangling to do -- only a cut.

main.rs loses --proton, --steamclient-so, --root and the game uid/gid,
and no longer ends by handing the process to a controller. It runs until
it is stopped. Deciding when the box is finished belongs to nesinit.

The ticket used to leave via that controller, so it needed a new way out:
neshub now serves it on a socket and nesinit dials for it. Listening
rather than dialling matches every other socket here and means no startup
ordering to get wrong.

Three tests, where there were none -- the ticket crosses a process
boundary as text now, so a round trip that drops a field would otherwise
be found by whoever cannot connect.

Cargo.toml

@@ -15,6 +15,7 @@ resolver = "3"
 members = [
   "apps/nescapture",
   "apps/nescope",
+  "apps/neshub",
   "apps/neswire",
   "crates/nesprotocol",
 ]

apps/neshub/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "neshub"
+version = "0.1.0"
+description = "Muxes captured video, audio and input into one QUIC connection to the client"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[[bin]]
+name = "neshub"
+path = "src/main.rs"
+
+[dependencies]
+# The transport. One endpoint per box; clients dial it with a ticket.
+iroh = "1.0.0"
+
+anyhow.workspace = true
+clap.workspace = true
+libc.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+
+base64 = "0.22"
+uuid = { version = "1", features = ["v7"] }
+
+nesprotocol = { path = "../../crates/nesprotocol" }

apps/neshub/README.md

@@ -0,0 +1,50 @@
+## neshub
+
+One connection out of the box.
+
+Everything inside the guest that produces or consumes a stream talks to neshub
+over a Unix socket. neshub muxes it all into a single [iroh](https://www.iroh.computer)
+QUIC endpoint and fans client input back the other way. A client dials that
+endpoint with a *ticket* and gets video, audio, cursor and stats on it.
+
+### The sockets
+
+| socket | default | direction | carries |
+| --- | --- | --- | --- |
+| `--video-ipc` | `/tmp/nestri-video.sock` | nescapture → neshub | encoded video frames |
+| `--audio-ipc` | `/tmp/nestri-audio.sock` | neswire → neshub | Opus packets |
+| `--input-ipc` | `/tmp/nestri-input.sock` | neshub ↔ nescope | input out, cursor and stats back |
+| `--stats-ipc` | `/tmp/nestri-stats.sock` | nescapture → neshub | encoder stats |
+| `--screenshot-ipc` | `/tmp/nestri-screenshot.sock` | neshub → nescope | a picture of the screen, on request |
+| `--ticket-ipc` | `/tmp/nestri-ticket.sock` | neshub → nesinit | the ticket, once |
+| — | `/tmp/nescapture-cmd.sock` | neshub → nescapture | IDR requests, encode settings |
+
+neshub is the listener on every one of them and the other side dials in. That
+is deliberate: it removes the startup ordering problem entirely, since a
+producer that is not running yet simply has not connected yet.
+
+### The ticket
+
+```
+nestri:<base64 of {endpoint_addr, stream_name}>
+```
+
+Generated once per boot, when the endpoint binds. It is served on a socket
+rather than printed because stdout here is a log file inside a virtual machine
+and the person who needs it is outside one — `nesinit` reads it and carries it
+to the host.
+
+### What it does not do
+
+neshub does not start the payload, know its name, or decide when the box is
+finished — `nesinit` owns all three, and shuts the VM down around this process.
+The same neshub binary serves a game, a desktop, or anything else that draws to
+`nescope`, because it never learns which one it is looking at.
+
+### Running it
+
+```bash
+cargo run --bin neshub                    # every socket at its default
+cargo run --bin neshub -- --relay none    # direct connections only
+RUST_LOG=neshub=debug cargo run --bin neshub
+```

apps/neshub/src/ipc_listener.rs

@@ -0,0 +1,293 @@
+use std::os::unix::io::AsRawFd;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{UnixDatagram, UnixListener};
+use tokio::sync::broadcast::error::RecvError;
+use tracing::{debug, error, info, warn};
+
+use crate::session::SessionManager;
+use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, STREAM_VIDEO, decode_ipc_frame};
+
+fn set_recv_buffer(socket: &UnixDatagram, size: libc::c_int) {
+    unsafe {
+        libc::setsockopt(
+            socket.as_raw_fd(),
+            libc::SOL_SOCKET,
+            libc::SO_RCVBUF,
+            &size as *const _ as *const libc::c_void,
+            std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+        );
+    }
+}
+
+pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc<SessionManager>) {
+    if socket_path.exists() {
+        let _ = std::fs::remove_file(&socket_path);
+    }
+
+    let socket = match UnixDatagram::bind(&socket_path) {
+        Ok(s) => {
+            let _ = std::fs::set_permissions(
+                &socket_path,
+                std::os::unix::fs::PermissionsExt::from_mode(0o666),
+            );
+            set_recv_buffer(&s, 4 * 1024 * 1024);
+            s
+        }
+        Err(e) => {
+            error!(
+                "Failed to bind video IPC socket {}: {e}",
+                socket_path.display()
+            );
+            return;
+        }
+    };
+
+    info!("Video IPC listening on {}", socket_path.display());
+
+    let mut buf = vec![0u8; 2 * 1024 * 1024];
+    loop {
+        match socket.recv(&mut buf).await {
+            Ok(n) => {
+                if let Some(decoded) = decode_ipc_frame(&buf[..n]) {
+                    if decoded.stream_type != STREAM_VIDEO {
+                        warn!(
+                            "unexpected stream type on video socket: {}",
+                            decoded.stream_type
+                        );
+                        continue;
+                    }
+                    let mut frame_data = Vec::with_capacity(1 + decoded.data.len());
+                    frame_data.push(decoded.codec);
+                    frame_data.push(decoded.flags);
+                    frame_data.extend_from_slice(&decoded.timestamp_ms.to_le_bytes());
+                    frame_data.extend_from_slice(&decoded.width.to_le_bytes());
+                    frame_data.extend_from_slice(&decoded.height.to_le_bytes());
+                    frame_data.extend_from_slice(decoded.data);
+
+                    session_manager.broadcast_video(frame_data).await;
+                } else {
+                    warn!("invalid video IPC frame ({} bytes)", n);
+                }
+            }
+            Err(e) => {
+                error!("video IPC recv error: {e}");
+                break;
+            }
+        }
+    }
+
+    info!("video IPC listener exited");
+}
+
+pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc<SessionManager>) {
+    if socket_path.exists() {
+        let _ = std::fs::remove_file(&socket_path);
+    }
+
+    let socket = match UnixDatagram::bind(&socket_path) {
+        Ok(s) => {
+            let _ = std::fs::set_permissions(
+                &socket_path,
+                std::os::unix::fs::PermissionsExt::from_mode(0o666),
+            );
+            set_recv_buffer(&s, 4 * 1024 * 1024);
+            s
+        }
+        Err(e) => {
+            error!(
+                "Failed to bind audio IPC socket {}: {e}",
+                socket_path.display()
+            );
+            return;
+        }
+    };
+
+    info!("Audio IPC listening on {}", socket_path.display());
+
+    let mut buf = vec![0u8; 65536];
+    loop {
+        match socket.recv(&mut buf).await {
+            Ok(n) => {
+                if let Some(decoded) = decode_ipc_frame(&buf[..n]) {
+                    if decoded.stream_type != STREAM_AUDIO {
+                        warn!(
+                            "unexpected stream type on audio socket: {}",
+                            decoded.stream_type
+                        );
+                        continue;
+                    }
+                    if decoded.codec != CODEC_OPUS {
+                        warn!("unexpected codec on audio socket: {}", decoded.codec);
+                        continue;
+                    }
+                    let audio_data = decoded.data.to_vec();
+                    session_manager.broadcast_audio(audio_data).await;
+                } else {
+                    warn!("invalid audio IPC frame ({} bytes)", n);
+                }
+            }
+            Err(e) => {
+                error!("audio IPC recv error: {e}");
+                break;
+            }
+        }
+    }
+
+    info!("audio IPC listener exited");
+}
+
+pub async fn run_input_ipc_listener(
+    socket_path: PathBuf,
+    input_tx: tokio::sync::broadcast::Sender<Vec<u8>>,
+    cursor_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    nescope_stats_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+) {
+    if socket_path.exists() {
+        let _ = std::fs::remove_file(&socket_path);
+    }
+
+    let listener = match UnixListener::bind(&socket_path) {
+        Ok(l) => {
+            let _ = std::fs::set_permissions(
+                &socket_path,
+                std::os::unix::fs::PermissionsExt::from_mode(0o666),
+            );
+            l
+        }
+        Err(e) => {
+            error!(
+                "Failed to bind input IPC socket {}: {e}",
+                socket_path.display()
+            );
+            return;
+        }
+    };
+
+    info!("Input IPC listening on {}", socket_path.display());
+
+    loop {
+        match listener.accept().await {
+            Ok((stream, peer_addr)) => {
+                debug!(?peer_addr, "nescope connected to input IPC");
+                if let Some(addr) = peer_addr.as_pathname() {
+                    debug!(?addr, "nescope input IPC peer address");
+                }
+
+                let (mut read_half, mut write_half) = stream.into_split();
+                let mut input_rx = input_tx.subscribe();
+                let _c_tx = cursor_tx.clone();
+
+                let write_handle = tokio::spawn(async move {
+                    loop {
+                        match input_rx.recv().await {
+                            Ok(data) => {
+                                let len = data.len() as u16;
+                                let mut frame = Vec::with_capacity(2 + data.len());
+                                frame.extend_from_slice(&len.to_le_bytes());
+                                frame.extend_from_slice(&data);
+                                if write_half.write_all(&frame).await.is_err() {
+                                    debug!("input IPC write failed");
+                                    break;
+                                }
+                            }
+                            Err(RecvError::Lagged(n)) => {
+                                warn!("input broadcast lagged by {n} messages");
+                            }
+                            Err(RecvError::Closed) => {
+                                debug!("input broadcast closed");
+                                break;
+                            }
+                        }
+                    }
+                });
+
+                let c_tx = cursor_tx.clone();
+                let ns_tx = nescope_stats_tx.clone();
+                let read_handle = tokio::spawn(async move {
+                    let mut len_buf = [0u8; 2];
+                    loop {
+                        if read_half.read_exact(&mut len_buf).await.is_err() {
+                            break;
+                        }
+                        let len = u16::from_le_bytes(len_buf) as usize;
+                        if len > 65535 {
+                            break;
+                        }
+                        let mut payload = vec![0u8; len];
+                        if read_half.read_exact(&mut payload).await.is_err() {
+                            break;
+                        }
+                        // Route: cursor updates start with 0x80, stats start with 0x00-0x02
+                        if !payload.is_empty()
+                            && (payload[0] == nesprotocol::input::CURSOR_UPDATE
+                                || payload[0] == nesprotocol::input::CURSOR_IMAGE)
+                        {
+                            let _ = c_tx.send(payload);
+                        } else {
+                            let _ = ns_tx.send(payload);
+                        }
+                    }
+                });
+
+                // When either task completes, the socket is broken.
+                // The other task will finish naturally on the next I/O error.
+                tokio::select! {
+                    _ = write_handle => {}
+                    _ = read_handle => {}
+                }
+
+                debug!("nescope disconnected from input IPC");
+            }
+            Err(e) => {
+                error!("input IPC accept error: {e}");
+                break;
+            }
+        }
+    }
+
+    info!("input IPC listener exited");
+}
+
+pub async fn run_stats_ipc_listener(
+    socket_path: PathBuf,
+    stats_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+) {
+    if socket_path.exists() {
+        let _ = std::fs::remove_file(&socket_path);
+    }
+
+    let socket = match UnixDatagram::bind(&socket_path) {
+        Ok(s) => {
+            let _ = std::fs::set_permissions(
+                &socket_path,
+                std::os::unix::fs::PermissionsExt::from_mode(0o666),
+            );
+            s
+        }
+        Err(e) => {
+            error!(
+                "Failed to bind stats IPC socket {}: {e}",
+                socket_path.display()
+            );
+            return;
+        }
+    };
+
+    info!("Stats IPC listening on {}", socket_path.display());
+
+    let mut buf = vec![0u8; 256];
+    loop {
+        match socket.recv(&mut buf).await {
+            Ok(n) => {
+                let _ = stats_tx.send(buf[..n].to_vec());
+            }
+            Err(e) => {
+                error!("stats IPC recv error: {e}");
+                break;
+            }
+        }
+    }
+    info!("stats IPC listener exited");
+}

apps/neshub/src/main.rs

@@ -0,0 +1,299 @@
+//! neshub — one connection out of the box.
+//!
+//! Four producers inside the guest send it frames over Unix sockets:
+//! nescapture (video, stats), neswire (audio), nescope (cursor). It muxes
+//! them into one iroh QUIC endpoint, and fans client input back the other
+//! way. That is the whole job.
+//!
+//! It does not know what is producing the pixels. The payload is started by
+//! nesinit and neshub never learns its name, which is what lets the same
+//! binary serve a game, a desktop, or something nobody here has thought of.
+
+mod ipc_listener;
+mod screenshot;
+mod session;
+mod ticket;
+
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use anyhow::Result;
+use clap::Parser;
+use iroh::endpoint::presets;
+use tracing::{info, warn};
+
+use crate::session::SessionManager;
+use crate::ticket::NestriTicket;
+use nesprotocol::ALPN;
+
+#[derive(Parser, Debug)]
+#[command(name = "neshub")]
+struct Args {
+    /// Relay mode: default, none, or a custom relay URL
+    #[arg(long, env = "NESTRI_RELAY", default_value = "default")]
+    relay: String,
+
+    /// Path for the video IPC socket (nescapture → neshub)
+    #[arg(
+        long,
+        env = "NESTRI_VIDEO_IPC",
+        default_value = "/tmp/nestri-video.sock"
+    )]
+    video_ipc: PathBuf,
+
+    /// Path for the audio IPC socket (neswire → neshub)
+    #[arg(
+        long,
+        env = "NESTRI_AUDIO_IPC",
+        default_value = "/tmp/nestri-audio.sock"
+    )]
+    audio_ipc: PathBuf,
+
+    /// Path for the input IPC socket (neshub → nescope).
+    #[arg(
+        long,
+        env = "NESTRI_INPUT_IPC",
+        default_value = "/tmp/nestri-input.sock"
+    )]
+    input_ipc: PathBuf,
+
+    /// Path for the stats IPC socket (nescapture → neshub stats).
+    #[arg(
+        long,
+        env = "NESTRI_STATS_IPC",
+        default_value = "/tmp/nestri-stats.sock"
+    )]
+    stats_ipc: PathBuf,
+
+    /// Socket the ticket is served on. neshub listens; nesinit dials and
+    /// carries the ticket to the host, because the person who needs it is
+    /// outside this VM and stdout here is a log file inside one.
+    #[arg(
+        long,
+        env = "NESTRI_TICKET_IPC",
+        default_value = "/tmp/nestri-ticket.sock"
+    )]
+    ticket_ipc: PathBuf,
+
+    /// Audio channels (from neswire config): 2 = stereo, 6 = 5.1, 8 = 7.1
+    #[arg(long, env = "NESTRI_AUDIO_CHANNELS", default_value_t = 2)]
+    audio_channels: u32,
+
+    /// Audio bitrate per channel in kbps
+    #[arg(long, env = "NESTRI_AUDIO_BITRATE", default_value_t = 64)]
+    audio_bitrate_per_channel: u32,
+
+    /// Socket nescope sends screenshots on. neshub listens; nescope dials out.
+    #[arg(
+        long,
+        env = "NESTRI_SCREENSHOT_IPC",
+        default_value = "/tmp/nestri-screenshot.sock"
+    )]
+    screenshot_ipc: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let args = Args::parse();
+
+    let mut builder = iroh::Endpoint::builder(presets::N0).alpns(vec![ALPN.to_vec()]);
+
+    match args.relay.as_str() {
+        "default" | "" => {
+            builder = builder.relay_mode(iroh::endpoint::RelayMode::Default);
+            info!("using default n0-computer relays");
+        }
+        "none" | "off" | "disabled" => {
+            builder = builder.relay_mode(iroh::endpoint::RelayMode::Disabled);
+            info!("relays disabled (direct connections only)");
+        }
+        url => {
+            let relay_url: iroh::RelayUrl = url.parse()?;
+            let relay_map = iroh::RelayMap::empty();
+            relay_map.insert(
+                relay_url.clone(),
+                Arc::new(iroh::RelayConfig::new(relay_url, None)),
+            );
+            builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map));
+            info!("using custom relay: {url}");
+        }
+    }
+
+    let endpoint = builder.bind().await?;
+    let endpoint_addr = endpoint.addr();
+    let ep_id = endpoint_addr.id;
+    info!("endpoint online: {}", ep_id.fmt_short());
+
+    // Input broadcast channel: input reader -> input IPC listener -> nescope
+    let (input_broadcast_tx, _) = tokio::sync::broadcast::channel::<Vec<u8>>(256);
+
+    // Cursor channel: IPC listener (read side) -> client sessions -> desktop-app
+    let (cursor_tx, mut cursor_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+
+    // Nescope stats channel: IPC listener -> client sessions
+    let (nescope_stats_tx, mut nescope_stats_rx) =
+        tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+
+    let session_manager = Arc::new(SessionManager::new());
+
+    // IDR / encode settings command channel: input reader → nescapture
+    let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+    {
+        tokio::spawn(async move {
+            let cmd_path = std::path::PathBuf::from("/tmp/nescapture-cmd.sock");
+            while let Some(bytes) = cmd_rx.recv().await {
+                if let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() {
+                    if sock.send_to(&bytes, &cmd_path).is_err() {
+                        warn!("nescapture cmd send failed at {}", cmd_path.display());
+                    }
+                }
+            }
+        });
+    }
+
+    // Spawn cursor relay
+    {
+        let mgr = session_manager.clone();
+        tokio::spawn(async move {
+            while let Some(data) = cursor_rx.recv().await {
+                mgr.broadcast_cursor(data).await;
+            }
+        });
+    }
+
+    // Spawn nescope stats relay
+    {
+        let mgr = session_manager.clone();
+        tokio::spawn(async move {
+            while let Some(data) = nescope_stats_rx.recv().await {
+                mgr.broadcast_stats(data).await;
+            }
+        });
+    }
+
+    // Spawn periodic hub stats
+    {
+        let mgr = session_manager.clone();
+        let audio_channels = args.audio_channels as u8;
+        let audio_kbps = args.audio_channels * args.audio_bitrate_per_channel;
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
+            loop {
+                interval.tick().await;
+                let clients = mgr.client_count().await as u8;
+                let bitrate = mgr.video_bitrate_bps();
+                let relay_ms = mgr.relay_ms();
+                let mut buf = Vec::with_capacity(15);
+                nesprotocol::stats::encode_hub_stats(
+                    &mut buf,
+                    clients,
+                    bitrate,
+                    relay_ms,
+                    audio_kbps,
+                    audio_channels,
+                );
+                mgr.broadcast_stats(buf).await;
+            }
+        });
+    }
+
+    // The ticket is how a client finds this endpoint. It is generated here
+    // because the endpoint is here, and served on a socket because the only
+    // thing that can carry it out of the VM is nesinit.
+    let stream_name = ticket::generate_stream_name();
+    let ticket = NestriTicket::new(endpoint_addr, stream_name);
+    info!(%ticket, "endpoint ticket generated");
+    ticket::serve(&args.ticket_ipc, ticket.to_string())?;
+
+    // Spawn IPC listeners
+    let video_ipc = args.video_ipc.clone();
+    let audio_ipc = args.audio_ipc.clone();
+    let input_ipc = args.input_ipc.clone();
+    let stats_ipc = args.stats_ipc.clone();
+    let stats_tx_clone = nescope_stats_tx.clone();
+    tokio::spawn({
+        let mgr = session_manager.clone();
+        async move { ipc_listener::run_video_listener(video_ipc, mgr).await }
+    });
+    tokio::spawn({
+        let mgr = session_manager.clone();
+        async move { ipc_listener::run_audio_listener(audio_ipc, mgr).await }
+    });
+    let input_ipc_tx = input_broadcast_tx.clone();
+    let cursor_ipc_tx = cursor_tx.clone();
+    let ns_tx = nescope_stats_tx.clone();
+    tokio::spawn({
+        async move {
+            ipc_listener::run_input_ipc_listener(input_ipc, input_ipc_tx, cursor_ipc_tx, ns_tx)
+                .await
+        }
+    });
+    tokio::spawn({
+        let stx = stats_tx_clone.clone();
+        async move { ipc_listener::run_stats_ipc_listener(stats_ipc, stx).await }
+    });
+
+    // Accept loop
+    let ep = endpoint.clone();
+    let mgr = session_manager.clone();
+    let accept_handle = tokio::spawn(async move {
+        while let Some(incoming) = ep.accept().await {
+            match incoming.await {
+                Ok(conn) => {
+                    let remote_id = conn.remote_id();
+                    info!(remote = %remote_id.fmt_short(), "client connected");
+                    let session = session::ClientSession::new(
+                        conn.clone(),
+                        input_broadcast_tx.clone(),
+                        session_manager.relay_ms_atomic(),
+                        cmd_tx.clone(),
+                    );
+                    mgr.add_session(remote_id, session).await;
+                    let mgr_clone = mgr.clone();
+                    let conn_clone = conn.clone();
+                    tokio::spawn(async move {
+                        conn_clone.closed().await;
+                        mgr_clone.remove_session(&remote_id).await;
+                    });
+                }
+                Err(e) => {
+                    warn!("incoming connection failed: {e}");
+                }
+            }
+        }
+        info!("accept loop exited");
+    });
+
+    // Not wired to anything today. Kept because the capture works and "show me
+    // what the guest is displaying" is the first question when a payload
+    // renders black.
+    let _screenshots = match screenshot::listen(&args.screenshot_ipc) {
+        Ok(connection) => Some(connection),
+        Err(e) => {
+            warn!("screenshots unavailable: {e:#}");
+            None
+        }
+    };
+
+    // neshub outlives every session and every payload. Nothing here decides
+    // when the box is done -- nesinit owns that, and shuts the VM down around
+    // this process.
+    info!("neshub running");
+    tokio::signal::ctrl_c().await?;
+
+    info!("shutting down..");
+    endpoint.close().await;
+    accept_handle.abort();
+
+    let _ = std::fs::remove_file(&args.video_ipc);
+    let _ = std::fs::remove_file(&args.audio_ipc);
+    let _ = std::fs::remove_file(&args.input_ipc);
+    let _ = std::fs::remove_file(&args.stats_ipc);
+    let _ = std::fs::remove_file(&args.ticket_ipc);
+    let _ = std::fs::remove_file("/tmp/nescapture-cmd.sock");
+    Ok(())
+}

apps/neshub/src/screenshot.rs

@@ -0,0 +1,105 @@
+//! Asking nescope what is on screen.
+//!
+//! neshub is the listener and nescope dials out, the same way the input socket
+//! works — so this waits for a connection rather than making one, and there is
+//! no race against a compositor that has not started yet.
+//!
+//! Nothing here interprets the pixels. It is transport only -- whatever asked
+//! for the picture decides what it means.
+//!
+//! **Nothing calls this today.** It was neshub's half of a login-QR capture,
+//! and that path is gone -- nessh's own token signs the client in, so no
+//! screenshot is needed to sign anybody in. It is kept because the capture
+//! works and "show me what the guest is displaying" is the first question when
+//! a payload renders black; `nescope-shot` covers the same ground from the
+//! other side. Wire it to a control message or delete it, but do not leave it
+//! half-connected.
+#![allow(dead_code)]
+
+use std::path::Path;
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::sync::Mutex;
+
+/// Ask for a picture of what is on screen.
+const REQUEST_CAPTURE: u8 = 0x01;
+
+/// Status bytes, as `nescope::screenshot_wire` defines them.
+const STATUS_OK: u8 = 0;
+
+/// The connection nescope made, once it has made one.
+///
+/// Held across captures rather than reconnecting: nescope connects once and
+/// serves every request on that connection, so dropping it would mean nothing
+/// could ask again.
+pub type Connection = Arc<Mutex<Option<UnixStream>>>;
+
+/// Listen for nescope, and keep the connection for whoever asks later.
+///
+/// Returns as soon as the socket is bound, not when nescope connects — the
+/// caller has other things to start, and a compositor that never dials is a
+/// failure the first capture reports rather than one that blocks startup.
+pub fn listen(path: &Path) -> Result<Connection> {
+    // A stale socket file makes bind fail with EADDRINUSE, which reads as
+    // "something is already listening" when nothing is.
+    let _ = std::fs::remove_file(path);
+    let listener = UnixListener::bind(path)
+        .with_context(|| format!("could not listen on {}", path.display()))?;
+    tracing::info!("listening for nescope screenshots on {}", path.display());
+
+    let connection: Connection = Arc::new(Mutex::new(None));
+    let slot = Arc::clone(&connection);
+    tokio::spawn(async move {
+        loop {
+            match listener.accept().await {
+                Ok((stream, _)) => {
+                    tracing::info!("nescope connected for screenshots");
+                    *slot.lock().await = Some(stream);
+                }
+                Err(e) => {
+                    tracing::warn!("accepting a screenshot connection failed: {e}");
+                    return;
+                }
+            }
+        }
+    });
+    Ok(connection)
+}
+
+/// One capture. `None` means nescope had nothing to show, which is not an
+/// error — a client that has not drawn yet is the normal case while waiting.
+pub async fn capture_on(connection: &Connection) -> Result<Option<(u32, u32, Vec<u8>)>> {
+    let mut guard = connection.lock().await;
+    let Some(stream) = guard.as_mut() else {
+        anyhow::bail!("nescope has not connected for screenshots yet");
+    };
+
+    stream
+        .write_all(&[REQUEST_CAPTURE])
+        .await
+        .context("could not ask nescope for a capture")?;
+
+    let mut header = [0u8; 9];
+    stream
+        .read_exact(&mut header)
+        .await
+        .context("nescope did not answer a capture request")?;
+
+    if header[0] != STATUS_OK {
+        // Every non-Ok status means "no pixels", and the distinctions between
+        // them are a debugging matter for `nescope-shot` rather than anything
+        // this can act on differently.
+        return Ok(None);
+    }
+    let width = u32::from_le_bytes(header[1..5].try_into().unwrap());
+    let height = u32::from_le_bytes(header[5..9].try_into().unwrap());
+    let mut rgba = vec![0u8; (width as usize) * (height as usize) * 4];
+    stream
+        .read_exact(&mut rgba)
+        .await
+        .context("a capture was cut short")?;
+    Ok(Some((width, height, rgba)))
+}

apps/neshub/src/session.rs

@@ -0,0 +1,567 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use iroh::endpoint::Connection;
+use tokio::sync::Mutex;
+use tracing::{debug, info, warn};
+
+use nesprotocol::input::{INPUT_KEY, INPUT_MOUSE_BUTTON, INPUT_MOUSE_MOVE, INPUT_MOUSE_WHEEL};
+use nesprotocol::{BIDI_INPUT, STREAM_AUDIO, STREAM_CURSOR, STREAM_STATS, STREAM_VIDEO};
+use nesprotocol::{FRAME_HDR_LEN, MSG_DATA, STREAM_VERSION, encode_frame};
+use nesprotocol::{MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, MSG_INPUT_BATCH};
+
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+
+pub struct ClientSession {
+    send_video: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    send_audio: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    send_cursor: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    send_stats: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    _video_task: tokio::task::JoinHandle<()>,
+    _audio_task: tokio::task::JoinHandle<()>,
+    _cursor_task: tokio::task::JoinHandle<()>,
+    _stats_task: tokio::task::JoinHandle<()>,
+    _input_task: tokio::task::JoinHandle<()>,
+}
+
+impl ClientSession {
+    pub fn new(
+        conn: Connection,
+        input_broadcast: tokio::sync::broadcast::Sender<Vec<u8>>,
+        relay_ms: Arc<AtomicU32>,
+        idr_cmd_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    ) -> Self {
+        let (video_tx, video_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        let (audio_tx, audio_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        let (cursor_tx, cursor_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+
+        let conn_v = conn.clone();
+        let _video_task =
+            tokio::spawn(async move { run_video_sender(conn_v, video_rx, relay_ms).await });
+
+        let conn_a = conn.clone();
+        let _audio_task = tokio::spawn(async move { run_audio_sender(conn_a, audio_rx).await });
+
+        let conn_c = conn.clone();
+        let _cursor_task = tokio::spawn(async move { run_cursor_sender(conn_c, cursor_rx).await });
+
+        let conn_s = conn.clone();
+        let _stats_task = tokio::spawn(async move { run_stats_sender(conn_s, stats_rx).await });
+
+        let conn_i = conn.clone();
+        let _input_task =
+            tokio::spawn(
+                async move { run_input_reader(conn_i, input_broadcast, idr_cmd_tx).await },
+            );
+
+        Self {
+            send_video: video_tx,
+            send_audio: audio_tx,
+            send_cursor: cursor_tx,
+            send_stats: stats_tx,
+            _video_task,
+            _audio_task,
+            _cursor_task,
+            _stats_task,
+            _input_task,
+        }
+    }
+
+    pub fn send_video_frame(&self, data: Vec<u8>) {
+        if let Err(e) = self.send_video.send(data) {
+            warn!("failed to send video data: {e}");
+        }
+    }
+
+    pub fn send_audio_packet(&self, data: Vec<u8>) {
+        if let Err(e) = self.send_audio.send(data) {
+            warn!("failed to send audio data: {e}");
+        }
+    }
+
+    pub fn send_cursor_data(&self, data: Vec<u8>) {
+        if let Err(e) = self.send_cursor.send(data) {
+            warn!("failed to send cursor data: {e}");
+        }
+    }
+
+    pub fn send_stats_data(&self, data: Vec<u8>) {
+        if let Err(e) = self.send_stats.send(data) {
+            warn!("failed to send stats data: {e}");
+        }
+    }
+}
+
+async fn run_input_reader(
+    conn: Connection,
+    input_broadcast: tokio::sync::broadcast::Sender<Vec<u8>>,
+    idr_cmd_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+) {
+    debug!("input reader started");
+    loop {
+        debug!("input reader opening bidi stream");
+        match conn.open_bi().await {
+            Ok((mut send, mut recv)) => {
+                debug!("input bidi stream opened, writing type+version byte");
+                if send.write_all(&[BIDI_INPUT, STREAM_VERSION]).await.is_err() {
+                    debug!("input type byte write failed");
+                    break;
+                }
+                let _ = send.finish();
+                debug!("input bidi stream ready, reading framed events");
+
+                loop {
+                    // Read uniform frame: [4B len][1B type][2B seq][payload]
+                    let mut len_buf = [0u8; 4];
+                    if recv.read_exact(&mut len_buf).await.is_err() {
+                        break;
+                    }
+                    let frame_len = u32::from_le_bytes(len_buf) as usize;
+                    if frame_len < 3 || frame_len > 65536 {
+                        break;
+                    }
+                    let mut frame = vec![0u8; frame_len];
+                    if recv.read_exact(&mut frame).await.is_err() {
+                        break;
+                    }
+                    let msg_type = frame[0];
+                    let _seq = u16::from_le_bytes([frame[1], frame[2]]);
+                    let payload = &frame[3..];
+
+                    match msg_type {
+                        MSG_INPUT_BATCH => {
+                            let mut offset = 0;
+                            while offset < payload.len() {
+                                if offset + 1 > payload.len() {
+                                    break;
+                                }
+                                match payload[offset] {
+                                    INPUT_KEY => {
+                                        if offset + 4 > payload.len() {
+                                            break;
+                                        }
+                                        let raw = vec![
+                                            INPUT_KEY,
+                                            payload[offset + 1],
+                                            payload[offset + 2],
+                                            payload[offset + 3],
+                                        ];
+                                        let _ = input_broadcast.send(raw);
+                                        offset += 4;
+                                    }
+                                    INPUT_MOUSE_MOVE => {
+                                        if offset + 5 > payload.len() {
+                                            break;
+                                        }
+                                        let mut raw = Vec::with_capacity(5);
+                                        raw.push(INPUT_MOUSE_MOVE);
+                                        raw.extend_from_slice(&payload[offset + 1..offset + 5]);
+                                        let _ = input_broadcast.send(raw);
+                                        offset += 5;
+                                    }
+                                    INPUT_MOUSE_BUTTON => {
+                                        if offset + 3 > payload.len() {
+                                            break;
+                                        }
+                                        let raw = vec![
+                                            INPUT_MOUSE_BUTTON,
+                                            payload[offset + 1],
+                                            payload[offset + 2],
+                                        ];
+                                        let _ = input_broadcast.send(raw);
+                                        offset += 3;
+                                    }
+                                    INPUT_MOUSE_WHEEL => {
+                                        if offset + 5 > payload.len() {
+                                            break;
+                                        }
+                                        let mut raw = Vec::with_capacity(5);
+                                        raw.push(INPUT_MOUSE_WHEEL);
+                                        raw.extend_from_slice(&payload[offset + 1..offset + 5]);
+                                        let _ = input_broadcast.send(raw);
+                                        offset += 5;
+                                    }
+                                    _ => {
+                                        debug!("unknown input event type: {}", payload[offset]);
+                                        break;
+                                    }
+                                }
+                            }
+                        }
+                        MSG_IDR_REQUEST => {
+                            info!("received IDR request from client");
+                            let _ = idr_cmd_tx.send(vec![MSG_IDR_REQUEST]);
+                        }
+                        MSG_ENCODE_SETTINGS => {
+                            info!(
+                                "received encode settings from client ({} bytes)",
+                                payload.len()
+                            );
+                            let mut cmd = Vec::with_capacity(1 + payload.len());
+                            cmd.push(MSG_ENCODE_SETTINGS);
+                            cmd.extend_from_slice(payload);
+                            let _ = idr_cmd_tx.send(cmd);
+                        }
+                        _ => {
+                            debug!("unknown bidi msg type: {}", msg_type);
+                        }
+                    }
+                }
+            }
+            Err(e) => {
+                debug!("input open_bi failed: {e}");
+                break;
+            }
+        }
+    }
+    debug!("input reader exiting");
+}
+
+async fn run_video_sender(
+    conn: Connection,
+    mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+    relay_ms: Arc<AtomicU32>,
+) {
+    loop {
+        let first = match rx.recv().await {
+            Some(data) => data,
+            None => {
+                debug!("video sender exiting (channel closed)");
+                return;
+            }
+        };
+
+        let mut send = match conn.open_uni().await {
+            Ok(s) => s,
+            Err(e) => {
+                debug!("video open_uni failed: {e}");
+                break;
+            }
+        };
+        debug!("video uni stream opened");
+
+        if send
+            .write_all(&[STREAM_VIDEO, STREAM_VERSION])
+            .await
+            .is_err()
+        {
+            let _ = send.finish();
+            break;
+        }
+
+        let mut seq: u16 = 0;
+        let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len());
+        encode_frame(&mut buf, MSG_DATA, seq, &first);
+        if send.write_all(&buf).await.is_err() {
+            let _ = send.finish();
+            break;
+        }
+        seq = seq.wrapping_add(1);
+
+        loop {
+            match rx.recv().await {
+                Some(bytes) => {
+                    let t0 = std::time::Instant::now();
+                    buf.clear();
+                    encode_frame(&mut buf, MSG_DATA, seq, &bytes);
+                    if send.write_all(&buf).await.is_err() {
+                        break;
+                    }
+                    let elapsed = t0.elapsed().as_secs_f32() * 1000.0;
+                    relay_ms.store(elapsed.to_bits(), Ordering::Relaxed);
+                    seq = seq.wrapping_add(1);
+                }
+                None => {
+                    let _ = send.finish();
+                    debug!("video sender exiting (channel closed)");
+                    return;
+                }
+            }
+        }
+        let _ = send.finish();
+    }
+    debug!("video sender exiting");
+}
+
+async fn run_audio_sender(conn: Connection, mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>) {
+    loop {
+        let first = match rx.recv().await {
+            Some(data) => data,
+            None => {
+                debug!("audio sender exiting (channel closed)");
+                return;
+            }
+        };
+
+        let mut send = match conn.open_uni().await {
+            Ok(s) => s,
+            Err(e) => {
+                debug!("audio open_uni failed: {e}");
+                break;
+            }
+        };
+        debug!("audio uni stream opened");
+
+        if send
+            .write_all(&[STREAM_AUDIO, STREAM_VERSION])
+            .await
+            .is_err()
+        {
+            let _ = send.finish();
+            break;
+        }
+
+        let mut seq: u16 = 0;
+        let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len());
+        encode_frame(&mut buf, MSG_DATA, seq, &first);
+        if send.write_all(&buf).await.is_err() {
+            let _ = send.finish();
+            break;
+        }
+        seq = seq.wrapping_add(1);
+
+        loop {
+            match rx.recv().await {
+                Some(bytes) => {
+                    buf.clear();
+                    encode_frame(&mut buf, MSG_DATA, seq, &bytes);
+                    if send.write_all(&buf).await.is_err() {
+                        break;
+                    }
+                    seq = seq.wrapping_add(1);
+                }
+                None => {
+                    let _ = send.finish();
+                    debug!("audio sender exiting (channel closed)");
+                    return;
+                }
+            }
+        }
+        let _ = send.finish();
+    }
+    debug!("audio sender exiting");
+}
+
+async fn run_cursor_sender(
+    conn: Connection,
+    mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+) {
+    loop {
+        let first = match rx.recv().await {
+            Some(data) => data,
+            None => {
+                debug!("cursor sender exiting (channel closed)");
+                return;
+            }
+        };
+
+        let mut send = match conn.open_uni().await {
+            Ok(s) => s,
+            Err(e) => {
+                debug!("cursor open_uni failed: {e}");
+                break;
+            }
+        };
+        debug!("cursor uni stream opened");
+
+        if send
+            .write_all(&[STREAM_CURSOR, STREAM_VERSION])
+            .await
+            .is_err()
+        {
+            let _ = send.finish();
+            break;
+        }
+
+        let msg_type = if first.is_empty() { 0 } else { first[0] };
+        let payload = if first.len() > 1 { &first[1..] } else { &[] };
+        let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len());
+        encode_frame(&mut buf, msg_type, 0, payload);
+        if send.write_all(&buf).await.is_err() {
+            let _ = send.finish();
+            break;
+        }
+
+        let mut sent: u64 = 1;
+        loop {
+            match rx.recv().await {
+                Some(bytes) => {
+                    sent += 1;
+                    if sent <= 3 {
+                        debug!(
+                            "cursor sender: sending update #{sent} ({} bytes)",
+                            bytes.len()
+                        );
+                    }
+                    buf.clear();
+                    let mt = if bytes.is_empty() { 0 } else { bytes[0] };
+                    let p = if bytes.len() > 1 { &bytes[1..] } else { &[] };
+                    encode_frame(&mut buf, mt, 0, p);
+                    if send.write_all(&buf).await.is_err() {
+                        break;
+                    }
+                }
+                None => {
+                    let _ = send.finish();
+                    debug!("cursor sender exiting (channel closed)");
+                    return;
+                }
+            }
+        }
+        let _ = send.finish();
+    }
+    debug!("cursor sender exiting");
+}
+
+async fn run_stats_sender(conn: Connection, mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>) {
+    loop {
+        let first = match rx.recv().await {
+            Some(data) => data,
+            None => {
+                debug!("stats sender exiting (channel closed)");
+                return;
+            }
+        };
+        let mut send = match conn.open_uni().await {
+            Ok(s) => s,
+            Err(e) => {
+                debug!("stats open_uni failed: {e}");
+                break;
+            }
+        };
+        debug!("stats uni stream opened");
+        if send
+            .write_all(&[STREAM_STATS, STREAM_VERSION])
+            .await
+            .is_err()
+        {
+            let _ = send.finish();
+            break;
+        }
+
+        let st = if first.is_empty() { 0 } else { first[0] };
+        let payload = if first.len() > 1 { &first[1..] } else { &[] };
+        let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len());
+        encode_frame(&mut buf, st, 0, payload);
+        if send.write_all(&buf).await.is_err() {
+            let _ = send.finish();
+            break;
+        }
+
+        loop {
+            match rx.recv().await {
+                Some(bytes) => {
+                    buf.clear();
+                    let mt = if bytes.is_empty() { 0 } else { bytes[0] };
+                    let p = if bytes.len() > 1 { &bytes[1..] } else { &[] };
+                    encode_frame(&mut buf, mt, 0, p);
+                    if send.write_all(&buf).await.is_err() {
+                        break;
+                    }
+                }
+                None => {
+                    let _ = send.finish();
+                    debug!("stats sender exiting (channel closed)");
+                    return;
+                }
+            }
+        }
+        let _ = send.finish();
+    }
+    debug!("stats sender exiting");
+}
+
+pub struct SessionManager {
+    sessions: Arc<Mutex<HashMap<iroh::EndpointId, ClientSession>>>,
+    video_bytes: AtomicU64,
+    last_video_bytes: AtomicU64,
+    video_bitrate: AtomicU64, // bytes/sec
+    relay_ms: Arc<AtomicU32>, // latest relay latency (f32 bits)
+}
+
+impl SessionManager {
+    pub fn new() -> Self {
+        Self {
+            sessions: Arc::new(Mutex::new(HashMap::new())),
+            video_bytes: AtomicU64::new(0),
+            last_video_bytes: AtomicU64::new(0),
+            video_bitrate: AtomicU64::new(0),
+            relay_ms: Arc::new(AtomicU32::new(0)),
+        }
+    }
+
+    pub async fn add_session(&self, id: iroh::EndpointId, session: ClientSession) {
+        let mut sessions = self.sessions.lock().await;
+        sessions.insert(id, session);
+        info!(remote = %id.fmt_short(), "client session added ({} total)", sessions.len());
+    }
+
+    pub async fn remove_session(&self, id: &iroh::EndpointId) {
+        let mut sessions = self.sessions.lock().await;
+        sessions.remove(id);
+        info!(remote = %id.fmt_short(), "client session removed ({} remaining)", sessions.len());
+    }
+
+    pub async fn broadcast_video(&self, data: Vec<u8>) {
+        self.video_bytes
+            .fetch_add(data.len() as u64, Ordering::Relaxed);
+        let sessions = self.sessions.lock().await;
+        if sessions.is_empty() {
+            return;
+        }
+        for session in sessions.values() {
+            session.send_video_frame(data.clone());
+        }
+    }
+
+    pub async fn broadcast_audio(&self, data: Vec<u8>) {
+        let sessions = self.sessions.lock().await;
+        if sessions.is_empty() {
+            return;
+        }
+        for session in sessions.values() {
+            session.send_audio_packet(data.clone());
+        }
+    }
+
+    pub async fn broadcast_cursor(&self, data: Vec<u8>) {
+        let sessions = self.sessions.lock().await;
+        if sessions.is_empty() {
+            return;
+        }
+        for session in sessions.values() {
+            session.send_cursor_data(data.clone());
+        }
+    }
+
+    pub async fn broadcast_stats(&self, data: Vec<u8>) {
+        let sessions = self.sessions.lock().await;
+        if sessions.is_empty() {
+            return;
+        }
+        for session in sessions.values() {
+            session.send_stats_data(data.clone());
+        }
+    }
+
+    pub fn relay_ms(&self) -> f32 {
+        f32::from_bits(self.relay_ms.swap(0, Ordering::Relaxed))
+    }
+
+    pub fn relay_ms_atomic(&self) -> Arc<AtomicU32> {
+        self.relay_ms.clone()
+    }
+
+    pub async fn client_count(&self) -> usize {
+        self.sessions.lock().await.len()
+    }
+
+    pub fn video_bitrate_bps(&self) -> u32 {
+        let current = self.video_bytes.load(Ordering::Relaxed);
+        let last = self.last_video_bytes.swap(current, Ordering::Relaxed);
+        let diff = current.saturating_sub(last);
+        self.video_bitrate.store(diff, Ordering::Relaxed);
+        (diff * 8) as u32 // bits per second
+    }
+}

apps/neshub/src/ticket.rs

@@ -0,0 +1,129 @@
+//! The ticket: how a client finds this box.
+//!
+//! neshub makes one per boot and serves it on a Unix socket. It listens and
+//! nesinit dials, the same way every other socket here works -- so there is no
+//! race against a process that has not started yet, and no file left on disk
+//! holding a live address after the box is gone.
+
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+use iroh::EndpointAddr;
+use serde::{Deserialize, Serialize};
+use tokio::io::AsyncWriteExt;
+use tokio::net::UnixListener;
+
+/// A Nestri connection ticket.
+///
+/// Format: `nestri:<base64_serialized_ticket>`
+///
+/// Example: `nestri:eyJlbmRwb2ludF9hZGRyIjp7...`
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NestriTicket {
+    pub endpoint_addr: EndpointAddr,
+    pub stream_name: String,
+}
+impl NestriTicket {
+    pub fn new(endpoint_addr: EndpointAddr, stream_name: String) -> Self {
+        Self {
+            endpoint_addr,
+            stream_name,
+        }
+    }
+
+    /// Encode the full ticket (including stream name) to a string.
+    pub fn encode(&self) -> String {
+        let ticket_bytes = serde_json::to_vec(self).unwrap_or_default();
+        let ticket_b64 = URL_SAFE_NO_PAD.encode(&ticket_bytes);
+        format!("nestri:{}", ticket_b64)
+    }
+
+    /// Decode a ticket string.
+    pub fn decode(ticket: &str) -> Option<Self> {
+        let rest = ticket.strip_prefix("nestri:")?;
+        let ticket_bytes = URL_SAFE_NO_PAD.decode(rest.as_bytes()).ok()?;
+        serde_json::from_slice(&ticket_bytes).ok()
+    }
+}
+impl std::fmt::Display for NestriTicket {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.encode())
+    }
+}
+impl std::str::FromStr for NestriTicket {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        NestriTicket::decode(s).ok_or_else(|| anyhow::anyhow!("invalid ticket format"))
+    }
+}
+
+/// Generate a unique stream ID using UUID v7.
+pub fn generate_stream_name() -> String {
+    format!("stream-{}", uuid::Uuid::now_v7().as_simple())
+}
+
+/// Serve the ticket to anything that dials, forever.
+///
+/// Returns once the socket is bound, not when someone connects: the caller has
+/// an endpoint to run and a reader that never arrives is not a reason to refuse
+/// clients. Every connection gets the ticket and a newline, then is dropped --
+/// there is nothing to say afterwards, and a reader blocked on more would hang.
+pub fn serve(path: &Path, ticket: String) -> Result<()> {
+    // A stale socket file makes bind fail with EADDRINUSE, which reads as
+    // "something is already listening" when nothing is.
+    let _ = std::fs::remove_file(path);
+    let listener = UnixListener::bind(path)
+        .with_context(|| format!("could not listen on {}", path.display()))?;
+    tracing::info!("serving the ticket on {}", path.display());
+
+    tokio::spawn(async move {
+        let line = format!("{ticket}\n");
+        loop {
+            match listener.accept().await {
+                Ok((mut stream, _)) => {
+                    if let Err(e) = stream.write_all(line.as_bytes()).await {
+                        tracing::warn!("could not hand over the ticket: {e}");
+                    }
+                }
+                Err(e) => {
+                    tracing::warn!("accepting a ticket connection failed: {e}");
+                    return;
+                }
+            }
+        }
+    });
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The ticket crosses a process boundary as text, so a round trip that
+    /// loses a field would be found by whoever cannot connect, not here.
+    #[test]
+    fn a_ticket_survives_the_round_trip() {
+        let addr = EndpointAddr::from(iroh::SecretKey::generate().public());
+        let ticket = NestriTicket::new(addr.clone(), "stream-abc".into());
+
+        let decoded = NestriTicket::decode(&ticket.encode()).expect("a ticket we just encoded");
+
+        assert_eq!(decoded.stream_name, "stream-abc");
+        assert_eq!(decoded.endpoint_addr.id, addr.id);
+    }
+
+    #[test]
+    fn anything_not_a_ticket_is_refused() {
+        assert!(NestriTicket::decode("nestri:@@@not-base64@@@").is_none());
+        assert!(NestriTicket::decode("stream-abc").is_none(), "no prefix");
+    }
+
+    /// Two boxes handing out the same stream name would collide in whatever is
+    /// keyed by it.
+    #[test]
+    fn stream_names_differ() {
+        assert_ne!(generate_stream_name(), generate_stream_name());
+    }
+}