holger-server-lib 0.6.9

//! The holger server crate: turns a RON config into a running artifact server.
//!
//! Config is a directed graph of three node kinds — [`Repository`] (a backend
//! plus its upstreams), [`ExposedEndpoint`] (a gRPC + optional HTTP/OCI gateway),
//! and [`StorageEndpoint`] (disk or a znippy archive). [`wire_holger`] resolves
//! that graph in passes: index every node by `ron_name`, wire forward refs
//! (repo→storage, repo→exposed) and reverse links (exposed→repos), then build a
//! per-endpoint [`FastRoutes`] table. Any repo with configured upstreams is
//! wrapped in a `ProxyBackend` at this stage so a 404 in the local archive falls
//! through to the upstream pull-through automatically.
//!
//! Two entry modes share the same wired graph: call [`Holger::fetch`]/[`Holger::put`]
//! in-process, or [`Holger::start`] to spin up the tonic gRPC services (and HTTP
//! gateways) fronted by shared auth ([`auth`]) and one append-only audit sink
//! ([`audit`]). Boot is deliberately fault-tolerant — a single bad archive is
//! disabled and logged (see [`Holger::instantiate_backends`]) rather than sinking
//! the whole server; only an all-repos failure hard-fails.
//!
//! Gotcha: the wiring uses raw `*const` pointers between nodes, so the graph is
//! valid only while the owning `Vec`s stay put — never mutate or reallocate
//! `repositories`/`exposed_endpoints`/`storage_endpoints` after `wire_holger`.
//! Related seams: `ron_url`/`ron_http_url` parse as bare `SocketAddr`s, so any
//! scheme'd URL must go through `bare_socket_addr` first (see [`dev_pair_ron`]).

// holger-server-lib/src/lib.rs
//
// Two modes of operation:
//   1. Rust function bindings — use Holger struct directly as a library
//   2. gRPC server — configure exposed_endpoints in RON, call holger.start()

use serde::{Deserialize, Serialize};

use std::{
    fs::File,
    io::BufReader,
    path::Path,
};

use anyhow::Result;
use ron::de::from_reader;

use std::collections::HashMap;
use std::sync::Arc;
pub use traits::RepositoryBackendTrait;
use crate::exposed::ExposedEndpoint;
use crate::exposed::fast_routes::FastRoutes;
pub use crate::repository::Repository;
pub use crate::storage::StorageEndpoint;

pub mod audit;
pub mod auth;
pub mod exposed;
pub mod group;
pub mod grpc;
pub mod metrics;
pub mod object;
pub mod promote;
pub mod properties;
pub mod proxy;
pub mod raw;
pub mod replication;
pub mod repository;
pub mod retention;
pub mod sbom;
pub mod search;
pub mod security;
pub mod storage;
/// Upstream-registry federation (the `full` tier's harbour clients). Gated behind
/// `federation`: the `mini`/`embedded` tiers do NOT compile it and do NOT link
/// Skidbladnir or any registry-client dependency.
#[cfg(feature = "federation")]
pub mod upstream;
// The `rust-remote` pull-through cache upstream — a SYNC `ureq` HTTP client, so an
// outbound-egress path. Gated behind `online` (like reqwest) so the zero-egress
// airgap tiers (`embedded`/`mini`) link NO HTTP client at all.
#[cfg(feature = "online")]
pub mod upstream_rust;

pub use crate::object::{LocalHolger, RemoteHolger};
pub use traits::HolgerObject;
// Re-export the core artifact types + the backend trait so in-process consumers
// (e.g. nornir's embedded-holger viz pane) can enumerate a repository's
// backend directly — `holger.repositories[i].backend_repository` is an
// `Arc<dyn RepositoryBackendTrait>`, so the trait must be reachable to call
// `list`/`fetch` on it without depending on `holger-traits` separately.
pub use traits::{ArtifactEntry, ArtifactId};

/// Lowercase-hex SHA-256 of `data` — the content id the retention executor
/// verifies before a backend unlinks (must match the backend's own recompute).
/// SHA-256 (not blake3) so it agrees with the loose-file backends' native
/// checksum path.
fn sha256_hex(data: &[u8]) -> String {
    // LAW #5 dedup: the shared content-hash primitive lives once in edda's
    // `nornir-hash` leaf; every backend consumes it so all recomputes agree.
    nornir_hash::sha256_hex(data)
}

/// `namespace/name@version` (or `name@version` when unscoped) — the artifact
/// label written into the GC audit record.
fn retention_artifact_label(id: &ArtifactId) -> String {
    match &id.namespace {
        Some(ns) => format!("{ns}/{}@{}", id.name, id.version),
        None => format!("{}@{}", id.name, id.version),
    }
}

// ========================= WIRING ENGINE =========================

pub fn wire_holger(holger: &mut Holger) -> Result<()> {
    let mut repo_map = HashMap::new();
    let mut exposed_map = HashMap::new();
    let mut storage_map = HashMap::new();

    for repo in &*holger.repositories {
        repo_map.insert(repo.ron_name.clone(), repo as *const Repository);
    }
    for exp in &*holger.exposed_endpoints {
        exposed_map.insert(exp.ron_name.clone(), exp as *const ExposedEndpoint);
    }
    for st in &*holger.storage_endpoints {
        storage_map.insert(st.ron_name.clone(), st as *const StorageEndpoint);
    }

    // Wire repository IO references
    for repo in &mut holger.repositories {
        for name in &repo.ron_upstreams {
            if let Some(ptr) = repo_map.get(name) {
                repo.wired_upstreams.push(*ptr);
            } else {
                return Err(anyhow::anyhow!("Missing upstream repo: {}", name));
            }
        }

        // Members of a virtual `group` repo (resolved in declaration order).
        for name in &repo.ron_members {
            if let Some(ptr) = repo_map.get(name) {
                repo.wired_members.push(*ptr);
            } else {
                return Err(anyhow::anyhow!(
                    "Missing member repo '{}' for group '{}'",
                    name,
                    repo.ron_name
                ));
            }
        }

        if let Some(io) = &mut repo.ron_in {
            io.wired_storage = *storage_map
                .get(&io.ron_storage_endpoint)
                .ok_or_else(|| anyhow::anyhow!("Missing storage endpoint: {}", io.ron_storage_endpoint))?;
            io.wired_exposed = *exposed_map
                .get(&io.ron_exposed_endpoint)
                .ok_or_else(|| anyhow::anyhow!("Missing exposed endpoint: {}", io.ron_exposed_endpoint))?;
        }

        if let Some(io) = &mut repo.ron_out {
            io.wired_storage = *storage_map
                .get(&io.ron_storage_endpoint)
                .ok_or_else(|| anyhow::anyhow!("Missing storage endpoint: {}", io.ron_storage_endpoint))?;
            io.wired_exposed = *exposed_map
                .get(&io.ron_exposed_endpoint)
                .ok_or_else(|| anyhow::anyhow!("Missing exposed endpoint: {}", io.ron_exposed_endpoint))?;
        }
    }

    // Wire reverse links (endpoint → repos)
    for exp in &mut holger.exposed_endpoints {
        for repo in &holger.repositories {
            if let Some(io) = &repo.ron_out {
                if io.ron_exposed_endpoint == exp.ron_name {
                    exp.wired_out_repositories.push(repo as *const Repository);
                }
            }
        }
    }

    // The effective serving backend for one repo: its raw backend, or — when it
    // declares upstreams — a `ProxyBackend` wrapping it so 404s fall through. A
    // repo with no materialized backend (e.g. a `group`, or one disabled on init)
    // yields `None`. Shared by the direct route and by group-member resolution, so
    // a group over a caching-proxy member sees the SAME pull-through behaviour.
    fn effective_backend(repo: &Repository) -> Option<Arc<dyn RepositoryBackendTrait>> {
        let backend_arc = repo.backend_repository.as_ref()?;
        if repo.wired_upstreams.is_empty() {
            return Some(backend_arc.clone());
        }
        let upstream_backends: Vec<Arc<dyn RepositoryBackendTrait>> = repo
            .wired_upstreams
            .iter()
            .filter_map(|&ptr| {
                if ptr.is_null() { return None; }
                let upstream: &Repository = unsafe { &*ptr };
                upstream.backend_repository.clone()
            })
            .collect();
        let mut proxy = crate::proxy::ProxyBackend::new(backend_arc.clone(), upstream_backends);
        // Optional negative cache: when the repo declares `ron_negative_cache_secs`,
        // a coordinate/path that misses primary + every upstream is remembered as a
        // miss for that TTL (0 / unset ⇒ disabled, unchanged pull-through).
        if let Some(secs) = repo.ron_negative_cache_secs {
            proxy = proxy.with_negative_cache_secs(secs);
        }
        Some(Arc::new(proxy))
    }

    // Build route tables for each exposed endpoint, wrapping repos that have
    // configured upstreams in a ProxyBackend so 404s fall through automatically.
    for exp in &mut holger.exposed_endpoints {
        let mut routes: Vec<(String, Arc<dyn RepositoryBackendTrait>)> = Vec::new();

        for &repo_ptr in &exp.wired_out_repositories {
            if repo_ptr.is_null() { continue; }
            let repo: &Repository = unsafe { &*repo_ptr };

            // A virtual `group` repo owns no `backend_repository`; it aggregates
            // its members' effective backends behind a single name (first hit wins).
            if repo.ron_repo_type == "group" {
                let members: Vec<Arc<dyn RepositoryBackendTrait>> = repo
                    .wired_members
                    .iter()
                    .filter_map(|&ptr| {
                        if ptr.is_null() { return None; }
                        let member: &Repository = unsafe { &*ptr };
                        effective_backend(member)
                    })
                    .collect();
                routes.push((
                    repo.ron_name.clone(),
                    Arc::new(crate::group::GroupBackend::new(repo.ron_name.clone(), members)),
                ));
                continue;
            }

            let Some(backend) = effective_backend(repo) else { continue };
            routes.push((repo.ron_name.clone(), backend));
        }

        exp.aggregated_routes = if routes.is_empty() {
            None
        } else {
            Some(FastRoutes::new(routes))
        };
    }

    Ok(())
}

// ========================= ROOT HOLGER =========================

#[derive(Serialize, Deserialize)]
pub struct Holger {
    pub repositories: Vec<Repository>,
    pub exposed_endpoints: Vec<ExposedEndpoint>,
    pub storage_endpoints: Vec<StorageEndpoint>,

    /// Optional auth configuration. Absent or empty methods = open access.
    #[serde(default)]
    pub auth: Option<auth::AuthConfig>,

    /// Audit-log configuration. Defaults to ON (append-only Arrow IPC) so a
    /// config that says nothing still gets an audit trail.
    #[serde(default)]
    pub audit: AuditSettings,
}

/// Audit-log configuration. On by default; the default backend is the
/// append-only Arrow IPC log ([`audit::ArrowIpcAuditLog`]).
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct AuditSettings {
    /// Directory for append-only Arrow IPC audit segments. Relative paths
    /// resolve against the server's working directory.
    #[serde(default = "default_audit_dir")]
    pub dir: String,
    /// Master switch. `false` installs a no-op audit log (audit disabled).
    #[serde(default = "default_true")]
    pub enabled: bool,
}

fn default_audit_dir() -> String {
    "holger-audit".to_string()
}

fn default_true() -> bool {
    true
}

impl Default for AuditSettings {
    fn default() -> Self {
        Self {
            dir: default_audit_dir(),
            enabled: true,
        }
    }
}

// ========================= LOAD RON CONFIG =========================

pub fn read_ron_config<P: AsRef<Path>>(path: P) -> Result<Holger> {
    let file = File::open(path)?;
    let reader = BufReader::new(file);
    let holger: Holger = from_reader(reader)?;
    Ok(holger)
}

/// Autogenerate the RON config for the **dev release pair** that nornir embeds:
///
/// * `/cache`    — a writable-znippy primary with a `crates-io` (`rust-remote`)
///   upstream → a transparent caching crates.io mirror (write-through fills it).
/// * `/sparring` — a writable-znippy registry → the release rehearsal target;
///   `seal` freezes it into a static drift snapshot.
///
/// nornir writes this to a file and launches `holger start --config <file>` as a
/// subprocess (no library linking, no version coupling). cargo speaks the HTTP
/// SPARSE protocol, so it talks to `http_addr` (the FastRoutes gateway), e.g.
/// `sparse+http://<http_addr>/cache/index/`; `grpc_url` is the control plane.
/// Archives live under `data_dir`.
/// Reduce a URL-ish string to a bare `host:port` socket address: strip any
/// `scheme://` prefix and any trailing `/path`. `start` parses `ron_url` /
/// `ron_http_url` as bare `SocketAddr`s, so a `--grpc https://host:port` (or
/// `grpc://…`) must be normalized here or the generated config won't parse.
fn bare_socket_addr(url: &str) -> &str {
    let no_scheme = url.split_once("://").map_or(url, |(_, rest)| rest);
    no_scheme.split(['/', '?', '#']).next().unwrap_or(no_scheme)
}

pub fn dev_pair_ron(data_dir: &Path, grpc_url: &str, http_addr: &str) -> String {
    let d = data_dir.display();
    let grpc_url = bare_socket_addr(grpc_url);
    let http_addr = bare_socket_addr(http_addr);
    format!(
        r#"(
    exposed_endpoints: [
        ( ron_name: "dev", ron_url: "{grpc_url}", ron_http_url: Some("{http_addr}") ),
    ],
    storage_endpoints: [
        ( ron_name: "cache-store", ron_storage_type: "znippy", ron_path: "{d}/cache/" ),
        ( ron_name: "sparring-store", ron_storage_type: "znippy", ron_path: "{d}/sparring/" ),
    ],
    repositories: [
        ( ron_name: "crates-io", ron_repo_type: "rust-remote", ron_upstreams: [], ron_in: None, ron_out: None ),
        ( ron_name: "cache", ron_repo_type: "rust", ron_writable_archive: Some("{d}/cache.znippy"), ron_base_url: Some("http://{http_addr}"), ron_upstreams: ["crates-io"], ron_in: None, ron_out: Some(( ron_storage_endpoint: "cache-store", ron_exposed_endpoint: "dev" )) ),
        ( ron_name: "sparring", ron_repo_type: "rust", ron_writable_archive: Some("{d}/sparring.znippy"), ron_base_url: Some("http://{http_addr}"), ron_upstreams: ["crates-io"], ron_in: None, ron_out: Some(( ron_storage_endpoint: "sparring-store", ron_exposed_endpoint: "dev" )) ),
    ],
)
"#
    )
}

impl Holger {
    // ─── Rust API (direct function bindings) ─────────────────────────

    /// Fetch an artifact by ID from a named repository
    pub fn fetch(&self, repository: &str, id: &traits::ArtifactId) -> Result<Option<Vec<u8>>> {
        let repo = self.get_repo(repository)?;
        repo.fetch(id)}

    /// Store an artifact (write-enabled repos only)
    pub fn put(&self, repository: &str, id: &traits::ArtifactId, data: &[u8]) -> Result<()> {
        let repo = self.get_repo(repository)?;
        if !repo.is_writable() {
            anyhow::bail!("Repository '{}' is read-only", repository);
        }
        repo.put(id, data)}

    /// List all configured repository names
    pub fn list_repositories(&self) -> Vec<&str> {
        self.repositories.iter().map(|r| r.ron_name.as_str()).collect()
    }

    /// Cross-repo artifact SEARCH (parity §H, `.nornir/search-design.md`) — the
    /// in-process entry the CLI's `search` verb drives. Collects every configured
    /// repository's live backend and runs the pure [`search::search_repos`]
    /// engine against `query`. Repos with no materialized backend (a virtual
    /// `group`, or one disabled on init) are skipped — their members are searched
    /// directly. Read-only: it never mutates a backend.
    pub fn search_all(&self, query: &search::SearchQuery) -> search::SearchResults {
        let repos = self.backend_pairs();
        let props = self.property_store();
        search::search_repos_with_properties(&repos, query, Some(&*props))
    }

    /// Collect this store's `(repo_name → live backend)` pairs — the raw material
    /// both `search_all` and the replication adapters aggregate over. Repos with
    /// no materialized backend (a virtual `group`, or one disabled on init) are
    /// skipped.
    fn backend_pairs(&self) -> Vec<(String, Arc<dyn RepositoryBackendTrait>)> {
        self.repositories
            .iter()
            .filter_map(|r| {
                r.backend_repository
                    .as_ref()
                    .map(|b| (r.ron_name.clone(), b.clone()))
            })
            .collect()
    }

    /// Replicate this store's artifacts INTO another holger store (parity §I,
    /// `.nornir/replication-design.md`) — the in-process entry the CLI's
    /// `replicate --from --to` verb drives. Reads only from `self` (the source)
    /// and writes only into `dest` (the destination); content-id idempotent, so a
    /// re-run over an unchanged source is a no-op.
    ///
    /// Repos are paired by NAME: a source repo `foo` replicates into the
    /// destination repo `foo`. A source repo with no same-named writable
    /// destination is reported (its artifacts fail closed), never silently
    /// dropped. `only` restricts the run to a single source repo name; `None`
    /// replicates every repo that has a destination match. See
    /// [`replication::replicate_pairs`].
    pub fn replicate_into(
        &self,
        dest: &Holger,
        only: Option<&str>,
        opts: &replication::ReplicationOptions,
    ) -> replication::ReplicationReport {
        let sources = self.backend_pairs();
        let dests = dest.backend_pairs();
        let mut pairs = Vec::new();
        for (sname, sback) in &sources {
            if let Some(want) = only {
                if sname != want {
                    continue;
                }
            }
            match dests.iter().find(|(dname, _)| dname == sname) {
                Some((dname, dback)) => pairs.push(replication::RepoPair {
                    source_name: sname.clone(),
                    source: sback.clone(),
                    dest_name: dname.clone(),
                    dest: dback.clone(),
                }),
                None => log::warn!(
                    "replicate: source repo '{sname}' has no same-named destination repo — skipped"
                ),
            }
        }
        replication::replicate_pairs(&pairs, opts)
    }

    /// The server-side operational data root — the parent of the configured audit
    /// dir (the promote PROD store + jera job ledger already live under it), or
    /// `.` when no meaningful audit dir is set. One derivation so the gRPC
    /// endpoint, the HTTP gateway, and the CLI all agree on where server state is.
    fn data_root(&self) -> std::path::PathBuf {
        std::path::Path::new(&self.audit.dir)
            .parent()
            .filter(|p| !p.as_os_str().is_empty())
            .map(|p| p.to_path_buf())
            .unwrap_or_else(|| std::path::PathBuf::from("."))
    }

    /// The hosted-SBOM store (`<data_root>/holger-sboms`). Derived once from the
    /// config so every surface — the `SbomService` gRPC RPCs, the `GET /-/sbom`
    /// HTTP door, and the `holger-server sbom` CLI verb — reads and writes the
    /// same on-disk store. See [`sbom::SbomStore`], `.nornir/sbom-hosting-design.md`.
    pub fn sbom_store(&self) -> sbom::SharedSbomStore {
        Arc::new(sbom::SbomStore::new(self.data_root().join("holger-sboms")))
    }

    /// The custom-property store (`<data_root>/holger-props`). Derived once from
    /// the config so every surface — the gRPC `Search` property axis, the
    /// `GET`/`PUT /-/properties` HTTP door, and the `holger-server properties` CLI
    /// verb — reads and writes the same on-disk store. See
    /// [`properties::PropertyStore`], `.nornir/properties-design.md`.
    pub fn property_store(&self) -> properties::SharedPropertyStore {
        Arc::new(properties::PropertyStore::new(self.data_root().join("holger-props")))
    }

    /// Set (or, with empty `values`, remove) one custom-property `key` on the
    /// artifact coordinate `(repo, id)` — the write side the HTTP door + CLI drive.
    /// Returns the resulting full property map. Last-wins per key; other keys are
    /// untouched.
    pub fn set_artifact_property(
        &self,
        repo: &str,
        id: &traits::ArtifactId,
        key: &str,
        values: Vec<String>,
    ) -> std::io::Result<properties::PropertyMap> {
        self.property_store().set(repo, id, key, values)
    }

    /// All custom properties attached to `(repo, id)` (empty when none) — the read
    /// side the HTTP door + CLI drive.
    pub fn get_artifact_properties(
        &self,
        repo: &str,
        id: &traits::ArtifactId,
    ) -> properties::PropertyMap {
        self.property_store().get(repo, id)
    }

    /// Plan retention/GC for one repository: gather its listing (`backend.list()`)
    /// as [`retention::ArtifactRecord`]s and run the pure [`retention::plan_retention`]
    /// planner against `policy` at `now` (unix seconds).
    ///
    /// This is a **planner only** — it computes which artifacts a retention policy
    /// would keep vs reclaim; it deletes nothing (holger's znippy archives are
    /// immutable, so carrying a plan out is an archive-set drop delegated to a jera
    /// job). Backends that can't enumerate return an empty listing, so the plan is
    /// empty rather than an error. `created_at` is left `None` (the listing carries
    /// no per-artifact ingest time today), so `keep_last_n` / `max_total_bytes`
    /// apply by version + size while `max_age_secs` no-ops — see
    /// `.nornir/retention-design.md`.
    pub fn plan_repo_retention(
        &self,
        repository: &str,
        policy: &retention::RetentionPolicy,
        now: i64,
    ) -> Result<retention::RetentionPlan> {
        let repo = self.get_repo(repository)?;
        // When the policy pins by a custom property, resolve each listed artifact's
        // properties against the shared store (the ONE I/O the planner itself never
        // does) and flag protected records — a tagged blob is then never reclaimed.
        let prop_store = policy
            .protect_property
            .as_ref()
            .map(|_| self.property_store());
        let records: Vec<retention::ArtifactRecord> = repo
            .list(None, usize::MAX)?
            .into_iter()
            .map(|entry| {
                let record = retention::ArtifactRecord::from_entry(entry);
                match (&policy.protect_property, &prop_store) {
                    (Some((key, value)), Some(store)) => {
                        let pinned = store.matches(repository, &record.id, key, value);
                        record.protected_by_property(pinned)
                    }
                    _ => record,
                }
            })
            .collect();
        Ok(retention::plan_retention(&records, policy, now))
    }

    /// Carry out a [`retention::RetentionPlan`] against one repository — the
    /// **executor** to `plan_repo_retention`'s planner. The planner decided
    /// *what* to reclaim; this acts on that decision and historizes the outcome.
    ///
    /// **Gated + safe:**
    /// * `opts.dry_run` defaults to `true` — a preview reclaims nothing. An actual
    ///   drop only happens when the caller passes [`retention::GcOptions::execute`].
    /// * A record the plan **kept** is never touched (only `decision.delete` rows
    ///   are considered), and — defense in depth against a hand-built or corrupted
    ///   plan — any delete row still carrying a `ProtectedLatest`/`MinPerNameFloor`
    ///   reason is **refused** (skipped), so a protected/latest generation can never
    ///   be dropped through this path.
    /// * Every delete is **content-id verified**: the executor reads the stored
    ///   bytes, computes their SHA-256, and hands it to the backend, which
    ///   re-verifies against what is on disk before it unlinks — a store that
    ///   changed under the plan is refused, never deleted blind.
    ///
    /// **Historized:** each actual reclaim is appended to `audit` as an
    /// [`audit::AuditAction::Delete`] event (identity `holger-retention-gc`, the
    /// reclaimed byte count, the plan's reasons as detail) — the append-only Arrow
    /// warehouse record of what the GC took. When a `jera` job ledger is attached
    /// (`jobs`), the whole run is wrapped in one tracked `retention-gc` job
    /// (`running → done|failed` + a reclaim summary), reusing the same durable job
    /// seam promotion uses.
    pub fn execute_repo_retention(
        &self,
        repository: &str,
        plan: &retention::RetentionPlan,
        opts: &retention::GcOptions,
        audit: &dyn audit::AuditLog,
        jobs: Option<&Arc<jera::jobs::JobStore>>,
    ) -> Result<retention::GcReport> {
        // Count this executor invocation as a retention run for `/metrics`
        // (dry-run or executing — both are a "run" an operator wants to see).
        metrics::global().record_retention_run();
        // Wrap the run in one durable jera job when a ledger is attached (the same
        // seam promotion uses). `None` ⇒ the fast path, no job row written.
        let job = jobs.map(|store| {
            store.start(
                "retention-gc",
                repository,
                repository,
                serde_json::json!({
                    "dry_run": opts.dry_run,
                    "planned_delete": plan.delete_count,
                    "planned_reclaim_bytes": plan.reclaimed_bytes,
                }),
            )
        });

        let out = self.execute_repo_retention_inner(repository, plan, opts, audit);

        if let Some(job) = job {
            match &out {
                Ok(report) => job.finish(
                    serde_json::json!({
                        "dry_run": report.dry_run,
                        "deleted": report.deleted_count,
                        "reclaimed_bytes": report.reclaimed_bytes,
                        "would_delete": report.would_delete_count,
                        "skipped": report.skipped_count,
                        "failed": report.failed_count,
                    }),
                    &format!("{} bytes reclaimed", report.reclaimed_bytes),
                ),
                Err(e) => job.fail_with_detail(serde_json::json!({ "error": e.to_string() })),
            }
        }
        out
    }

    /// The reclaim loop itself — [`execute_repo_retention`](Self::execute_repo_retention)
    /// wraps this in the optional job ledger. Every safety/historization behaviour
    /// lives here.
    fn execute_repo_retention_inner(
        &self,
        repository: &str,
        plan: &retention::RetentionPlan,
        opts: &retention::GcOptions,
        audit: &dyn audit::AuditLog,
    ) -> Result<retention::GcReport> {
        use retention::{GcAction, GcOutcome, RetentionReason};

        let repo = self.get_repo(repository)?;

        let mut actions: Vec<GcAction> = Vec::new();
        let mut deleted_count = 0usize;
        let mut reclaimed_bytes = 0u64;
        let mut would_delete_count = 0usize;
        let mut skipped_count = 0usize;
        let mut failed_count = 0usize;

        for d in &plan.decisions {
            if !d.delete {
                continue; // the plan KEPT this — never in scope for the executor.
            }
            let id = &d.record.id;

            // Defense in depth: refuse to act on anything the planner protected,
            // even if a (hand-built / corrupted) plan wrongly flagged it delete.
            if d.reasons.iter().any(|r| {
                matches!(
                    r,
                    RetentionReason::ProtectedLatest
                        | RetentionReason::MinPerNameFloor
                        | RetentionReason::ProtectedByProperty
                )
            }) {
                skipped_count += 1;
                actions.push(GcAction {
                    id: id.clone(),
                    size_bytes: d.record.size_bytes,
                    content_id: None,
                    outcome: GcOutcome::Skipped(
                        "protected generation carried a delete flag — refused".into(),
                    ),
                    reasons: d.reasons.clone(),
                });
                continue;
            }

            // Read the stored bytes: verifies the artifact still exists AND gives
            // us the content id to check before any unlink.
            let bytes = match repo.fetch(id) {
                Ok(Some(b)) => b,
                Ok(None) => {
                    skipped_count += 1;
                    actions.push(GcAction {
                        id: id.clone(),
                        size_bytes: d.record.size_bytes,
                        content_id: None,
                        outcome: GcOutcome::Skipped("already absent from the store".into()),
                        reasons: d.reasons.clone(),
                    });
                    continue;
                }
                Err(e) => {
                    failed_count += 1;
                    actions.push(GcAction {
                        id: id.clone(),
                        size_bytes: d.record.size_bytes,
                        content_id: None,
                        outcome: GcOutcome::Failed(format!("read failed: {e}")),
                        reasons: d.reasons.clone(),
                    });
                    continue;
                }
            };
            let content_id = sha256_hex(&bytes);

            if opts.dry_run {
                would_delete_count += 1;
                actions.push(GcAction {
                    id: id.clone(),
                    size_bytes: bytes.len() as u64,
                    content_id: Some(content_id),
                    outcome: GcOutcome::WouldDelete,
                    reasons: d.reasons.clone(),
                });
                continue;
            }

            // Actual reclaim: the backend re-verifies the content id against what
            // is on disk and unlinks only on a match.
            match repo.delete_artifact(id, &content_id) {
                Ok(()) => {
                    deleted_count += 1;
                    reclaimed_bytes += bytes.len() as u64;
                    // Historize the reclaim to the warehouse (append-only audit).
                    let detail = format!(
                        "retention-gc reclaim: {}",
                        d.reasons
                            .iter()
                            .map(|r| r.as_str())
                            .collect::<Vec<_>>()
                            .join(",")
                    );
                    let _ = audit.record(
                        audit::AuditEvent::new(
                            "holger-retention-gc",
                            audit::AuditAction::Delete,
                            repository,
                            retention_artifact_label(id),
                            "",
                            200,
                            bytes.len() as u64,
                        )
                        .with_detail(detail),
                    );
                    actions.push(GcAction {
                        id: id.clone(),
                        size_bytes: bytes.len() as u64,
                        content_id: Some(content_id),
                        outcome: GcOutcome::Deleted,
                        reasons: d.reasons.clone(),
                    });
                }
                Err(e) => {
                    failed_count += 1;
                    actions.push(GcAction {
                        id: id.clone(),
                        size_bytes: bytes.len() as u64,
                        content_id: Some(content_id),
                        outcome: GcOutcome::Failed(e.to_string()),
                        reasons: d.reasons.clone(),
                    });
                }
            }
        }

        Ok(retention::GcReport {
            repository: repository.to_string(),
            dry_run: opts.dry_run,
            actions,
            deleted_count,
            reclaimed_bytes,
            would_delete_count,
            skipped_count,
            failed_count,
            evaluated_at: plan.evaluated_at,
        })
    }

    /// Plan **and** execute retention for one repository in a single call — a
    /// convenience over [`plan_repo_retention`](Self::plan_repo_retention) +
    /// [`execute_repo_retention`](Self::execute_repo_retention). Still gated:
    /// `opts.dry_run` decides whether anything is actually reclaimed.
    pub fn run_repo_retention(
        &self,
        repository: &str,
        policy: &retention::RetentionPolicy,
        now: i64,
        opts: &retention::GcOptions,
        audit: &dyn audit::AuditLog,
        jobs: Option<&Arc<jera::jobs::JobStore>>,
    ) -> Result<retention::GcReport> {
        let plan = self.plan_repo_retention(repository, policy, now)?;
        self.execute_repo_retention(repository, &plan, opts, audit, jobs)
    }

    /// Build the shared audit sink from this config (append-only Arrow IPC, or a
    /// no-op when audit is disabled) — the same sink the server installs at boot.
    /// Public so one-shot tools (the `retention` GC executor CLI) can historize
    /// into the very same warehouse the running server writes.
    pub fn audit_log(&self) -> Arc<dyn audit::AuditLog> {
        self.build_audit_log()
    }

    /// Get a repository backend by name
    fn get_repo(&self, name: &str) -> Result<&Arc<dyn RepositoryBackendTrait>> {
        for repo in &self.repositories {
            if repo.ron_name == name {
                return repo.backend_repository.as_ref()
                    .ok_or_else(|| anyhow::anyhow!("Repository '{}' has no backend", name));
            }
        }
        anyhow::bail!("Repository '{}' not found", name)
    }

    // ─── gRPC server (when configured) ──────────────────────────────

    /// Start gRPC servers (and optional HTTP gateways) for all configured
    /// exposed_endpoints
    pub fn start(&self) -> Result<()> {
        crate::exposed::tls::ensure_crypto_provider();
        let auth_config = Arc::new(self.auth.clone().unwrap_or_default());
        // M12/P18: make the mTLS trust model explicit at boot (client_ca is
        // authoritative; the per-method ca_cert is not independently enforced).
        auth_config.warn_unenforced_mtls_ca();
        // One shared, append-only audit sink across every endpoint.
        let audit = self.build_audit_log();
        for ep in &self.exposed_endpoints {
            if let Some(routes) = &ep.aggregated_routes {
                self.start_grpc_endpoint(
                    &ep.ron_url,
                    routes.clone(),
                    auth_config.clone(),
                    audit.clone(),
                    ep.ron_tls.as_ref(),
                    ep.ron_max_body_bytes.unwrap_or(1024 * 1024 * 1024),
                )?;

                if let Some(http_url) = &ep.ron_http_url {
                    let addr: std::net::SocketAddr = http_url.parse()
                        .map_err(|e| anyhow::anyhow!("Invalid HTTP address '{}': {}", http_url, e))?;
                    let tls = match &ep.ron_tls {
                        Some(t) => Some(t.rustls_gateway_config()?),
                        None => None,
                    };
                    let settings = Arc::new(crate::exposed::http::GatewaySettings {
                        routes: routes.clone(),
                        auth_config: auth_config.clone(),
                        tls,
                        // 1 GiB default body cap.
                        max_body_bytes: ep.ron_max_body_bytes.unwrap_or(1024 * 1024 * 1024),
                        audit: audit.clone(),
                        sboms: Some(self.sbom_store()),
                        properties: Some(self.property_store()),
                        // Serve-time quarantine gate — off unless
                        // HOLGER_QUARANTINE_PROPERTY is set (see serve_grpc).
                        quarantine: std::env::var("HOLGER_QUARANTINE_PROPERTY")
                            .ok()
                            .and_then(|raw| crate::properties::parse_predicate(&raw)),
                    });
                    crate::exposed::http::start_http_gateway(addr, settings)?;
                }
            }
        }
        Ok(())
    }

    /// Build the shared audit sink from config. If the Arrow-IPC backend cannot
    /// be opened, log and fall back to a no-op log rather than refusing to start
    /// — availability of the server outweighs a missing audit directory, and the
    /// failure is loud.
    fn build_audit_log(&self) -> Arc<dyn audit::AuditLog> {
        if !self.audit.enabled {
            log::info!("audit log disabled by config");
            return Arc::new(audit::NoopAuditLog);
        }
        match audit::default_audit_log(&self.audit.dir) {
            Ok(log) => {
                log::info!(
                    "audit log: append-only Arrow IPC segments under {:?}",
                    self.audit.dir
                );
                log
            }
            Err(e) => {
                log::error!(
                    "audit DISABLED — could not open audit dir {:?}: {e}",
                    self.audit.dir
                );
                Arc::new(audit::NoopAuditLog)
            }
        }
    }

    fn start_grpc_endpoint(
        &self,
        addr: &str,
        routes: FastRoutes,
        auth_config: Arc<auth::AuthConfig>,
        audit: Arc<dyn audit::AuditLog>,
        tls: Option<&crate::exposed::tls::TlsSettings>,
        max_body_bytes: usize,
    ) -> Result<()> {
        use crate::grpc::holger_proto::{
            repository_service_server::RepositoryServiceServer,
            archive_service_server::ArchiveServiceServer,
            admin_service_server::AdminServiceServer,
            promotion_service_server::PromotionServiceServer,
            search_service_server::SearchServiceServer,
            sbom_service_server::SbomServiceServer,
            property_service_server::PropertyServiceServer,
            replication_service_server::ReplicationServiceServer,
        };
        use crate::grpc::HolgerGrpc;

        // Immutable PROD store for the promote workflow: a sibling of the audit
        // dir (both are server-side operational state under the data root).
        let data_root = {
            let audit_dir = std::path::Path::new(&self.audit.dir);
            audit_dir
                .parent()
                .filter(|p| !p.as_os_str().is_empty())
                .map(|p| p.to_path_buf())
        };
        let prod_root = match &data_root {
            Some(p) => p.join("holger-prod-store"),
            None => std::path::PathBuf::from("holger-prod-store"),
        };
        // Promotion engine, optionally gated by the policy gate (design §4). The
        // gate is OFF unless `HOLGER_PROMOTE_ENFORCE` is configured; a configured
        // but unreadable/unparseable advisory feed fails closed (serve refuses to
        // start) rather than waving promotions through.
        let mut engine = crate::promote::PromotionEngine::new(prod_root);
        match crate::promote::PolicyGate::from_env()? {
            Some(gate) => {
                log::info!("promotion policy gate ENABLED (HOLGER_PROMOTE_ENFORCE)");
                engine = engine.with_gate(gate);
            }
            None => log::info!(
                "promotion policy gate off (set HOLGER_PROMOTE_ENFORCE=block|warn to enable)"
            ),
        }
        // Property-requirement gate (property-policy §): only promote artifacts
        // whose SOURCE coordinate carries a required custom property. OFF unless
        // `HOLGER_PROMOTE_REQUIRE_PROPERTY=key=value` (or bare `key`) is set —
        // mirrors the `HOLGER_PROMOTE_ENFORCE` env surface.
        if let Ok(raw) = std::env::var("HOLGER_PROMOTE_REQUIRE_PROPERTY") {
            if let Some((key, value)) = crate::promote::PromotionEngine::parse_property_requirement(&raw) {
                log::info!("promotion property requirement ENABLED ('{key}={value}')");
                engine = engine.with_required_property(key, value);
            }
        }
        // JOB LEDGER (jera): route every promotion through the shared `jera::jobs`
        // ledger so long-running server operations land in one redb list a reader
        // can enumerate. jera keeps the list bounded (recent jobs); the permanent
        // record is the audit log. Best-effort + additive: if the ledger can't
        // be opened (e.g. a read-only data dir) the server logs and serves exactly
        // as before, promotions just aren't recorded. The ledger file lives beside
        // the prod store under the data root (`holger-jobs/jobs.redb`).
        let jobs_root = match &data_root {
            Some(p) => p.join("holger-jobs"),
            None => std::path::PathBuf::from("holger-jobs"),
        };
        match std::fs::create_dir_all(&jobs_root)
            .map_err(anyhow::Error::from)
            .and_then(|()| jera::jobs::JobStore::open(&jobs_root))
        {
            Ok(store) => {
                log::info!("promotion job ledger at {}/jobs.redb", jobs_root.display());
                engine = engine.with_job_ledger(Arc::new(store));
            }
            Err(e) => log::warn!(
                "promotion job ledger unavailable ({e}); promotions serve un-recorded"
            ),
        }
        let promotion = Arc::new(engine);

        // Hosted-SBOM store (parity §6/§11) under the same data root as the
        // promote store / job ledger; shared with the HTTP `/-/sbom` door and the
        // CLI so all three surfaces host the same documents.
        let sboms = self.sbom_store();

        // Custom-property store (parity §8) under the same data root; shared with
        // the HTTP `/-/properties` door + the CLI so the `property` search axis and
        // the write path all see one on-disk store.
        let props = self.property_store();

        // Serve-time quarantine gate (property-policy §): a blob whose properties
        // match `HOLGER_QUARANTINE_PROPERTY` (`key=value` / bare `key`) is refused at
        // the fetch boundary — a bad/unverified artifact is blocked without being
        // deleted. OFF unless the env var is set (zero hot-path cost by default).
        let quarantine = std::env::var("HOLGER_QUARANTINE_PROPERTY")
            .ok()
            .and_then(|raw| crate::properties::parse_predicate(&raw));
        if let Some((k, v)) = &quarantine {
            log::info!("serve-time quarantine gate ENABLED ('{k}={v}')");
        }

        let mut grpc_state_builder = HolgerGrpc::with_auth(routes, auth_config)
            .with_audit(audit)
            .with_max_body_bytes(max_body_bytes)
            .with_promotion(promotion)
            .with_sboms(sboms)
            .with_properties(props);
        if let Some((key, value)) = quarantine {
            grpc_state_builder = grpc_state_builder.with_quarantine(key, value);
        }
        let grpc_state = Arc::new(grpc_state_builder);
        let listen_addr: std::net::SocketAddr = addr.parse()
            .map_err(|e| anyhow::anyhow!("Invalid gRPC address '{}': {}", addr, e))?;

        // The TLS terminates on the tonic builder itself; build the optional
        // `ServerTlsConfig` here (holger's own PKI), then hand it — with the
        // per-service message-size-tuned routes — to the SHARED serve loop
        // `nornir_service::serve_with_tls` (edda). This is the ONE canonical
        // `Server::builder()…serve(addr)` bootstrap tail nornir/korp/dwarves also
        // consume (Law #1); holger's richer auth (OIDC/JWT + mTLS-CN + RBAC) lives
        // in `HolgerGrpc`, not here, and is untouched.
        let tls_was_set = tls.is_some();
        let tls_config = match tls {
            Some(t) => Some(
                t.tonic_config()
                    .map_err(|e| anyhow::anyhow!("gRPC TLS config: {}", e))?,
            ),
            None => {
                log::warn!(
                    "SECURITY: gRPC server on {} runs WITHOUT TLS (cleartext h2c). \
                     Set ron_tls for any non-loopback use.",
                    listen_addr
                );
                None
            }
        };

        // Per-service message-size caps stay applied to each service BEFORE it is
        // collected into tonic's dynamic `Routes` — nothing app-specific is lost
        // by sharing the serve tail.
        let mut routes = tonic::service::Routes::builder();
        routes.add_service(
            RepositoryServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            ArchiveServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            AdminServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            PromotionServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            SearchServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            SbomServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            PropertyServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );
        routes.add_service(
            ReplicationServiceServer::new(grpc_state.clone())
                .max_decoding_message_size(max_body_bytes)
                .max_encoding_message_size(max_body_bytes),
        );

        tokio::spawn(async move {
            let scheme = if tls_was_set { "https/h2" } else { "h2c (cleartext)" };
            println!("gRPC server listening on {} [{}]", listen_addr, scheme);
            if let Err(e) =
                nornir_service::serve_with_tls(routes.routes(), listen_addr, tls_config).await
            {
                eprintln!("gRPC server error: {}", e);
            }
        });

        Ok(())
    }

    pub fn instantiate_backends(&mut self) -> Result<()> {
        for se in &mut self.storage_endpoints {
            se.backend_from_config()?;
        }

        // A single repo failing to instantiate (e.g. a corrupt/incompatible
        // znippy archive) must NOT take the whole server down — disable that one
        // repo and serve the rest. `wire_holger` already skips repos whose
        // `backend_repository` stayed `None`. We only hard-fail if *every* repo
        // failed (nothing left to serve).
        let total = self.repositories.len();
        let mut failed = 0usize;
        for repo in &mut self.repositories {
            if let Err(e) = repo.backend_from_config() {
                failed += 1;
                log::error!(
                    "repository '{}' DISABLED — backend init failed: {:#}",
                    repo.ron_name, e
                );
            }
        }
        if total > 0 && failed == total {
            anyhow::bail!(
                "all {} repositories failed to initialize — nothing to serve (see errors above)",
                total
            );
        }
        if failed > 0 {
            log::warn!(
                "{}/{} repositories disabled (backend init failed); serving the remaining {}",
                failed, total, total - failed
            );
        }
        Ok(())
    }
}

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

    /// The autogenerated dev pair must (a) parse, (b) instantiate every backend,
    /// (c) wire (upstreams + endpoints all resolve), and (d) behave: `/sparring`
    /// is writable + serves what it's given, `/crates-io` is read-only.
    #[test]
    fn dev_pair_ron_parses_instantiates_and_wires() {
        let dir = tempfile::tempdir().unwrap();
        let ron = dev_pair_ron(dir.path(), "https://127.0.0.1:18443", "127.0.0.1:18444");

        // A scheme'd `--grpc` must still yield a bare `host:port` `ron_url`,
        // since `start` parses it as a raw SocketAddr (regression: dev-pair
        // crash-looped on `Invalid gRPC address 'https://…'`).
        assert!(
            ron.contains(r#"ron_url: "127.0.0.1:18443""#),
            "ron_url must be a bare host:port with no scheme, got:\n{ron}"
        );

        let mut holger: Holger =
            from_reader(std::io::Cursor::new(ron.as_bytes())).expect("dev-pair RON must parse");
        assert_eq!(holger.list_repositories(), vec!["crates-io", "cache", "sparring"]);

        holger.instantiate_backends().expect("every backend must build");
        wire_holger(&mut holger).expect("upstreams + endpoints must all wire");

        // /sparring is writable and round-trips a crate.
        let id = traits::ArtifactId { namespace: None, name: "demo".into(), version: "0.1.0".into() };
        holger.put("sparring", &id, b"crate-bytes").expect("sparring is writable");
        assert_eq!(
            holger.fetch("sparring", &id).unwrap().as_deref(),
            Some(b"crate-bytes".as_ref()),
        );

        // /crates-io upstream is read-only (a put must be rejected).
        assert!(holger.put("crates-io", &id, b"x").is_err(), "crates-io upstream is read-only");
    }

    /// A virtual `group` repo wires end-to-end: its members resolve, the route
    /// table exposes a `GroupBackend` under the group name, and a fetch through
    /// the group resolves across members (first hit wins), while the group itself
    /// is read-only.
    #[test]
    fn group_repo_wires_and_resolves_across_members() {
        let dir = tempfile::tempdir().unwrap();
        let d = dir.path().display();
        let ron = format!(
            r#"(
    exposed_endpoints: [ ( ron_name: "dev", ron_url: "127.0.0.1:0", ron_http_url: None ) ],
    storage_endpoints: [ ( ron_name: "inert", ron_storage_type: "znippy", ron_path: "{d}/inert/" ) ],
    repositories: [
        ( ron_name: "r1", ron_repo_type: "rust", ron_data_dir: Some("{d}/r1"), ron_base_url: Some("http://x"), ron_upstreams: [], ron_in: None, ron_out: None ),
        ( ron_name: "r2", ron_repo_type: "rust", ron_data_dir: Some("{d}/r2"), ron_base_url: Some("http://x"), ron_upstreams: [], ron_in: None, ron_out: None ),
        ( ron_name: "all", ron_repo_type: "group", ron_members: ["r1", "r2"], ron_upstreams: [], ron_in: None, ron_out: Some(( ron_storage_endpoint: "inert", ron_exposed_endpoint: "dev" )) ),
    ],
)
"#
        );

        let mut holger: Holger =
            from_reader(std::io::Cursor::new(ron.as_bytes())).expect("group RON must parse");
        holger.instantiate_backends().expect("member + group backends build");
        wire_holger(&mut holger).expect("group members + endpoint must wire");

        // Seed r2 only (via its own backend) — the group must still resolve it.
        let id = traits::ArtifactId { namespace: None, name: "widget".into(), version: "1.0.0".into() };
        holger.put("r2", &id, b"from-r2").expect("r2 is writable");

        // The exposed endpoint's route table carries a "all" route (the group).
        let exp = &holger.exposed_endpoints[0];
        let routes = exp.aggregated_routes.as_ref().expect("routes built");
        let group = routes.lookup("all").expect("group route present");
        assert!(!group.is_writable(), "group is a read-only view");
        assert_eq!(
            group.fetch(&id).unwrap().as_deref(),
            Some(b"from-r2".as_ref()),
            "group resolves an artifact held only by member r2",
        );
        // An artifact in neither member → None across the group.
        let absent = traits::ArtifactId { namespace: None, name: "ghost".into(), version: "9.9.9".into() };
        assert_eq!(group.fetch(&absent).unwrap(), None);
    }

    /// A single writable file-backed `rust` repo named `mirror`, rooted under
    /// `dir` — the on-disk store both replication endpoints use in the test below.
    fn single_rust_store(dir: &std::path::Path) -> Holger {
        let d = dir.display();
        let ron = format!(
            r#"(
    exposed_endpoints: [ ( ron_name: "dev", ron_url: "127.0.0.1:0", ron_http_url: None ) ],
    storage_endpoints: [],
    repositories: [
        ( ron_name: "mirror", ron_repo_type: "rust", ron_data_dir: Some("{d}/mirror"), ron_base_url: Some("http://x"), ron_upstreams: [], ron_in: None, ron_out: None ),
    ],
)
"#
        );
        let mut holger: Holger =
            from_reader(std::io::Cursor::new(ron.as_bytes())).expect("store RON must parse");
        holger.instantiate_backends().expect("backend builds");
        wire_holger(&mut holger).expect("wires");
        holger
    }

    /// End-to-end replication between TWO SEPARATE file-backed holger stores (the
    /// laptop-testable form the CLI drives): seed a source store, replicate into a
    /// fresh destination store, and assert the destination holds every source
    /// artifact byte-identical — then that a re-run is a pure no-op.
    #[test]
    fn replicate_between_two_file_backed_stores_is_complete_and_idempotent() {
        use crate::replication::ReplicationOptions;
        let src_dir = tempfile::tempdir().unwrap();
        let dst_dir = tempfile::tempdir().unwrap();
        let source = single_rust_store(src_dir.path());
        let dest = single_rust_store(dst_dir.path());

        // Seed the SOURCE's `mirror` repo with three crates.
        let ids = [
            traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() },
            traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.1".into() },
            traits::ArtifactId { namespace: None, name: "tokio".into(), version: "1.40.0".into() },
        ];
        let bodies: [&[u8]; 3] = [b"serde-a", b"serde-b", b"tokio-c"];
        for (id, body) in ids.iter().zip(bodies.iter()) {
            source.put("mirror", id, body).expect("source is writable");
        }

        // Dry run writes nothing.
        let preview = source.replicate_into(&dest, None, &ReplicationOptions::preview());
        assert_eq!(preview.would_copy(), 3, "all three would copy");
        assert_eq!(preview.copied(), 0);
        for id in &ids {
            assert!(dest.fetch("mirror", id).unwrap().is_none(), "dry run wrote nothing");
        }

        // Execute: copy source → dest.
        let run = source.replicate_into(&dest, None, &ReplicationOptions::executing());
        assert_eq!(run.copied(), 3, "all three copied");
        assert!(run.is_complete(), "no failures");

        // CORE RED-when-broken assertion: destination holds every source artifact,
        // byte-identical. A dropped or garbled copy fails right here.
        for (id, body) in ids.iter().zip(bodies.iter()) {
            let got = dest
                .fetch("mirror", id)
                .unwrap()
                .unwrap_or_else(|| panic!("destination MISSING {id:?} after replicate"));
            assert_eq!(&got, body, "destination bytes must equal source bytes for {id:?}");
        }

        // Re-run over the unchanged source is a no-op (idempotent by content id).
        let again = source.replicate_into(&dest, None, &ReplicationOptions::executing());
        assert_eq!(again.copied(), 0, "re-run copies nothing");
        assert_eq!(again.skipped(), 3, "every artifact already present by content id");
    }
}