// The registry side: handles -> DIDs -> module records and blobs (checked against their CIDs), the
// lockfile and vendored copies (`add`), and staging them for a static site or a server (`stage`,
// `oauth`). Modules are com.fedfork.exec.wasm.module records; a lib is <handle>/<rkey>.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use atlibs_host::compose::{self, Composition};
use atlibs_host::run::{Hook, module_key_ok, module_pin};
use atlibs_host::host::Fetch;
use atlibs_host::oauth::{AppRepo, ClientMetadataHandler, TEMPLATE_ORIGIN, compose_record};
use atlibs_host::wasm::sha256_hex;
use atlibs_host::{iso_now, pretty};
use serde_json::{Map, Value, json};

use crate::{Args, Cli, env};

pub const LOCK: &str = "atlibs.lock.json";

pub use atlibs_host::registry::{ABI, CLIENT, COMPOSITION, Client, Lib, MODULE, Net, check_pin, fetch_lib, named, resolve, spec_of, split_uri};

/// Net with the environment's PLC, handle resolver and private-address switch (tests, mirrors).
pub fn net_from_env(fetch: Arc<dyn Fetch>) -> Net {
    Net {
        fetch,
        plc: env("ATLIBS_PLC_URL").unwrap_or_else(|| "https://plc.directory".into()),
        resolver: env("ATLIBS_HANDLE_RESOLVER").unwrap_or_else(|| "https://public.api.bsky.app".into()),
        allow_private: env("ATLIBS_ALLOW_PRIVATE").as_deref() == Some("1"),
    }
}

/// The first `rel` in `start` or a directory above it.
pub fn find_up(start: &Path, rel: &str) -> Option<PathBuf> {
    let start = std::fs::canonicalize(start).ok()?;
    start.ancestors().map(|d| d.join(rel)).find(|p| p.exists())
}

/// A module directory: a path holding module.json, or a bare name for <repo>/modules/<name> (a module
/// repo) or <repo>/examples/rust/<name> (this one), the nearest above `from`.
pub fn module_dir(from: &Path, name: &str) -> Option<PathBuf> {
    let direct = from.join(name);
    if direct.join("module.json").is_file() {
        return Some(direct);
    }
    if name.contains('/') {
        return None;
    }
    let start = std::fs::canonicalize(from).ok()?;
    start.ancestors().flat_map(|d| ["modules", "examples/rust"].map(|base| d.join(base).join(name))).find(|d| d.join("module.json").is_file())
}

/// A local module: module.json, its `cargo wasm` build above it, records/<session>.json if any.
pub fn local_lib(dir: &Path) -> Result<Lib, String> {
    let read = |p: &Path| std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()));
    let spec: Value = serde_json::from_slice(&read(&dir.join("module.json"))?).map_err(|e| format!("{}/module.json: {e}", dir.display()))?;
    let spec = Value::Object(spec_of(&spec));
    let dir = std::fs::canonicalize(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    let key = dir.file_name().and_then(|n| n.to_str()).unwrap_or_default().to_string();
    let rel = format!("target/wasm32-unknown-unknown/release/{}.wasm", key.replace('-', "_"));
    let path = find_up(&dir, &rel).ok_or_else(|| format!("{}: no {rel} in it or above it: run cargo wasm", dir.display()))?;
    let wasm = read(&path)?;
    let client = records_client(&dir, &spec)?;
    Ok(Lib { key, reference: None, uri: None, cid: None, sha256: sha256_hex(&wasm), wasm, spec, client, id: std::fs::canonicalize(&path).unwrap_or(path).display().to_string(), pinned: None })
}

/// A local module.json's session record: records/<session>.json in `dir` or a directory above it.
fn records_client(dir: &Path, spec: &Value) -> Result<Option<Client>, String> {
    let Some(s) = spec["session"].as_str() else { return Ok(None) };
    let Some(p) = find_up(dir, &format!("records/{s}.json")) else { return Ok(None) };
    let record = serde_json::from_slice(&std::fs::read(&p).map_err(|e| format!("{}: {e}", p.display()))?).map_err(|e| format!("{}: {e}", p.display()))?;
    Ok(Some(Client { rkey: s.into(), record, uri: None, cid: None }))
}

// ---- lockfile ------------------------------------------------------------------------------

pub fn read_lock(path: &Path) -> Result<Value, String> {
    match std::fs::read(path) {
        Ok(bytes) => serde_json::from_slice(&bytes).map_err(|e| format!("{}: {e}", path.display())),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(json!({ "modules": {}, "sessions": {}, "compositions": {} })),
        Err(e) => Err(format!("{}: {e}", path.display())),
    }
}

pub fn write_json(path: &Path, value: &Value) -> Result<(), String> {
    if let Some(dir) = path.parent() {
        std::fs::create_dir_all(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    }
    std::fs::write(path, format!("{}\n", pretty(value))).map_err(|e| format!("{}: {e}", path.display()))
}

fn write_bytes(path: &Path, bytes: &[u8]) -> Result<(), String> {
    if let Some(dir) = path.parent() {
        std::fs::create_dir_all(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    }
    std::fs::write(path, bytes).map_err(|e| format!("{}: {e}", path.display()))
}

/// A module from the lockfile (or stage.json) in `dir`: its vendored wasm and record fields, re-checked
/// against the entry's pin (module_pin: a changed wasm, calls or hooks no longer match).
pub fn lock_lib(dir: &Path, lock: &Value, key: &str) -> Result<Lib, String> {
    let entry = &lock["modules"][key];
    let file = |k: &str| dir.join(entry[k].as_str().unwrap_or_default());
    let wasm = std::fs::read(file("wasm")).map_err(|e| format!("{}: {e}", file("wasm").display()))?;
    let sha256 = sha256_hex(&wasm);
    let spec: Value = serde_json::from_slice(&std::fs::read(file("spec")).map_err(|e| format!("{}: {e}", file("spec").display()))?).map_err(|e| e.to_string())?;
    if entry["pin"].as_str() != Some(module_pin(&wasm, &spec).as_str()) {
        return Err(format!("{key}: {} and {} do not match the pin {} has for it", file("wasm").display(), file("spec").display(), dir.display()));
    }
    let client = entry["session"].as_str().and_then(|s| {
        let session = &lock["sessions"][s];
        session.get("record").map(|record| Client { rkey: s.into(), record: record.clone(), uri: session["uri"].as_str().map(String::from), cid: session["cid"].as_str().map(String::from) })
    });
    let uri = entry["uri"].as_str().map(String::from);
    let id = uri.clone().unwrap_or_else(|| std::fs::canonicalize(file("wasm")).unwrap_or(file("wasm")).display().to_string());
    Ok(Lib { key: key.into(), reference: entry["ref"].as_str().map(String::from), uri, cid: entry["cid"].as_str().map(String::from), wasm, sha256, spec, client, id, pinned: Some(dir.to_path_buf()) })
}

fn vendor(dir: &Path, lock: &mut Value, lib: &Lib) -> Result<(), String> {
    if let Some(other) = lock["modules"][&lib.key]["uri"].as_str().filter(|u| Some(*u) != lib.uri.as_deref()) {
        return Err(format!("{} is already {other} in {LOCK}; remove it there first", lib.key));
    }
    let (wasm, spec) = (format!("modules/{}.wasm", lib.key), format!("modules/{}.json", lib.key));
    write_bytes(&dir.join(&wasm), &lib.wasm)?;
    write_json(&dir.join(&spec), &lib.spec)?;
    lock["modules"][&lib.key] = json!({ "ref": lib.reference, "uri": lib.uri, "cid": lib.cid, "pin": lib.pin(), "session": lib.client.as_ref().map(|c| &c.rkey), "wasm": wasm, "spec": spec });
    if let Some(c) = &lib.client {
        lock["sessions"][&c.rkey] = json!({ "uri": c.uri, "cid": c.cid, "record": c.record });
    }
    Ok(())
}

// ---- refs ----------------------------------------------------------------------------------

/// A published ref as people read it (add's output and lockfile `ref`): <handle>/<lib>[@pin], a did:.../<lib>
/// or at:// ref named by its handle (`named`). The uri, cid and pin stay what identifies it.
pub async fn shown(net: &Net, reference: &str, uri: &str) -> String {
    let (path, pin) = reference.split_once('@').map_or((reference, None), |(p, pin)| (p, Some(pin)));
    let name = if path.starts_with("at://") || path.starts_with("did:") { named(net, uri).await } else { path.to_string() };
    pin.map_or(name.clone(), |p| format!("{name}@{p}"))
}

/// <handle>/<lib>[@pin] or did:.../<lib>, as opposed to a path.
pub fn is_remote(reference: &str) -> bool {
    if reference.starts_with("at://") {
        return true;
    }
    let Some((head, lib)) = reference.split_once('/') else { return false };
    !lib.is_empty() && !lib.contains('/') && (head.starts_with("did:") || head.contains('.')) && !Path::new(reference).exists() && !head.starts_with('.')
}


/// A module vendored in `base`: its atlibs.lock.json entry (lock_lib), else `atlibs stage` output,
/// modules/<key>.wasm + .json, whose id is where the wasm is (never something the json names).
fn vendored_lib(base: &Path, key: &str) -> Result<Option<Lib>, String> {
    if base.join(LOCK).is_file() {
        let lock = read_lock(&base.join(LOCK))?;
        if lock["modules"].get(key).is_some() {
            return lock_lib(base, &lock, key).map(Some);
        }
    }
    if !module_key_ok(key) || !base.join(format!("modules/{key}.wasm")).is_file() {
        return Ok(None);
    }
    pair_lib(&base.join(format!("modules/{key}.wasm")), &base.join(format!("modules/{key}.json"))).map(Some)
}

/// <module.wasm> <module.json>: the module.json fields a host reads, keyed by the wasm's file stem; its session
/// record as local_lib finds one (records/<session>.json at or above the module.json).
fn pair_lib(wasm_path: &Path, spec_path: &Path) -> Result<Lib, String> {
    let read = |p: &Path| std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()));
    let spec: Value = serde_json::from_slice(&read(spec_path)?).map_err(|e| format!("{}: {e}", spec_path.display()))?;
    let wasm = read(wasm_path)?;
    let id = std::fs::canonicalize(wasm_path).map_err(|e| format!("{}: {e}", wasm_path.display()))?.display().to_string();
    let key = wasm_path.file_stem().and_then(|s| s.to_str()).unwrap_or_default().to_string();
    let spec = Value::Object(spec_of(&spec));
    let client = records_client(spec_path.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new(".")), &spec)?;
    Ok(Lib { key, reference: None, uri: None, cid: None, sha256: sha256_hex(&wasm), wasm, spec, client, id, pinned: None })
}

/// One instance's module ref as a lib: an at:// URI; else a key of the publisher's (`publisher`); else (a
/// local manifest) <handle>/<lib>[@pin]; else a module vendored in `base` (vendored_lib); else a local module directory.
async fn instance_lib(net: &Net, reference: &str, publisher: Option<&str>, base: &Path, pin: Option<&str>) -> Result<Lib, String> {
    match (reference.starts_with("at://"), publisher) {
        (true, _) => fetch_lib(net, reference, None, pin).await,
        (false, Some(did)) => atlibs_host::registry::instance_lib(net, reference, did, pin).await,
        (false, None) if is_remote(reference) => {
            let (uri, ref_pin) = resolve(net, reference).await?;
            let lib = fetch_lib(net, &uri, Some(reference), pin.or(ref_pin.as_deref())).await?;
            check_pin(&lib, ref_pin.as_deref())?;
            Ok(lib)
        }
        (false, None) => match vendored_lib(base, reference)? {
            Some(lib) => Ok(lib),
            None => {
                let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
                let dir = module_dir(base, reference).or_else(|| module_dir(&cwd, reference));
                local_lib(&dir.ok_or_else(|| format!("no module {reference} (modules/{reference}.wasm beside it, a module directory, modules/<name> or examples/rust/<name>, or an at:// URI)"))?)
            }
        },
    }
}

/// A composition as it runs (compose::load through instance_lib, hook defaults added; a run checks this
/// build's transports: run::target_of), its libs, and the manifest stage and add write: each instance's
/// module its lib's key, pinned (what vendored_lib reads next to modules/). Two modules that share a key are
/// refused.
pub async fn composition_libs(net: &Net, manifest: &Value, publisher: Option<&str>, base: &Path) -> Result<(Composition, Value, Vec<Lib>), String> {
    let lib_of: atlibs_host::registry::LibOf = {
        let (net, publisher, base) = (net.clone(), publisher.map(String::from), base.to_path_buf());
        Arc::new(move |reference: String, pin: Option<String>| {
            let (net, publisher, base) = (net.clone(), publisher.clone(), base.clone());
            Box::pin(async move { instance_lib(&net, &reference, publisher.as_deref(), &base, pin.as_deref()).await })
        })
    };
    let (comp, found) = atlibs_host::registry::load_with(manifest, lib_of).await?;
    // routes now; transports are the running host's to check (any it may serve counts here)
    let hooked = comp.modules.values().flat_map(|m| m.hooks.iter().map(|h| h.nsid.clone())).collect();
    compose::check(&comp, &hooked)?;
    let mut libs: Vec<Lib> = vec![];
    for (reference, lib) in &found {
        match libs.iter().find(|l| l.key == lib.key) {
            Some(l) if l.pin() == lib.pin() => {}
            Some(l) => return Err(format!("composition: {} and {reference} are both module {}: one key, two modules", l.reference.as_deref().unwrap_or(&l.id), lib.key)),
            None => libs.push(lib.clone()),
        }
    }
    let mut out = comp.manifest.clone();
    for inst in out["instances"].as_array_mut().into_iter().flatten() {
        let lib = &found.iter().find(|(r, _)| inst["module"] == r.as_str()).expect("loaded").1;
        inst["module"] = json!(lib.key);
        inst["pin"] = json!(lib.pin());
    }
    if let Some(m) = out.as_object_mut() {
        m.shift_remove("$type");
        m.shift_remove("createdAt");
    }
    Ok((comp, out, libs))
}

/// The lib a hook default `key` of module `id` names, resolved as runs resolve it: from
/// `pinned`'s atlibs.lock.json alone when the module is pinned (vendored with the same publisher's URI);
/// else the same publisher's record for an at:// module; else examples/rust/<key> above its build.
pub async fn default_lib(net: &Net, id: &str, pinned: Option<&Path>, key: &str) -> Result<Lib, String> {
    if !module_key_ok(key) {
        return Err(format!("hook default {key} is not a module key"));
    }
    let sibling = split_uri(id).map(|(did, collection, _)| format!("at://{did}/{collection}/{key}"));
    if let Some(dir) = pinned {
        let lock = read_lock(&dir.join(LOCK))?;
        let entry = &lock["modules"][key];
        if entry.is_null() || (sibling.is_some() && entry["uri"].as_str() != sibling.as_deref()) {
            return Err(format!("hook default {key}: {id} is pinned, but {} does not vendor {} (`atlibs add` the module again: it vendors its defaults)", dir.join(LOCK).display(), sibling.as_deref().unwrap_or(key)));
        }
        return lock_lib(dir, &lock, key);
    }
    match &sibling {
        Some(uri) => fetch_lib(net, uri, None, None).await,
        None => {
            let from = Path::new(id).parent().unwrap_or(Path::new("."));
            local_lib(&module_dir(from, key).ok_or_else(|| format!("hook default {key}: no modules/{key} or examples/rust/{key} above {id}"))?)
        }
    }
}

/// `lib`'s hook defaults, and theirs, as default_lib resolves them (live for an unpinned lib: `add` and
/// `stage` take them with it, so the lockfile or the staged site pins them too).
async fn defaults_of(net: &Net, lib: &Lib, pinned: Option<&Path>) -> Result<Vec<Lib>, String> {
    let mut out: Vec<Lib> = vec![];
    let mut todo = vec![lib.clone()];
    while let Some(next) = todo.pop() {
        for key in hooks_of(&next.spec)?.into_iter().filter_map(|h| h.default) {
            if out.iter().any(|l| l.key == key) || key == lib.key {
                continue;
            }
            let found = default_lib(net, &next.id, pinned, &key).await?;
            todo.push(found.clone());
            out.push(found);
        }
    }
    Ok(out)
}

fn hooks_of(spec: &Value) -> Result<Vec<Hook>, String> {
    atlibs_host::run::Module::from_manifest(vec![], spec).map(|m| m.hooks)
}

/// What a ref names: every command (run, serve, add, stage, oauth, verify, run --handler) resolves its
/// refs with resolve_ref, so a ref form means the same thing, with the same errors, everywhere.
pub enum Ref {
    Module(Lib),
    /// `name`: its rkey, lockfile key or file stem (<name>[.composition].json); `id`: its at:// URI, else
    /// the file's real path; `comp`: as it runs; `manifest`: as stage and add write it (composition_libs)
    Composition { name: String, id: String, cid: Option<String>, comp: Composition, manifest: Value, libs: Vec<Lib> },
    Client(Client),
}

impl Ref {
    pub fn libs(&self) -> &[Lib] {
        match self {
            Ref::Module(lib) => std::slice::from_ref(lib),
            Ref::Composition { libs, .. } => libs,
            Ref::Client(_) => &[],
        }
    }
}

pub const REF_FORMS: &str = "<handle>/<lib>[@pin|@cid] (or <did>/<lib>), an at:// URI, a name in atlibs.lock.json, a module directory or a name under examples/rust, a composition or clientMetadata .json, <module.wasm> <module.json>, or <stage.json|atlibs.lock.json> <name>";

fn read_json(path: &Path) -> Result<Value, String> {
    serde_json::from_slice(&std::fs::read(path).map_err(|e| format!("{}: {e}", path.display()))?).map_err(|e| format!("{}: {e}", path.display()))
}

fn real_path(path: &Path) -> Result<String, String> {
    std::fs::canonicalize(path).map(|p| p.display().to_string()).map_err(|e| format!("{}: {e}", path.display()))
}

/// Resolves the ref at the front of `words`; returns it and the words after it (after the second word a
/// .wasm, stage.json or lockfile takes). Names in a lockfile are looked up in `dir`'s; paths are the cwd's.
pub async fn resolve_ref<'a>(net: &Net, dir: &Path, words: &'a [String]) -> Result<(Ref, &'a [String]), String> {
    let first = words.first().ok_or_else(|| format!("name a ref: {REF_FORMS}"))?;
    let second = || words.get(1).ok_or_else(|| format!("{first} needs a second word: {REF_FORMS}"));
    if is_remote(first) {
        let (uri, pin) = resolve(net, first).await?;
        let (did, collection, rkey) = split_uri(&uri).ok_or_else(|| format!("{uri} is not an at:// record URI"))?;
        let found = match collection {
            MODULE => {
                let mut lib = fetch_lib(net, &uri, Some(first), pin.as_deref()).await?;
                // pinned: its hook defaults only from the lockfile here (default_lib)
                lib.pinned = pin.map(|_| dir.to_path_buf());
                Ref::Module(lib)
            }
            COMPOSITION => {
                let (cid, value) = atlibs_host::registry::composition(net, &uri, pin.as_deref()).await?;
                let (comp, manifest, libs) = composition_libs(net, &value, Some(did), dir).await?;
                Ref::Composition { name: rkey.into(), id: uri.clone(), cid: Some(cid), comp, manifest, libs }
            }
            CLIENT => {
                let (cid, value) = net.record(&uri).await?.ok_or_else(|| format!("{uri} not found"))?;
                if let Some(p) = pin.filter(|p| *p != cid) {
                    return Err(format!("{rkey}: pinned to {p}, but it is record {cid}"));
                }
                Ref::Client(Client { rkey: rkey.into(), record: value, uri: Some(uri.clone()), cid: Some(cid) })
            }
            _ => return Err(format!("{uri} is not a {MODULE}, {COMPOSITION} or {CLIENT} record")),
        };
        return Ok((found, &words[1..]));
    }
    let path = Path::new(first);
    if first.ends_with(".wasm") {
        return Ok((Ref::Module(pair_lib(path, Path::new(second()?))?), &words[2..]));
    }
    if first.ends_with(".json") {
        let value = read_json(path)?;
        let stem = path.file_name().and_then(|s| s.to_str()).unwrap_or_default();
        let stem = stem.strip_suffix(".composition.json").or(stem.strip_suffix(".json")).unwrap_or(stem).to_string();
        if value["$type"] == CLIENT {
            // records/<rkey>.json: a client with no module behind it
            return Ok((Ref::Client(Client { rkey: stem, record: value, uri: None, cid: None }), &words[1..]));
        }
        let base = path.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new("."));
        if value.get("instances").is_some() {
            let (comp, manifest, libs) = composition_libs(net, &value, None, base).await?;
            return Ok((Ref::Composition { name: stem, id: real_path(path)?, cid: None, comp, manifest, libs }, &words[1..]));
        }
        // stage.json or atlibs.lock.json: <file> <name>
        return Ok((from_lock(net, base, &value, second()?).await?, &words[2..]));
    }
    let lock = read_lock(&dir.join(LOCK))?;
    if lock["modules"].get(first).is_some() || lock["compositions"].get(first).is_some() {
        return Ok((from_lock(net, dir, &lock, first).await?, &words[1..]));
    }
    let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
    let found = module_dir(&cwd, first).ok_or_else(|| format!("no module {first}: not in {}, not a module directory (module.json), an at:// URI or <handle>/<lib>", dir.join(LOCK).display()))?;
    Ok((Ref::Module(local_lib(&found)?), &words[1..]))
}

/// `name` in a lockfile (or stage.json) in `dir`: a vendored module, or a composition next to its modules.
async fn from_lock(net: &Net, dir: &Path, lock: &Value, name: &str) -> Result<Ref, String> {
    if lock["modules"].get(name).is_some() {
        return Ok(Ref::Module(lock_lib(dir, lock, name)?));
    }
    if let Some(file) = lock["compositions"][name]["file"].as_str() {
        let path = dir.join(file);
        let (comp, manifest, libs) = composition_libs(net, &read_json(&path)?, None, dir).await?;
        let entry = &lock["compositions"][name];
        let id = match entry["uri"].as_str() {
            Some(uri) => uri.to_string(),
            None => real_path(&path)?,
        };
        return Ok(Ref::Composition { name: name.into(), id, cid: entry["cid"].as_str().map(String::from), comp, manifest, libs });
    }
    let known: Vec<&str> = ["modules", "compositions"].iter().flat_map(|k| lock[k].as_object().into_iter().flatten().map(|(k, _)| k.as_str())).collect();
    Err(format!("no module {name} in {} (it has: {})", dir.display(), known.join(", ")))
}

/// What `stage`/`oauth` work from: modules, and compositions (name -> rewritten manifest).
#[derive(Default)]
pub struct Set {
    pub libs: Vec<Lib>,
    pub compositions: Vec<(String, Value)>,
    /// clientMetadata records named directly (a page that runs no module, only signs in)
    pub clients: Vec<Client>,
}

impl Set {
    fn push(&mut self, lib: Lib) -> Result<(), String> {
        match self.libs.iter().find(|l| l.key == lib.key) {
            Some(l) if l.pin() == lib.pin() => Ok(()),
            Some(_) => Err(format!("module {} named twice", lib.key)),
            None => Ok(self.libs.push(lib)),
        }
    }

    /// A module comes with its hook defaults (defaults_of: from its lockfile when pinned), so the staged
    /// site has them beside it.
    async fn add(&mut self, net: &Net, found: Ref) -> Result<(), String> {
        match found {
            Ref::Module(lib) => {
                for d in defaults_of(net, &lib, lib.pinned.as_deref()).await? {
                    self.push(d)?;
                }
                self.push(lib)
            }
            Ref::Composition { name, manifest, libs, .. } => {
                for lib in libs {
                    self.push(lib)?;
                }
                Ok(self.compositions.push((name, manifest)))
            }
            Ref::Client(c) => Ok(self.clients.push(c)),
        }
    }
}

/// Refs (or, with none, everything in the lockfile in `dir`) as a Set.
pub async fn gather(net: &Net, dir: &Path, refs: &[String]) -> Result<Set, String> {
    let mut set = Set::default();
    if refs.is_empty() {
        let lock_path = dir.join(LOCK);
        if !lock_path.exists() {
            return Err(format!("no {}: `atlibs add <handle>/<lib>` first, or name modules", lock_path.display()));
        }
        let lock = read_lock(&lock_path)?;
        for k in ["modules", "compositions"].iter().flat_map(|k| lock[k].as_object().into_iter().flatten().map(|(k, _)| k)) {
            set.add(net, from_lock(net, dir, &lock, k).await?).await?;
        }
        return Ok(set);
    }
    let mut words = refs;
    while !words.is_empty() {
        let (found, rest) = resolve_ref(net, dir, words).await?;
        set.add(net, found).await?;
        words = rest;
    }
    Ok(set)
}

// ---- commands ------------------------------------------------------------------------------

/// atlibs add <handle>/<lib>[@pin]... [--dir .]
pub async fn add(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&["dir"], "add")?;
    if args.pos.is_empty() {
        return Err(crate::usage("add"));
    }
    let dir = PathBuf::from(args.flag("dir").unwrap_or("."));
    let lock_path = dir.join(LOCK);
    let mut lock = read_lock(&lock_path)?;
    for (i, r) in args.pos.iter().enumerate() {
        if !is_remote(r) {
            return Err(format!("{r}: add takes published refs, <handle>/<lib>[@pin|@cid] or an at:// URI\n{}", crate::usage("add")));
        }
        match resolve_ref(&cli.net, &dir, &args.pos[i..]).await?.0 {
            Ref::Composition { name, id, cid, manifest, mut libs, .. } => {
                for lib in libs.iter_mut() {
                    if lib.reference.is_none() && lib.uri.is_some() {
                        lib.reference = Some(named(&cli.net, &lib.id).await);
                    }
                    vendor(&dir, &mut lock, lib)?;
                }
                let file = format!("{name}.composition.json");
                write_json(&dir.join(&file), &json!(manifest.as_object().cloned().unwrap_or_default().into_iter().chain([("uri".to_string(), json!(id))]).collect::<Map<_, _>>()))?;
                let shown = shown(&cli.net, r, &id).await;
                lock["compositions"][&name] = json!({ "ref": shown, "uri": id, "cid": cid, "file": file });
                eprintln!("added composition {name} ({}): {}", bare(&shown), libs.iter().map(|l| l.key.as_str()).collect::<Vec<_>>().join(", "));
            }
            Ref::Module(mut lib) => {
                lib.reference = Some(shown(&cli.net, r, &lib.id).await);
                // its hook defaults as published now, vendored beside it: a run of it takes them from here
                for mut d in defaults_of(&cli.net, &lib, None).await? {
                    d.reference = Some(named(&cli.net, &d.id).await);
                    vendor(&dir, &mut lock, &d)?;
                    eprintln!("added {} ({}) pin {}, a hook default of {}", d.key, bare(d.reference.as_deref().unwrap_or_default()), d.pin(), lib.key);
                }
                vendor(&dir, &mut lock, &lib)?;
                eprintln!("added {} ({}) pin {}", lib.key, bare(lib.reference.as_deref().unwrap_or_default()), lib.pin());
            }
            Ref::Client(c) => return Err(format!("{} is not a {MODULE} or {COMPOSITION} record", c.uri.unwrap_or_default())),
        }
    }
    write_json(&lock_path, &lock)?;
    eprintln!("wrote {}", lock_path.display());
    write_bindings(&dir, &[])
}

pub const BINDINGS: &str = "atlibs.modules";

/// atlibs.modules.js and .d.ts in `dir`: a binding per module atlibs.lock.json vendors (pinned), per module
/// directory under modules/ (a module repo's own, unpinned) and per one in `extra`, for compose/build.js. The
/// .d.ts is the same JSON as a literal type.
pub fn write_bindings(dir: &Path, extra: &[PathBuf]) -> Result<(), String> {
    let read = |p: &Path| -> Result<Value, String> { serde_json::from_slice(&std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()))?).map_err(|e| format!("{}: {e}", p.display())) };
    let lock = read_lock(&dir.join(LOCK))?;
    let mut modules: Vec<(String, Option<String>, Value)> = vec![];
    for (key, e) in lock["modules"].as_object().into_iter().flatten() {
        let spec = e["spec"].as_str().ok_or_else(|| format!("{key}: no spec in {LOCK}"))?;
        modules.push((key.clone(), e["pin"].as_str().map(String::from), read(&dir.join(spec))?));
    }
    let mut own: Vec<PathBuf> = std::fs::read_dir(dir.join("modules")).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.join("module.json").is_file()).collect();
    own.sort();
    for d in own.iter().chain(extra) {
        let key = std::fs::canonicalize(d).map_err(|e| format!("{}: {e}", d.display()))?.file_name().and_then(|n| n.to_str()).unwrap_or_default().to_string();
        if !modules.iter().any(|(k, ..)| *k == key) {
            modules.push((key, None, read(&d.join("module.json"))?));
        }
    }
    let head = "// generated by atlibs compose bindings from module.json: do not edit";
    let (mut js, mut dts) = (vec!["/* @ts-self-types=\"./atlibs.modules.d.ts\" */".to_string(), head.into(), "const lock = {};".into()], vec![head.to_string()]);
    for (key, pin, spec) in &modules {
        let mut b = Map::new();
        b.insert("key".into(), json!(key));
        if let Some(p) = pin {
            b.insert("pin".into(), json!(p));
        }
        for k in ["session", "calls", "anyHooks", "hooks", "implements", "config"] {
            if let Some(v) = spec.get(k).filter(|v| !v.is_null()) {
                b.insert(k.into(), v.clone());
            }
        }
        b.insert("entries".into(), json!(spec["actions"].as_array().into_iter().flatten().filter_map(|a| a["entry"].as_str()).collect::<Vec<_>>()));
        let (name, body) = (camel(key), Value::Object(b).to_string());
        js.push(format!("export const {name} = (lock[{}] = {body});", json!(key)));
        dts.push(format!("export declare const {name}: {body};"));
    }
    js.push("for (const m of Object.values(lock)) Object.defineProperty(m, \"lock\", { value: lock });".into());
    for (ext, lines) in [("js", js), ("d.ts", dts)] {
        let path = dir.join(format!("{BINDINGS}.{ext}"));
        std::fs::write(&path, format!("{}\n", lines.join("\n"))).map_err(|e| format!("{}: {e}", path.display()))?;
    }
    eprintln!("wrote {}", dir.join(format!("{BINDINGS}.js")).display());
    Ok(())
}

/// A module key as a JS name: vm-mcp -> vmMcp.
fn camel(key: &str) -> String {
    let mut out = String::new();
    let mut up = false;
    for c in key.chars() {
        if c == '-' || c == '_' {
            up = !out.is_empty();
        } else {
            out.push(if up { c.to_ascii_uppercase() } else { c });
            up = false;
        }
    }
    out
}

/// `atlibs compose bindings [--dir .] [<module dir> ...]`: write_bindings.
pub fn compose_bindings(argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&["dir"], "compose bindings")?;
    let dirs: Vec<PathBuf> = args.pos.iter().map(PathBuf::from).collect();
    write_bindings(Path::new(args.flag("dir").unwrap_or(".")), &dirs)
}

/// A shown ref without its @pin (add prints the pin itself).
fn bare(shown: &str) -> &str {
    shown.split('@').next().unwrap_or(shown)
}

fn origin_flag(args: &Args) -> Result<String, String> {
    let origin = args.flag("origin").ok_or("--origin <https://site/base/> is required")?;
    let url = url::Url::parse(origin).map_err(|_| format!("--origin {origin} is not a URL"))?;
    if url.scheme() != "https" {
        return Err("--origin must be https (localhost pages use the loopback client: clientId() in javascript/src/oauth)".into());
    }
    Ok(url.to_string())
}

/// session -> record, in the order modules name them.
fn records(set: &Set, need: bool) -> Result<Map<String, Value>, String> {
    let mut records = Map::new();
    for lib in &set.libs {
        match (&lib.client, lib.spec["session"].as_str()) {
            (Some(c), _) => {
                records.entry(c.rkey.clone()).or_insert_with(|| c.record.clone());
            }
            (None, Some(s)) if need => return Err(format!("{}: no records/{s}.json (its session's clientMetadata record)", lib.key)),
            _ => {}
        }
    }
    for c in &set.clients {
        if records.insert(c.rkey.clone(), c.record.clone()).is_some_and(|had| had != c.record) {
            return Err(format!("session {} named twice, with different records", c.rkey));
        }
    }
    Ok(records)
}

/// atlibs stage [--target spa|server] [--out dist] [--origin URL (spa only; default a TEMPLATE_ORIGIN template)] [--oauth-path oauth/] [--name N [--record F]] [--dir .] [ref...]
pub async fn stage(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&["target", "out", "origin", "oauth-path", "name", "record", "dir"], "stage")?;
    let target = args.flag("target").unwrap_or("spa");
    if target != "spa" && target != "server" {
        return Err(format!("--target must be spa or server\n{}", crate::usage("stage")));
    }
    let server = target == "server";
    if server && args.has("name") {
        return Err("--name composes OAuth clients for a page; --target server mounts each session's (ClientMetadataHandler)".into());
    }
    if server && args.has("origin") {
        return Err("--origin is a page's; a --target server is told its origin when it serves (atlibs serve --origin, ATLIBS_ORIGIN)".into());
    }
    // spa with no --origin: a template on TEMPLATE_ORIGIN, filled at deploy time
    let origin = match (server, args.has("origin")) {
        (true, _) => None,
        (false, true) => Some(origin_flag(&args)?),
        (false, false) => Some(format!("{TEMPLATE_ORIGIN}/")),
    };
    let template = origin.as_deref() == Some(format!("{TEMPLATE_ORIGIN}/").as_str());
    let out = PathBuf::from(args.flag("out").unwrap_or("dist"));
    let oauth_path = args.flag("oauth-path").unwrap_or("oauth/").trim_matches('/').to_string();
    let set = gather(&cli.net, Path::new(args.flag("dir").unwrap_or(".")), &args.pos).await?;
    let records = records(&set, !server)?;

    let mut modules = Map::new();
    for lib in &set.libs {
        let (wasm, spec) = (format!("modules/{}.wasm", lib.key), format!("modules/{}.json", lib.key));
        write_bytes(&out.join(&wasm), &lib.wasm)?;
        write_json(&out.join(&spec), &lib.spec)?;
        modules.insert(lib.key.clone(), json!({ "session": lib.client.as_ref().map(|c| &c.rkey).or(lib.spec["session"].as_str().map(String::from).as_ref()), "wasm": wasm, "spec": spec, "sha256": lib.sha256, "pin": lib.pin() }));
    }
    let manifest = if let Some(origin) = origin {
        let handler = ClientMetadataHandler { records: records.clone(), path: Some(format!("/{oauth_path}/")), origin: Some(origin.clone()), redirect_uri: None, name: args.flag("name").map(String::from) };
        let base = origin.trim_end_matches('/');
        let mut sessions = Map::new();
        for session in records.keys() {
            let doc = handler.document(base, session).expect("a record")?;
            write_json(&out.join(format!("{oauth_path}/{session}/client-metadata.json")), &doc)?;
            let mut entry = json!({ "clientId": doc["client_id"], "scope": doc["scope"] });
            let optional: Vec<&str> = records[session]["scopeGroups"].as_array().into_iter().flatten().filter(|g| g["optional"] == true).flat_map(|g| g["scopes"].as_array().into_iter().flatten().filter_map(Value::as_str)).collect();
            if !optional.is_empty() {
                entry["required"] = json!(doc["scope"].as_str().unwrap_or_default().split(' ').filter(|s| !optional.contains(s)).collect::<Vec<_>>().join(" "));
            }
            sessions.insert(session.clone(), entry);
        }
        let mut manifest = json!({ "origin": origin, "clientId": null, "scope": null, "sessions": sessions, "modules": modules });
        if let Some(name) = args.flag("name") {
            if records.contains_key(name) {
                return Err(format!("--name {name} is already a session name"));
            }
            let doc = handler.document(base, name).expect("the composite")?;
            write_json(&out.join(format!("{oauth_path}/{name}/client-metadata.json")), &doc)?;
            manifest["clientId"] = doc["client_id"].clone();
            manifest["scope"] = doc["scope"].clone();
            if let Some(file) = args.flag("record") {
                write_json(Path::new(file), &compose_record(&records, name, doc["client_id"].as_str().unwrap_or_default(), &format!("{base}/"), &iso_now())?)?;
            }
        } else if let [(_, one)] = &sessions.iter().collect::<Vec<_>>()[..] {
            manifest["clientId"] = one["clientId"].clone();
            manifest["scope"] = one["scope"].clone();
        }
        if !template && records.values().any(|r| r.get("app").is_some()) {
            manifest["app"] = stage_app(cli, &records, &origin, &out).await?;
        }
        manifest
    } else {
        let sessions: Map<String, Value> = records.into_iter().map(|(k, record)| (k, json!({ "record": record }))).collect();
        json!({ "target": target, "oauthPath": format!("/{oauth_path}/"), "sessions": sessions, "modules": modules })
    };
    let single = set.compositions.len() == 1;
    for (name, comp) in &set.compositions {
        let mut comp = comp.clone();
        comp["target"] = json!(target);
        write_json(&out.join(if single { "composition.json".into() } else { format!("{name}.composition.json") }), &comp)?;
    }
    write_json(&out.join("stage.json"), &manifest)?;
    out!("{}", pretty(&manifest));
    if template {
        let dir = out.display();
        eprintln!("no --origin: {TEMPLATE_ORIGIN} is a placeholder; at deploy, with ORIGIN=https://host/page (no trailing slash):\n  find {dir}/stage.json {dir}/{oauth_path} -name '*.json' -exec sed -i.bak \"s#{TEMPLATE_ORIGIN}#$ORIGIN#g\" {{}} + && find {dir} -name '*.json.bak' -delete\n  (or stage again with --origin)");
    }
    Ok(())
}

/// The app identity (AppRepo) as files a static host serves at the origin's root: the DID document, atproto-did,
/// the wellKnown files, and the repo reads a static host can answer whatever the query (describeRepo; getRecord
/// and listRecords only when there is one record: a static file cannot pick by collection and rkey).
async fn stage_app(cli: &Cli, records: &Map<String, Value>, origin: &str, out: &Path) -> Result<Value, String> {
    let site = url::Url::parse(origin).map_err(|_| format!("--origin {origin} is not a URL"))?;
    let root = site.origin().ascii_serialization();
    let key = cli.store()?.didkey(&format!("app:{root}")).await?;
    let app = AppRepo::new(records, &root, key.did.trim_start_matches("did:key:"))?.ok_or("no app record")?;
    let did = app.did.clone();
    let mut paths: Vec<String> = ["/.well-known/did.json", "/.well-known/atproto-did"].map(String::from).to_vec();
    paths.extend(app.well_known.keys().map(|f| format!("/.well-known/{f}")));
    paths.push(format!("/xrpc/com.atproto.repo.describeRepo?repo={did}"));
    if let [(collection, rkey, ..)] = &app.records[..] {
        paths.push(format!("/xrpc/com.atproto.repo.getRecord?repo={did}&collection={collection}&rkey={rkey}"));
        paths.push(format!("/xrpc/com.atproto.repo.listRecords?repo={did}&collection={collection}"));
    } else {
        eprintln!("app {did}: {} records; a static host serves no getRecord or listRecords for more than one (serve them: atlibs serve --origin)", app.records.len());
    }
    let mut files = vec![];
    for path in paths {
        let (_, _, body) = app.answer(&path).ok_or("an app path")?;
        let file = path.split('?').next().unwrap_or_default().trim_start_matches('/').to_string();
        write_bytes(&out.join(&file), body.as_bytes())?;
        files.push(file);
    }
    if site.path() != "/" {
        eprintln!("app {did}: a did:web resolves at the host root: serve {} from {root}/, not {origin}", files.join(", "));
    }
    Ok(json!({ "did": did, "files": files }))
}

/// atlibs oauth --origin URL [--oauth-path oauth/] [--name N] [--dir .] [ref...]: the client metadata
/// documents an origin serves, by session.
pub async fn oauth(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&["origin", "oauth-path", "name", "dir"], "oauth")?;
    let origin = args.flag("origin").ok_or_else(|| format!("--origin <https://site/> is required\n{}", crate::usage("oauth")))?;
    url::Url::parse(origin).map_err(|_| format!("--origin {origin} is not a URL"))?;
    let set = gather(&cli.net, Path::new(args.flag("dir").unwrap_or(".")), &args.pos).await?;
    let handler = ClientMetadataHandler {
        records: records(&set, false)?,
        path: Some(format!("/{}/", args.flag("oauth-path").unwrap_or("oauth/").trim_matches('/'))),
        name: args.flag("name").map(String::from),
        ..Default::default()
    };
    let mut docs = Map::new();
    for session in handler.records.keys().chain(handler.name.as_ref()) {
        docs.insert(session.clone(), handler.document(origin.trim_end_matches('/'), session).expect("known")?);
    }
    out!("{}", pretty(&Value::Object(docs)));
    Ok(())
}

/// atlibs verify <handle>/<lib>: rebuild a published module from its source records in a container
/// (`engine()`) and compare the sha256 with its manifest and its blob.
/// `atlibs compose check <composition.json>`: the hosts' load checks with no wasm (compose::check_specs).
/// Prints the composition as it would run (hook defaults added); a failure is the error, exit 1.
pub async fn compose_check(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&[], "compose check")?;
    let [file] = &args.pos[..] else { return Err(crate::usage("compose check")) };
    let path = Path::new(file);
    let manifest: Value = serde_json::from_slice(&std::fs::read(path).map_err(|e| format!("{file}: {e}"))?).map_err(|e| format!("{file}: {e}"))?;
    let base = path.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new(".")).to_path_buf();
    let net = cli.net.clone();
    let loader: compose::ModuleLoader = Arc::new(move |reference: String, pin: Option<String>| {
        let (net, base) = (net.clone(), base.clone());
        Box::pin(async move { spec_module_at(&net, &reference, &base, pin.as_deref()).await })
    });
    out!("{}", pretty(&compose::check_specs(&manifest, &loader).await?));
    Ok(())
}

/// A module ref's module for compose check, wasm read only where it is fetched: atlibs.lock.json's entry in
/// `base` (its spec and pin), stage.json's, a published one (fetched and checked like `add`), else a module
/// directory's module.json (no pin).
async fn spec_module_at(net: &Net, reference: &str, base: &Path, pin: Option<&str>) -> Result<atlibs_host::run::Module, String> {
    let read = |p: &Path| -> Result<Value, String> { serde_json::from_slice(&std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()))?).map_err(|e| format!("{}: {e}", p.display())) };
    for file in [LOCK, "stage.json"] {
        if !base.join(file).is_file() {
            continue;
        }
        let entry = &read(&base.join(file))?["modules"][reference];
        if let Some(spec) = entry["spec"].as_str() {
            return compose::spec_module(&Value::Object(spec_of(&read(&base.join(spec))?)), entry["pin"].as_str());
        }
    }
    if is_remote(reference) {
        return atlibs_host::registry::module_of(&instance_lib(net, reference, None, base, pin).await?);
    }
    let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
    let dir = module_dir(base, reference).or_else(|| module_dir(&cwd, reference)).ok_or_else(|| format!("no module {reference} (atlibs.lock.json or stage.json beside the composition, a module directory, modules/<name> or examples/rust/<name>, or a published ref)"))?;
    compose::spec_module(&Value::Object(spec_of(&read(&dir.join("module.json"))?)), None)
}

pub async fn verify(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let args = Args::parse(argv, &[])?;
    args.only(&[], "verify")?;
    let [reference] = &args.pos[..] else { return Err(crate::usage("verify")) };
    let found = if is_remote(reference) { Some(resolve_ref(&cli.net, Path::new("."), &args.pos).await?.0) } else { None };
    let Some(Ref::Module(lib)) = found else { return Err(format!("{reference}: verify takes a published module\n{}", crate::usage("verify"))) };
    let engine = engine()?;
    let work = std::env::temp_dir().join(format!("atlibs-verify-{}", atlibs_host::store::uuid()));
    let report = rebuild(cli, &lib, &engine, &work).await;
    let _ = std::fs::remove_dir_all(&work);
    let report = report?;
    out!("{}", pretty(&report));
    if report["reproducible"] == true { Ok(()) } else { Err("not reproducible".into()) }
}

/// The container engine: CONTAINER_ENGINE, else the first of podman, docker and Apple's `container` (macOS)
/// that answers `info` (`container system status`). Not `--version`: that also passes for a docker CLI whose
/// daemon is down, or a docker look-alike whose `run` ends before the command does and still exits 0.
pub fn engine() -> Result<String, String> {
    if let Some(engine) = env("CONTAINER_ENGINE") {
        return Ok(engine);
    }
    let up = |engine: &str, probe: &[&str]| std::process::Command::new(engine).args(probe).stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null()).status().is_ok_and(|s| s.success());
    [("podman", &["info"][..]), ("docker", &["info"]), ("container", &["system", "status"])]
        .into_iter()
        .find(|(engine, probe)| up(engine, probe))
        .map(|(engine, _)| engine.to_string())
        .ok_or_else(|| "no container engine answers (podman info, docker info, container system status): start one, or set CONTAINER_ENGINE".into())
}

/// The reference build, shared by `atlibs publish` and `verify`: `command` in docker.io/library/rust:<the
/// version of `rustc --version`> with `src` at /src. Not on the host: cargo hashes `rustc -vV` (its `host:`
/// line too) into every crate's -C metadata, which changes the wasm, so a macOS build never equals a Linux rebuild.
pub fn container_build(engine: &str, src: &Path, rustc_version: &str, command: &str) -> Result<(), String> {
    let image = format!("docker.io/library/rust:{}", rustc_version.split(' ').nth(1).unwrap_or("latest"));
    eprintln!("building with '{command}' in {image} ({engine})");
    let status = std::process::Command::new(engine)
        .args(["run", "--rm", "-v", &format!("{}:/src:Z", src.display()), "-w", "/src", &image, "bash", "-c", &format!("rustup target add wasm32-unknown-unknown >/dev/null && {command}")])
        .stdout(std::io::stderr())
        .status()
        .map_err(|e| format!("{engine}: {e}"))?;
    if !status.success() {
        return Err(format!("the build in {image} failed"));
    }
    Ok(())
}

/// Writes the manifest's source files under `work`/src, builds them in its image with `engine`, and reports
/// {built, manifest, published, reproducible}.
async fn rebuild(cli: &Cli, lib: &Lib, engine: &str, work: &Path) -> Result<Value, String> {
    let uri = lib.id.clone();
    let (_, module) = cli.net.record(&uri).await?.ok_or_else(|| format!("{uri} not found"))?;
    let source = module["source"]["uri"].as_str().ok_or_else(|| format!("{uri} has no source"))?;
    let (cid, manifest) = cli.net.record(source).await?.ok_or_else(|| format!("{source} not found"))?;
    if module["source"]["cid"] != cid.as_str() {
        return Err("the manifest's CID does not match the module's source reference".into());
    }
    for entry in manifest["files"].as_array().into_iter().flatten() {
        let path = entry["path"].as_str().unwrap_or_default();
        if path.is_empty() || path.starts_with('/') || path.split('/').any(|p| p == "..") {
            return Err(format!("refusing path {path}"));
        }
        let file_uri = entry["file"]["uri"].as_str().unwrap_or_default();
        let (file_cid, file) = cli.net.record(file_uri).await?.ok_or_else(|| format!("{file_uri} not found"))?;
        let text = file["text"].as_str().unwrap_or_default();
        if entry["file"]["cid"] != file_cid.as_str() || file["sha256"] != sha256_hex(text.as_bytes()).as_str() {
            return Err(format!("{path}: does not match its record or the manifest"));
        }
        write_bytes(&work.join("src").join(path), text.as_bytes())?;
    }
    let build = &manifest["build"];
    if build["tool"] != "cargo" {
        return Err(format!("unknown build tool {}", build["tool"]));
    }
    let artifact = build["artifact"].as_str().unwrap_or_default();
    eprintln!("rebuilding {}", manifest["name"]);
    container_build(engine, &work.join("src"), build["version"].as_str().unwrap_or_default(), build["command"].as_str().unwrap_or_default())?;
    let built = std::fs::read(work.join("src").join(artifact)).map_err(|e| format!("{engine} run exited 0 but the build left no {artifact} ({e}): is {engine} a working docker-compatible engine? (CONTAINER_ENGINE)"))?;
    let built = sha256_hex(&built);
    let expected = manifest["output"]["sha256"].as_str().unwrap_or_default();
    Ok(json!({ "built": built, "manifest": expected, "published": lib.sha256, "reproducible": built == expected && built == lib.sha256 }))
}

#[cfg(test)]
pub mod tests {
    use super::*;
    use crate::fake::Pds;

    #[tokio::test]
    async fn add_pins_vendors_and_checks_blobs_then_stage_writes_clients_at_the_paths() {
        let pds = Pds::new();
        let module = pds.publish_module("alice.test", "echo", b"\0asm-echo", Some("echo"));
        let dir = tempdir("add");
        let cli = pds.cli(&dir);
        add(&cli, &["alice.test/echo".into(), "--dir".into(), dir.display().to_string()]).await.unwrap();
        let lock = read_lock(&dir.join(LOCK)).unwrap();
        assert_eq!(lock["modules"]["echo"]["uri"], module);
        let spec: Value = serde_json::from_slice(&std::fs::read(dir.join("modules/echo.json")).unwrap()).unwrap();
        assert_eq!(spec["uri"], module);
        assert!(spec.get("owner").is_none() && spec.get("id").is_none(), "a record cannot choose its identity: {spec}");
        assert_eq!(lock["modules"]["echo"]["ref"], "alice.test/echo");
        assert_eq!(std::fs::read(dir.join("modules/echo.wasm")).unwrap(), b"\0asm-echo");
        assert_eq!(lock["sessions"]["echo"]["record"]["name"], "echo");
        // and the bindings compose/build.js imports: the module, pinned
        let js = std::fs::read_to_string(dir.join("atlibs.modules.js")).unwrap();
        let pin = lock["modules"]["echo"]["pin"].as_str().unwrap();
        assert!(js.contains(&format!(r#"export const echo = (lock["echo"] = {{"key":"echo","pin":"{pin}""#)), "{js}");
        assert!(std::fs::read_to_string(dir.join("atlibs.modules.d.ts")).unwrap().contains(&format!(r#"export declare const echo: {{"key":"echo","pin":"{pin}""#)));
        // a pin that does not match, a handle the DID document does not claim, a blob that does not match
        assert!(add(&cli, &["alice.test/echo@deadbeef00".into(), "--dir".into(), dir.display().to_string()]).await.unwrap_err().contains("too short"));
        assert!(add(&cli, &[format!("alice.test/echo@{}", "d".repeat(32)), "--dir".into(), dir.display().to_string()]).await.unwrap_err().contains(&format!("pin {} not published", "d".repeat(32))));
        assert!(add(&cli, &["mallory.test/echo".into(), "--dir".into(), dir.display().to_string()]).await.unwrap_err().contains("does not claim"));
        pds.corrupt_blob(&module);
        assert!(add(&cli, &["alice.test/echo".into(), "--dir".into(), dir.display().to_string()]).await.unwrap_err().contains("does not match its CID"));

        let out = dir.join("dist");
        let argv = |extra: &[&str]| -> Vec<String> { ["--out", out.to_str().unwrap(), "--dir", dir.to_str().unwrap()].iter().chain(extra).map(|s| s.to_string()).collect() };
        stage(&cli, &argv(&["--origin", "https://site.example/app", "--oauth-path", "auth/clients"])).await.unwrap();
        let doc: Value = serde_json::from_slice(&std::fs::read(out.join("auth/clients/echo/client-metadata.json")).unwrap()).unwrap();
        assert_eq!(doc["client_id"], "https://site.example/app/auth/clients/echo/client-metadata.json");
        assert_eq!(doc["redirect_uris"], json!(["https://site.example/app/"]));
        let staged: Value = serde_json::from_slice(&std::fs::read(out.join("stage.json")).unwrap()).unwrap();
        assert_eq!(staged["clientId"], doc["client_id"]);
        assert_eq!(std::fs::read(out.join("modules/echo.wasm")).unwrap(), b"\0asm-echo");
        assert!(stage(&cli, &argv(&["--origin", "http://site.example"])).await.unwrap_err().contains("must be https"));

        // no --origin: a template whose one literal the deploy step replaces
        stage(&cli, &argv(&["--oauth-path", "auth/clients"])).await.unwrap();
        let fill = |p: &str| -> Value { serde_json::from_str(&std::fs::read_to_string(out.join(p)).unwrap().replace(TEMPLATE_ORIGIN, "https://site.example/app")).unwrap() };
        let (staged, filled) = (fill("stage.json"), fill("auth/clients/echo/client-metadata.json"));
        assert_eq!((&staged["clientId"], &filled["client_id"], &filled["redirect_uris"]), (&doc["client_id"], &doc["client_id"], &doc["redirect_uris"]));

        assert!(stage(&cli, &argv(&["--target", "server", "--origin", "https://site.example/"])).await.unwrap_err().starts_with("--origin is a page's"));
        stage(&cli, &argv(&["--target", "server"])).await.unwrap();
        let staged: Value = serde_json::from_slice(&std::fs::read(out.join("stage.json")).unwrap()).unwrap();
        assert_eq!(staged["target"], "server");
        assert_eq!(staged["sessions"]["echo"]["record"]["name"], "echo");
        let handler = ClientMetadataHandler::from_stage(&staged);
        assert_eq!(handler.document("https://srv.example", "echo").unwrap().unwrap()["client_id"], "https://srv.example/oauth/echo/client-metadata.json");
        oauth(&cli, &["--origin".into(), "https://x.example".into(), "--dir".into(), dir.display().to_string()]).await.unwrap();
        assert!(oauth(&cli, &["--dir".into(), dir.display().to_string()]).await.unwrap_err().contains("--origin"));
        assert!(stage(&cli, &argv(&["--orign", "x"])).await.unwrap_err().contains("unknown flag --orign"));
    }

    #[tokio::test]
    async fn stage_a_client_record_with_no_module() {
        let pds = Pds::new();
        let dir = tempdir("client");
        let cli = pds.cli(&dir);
        let record = dir.join("agents-page.json");
        write_json(&record, &json!({ "$type": CLIENT, "name": "agents page", "clientMetadata": { "client_id": "https://tpl.test/c.json", "scope": "atproto rpc:a.b.c?aud=*" }, "createdAt": "x" })).unwrap();
        let (uri, _) = pds.put("alice.test", CLIENT, "remote", json!({ "$type": CLIENT, "name": "remote", "clientMetadata": { "client_id": "https://tpl.test/c.json", "scope": "atproto" }, "createdAt": "x" }));
        let out = dir.join("site");
        let argv = |refs: &[&str]| -> Vec<String> { ["--origin", "https://site.example/agents/", "--out", out.to_str().unwrap()].iter().chain(refs).map(|s| s.to_string()).collect() };
        stage(&cli, &argv(&[record.to_str().unwrap()])).await.unwrap();
        let staged: Value = serde_json::from_slice(&std::fs::read(out.join("stage.json")).unwrap()).unwrap();
        assert_eq!(staged["clientId"], "https://site.example/agents/oauth/agents-page/client-metadata.json");
        assert_eq!(staged["scope"], "atproto rpc:a.b.c?aud=*");
        assert_eq!(staged["modules"], json!({}));
        stage(&cli, &argv(&[record.to_str().unwrap(), &uri, "--name", "both"])).await.unwrap();
        assert!(out.join("oauth/remote/client-metadata.json").is_file());
        let both: Value = serde_json::from_slice(&std::fs::read(out.join("oauth/both/client-metadata.json")).unwrap()).unwrap();
        assert_eq!(both["scope"], "atproto rpc:a.b.c?aud=*");
    }

    #[tokio::test]
    async fn add_a_composition_vendors_its_modules_next_to_it() {
        let pds = Pds::new();
        pds.publish_module("alice.test", "echo", b"\0asm-echo", None);
        pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo" }, { "name": "b", "module": "echo" }], "createdAt": "x" }));
        let dir = tempdir("comp");
        let cli = pds.cli(&dir);
        add(&cli, &["alice.test/pair".into(), "--dir".into(), dir.display().to_string()]).await.unwrap();
        let comp: Value = serde_json::from_slice(&std::fs::read(dir.join("pair.composition.json")).unwrap()).unwrap();
        assert_eq!(comp["instances"][0]["module"], "echo");
        let lock = read_lock(&dir.join(LOCK)).unwrap();
        assert_eq!(comp["instances"][0]["pin"], lock["modules"]["echo"]["pin"]);
        let words = [dir.join("pair.composition.json").display().to_string()];
        let Ref::Composition { comp, .. } = resolve_ref(&cli.net, &dir, &words).await.unwrap().0 else { panic!("a composition") };
        assert_eq!(comp.modules.len(), 2);
    }

    /// atlibs_host::registry::published, what an embedding host (knotless-run) loads a binding's composition with.
    #[tokio::test]
    async fn a_host_loads_a_published_composition_pinned_by_cid() {
        let pds = Pds::new();
        pds.publish_module("alice.test", "echo", b"\0asm-echo", None);
        let (uri, cid) = pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo" }], "createdAt": "x" }));
        let cli = pds.cli(&tempdir("published"));
        let (found, comp, libs) = atlibs_host::registry::published(&cli.net, &uri, Some(&cid)).await.unwrap();
        assert_eq!((found, comp.modules.len(), libs[0].1.key.as_str()), (cid.clone(), 1, "echo"));
        let err = atlibs_host::registry::published(&cli.net, &uri, Some("bafyother")).await.err().unwrap();
        assert_eq!(err, format!("{uri}: version bafyother not published ({uri}.bafyother not found)"));
        // republished: the binding's CID still loads its version; unpinned, the latest
        pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo" }, { "name": "b", "module": "echo" }], "createdAt": "x" }));
        assert_eq!(atlibs_host::registry::published(&cli.net, &uri, Some(&cid)).await.unwrap().1.modules.len(), 1);
        assert_eq!(atlibs_host::registry::published(&cli.net, &uri, None).await.unwrap().1.modules.len(), 2);
        assert!(atlibs_host::registry::published(&cli.net, &uri, Some("../x")).await.err().unwrap().contains("is not a record CID"));
    }

    /// A republished module leaves every composition that pins it as it was (knotless-run's binding broke on it).
    #[tokio::test]
    async fn a_pinned_instance_survives_its_modules_republish() {
        let pds = Pds::new();
        let module = pds.publish_module("alice.test", "echo", b"\0asm-echo", None);
        let cli = pds.cli(&tempdir("republish"));
        let pin = fetch_lib(&cli.net, &module, None, None).await.unwrap().pin();
        let (uri, cid) = pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo", "pin": pin }], "createdAt": "x" }));
        pds.publish_module("alice.test", "echo", b"\0asm-echo-2", None);
        let (_, comp, libs) = atlibs_host::registry::published(&cli.net, &uri, Some(&cid)).await.unwrap();
        assert_eq!((comp.modules["a"].pin.as_str(), libs[0].1.id.as_str(), libs[0].1.key.as_str()), (pin.as_str(), module.as_str(), "echo"));
        // a pin publish never wrote (published before pins were kept): refused, by name
        let gone = "0".repeat(64);
        pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo", "pin": gone }], "createdAt": "x" }));
        let err = atlibs_host::registry::published(&cli.net, &uri, None).await.err().unwrap();
        assert_eq!(err, format!("{module}: pin {gone} not published ({module}.{} not found)", &gone[..32]));
    }

    /// Every ref form, through resolve_ref, as stage/oauth (gather), run and serve (run::target_of) see it.
    #[tokio::test]
    async fn every_ref_form_resolves_the_same_for_every_command() {
        let pds = Pds::new();
        let module = pds.publish_module("alice.test", "echo", b"\0asm-echo", Some("echo"));
        let (comp_uri, comp_cid) = pds.put_composition("alice.test", "pair", json!({ "$type": COMPOSITION, "name": "pair", "instances": [{ "name": "a", "module": "echo" }], "createdAt": "x" }));
        let (client_uri, _) = pds.put("alice.test", CLIENT, "page", json!({ "$type": CLIENT, "name": "page", "clientMetadata": { "client_id": "https://tpl.test/c.json", "scope": "atproto" }, "createdAt": "x" }));
        let dir = tempdir("refs");
        let cli = pds.cli(&dir);
        let words = |w: &[&str]| -> Vec<String> { w.iter().map(|s| s.to_string()).collect() };
        let resolve = |w: Vec<String>| {
            let (net, dir) = (cli.net.clone(), dir.clone());
            async move { resolve_ref(&net, &dir, &w).await.map(|(found, rest)| (found, rest.len())) }
        };
        let kind = |found: &Ref| match found {
            Ref::Module(lib) => format!("module {}", lib.key),
            Ref::Composition { name, libs, .. } => format!("composition {name} [{}]", libs.iter().map(|l| l.key.as_str()).collect::<Vec<_>>().join(",")),
            Ref::Client(c) => format!("client {}", c.rkey),
        };

        // published: <handle>/<lib>, pins (sha256 prefix, record CID), at:// URIs of each collection
        let cid = pds.get(&module).map(|_| fetch_lib(&cli.net, &module, None, None)).unwrap().await.unwrap().cid.unwrap();
        let pin = fetch_lib(&cli.net, &module, None, None).await.unwrap().pin();
        for (r, want) in [("alice.test/echo", "module echo"), (&format!("alice.test/echo@{}", &pin[..32]), "module echo"), (&format!("alice.test/echo@{pin}"), "module echo"), (&format!("alice.test/echo@{cid}"), "module echo"), (&module, "module echo"), ("alice.test/pair", "composition pair [echo]"), (&format!("alice.test/pair@{comp_cid}"), "composition pair [echo]"), (&comp_uri, "composition pair [echo]"), (&client_uri, "client page")] {
            let (found, rest) = resolve(words(&[r, "next"])).await.unwrap_or_else(|e| panic!("{r}: {e}"));
            assert_eq!((kind(&found), rest), (want.to_string(), 1), "{r}");
        }
        let Ref::Module(lib) = resolve(words(&["alice.test/echo"])).await.unwrap().0 else { panic!() };
        assert_eq!((lib.id.as_str(), lib.reference.as_deref()), (module.as_str(), Some("alice.test/echo")), "the run identity is the at:// URI");
        // <did>/<lib> and the at:// URI name the same record as <handle>/<lib> (add shows them by the handle, below)
        let did = crate::fake::did_of("alice.test");
        for r in [format!("{did}/echo"), module.clone(), format!("{did}/echo@{cid}")] {
            let Ref::Module(same) = resolve(words(&[&r])).await.unwrap().0 else { panic!("{r}") };
            assert_eq!((same.id.as_str(), same.cid.as_deref(), same.pin()), (module.as_str(), Some(cid.as_str()), pin.clone()), "{r}");
            assert_eq!(shown(&cli.net, &r, &same.id).await, if r.contains('@') { format!("alice.test/echo@{cid}") } else { "alice.test/echo".into() }, "{r}");
        }
        assert!(resolve(words(&["alice.test/echo@deadbeef00"])).await.err().unwrap().contains("pin deadbeef00 is too short"));
        assert!(resolve(words(&["alice.test/pair@bafynope"])).await.err().unwrap().contains("version bafynope not published"));
        let (other, _) = pds.put("alice.test", "com.example.other", "x", json!({}));
        assert!(resolve(words(&[&other])).await.err().unwrap().contains("is not a com.fedfork.exec.wasm.module, com.fedfork.exec.composition or com.fedfork.exec.oauth.clientMetadata record"));

        // local: a module directory (its cargo wasm build above it), <module.wasm> <module.json>
        let local = dir.join("local");
        write_bytes(&local.join("target/wasm32-unknown-unknown/release/toy.wasm"), b"\0asm-toy").unwrap();
        write_json(&local.join("toy/module.json"), &json!({ "name": "toy", "actions": [{ "entry": "go" }], "id": "/etc/passwd" })).unwrap();
        let toy_dir = local.join("toy").display().to_string();
        let toy_wasm = local.join("target/wasm32-unknown-unknown/release/toy.wasm").display().to_string();
        let toy_json = local.join("toy/module.json").display().to_string();
        assert_eq!(kind(&resolve(words(&[&toy_dir, "go"])).await.unwrap().0), "module toy");
        let (pair, rest) = resolve(words(&[&toy_wasm, &toy_json, "go"])).await.unwrap();
        assert_eq!((kind(&pair), rest), ("module toy".into(), 1));
        assert!(pair.libs()[0].spec.get("id").is_none() && pair.libs()[0].id.ends_with("toy.wasm"), "the id is where the wasm is");
        assert!(resolve(words(&[&toy_wasm])).await.err().unwrap().contains("needs a second word"));

        // a composition .json of module dirs; the same, staged (modules/ beside it, sha256 pinned)
        let comp_file = local.join("duo.composition.json");
        write_json(&comp_file, &json!({ "instances": [{ "name": "x", "module": toy_dir }, { "name": "y", "module": toy_dir }] })).unwrap();
        let (found, _) = resolve(words(&[comp_file.to_str().unwrap()])).await.unwrap();
        assert_eq!(kind(&found), "composition duo [toy]");
        // ... and of published modules by <handle>/<lib> or <did>/<lib>, pins checked
        let remote_comp = local.join("remote.composition.json");
        write_json(&remote_comp, &json!({ "instances": [{ "name": "e", "module": "alice.test/echo" }, { "name": "f", "module": format!("{did}/echo@{pin}") }] })).unwrap();
        assert_eq!(kind(&resolve(words(&[remote_comp.to_str().unwrap()])).await.unwrap().0), "composition remote [echo]");
        write_json(&remote_comp, &json!({ "instances": [{ "name": "e", "module": format!("alice.test/echo@{}", "0".repeat(32)) }] })).unwrap();
        assert!(resolve(words(&[remote_comp.to_str().unwrap()])).await.err().unwrap().contains(&format!("pin {} not published", "0".repeat(32))));
        let staged = dir.join("staged");
        stage(&cli, &words(&["--target", "server", "--out", staged.to_str().unwrap(), comp_file.to_str().unwrap()])).await.unwrap();
        let staged_comp = staged.join("composition.json").display().to_string();
        let (found, _) = resolve(words(&[&staged_comp])).await.unwrap();
        assert_eq!(kind(&found), "composition composition [toy]");
        assert!(matches!(crate::run::target_of(found).unwrap(), crate::run::Target::Composition { .. }));
        let mut tampered: Value = serde_json::from_str(&std::fs::read_to_string(&staged_comp).unwrap()).unwrap();
        tampered["instances"][0]["pin"] = json!("0".repeat(64));
        write_json(Path::new(&staged_comp), &tampered).unwrap();
        assert_eq!(resolve(words(&[&staged_comp])).await.err().unwrap(), "composition: instance x module does not match its pin");
        // stage.json <name>
        let stage_json = staged.join("stage.json").display().to_string();
        assert_eq!(resolve(words(&[&stage_json, "toy", "go"])).await.map(|(f, r)| (kind(&f), r)).unwrap(), ("module toy".into(), 1));
        assert!(resolve(words(&[&stage_json, "nope"])).await.err().unwrap().contains("no module nope in"));

        // a clientMetadata .json, and names in `dir`'s lockfile (module and composition); then gather
        let client_file = dir.join("agents-page.json");
        write_json(&client_file, &json!({ "$type": CLIENT, "name": "agents page", "clientMetadata": { "client_id": "https://tpl.test/c.json", "scope": "atproto" }, "createdAt": "x" })).unwrap();
        assert_eq!(kind(&resolve(words(&[client_file.to_str().unwrap()])).await.unwrap().0), "client agents-page");
        assert!(crate::run::target_of(resolve(words(&[client_file.to_str().unwrap()])).await.unwrap().0).err().unwrap().contains("no module to run"));
        add(&cli, &words(&[&format!("{did}/echo"), &comp_uri, "--dir", dir.to_str().unwrap()])).await.unwrap();
        let lock = read_lock(&dir.join(LOCK)).unwrap();
        assert_eq!((&lock["modules"]["echo"]["ref"], &lock["modules"]["echo"]["uri"], &lock["compositions"]["pair"]["ref"]), (&json!("alice.test/echo"), &json!(module), &json!("alice.test/pair")), "the lockfile shows the handle, keeps the at:// URI");
        add(&cli, &words(&["alice.test/echo", "alice.test/pair", "--dir", dir.to_str().unwrap()])).await.unwrap();
        assert!(add(&cli, &words(&[&toy_dir, "--dir", dir.to_str().unwrap()])).await.err().unwrap().contains("add takes published refs"));
        assert!(add(&cli, &words(&[&client_uri, "--dir", dir.to_str().unwrap()])).await.err().unwrap().contains("is not a com.fedfork.exec.wasm.module or com.fedfork.exec.composition record"));
        let (echo, _) = resolve(words(&["echo"])).await.unwrap();
        assert_eq!((kind(&echo), echo.libs()[0].id.as_str()), ("module echo".into(), module.as_str()));
        let Ref::Composition { id, cid, .. } = resolve(words(&["pair"])).await.unwrap().0 else { panic!("a composition") };
        assert_eq!((id, cid), (comp_uri.clone(), Some(comp_cid.clone())), "a lockfile composition keeps its at:// identity");
        let lock_json = dir.join(LOCK).display().to_string();
        assert_eq!(kind(&resolve(words(&[&lock_json, "pair"])).await.unwrap().0), "composition pair [echo]");
        assert!(resolve(words(&["no-such-module"])).await.err().unwrap().contains("no module no-such-module: not in"));
        assert!(resolve(vec![]).await.err().unwrap().starts_with("name a ref"));

        let set = gather(&cli.net, &dir, &words(&["echo", &toy_wasm, &toy_json, client_file.to_str().unwrap(), "pair"])).await.unwrap();
        assert_eq!(set.libs.iter().map(|l| l.key.as_str()).collect::<Vec<_>>(), ["echo", "toy"]);
        assert_eq!((set.clients[0].rkey.as_str(), set.compositions[0].0.as_str()), ("agents-page", "pair"));
        let all = gather(&cli.net, &dir, &[]).await.unwrap();
        assert_eq!((all.libs.len(), all.compositions.len()), (1, 1));
    }

    /// A module record with hooks: bob's asker, whose policy hook defaults to `pol` (review 4a H1, 4b #1).
    fn publish_asker(pds: &Pds, handle: &str, rkey: &str, wasm: &[u8], extra: Value) -> String {
        let mut value = json!({ "$type": MODULE, "name": rkey, "module": pds.blob(wasm), "abi": ABI, "actions": [], "calls": [], "createdAt": "x" });
        value.as_object_mut().unwrap().extend(extra.as_object().cloned().unwrap_or_default());
        pds.put_module(handle, rkey, value).0
    }
    const H: &str = "com.example.hooks.check";

    #[tokio::test]
    async fn a_hook_default_is_its_modules_publishers_and_pinned_with_it() {
        let pds = Pds::new();
        let asker = publish_asker(&pds, "bob.test", "asker", b"\0asm-asker", json!({ "hooks": [{ "nsid": H, "default": "pol", "combine": "all-allow", "reason": "r" }] }));
        let pol = publish_asker(&pds, "bob.test", "pol", b"\0asm-pol", json!({ "calls": ["com.atproto.repo.getRecord"], "implements": [{ "nsid": H, "entry": "evaluate" }] }));
        let dir = tempdir("defaults");
        let cli = pds.cli(&dir);
        // a composition of bob's module, local or published by carol: the default is bob's pol (4b #1)
        let manifest = json!({ "$type": COMPOSITION, "instances": [{ "name": "a", "module": asker }], "createdAt": "x" });
        write_json(&dir.join("c.composition.json"), &manifest).unwrap();
        pds.put("carol.test", COMPOSITION, "c", manifest);
        for r in [dir.join("c.composition.json").display().to_string(), "carol.test/c".to_string()] {
            let Ref::Composition { comp, manifest, libs, .. } = resolve_ref(&cli.net, &dir, &[r.clone()]).await.unwrap_or_else(|e| panic!("{r}: {e}")).0 else { panic!("{r}") };
            assert_eq!(comp.manifest["instances"][1], json!({ "name": "a-pol", "module": pol }), "{r}");
            assert_eq!(comp.modules["a-pol"].id.as_deref(), Some(pol.as_str()));
            assert_eq!((manifest["instances"][1]["module"].as_str(), libs.iter().map(|l| l.id.as_str()).collect::<Vec<_>>()), (Some("pol"), vec![asker.as_str(), pol.as_str()]));
        }
        // add vendors the module's defaults; a pinned module's default comes from the lockfile only
        add(&cli, &["bob.test/asker".into(), "--dir".into(), dir.display().to_string()]).await.unwrap();
        let lock = read_lock(&dir.join(LOCK)).unwrap();
        assert_eq!(lock["modules"]["pol"]["uri"], pol);
        let vendored = lock["modules"]["pol"]["pin"].clone();
        // bob (or whoever holds his PDS) republishes pol: no calls, a new wasm
        publish_asker(&pds, "bob.test", "pol", b"\0asm-allow-all", json!({ "implements": [{ "nsid": H, "entry": "evaluate" }] }));
        let Ref::Module(lib) = resolve_ref(&cli.net, &dir, &["asker".to_string()]).await.unwrap().0 else { panic!() };
        let found = default_lib(&cli.net, &lib.id, lib.pinned.as_deref(), "pol").await.unwrap();
        assert_eq!((found.wasm.as_slice(), json!(found.pin())), (&b"\0asm-pol"[..], vendored), "the vendored default, not the republished one");
        // a ref pinned by @pin in a dir whose lockfile does not vendor its default: refused
        let elsewhere = tempdir("defaults-elsewhere");
        let pinned = format!("bob.test/asker@{}", lib.pin());
        let Ref::Module(lib) = resolve_ref(&cli.net, &elsewhere, &[pinned]).await.unwrap().0 else { panic!() };
        let refused = default_lib(&cli.net, &lib.id, lib.pinned.as_deref(), "pol").await.err().unwrap();
        assert!(refused.starts_with(&format!("hook default pol: {asker} is pinned, but")), "{refused}");
        // unpinned: the live record
        let Ref::Module(lib) = resolve_ref(&cli.net, &elsewhere, &["bob.test/asker".to_string()]).await.unwrap().0 else { panic!() };
        assert_eq!(default_lib(&cli.net, &lib.id, lib.pinned.as_deref(), "pol").await.unwrap().wasm, b"\0asm-allow-all");
        // two publishers' modules under one key cannot be staged or vendored side by side (review 4a L4)
        let carols = publish_asker(&pds, "carol.test", "pol", b"\0asm-carol", json!({}));
        write_json(&dir.join("two.composition.json"), &json!({ "instances": [{ "name": "a", "module": pol }, { "name": "b", "module": carols }] })).unwrap();
        let two = resolve_ref(&cli.net, &dir, &[dir.join("two.composition.json").display().to_string()]).await.err().unwrap();
        assert!(two.ends_with("are both module pol: one key, two modules"), "{two}");
    }

    #[tokio::test]
    async fn a_local_default_is_found_in_its_modules_repo() {
        // a module repo (atlibs-socialweb-computer-requester): modules/<key>, one target above them
        let pds = Pds::new();
        let repo = tempdir("module-repo");
        let built = repo.join("target/wasm32-unknown-unknown/release");
        std::fs::create_dir_all(&built).unwrap();
        for key in ["asker", "policy-open"] {
            std::fs::create_dir_all(repo.join("modules").join(key)).unwrap();
            std::fs::write(repo.join("modules").join(key).join("module.json"), r#"{"name":"m","calls":[]}"#).unwrap();
            std::fs::write(built.join(format!("{}.wasm", key.replace('-', "_"))), b"\0asm").unwrap();
        }
        let id = built.join("asker.wasm").display().to_string();
        assert_eq!(default_lib(&pds.cli(&repo).net, &id, None, "policy-open").await.unwrap().key, "policy-open");
        let missing = default_lib(&pds.cli(&repo).net, &id, None, "nope").await.err().unwrap();
        assert!(missing.starts_with("hook default nope: no modules/nope or examples/rust/nope above"), "{missing}");
    }

    #[tokio::test]
    async fn a_pin_covers_the_record_and_needs_32_hex_characters() {
        let pds = Pds::new();
        publish_asker(&pds, "bob.test", "m", b"\0asm-m", json!({ "calls": ["com.atproto.repo.getRecord"] }));
        let cli = pds.cli(&tempdir("pins"));
        let resolve = |r: String| {
            let net = cli.net.clone();
            async move { resolve_ref(&net, Path::new("."), &[r]).await.map(|_| ()) }
        };
        let Ref::Module(lib) = resolve_ref(&cli.net, Path::new("."), &["bob.test/m".to_string()]).await.unwrap().0 else { panic!() };
        let pin = lib.pin();
        assert!(resolve(format!("bob.test/m@{pin}")).await.is_ok());
        assert!(resolve(format!("bob.test/m@{}", &pin[..32])).await.is_ok());
        for short in [7, 8, 31] {
            assert!(resolve(format!("bob.test/m@{}", &pin[..short])).await.unwrap_err().contains("is too short"), "{short}");
        }
        // the same blob, its record's calls dropped (so it may call anything): a new pin; the old one still loads its version
        publish_asker(&pds, "bob.test", "m", b"\0asm-m", json!({}));
        let Ref::Module(old) = resolve_ref(&cli.net, Path::new("."), &[format!("bob.test/m@{pin}")]).await.unwrap().0 else { panic!() };
        assert_eq!((old.pin(), old.spec["calls"].clone()), (pin.clone(), json!(["com.atproto.repo.getRecord"])));
        // a record CID pins the record at its rkey, so a republish moves it
        assert!(resolve(format!("bob.test/m@{}", lib.cid.unwrap())).await.unwrap_err().contains("pinned to"));
    }

    /// verify end to end against the fake registry, offline: source file records -> manifest -> module, rebuilt
    /// by a stand-in engine that runs the container's command on the host in the bind-mounted dir (rustup a
    /// no-op); and one that exits 0 without running it (a docker look-alike seen on macOS) is named as the cause.
    /// The real engine: rust/host/cli.test.mjs.
    #[cfg(unix)]
    #[tokio::test]
    async fn verify_rebuilds_from_the_source_records_and_names_an_engine_that_built_nothing() {
        use std::os::unix::fs::PermissionsExt;
        let pds = Pds::new();
        let wasm = b"\0asm-built";
        let script = "mkdir -p target/out && printf '\\0asm-built' > target/out/m.wasm\n";
        let (file, file_cid) = pds.put("alice.test", "com.fedfork.exec.source.file", "f", json!({ "sha256": sha256_hex(script.as_bytes()), "text": script }));
        let manifest = json!({ "name": "m", "files": [{ "path": "rust/src/build.sh", "file": { "uri": file, "cid": file_cid } }], "output": { "sha256": sha256_hex(wasm) },
            "build": { "tool": "cargo", "version": "rustc 1.98.1 (x 2026-08-05)", "command": "bash rust/src/build.sh", "artifact": "target/out/m.wasm" } });
        let (source, source_cid) = pds.put("alice.test", "com.fedfork.exec.source.manifest", "m", manifest);
        let uri = pds.publish_module("alice.test", "m", wasm, None);
        let mut module = pds.get(&uri).unwrap();
        module["source"] = json!({ "uri": source, "cid": source_cid });
        pds.put("alice.test", MODULE, "m", module);
        let dir = tempdir("verify");
        let cli = pds.cli(&dir);
        let Ref::Module(lib) = resolve_ref(&cli.net, Path::new("."), &["alice.test/m".to_string()]).await.unwrap().0 else { panic!() };
        let engine = |name: &str, body: &str| {
            let path = dir.join(name);
            std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).unwrap();
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
            path.display().to_string()
        };
        // run --rm -v <dir>:/src:Z -w /src <image> bash -c <command>
        let host = engine("host-engine", r#"[ "$7" = docker.io/library/rust:1.98.1 ] && cd "${4%%:/src:Z}" && exec bash -c "rustup() { :; }; ${10}""#);
        let report = rebuild(&cli, &lib, &host, &dir.join("work")).await.unwrap();
        assert_eq!(report, json!({ "built": sha256_hex(wasm), "manifest": sha256_hex(wasm), "published": sha256_hex(wasm), "reproducible": true }));
        let lying = engine("lying-engine", "exit 0");
        let err = rebuild(&cli, &lib, &lying, &dir.join("work2")).await.unwrap_err();
        assert!(err.contains("exited 0 but the build left no target/out/m.wasm") && err.contains("CONTAINER_ENGINE"), "{err}");
    }

    pub fn tempdir(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!("atlibs-cli-{name}-{}", atlibs_host::store::uuid()));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }
}