From 12ae553ca1d035948613fe4429022ab586be8872 Mon Sep 17 00:00:00 2001
From: leetcrypt <leetcrypt@users.noreply.github.com>
Date: Sat, 27 Jun 2026 15:21:53 -0700
Subject: [PATCH] =?UTF-8?q?feat(operator):=20VM=20marketplace=20=E2=80=94?=
 =?UTF-8?q?=20Phase=20A=20registry=20+=20Phase=20B=20trading?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase A: host-global VM registry (~/.hh/registry.json) joining opaque
snapshots to their .hh-agent manifests. New src/registry.rs (serde_json),
one Entry per saved VM with cached purpose/status/todo scraped from the
live container at save time. Reconcile-on-read self-evicts pruned images.
/sbx browse TUI lister + operator-side `registry list|show` reader.

Phase B: trading/skill VMs. Entry gains shareable/tags/share_path +
publish/get/list_shareable helpers. New TUI verbs /sbx publish <label>
[tag...], /sbx catalog @user, /sbx pull @user <label>. Wire protocol adds
_sbx:catreq/catalog/pullreq frames (parse_sbx). Receiver auto-loads a
received hh-snap-*.tar, reads the in-image manifest, and self-registers
(kept shareable for re-trade). Reuses the existing E2E /send + ft.rs
streaming transport.

podman gotcha fixed: `podman load` prints registry-qualified
`localhost/hh-snap:<label>` vs docker's bare tag — parse_loaded_tag now
finds the hh-snap: marker anywhere in the line (unit-tested both shapes).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 cmd_chat/operator/__main__.py |  61 ++++++
 hh/src/app.rs                 | 386 ++++++++++++++++++++++++++++++++-
 hh/src/main.rs                |   1 +
 hh/src/net.rs                 |  82 ++++++-
 hh/src/registry.rs            | 387 ++++++++++++++++++++++++++++++++++
 hh/src/sbx.rs                 | 109 ++++++++++
 6 files changed, 1019 insertions(+), 7 deletions(-)
 create mode 100644 hh/src/registry.rs

diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py
index 1ebc269..b83138d 100644
--- a/cmd_chat/operator/__main__.py
+++ b/cmd_chat/operator/__main__.py
@@ -151,6 +151,14 @@ def _build_parser() -> argparse.ArgumentParser:
     mf.add_argument("--replace", action="store_true", help="replace state lists instead of appending")
     mf.add_argument("--session", default=None)
 
+    # ── VM registry (discovery: what saved VMs exist + what each is for) ──
+    rg = sub.add_parser("registry",
+                        help="browse the host-global saved-VM registry (~/.hh/registry.json)")
+    rg.add_argument("action", choices=["list", "show"], nargs="?", default="list")
+    rg.add_argument("--label", default=None, help="for show: which VM")
+    rg.add_argument("--repo", default=None, help="for list: filter to this origin repo")
+    rg.add_argument("--json", action="store_true", help="machine-readable output")
+
     for v in ("roster", "status", "down"):
         sub.add_parser(v).add_argument("--session", default=None)
     return ap
@@ -396,6 +404,57 @@ def _run_manifest(args) -> int:
     return 0 if resp.get("ok") else 1
 
 
+def _run_registry(args) -> int:
+    """Read the host-global VM registry written by the TUI's `/sbx save`. Standalone
+    (no daemon/room needed) so an agent can survey available work before joining."""
+    path = Path.home() / ".hh" / "registry.json"
+    try:
+        data = json.loads(path.read_text())
+    except FileNotFoundError:
+        print("no VM registry yet (~/.hh/registry.json) — save a VM with `/sbx save`",
+              file=sys.stderr)
+        return 0 if args.action == "list" else 1
+    except (json.JSONDecodeError, OSError) as e:
+        print(f"registry unreadable: {e}", file=sys.stderr)
+        return 1
+
+    entries = list((data.get("entries") or {}).values())
+    entries.sort(key=lambda e: e.get("created_unix", 0), reverse=True)
+
+    if args.action == "show":
+        if not args.label:
+            print("usage: registry show --label <name>", file=sys.stderr)
+            return 2
+        match = next((e for e in entries if e.get("label") == args.label), None)
+        if not match:
+            print(f"no saved VM '{args.label}'", file=sys.stderr)
+            return 1
+        print(json.dumps(match, indent=2))
+        return 0
+
+    if args.repo:
+        entries = [e for e in entries if e.get("repo") == args.repo]
+    if args.json:
+        print(json.dumps(entries, indent=2))
+        return 0
+    if not entries:
+        print("no saved VMs yet")
+        return 0
+    for e in entries:
+        todo = e.get("todo") or []
+        nxt = f" · next: {todo[0]}" if todo else ""
+        # Phase B: flag tradeable VMs so an agent knows what it can `/sbx pull`.
+        share = ""
+        if e.get("shareable"):
+            tags = e.get("tags") or []
+            share = " · shareable" + (f" {{{', '.join(tags)}}}" if tags else "")
+        print(f"• {e.get('label')} · [{e.get('status') or '?'}] · "
+              f"{e.get('purpose') or '—'}{nxt}{share}")
+        print(f"    backend={e.get('backend')} artifact={e.get('artifact_ref')} "
+              f"by={e.get('created_by')}")
+    return 0
+
+
 def _run_simple(args, op: str) -> int:
     resp = _client_request(args, {"op": op})
     print(json.dumps(resp))
@@ -427,6 +486,8 @@ def main(argv: list[str] | None = None) -> int:
         return _run_spawn(args)
     if verb == "manifest":
         return _run_manifest(args)
+    if verb == "registry":
+        return _run_registry(args)
     if verb == "screen":
         return _run_screen(args)
     if verb == "watch":
diff --git a/hh/src/app.rs b/hh/src/app.rs
index 1ae3979..1cdc0e4 100644
--- a/hh/src/app.rs
+++ b/hh/src/app.rs
@@ -3,6 +3,7 @@
 use crate::ft;
 use crate::layout::{Dir, Layout, Resize};
 use crate::net::{self, Session};
+use crate::registry;
 use crate::sbx;
 use crate::theme::Theme;
 use crate::ui;
@@ -144,9 +145,41 @@ pub enum Net {
         by: String,
         vm: String,
     },
+    /// A peer asked for our shareable-VM catalog (`/sbx catalog @me`). `by` is the
+    /// server-authenticated requester; `to` is whom the request named (we act only
+    /// if it's us). We reply with a `_sbx:catalog` frame of our published VMs.
+    SbxCatReq {
+        by: String,
+        to: String,
+    },
+    /// A peer answered our `/sbx catalog @them` with their published-VM slice.
+    SbxCatalog {
+        by: String,
+        to: String,
+        items: Vec<CatalogItem>,
+    },
+    /// A peer asked us to hand over a published VM (`/sbx pull @me <label>`). If we
+    /// published `label`, we `/send` them its portable artifact.
+    SbxPullReq {
+        by: String,
+        to: String,
+        label: String,
+    },
     Closed,
 }
 
+/// One row of a peer's shareable-VM catalog, as it rides the `_sbx:catalog` frame.
+/// A trimmed view of `registry::Entry` — enough for the requester to decide what
+/// to `/sbx pull`, without leaking host-local paths.
+#[derive(Clone, Debug, Default)]
+pub struct CatalogItem {
+    pub label: String,
+    pub purpose: String,
+    pub status: String,
+    pub tags: Vec<String>,
+    pub size_bytes: Option<u64>,
+}
+
 pub struct SbxView {
     pub parser: vt100::Parser,
     pub backend: String,
@@ -567,6 +600,34 @@ impl App {
                     ));
                 }
             }
+            // catreq/pullreq need to send frames + read disk → handled in the run
+            // loop; they never reach here. Arms kept for match exhaustiveness.
+            Net::SbxCatReq { .. } | Net::SbxPullReq { .. } => {}
+            Net::SbxCatalog { by, to, items } => {
+                // Only render a catalog that was sent to us (everyone sees the
+                // broadcast, but it's a private reply to one requester).
+                if to != self.me {
+                    return;
+                }
+                if items.is_empty() {
+                    self.sys(format!("† {by} has no shareable VMs"));
+                } else {
+                    self.sys(format!("† {by}'s shareable VMs ({}):", items.len()));
+                    for it in items {
+                        let purpose = if it.purpose.is_empty() { "—" } else { it.purpose.as_str() };
+                        let status = if it.status.is_empty() { "?" } else { it.status.as_str() };
+                        let mut line = format!("  • {} · [{}] · {}", it.label, status, purpose);
+                        if !it.tags.is_empty() {
+                            line.push_str(&format!(" · {{{}}}", it.tags.join(", ")));
+                        }
+                        if let Some(sz) = it.size_bytes {
+                            line.push_str(&format!(" · {}", crate::ft::human(sz as usize)));
+                        }
+                        self.sys(line);
+                        self.sys(format!("      pull with `/sbx pull @{by} {}`", it.label));
+                    }
+                }
+            }
             Net::Closed => {
                 self.connected = false;
                 self.sys("connection closed — press Ctrl-R to reconnect");
@@ -1513,8 +1574,38 @@ pub async fn run(params: net::ConnParams, mut session: Session, mut theme: Theme
                                         };
                                     });
                                 }
+                                // A pulled OCI snapshot (`hh-snap-<label>.tar`) loads
+                                // into the local engine and self-registers — the
+                                // manifest rides inside the image, so the receiver's
+                                // `/sbx browse`/`load` light up without a human briefing.
+                                let is_snap_tar = ext.as_deref() == Some("tar")
+                                    && path
+                                        .file_name()
+                                        .and_then(|s| s.to_str())
+                                        .is_some_and(|n| n.starts_with("hh-snap-"));
+                                if is_snap_tar {
+                                    let tx = app_tx.clone();
+                                    let tar = path.clone();
+                                    tokio::spawn(async move {
+                                        let res = tokio::task::spawn_blocking(move || {
+                                            register_received_snapshot(&tar)
+                                        })
+                                        .await;
+                                        let _ = match res {
+                                            Ok(Ok(msg)) => tx.send(Net::Sys(msg)),
+                                            Ok(Err(e)) => tx.send(Net::Sys(format!(
+                                                "(received snapshot not imported: {e:#})"
+                                            ))),
+                                            Err(e) => tx.send(Net::Err(format!("import task: {e}"))),
+                                        };
+                                    });
+                                }
+                                // Bridge any *other* received file into a hosted
+                                // sandbox so the room can use it from the shared
+                                // shell. Snapshot archives are skipped — they're
+                                // loaded as images above, not dropped into a shell.
                                 if let Some((be, name)) = &broker_meta {
-                                    if !matches!(be, sbx::Backend::Local) {
+                                    if !is_snap_tar && !matches!(be, sbx::Backend::Local) {
                                         let (be, name) = (*be, name.clone());
                                         let run_user = sbx::run_user_for(
                                             be,
@@ -1593,6 +1684,51 @@ pub async fn run(params: net::ConnParams, mut session: Session, mut theme: Theme
                                 )),
                             }
                         }
+                        // A peer asked for our shareable-VM catalog — reply (only
+                        // if the request named us) with our published slice.
+                        Net::SbxCatReq { by, to } => {
+                            if to == app.me {
+                                let items: Vec<serde_json::Value> = registry::list_shareable()
+                                    .into_iter()
+                                    .map(|e| json!({
+                                        "label": e.label, "purpose": e.purpose,
+                                        "status": e.status, "tags": e.tags, "size": e.size_bytes,
+                                    }))
+                                    .collect();
+                                send_frame(&out_tx, &session.room, json!({
+                                    "_sbx":"catalog","to": by,"items": items
+                                }));
+                            }
+                        }
+                        // A peer asked us to hand over a published VM — if we
+                        // actually published it (and the artifact's still on disk),
+                        // `/send` it to them; their ft path auto-imports + registers.
+                        Net::SbxPullReq { by, to, label } => {
+                            if to == app.me {
+                                match registry::get(&label) {
+                                    Some(e)
+                                        if e.shareable
+                                            && !e.share_path.is_empty()
+                                            && std::path::Path::new(&e.share_path).exists() =>
+                                    {
+                                        app.sys(format!(
+                                            "† {by} is pulling ‘{label}’ — sending it over…"
+                                        ));
+                                        let path = e.share_path.clone();
+                                        offer_payload(
+                                            &mut app, &mut send_seq, &out_tx, &session.room,
+                                            &app_tx, &path, Some(&by),
+                                        );
+                                    }
+                                    Some(_) => app.sys(format!(
+                                        "† {by} asked for ‘{label}’ but it isn't published — `/sbx publish {label}` first"
+                                    )),
+                                    None => app.sys(format!(
+                                        "† {by} asked for ‘{label}’ but no such saved VM exists here"
+                                    )),
+                                }
+                            }
+                        }
                         other => app.apply(other),
                     }
                 }
@@ -2178,11 +2314,20 @@ fn handle_command(
                         app.sys("multipass must power off to snapshot — stopping the shared shell, then saving (reload it with `/sbx load`)");
                     }
                     let (tx, lbl) = (app_tx.clone(), label.clone());
+                    let created_by = app.me.clone();
                     tokio::spawn(async move {
-                        let res = tokio::task::spawn_blocking(move || sbx::save_state(be, &name, &label, local)).await;
+                        let res = tokio::task::spawn_blocking(move || {
+                            let desc = sbx::save_state(be, &name, &label, local)?;
+                            // Index the snapshot in the host-global registry, caching
+                            // its .hh-agent manifest summary. Best-effort: an index
+                            // hiccup must never fail the save the user just asked for.
+                            register_saved_snapshot(be, &name, &label, &created_by);
+                            Ok::<String, anyhow::Error>(desc)
+                        })
+                        .await;
                         let _ = match res {
                             Ok(Ok(desc)) => tx.send(Net::Sys(format!(
-                                "† saved sandbox → {desc} · reload with `/sbx load {lbl}`"))),
+                                "† saved sandbox → {desc} · reload with `/sbx load {lbl}` · `/sbx browse`"))),
                             Ok(Err(e)) => tx.send(Net::Err(format!("save failed: {e}"))),
                             Err(e) => tx.send(Net::Err(format!("save task: {e}"))),
                         };
@@ -2298,6 +2443,97 @@ fn handle_command(
                     };
                 });
             }
+            Some("browse") | Some("registry") => {
+                // The discovery layer: every saved VM with WHAT it's for + where the
+                // work stands, read from the host-global registry. Reconciled against
+                // the engine so pruned snapshots drop out (no passive accumulation).
+                let tx = app_tx.clone();
+                tokio::spawn(async move {
+                    let res = tokio::task::spawn_blocking(|| {
+                        registry::list_reconciled(oci_image_exists)
+                    })
+                    .await;
+                    match res {
+                        Ok(entries) if !entries.is_empty() => {
+                            let _ = tx.send(Net::Sys(format!("saved VMs ({}):", entries.len())));
+                            for e in entries {
+                                let purpose = if e.purpose.is_empty() { "—" } else { e.purpose.as_str() };
+                                let status = if e.status.is_empty() { "?" } else { e.status.as_str() };
+                                let mut line = format!("  • {} · [{}] · {}", e.label, status, purpose);
+                                let nt = e.next_todo();
+                                if !nt.is_empty() {
+                                    line.push_str(&format!(" · next: {nt}"));
+                                }
+                                let _ = tx.send(Net::Sys(line));
+                                let _ = tx.send(Net::Sys(format!(
+                                    "      load with `/sbx load {}`", e.label)));
+                            }
+                        }
+                        Ok(_) => {
+                            let _ = tx.send(Net::Sys(
+                                "no saved VMs yet — `/sbx save [label]` records one with its manifest".into()));
+                        }
+                        Err(e) => {
+                            let _ = tx.send(Net::Err(format!("browse task: {e}")));
+                        }
+                    }
+                });
+            }
+            Some("publish") => {
+                // Mark a saved VM shareable so peers can pull it. Ensures a
+                // portable artifact exists (exports an image-backed snap to a
+                // `.tar`), then flips the registry entry's `shareable` flag.
+                let args: Vec<String> = p.map(str::to_string).collect();
+                let mut pos = args.iter().filter(|a| !a.starts_with('-'));
+                match pos.next() {
+                    None => app.sys(
+                        "usage: /sbx publish <label> [tag…]  (mark a saved VM shareable so peers can /sbx pull it)",
+                    ),
+                    Some(label) => {
+                        let label = label.clone();
+                        let tags: Vec<String> = pos.cloned().collect();
+                        app.sys(format!("publishing '{label}'…"));
+                        let tx = app_tx.clone();
+                        tokio::spawn(async move {
+                            let res = tokio::task::spawn_blocking(move || {
+                                publish_snapshot(&label, &tags)
+                            })
+                            .await;
+                            let _ = match res {
+                                Ok(Ok(msg)) => tx.send(Net::Sys(msg)),
+                                Ok(Err(e)) => tx.send(Net::Err(format!("publish failed: {e:#}"))),
+                                Err(e) => tx.send(Net::Err(format!("publish task: {e}"))),
+                            };
+                        });
+                    }
+                }
+            }
+            Some("catalog") => {
+                // Ask a peer what VMs they offer. They reply with a `_sbx:catalog`
+                // frame (handled in apply → printed). Leading '@' optional.
+                match p.next().map(|s| s.trim_start_matches('@')) {
+                    Some(u) if !u.is_empty() => {
+                        send_frame(out_tx, room, json!({"_sbx":"catreq","to": u}));
+                        app.sys(format!("requested {u}'s shareable VMs…"));
+                    }
+                    _ => app.sys("usage: /sbx catalog @<user>  (ask a peer which VMs they offer)"),
+                }
+            }
+            Some("pull") => {
+                // Pull a published VM from a peer: send a pullreq; they `/send` the
+                // artifact and our ft path auto-imports + registers it on /accept.
+                let who = p.next().map(|s| s.trim_start_matches('@').to_string());
+                let label = p.next().map(str::to_string);
+                match (who, label) {
+                    (Some(u), Some(l)) if !u.is_empty() && !l.is_empty() => {
+                        send_frame(out_tx, room, json!({"_sbx":"pullreq","to": u,"label": l}));
+                        app.sys(format!(
+                            "asked {u} for ‘{l}’ — when it offers, `/accept` to receive + import it"
+                        ));
+                    }
+                    _ => app.sys("usage: /sbx pull @<user> <label>  (pull a published VM from a peer)"),
+                }
+            }
             Some("vms") => {
                 let tx = app_tx.clone();
                 tokio::spawn(async move {
@@ -2535,7 +2771,7 @@ fn handle_command(
                 }
             }
             other => {
-                let usage = "usage: /sbx <type> <option> — /sbx docker|podman|multipass|local [image] · /sbx vbox [gui] <vm> [yes] (host opens directly; non-host appends yes) · /sbx vbox new [name] (fresh VM) · vmlib [<id> [install [--iso path]]] (VM library — build locally) · vms · stop · save [label] [--local] · load <label> · snaps · vmsave <vm> [label] [--local] · vmload <vm> [label] · vmsnaps <vm>";
+                let usage = "usage: /sbx <type> <option> — /sbx docker|podman|multipass|local [image] · /sbx vbox [gui] <vm> [yes] (host opens directly; non-host appends yes) · /sbx vbox new [name] (fresh VM) · vmlib [<id> [install [--iso path]]] (VM library — build locally) · vms · stop · save [label] [--local] · load <label> · snaps · browse (saved VMs + what each is for) · publish <label> [tag…] (mark shareable) · catalog @user (peer's VMs) · pull @user <label> (fetch a peer's VM) · vmsave <vm> [label] [--local] · vmload <vm> [label] · vmsnaps <vm>";
                 // Bare `/sbx` → usage. An unrecognised subcommand → suggest the
                 // closest documented one before the usage line.
                 match other.and_then(|bad| closest(bad, SBX_SUBCOMMANDS).map(|s| (bad, s))) {
@@ -2830,8 +3066,8 @@ const KNOWN_COMMANDS: &[&str] = &[
 /// "did you mean". Aliases (`launch`, `virtualbox`, `snapshots`, `gui`) are
 /// omitted so suggestions point at the documented form.
 const SBX_SUBCOMMANDS: &[&str] = &[
-    "docker", "podman", "multipass", "local", "vbox", "stop", "save", "load", "snaps", "vms",
-    "vmsave", "vmload", "vmsnaps", "vmlib",
+    "docker", "podman", "multipass", "local", "vbox", "stop", "save", "load", "snaps", "browse",
+    "publish", "catalog", "pull", "vms", "vmsave", "vmload", "vmsnaps", "vmlib",
 ];
 
 /// Classic Levenshtein edit distance (insert/delete/substitute, each cost 1).
@@ -2866,6 +3102,144 @@ fn closest<'a>(input: &str, candidates: &[&'a str]) -> Option<&'a str> {
         .map(|(_, c)| c)
 }
 
+/// Record a freshly-saved snapshot in the host-global VM registry, caching the
+/// `.hh-agent` manifest summary read out of the (still-running, for OCI backends)
+/// container. Best-effort — never panics, never fails the save. Blocking; call
+/// from inside `spawn_blocking`.
+fn register_saved_snapshot(be: sbx::Backend, name: &str, label: &str, created_by: &str) {
+    let (artifact_kind, artifact_ref, size_bytes, manifest) = match be {
+        sbx::Backend::Docker | sbx::Backend::Podman => {
+            let engine = be.engine();
+            let image = format!("{}:{}", sbx::SNAP_REPO, label);
+            let size = oci_image_size(engine, &image);
+            let manifest = registry::read_container_manifest(engine, name, "/root");
+            ("image".to_string(), image, size, manifest)
+        }
+        // Multipass snapshots live in multipass's own store; the instance is
+        // powered off by save time, so there's no container to read a manifest
+        // from. Record the pointer; reconcile leaves it (no image to probe).
+        sbx::Backend::Multipass => ("snapshot".to_string(), label.to_string(), None, None),
+        sbx::Backend::Local => return, // nothing persistent to index
+    };
+    let (purpose, status, todo) = manifest
+        .as_deref()
+        .map(registry::scan_manifest)
+        .unwrap_or_default();
+    let entry = registry::Entry {
+        label: label.to_string(),
+        backend: be.engine().to_string(),
+        artifact_kind,
+        artifact_ref,
+        size_bytes,
+        created_unix: registry::now_unix(),
+        created_by: created_by.to_string(),
+        repo: registry::cwd_repo(),
+        purpose,
+        status,
+        todo,
+        ..Default::default()
+    };
+    if let Err(e) = registry::upsert(entry) {
+        eprintln!("registry upsert failed for '{label}': {e}");
+    }
+}
+
+/// Load a pulled `hh-snap-<label>.tar` into the local engine and record it in the
+/// host registry, scraping the `.hh-agent` manifest that rides inside the image so
+/// the receiver's `/sbx browse`/`load` know what the VM is for. Blocking — runs
+/// off the UI thread. Returns a status line for the room. The kept `.tar` doubles
+/// as this host's portable artifact, so a pulled VM is immediately re-shareable.
+fn register_received_snapshot(tar: &std::path::Path) -> anyhow::Result<String> {
+    let (engine, image, label) = sbx::import_image_archive(tar)?;
+    let (purpose, status, todo) = sbx::read_image_manifest(&engine, &image)
+        .as_deref()
+        .map(registry::scan_manifest)
+        .unwrap_or_default();
+    let entry = registry::Entry {
+        label: label.clone(),
+        backend: engine.clone(),
+        artifact_kind: "image".to_string(),
+        artifact_ref: image.clone(),
+        size_bytes: oci_image_size(&engine, &image),
+        created_unix: registry::now_unix(),
+        created_by: "pulled".to_string(),
+        repo: registry::cwd_repo(),
+        purpose,
+        status,
+        // The received file is already a portable artifact — keep it shareable so
+        // a pulled skill VM can be re-traded onward without re-exporting.
+        share_path: tar.to_string_lossy().into_owned(),
+        shareable: true,
+        todo,
+        ..Default::default()
+    };
+    registry::upsert(entry)?;
+    Ok(format!(
+        "† imported snapshot ‘{label}’ — `/sbx load {label}` to boot it · `/sbx browse` for details"
+    ))
+}
+
+/// Mark a saved snapshot shareable and ensure it has a portable, sendable file.
+/// File-backed snaps (`.tar`/`.ova`) are already tradeable; image-backed snaps
+/// are exported to `hh-snapshots/hh-snap-<label>.tar` first so a peer can receive
+/// them over `/send`. Blocking — run off the UI thread. Returns a status line.
+fn publish_snapshot(label: &str, tags: &[String]) -> anyhow::Result<String> {
+    use anyhow::Context;
+    let entry = registry::get(label)
+        .with_context(|| format!("no saved VM labelled '{label}' — `/sbx browse` to list"))?;
+    let share_path = match entry.artifact_kind.as_str() {
+        // Already a standalone file the owner controls — send it as-is.
+        "file" => entry.artifact_ref.clone(),
+        // Lives only in the engine image store — export a portable copy.
+        "image" => {
+            let backend = sbx::Backend::parse(&entry.backend)
+                .with_context(|| format!("unknown backend '{}' for '{label}'", entry.backend))?;
+            sbx::export_image(backend, label)?
+                .to_string_lossy()
+                .into_owned()
+        }
+        other => anyhow::bail!(
+            "'{label}' is a {other} snapshot — only file/image snapshots are tradeable"
+        ),
+    };
+    let published = registry::publish(label, &share_path, tags)?;
+    let tagnote = if published.tags.is_empty() {
+        String::new()
+    } else {
+        format!(" [{}]", published.tags.join(", "))
+    };
+    Ok(format!(
+        "✓ published '{label}'{tagnote} — shareable artifact at {share_path}; peers can `/sbx pull @<you> {label}`"
+    ))
+}
+
+/// Size in bytes of an OCI image (`<engine> image inspect … --format {{.Size}}`),
+/// or None if the engine/image isn't available. Blocking.
+fn oci_image_size(engine: &str, image: &str) -> Option<u64> {
+    let out = std::process::Command::new(engine)
+        .args(["image", "inspect", image, "--format", "{{.Size}}"])
+        .stderr(std::process::Stdio::null())
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    String::from_utf8_lossy(&out.stdout).trim().parse::<u64>().ok()
+}
+
+/// Does an OCI image still exist? Used to reconcile the registry so entries whose
+/// snapshot was pruned drop out of `/sbx browse`. Blocking.
+fn oci_image_exists(backend: &str, image: &str) -> bool {
+    let engine = if backend == "podman" { "podman" } else { "docker" };
+    std::process::Command::new(engine)
+        .args(["image", "inspect", image])
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null())
+        .status()
+        .map(|s| s.success())
+        .unwrap_or(false)
+}
+
 /// A safe snapshot label — compatible with both a Docker image tag and a
 /// multipass snapshot name (alphanumerics plus `.`, `_`, `-`).
 fn is_snap_label(s: &str) -> bool {
diff --git a/hh/src/main.rs b/hh/src/main.rs
index 3be6f56..4ed500a 100644
--- a/hh/src/main.rs
+++ b/hh/src/main.rs
@@ -9,6 +9,7 @@ mod crypto;
 mod ft;
 mod layout;
 mod net;
+mod registry;
 mod sbx;
 mod theme;
 mod ui;
diff --git a/hh/src/net.rs b/hh/src/net.rs
index 944eed4..8f33f53 100644
--- a/hh/src/net.rs
+++ b/hh/src/net.rs
@@ -1,7 +1,7 @@
 //! SRP authentication (blocking, one-shot) + async websocket transport and the
 //! reader task that decrypts/parses server frames into `Net` events.
 
-use crate::app::{ChatLine, Net, User};
+use crate::app::{CatalogItem, ChatLine, Net, User};
 use crate::crypto;
 use anyhow::{Context, Result};
 use base64::engine::general_purpose::STANDARD;
@@ -238,10 +238,45 @@ fn parse_sbx(text: &str, sender: &str) -> Option<Net> {
             by: sender.to_string(),
             vm: v["vm"].as_str().unwrap_or("a VM").to_string(),
         }),
+        // VM-trading (Phase B). `by` is the server-authenticated sender; `to`
+        // names the intended recipient (the handler acts only if it's itself).
+        "catreq" => Some(Net::SbxCatReq {
+            by: sender.to_string(),
+            to: v["to"].as_str().unwrap_or("").to_string(),
+        }),
+        "catalog" => Some(Net::SbxCatalog {
+            by: sender.to_string(),
+            to: v["to"].as_str().unwrap_or("").to_string(),
+            items: v["items"]
+                .as_array()
+                .map(|a| a.iter().map(parse_catalog_item).collect())
+                .unwrap_or_default(),
+        }),
+        "pullreq" => Some(Net::SbxPullReq {
+            by: sender.to_string(),
+            to: v["to"].as_str().unwrap_or("").to_string(),
+            label: v["label"].as_str()?.to_string(),
+        }),
         _ => None,
     }
 }
 
+/// Parse one `_sbx:catalog` item object into a `CatalogItem`. Tolerant: missing
+/// fields default to empty so a malformed row degrades rather than dropping the
+/// whole catalog.
+fn parse_catalog_item(v: &Value) -> CatalogItem {
+    CatalogItem {
+        label: v["label"].as_str().unwrap_or("").to_string(),
+        purpose: v["purpose"].as_str().unwrap_or("").to_string(),
+        status: v["status"].as_str().unwrap_or("").to_string(),
+        tags: v["tags"]
+            .as_array()
+            .map(|a| a.iter().filter_map(|t| t.as_str().map(str::to_string)).collect())
+            .unwrap_or_default(),
+        size_bytes: v["size"].as_u64(),
+    }
+}
+
 /// Parse a decrypted `{"_ai":...}` frame from an AI agent. `"typing"` toggles the
 /// thinking spinner; `"stream"` carries the cumulative reply text for a live
 /// preview bubble (`done` clears it once the final message is posted).
@@ -408,4 +443,49 @@ mod tests {
             let _ = parse_users(&v);
         }
     }
+
+    // ── Phase B VM-trading frames ──────────────────────────────────────────
+    #[test]
+    fn parse_sbx_catreq_uses_authenticated_sender() {
+        // The frame can claim any `to`, but `by` is the server-stamped sender.
+        let frame = json!({"_sbx":"catreq","to":"bob"}).to_string();
+        let Some(Net::SbxCatReq { by, to }) = parse_sbx(&frame, "alice") else {
+            panic!("expected SbxCatReq");
+        };
+        assert_eq!(by, "alice");
+        assert_eq!(to, "bob");
+    }
+
+    #[test]
+    fn parse_sbx_catalog_parses_items() {
+        let frame = json!({
+            "_sbx":"catalog","to":"alice",
+            "items":[
+                {"label":"kali-recon","purpose":"recon box","status":"in_progress",
+                 "tags":["recon","kali"],"size": 1024},
+                {"label":"bare"} // missing fields must default, not drop the row
+            ]
+        }).to_string();
+        let Some(Net::SbxCatalog { by, to, items }) = parse_sbx(&frame, "bob") else {
+            panic!("expected SbxCatalog");
+        };
+        assert_eq!(by, "bob");
+        assert_eq!(to, "alice");
+        assert_eq!(items.len(), 2);
+        assert_eq!(items[0].label, "kali-recon");
+        assert_eq!(items[0].tags, vec!["recon".to_string(), "kali".to_string()]);
+        assert_eq!(items[0].size_bytes, Some(1024));
+        assert_eq!(items[1].label, "bare");
+        assert!(items[1].purpose.is_empty());
+        assert_eq!(items[1].size_bytes, None);
+    }
+
+    #[test]
+    fn parse_sbx_pullreq_requires_label() {
+        let ok = json!({"_sbx":"pullreq","to":"bob","label":"kali-recon"}).to_string();
+        assert!(matches!(parse_sbx(&ok, "alice"), Some(Net::SbxPullReq { .. })));
+        // No label → reject the frame rather than pull a nameless VM.
+        let bad = json!({"_sbx":"pullreq","to":"bob"}).to_string();
+        assert!(parse_sbx(&bad, "alice").is_none());
+    }
 }
diff --git a/hh/src/registry.rs b/hh/src/registry.rs
new file mode 100644
index 0000000..131171d
--- /dev/null
+++ b/hh/src/registry.rs
@@ -0,0 +1,387 @@
+//! Host-global VM registry — the join between a saved snapshot and what it's FOR.
+//!
+//! `/sbx save` only ever produced an opaque engine artifact (a `hh-snap:<label>`
+//! image tag, an `hh-snapshots/*.tar`, …); the rich `.hh-agent` manifest that
+//! describes *why* a VM exists lives **inside** the VM and is invisible until you
+//! boot it. This module records, in one host-global file, an entry per saved VM:
+//! where its artifact is, plus a cached summary (purpose / status / next-todo)
+//! scraped from the VM's manifest at save time. That makes saved work browsable
+//! (`/sbx browse`) and selectable by an agent (`operator registry list`).
+//!
+//! Scope is deliberately **global** (`~/.hh/registry.json`), not per-room/per-repo:
+//! the artifacts it indexes are already host-level (one docker image store, one
+//! `hh-snapshots/` dir visible to every room), so a per-room split would hide the
+//! same image from other rooms. Each entry is tagged with its origin (`created_by`,
+//! `repo`) instead, and `list()` reconciles against what still exists so the file
+//! never grows past the live snapshots.
+
+use std::collections::BTreeMap;
+use std::path::PathBuf;
+use std::process::{Command, Stdio};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+
+const REGISTRY_VERSION: u32 = 1;
+
+/// One saved VM: its artifact location plus a cached, human/agent-readable summary
+/// of the work it carries. Unknown future fields on disk are ignored on read.
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct Entry {
+    pub label: String,
+    /// docker | podman | multipass | vbox
+    pub backend: String,
+    /// "image" (engine tag in `artifact_ref`) or "file" (path in `artifact_ref`)
+    pub artifact_kind: String,
+    pub artifact_ref: String,
+    #[serde(default)]
+    pub size_bytes: Option<u64>,
+    #[serde(default)]
+    pub created_unix: u64,
+    #[serde(default)]
+    pub created_by: String,
+    /// Originating working dir (basename) — origin tag, not a scoping key.
+    #[serde(default)]
+    pub repo: String,
+    // ── cached manifest summary (best-effort; empty if the VM had no .hh-agent) ──
+    #[serde(default)]
+    pub purpose: String,
+    #[serde(default)]
+    pub status: String,
+    #[serde(default)]
+    pub todo: Vec<String>,
+    // ── trading (Phase B) ──────────────────────────────────────────────────
+    /// Marked shareable by `/sbx publish`. Only shareable entries appear in the
+    /// catalog a peer receives; a non-shareable entry is private to this host.
+    #[serde(default)]
+    pub shareable: bool,
+    /// Free-form skill tags (e.g. "recon", "kali") for catalog filtering.
+    #[serde(default)]
+    pub tags: Vec<String>,
+    /// Portable, sendable artifact path (a `.tar`/`.ova` under `hh-snapshots/`).
+    /// An image-backed entry only becomes truly tradeable once it's been exported
+    /// to a file a peer can receive over `/send`; `publish` fills this in.
+    #[serde(default)]
+    pub share_path: String,
+}
+
+impl Entry {
+    /// The single most useful "what's next" line for a one-row table cell.
+    pub fn next_todo(&self) -> &str {
+        self.todo.first().map(String::as_str).unwrap_or("")
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+struct Registry {
+    version: u32,
+    entries: BTreeMap<String, Entry>,
+}
+
+impl Default for Registry {
+    fn default() -> Self {
+        Registry { version: REGISTRY_VERSION, entries: BTreeMap::new() }
+    }
+}
+
+/// `~/.hh/registry.json`. Falls back to `./.hh/registry.json` if `$HOME` is unset.
+pub fn registry_path() -> PathBuf {
+    let base = std::env::var("HOME").map(PathBuf::from).unwrap_or_else(|_| PathBuf::from("."));
+    base.join(".hh").join("registry.json")
+}
+
+fn load() -> Registry {
+    let path = registry_path();
+    let Ok(text) = std::fs::read_to_string(&path) else {
+        return Registry::default();
+    };
+    serde_json::from_str(&text).unwrap_or_default()
+}
+
+fn store(reg: &Registry) -> Result<()> {
+    let path = registry_path();
+    if let Some(dir) = path.parent() {
+        std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
+    }
+    let text = serde_json::to_string_pretty(reg).context("serializing registry")?;
+    std::fs::write(&path, text).with_context(|| format!("writing {}", path.display()))?;
+    Ok(())
+}
+
+/// Insert or replace the entry for `entry.label`.
+pub fn upsert(entry: Entry) -> Result<()> {
+    let mut reg = load();
+    reg.entries.insert(entry.label.clone(), entry);
+    store(&reg)
+}
+
+/// Read one entry by label (None if unknown). Used by `/sbx pull`/`publish` to
+/// look up an artifact without dragging in the engine reconcile probe.
+pub fn get(label: &str) -> Option<Entry> {
+    load().entries.remove(label)
+}
+
+/// Mark `label` shareable (Phase B `/sbx publish`): records the portable
+/// `share_path` a peer can receive, merges in any `tags`, and flips `shareable`.
+/// Errors if the label isn't in the registry. Idempotent.
+pub fn publish(label: &str, share_path: &str, tags: &[String]) -> Result<Entry> {
+    let mut reg = load();
+    let entry = reg
+        .entries
+        .get_mut(label)
+        .with_context(|| format!("no saved VM labelled '{label}' — /sbx browse to list"))?;
+    mark_published(entry, share_path, tags);
+    let out = entry.clone();
+    store(&reg)?;
+    Ok(out)
+}
+
+/// Pure mutation behind `publish` — flips shareable, sets the portable artifact
+/// path, and merges tags without duplicating. Split out so it's testable without
+/// touching the global on-disk registry.
+fn mark_published(entry: &mut Entry, share_path: &str, tags: &[String]) {
+    entry.shareable = true;
+    entry.share_path = share_path.to_string();
+    for t in tags {
+        if !entry.tags.contains(t) {
+            entry.tags.push(t.clone());
+        }
+    }
+}
+
+/// Whether an entry *names* a sendable artifact (shareable + has a `share_path`).
+/// The on-disk existence check is layered on top in `list_shareable`.
+fn is_offerable(e: &Entry) -> bool {
+    e.shareable && !e.share_path.is_empty()
+}
+
+/// The catalog this host offers: shareable entries whose portable artifact still
+/// exists on disk (a peer can only pull what we can actually send). Newest first.
+pub fn list_shareable() -> Vec<Entry> {
+    let mut v: Vec<Entry> = load()
+        .entries
+        .into_values()
+        .filter(|e| is_offerable(e) && PathBuf::from(&e.share_path).exists())
+        .collect();
+    v.sort_by(|a, b| b.created_unix.cmp(&a.created_unix));
+    v
+}
+
+/// Drops entries whose backing artifact no longer exists
+/// (image pruned, file deleted). Persists the pruned set so the file self-limits.
+/// `image_exists` is injected so the engine probe stays out of this module's deps;
+/// it is asked only about `artifact_kind == "image"` entries.
+pub fn list_reconciled(image_exists: impl Fn(&str, &str) -> bool) -> Vec<Entry> {
+    let mut reg = load();
+    let before = reg.entries.len();
+    reg.entries.retain(|_, e| match e.artifact_kind.as_str() {
+        "file" => PathBuf::from(&e.artifact_ref).exists(),
+        "image" => image_exists(&e.backend, &e.artifact_ref),
+        _ => true,
+    });
+    if reg.entries.len() != before {
+        let _ = store(&reg);
+    }
+    let mut v: Vec<Entry> = reg.entries.into_values().collect();
+    v.sort_by(|a, b| b.created_unix.cmp(&a.created_unix));
+    v
+}
+
+pub fn now_unix() -> u64 {
+    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
+}
+
+/// Basename of the current working dir — the origin tag for a new entry.
+pub fn cwd_repo() -> String {
+    std::env::current_dir()
+        .ok()
+        .and_then(|p| p.file_name().map(|s| s.to_string_lossy().into_owned()))
+        .unwrap_or_default()
+}
+
+/// Read `<root>/.hh-agent/manifest.yaml` out of a *running* OCI container via
+/// `<engine> exec <name> cat …`. Returns the raw YAML text, or None if the
+/// container isn't running / has no manifest. Blocking.
+pub fn read_container_manifest(engine: &str, name: &str, root: &str) -> Option<String> {
+    let path = format!("{}/.hh-agent/manifest.yaml", root.trim_end_matches('/'));
+    let out = Command::new(engine)
+        .args(["exec", name, "cat", &path])
+        .stderr(Stdio::null())
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    let text = String::from_utf8_lossy(&out.stdout).into_owned();
+    if text.trim().is_empty() {
+        None
+    } else {
+        Some(text)
+    }
+}
+
+/// Extract `(purpose, status, todo)` from an `.hh-agent` `manifest.yaml`.
+///
+/// This is a deliberately small, format-specific scraper rather than a real YAML
+/// parser (the crate has no YAML dep, and the file is machine-written by *our*
+/// `manifest.py` with a stable, block-style shape). It tolerates anything it
+/// doesn't recognise by returning empty fields — the entry is still recorded.
+pub fn scan_manifest(yaml: &str) -> (String, String, Vec<String>) {
+    let mut purpose = String::new();
+    let mut status = String::new();
+    let mut todo: Vec<String> = Vec::new();
+
+    let lines: Vec<&str> = yaml.lines().collect();
+    let mut i = 0;
+    while i < lines.len() {
+        let line = lines[i];
+        let trimmed = line.trim_start();
+
+        // `purpose:` is a top-level (column-0) scalar.
+        if !line.starts_with(char::is_whitespace) {
+            if let Some(rest) = line.strip_prefix("purpose:") {
+                purpose = unquote(rest.trim());
+            }
+        }
+        // `status:` only appears nested under `state:`; match it wherever indented.
+        if status.is_empty() {
+            if let Some(rest) = trimmed.strip_prefix("status:") {
+                status = unquote(rest.trim());
+            }
+        }
+        // `todo:` opens a block of `- item` bullets (or is inline `todo: []`).
+        if let Some(rest) = trimmed.strip_prefix("todo:") {
+            let after = rest.trim();
+            if !after.is_empty() && after != "[]" {
+                // inline flow list e.g. `todo: [a, b]` — best-effort split
+                todo = parse_inline_list(after);
+            } else if after.is_empty() {
+                let mut j = i + 1;
+                while j < lines.len() {
+                    let b = lines[j].trim_start();
+                    if let Some(item) = b.strip_prefix("- ") {
+                        todo.push(unquote(item.trim()));
+                        j += 1;
+                    } else {
+                        break;
+                    }
+                }
+            }
+        }
+        i += 1;
+    }
+    (purpose, status, todo)
+}
+
+fn parse_inline_list(s: &str) -> Vec<String> {
+    s.trim_start_matches('[')
+        .trim_end_matches(']')
+        .split(',')
+        .map(|p| unquote(p.trim()))
+        .filter(|p| !p.is_empty())
+        .collect()
+}
+
+/// Strip one layer of surrounding YAML quoting (single or double) and undo the
+/// doubled-single-quote escape PyYAML emits.
+fn unquote(s: &str) -> String {
+    let s = s.trim();
+    if s.len() >= 2 && s.starts_with('"') && s.ends_with('"') {
+        return s[1..s.len() - 1].replace("\\\"", "\"");
+    }
+    if s.len() >= 2 && s.starts_with('\'') && s.ends_with('\'') {
+        return s[1..s.len() - 1].replace("''", "'");
+    }
+    s.to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const SAMPLE: &str = "\
+schema: hh-agent/v1
+id: hh-abc123
+name: kali-recon-vm
+purpose: Kali recon sandbox, summoned in-room and stamped live
+created_by: operator
+vm:
+  engine: podman
+  image: kalilinux/kali-rolling
+goals:
+  objective: build the manifest tooling
+  user_intent: tradeable VM state
+  stop_conditions: []
+state:
+  status: in_progress
+  progress: tooling built
+  done:
+  - wrote manifest.py
+  todo:
+  - pivot to 10.0.0.5
+  - dump creds
+  blockers: []
+provenance:
+- by: operator
+  action: init
+";
+
+    #[test]
+    fn scans_purpose_status_todo() {
+        let (purpose, status, todo) = scan_manifest(SAMPLE);
+        assert_eq!(purpose, "Kali recon sandbox, summoned in-room and stamped live");
+        assert_eq!(status, "in_progress");
+        assert_eq!(todo, vec!["pivot to 10.0.0.5".to_string(), "dump creds".to_string()]);
+    }
+
+    #[test]
+    fn empty_todo_list_is_empty() {
+        let y = "purpose: x\nstate:\n  status: done\n  todo: []\n  blockers: []\n";
+        let (purpose, status, todo) = scan_manifest(y);
+        assert_eq!(purpose, "x");
+        assert_eq!(status, "done");
+        assert!(todo.is_empty());
+    }
+
+    #[test]
+    fn handles_quoted_scalars() {
+        let y = "purpose: 'a, b: c'\nstate:\n  status: \"in_progress\"\n";
+        let (purpose, status, _) = scan_manifest(y);
+        assert_eq!(purpose, "a, b: c");
+        assert_eq!(status, "in_progress");
+    }
+
+    #[test]
+    fn missing_fields_are_empty_not_panic() {
+        let (purpose, status, todo) = scan_manifest("name: bare\n");
+        assert!(purpose.is_empty());
+        assert!(status.is_empty());
+        assert!(todo.is_empty());
+    }
+
+    #[test]
+    fn mark_published_sets_fields_and_dedups_tags() {
+        let mut e = Entry { tags: vec!["recon".into()], ..Default::default() };
+        mark_published(&mut e, "/snaps/x.tar", &["recon".into(), "kali".into()]);
+        assert!(e.shareable);
+        assert_eq!(e.share_path, "/snaps/x.tar");
+        // "recon" already present must not duplicate; "kali" appended.
+        assert_eq!(e.tags, vec!["recon".to_string(), "kali".to_string()]);
+    }
+
+    #[test]
+    fn offerable_requires_shareable_flag_and_path() {
+        let base = Entry { share_path: "/snaps/x.tar".into(), shareable: true, ..Default::default() };
+        assert!(is_offerable(&base));
+        assert!(!is_offerable(&Entry { shareable: false, ..base.clone() }));
+        assert!(!is_offerable(&Entry { share_path: String::new(), ..base }));
+    }
+
+    #[test]
+    fn next_todo_picks_first() {
+        let e = Entry { todo: vec!["one".into(), "two".into()], ..Default::default() };
+        assert_eq!(e.next_todo(), "one");
+        assert_eq!(Entry::default().next_todo(), "");
+    }
+}
diff --git a/hh/src/sbx.rs b/hh/src/sbx.rs
index bf48e89..780a776 100644
--- a/hh/src/sbx.rs
+++ b/hh/src/sbx.rs
@@ -934,6 +934,99 @@ pub fn save_state(backend: Backend, name: &str, label: &str, local: bool) -> Res
     }
 }
 
+/// Export an already-committed `hh-snap:<label>` image to a portable `.tar` under
+/// `hh-snapshots/` — the standalone file a peer can receive over `/send`. This is
+/// the `--local` half of `save_state` made reusable for `/sbx publish`, so an
+/// image saved *without* `--local` can still be made tradeable later. Idempotent:
+/// returns the existing file if it's already there. Blocking.
+pub fn export_image(backend: Backend, label: &str) -> Result<std::path::PathBuf> {
+    let engine = match backend {
+        Backend::Docker | Backend::Podman => engine_bin(backend),
+        _ => anyhow::bail!("only docker/podman snapshots export to a portable .tar"),
+    };
+    let path = snap_dir()?.join(format!("hh-snap-{label}.tar"));
+    if path.exists() {
+        return Ok(path);
+    }
+    let tag = format!("{SNAP_REPO}:{label}");
+    let out = Command::new(engine)
+        .args(["save", &tag, "-o"])
+        .arg(&path)
+        .output()
+        .with_context(|| format!("{engine} save"))?;
+    if !out.status.success() {
+        let err = String::from_utf8_lossy(&out.stderr);
+        anyhow::bail!(
+            "exporting {tag} failed: {}",
+            err.lines().last().unwrap_or("").trim()
+        );
+    }
+    Ok(path)
+}
+
+/// Pull the `(tag, label)` of a loaded snapshot out of `<engine> load` stdout.
+/// docker prints `Loaded image: hh-snap:<label>`; podman prints
+/// `Loaded image: localhost/hh-snap:<label>` (registry-qualified). We locate the
+/// `hh-snap:` marker anywhere in a line and normalise to the bare `hh-snap:<label>`
+/// tag — podman resolves the unqualified form back to `localhost/…` on use, and
+/// docker stores it bare, so the bare tag is the portable handle. Pure (no engine
+/// call) so the dual-engine output shapes stay under test.
+fn parse_loaded_tag(stdout: &str) -> Option<(String, String)> {
+    let marker = format!("{SNAP_REPO}:");
+    let tag = stdout
+        .lines()
+        .find_map(|l| l.find(&marker).map(|i| l[i..].trim().to_string()))?;
+    let label = tag.split_once(':').map(|(_, l)| l.to_string()).unwrap_or_default();
+    Some((tag, label))
+}
+
+/// Load a received `hh-snap-<label>.tar` image archive into the local engine —
+/// the receive-side inverse of `export_image`. Tries docker first, then podman
+/// (a docker `save` archive loads into either). Returns `(engine, image_tag,
+/// label)` parsed from the engine's `Loaded image:` line. Blocking.
+pub fn import_image_archive(path: &std::path::Path) -> Result<(String, String, String)> {
+    let mut last_err = String::new();
+    for engine in ["docker", "podman"] {
+        let out = match Command::new(engine).args(["load", "-i"]).arg(path).output() {
+            Ok(o) => o,
+            Err(_) => continue, // engine not installed — try the next
+        };
+        if !out.status.success() {
+            last_err = String::from_utf8_lossy(&out.stderr).trim().to_string();
+            continue;
+        }
+        let stdout = String::from_utf8_lossy(&out.stdout);
+        let (tag, label) = parse_loaded_tag(&stdout)
+            .with_context(|| format!("{engine} load: no {SNAP_REPO}:… tag in output"))?;
+        return Ok((engine.to_string(), tag, label));
+    }
+    anyhow::bail!(
+        "couldn't load image archive (no docker/podman, or load failed: {})",
+        if last_err.is_empty() { "engine missing" } else { &last_err }
+    )
+}
+
+/// Run a throwaway container from `image` to read its `.hh-agent` manifest, so a
+/// just-imported snapshot can be registered with its purpose/status/todo without
+/// booting it into the room. `<engine> run --rm <image> cat …`; None if the image
+/// carries no manifest. Blocking.
+pub fn read_image_manifest(engine: &str, image: &str) -> Option<String> {
+    let out = Command::new(engine)
+        .args(["run", "--rm", image, "cat", "/root/.hh-agent/manifest.yaml"])
+        .stderr(Stdio::null())
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    let text = String::from_utf8_lossy(&out.stdout).into_owned();
+    if text.trim().is_empty() {
+        None
+    } else {
+        Some(text)
+    }
+}
+
 /// Restore a Multipass instance to a saved snapshot — the load-side inverse of
 /// the multipass branch of `save_state`. `multipass restore <name>.<label>`
 /// requires the instance to exist and be **stopped**, which is exactly the
@@ -1505,6 +1598,22 @@ mod tests {
     use super::*;
     use std::time::{Duration, Instant};
 
+    #[test]
+    fn parse_loaded_tag_handles_docker_and_podman() {
+        // docker: bare tag.
+        assert_eq!(
+            parse_loaded_tag("Loaded image: hh-snap:kali-recon\n"),
+            Some(("hh-snap:kali-recon".to_string(), "kali-recon".to_string()))
+        );
+        // podman: registry-qualified — must strip `localhost/` to the bare tag.
+        assert_eq!(
+            parse_loaded_tag("Loaded image: localhost/hh-snap:kali-recon\n"),
+            Some(("hh-snap:kali-recon".to_string(), "kali-recon".to_string()))
+        );
+        // no snapshot tag in the output → None (caller surfaces a clear error).
+        assert_eq!(parse_loaded_tag("Loaded image: alpine:latest\n"), None);
+    }
+
     /// Proves the PTY pipeline: spawn a real shell, send a command, read its
     /// output back off the channel. (Local backend — no container needed.)
     #[test]