yah-mesofact-bundle 0.8.26

//! The **node runtime-asset tier** (R746-F1, W272 §2/§7): the serve binary a
//! *vanilla* bundle does not carry.
//!
//! A self-contained bundle (`runtime = "self"`) closes over its own
//! `bins/<triple>/serve`, so its binary's bytes are covered by the bundle
//! digest and resolution is a path join. A vanilla bundle
//! (`runtime = "mesofact/<ver>"`) deliberately carries nothing — the whole
//! point is that one ~70MB V8-carrying binary is a **node-level** asset shared
//! by every site at that version, not a per-site blob re-uploaded and
//! re-downloaded per deploy. That trade only works if the node can *get* the
//! binary, which is what this module does. Until it existed, a vanilla bundle
//! assembled fine and then had nothing to exec.
//!
//! ```text
//! <cache>/bundles/<digest>/…                       # materialized bundle trees
//! <cache>/runtimes/mesofact/0.8.20/<triple>/serve   # stock runtime asset
//! <cache>/runtimes/acme/site/1.2.3/<triple>/serve   # namespaced CUSTOM runtime
//! <cache>/runtimes/almanac-feed/0.8.22/<triple>/almanac-feed   # a SIDECAR asset
//! ```
//!
//! **The tier is not serve-only** (R746-T3). One runtime ref names one binary,
//! whose filename the published manifest records — `serve` for a mesofact
//! runtime, `almanac-feed` for the feed-fetch sidecar. Before this, the feed
//! fetcher could only reach a node inside a self-contained bundle's `bins/`,
//! which meant a *vanilla* bundle with a feed tier still needed a cross-built
//! musl binary on the syncing machine. That single path was the last thing
//! standing between yah.dev and "a templates-only edit deploys from a laptop
//! with no Rust toolchain".
//!
//! **The path is namespaced from day one**, though nothing publishes a custom
//! runtime yet (W272 §7): custom runtimes cache exactly like stock ones, and
//! generalizing the key costs nothing while no node has written one to disk and
//! costs a fleet-wide migration afterwards. Stock is the unnamespaced case —
//! `mesofact/<ver>` is `<name>/<ver>`, `acme/site/1.2.3` is
//! `<namespace>/<name>/<ver>`. See [`RuntimeRef`].
//!
//! **Trust comes from the content address, not the transport.** The by-name key
//! (`runtimes/…/<triple>.toml`) resolves to a small published manifest naming
//! the binary's blake3; the bytes themselves come from the same
//! `blobs/<blake3>` namespace bundles use, and are verified against that hash
//! before anything is written where a fork could reach it. This is the property
//! the self-contained shape got for free from the bundle digest, and losing it
//! would make vanilla a strictly worse trust posture than the shape it
//! replaces. It also means the runtime asset dedupes against a self-contained
//! bundle carrying the identical binary — same bytes, same blob.
//!
//! Sharing the bundle store rather than inventing a second one is deliberate:
//! the node already reads it unauthenticated over plain HTTPS
//! ([`yah_object_store::HttpReadOnlyObjectStore`]), so a runtime asset needs no
//! new origin, no new credential, and no new cache-invalidation story.

use std::fmt;
use std::path::PathBuf;

use crate::BundleError;

/// A runtime the node resolves by *name*, as opposed to `runtime = "self"`
/// which resolves inside the bundle.
///
/// Wire form is slash-separated, either `<name>/<version>` (stock, e.g.
/// `mesofact/0.8.20`) or `<namespace>/<name>/<version>` (a custom runtime
/// published under an org/project, e.g. `acme/site/1.2.3`). The namespace is
/// what keeps two orgs' `site` runtimes from colliding in one node cache.
///
/// This parses the *general* form even though [`BundleRuntime::parse`] accepts
/// only `mesofact/<ver>` today: the node cache layout is the thing that is
/// expensive to change later (bytes on disk across a fleet), the manifest field
/// is not (R746-F5/F6 rework it into a constraint expression). Accepting a
/// namespaced ref here early is forward-compatible and unreachable in practice
/// until a manifest can express one.
///
/// [`BundleRuntime::parse`]: crate::BundleRuntime::parse
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RuntimeRef {
    namespace: Option<String>,
    name: String,
    version: String,
}

impl RuntimeRef {
    /// Parse the wire form (`<name>/<ver>` or `<ns>/<name>/<ver>`).
    ///
    /// Every segment is validated to `[A-Za-z0-9._+-]+` and rejected if it is
    /// `.` or `..`. That is not decoration: these segments are joined onto a
    /// node's cache directory, so a permissive parse would let a published
    /// manifest name a write target outside the cache. Same reasoning as
    /// `store::checked_rel` for bundle content paths.
    pub fn parse(s: &str) -> Result<Self, BundleError> {
        let segments: Vec<&str> = s.split('/').collect();
        let (namespace, name, version) = match segments.as_slice() {
            [name, version] => (None, *name, *version),
            [ns, name, version] => (Some(*ns), *name, *version),
            _ => {
                return Err(BundleError::Runtime(format!(
                    "{s:?} is not a resolvable runtime reference (expected \
                     \"<name>/<version>\" or \"<namespace>/<name>/<version>\"; \
                     \"self\" is resolved inside the bundle, not here)"
                )))
            }
        };
        for seg in namespace.into_iter().chain([name, version]) {
            check_segment(seg, s)?;
        }
        Ok(Self {
            namespace: namespace.map(str::to_string),
            name: name.to_string(),
            version: version.to_string(),
        })
    }

    /// Publishing namespace (`None` for a stock runtime).
    pub fn namespace(&self) -> Option<&str> {
        self.namespace.as_deref()
    }

    /// Runtime name, e.g. `mesofact`.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Exact runtime version, e.g. `0.8.20`.
    pub fn version(&self) -> &str {
        &self.version
    }

    /// The wire string this was parsed from.
    pub fn as_wire(&self) -> String {
        match &self.namespace {
            Some(ns) => format!("{ns}/{}/{}", self.name, self.version),
            None => format!("{}/{}", self.name, self.version),
        }
    }

    /// Cache-relative path to the binary named `bin`:
    /// `runtimes/[<ns>/]<name>/<ver>/<triple>/<bin>`.
    ///
    /// `bin` is a bare filename, never a path — it comes from a published
    /// manifest, and joining an unchecked one onto the cache dir would be the
    /// same write-outside-the-cache primitive [`RuntimeRef::parse`] exists to
    /// deny. Callers pass a compile-time constant ([`SERVE_BIN`],
    /// kamaji's `FEED_BIN_NAME`); [`check_bin_name`] enforces it for the
    /// manifest-supplied side.
    pub fn cache_rel_bin(&self, triple: &str, bin: &str) -> PathBuf {
        self.cache_rel_dir(triple).join(bin)
    }

    /// Cache-relative directory holding this runtime's binaries for `triple`:
    /// `runtimes/[<ns>/]<name>/<ver>/<triple>`.
    pub fn cache_rel_dir(&self, triple: &str) -> PathBuf {
        let mut p = PathBuf::from("runtimes");
        if let Some(ns) = &self.namespace {
            p.push(ns);
        }
        p.push(&self.name);
        p.push(&self.version);
        p.push(triple);
        p
    }

    /// Cache-relative path to the serve binary:
    /// `runtimes/[<ns>/]<name>/<ver>/<triple>/serve`. The [`SERVE_BIN`] case of
    /// [`cache_rel_bin`](Self::cache_rel_bin).
    pub fn cache_rel_serve(&self, triple: &str) -> PathBuf {
        self.cache_rel_bin(triple, SERVE_BIN)
    }

    /// Object-store key of the published asset manifest for `triple`:
    /// `runtimes/[<ns>/]<name>/<ver>/<triple>.toml`.
    ///
    /// By *name*, not by content — this is the one mutable-shaped key in the
    /// scheme, and it is what turns "mesofact/0.8.20 on x86_64 musl" into a
    /// blake3 the node can then verify bytes against.
    pub fn asset_manifest_key(&self, triple: &str) -> String {
        let mut key = String::from("runtimes/");
        if let Some(ns) = &self.namespace {
            key.push_str(ns);
            key.push('/');
        }
        key.push_str(&self.name);
        key.push('/');
        key.push_str(&self.version);
        key.push('/');
        key.push_str(triple);
        key.push_str(".toml");
        key
    }
}

impl fmt::Display for RuntimeRef {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.as_wire())
    }
}

/// Filename of a mesofact runtime's binary inside a resolved runtime-asset
/// directory. Mirrors `bins/<triple>/serve` in a self-contained bundle so the
/// two arms of resolution hand the same shape to the fork.
pub const SERVE_BIN: &str = "serve";

/// Reject a binary name that is anything other than a plain filename.
///
/// Applied to the *manifest-supplied* name in [`ensure_runtime_asset`]: a
/// published manifest is remote input, and its `bin` field is joined onto the
/// node's cache dir. Same reasoning — and the same character class — as
/// [`RuntimeRef::parse`]'s segment check.
pub fn check_bin_name(bin: &str) -> Result<(), BundleError> {
    check_segment(bin, bin).map_err(|_| {
        BundleError::Runtime(format!(
            "runtime asset binary name {bin:?} is not a plain filename \
             (must be non-empty [A-Za-z0-9._+-] and never \".\" or \"..\")"
        ))
    })
}

/// Reject anything that isn't a plain, path-safe identifier segment.
fn check_segment(seg: &str, whole: &str) -> Result<(), BundleError> {
    let ok = !seg.is_empty()
        && seg != "."
        && seg != ".."
        && seg
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'+' | b'-'));
    if ok {
        return Ok(());
    }
    Err(BundleError::Runtime(format!(
        "runtime reference {whole:?} has an unusable segment {seg:?} \
         (segments must be non-empty [A-Za-z0-9._+-] and never \".\" or \"..\")"
    )))
}

#[cfg(feature = "store")]
pub use asset::{
    ensure_runtime_asset, publish_runtime_asset, RuntimeAsset, RuntimeAssetManifest,
    RUNTIME_ASSET_SCHEMA_VERSION,
};

/// Runtime ref + binary name for the almanac feed-fetch sidecar (R746-T3).
///
/// The fetcher is versioned with yubaba, so it gets its own unnamespaced
/// runtime ref (`almanac-feed/<ver>`) rather than riding the mesofact runtime's.
pub const FEED_BIN: &str = "almanac-feed";

#[cfg(feature = "store")]
mod asset {
    use std::fs;
    use std::path::{Path, PathBuf};

    use serde::{Deserialize, Serialize};
    use yah_object_store::ObjectStore;

    use super::{check_bin_name, RuntimeRef};
    use crate::store::{io_ctx, store_err, touch_access};
    use crate::{blob_key, BundleError, BundleHash};

    /// Wire version of the runtime-asset manifest. Separate from the bundle
    /// manifest's [`SCHEMA_VERSION`](crate::SCHEMA_VERSION): they version
    /// independent objects and a node speaking one may not speak the other.
    pub const RUNTIME_ASSET_SCHEMA_VERSION: u32 = 1;

    /// The small published object at `runtimes/…/<triple>.toml` — the by-name
    /// pointer into the content-addressed blob space.
    ///
    /// ```toml
    /// schema_version = 1
    /// runtime = "mesofact/0.8.20"
    /// triple  = "x86_64-unknown-linux-musl"
    /// bin     = "serve"   # filename the bytes land under in the node cache
    /// serve   = "9f3c…"   # blake3 of that binary, in blobs/<blake3>
    /// ```
    ///
    /// It carries `runtime` and `triple` redundantly with its own key on
    /// purpose: a node checks them against what it asked for, so an asset
    /// misfiled under the wrong version (a publish-script bug, a copy-paste in
    /// a pipeline) is caught by the fetcher instead of becoming a site served
    /// by the wrong binary. `bin` is checked the same way, which is what makes
    /// a sidecar asset (`almanac-feed`) safe to resolve through the same tier.
    ///
    /// **One ref, one binary.** The key is per (runtime × triple), so a runtime
    /// that shipped two executables would need two refs — which is the right
    /// shape anyway: `almanac-feed` is versioned with yubaba, not with
    /// mesofact, and pretending otherwise would tie their release cadences
    /// together for no reason.
    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
    pub struct RuntimeAssetManifest {
        /// Wire-format version. Always [`RUNTIME_ASSET_SCHEMA_VERSION`] today.
        pub schema_version: u32,
        /// The runtime this asset is for, in [`RuntimeRef`] wire form.
        pub runtime: String,
        /// Target triple these bytes were built for.
        pub triple: String,
        /// Filename these bytes are written under in the node's runtime cache,
        /// and the name a resolver asks for. `serve` for a mesofact runtime.
        ///
        /// Defaulted rather than required so a manifest published before the
        /// field existed still parses as the serve-only case it was.
        #[serde(default = "default_bin")]
        pub bin: String,
        /// BLAKE3 of the binary; its blob lives at `blobs/<serve>`.
        pub serve: BundleHash,
    }

    fn default_bin() -> String {
        super::SERVE_BIN.to_string()
    }

    impl RuntimeAssetManifest {
        /// Parse a published asset manifest, rejecting an unknown schema version.
        pub fn from_toml_str(s: &str) -> Result<Self, BundleError> {
            let m: RuntimeAssetManifest =
                toml::from_str(s).map_err(|e| BundleError::Manifest(e.to_string()))?;
            if m.schema_version != RUNTIME_ASSET_SCHEMA_VERSION {
                return Err(BundleError::SchemaVersion {
                    found: m.schema_version,
                });
            }
            Ok(m)
        }

        /// Serialize to the published TOML form.
        pub fn to_toml_string(&self) -> Result<String, BundleError> {
            toml::to_string_pretty(self).map_err(|e| BundleError::Manifest(e.to_string()))
        }
    }

    /// Outcome of [`publish_runtime_asset`].
    #[derive(Debug, Clone)]
    pub struct RuntimeAsset {
        /// The published manifest.
        pub manifest: RuntimeAssetManifest,
        /// Key of the manifest object.
        pub manifest_key: String,
        /// Key of the binary blob.
        pub blob_key: String,
        /// False when the store already held the binary's bytes — the whole
        /// point of sharing `blobs/` with bundles. A patch release that did not
        /// change the serve binary uploads nothing.
        pub blob_uploaded: bool,
    }

    /// Publish `serve_bin` as the runtime asset for `runtime` × `triple`,
    /// landing on the node under the filename `bin` ([`SERVE_BIN`] for a
    /// mesofact runtime; `almanac-feed` for the feed-fetch sidecar).
    ///
    /// PUTs the binary at `blobs/<blake3>` (skipped when already present, the
    /// same append-only dedupe `publish_bundle` uses) and the naming manifest at
    /// `runtimes/…/<triple>.toml`. The manifest key is by-name and therefore
    /// overwritable: re-publishing the same version with different bytes is a
    /// *repoint*, which is exactly what a re-cut release is. Nodes that already
    /// cached the old bytes keep them — the cache is keyed by name, so a repoint
    /// only reaches nodes that had not fetched it yet. Publish under a new
    /// version rather than repointing one that is live.
    pub fn publish_runtime_asset(
        store: &dyn ObjectStore,
        runtime: &RuntimeRef,
        triple: &str,
        bin: &str,
        serve_bin: &Path,
    ) -> Result<RuntimeAsset, BundleError> {
        check_bin_name(bin)?;
        let bytes = fs::read(serve_bin).map_err(io_ctx(serve_bin))?;
        let hash = BundleHash::of(&bytes);
        let bkey = blob_key(&hash);
        let blob_uploaded = if store.head(&bkey).map_err(store_err)? {
            false
        } else {
            store.put(&bkey, bytes).map_err(store_err)?;
            true
        };

        let manifest = RuntimeAssetManifest {
            schema_version: RUNTIME_ASSET_SCHEMA_VERSION,
            runtime: runtime.as_wire(),
            triple: triple.to_string(),
            bin: bin.to_string(),
            serve: hash,
        };
        let mkey = runtime.asset_manifest_key(triple);
        store
            .put(&mkey, manifest.to_toml_string()?.into_bytes())
            .map_err(store_err)?;

        Ok(RuntimeAsset {
            manifest,
            manifest_key: mkey,
            blob_key: bkey,
            blob_uploaded,
        })
    }

    /// Resolve the binary named `bin` for `runtime` × `triple` under
    /// `cache_dir`, fetching and verifying it from `store` on a miss. Returns
    /// the executable path.
    ///
    /// `bin` is the caller's *expectation* — a compile-time constant at every
    /// call site — and the published manifest's own `bin` is checked against
    /// it, exactly as `runtime` and `triple` are. That is what keeps the
    /// cache-hit fast path possible: the resolver knows the filename before it
    /// has fetched anything, so a warm resolve touches the network zero times.
    ///
    /// Resolution is **purely** `<cache_dir>/runtimes/…/<bin>`: no `PATH`
    /// lookup, no scan for a plausible binary elsewhere on the box, no reuse of
    /// another bundle's `bins/`. A node that cannot fetch the named version
    /// fails the deploy — serving a site with a binary nobody named is a worse
    /// outcome than not serving it, because it succeeds silently.
    ///
    /// A cache hit returns immediately without re-hashing: the fetch verifies
    /// before an atomic rename, so a file visible at the final path was already
    /// checked, and re-blake3-ing ~70MB on every deploy buys nothing against a
    /// threat model where an attacker who can write the cache dir can also
    /// write the process that reads it.
    ///
    /// **Not evicted by [`BundleCache`](crate::BundleCache).** Runtime assets
    /// live outside `bundles/`, so the bundle LRU neither counts nor reclaims
    /// them — deliberately: one asset backs every resident serve process at that
    /// version, and evicting it under a bundle-shaped budget would break the
    /// next restart of sites that had nothing to do with the deploy that
    /// tripped the budget. An access marker is touched on every resolve so a
    /// future runtime-tier reclaim has recency to work from.
    pub fn ensure_runtime_asset(
        store: &dyn ObjectStore,
        cache_dir: &Path,
        runtime: &RuntimeRef,
        triple: &str,
        bin: &str,
    ) -> Result<PathBuf, BundleError> {
        check_bin_name(bin)?;
        let dir = cache_dir.join(runtime.cache_rel_dir(triple));
        let dest = dir.join(bin);
        if dest.is_file() {
            let _ = touch_access(&dir);
            return Ok(dest);
        }

        // By-name pointer → blake3.
        let mkey = runtime.asset_manifest_key(triple);
        let manifest_bytes = store
            .get(&mkey)
            .map_err(store_err)?
            .ok_or_else(|| BundleError::RuntimeAssetMissing {
                runtime: runtime.as_wire(),
                triple: triple.to_string(),
                location: store.locate(&mkey),
            })?;
        let manifest_text = String::from_utf8(manifest_bytes).map_err(|e| {
            BundleError::Manifest(format!("runtime asset manifest {mkey} not utf-8: {e}"))
        })?;
        let manifest = RuntimeAssetManifest::from_toml_str(&manifest_text)?;
        if manifest.runtime != runtime.as_wire() || manifest.triple != triple {
            return Err(BundleError::Manifest(format!(
                "runtime asset at {mkey} declares {}/{} but was published under {}/{triple} \
                 — refusing to serve a bundle with a binary built for something else",
                manifest.runtime,
                manifest.triple,
                runtime.as_wire(),
            )));
        }
        if manifest.bin != bin {
            return Err(BundleError::Manifest(format!(
                "runtime asset at {mkey} publishes the binary {:?} but {} was asked for \
                 {bin:?} — refusing to fork a binary under a name it was not published as",
                manifest.bin,
                runtime.as_wire(),
            )));
        }

        // Content address → bytes, verified before they land anywhere forkable.
        let bkey = blob_key(&manifest.serve);
        let bytes = store
            .get(&bkey)
            .map_err(store_err)?
            .ok_or_else(|| BundleError::MissingBlob {
                key: store.locate(&bkey),
                path: format!("{}/{triple}/{bin}", runtime.as_wire()),
            })?;
        let actual = BundleHash::of(&bytes);
        if actual != manifest.serve {
            return Err(BundleError::HashMismatch {
                what: format!("runtime asset {}/{triple}", runtime.as_wire()),
                expected: manifest.serve.to_string(),
                actual: actual.to_string(),
            });
        }

        // Stage beside the destination and rename: the final path is never
        // visible holding a partial or unverified binary, and two concurrent
        // deploys of the same version race harmlessly (both wrote identical,
        // verified bytes).
        fs::create_dir_all(&dir).map_err(io_ctx(&dir))?;
        let staging = dir.join(format!(".staging-{}", manifest.serve.as_str()));
        fs::write(&staging, &bytes).map_err(io_ctx(&staging))?;
        set_executable(&staging)?;
        fs::rename(&staging, &dest).map_err(io_ctx(&dest))?;
        let _ = touch_access(&dir);
        Ok(dest)
    }

    /// Mark a staged runtime binary executable before it becomes visible.
    #[cfg(unix)]
    fn set_executable(path: &Path) -> Result<(), BundleError> {
        use std::os::unix::fs::PermissionsExt;
        let mut perms = fs::metadata(path).map_err(io_ctx(path))?.permissions();
        perms.set_mode(0o755);
        fs::set_permissions(path, perms).map_err(io_ctx(path))
    }

    #[cfg(not(unix))]
    fn set_executable(_path: &Path) -> Result<(), BundleError> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn stock_is_the_unnamespaced_case_and_custom_is_namespaced() {
        let stock = RuntimeRef::parse("mesofact/0.8.20").unwrap();
        assert_eq!(stock.namespace(), None);
        assert_eq!(stock.name(), "mesofact");
        assert_eq!(stock.version(), "0.8.20");
        assert_eq!(stock.as_wire(), "mesofact/0.8.20");

        let custom = RuntimeRef::parse("acme/site/1.2.3").unwrap();
        assert_eq!(custom.namespace(), Some("acme"));
        assert_eq!(custom.name(), "site");
        assert_eq!(custom.version(), "1.2.3");
        assert_eq!(custom.as_wire(), "acme/site/1.2.3");
    }

    /// The layout this ticket commits a fleet to. Changing it after nodes have
    /// written to disk is a migration, so pin both arms.
    #[test]
    fn cache_layout_is_namespace_name_version_triple() {
        let triple = "x86_64-unknown-linux-musl";
        assert_eq!(
            RuntimeRef::parse("mesofact/0.8.20")
                .unwrap()
                .cache_rel_serve(triple),
            PathBuf::from("runtimes/mesofact/0.8.20/x86_64-unknown-linux-musl/serve")
        );
        assert_eq!(
            RuntimeRef::parse("acme/site/1.2.3")
                .unwrap()
                .cache_rel_serve(triple),
            PathBuf::from("runtimes/acme/site/1.2.3/x86_64-unknown-linux-musl/serve")
        );
        assert_eq!(
            RuntimeRef::parse("mesofact/0.8.20")
                .unwrap()
                .asset_manifest_key(triple),
            "runtimes/mesofact/0.8.20/x86_64-unknown-linux-musl.toml"
        );
    }

    /// These segments are joined onto a node's cache dir, so a permissive parse
    /// is a write-outside-the-cache primitive handed to whoever publishes a
    /// manifest.
    #[test]
    fn traversal_and_junk_segments_are_rejected() {
        assert!(RuntimeRef::parse("mesofact/../../etc").is_err());
        assert!(RuntimeRef::parse("../mesofact/0.1.0").is_err());
        assert!(RuntimeRef::parse("mesofact/.").is_err());
        assert!(RuntimeRef::parse("mesofact/").is_err());
        assert!(RuntimeRef::parse("/0.1.0").is_err());
        assert!(RuntimeRef::parse("a/b/c/d").is_err());
        assert!(RuntimeRef::parse("mesofact").is_err());
        // `self` is resolved inside the bundle and must never reach this path.
        assert!(RuntimeRef::parse("self").is_err());
        // A whitespace or shell-metacharacter segment can't become a path.
        assert!(RuntimeRef::parse("mesofact/0.8 20").is_err());
        assert!(RuntimeRef::parse("mesofact/$(id)").is_err());
    }

    #[cfg(feature = "store")]
    mod asset {
        use super::*;
        use crate::runtime::{ensure_runtime_asset, publish_runtime_asset, RuntimeAssetManifest};
        use crate::{blob_key, BundleError, BundleHash};
        use tempfile::TempDir;
        use yah_object_store::{InMemoryObjectStore, ObjectStore};

        const TRIPLE: &str = "x86_64-unknown-linux-musl";

        fn write_bin(dir: &TempDir, bytes: &[u8]) -> std::path::PathBuf {
            let p = dir.path().join("mesofact-serve");
            std::fs::write(&p, bytes).unwrap();
            p
        }

        #[test]
        fn publish_then_resolve_round_trips_and_lands_executable() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/0.8.20").unwrap();

            let published =
                publish_runtime_asset(&store, &rref, TRIPLE, SERVE_BIN, &write_bin(&src, b"ELF serve")).unwrap();
            assert!(published.blob_uploaded);

            let resolved = ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap();
            assert_eq!(
                resolved,
                cache.path().join("runtimes/mesofact/0.8.20").join(TRIPLE).join("serve")
            );
            assert_eq!(std::fs::read(&resolved).unwrap(), b"ELF serve");
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let mode = std::fs::metadata(&resolved).unwrap().permissions().mode();
                assert!(mode & 0o111 != 0, "a runtime asset must land forkable, mode={mode:o}");
            }
        }

        /// The sharing property the vanilla shape exists for: a second site at
        /// the same runtime version fetches nothing. A per-bundle copy would be
        /// the self-contained shape wearing a different manifest.
        #[test]
        fn a_second_resolve_at_the_same_version_is_a_cache_hit() {
            struct CountingStore {
                inner: InMemoryObjectStore,
                gets: std::sync::atomic::AtomicUsize,
            }
            impl ObjectStore for CountingStore {
                fn put(&self, key: &str, data: Vec<u8>) -> Result<(), yah_object_store::Error> {
                    self.inner.put(key, data)
                }
                fn get(&self, key: &str) -> Result<Option<Vec<u8>>, yah_object_store::Error> {
                    self.gets.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    self.inner.get(key)
                }
                fn delete(&self, key: &str) -> Result<(), yah_object_store::Error> {
                    self.inner.delete(key)
                }
                fn list_prefix(&self, p: &str) -> Result<Vec<String>, yah_object_store::Error> {
                    self.inner.list_prefix(p)
                }
            }

            let store = CountingStore {
                inner: InMemoryObjectStore::new(),
                gets: std::sync::atomic::AtomicUsize::new(0),
            };
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/0.8.20").unwrap();
            publish_runtime_asset(&store, &rref, TRIPLE, SERVE_BIN, &write_bin(&src, b"ELF serve")).unwrap();

            ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap();
            let after_cold = store.gets.load(std::sync::atomic::Ordering::SeqCst);
            assert!(after_cold >= 2, "cold resolve fetches manifest + blob");

            // Second site, same node, same runtime version.
            ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap();
            assert_eq!(
                store.gets.load(std::sync::atomic::Ordering::SeqCst),
                after_cold,
                "a second bundle at the same runtime version must fetch nothing"
            );
        }

        /// Two versions are two assets; resolving one must not satisfy the other.
        #[test]
        fn versions_do_not_alias() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let old = RuntimeRef::parse("mesofact/0.8.19").unwrap();
            let new = RuntimeRef::parse("mesofact/0.8.20").unwrap();
            publish_runtime_asset(&store, &old, TRIPLE, SERVE_BIN, &write_bin(&src, b"old serve")).unwrap();
            ensure_runtime_asset(&store, cache.path(), &old, TRIPLE, SERVE_BIN).unwrap();

            // 0.8.20 was never published — the cached 0.8.19 must not stand in.
            let err = ensure_runtime_asset(&store, cache.path(), &new, TRIPLE, SERVE_BIN).unwrap_err();
            assert!(
                matches!(err, BundleError::RuntimeAssetMissing { .. }),
                "got {err:?}"
            );
        }

        /// Verify #2: an unfetchable version fails loudly, naming the version
        /// and where it looked, and falls back to nothing.
        #[test]
        fn an_unpublished_version_fails_naming_version_and_location() {
            let store = InMemoryObjectStore::new();
            let cache = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/9.9.9").unwrap();

            let err = ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap_err();
            let msg = err.to_string();
            assert!(msg.contains("mesofact/9.9.9"), "got {msg}");
            assert!(msg.contains(TRIPLE), "got {msg}");
            assert!(
                msg.contains("runtimes/mesofact/9.9.9/x86_64-unknown-linux-musl.toml"),
                "the message must name what it looked for, got {msg}"
            );
            assert!(
                !cache.path().join("runtimes").exists(),
                "a failed resolve must not leave a partial runtime dir"
            );
        }

        /// The trust property vanilla must not lose relative to a self-contained
        /// bundle, whose binary is covered by the bundle digest.
        #[test]
        fn tampered_bytes_are_refused_and_never_written() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/0.8.20").unwrap();
            let published =
                publish_runtime_asset(&store, &rref, TRIPLE, SERVE_BIN, &write_bin(&src, b"ELF serve")).unwrap();

            // A hostile origin swaps the blob's bytes under its content address.
            store.put(&published.blob_key, b"malware".to_vec()).unwrap();

            let err = ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap_err();
            assert!(matches!(err, BundleError::HashMismatch { .. }), "got {err:?}");
            assert!(
                !cache.path().join(rref.cache_rel_serve(TRIPLE)).exists(),
                "unverified bytes must never reach the resolved path"
            );
        }

        /// An asset misfiled under the wrong version is caught by the fetcher
        /// rather than becoming a site served by the wrong binary.
        #[test]
        fn a_misfiled_asset_manifest_is_refused() {
            let store = InMemoryObjectStore::new();
            let cache = TempDir::new().unwrap();
            let asked = RuntimeRef::parse("mesofact/0.8.20").unwrap();

            // Published at 0.8.20's key but declaring 0.8.19 inside.
            let bytes = b"ELF serve".to_vec();
            let hash = BundleHash::of(&bytes);
            store.put(&blob_key(&hash), bytes).unwrap();
            let manifest = RuntimeAssetManifest {
                schema_version: 1,
                runtime: "mesofact/0.8.19".to_string(),
                triple: TRIPLE.to_string(),
                bin: SERVE_BIN.to_string(),
                serve: hash,
            };
            store
                .put(
                    &asked.asset_manifest_key(TRIPLE),
                    manifest.to_toml_string().unwrap().into_bytes(),
                )
                .unwrap();

            let err = ensure_runtime_asset(&store, cache.path(), &asked, TRIPLE, SERVE_BIN).unwrap_err();
            assert!(err.to_string().contains("0.8.19"), "got {err}");
        }

        /// A runtime asset and a self-contained bundle carrying the identical
        /// binary share one blob — the dedupe that makes sharing the bundle
        /// store rather than inventing a second one worth it.
        #[test]
        fn identical_binaries_dedupe_against_the_bundle_blob_space() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/0.8.20").unwrap();
            let bin = write_bin(&src, b"ELF serve");

            let first = publish_runtime_asset(&store, &rref, TRIPLE, SERVE_BIN, &bin).unwrap();
            assert!(first.blob_uploaded);
            let other = RuntimeRef::parse("acme/site/1.0.0").unwrap();
            let second = publish_runtime_asset(&store, &other, TRIPLE, SERVE_BIN, &bin).unwrap();
            assert!(!second.blob_uploaded, "identical bytes must not re-upload");
            assert_eq!(first.blob_key, second.blob_key);
        }

        /// R746-T3: the tier is not serve-only. The feed fetcher resolves
        /// through it under its own name, which is what lets a *vanilla*
        /// bundle carry a feed tier without carrying a binary.
        #[test]
        fn a_sidecar_asset_resolves_under_its_own_binary_name() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let feed = RuntimeRef::parse("almanac-feed/0.8.22").unwrap();

            publish_runtime_asset(&store, &feed, TRIPLE, FEED_BIN, &write_bin(&src, b"ELF feed"))
                .unwrap();
            let resolved = ensure_runtime_asset(&store, cache.path(), &feed, TRIPLE, FEED_BIN)
                .unwrap();

            assert_eq!(
                resolved,
                cache
                    .path()
                    .join("runtimes/almanac-feed/0.8.22")
                    .join(TRIPLE)
                    .join(FEED_BIN)
            );
            assert_eq!(std::fs::read(&resolved).unwrap(), b"ELF feed");
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let mode = std::fs::metadata(&resolved).unwrap().permissions().mode();
                assert!(mode & 0o111 != 0, "a sidecar asset must land forkable");
            }
        }

        /// The binary name is checked like `runtime` and `triple` are: asking
        /// for `serve` and being handed the feed fetcher is exactly the
        /// misfiling the by-name key makes possible, and it would fork a
        /// binary that exits on `serve --bundle`.
        #[test]
        fn an_asset_published_under_another_binary_name_is_refused() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let rref = RuntimeRef::parse("mesofact/0.8.22").unwrap();

            publish_runtime_asset(&store, &rref, TRIPLE, FEED_BIN, &write_bin(&src, b"ELF feed"))
                .unwrap();

            let err =
                ensure_runtime_asset(&store, cache.path(), &rref, TRIPLE, SERVE_BIN).unwrap_err();
            let msg = err.to_string();
            assert!(msg.contains(FEED_BIN) && msg.contains(SERVE_BIN), "got {msg}");
            assert!(
                !cache.path().join(rref.cache_rel_serve(TRIPLE)).exists(),
                "a refused resolve must leave nothing forkable behind"
            );
        }

        /// Two assets at one version × triple do not collide, because they are
        /// two refs — and the cache dirs are disjoint, so the mesofact runtime
        /// never shadows the fetcher.
        #[test]
        fn serve_and_feed_assets_do_not_share_a_cache_slot() {
            let store = InMemoryObjectStore::new();
            let src = TempDir::new().unwrap();
            let cache = TempDir::new().unwrap();
            let serve_ref = RuntimeRef::parse("mesofact/0.8.22").unwrap();
            let feed_ref = RuntimeRef::parse("almanac-feed/0.8.22").unwrap();

            publish_runtime_asset(&store, &serve_ref, TRIPLE, SERVE_BIN, &write_bin(&src, b"S"))
                .unwrap();
            let feed_src = TempDir::new().unwrap();
            publish_runtime_asset(&store, &feed_ref, TRIPLE, FEED_BIN, &write_bin(&feed_src, b"F"))
                .unwrap();

            let s = ensure_runtime_asset(&store, cache.path(), &serve_ref, TRIPLE, SERVE_BIN)
                .unwrap();
            let f =
                ensure_runtime_asset(&store, cache.path(), &feed_ref, TRIPLE, FEED_BIN).unwrap();
            assert_ne!(s, f);
            assert_eq!(std::fs::read(&s).unwrap(), b"S");
            assert_eq!(std::fs::read(&f).unwrap(), b"F");
        }

        /// A published manifest is remote input; its `bin` is joined onto the
        /// node's cache dir. Same traversal posture as the ref's segments.
        #[test]
        fn a_traversing_binary_name_is_refused() {
            for bad in ["../serve", "a/b", "..", ".", "", "serve bin"] {
                assert!(
                    crate::runtime::check_bin_name(bad).is_err(),
                    "{bad:?} must not become a cache path"
                );
            }
            assert!(crate::runtime::check_bin_name(FEED_BIN).is_ok());
            assert!(crate::runtime::check_bin_name(SERVE_BIN).is_ok());
        }
    }
}