feat(#2391): btrfs SnapshotAgentSubvolume/DeleteAgentSnapshot priv ops

Adds the first missing piece from #2391's migration-gaps list: a
read-only btrfs snapshot priv op so hivectl migrate can freeze a
consistent point-in-time copy of an agent's state subvolume for
btrfs send, without stopping the live agent.

- PrivRequest::SnapshotAgentSubvolume / DeleteAgentSnapshot (hive-sh4re)
- hive-priv handlers: btrfs subvolume snapshot -r / delete, sibling
  dot-prefixed path (<AGENT_STATE_ROOT>/.<agent>.snapshot.<label>)
- hive-c0re::priv_client wrappers
- hivectl subvol snapshot / delete-snapshot verbs (no agent stop needed
  — btrfs snapshots are atomic against a live subvolume)

Does not yet wire actual btrfs send/receive or the hivectl migrate
verb — those stay tracked on #2391 as separate follow-up pieces.

hive-c0re/src/bin/hivectl.rs

@@ -620,6 +620,24 @@ enum SubvolCmd {
         #[arg(long)]
         yes: bool,
     },
+    /// Create a read-only snapshot of an agent's state subvolume — the
+    /// first step of the (in-progress) inter-hive migration path, or a
+    /// manual point-in-time backup. Agent must already be a subvolume
+    /// (`subvol upgrade` first). Prints the snapshot's host path.
+    Snapshot {
+        /// Agent name (e.g. `damocles`, `iris`).
+        name: String,
+        /// Snapshot label (`[A-Za-z0-9_.-]`); defaults to a UTC timestamp.
+        #[arg(long)]
+        label: Option<String>,
+    },
+    /// Delete a snapshot created by `subvol snapshot`.
+    DeleteSnapshot {
+        /// Agent name the snapshot belongs to.
+        name: String,
+        /// Snapshot label passed to `subvol snapshot --label`.
+        label: String,
+    },
 }
 
 #[tokio::main]
@@ -693,6 +711,10 @@ async fn main() -> Result<()> {
         Cmd::Restart { scope, graceful } => restart(&socket, scope.to_scope(), graceful).await,
         Cmd::Subvol { cmd } => match cmd {
             SubvolCmd::Upgrade { name, yes } => subvol_upgrade(&socket, &name, yes).await,
+            SubvolCmd::Snapshot { name, label } => subvol_snapshot(&name, label).await,
+            SubvolCmd::DeleteSnapshot { name, label } => {
+                subvol_delete_snapshot(&name, &label).await
+            }
         },
         Cmd::Choom {
             name,
@@ -1703,6 +1725,38 @@ async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> {
     Ok(())
 }
 
+/// `subvol snapshot <agent> [--label <label>]` — create a read-only btrfs
+/// snapshot of an agent's state subvolume. Unlike `upgrade`, this does NOT
+/// stop the agent: btrfs snapshots are atomic + consistent to take against a
+/// live subvolume. Default label is a unix-timestamp so repeated calls don't
+/// collide without the caller having to think of a name.
+async fn subvol_snapshot(name: &str, label: Option<String>) -> Result<()> {
+    if !agent_exists(name)? {
+        bail!("no agent named {name:?} (no state dir under the agents root)");
+    }
+    let label = label.unwrap_or_else(|| {
+        let secs = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |d| d.as_secs());
+        format!("snap-{secs}")
+    });
+    let path = hive_c0re::priv_client::snapshot_agent_subvolume(name, &label)
+        .await
+        .with_context(|| format!("snapshot {name} state subvolume (label {label:?})"))?;
+    println!("{path}");
+    Ok(())
+}
+
+/// `subvol delete-snapshot <agent> <label>` — remove a snapshot created by
+/// `subvol snapshot`.
+async fn subvol_delete_snapshot(name: &str, label: &str) -> Result<()> {
+    hive_c0re::priv_client::delete_agent_snapshot(name, label)
+        .await
+        .with_context(|| format!("delete {name} snapshot (label {label:?})"))?;
+    println!("deleted snapshot {label:?} for {name}");
+    Ok(())
+}
+
 /// Render a hive-wide stop/start response: one `<verb>: <name>` line per
 /// touched container, then surface any aggregated per-target failure as a
 /// non-zero exit. `verb` is the past-tense word printed per item

hive-c0re/src/priv_client.rs

@@ -439,6 +439,41 @@ pub async fn upgrade_agent_subvolume(agent_name: &str) -> Result<()> {
     .await?)
 }
 
+/// Create a read-only btrfs snapshot of an agent's state subvolume (via
+/// hive-priv as root) — the first step of `hivectl migrate`'s send/receive
+/// path. Returns the snapshot's absolute host path. Fails if the agent's
+/// state root isn't a subvolume yet, or a snapshot with the same
+/// `snapshot_name` already exists.
+///
+/// # Errors
+/// Returns an error if the hive-priv call fails, the state dir isn't a
+/// btrfs subvolume, or the snapshot already exists.
+pub async fn snapshot_agent_subvolume(agent_name: &str, snapshot_name: &str) -> Result<String> {
+    let (stdout, _) = check(
+        call(&PrivRequest::SnapshotAgentSubvolume {
+            agent_name: agent_name.to_owned(),
+            snapshot_name: snapshot_name.to_owned(),
+        })
+        .await?,
+    )?;
+    Ok(stdout)
+}
+
+/// Delete a previously-created read-only agent-state snapshot (cleanup
+/// counterpart to [`snapshot_agent_subvolume`]). No-op if the snapshot
+/// doesn't exist. Via hive-priv as root.
+///
+/// # Errors
+/// Returns an error if the hive-priv call fails or the underlying
+/// `btrfs subvolume delete` fails.
+pub async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<()> {
+    ok(call(&PrivRequest::DeleteAgentSnapshot {
+        agent_name: agent_name.to_owned(),
+        snapshot_name: snapshot_name.to_owned(),
+    })
+    .await?)
+}
+
 /// Write `/etc/tmpfiles.d/hyperhive-agents.conf` for `agents` (logical names,
 /// e.g. `"atlas"`) and immediately apply it with `systemd-tmpfiles --create`.
 /// See [`PrivRequest::SyncAgentTmpfiles`] for the full semantics.

hive-priv/src/main.rs

@@ -365,6 +365,24 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String,
             upgrade_agent_subvolume(agent_name).await
         }
 
+        PrivRequest::SnapshotAgentSubvolume {
+            ref agent_name,
+            ref snapshot_name,
+        } => {
+            validate_agent_name(agent_name)?;
+            validate_credential_name(snapshot_name)?;
+            snapshot_agent_subvolume(agent_name, snapshot_name).await
+        }
+
+        PrivRequest::DeleteAgentSnapshot {
+            ref agent_name,
+            ref snapshot_name,
+        } => {
+            validate_agent_name(agent_name)?;
+            validate_credential_name(snapshot_name)?;
+            delete_agent_snapshot(agent_name, snapshot_name).await
+        }
+
         PrivRequest::SyncAgentTmpfiles { ref agents } => sync_agent_tmpfiles(agents).await,
     }
 }
@@ -1059,6 +1077,86 @@ async fn upgrade_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
     ))
 }
 
+/// Derive a snapshot's path from the agent name + label: a dot-prefixed
+/// sibling of the agent's state root so it can never collide with a real
+/// agent directory (`validate_agent_name` rejects dot-prefixed names).
+fn snapshot_path(agent_name: &str, snapshot_name: &str) -> PathBuf {
+    PathBuf::from(AGENT_STATE_ROOT).join(format!(".{agent_name}.snapshot.{snapshot_name}"))
+}
+
+/// `SnapshotAgentSubvolume` — create a read-only btrfs snapshot of an
+/// agent's state subvolume, for `btrfs send` to stream from during
+/// inter-hive migration. See the wire doc.
+async fn snapshot_agent_subvolume(
+    agent_name: &str,
+    snapshot_name: &str,
+) -> Result<(String, String)> {
+    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
+    if !is_btrfs_subvolume(&agent_root) {
+        bail!(
+            "{} is not a btrfs subvolume — nothing to snapshot (run `hivectl subvol upgrade` first)",
+            agent_root.display()
+        );
+    }
+    let snap = snapshot_path(agent_name, snapshot_name);
+    if snap.exists() {
+        bail!(
+            "snapshot {} already exists — delete it first or pick a different name",
+            snap.display()
+        );
+    }
+    let out = Command::new("btrfs")
+        .args(["subvolume", "snapshot", "-r"])
+        .arg(&agent_root)
+        .arg(&snap)
+        .output()
+        .await
+        .with_context(|| {
+            format!(
+                "spawn btrfs subvolume snapshot -r {} {}",
+                agent_root.display(),
+                snap.display()
+            )
+        })?;
+    if !out.status.success() {
+        bail!(
+            "btrfs subvolume snapshot -r {} {} failed: {}",
+            agent_root.display(),
+            snap.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    tracing::info!(
+        agent = %agent_name, snapshot = %snap.display(),
+        "created read-only agent state snapshot"
+    );
+    Ok((snap.display().to_string(), String::new()))
+}
+
+/// `DeleteAgentSnapshot` — delete a previously-created read-only snapshot.
+/// No-op if the path doesn't exist. See the wire doc.
+async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<(String, String)> {
+    let snap = snapshot_path(agent_name, snapshot_name);
+    if !snap.exists() {
+        return Ok((String::new(), String::new()));
+    }
+    let out = Command::new("btrfs")
+        .args(["subvolume", "delete"])
+        .arg(&snap)
+        .output()
+        .await
+        .with_context(|| format!("spawn btrfs subvolume delete {}", snap.display()))?;
+    if !out.status.success() {
+        bail!(
+            "btrfs subvolume delete {} failed: {}",
+            snap.display(),
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    tracing::info!(agent = %agent_name, snapshot = %snap.display(), "deleted agent state snapshot");
+    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.

hive-sh4re/src/priv_proto.rs

@@ -580,6 +580,38 @@ pub enum PrivRequest {
         agent_name: String,
     },
 
+    /// Create a read-only snapshot of an agent's state subvolume
+    /// (`btrfs subvolume snapshot -r <agent_root> <snapshot_path>`). Used as
+    /// the first step of inter-hive migration (`hivectl migrate`): freezing
+    /// a consistent point-in-time copy that `btrfs send` can stream from
+    /// while the source subvolume keeps running underneath the live agent.
+    ///
+    /// The snapshot is created as a sibling of the agent's state root
+    /// (`<AGENT_STATE_ROOT>/.<agent_name>.snapshot.<snapshot_name>`, dot-prefixed
+    /// so it never collides with a real agent name) and its path is returned
+    /// verbatim in the response's `stdout`. Fails if the agent's state root
+    /// isn't a btrfs subvolume (nothing to snapshot) or a snapshot with the
+    /// same name already exists. Requires root.
+    SnapshotAgentSubvolume {
+        /// Logical agent name (validated by `validate_agent_name`).
+        agent_name: String,
+        /// Snapshot label (validated like a credential name:
+        /// non-empty `[A-Za-z0-9_.-]`); becomes part of the snapshot path.
+        snapshot_name: String,
+    },
+
+    /// Delete a previously-created read-only snapshot
+    /// (`btrfs subvolume delete <snapshot_path>`). Cleanup counterpart to
+    /// [`PrivRequest::SnapshotAgentSubvolume`] — called once a migration's
+    /// `btrfs send` has completed (or aborted) and the frozen copy is no
+    /// longer needed. No-op if the snapshot path doesn't exist. Requires root.
+    DeleteAgentSnapshot {
+        /// Logical agent name (validated by `validate_agent_name`).
+        agent_name: String,
+        /// Snapshot label, same validation as `SnapshotAgentSubvolume`.
+        snapshot_name: 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