docs(#2862): document the push side and regenerate the CLI reference

docs/tools/hivectl-cli.md is generated by `hivectl markdown-docs` and
diffed against a fresh run by the hivectl-docs flake check, so adding
the push verb without regenerating it would have failed CI. The
regeneration also retires two copies of a sentence describing the
cross-hive leg as an ssh pipe that "isn't wired up yet" -- ssh was
dropped when the WireGuard mesh became the authentication, and the leg
is wired up now.

snapshot-store.md documented only the receiving host, so a reader had
no way to learn how a pushing hive is told where the store is. It now
covers services.hyperhive.swarm.snapshotStore, including why address
has no default and port does: an address is a deployment fact that
cannot be guessed, a port is a convention both ends read from the same
docs. It also states the namespace rule the two options illustrate --
swarm.* describes the swarm as seen from here, a bare
services.hyperhive.<service> describes a role this host performs.

swarm.md never mentioned the store even though the option lives in its
namespace, so a reader configuring swarm.peers had no signal it exists.

docs/snapshot-store.md

@@ -41,6 +41,41 @@ Note that the mesh is gated on `swarm.wireguard.enable`, **not** on
 `c0re.enable` --- a store host runs no hive and would otherwise get no
 `wg-hive` interface at all.
 
+## Pointing a hive at it
+
+The block above configures the host that *receives*. Every hive that
+*pushes* separately needs to be told where the store is:
+
+```nix
+services.hyperhive.swarm.snapshotStore = {
+  address = "10.100.0.9";  # the store's mesh address, no prefix
+  port = 51821;            # optional; must match the receiver's port
+};
+```
+
+Two deliberate asymmetries in that pair, both easy to misread as
+inconsistency:
+
+- **`address` has no default.** It is a deployment fact a pushing hive
+  cannot derive, and a wrong guess means streaming an agent's state at
+  whatever happens to answer. Unset, a push fails naming this option.
+- **`port` does default** (`51821`), because it is a convention both
+  ends read from the same option docs --- a default there is
+  coordination, not a guess.
+
+Note the option lives under `swarm.*` while the receiving host's lives
+under `services.hyperhive.snapshotStore`. That is the distinction the
+two namespaces carry throughout: `swarm.*` describes *the swarm* as seen
+from this host, and a bare `services.hyperhive.<service>` describes *a
+role this host performs*. A store host sets both --- one to run the
+receiver, one only if it also runs a hive that pushes.
+
+With it set, `hivectl agent <name> subvol snapshot push <label>
+[--parent <label>]` streams a snapshot straight into the store. There is
+no destination argument, because a swarm has exactly one store (see
+[One subvolume per agent, not per hive](#one-subvolume-per-agent-not-per-hive)),
+and no credential argument, because the mesh is the authentication.
+
 ## The mesh is the authentication
 
 There are no certificates here, and no key material of its own. That is

docs/swarm.md

@@ -221,8 +221,22 @@ other side initiates. With keepalive on, the NAT hole stays open.
 If both hosts are behind NAT, a STUN relay or a third host (exit node)
 is required. Out of scope for v0.
 
+## Snapshot store
+
+One further option lives in this namespace but is documented with the
+service it points at: `services.hyperhive.swarm.snapshotStore.{address,
+port}` tells this hive where the swarm's `btrfs receive` endpoint is, so
+`hivectl agent <name> subvol snapshot push` has somewhere to stream to.
+
+It is genuinely swarm-scoped rather than per-peer — a swarm has exactly
+one store, because the receiver keys destinations by *agent* so a
+migrating agent keeps one unbroken incremental chain. See
+[snapshot-store.md](snapshot-store.md).
+
 ## Cross-references
 
+- `docs/snapshot-store.md` — the swarm's `btrfs receive` endpoint, and
+  the `swarm.snapshotStore` option that points a hive at it
 - `docs/conventions.md` § Hive identity — env vars, qualified labels
 - `docs/matrix.md` — matrix federation, TLS cert auto-generation,
   firewall posture

docs/tools/hivectl-cli.md

@@ -42,6 +42,7 @@ This document contains the help content for the `hivectl` command-line program.
 * [`hivectl agent subvol snapshot create`↴](#hivectl-agent-subvol-snapshot-create)
 * [`hivectl agent subvol snapshot delete`↴](#hivectl-agent-subvol-snapshot-delete)
 * [`hivectl agent subvol snapshot send`↴](#hivectl-agent-subvol-snapshot-send)
+* [`hivectl agent subvol snapshot push`↴](#hivectl-agent-subvol-snapshot-push)
 * [`hivectl list-agents`↴](#hivectl-list-agents)
 * [`hivectl quota-enable`↴](#hivectl-quota-enable)
 * [`hivectl approvals`↴](#hivectl-approvals)
@@ -562,7 +563,8 @@ Read-only snapshots of this agent's state subvolume
 
 * `create` — Create a read-only snapshot (agent must already be a subvolume)
 * `delete` — Delete a snapshot created by `subvol snapshot create`
-* `send` — Export a snapshot to a local file via `btrfs send` (the local-file half of inter-hive migration transport; the cross-hive `ssh ... btrfs receive` leg isn't wired up yet). Also useful standalone as a point-in-time backup: a full send with no `--parent` produces a self-contained archive of the snapshot
+* `send` — Export a snapshot to a local file via `btrfs send` — the local-file half of the inter-hive migration transport (`push` is the network half). Also useful standalone as a point-in-time backup: a full send with no `--parent` produces a self-contained archive of the snapshot
+* `push` — Stream a snapshot to the swarm's snapshot store over the WireGuard mesh — the network half of the migration transport
 
 
 
@@ -592,7 +594,7 @@ Delete a snapshot created by `subvol snapshot create`
 
 ## `hivectl agent subvol snapshot send`
 
-Export a snapshot to a local file via `btrfs send` (the local-file half of inter-hive migration transport; the cross-hive `ssh ... btrfs receive` leg isn't wired up yet). Also useful standalone as a point-in-time backup: a full send with no `--parent` produces a self-contained archive of the snapshot
+Export a snapshot to a local file via `btrfs send` — the local-file half of the inter-hive migration transport (`push` is the network half). Also useful standalone as a point-in-time backup: a full send with no `--parent` produces a self-contained archive of the snapshot
 
 **Usage:** `hivectl agent subvol snapshot send [OPTIONS] --dest <DEST> <LABEL>`
 
@@ -607,6 +609,26 @@ Export a snapshot to a local file via `btrfs send` (the local-file half of inter
 
 
 
+## `hivectl agent subvol snapshot push`
+
+Stream a snapshot to the swarm's snapshot store over the WireGuard mesh — the network half of the migration transport.
+
+Nothing is staged locally: `btrfs send` writes straight into the connection, so a multi-gigabyte agent needs no scratch space on this host. The mesh is the authentication (cryptokey routing binds the sender's address to its key), so there is no credential to pass here.
+
+There is no destination argument: a swarm has one store, read from `services.hyperhive.swarm.snapshotStore`.
+
+**Usage:** `hivectl agent subvol snapshot push [OPTIONS] <LABEL>`
+
+###### **Arguments:**
+
+* `<LABEL>` — Snapshot label passed to `subvol snapshot create --label`
+
+###### **Options:**
+
+* `--parent <PARENT>` — Optional parent snapshot label for an incremental send (`btrfs send -p`) — must be an existing, older snapshot of the same agent, and must already be present on the receiver. Omit for a full send
+
+
+
 ## `hivectl list-agents`
 
 Show all managed agents with their status and technical state.

hive-c0re/src/main.rs

@@ -31,6 +31,7 @@ mod paths;
 mod priv_client;
 mod questions;
 mod server;
+mod snapshot_push;
 mod socket_server;
 mod stats;
 mod stores;

hive-c0re/src/priv_client.rs

@@ -11,7 +11,9 @@ use hive_priv_sock::{
     BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, NetworkIsolation,
     PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
 };
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use std::os::fd::{AsRawFd as _, OwnedFd, RawFd};
+
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest};
 use tokio::net::UnixStream;
 
 /// Send a single request to `hive-priv` and return the response.
@@ -40,6 +42,140 @@ pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
     }
 }
 
+/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS`
+/// message. `CMSG_SPACE` is not a `const fn`, so the size is a literal
+/// with room to spare; the union member supplies the `cmsghdr`
+/// alignment `CMSG_FIRSTHDR` requires.
+#[repr(C)]
+union CmsgSpace {
+    _align: libc::cmsghdr,
+    bytes: [u8; 32],
+}
+
+/// `sendmsg` `bytes` with `fd` attached as `SCM_RIGHTS`, returning how
+/// many bytes were accepted.
+///
+/// The descriptor rides on this one call — ancillary data cannot be
+/// sent separately from payload — so the caller must not have written
+/// any of `bytes` beforehand.
+fn send_with_fd(sock: RawFd, bytes: &[u8], fd: RawFd) -> std::io::Result<usize> {
+    const FD_SIZE: usize = std::mem::size_of::<RawFd>();
+
+    let mut iov = libc::iovec {
+        iov_base: bytes.as_ptr().cast::<libc::c_void>().cast_mut(),
+        iov_len: bytes.len(),
+    };
+    let mut cmsg = CmsgSpace { bytes: [0; 32] };
+    // SAFETY: msghdr is a plain C struct with no invalid bit patterns;
+    // every field we rely on is set immediately below.
+    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
+    msg.msg_iov = &raw mut iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast();
+    // SAFETY: CMSG_SPACE is a pure size computation.
+    let space = unsafe { libc::CMSG_SPACE(u32::try_from(FD_SIZE).unwrap_or(4)) };
+    msg.msg_controllen = space as _;
+
+    // SAFETY: the control buffer is live, aligned, and long enough for
+    // the header CMSG_SPACE just sized.
+    let hdr = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) };
+    if hdr.is_null() {
+        return Err(std::io::Error::other(
+            "control buffer too small for SCM_RIGHTS",
+        ));
+    }
+    // SAFETY: CMSG_LEN is a pure size computation; `hdr` points into
+    // our own buffer, and write_unaligned tolerates its alignment.
+    unsafe {
+        let len = libc::CMSG_LEN(u32::try_from(FD_SIZE).unwrap_or(4));
+        std::ptr::write_unaligned(
+            hdr,
+            libc::cmsghdr {
+                cmsg_len: len as _,
+                cmsg_level: libc::SOL_SOCKET,
+                cmsg_type: libc::SCM_RIGHTS,
+            },
+        );
+        // Copied in byte-wise: the control buffer is only cmsghdr-
+        // aligned, so casting CMSG_DATA to a *mut RawFd would be
+        // unsound even where it happens to work.
+        std::ptr::copy_nonoverlapping(
+            std::ptr::from_ref(&fd).cast::<u8>(),
+            libc::CMSG_DATA(hdr),
+            FD_SIZE,
+        );
+    }
+
+    // SAFETY: `msg` points at a live iovec over `bytes` and the control
+    // buffer we just filled in.
+    let n = unsafe { libc::sendmsg(sock, &raw const msg, 0) };
+    if n < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    Ok(usize::try_from(n).unwrap_or_default())
+}
+
+/// Send a request to `hive-priv` with an open file descriptor attached,
+/// and return the response.
+///
+/// The helper receives the descriptor itself — not a path or an address
+/// — so it can act on something it was handed without being told what
+/// that thing is or how to reach it. Used for
+/// [`PrivRequest::SendAgentSnapshotToFd`], where the descriptor is a
+/// socket already connected to a peer hive's snapshot store.
+///
+/// ⚠️ Takes the descriptor by value and closes it as soon as the kernel
+/// has it, *before* awaiting the response. That is not tidiness: a
+/// socket stays open until every copy of it is closed, so a caller
+/// holding one back would leave the receiving end waiting for an EOF
+/// that never comes — `btrfs receive` blocks, and this side reports
+/// success for a transfer the peer has not committed. Passing ownership
+/// makes that mistake unrepresentable.
+///
+/// # Errors
+///
+/// Fails if the socket is unreachable, the descriptor cannot be
+/// attached, or hive-priv answers with something other than a terminal
+/// event.
+pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
+    let mut stream = UnixStream::connect(PRIV_SOCK)
+        .await
+        .context("connect to hive-priv socket (fd-passing)")?;
+    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
+    let bytes = line.as_bytes();
+
+    let sock = stream.as_raw_fd();
+    let raw_fd = fd.as_raw_fd();
+    let sent = stream
+        .async_io(Interest::WRITABLE, || send_with_fd(sock, bytes, raw_fd))
+        .await
+        .context("send request + descriptor to hive-priv")?;
+    // The kernel has duplicated the descriptor into hive-priv's queue,
+    // so our copy has done its job. Close it now, before waiting on the
+    // response: see the EOF note on this function.
+    drop(fd);
+
+    // A short sendmsg is legal; the descriptor went with the first
+    // call, so the tail is an ordinary write.
+    if sent < bytes.len() {
+        stream
+            .write_all(&bytes[sent..])
+            .await
+            .context("send remainder of request to hive-priv")?;
+    }
+    stream.shutdown().await.context("shutdown write half")?;
+
+    let mut resp_line = String::new();
+    BufReader::new(stream)
+        .read_line(&mut resp_line)
+        .await
+        .context("read response from hive-priv")?;
+    match serde_json::from_str::<PrivEvent>(&resp_line).context("parse PrivResponse")? {
+        PrivEvent::Done(resp) => Ok(resp),
+        PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"),
+    }
+}
+
 /// Send a streaming request to `hive-priv`, calling `on_line` for each
 /// `PrivEvent::Line` as it arrives, then returning the terminal
 /// `PrivResponse`. Used for long-running ops (`create` / `update`).

hive-c0re/src/server.rs

@@ -264,6 +264,11 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
                 parent,
                 dest,
             } => handle_send_snapshot(name.as_str(), label, parent.as_deref(), dest).await?,
+            HostRequest::PushSnapshot {
+                name,
+                label,
+                parent,
+            } => handle_push_snapshot(name.as_str(), label, parent.as_deref()).await?,
         })
     }
     .await;
@@ -661,6 +666,20 @@ async fn handle_send_snapshot(
     Ok(HostResponse::messages(vec![path]))
 }
 
+/// Push a snapshot to the swarm's store. The network sibling of
+/// [`handle_send_snapshot`]: nothing lands on this host, so success is
+/// bare rather than a path.
+async fn handle_push_snapshot(
+    name: &str,
+    label: &str,
+    parent: Option<&str>,
+) -> Result<HostResponse> {
+    crate::snapshot_push::push_agent_snapshot(name, label, parent)
+        .await
+        .with_context(|| format!("push {name} snapshot (label {label:?}) to the swarm store"))?;
+    Ok(HostResponse::success())
+}
+
 async fn handle_matrix_sync_admin() -> Result<HostResponse> {
     require_matrix_present().await?;
     let register_token =

hive-c0re/src/snapshot_push.rs

@@ -0,0 +1,148 @@
+//! Push an agent snapshot to the swarm's snapshot store.
+//!
+//! This is the c0re half of the transport: it resolves the store,
+//! opens the connection, writes the protocol header, and hands the
+//! connected socket to `hive-priv`, which runs `btrfs send` straight
+//! into it.
+//!
+//! The split is the point. Everything that requires knowing *where* the
+//! store is and *what* the wire protocol looks like happens here, in the
+//! unprivileged daemon. hive-priv only ever receives an already-open
+//! descriptor, so the root helper never learns an address, never parses
+//! a peer list, and never decides who to trust — and once `btrfs send`
+//! starts, neither process is in the data path.
+//!
+//! There is exactly **one** store per swarm, not one per peer hive. The
+//! receiver keys destinations by *agent*, so an agent that migrates
+//! between hives keeps a single unbroken incremental chain; per-hive
+//! stores would split that chain in two, which is the case the store
+//! exists to serve.
+
+use anyhow::{Context as _, Result, bail};
+use hive_priv_sock::PrivRequest;
+use tokio::io::AsyncWriteExt as _;
+use tokio::net::TcpStream;
+
+use crate::priv_client;
+
+/// `host:port` of the swarm's snapshot store, rendered by
+/// `nix/host-modules/hive-c0re/environment.nix` from
+/// `services.hyperhive.swarm.snapshotStore`.
+const STORE_ENV: &str = "HYPERHIVE_SNAPSHOT_STORE";
+
+/// Send `snapshot` of `agent` to the swarm's snapshot store, optionally
+/// as an incremental against `parent`.
+///
+/// # Errors
+///
+/// Fails when no store is configured, it is unreachable, or hive-priv
+/// reports the `btrfs send` failed.
+pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&str>) -> Result<()> {
+    let configured = std::env::var(STORE_ENV).ok();
+    let addr = require_store_addr(configured.as_deref())?;
+
+    let mut sock = TcpStream::connect(addr)
+        .await
+        .with_context(|| format!("connect to the swarm snapshot store at {addr}"))?;
+
+    // The header names which agent this stream belongs to: a btrfs
+    // stream carries the sender's own subvolume name, not the hive's
+    // notion of the agent, so the receiver cannot infer it. One line,
+    // because the receiver reads it with `read` and must not buffer
+    // past the newline into the byte stream.
+    sock.write_all(format!("agent {agent}\n").as_bytes())
+        .await
+        .context("send header to the swarm snapshot store")?;
+    sock.flush()
+        .await
+        .context("flush header to the swarm snapshot store")?;
+
+    // Hand the connected socket over. `call_with_fd` takes ownership and
+    // closes our copy as soon as the kernel has it, so the receiver sees
+    // EOF when `btrfs send` finishes rather than hanging on a descriptor
+    // this process is still holding.
+    let fd = sock
+        .into_std()
+        .context("detach snapshot-store socket for hand-off")?
+        .into();
+
+    let resp = priv_client::call_with_fd(
+        &PrivRequest::SendAgentSnapshotToFd {
+            agent_name: agent.to_owned(),
+            snapshot_name: snapshot.to_owned(),
+            parent_snapshot_name: parent.map(str::to_owned),
+        },
+        fd,
+    )
+    .await
+    .with_context(|| format!("stream {agent}/{snapshot} to the swarm snapshot store"))?;
+
+    if !resp.ok {
+        bail!(
+            "push of {agent}/{snapshot} failed: {}",
+            resp.error.unwrap_or_else(|| resp.stderr.clone())
+        );
+    }
+    tracing::info!(
+        agent,
+        snapshot,
+        parent,
+        store = addr,
+        "pushed agent snapshot to the swarm snapshot store"
+    );
+    Ok(())
+}
+
+/// Require a configured store address.
+///
+/// Split out of [`push_agent_snapshot`] so the "not configured" path is
+/// testable without setting a process-wide env var, which would race
+/// other tests.
+///
+/// # Errors
+///
+/// Fails when unset, naming the option to set. There is deliberately no
+/// fallback: an address is a deployment fact this host cannot derive,
+/// and guessing one means pushing an agent's state at whatever happens
+/// to answer.
+fn require_store_addr(configured: Option<&str>) -> Result<&str> {
+    match configured {
+        Some(addr) if !addr.is_empty() => Ok(addr),
+        _ => bail!(
+            "no snapshot store configured for this swarm — set \
+             services.hyperhive.swarm.snapshotStore.address (and .port if it \
+             differs from the default) on this host"
+        ),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::require_store_addr;
+
+    #[test]
+    fn a_configured_address_passes_through() {
+        assert_eq!(
+            require_store_addr(Some("10.100.0.1:51821")).expect("configured"),
+            "10.100.0.1:51821"
+        );
+    }
+
+    #[test]
+    fn an_unconfigured_store_names_the_option_and_never_guesses() {
+        let err = require_store_addr(None).expect_err("no store means no push");
+        let msg = format!("{err:#}");
+        assert!(msg.contains("swarm.snapshotStore.address"), "{msg}");
+        // A guessed address or port is the failure this is guarding
+        // against: it would push an agent's state at whatever answers.
+        assert!(!msg.contains("51821"), "must not suggest a value: {msg}");
+    }
+
+    #[test]
+    fn an_empty_value_is_treated_as_unset() {
+        // An empty env var is a misconfiguration, not an address; it
+        // must fail the same way rather than reaching connect().
+        let err = require_store_addr(Some("")).expect_err("empty is not an address");
+        assert!(format!("{err:#}").contains("no snapshot store configured"));
+    }
+}

hive-host-sock/src/lib.rs

@@ -375,6 +375,23 @@ pub enum HostRequest {
         parent: Option<String>,
         dest: String,
     },
+    /// Push a snapshot to the swarm's snapshot store over the WireGuard
+    /// mesh (`hivectl agent <name> subvol snapshot push`). The network
+    /// sibling of [`HostRequest::SendSnapshot`]: same snapshot and
+    /// optional incremental `parent`, but the stream goes to the store's
+    /// receiver instead of a local file.
+    ///
+    /// There is no destination field: a swarm has exactly one store, and
+    /// the daemon reads its address from
+    /// `services.hyperhive.swarm.snapshotStore`, failing if none is
+    /// configured. Bare success — nothing is written on this host to
+    /// report a path for.
+    PushSnapshot {
+        name: Ident,
+        label: String,
+        #[serde(default)]
+        parent: Option<String>,
+    },
 }
 
 /// One agent's btrfs qgroup usage row — the [`HostRequest::QuotaShow`]

hive-priv-sock/src/lib.rs

@@ -715,6 +715,35 @@ pub enum PrivRequest {
         dest_file_name: String,
     },
 
+    /// Stream a previously-created read-only snapshot into a file
+    /// descriptor the caller passes alongside this request (`SCM_RIGHTS`
+    /// ancillary data on the same socket): `btrfs send [-p <parent>]
+    /// <snapshot> >&<passed fd>`.
+    ///
+    /// The network half of the inter-hive migration transport. hive-c0re
+    /// connects to the peer hive's snapshot store, writes the header
+    /// itself, and hands the **connected socket** over — so hive-priv
+    /// never learns an address, a protocol, or that a network is
+    /// involved, and nobody sits in the data path once the send starts
+    /// (which is what makes a multi-gigabyte transfer survive a
+    /// hive-c0re restart).
+    ///
+    /// Exactly one descriptor must accompany this request. hive-priv
+    /// rejects the request if none arrived, if more than one did, or if a
+    /// descriptor arrives alongside any *other* operation — no guessing
+    /// when the caller didn't say. Requires root.
+    SendAgentSnapshotToFd {
+        /// Logical agent name (validated by `validate_agent_name`).
+        agent_name: String,
+        /// Snapshot label to send, same validation as `SnapshotAgentSubvolume`.
+        snapshot_name: String,
+        /// Optional parent snapshot label for an incremental
+        /// (`btrfs send -p`) send — must be an older read-only snapshot of
+        /// the same agent, still present on disk. `None` sends the full
+        /// snapshot.
+        parent_snapshot_name: Option<String>,
+    },
+
     /// Write `/etc/tmpfiles.d/hyperhive-agents.conf` for the given agent set
     /// and immediately apply it with `systemd-tmpfiles --create`. Each entry
     /// declares the per-agent runtime dirs (`/run/hyperhive/agents/<name>` and

hive-priv/src/main.rs

@@ -17,6 +17,7 @@
 //! `LISTEN_FDS=1` + `LISTEN_PID=<self>`, the inherited fd 3 is used
 //! instead of binding a fresh socket.
 
+use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};
 use std::path::{Path, PathBuf};
 
 use anyhow::{Context as _, Result, bail};
@@ -26,7 +27,7 @@ use hive_priv_sock::{
     PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
     PrivStreamLine, SIBLING_CONTAINERS,
 };
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncWriteExt, BufReader};
 use tokio::net::unix::OwnedWriteHalf;
 use tokio::net::{UnixListener, UnixStream};
 use tokio::process::Command;
@@ -92,11 +93,162 @@ fn socket_listener() -> Result<UnixListener> {
     Ok(listener)
 }
 
+/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS` message.
+///
+/// `CMSG_SPACE` is not a `const fn`, so the size is a literal with room
+/// to spare (24 bytes are needed for a single descriptor on x86-64). The
+/// union member gives the `cmsghdr` alignment `CMSG_FIRSTHDR` requires —
+/// a bare `[u8; N]` is only byte-aligned and would be undefined behaviour
+/// to walk.
+#[repr(C)]
+union CmsgSpace {
+    _align: libc::cmsghdr,
+    bytes: [u8; 32],
+}
+
+/// One `recvmsg` into `buf`, returning the bytes read plus any file
+/// descriptors that rode along as `SCM_RIGHTS`.
+///
+/// Why not a plain read: ancillary data is attached to a *specific*
+/// `recvmsg` call, so a buffered line reader cannot surface it — it
+/// reads the bytes and silently drops the descriptor.
+///
+/// `MSG_CMSG_CLOEXEC` is not optional: without it a received descriptor
+/// is inherited by every `btrfs` / `nixos-container` child this helper
+/// later spawns.
+fn recv_with_fds(sock: RawFd, buf: &mut [u8]) -> std::io::Result<(usize, Vec<OwnedFd>)> {
+    const FD_SIZE: usize = std::mem::size_of::<RawFd>();
+
+    let mut iov = libc::iovec {
+        iov_base: buf.as_mut_ptr().cast(),
+        iov_len: buf.len(),
+    };
+    let mut cmsg = CmsgSpace { bytes: [0; 32] };
+    // SAFETY: msghdr is a plain C struct with no invalid bit patterns;
+    // every field we care about is set immediately below.
+    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
+    msg.msg_iov = &raw mut iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast();
+    msg.msg_controllen = 32;
+
+    // SAFETY: `msg` points at a live iovec covering `buf` and a live,
+    // correctly aligned control buffer of the length we just declared.
+    let n = unsafe { libc::recvmsg(sock, &raw mut msg, libc::MSG_CMSG_CLOEXEC) };
+    if n < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+
+    // Take ownership of every descriptor the kernel attached, even ones
+    // this protocol never expects: an `OwnedFd` we drop is closed, an
+    // fd we fail to claim is leaked for the lifetime of the process.
+    let mut fds = Vec::new();
+    // SAFETY: `msg` was just filled in by a successful `recvmsg`.
+    let mut cmsgp = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) };
+    while !cmsgp.is_null() {
+        // SAFETY: CMSG_FIRSTHDR / CMSG_NXTHDR only ever return a pointer
+        // to a complete header inside the control buffer.
+        let hdr = unsafe { std::ptr::read_unaligned(cmsgp) };
+        if hdr.cmsg_level == libc::SOL_SOCKET && hdr.cmsg_type == libc::SCM_RIGHTS {
+            // SAFETY: same, and CMSG_LEN(0) is the header's own length.
+            let payload = hdr.cmsg_len as usize - unsafe { libc::CMSG_LEN(0) } as usize;
+            let count = payload / FD_SIZE;
+            // SAFETY: CMSG_DATA points at `payload` bytes of descriptors.
+            let data = unsafe { libc::CMSG_DATA(cmsgp) };
+            for i in 0..count {
+                // Copied out byte-wise rather than read through a
+                // `*const RawFd`: the control buffer is only guaranteed
+                // `cmsghdr`-aligned, so casting to a more strictly
+                // aligned pointer would be unsound even where it happens
+                // to work.
+                let mut raw = [0u8; FD_SIZE];
+                // SAFETY: i < count, so this reads inside the payload.
+                unsafe {
+                    std::ptr::copy_nonoverlapping(data.add(i * FD_SIZE), raw.as_mut_ptr(), FD_SIZE);
+                }
+                // SAFETY: the kernel just created this descriptor for
+                // us — we are its only owner.
+                fds.push(unsafe { OwnedFd::from_raw_fd(RawFd::from_ne_bytes(raw)) });
+            }
+        }
+        // SAFETY: `cmsgp` came from this same message.
+        cmsgp = unsafe { libc::CMSG_NXTHDR(&raw const msg, cmsgp) };
+    }
+
+    // `n >= 0` was checked above, so the conversion cannot fail; going
+    // through `try_from` keeps it a cast-free, lint-clean widening.
+    let read = usize::try_from(n).unwrap_or_default();
+    Ok((read, fds))
+}
+
+/// Reads newline-delimited requests off one connection, pairing each
+/// with the descriptor that arrived with it.
+///
+/// The pairing is deliberately trivial, because the protocol is:
+/// `hive-sock-client` connects per request, so a connection carries one
+/// line and at most one descriptor. The loop below still handles several
+/// sequential requests (the server always has), but it refuses to guess
+/// — a second descriptor arriving before its line is a protocol error,
+/// not something to queue and hope about.
+struct Requests<'a> {
+    sock: &'a UnixStream,
+    buf: Vec<u8>,
+    fd: Option<OwnedFd>,
+}
+
+impl Requests<'_> {
+    /// Next complete request line and its descriptor, or `None` at EOF.
+    async fn next(&mut self) -> Result<Option<(String, Option<OwnedFd>)>> {
+        loop {
+            if let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
+                let line: Vec<u8> = self.buf.drain(..=nl).take(nl).collect();
+                let line = String::from_utf8(line).context("request line was not valid UTF-8")?;
+                return Ok(Some((line, self.fd.take())));
+            }
+
+            let mut chunk = [0u8; 8192];
+            let raw = self.sock.as_raw_fd();
+            let (n, fds) = self
+                .sock
+                .async_io(tokio::io::Interest::READABLE, || {
+                    recv_with_fds(raw, &mut chunk)
+                })
+                .await
+                .context("recvmsg on the priv socket")?;
+
+            for fd in fds {
+                if self.fd.replace(fd).is_some() {
+                    bail!("more than one file descriptor passed for a single request");
+                }
+            }
+            if n == 0 {
+                if !self.buf.is_empty() {
+                    bail!("connection closed mid-request ({} bytes)", self.buf.len());
+                }
+                return Ok(None);
+            }
+            self.buf.extend_from_slice(&chunk[..n]);
+        }
+    }
+}
+
 async fn handle(stream: UnixStream) {
     let (reader, mut writer) = stream.into_split();
-    let mut lines = BufReader::new(reader).lines();
-    while let Ok(Some(line)) = lines.next_line().await {
-        let resp = dispatch(&line, &mut writer).await;
+    let mut requests = Requests {
+        sock: reader.as_ref(),
+        buf: Vec::new(),
+        fd: None,
+    };
+    loop {
+        let (line, fd) = match requests.next().await {
+            Ok(Some(req)) => req,
+            Ok(None) => break,
+            Err(e) => {
+                tracing::warn!(error = %format!("{e:#}"), "reading request failed");
+                break;
+            }
+        };
+        let resp = dispatch(&line, fd, &mut writer).await;
         // Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
         // to a bare PrivResponse (untagged), so old hive-c0re callers that
         // deserialise directly to PrivResponse continue to work.
@@ -112,31 +264,54 @@ async fn handle(stream: UnixStream) {
     }
 }
 
-async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
-    match serde_json::from_str::<PrivRequest>(line) {
-        Ok(req) => match exec(req, writer).await {
-            Ok((stdout, stderr)) => PrivResponse {
-                ok: true,
-                stdout,
-                stderr,
-                error: None,
-            },
-            Err(e) => PrivResponse {
-                ok: false,
-                stdout: String::new(),
-                stderr: String::new(),
-                error: Some(format!("{e:#}")),
-            },
+/// Reject a request whose descriptor and operation disagree, in either
+/// direction.
+///
+/// No guessing when the caller didn't say: an op that streams into a
+/// passed descriptor cannot invent one, and an op that takes none must
+/// not silently accept one. Returning the `Err` here drops the
+/// `OwnedFd`, which closes it.
+fn check_fd_agreement(req: &PrivRequest, fd: Option<&OwnedFd>) -> Result<()> {
+    let wants_fd = matches!(req, PrivRequest::SendAgentSnapshotToFd { .. });
+    match (wants_fd, fd.is_some()) {
+        (true, false) => bail!("this operation requires a passed file descriptor, none arrived"),
+        (false, true) => bail!("this operation does not take a passed file descriptor"),
+        _ => Ok(()),
+    }
+}
+
+async fn dispatch(line: &str, fd: Option<OwnedFd>, writer: &mut OwnedWriteHalf) -> PrivResponse {
+    match run(line, fd, writer).await {
+        Ok((stdout, stderr)) => PrivResponse {
+            ok: true,
+            stdout,
+            stderr,
+            error: None,
         },
         Err(e) => PrivResponse {
             ok: false,
             stdout: String::new(),
             stderr: String::new(),
-            error: Some(format!("parse request: {e}")),
+            error: Some(format!("{e:#}")),
         },
     }
 }
 
+/// Parse one request line, check it agrees with the descriptor that
+/// arrived with it, and execute it.
+///
+/// Split out of [`dispatch`] so the three failure modes collapse into one
+/// `Result` instead of three nested matches building the same struct.
+async fn run(
+    line: &str,
+    fd: Option<OwnedFd>,
+    writer: &mut OwnedWriteHalf,
+) -> Result<(String, String)> {
+    let req = serde_json::from_str::<PrivRequest>(line).context("parse request")?;
+    check_fd_agreement(&req, fd.as_ref())?;
+    exec(req, fd, writer).await
+}
+
 /// Write one `PrivEvent::Line` to the client. Best-effort: a write
 /// failure is logged but doesn't abort the running subprocess.
 async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
@@ -159,7 +334,17 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data:
 // One match arm per priv op — a flat 1:1 dispatch table. The length tracks
 // the op count, not complexity; splitting it would just scatter the mapping.
 #[allow(clippy::too_many_lines)]
-async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
+/// Execute one validated request.
+///
+/// `fd` is the descriptor that arrived with this request, already checked
+/// against the operation by [`check_fd_agreement`]: `Some` exactly for
+/// the variants that stream into a caller-supplied descriptor, `None`
+/// for every other operation.
+async fn exec(
+    req: PrivRequest,
+    fd: Option<OwnedFd>,
+    writer: &mut OwnedWriteHalf,
+) -> Result<(String, String)> {
     match req {
         PrivRequest::StartContainer { ref name } => {
             validate_container_name(name)?;
@@ -417,6 +602,26 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
             .await
         }
 
+        PrivRequest::SendAgentSnapshotToFd {
+            ref agent_name,
+            ref snapshot_name,
+            ref parent_snapshot_name,
+        } => {
+            validate_agent_name(agent_name)?;
+            validate_snapshot_name(snapshot_name)?;
+            if let Some(parent) = parent_snapshot_name {
+                validate_snapshot_name(parent)?;
+            }
+            let dest = fd.context("no descriptor to stream into")?;
+            send_agent_snapshot_to_fd(
+                agent_name,
+                snapshot_name,
+                parent_snapshot_name.as_deref(),
+                dest,
+            )
+            .await
+        }
+
         PrivRequest::SyncAgentTmpfiles { ref agents } => sync_agent_tmpfiles(agents).await,
     }
 }
@@ -1424,6 +1629,70 @@ async fn send_agent_snapshot_to_file(
     Ok((dest.display().to_string(), String::new()))
 }
 
+/// `SendAgentSnapshotToFd` — stream a read-only snapshot (optionally
+/// incremental against `parent_name`) straight into a descriptor the
+/// caller passed us.
+///
+/// The network half of the inter-hive migration transport, arranged so
+/// this helper never learns there *is* a network: hive-c0re connects to
+/// the peer's snapshot store, writes the header itself, and hands the
+/// connected socket over. We only ever see "a thing to write bytes into",
+/// which keeps a root process out of any address, protocol or trust
+/// decision — and keeps everyone out of the data path once `btrfs send`
+/// starts, which matters at multi-gigabyte sizes.
+async fn send_agent_snapshot_to_fd(
+    agent_name: &str,
+    snapshot_name: &str,
+    parent_name: Option<&str>,
+    dest: OwnedFd,
+) -> Result<(String, String)> {
+    let snap = snapshot_path(agent_name, snapshot_name);
+    if !snap.exists() {
+        bail!(
+            "snapshot {} does not exist — create it with `subvol snapshot create` first",
+            snap.display()
+        );
+    }
+
+    let mut cmd = Command::new("btrfs");
+    cmd.arg("send");
+    if let Some(parent) = parent_name {
+        let parent_path = snapshot_path(agent_name, parent);
+        if !parent_path.exists() {
+            bail!(
+                "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
+                parent_path.display()
+            );
+        }
+        cmd.arg("-p").arg(&parent_path);
+    }
+    cmd.arg(&snap);
+    cmd.stdout(std::process::Stdio::from(dest));
+    cmd.stderr(std::process::Stdio::piped());
+
+    let out = cmd
+        .spawn()
+        .with_context(|| format!("spawn btrfs send {}", snap.display()))?
+        .wait_with_output()
+        .await
+        .with_context(|| format!("wait on btrfs send {}", snap.display()))?;
+    if !out.status.success() {
+        // Nothing to clean up: the destination isn't ours. A partial
+        // stream is the receiving end's problem, and `btrfs receive`
+        // refuses to commit an incomplete subvolume anyway.
+        bail!(
+            "btrfs send {} failed: {}",
+            snap.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    tracing::info!(
+        agent = %agent_name, snapshot = %snap.display(), parent = ?parent_name,
+        "streamed agent snapshot into a passed descriptor"
+    );
+    Ok((String::new(), String::new()))
+}
+
 /// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent
 /// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
 /// wire doc.
@@ -2217,12 +2486,68 @@ async fn sync_agent_tmpfiles(agents: &[String]) -> Result<(String, String)> {
 #[cfg(test)]
 mod tests {
     use super::{
-        PAUSED_MARKER_FILE, limits_dropin_body, redact_password_line, remove_marker_in,
-        write_state_file_nofollow,
+        OwnedFd, PAUSED_MARKER_FILE, PrivRequest, check_fd_agreement, limits_dropin_body,
+        redact_password_line, remove_marker_in, write_state_file_nofollow,
     };
     use std::path::PathBuf;
     use std::sync::atomic::{AtomicU32, Ordering};
 
+    /// A request that streams into a caller-supplied descriptor.
+    fn fd_taking_request() -> PrivRequest {
+        PrivRequest::SendAgentSnapshotToFd {
+            agent_name: "atlas".to_owned(),
+            snapshot_name: "hive-migrate".to_owned(),
+            parent_snapshot_name: None,
+        }
+    }
+
+    /// A real descriptor — `/dev/null` rather than a fake, so the drop
+    /// that closes it on a rejection path is genuinely exercised.
+    fn some_fd() -> OwnedFd {
+        std::fs::File::open("/dev/null")
+            .expect("open /dev/null")
+            .into()
+    }
+
+    /// An op that streams into a passed descriptor cannot invent one:
+    /// falling back to anything (a temp file, the response socket) would
+    /// send an agent's state somewhere the caller never asked for.
+    #[test]
+    fn an_fd_taking_op_without_a_descriptor_is_rejected() {
+        let err = check_fd_agreement(&fd_taking_request(), None)
+            .expect_err("no descriptor arrived, so this must not proceed");
+        let msg = format!("{err:#}");
+        assert!(msg.contains("requires a passed file descriptor"), "{msg}");
+    }
+
+    /// The mirror case: a descriptor sent alongside an op that takes
+    /// none is a protocol error, not something to ignore. Returning the
+    /// error drops the `OwnedFd`, which closes it — the alternative
+    /// leaks one descriptor per stray request in a long-lived root
+    /// process.
+    #[test]
+    fn a_descriptor_sent_to_an_op_that_takes_none_is_rejected() {
+        let fd = some_fd();
+        let err = check_fd_agreement(&PrivRequest::DaemonReload, Some(&fd))
+            .expect_err("an unexpected descriptor must not be silently ignored");
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("does not take a passed file descriptor"),
+            "{msg}"
+        );
+    }
+
+    /// Both agreeing combinations pass, so the check rejects mismatches
+    /// rather than descriptors in general.
+    #[test]
+    fn agreeing_combinations_are_accepted() {
+        let fd = some_fd();
+        check_fd_agreement(&fd_taking_request(), Some(&fd))
+            .expect("an fd-taking op with its descriptor is the normal case");
+        check_fd_agreement(&PrivRequest::DaemonReload, None)
+            .expect("every ordinary request arrives without a descriptor");
+    }
+
     /// An unset weight is "not configured": the drop-in must come out
     /// byte-identical to the pre-weights two-setting body, so a hive-c0re
     /// older than this field can't change what lands on disk.

hivectl/src/cli.rs

@@ -620,10 +620,10 @@ pub enum SnapshotCmd {
         /// Snapshot label passed to `subvol snapshot create --label`.
         label: String,
     },
-    /// Export a snapshot to a local file via `btrfs send` (the local-file
-    /// half of inter-hive migration transport; the cross-hive `ssh ...
-    /// btrfs receive` leg isn't wired up yet). Also useful standalone as a
-    /// point-in-time backup: a full send with no `--parent` produces a
+    /// Export a snapshot to a local file via `btrfs send` — the
+    /// local-file half of the inter-hive migration transport (`push`
+    /// is the network half). Also useful standalone as a point-in-time
+    /// backup: a full send with no `--parent` produces a
     /// self-contained archive of the snapshot.
     Send {
         /// Snapshot label passed to `subvol snapshot create --label`.
@@ -638,4 +638,25 @@ pub enum SnapshotCmd {
         #[arg(long)]
         dest: String,
     },
+    /// Stream a snapshot to the swarm's snapshot store over the
+    /// WireGuard mesh — the network half of the migration transport.
+    ///
+    /// Nothing is staged locally: `btrfs send` writes straight into the
+    /// connection, so a multi-gigabyte agent needs no scratch space on
+    /// this host. The mesh is the authentication (cryptokey routing
+    /// binds the sender's address to its key), so there is no credential
+    /// to pass here.
+    ///
+    /// There is no destination argument: a swarm has one store, read
+    /// from `services.hyperhive.swarm.snapshotStore`.
+    Push {
+        /// Snapshot label passed to `subvol snapshot create --label`.
+        label: String,
+        /// Optional parent snapshot label for an incremental send
+        /// (`btrfs send -p`) — must be an existing, older snapshot of the
+        /// same agent, and must already be present on the receiver.
+        /// Omit for a full send.
+        #[arg(long)]
+        parent: Option<String>,
+    },
 }

hivectl/src/subvol.rs

@@ -44,6 +44,9 @@ pub(crate) async fn dispatch_subvol(socket: &Path, name: &str, cmd: SubvolCmd) -
                 parent,
                 dest,
             } => subvol_snapshot_send(socket, name, &label, parent.as_deref(), &dest).await,
+            SnapshotCmd::Push { label, parent } => {
+                subvol_snapshot_push(socket, name, &label, parent.as_deref()).await
+            }
         },
     }
 }
@@ -217,3 +220,37 @@ async fn subvol_snapshot_send(
     )
     .await
 }
+
+/// `subvol snapshot push <agent> <label> [--parent <label>]` — stream a
+/// snapshot to the swarm's snapshot store over the mesh.
+///
+/// The network sibling of `send`. Nothing is staged on this host, so
+/// there is no path to print: success is silent apart from the
+/// confirmation below. No destination argument — a swarm has one store,
+/// and the daemon reads its address from the host config.
+async fn subvol_snapshot_push(
+    socket: &Path,
+    name: &str,
+    label: &str,
+    parent: Option<&str>,
+) -> Result<()> {
+    daemon_request(
+        socket,
+        hive_host_sock::HostRequest::PushSnapshot {
+            name: crate::util::parse_ident(name)?,
+            label: label.to_owned(),
+            parent: parent.map(str::to_owned),
+        },
+        "snapshot push",
+    )
+    .await?;
+    match parent {
+        Some(p) => {
+            println!(
+                "pushed {name} snapshot {label:?} (incremental from {p:?}) to the swarm store"
+            );
+        }
+        None => println!("pushed {name} snapshot {label:?} (full) to the swarm store"),
+    }
+    Ok(())
+}

nix/host-modules/hive-c0re/environment.nix

@@ -149,6 +149,18 @@ in
       # or no gatewayHost is set (no browser-reachable matrix vhost).
       HIVE_MATRIX_PUBLIC_URL = "https://${config.services.hyperhive.matrix.gatewayHost}/";
     }
+// lib.optionalAttrs (config.services.hyperhive.swarm.snapshotStore.address != null) {
+  # `host:port` of the swarm's single snapshot store, for pushing agent
+  # snapshots (hive-c0re::snapshot_push). One per swarm, not one per
+  # peer — the receiver keys destinations by agent so a migrating agent
+  # keeps one incremental chain. Absent when no store is configured, and
+  # a push then fails naming the option rather than guessing.
+  HYPERHIVE_SNAPSHOT_STORE =
+    let
+      s = config.services.hyperhive.swarm.snapshotStore;
+    in
+    "${s.address}:${toString s.port}";
+}
 // lib.optionalAttrs (config.services.hyperhive.swarm.peers != { }) {
   # Peer hives serialised as a JSON array of {domain, cert_fingerprint,
   # wireguard_address?} objects. Consumed by hive-agent::identity::peers()

nix/host-modules/swarm.nix

@@ -105,6 +105,7 @@
               are silently excluded from `wg-hive`).
             '';
           };
+
         };
       }
     );
@@ -125,4 +126,41 @@
     '';
   };
 
+  options.services.hyperhive.swarm.snapshotStore = {
+    address = lib.mkOption {
+      type = lib.types.nullOr lib.types.str;
+      default = null;
+      example = "10.100.0.1";
+      description = ''
+        Mesh address of the swarm's snapshot store — the single
+        `btrfs receive` endpoint every hive in this swarm pushes agent
+        snapshots to. Bare IP, no prefix.
+
+        There is exactly **one** store per swarm, not one per peer: the
+        receiver keys destinations by *agent*, so an agent that migrates
+        between hives keeps a single unbroken incremental chain. Per-hive
+        stores would split that chain in two, which is the case the store
+        exists to serve.
+
+        Null means this swarm has no store configured, and pushing fails
+        saying so rather than guessing an address. Set it on every hive
+        that pushes; the receiving host separately sets
+        `services.hyperhive.snapshotStore.enable`.
+      '';
+    };
+
+    port = lib.mkOption {
+      type = lib.types.port;
+      default = 51821;
+      description = ''
+        TCP port the swarm's snapshot store listens on. Must match the
+        receiving host's `services.hyperhive.snapshotStore.port`.
+
+        Defaulted (unlike `address`) because it is a shared convention
+        both sides read from the same option docs — whereas an address
+        is deployment-specific and cannot be guessed.
+      '';
+    };
+  };
+
 }