gunnar-server 2.0.0

//! The administrative verbs: namespaces, repositories, principals, keys,
//! grants, and `gc`.
//!
//! These run against the same [`RepoCatalog`] and [`AuthTable`] the gRPC
//! control plane serves, so a repository created here is immediately visible
//! there and vice versa. There is no second model and no separate file format
//! (one implementation).

use anyhow::{bail, Context, Result};
use gunnar_core::observe::TARGET_STORE;
use gunnar_core::{Fingerprint, Grant, NewRepo, Repo, RepoRef, Scope, Visibility};

use crate::storage::MaintenanceReport;
use crate::stores::ServerStores;
use gunnar_common::cli::{
    GcArgs, GrantArgs, KeyCommand, NamespaceCommand, PrincipalCommand, ReflogArgs, RepoCommand,
    RevokeArgs,
};

/// `<namespace>/<name>`, through the same sandbox the SSH listener uses.
///
/// One resolver, not two: a path the CLI accepts and the SSH path sandbox
/// rejects would be a repository nobody could clone (one implementation).
fn repo_ref(path: &str) -> Result<RepoRef> {
    gunnar_core::resolve_repo_path(path)
        .with_context(|| format!("`{path}` is not a valid <namespace>/<name>"))
}

/// A certificate's DER, read from a file holding either PEM or DER.
///
/// The unwrap itself lives in `gunnar-core` beside the fingerprint, because the
/// two must agree about what the certificate bytes are.
fn read_certificate(path: &std::path::Path) -> Result<Vec<u8>> {
    let raw = std::fs::read(path)
        .with_context(|| format!("reading the certificate at {}", path.display()))?;
    Ok(gunnar_core::auth::certificate_der(&raw)?)
}

/// An OpenSSH public key line, or `@<path>`.
fn key_argument(spec: &str) -> Result<String> {
    match spec.strip_prefix('@') {
        Some(path) => std::fs::read_to_string(path)
            .with_context(|| format!("reading the public key from {path}"))
            .map(|text| text.trim().to_owned()),
        None => Ok(spec.trim().to_owned()),
    }
}

// ── the certificate issuer ─────────────────────────────────────────────────

/// **The PKI document, as one value** — the issuer and the trusted-CA list.
///
/// The ONE reader (one implementation). `gunnar issuer show` prints it, and
/// `CoreControlPlane::issuer_info` puts it on the wire for `Admin.IssuerInfo`;
/// neither assembles its own, so a box cannot say one thing on its own disk
/// and another over the control plane.
///
/// Takes an already-opened [`gunnar_core::Pki`] rather than the stores,
/// because the CAPABILITY is the caller's decision: both callers hand it
/// `ServerStores::pki_readonly()`, whose secret store refuses every call, and
/// this function has no way to reach a private key whatever it is given — the
/// two types it reads out of, `Issuer` and `TrustedCa`, have no field that
/// could hold one.
pub fn issuer_document(pki: &gunnar_core::Pki) -> Result<gunnar_proto::IssuerInfoResponse> {
    let trusted = pki
        .trusted_cas()
        .context("reading the trusted-CA list")?
        .into_iter()
        .map(|ca| gunnar_proto::TrustedCaInfo {
            label: ca.label,
            fingerprint: ca.fingerprint.to_string(),
            cert_pem: ca.cert_pem,
            added_unix_ms: ca.added,
            is_issuer: ca.is_issuer,
        })
        .collect();
    Ok(match pki.issuer().context("reading the issuer")? {
        Some(live) => gunnar_proto::IssuerInfoResponse {
            has_issuer: true,
            subject: live.subject,
            fingerprint: live.fingerprint.to_string(),
            cert_pem: live.cert_pem,
            created_unix_ms: live.created,
            origin: live.origin.to_string(),
            trusted,
        },
        // Every other field empty, and `has_issuer` is what says so: a data
        // root that has never been served has never minted one, and that is a
        // different fact from "the document could not be read" — which is an
        // error here, never an empty answer.
        None => gunnar_proto::IssuerInfoResponse {
            has_issuer: false,
            trusted,
            ..Default::default()
        },
    })
}

/// Run a `gunnar issuer …` verb.
pub fn issuer(stores: &ServerStores, command: gunnar_common::cli::IssuerCommand) -> Result<()> {
    use gunnar_common::cli::IssuerCommand;

    match command {
        IssuerCommand::Create { subject, days } => {
            // Refused, loudly, on a server that already has one. That refusal
            // lives in `gunnar_core::issuer` — in the document's own
            // compare-and-swap, not only in a check this command could skip.
            let created = stores.pki()?.create(&subject, days)?;
            println!(
                "issuer created: {} ({})",
                created.subject, created.fingerprint
            );
            eprintln!(
                "its private key is sealed in the secret store and is not in the control \
                 database. There is no way to print it, and that is deliberate: a key that can \
                 mint any user is the one whose loss means an attacker issues themselves an \
                 administrator"
            );
        }
        IssuerCommand::Import { cert, key } => {
            let cert_pem = std::fs::read_to_string(&cert)
                .with_context(|| format!("reading the CA certificate at {}", cert.display()))?;
            let key_pem = std::fs::read_to_string(&key)
                .with_context(|| format!("reading the CA private key at {}", key.display()))?;
            let imported = stores.pki()?.import(&cert_pem, &key_pem)?;
            println!(
                "issuer imported: {} ({})",
                imported.subject, imported.fingerprint
            );
        }
        IssuerCommand::Show { pem, at, identity: _ } => {
            // `--at` is the SAME verb asked over the wire, and it is dispatched
            // in `gunnar-cli` before this arm is ever reached. Getting here
            // with one means this binary was built without the control client,
            // and the honest answer is to say so — not to silently read the
            // local data root and let the operator believe they were told
            // about the box they named.
            if let Some(at) = at {
                bail!(
                    "`--at {at}` asks a running server, and this binary carries no control                      client (built without the `control` feature). Without it, `gunnar issuer                      show` can only read the data root it is standing in"
                );
            }
            // The READ-ONLY handle: `show` is a control-plane read and must
            // work on a binary with no storage engine, and it must be unable to
            // reach the private key even if this arm were later edited.
            //
            // Through the SAME reader and the SAME renderer `Admin.IssuerInfo`
            // uses (one implementation), so the lines printed here and the
            // lines `--at` prints cannot disagree about one document.
            print!("{}", issuer_document(&stores.pki_readonly()?)?.render(pem));
        }
        IssuerCommand::Trust { label, cert } => {
            let cert_pem = std::fs::read_to_string(&cert)
                .with_context(|| format!("reading the CA certificate at {}", cert.display()))?;
            let label = if label.is_empty() {
                cert.file_name()
                    .map(|name| name.to_string_lossy().into_owned())
                    .unwrap_or_else(|| "ca".to_owned())
            } else {
                label
            };
            let added = stores.pki()?.trust_ca(&label, &cert_pem)?;
            println!("trusting {} ({})", added.label, added.fingerprint);
        }
        // Printed through `Display` rather than assembled here: `serve.rs`
        // refuses with the same words out of the same type, and a second
        // renderer is a second story about one server.
        IssuerCommand::Readiness => {
            print!("{}", crate::tls_preflight::preflight(stores)?);
            println!();
        }
        IssuerCommand::Untrust { fingerprint } => {
            let parsed = Fingerprint::parse(&fingerprint)?;
            match stores.pki()?.untrust_ca(&parsed)? {
                Some(removed) => println!("no longer trusting {} ({parsed})", removed.label),
                None => bail!("no trusted CA has fingerprint {parsed}"),
            }
        }
    }
    Ok(())
}

/// Write an issued certificate — and, on the weaker path, its key — and
/// register it to the principal it names.
///
/// The file names match what `cert-helper` writes and therefore what
/// `bench/src/trust.rs` produces, so an operator holding both kinds of material
/// does not have to know which tool made which.
///
/// The private key is written at `0600` on unix, and the mode is set at
/// **creation** rather than after: a key that is world-readable for the
/// microseconds between `write` and `set_permissions` is a key that was
/// world-readable.
fn hand_over(
    stores: &ServerStores,
    dir: &std::path::Path,
    issued: &gunnar_core::Issued,
) -> Result<()> {
    std::fs::create_dir_all(dir)
        .with_context(|| format!("creating the output directory {}", dir.display()))?;

    let cert_path = dir.join(format!("{}_cert.pem", issued.principal_id));
    std::fs::write(&cert_path, &issued.cert_pem)
        .with_context(|| format!("writing {}", cert_path.display()))?;

    // Registered as well as issued. The certificate-identity plan keeps the
    // stronger form available at no cost: the CN gives the identity and the
    // fingerprint gives the registration, and a deployment that wants both to
    // agree needs the fingerprint to be in the table.
    let registered =
        stores
            .auth()
            .register_certificate(&issued.principal_id, &issued.cert_der, None)?;

    println!(
        "issued {} -> {}",
        registered.fingerprint_sha256,
        cert_path.display()
    );

    if let Some(key_pem) = &issued.key_pem {
        let key_path = dir.join(format!("{}_pkey.pem", issued.principal_id));
        write_private(&key_path, key_pem)?;
        println!("private key -> {}", key_path.display());
        eprintln!(
            "this key was generated HERE and is now on the wire and on this disk. \
             `--csr` is the form where it never leaves {}'s machine",
            issued.principal_id
        );
    }
    Ok(())
}

/// Write a private key so that it is never, at any instant, readable by anyone
/// but its owner.
fn write_private(path: &std::path::Path, contents: &str) -> Result<()> {
    use std::io::Write as _;

    let mut options = std::fs::OpenOptions::new();
    options.write(true).create(true).truncate(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        options.mode(0o600);
    }
    // `mode()` applies only when the file is CREATED, so an existing file would
    // keep whatever permissions it had. Remove it first.
    if path.exists() {
        std::fs::remove_file(path)
            .with_context(|| format!("removing the previous {}", path.display()))?;
    }
    let mut file = options
        .open(path)
        .with_context(|| format!("creating {}", path.display()))?;
    file.write_all(contents.as_bytes())
        .with_context(|| format!("writing {}", path.display()))?;
    Ok(())
}

// ── namespaces ──────────────────────────────────────────────────────────────

/// Run a `gunnar namespace …` verb.
pub fn namespace(stores: &ServerStores, command: NamespaceCommand) -> Result<()> {
    match command {
        NamespaceCommand::Create { name } => {
            let created = stores.catalog().create_namespace(&name)?;
            println!("created namespace {}", created.name);
        }
        NamespaceCommand::List => {
            let mut names: Vec<String> = stores
                .catalog()
                .namespaces()?
                .into_iter()
                .map(|ns| ns.name)
                .collect();
            names.sort();
            for name in names {
                println!("{name}");
            }
        }
        NamespaceCommand::Delete { name, delete_repos } => {
            let gone = stores.catalog().delete_namespace(&name, delete_repos)?;
            println!("deleted namespace {name} and {} repositories", gone.len());
        }
        NamespaceCommand::SetBudget {
            name,
            explode_budget,
        } => {
            let Some(spec) = explode_budget else {
                anyhow::bail!(
                    "`namespace set-budget` needs a budget to set; today that is \
                     `--explode-budget <8G|512M|unlimited|default>`"
                );
            };
            let bytes = match spec.trim().to_ascii_lowercase().as_str() {
                "default" | "" => None,
                #[cfg(feature = "engine-znippy")]
                other => Some(
                    crate::storage::znippy_engine::parse_explode_budget(other)
                        .context("reading the explode budget")?,
                ),
                #[cfg(not(feature = "engine-znippy"))]
                other => anyhow::bail!("an explode budget ({other}) is the znippy engine's; this server was built without it"),
            };
            let ns = stores
                .catalog()
                .set_namespace_explode_budget(&name, bytes)?;
            // Applied to the stores that are open right now, not only to the
            // ones opened later — an admin who raises a quota expects the
            // repository being read this second to get it.
            for repo in stores.catalog().list(Some(&name))? {
                if let Ok(dir) = stores.repo_dir(&repo.path()) {
                    stores.apply_explode_quota(&name, &dir);
                }
            }
            match ns.explode_budget_bytes {
                Some(0) => println!("{name}: explode budget unlimited"),
                #[cfg(feature = "engine-znippy")]
                Some(b) => println!(
                    "{name}: explode budget {}",
                    crate::storage::znippy_engine::human_bytes(b)
                ),
                #[cfg(not(feature = "engine-znippy"))]
                Some(b) => println!("{name}: explode budget {b} bytes"),
                None => println!("{name}: explode budget back to the server default"),
            }
        }
    }
    Ok(())
}

// ── repositories ────────────────────────────────────────────────────────────

/// Run a `gunnar repo …` verb.
pub fn repo(stores: &ServerStores, command: RepoCommand) -> Result<()> {
    match command {
        RepoCommand::Create {
            path,
            hash,
            default_branch,
            public,
            description,
            adopt,
        } => {
            let path = repo_ref(&path)?;
            let mut spec = NewRepo::new(path.clone())
                .hash_kind(hash.into())
                .visibility(if public {
                    Visibility::Public
                } else {
                    Visibility::Private
                });
            spec.description = description;
            if let Some(branch) = default_branch {
                spec = spec.default_branch(full_branch(&branch));
            }
            // The catalog row and the on-disk repository, together and
            // transactionally, refusing a store already at the name unless
            // `--adopt`. A catalog entry with nothing behind it is listed by
            // `repo list` and refused by `git clone`; a stranger's directory
            // adopted by a create is a deleted repository resurrected.
            let created = stores.create_repo(spec, adopt)?;
            let store = stores.repo_store(&created)?;
            println!(
                "created {} ({}, default branch {}) at {}",
                created.key(),
                created.hash_kind,
                created.default_branch,
                store.git_dir().display()
            );
        }
        RepoCommand::List { namespace } => {
            let mut repos = stores.catalog().list(namespace.as_deref())?;
            repos.sort_by_key(|repo| repo.key());
            for repo in repos {
                println!(
                    "{}\t{}\t{}\t{:?}",
                    repo.key(),
                    repo.hash_kind,
                    repo.default_branch,
                    repo.visibility
                );
            }
        }
        RepoCommand::Delete {
            path,
            purge,
            retention_days,
        } => {
            let path = repo_ref(&path)?;
            // **An ORPHAN is a delete to finish, not a not-found** (2026-08-30):
            // a store on disk with no row and no tombstone is what an old
            // catalog-only delete left behind. `--purge` finishes it; a soft
            // delete of one is refused, because a tombstone that records no
            // row promises an undelete the server cannot honour. Same rule as
            // the control plane's `delete_repo`, from the same two helpers.
            let orphan = stores.is_orphan(&path)?;
            anyhow::ensure!(
                purge || !orphan,
                "{}",
                crate::stores::ServerStores::orphan_soft_delete_refusal(&path)
            );
            // Read the row BEFORE removing it: the tombstone records what the
            // repository was, and an undelete that guessed the hash kind would
            // reopen the resurrection the reclaim rules closed from the other side.
            let removed = match stores.catalog().delete(&path) {
                Ok(repo) => Some(repo),
                Err(gunnar_core::Error::NoSuchRepo(_)) if orphan => None,
                Err(err) => return Err(err.into()),
            };
            // Release the handle before anything else, so the store lock is
            // not left holding a file inside a directory being removed.
            stores.close_repo(&path);
            if purge {
                let dir = stores.repo_dir(&path)?;
                let (before, after) = stores.purge_repo(&path)?;
                println!(
                    "deleted {}{} and purged {} ({} bytes reclaimed)",
                    path,
                    if removed.is_none() { " (an ORPHAN: it had no catalog row)" } else { "" },
                    dir.display(),
                    before.saturating_sub(after)
                );
            } else {
                let removed = removed.expect("a soft delete of an orphan is refused above");
                let days = retention_days.unwrap_or(crate::stores::DEFAULT_DELETE_RETENTION_DAYS);
                let record = stores.soft_delete(
                    &path,
                    Some(&removed),
                    "",
                    std::time::Duration::from_secs(u64::from(days) * 86_400),
                )?;
                let bytes = crate::stores::dir_bytes(&stores.repo_dir(&path)?);
                println!(
                    "deleted {} — its {bytes} bytes are KEPT for {days} day(s)",
                    removed.key()
                );
                println!(
                    "  undelete: gunnar repo undelete {}    purge now: gunnar repo purge {}",
                    removed.key(),
                    removed.key()
                );
                if let Some(at) = record.purge_after_unix_ms {
                    println!("  the sweep may reclaim them after unix_ms {at}");
                }
                println!("  until then they still count against the account's storage.");
            }
        }
        RepoCommand::Undelete { path } => {
            let path = repo_ref(&path)?;
            let spec = stores.undelete_spec(&path)?;
            // `adopt`: the store on disk is the one this repository always
            // had, and the tombstone carries the hash kind the adopt path
            // checks the store's own marker against.
            let repo = stores.create_repo(spec, true)?;
            stores.clear_tombstone(&path)?;
            println!(
                "undeleted {} ({}, default branch {})",
                repo.key(),
                repo.hash_kind,
                repo.default_branch
            );
        }
        RepoCommand::Purge { path } => {
            let path = repo_ref(&path)?;
            anyhow::ensure!(
                stores.catalog().get(&path)?.is_none(),
                "{path} is a LIVE repository, not a deleted one. `repo delete {path} --purge` \
                 deletes and reclaims it in one step; purge alone only ends a retention window."
            );
            anyhow::ensure!(
                stores.tombstone_of(&path)?.is_some(),
                "{path} is neither catalogued nor tombstoned; there is nothing to purge"
            );
            let (before, after) = stores.purge_repo(&path)?;
            println!(
                "purged {path} ({} bytes reclaimed); it can no longer be undeleted",
                before.saturating_sub(after)
            );
        }
        RepoCommand::ListDeleted {} => {
            let rows = stores.tombstones()?;
            if rows.is_empty() {
                println!("nothing is deleted");
            }
            let now = gunnar_core::now_ms();
            let mut total = 0u64;
            for row in &rows {
                total += row.bytes;
                let left = match row.tombstone.seconds_left(now) {
                    Some(secs) => format!("{} day(s) left", secs / 86_400),
                    None => "no expiry".to_owned(),
                };
                println!("{}\t{} bytes\t{left}", row.path, row.bytes);
            }
            if !rows.is_empty() {
                println!(
                    "{} deleted repository(ies) holding {total} bytes — still on the account",
                    rows.len()
                );
            }
        }
        RepoCommand::ListOrphans { namespace } => {
            let rows = stores.orphans()?;
            let rows: Vec<_> = rows
                .into_iter()
                .filter(|row| {
                    namespace
                        .as_deref()
                        .is_none_or(|want| row.path.namespace() == want)
                })
                .collect();
            if rows.is_empty() {
                println!("no orphan stores: every directory under the repository root is either \
                          catalogued (`repo list`) or soft-deleted (`repo list-deleted`)");
            }
            let mut total = 0u64;
            for row in &rows {
                total += row.bytes;
                println!(
                    "{}\t{} bytes\t{}\tmtime {}\t{}{}",
                    row.path,
                    row.bytes,
                    row.hash_marker.as_deref().unwrap_or("-"),
                    row.modified_unix_ms,
                    row.dir.display(),
                    match &row.failed_reclaim {
                        Some(note) => format!("\tRECLAIM FAILED: {note}"),
                        None => String::new(),
                    }
                );
            }
            if !rows.is_empty() {
                println!(
                    "{} orphan store(s) holding {total} bytes — no catalog row, no tombstone",
                    rows.len()
                );
                println!(
                    "  reclaim the bytes: gunnar repo reclaim <path>    keep them: gunnar repo \
                     create <path> --adopt --hash <the HASH column>"
                );
            }
        }
        RepoCommand::Reclaim { path } => {
            let path = repo_ref(&path)?;
            let bytes = stores.reclaim_orphan(&path)?;
            println!("reclaimed {path} ({bytes} bytes); the name is usable again");
        }
        #[cfg(not(feature = "engine-znippy"))]
        RepoCommand::ArchiveOut { .. } | RepoCommand::ArchiveIn { .. } => {
            anyhow::bail!("repository archives are the znippy engine's; this server was built without it")
        }
        #[cfg(feature = "engine-znippy")]
        RepoCommand::ArchiveOut { .. } => {
            // Dispatched before the stores are opened (see `lib.rs`): an
            // archive must be takeable from a box that is serving, and opening
            // the control store would fight the running server for redb's lock.
            unreachable!("archive-out is handled by repo_archive_out_without_stores")
        }
        #[cfg(feature = "engine-znippy")]
        RepoCommand::ArchiveIn {
            file,
            as_path,
            no_catalog,
        } => {
            let path = repo_ref(&as_path)?;
            let dir = stores.repo_dir(&path)?;
            let r = crate::storage::znippy_engine::archive_in(&file, &dir)?;
            println!(
                "restored {} bytes into {} (generation {})",
                r.bytes_in,
                r.repo_dir.display(),
                r.to_generation
            );
            if !no_catalog {
                // The hash kind comes from the restored marker, never from a
                // flag: a repository restored as the wrong hash is a repository
                // no client can push to, and the archive already carries the
                // answer.
                let marker =
                    std::fs::read_to_string(dir.join(crate::storage::znippy_engine::OBJECT_FORMAT))
                        .unwrap_or_default();
                let hash = match marker.trim() {
                    "sha1" => gunnar_core::HashKind::Sha1,
                    "sha256" | "" => gunnar_core::HashKind::Sha256,
                    other => bail!("the archive names an unknown object format {other:?}"),
                };
                let spec = NewRepo::new(path.clone()).hash_kind(hash);
                // `adopt`: the store on disk is the one this command just
                // wrote, which is the single case where adopting is exactly
                // right rather than a resurrection.
                //
                // **A row that is already there is a SUCCESS, not a clash.**
                // The control document is restored separately (it carries the
                // default branch, the visibility and the grants, none of which
                // are in a repository archive), and when it has been the row
                // for this repository is already correct. Failing here would
                // make the two halves of a restore refuse to be done in the
                // order that keeps the most metadata.
                match stores.create_repo(spec, true) {
                    Ok(created) => {
                        println!(
                            "catalogued {} ({}, default branch {})",
                            created.key(),
                            created.hash_kind,
                            created.default_branch
                        );
                        println!("{}", encrypted_flag_note(&path));
                    }
                    Err(e) if format!("{e}").contains("already exists") => {
                        match stores.catalog().get(&path)? {
                            Some(have) => println!(
                                "the catalog already has {} ({}, default branch {}) — restored \
                                 from the control document, and left alone",
                                have.key(),
                                have.hash_kind,
                                have.default_branch
                            ),
                            None => println!("the catalog already has {path}"),
                        }
                    }
                    Err(e) => return Err(e),
                }
            }
        }
    }
    Ok(())
}

/// The first eight bytes of a digest, for a line a human reads.
#[cfg(feature = "engine-znippy")]
fn hex_head(d: &[u8; 32]) -> String {
    d[..8].iter().map(|b| format!("{b:02x}")).collect()
}

/// **`gunnar repo archive-out`, WITHOUT opening the control store.**
///
/// A backup that can only be taken from a stopped server is a backup nobody
/// takes, so this verb does not go through [`ServerStores`]: it resolves the
/// repository directory through the same `resolve_repo_path` sandbox the SSH
/// listener uses and reads three files out of it. Nothing it touches is
/// written by the running server except by *append*, which is exactly why no
/// quiesce is needed — see [`crate::storage::archive`].
#[cfg(feature = "engine-znippy")]
pub fn repo_archive_out_without_stores(
    data_root: &std::path::Path,
    command: RepoCommand,
) -> Result<()> {
    let RepoCommand::ArchiveOut {
        path,
        to,
        from_generation,
    } = command
    else {
        unreachable!("called with something other than archive-out")
    };
    let path = repo_ref(&path)?;
    let dir = path
        .path_under(&data_root.join("repos"))
        .with_context(|| format!("resolving the directory for {path}"))?;
    if !dir.is_dir() {
        bail!("{path} has nothing on disk at {}", dir.display());
    }
    let r = crate::storage::znippy_engine::archive_out(&dir, &to, from_generation)?;
    for s in &r.segments {
        println!(
            "  {:<44} {:>12} bytes  from {}",
            s.name, s.len, s.from_offset
        );
    }
    println!(
        "archived {path} -> {}  {} bytes, generation {} -> {}  sha256 {}",
        to.display(),
        r.bytes_out,
        r.from_generation,
        r.to_generation,
        hex_head(&r.file_sha256)
    );
    println!("next incremental: --from-generation {}", r.to_generation);
    Ok(())
}

/// `main` → `refs/heads/main`; a full ref name is left alone.
fn full_branch(name: &str) -> String {
    if name.starts_with("refs/") {
        name.to_owned()
    } else {
        format!("refs/heads/{name}")
    }
}

// ── principals and keys ─────────────────────────────────────────────────────

/// Run a `gunnar principal …` verb.
pub fn principal(stores: &ServerStores, command: PrincipalCommand) -> Result<()> {
    match command {
        PrincipalCommand::Add {
            id,
            name,
            kind,
            csr,
            server_generated_key,
            out,
            days,
        } => {
            let display = name.unwrap_or_else(|| id.clone());
            let created = stores.auth().add_principal(gunnar_core::Principal::new(
                &id,
                display,
                kind.into(),
            )?)?;
            println!("added principal {} ({:?})", created.id, created.kind);

            // The certificate, if one was asked for. **After** the principal
            // exists, because `register_certificate` refuses an unknown
            // principal and the whole point of the certificate-identity plan is
            // that the two are one act.
            match (csr, server_generated_key) {
                (Some(path), _) => {
                    let request = std::fs::read_to_string(&path).with_context(|| {
                        format!("reading the signing request at {}", path.display())
                    })?;
                    if let Some(claimed) = gunnar_core::issuer::csr_claimed_common_name(&request)? {
                        if claimed != created.id {
                            // Not an error: the request is being issued under
                            // the id that was asked for, which is the correct
                            // outcome. But an operator has to be able to see
                            // that the request asked for something else.
                            println!(
                                "note: the request claims CN={claimed}; it is being issued as \
                                 {} — a request does not choose its own name",
                                created.id
                            );
                        }
                    }
                    let issued = stores.pki()?.issue_from_csr(&created.id, &request, days)?;
                    hand_over(stores, &out, &issued)?;
                }
                (None, true) => {
                    let issued = stores.pki()?.issue_with_server_key(&created.id, days)?;
                    hand_over(stores, &out, &issued)?;
                }
                (None, false) => {
                    if stores
                        .control_store()
                        .read_doc(gunnar_core::ISSUER_REF)?
                        .is_some()
                    {
                        println!(
                            "no certificate was issued. `--csr <file>` signs a request whose \
                             key never leaves {}'s machine; `--server-generated-key` mints one \
                             here instead",
                            created.id
                        );
                    }
                }
            }
        }
        PrincipalCommand::List => {
            let mut principals = stores.auth().principals()?;
            principals.sort_by(|a, b| a.id.cmp(&b.id));
            for principal in principals {
                println!("{}\t{}\t{:?}", principal.id, principal.name, principal.kind);
            }
        }
        PrincipalCommand::Remove { id } => {
            let removed = stores.auth().remove_principal(&id)?;
            println!("removed principal {}", removed.id);
        }
    }
    Ok(())
}

/// Run a `gunnar key …` verb.
pub fn key(stores: &ServerStores, command: KeyCommand) -> Result<()> {
    match command {
        KeyCommand::Add {
            principal,
            key,
            cert,
        } => {
            let registered = match (key, cert) {
                (_, Some(path)) => {
                    let der = read_certificate(&path)?;
                    stores.auth().register_certificate(&principal, &der, None)?
                }
                (Some(key), None) => {
                    let line = key_argument(&key)?;
                    stores.auth().register_key(&principal, &line, None)?
                }
                (None, None) => bail!("give a public key or --cert"),
            };
            println!(
                "registered {} ({}) to {}",
                registered.fingerprint_sha256, registered.algo, registered.principal_id
            );
        }
        KeyCommand::List { principal } => {
            let mut keys = match principal {
                Some(id) => stores.auth().keys_of(&id)?,
                None => stores.auth().keys()?,
            };
            keys.sort_by(|a, b| a.fingerprint_sha256.cmp(&b.fingerprint_sha256));
            for key in keys {
                println!(
                    "{}\t{}\t{}",
                    key.fingerprint_sha256, key.algo, key.principal_id
                );
            }
        }
        KeyCommand::Revoke { fingerprint } => {
            let parsed = Fingerprint::parse(&fingerprint)?;
            match stores.auth().revoke_key(&parsed)? {
                Some(key) => println!("revoked {} from {}", parsed, key.principal_id),
                None => bail!("no key with fingerprint {parsed} is registered"),
            }
        }
    }
    Ok(())
}

/// `gunnar token`.
///
/// # The secret is printed once and then does not exist
///
/// `gunnar_core::mint_token_secret` produces it, `register_token` stores only
/// its sha-256, and this is the only place it is ever written out. It goes to
/// **stdout alone**, with the explanation on stderr, so `TOKEN=$(gunnar token
/// add ci)` captures the credential and nothing else.
pub fn token(stores: &ServerStores, command: gunnar_common::cli::TokenCommand) -> Result<()> {
    use gunnar_common::cli::TokenCommand;

    match command {
        TokenCommand::Add {
            principal,
            label,
            expires_in,
        } => {
            let secret = gunnar_core::mint_token_secret();
            let expires =
                expires_in.map(|days| gunnar_core::now_ms() + (days as i64) * 24 * 60 * 60 * 1000);
            let registered = stores
                .auth()
                .register_token(&principal, &secret, &label, expires)?;
            // stdout, once. The hash goes to the operator too, because it is
            // what `revoke` takes and it will never be derivable from anything
            // they still have.
            println!("{secret}");
            eprintln!(
                "minted for {} (hash {}); this is the only time the token is printed",
                registered.principal_id, registered.hash
            );
        }
        TokenCommand::List { principal } => {
            let mut tokens = stores.auth().tokens()?;
            if let Some(id) = &principal {
                tokens.retain(|token| &token.principal_id == id);
            }
            tokens.sort_by(|a, b| a.hash.cmp(&b.hash));
            for token in tokens {
                println!("{}\t{}\t{}", token.hash, token.principal_id, token.label);
            }
        }
        TokenCommand::Revoke { hash } => match stores.auth().revoke_token(&hash)? {
            Some(token) => println!("revoked {} from {}", token.hash, token.principal_id),
            None => bail!("no token with hash {hash} is registered"),
        },
    }
    Ok(())
}

// ── grants ──────────────────────────────────────────────────────────────────

fn scope_of(server: bool, namespace: Option<&str>, repo: Option<&str>) -> Result<Scope> {
    match (server, namespace, repo) {
        (true, _, _) => Ok(Scope::Server),
        (_, Some(name), _) => Ok(Scope::Namespace(name.to_owned())),
        (_, _, Some(path)) => Ok(Scope::Repo(repo_ref(path)?)),
        _ => bail!("a grant needs a scope: --server, --namespace <ns>, or --repo <ns>/<name>"),
    }
}

/// `gunnar grant`.
pub fn grant(stores: &ServerStores, args: GrantArgs) -> Result<()> {
    let scope = scope_of(args.server, args.namespace.as_deref(), args.repo.as_deref())?;
    let granted = stores.auth().grant(Grant {
        principal: args.principal.clone(),
        scope,
        level: args.level.into(),
    })?;
    println!(
        "{} now holds {} on {}",
        granted.principal, granted.level, granted.scope
    );
    // Nested namespaces do NOT inherit: `team` does not reach
    // `team/sub`. Said out loud at the moment the grant is made, because the
    // alternative is an operator discovering it as a permission denial.
    if let Scope::Namespace(name) = &granted.scope {
        if name.contains('/')
            || stores
                .catalog()
                .namespaces()?
                .iter()
                .any(|ns| ns.name.starts_with(&format!("{name}/")))
        {
            println!(
                "note: namespace grants match EXACTLY; this does not reach nested namespaces \
                 under {name}/"
            );
        }
    }
    Ok(())
}

/// `gunnar revoke`.
pub fn revoke(stores: &ServerStores, args: RevokeArgs) -> Result<()> {
    let scope = scope_of(args.server, args.namespace.as_deref(), args.repo.as_deref())?;
    if stores.auth().revoke_grant(&args.principal, &scope)? {
        println!("revoked {}'s access to {scope}", args.principal);
        Ok(())
    } else {
        bail!("{} holds no grant on {scope}", args.principal)
    }
}

// ── gc ──────────────────────────────────────────────────────────────────────

// `percent(Option<f32>) -> String` stood here, rendering a measured fraction or
// `?` for one that could not be read. Its whole reason to exist was that `?` and
// `0%` are different facts — and both facts were `RepackOptions`' delta figures,
// which the storage restart amputated with the packfile engine. `MaintenanceReport` carries
// no fraction: it reports `bytes_before`/`bytes_after`, `rows`, `retired_packs`,
// `verified` and `sections_sealed`, every one of them a count or a pair, which
// is the standing preference for a count over a ratio arrived at from the other
// direction. Nothing left in this crate has an `Option<f32>` to render.

/// `gunnar gc`, the repack seam, now driving `gunnar-store`'s `repack()`.
///
/// Each repository's pushed packs and loose objects are folded into **one new
/// pack**, which is put in place and verified before anything is removed
/// (never in place). A repository that is already consolidated is
/// reported as such and left alone: rewriting it would churn the pack id and
/// reclaim nothing.
///
/// **This is not the znippy cold tier `repack()` describes**, and the reason is
/// That tier as specified stores *inflated* object content, which
/// forecloses pack-copy permanently and makes every cold-tier fetch pay a full
/// recompression. `gunnar-store::repack`'s module documentation carries the
/// argument. What this verb does is git's own answer (many immutable
/// containers plus a periodic merge) with the client's deltas and deflate
/// preserved, which is what verbatim pack copy needs to exist at all.
///
/// The per-repository line reports **applied outputs**: objects, what was
/// folded, what was copied rather than recompressed, and the bytes each way. A
/// `gc` that printed only "done" would be indistinguishable from one that did
/// nothing (measured, not asserted).
///
/// Unreachable objects are **kept unless `--prune` says otherwise**.
/// `RepackOptions::prune_unreachable` had been implemented, tested
/// (`gunnar-store/tests/repack.rs`) and reachable from nothing: no flag, no
/// caller, so nothing gunnar ever reclaimed the objects of a rejected push, a
/// deleted branch or a force-pushed-away history. It is a separate and
/// destructive decision, which is why it is a flag and not a default, and the
/// number of objects it dropped is reported rather than assumed.
pub fn gc(stores: &ServerStores, args: GcArgs) -> Result<()> {
    let repos = match &args.path {
        Some(path) => {
            let path = repo_ref(path)?;
            vec![stores.catalog().open(&path)?]
        }
        None => stores.catalog().list(None)?,
    };

    // ── the storage restart: the tier maintains itself; there is no repack ──
    //
    // What stood here drove `gunnar_store::repack()` — consolidate the
    // packfiles in `objects/pack`, optionally re-encoding deltas, with a
    // `RepackOptions` carrying window, depth, delta order and policy. Every one
    // of those is a packfile concept and the packfile engine is amputated.
    //
    // The engine's own two verbs replace all of it: `seal` folds the live logs
    // into the reserved Arrow sections, `gc` computes the live set, tombstones
    // the packs it found wholly dead, drops the dead index rows and compacts.
    // Neither takes a window or a delta policy, which is why `GcArgs` no longer
    // offers them — a flag that parses and is ignored is worse than one that is
    // not there.
    let _ = &args.prune;
    let _ = &args.prune_grace;
    let mut done = 0usize;
    let mut failed = 0usize;
    let mut reclaimed: i64 = 0;
    for repo in &repos {
        match gc_one(stores, repo) {
            Ok(report) => {
                done += 1;
                reclaimed += report.bytes_before as i64 - report.bytes_after as i64;
                println!(
                    "{repo}\t{} -> {} bytes, {} live rows, {} pack(s) retired, {} section(s) \
                     sealed [{}]",
                    report.bytes_before,
                    report.bytes_after,
                    report.rows,
                    report.retired_packs,
                    report.sections_sealed,
                    report.strategy,
                );
            }
            Err(err) => {
                failed += 1;
                eprintln!("{repo}\tgc failed: {err:#}");
            }
        }
    }
    println!(
        "{done} of {} repository(ies) maintained, {reclaimed} bytes reclaimed",
        repos.len()
    );
    if failed > 0 {
        bail!("{failed} repository(ies) could not be maintained; see above");
    }
    Ok(())
}

/// **Maintain ONE repository. The only writer, and both doors call it.**
///
/// [`gc`] above is the CLI verb: it resolves what to maintain, loops, and
/// prints. `Repos.Gc` on the gRPC control plane resolves one repository and
/// reports bytes over the wire. Neither of those is maintenance — the
/// maintenance is here, once, so there is no second answer to "what does gc
/// do" to drift out of step with the first (one implementation). The split is deliberate:
/// **everything that is common is in this function and everything that
/// differs — argument parsing, the loop, the printed lines, the exit code — is
/// in the caller**, which is why extracting it changed no CLI output.
///
/// The span, the timing and the two log records live here rather than in the
/// callers, because the operator asking "what did the box just spend that I/O
/// on" gets the same record whether the run came from a shell or from a console
/// button. A failure logs `store.maintain_failed` here and is returned, so the
/// CLI can count it toward its exit code and the RPC can turn it into a
/// `Status` — one record, two policies.
pub fn gc_one(stores: &ServerStores, repo: &Repo) -> Result<MaintenanceReport> {
    // A snapshot streaming from its cut reads the objects reachable from the cut's tips, which
    // a gc of a moved ref could drop (R5.1-A): it waits for the stream.
    stores.wait_unpinned("a gc");
    let span = tracing::info_span!(target: TARGET_STORE, "store.maintain", repo = %repo);
    let _entered = span.enter();
    let started = std::time::Instant::now();
    let store = stores.repo_store(repo)?;
    match stores.engine().maintain(store.git_dir(), store.hash_kind()) {
        Ok(report) => {
            tracing::info!(
                target: TARGET_STORE,
                event = "store.maintained",
                strategy = %report.strategy,
                bytes_before = report.bytes_before,
                bytes_after = report.bytes_after,
                rows = report.rows,
                retired_packs = report.retired_packs,
                verified = report.verified,
                sections_sealed = report.sections_sealed,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "the store was sealed and compacted"
            );
            Ok(report)
        }
        Err(err) => {
            tracing::error!(
                target: TARGET_STORE,
                event = "store.maintain_failed",
                %err,
                "maintaining the store failed"
            );
            Err(err)
        }
    }
}

/// The bytes [`gc_one`] took off the filesystem, as one number.
///
/// `MaintenanceReport` carries the before/after PAIR on purpose — a lone
/// "reclaimed" figure cannot tell a run that compacted from one that did
/// nothing — so this is the lossy projection the wire asks for and not the
/// shape the report should have had. A run that grew the store (an in-place
/// compaction that had to write before it could retire) reports `0` rather
/// than an underflowed `u64`.
pub fn reclaimed_bytes(report: &MaintenanceReport) -> u64 {
    report.bytes_before.saturating_sub(report.bytes_after)
}

/// `gunnar reflog` — read the reference history, and optionally seal it.
///
/// This verb exists because the history it reads was **write-only**. Every
/// compaction wrote the reflog it took out of the live log to `refs-archive/`
/// as one complete Arrow IPC stream, and no production code in the workspace
/// ever opened one; a library reader with no caller would have been the same
/// bug one layer up.
///
/// It opens **no store, and no catalog**. `History::read` globs the archive and
/// reads the live log through the same torn-tail scan an open uses, taking no
/// lock, so this works while the server is serving — which is the only state a
/// repository worth asking about is ever in.
///
/// The catalog exclusion was found by running this verb against a live server
/// and being refused, *after* the library underneath it had been proven
/// lock-free: `ServerStores::open` opens the control plane's own store, so
/// resolving `<ns>/<name>` through the catalog took the very lock this verb
/// exists to avoid. That was the control *repository*'s lock when it was found;
/// since 2026-08-10 it is redb's exclusive lock on `control.redb`, which the
/// catalog, the identity table and now the secrets all sit behind. Same
/// exclusion, one file instead of two. The repository directory is resolved
/// through the same
/// `RepoRef::path_under` sandbox the SSH listener uses, and the hash kind comes
/// from the repository's own `config` (`gunnar_store::fs::stored_hash_kind`),
/// which is where `FsStore` already reads it.
///
/// The listing reports **applied output**: real oids either side of each move,
/// the actor, and whether the row is a compaction's snapshot rather than a
/// reference actually moving. A verb that printed only a count would be
/// indistinguishable from one that read nothing (measured, not asserted).
///
/// # 🔴 It reads a **retired** format, and it now says so instead of guessing
///
/// **2026-08-10.** What stood here was
/// `let store_dir = todo!("reflog: ask the engine for its ref journal")` — a
/// live `todo!()` in a shipped admin verb, which panicked the moment anybody
/// ran it. It was invisible until today because a crate that fails at name
/// resolution never reaches the lint passes.
///
/// The obvious repair — hand [`gunnar_core::History::read`] a plausible path —
/// is the **worst** available answer, and measurably so:
/// [`History::read`](gunnar_core::History::read) globs an archive directory and
/// tests the live log with `exists()`, so a wrong path is not an error. It is
/// **`0 pushes`**, printed with a straight face, and `--seal` would write a
/// valid, empty Arrow artifact over it.
///
/// So the path is **probed rather than assumed**, and the probe has exactly two
/// outcomes:
///
/// * `<git-dir>/gunnar-store/` holds [`REFS_LOG`] or [`REFS_ARCHIVE_DIR`] — the
///   retired filesystem backend's Arrow reflog is really there, and this verb
///   reads it, which is what it has always done and the only reader those bytes
///   have.
/// * it does not — the verb **refuses for that repository**, names the engine
///   this process runs and says where that engine's ref history actually is.
///   `read == 0` across every repository is a non-zero exit, because a script
///   must not read "nothing to say" as success.
///
/// # Why this is not re-pointed at the engine's journal
///
/// It is not a missing path; it is a missing capability, and the gap is three
/// deep:
///
/// 1. **Neither `GitOps` nor `GitServe` carries a ref-history accessor**, and
///    one does not belong on the contract: the `GitServe` reading contract fixes the
///    ordering rule that `GitServe` is *"defined by what the Arrow arm needs to
///    serve a clone well"*, and ref history serves no clone. Worse, the gix arm
///    could never conform — `storage-git-gix` opens its ref store with
///    `WriteReflog::Disable` and says so deliberately (*"a git server has no use
///    for a per-ref reflog file tree, and `GitOps` does not expose one"*). A
///    contract method one arm is structurally unable to answer is the
///    two-arms-disagree defect the `Harness` exists to catch, written into the
///    contract on purpose.
/// 2. **The znippy arm's journal is `pub(crate)`.** `refs.log` is one Arrow IPC
///    frame per push under the store root, complete (a seal *copies* it into the
///    archive and a compaction preserves row order), but `GitStore::ref_log()`
///    is not public and the file name is `GitStore`'s layout, not this crate's.
/// 3. **Its rows do not carry what this verb prints.** The columns are `name`,
///    `target`, `peeled`, `symref_target`, `push_seq`, `updated_ms`. The
///    previous value is derivable by replay; **`actor` does not exist at all**,
///    and `--seal`'s artifact is a `gunnar-core` format the engine has its own
///    answer to. Re-pointing this verb is therefore a redesign of what
///    `gunnar reflog` reports, which is a design decision and not a repair.
///
/// [`REFS_LOG`]: gunnar_core::arrow_log::REFS_LOG
/// [`REFS_ARCHIVE_DIR`]: gunnar_core::arrow_log::REFS_ARCHIVE_DIR
pub fn reflog(data_root: &std::path::Path, args: ReflogArgs) -> Result<()> {
    let repos_root = data_root.join(crate::stores::REPOS_DIR);
    let repos: Vec<(String, std::path::PathBuf)> = match &args.path {
        Some(path) => {
            let path = repo_ref(path)?;
            let dir = path
                .path_under(&repos_root)
                .with_context(|| format!("resolving the directory for {path}"))?;
            if !dir.join("HEAD").exists() {
                bail!("there is no repository at {}", dir.display());
            }
            vec![(path.key(), dir)]
        }
        None => bare_repositories(&repos_root)?,
    };
    if repos.is_empty() {
        println!("no repositories under {}", repos_root.display());
        return Ok(());
    }
    // One repository gets the path it was given; many get one artifact each
    // under it, because a single file would silently keep only the last.
    let seal_is_a_directory = args.seal.is_some() && repos.len() > 1;

    let mut read = 0usize;
    let mut failed = 0usize;
    let mut refused = 0usize;
    for (key, git_dir) in &repos {
        let store_dir = git_dir.join(LEGACY_STORE_DIR);
        // **Probed, never assumed.** `History::read` answers `0 pushes` for a
        // directory that is not there, so pointing it somewhere plausible would
        // print a number that means nothing. See this function's docs.
        if !legacy_reflog_present(&store_dir) {
            refused += 1;
            eprintln!(
                "{key}\tREFUSED: no reference history this verb can read — nothing at {}",
                store_dir.display()
            );
            continue;
        }
        let hash_kind = match stored_hash_kind(git_dir) {
            Ok(kind) => kind,
            Err(err) => {
                failed += 1;
                eprintln!("{key}\tFAILED: {err:#}");
                continue;
            }
        };

        let history = match gunnar_core::History::read(&store_dir, hash_kind) {
            Ok(history) => history,
            Err(err) => {
                failed += 1;
                eprintln!("{key}\tFAILED: {err}");
                continue;
            }
        };
        read += 1;

        match &args.seal {
            Some(target) => {
                let path = if seal_is_a_directory {
                    target.join(format!("{}.arrow", key.replace('/', "_")))
                } else {
                    target.clone()
                };
                let report = history.seal(&path)?;
                println!(
                    "{key}\t{} pushes ({} archived, {} live)\t{} rows\t{} B\t{:.1} B/push\t{:.2}x smaller than the {} B it read\t{}",
                    report.pushes,
                    history.archived_pushes(),
                    history.live_pushes(),
                    report.rows,
                    report.bytes,
                    report.bytes_per_push(),
                    report.shrink_x(),
                    report.bytes_in,
                    path.display(),
                );
            }
            None => {
                let transactions = history.transactions()?;
                // **`bytes_in` is on the line for a reason.** `History` reads a
                // live log through the same torn-tail scan an open uses, so a
                // log whose very first frame never landed has a valid prefix of
                // zero and is *correctly* replayed as no pushes at all. Without
                // the byte count, "0 pushes" out of a 40 KiB file and "0 pushes"
                // out of an empty directory print identically — and one of those
                // is a history an operator has lost (measured, not asserted).
                println!(
                    "{key}\t{} pushes ({} from the archive, {} live) in {} segment(s)\t{} rows\t{} B read",
                    history.pushes(),
                    history.archived_pushes(),
                    history.live_pushes(),
                    history.segments_read(),
                    transactions.len(),
                    history.bytes_in(),
                );
                let show = if args.all {
                    transactions.as_slice()
                } else {
                    let from = transactions.len().saturating_sub(args.limit);
                    &transactions[from..]
                };
                if show.len() < transactions.len() {
                    println!(
                        "  … {} older transaction(s) not shown; --all for every one",
                        transactions.len() - show.len()
                    );
                }
                for entry in show {
                    println!(
                        "  {:>6} {} {} -> {}{}",
                        entry.push_seq,
                        entry.refname,
                        value(entry.old_oid.as_ref(), entry.old_symref.as_deref()),
                        value(entry.new_oid.as_ref(), entry.new_symref.as_deref()),
                        if entry.is_snapshot() {
                            "  (compaction snapshot, not a move)"
                        } else {
                            ""
                        },
                    );
                }
            }
        }
    }

    // Printed once rather than per repository: on a server with two hundred
    // repositories the explanation is the same explanation two hundred times,
    // and the operator's next action is one action.
    if refused > 0 {
        eprintln!("\n{}", refusal_explanation());
    }
    if failed > 0 {
        bail!(
            "{failed} of {} repositor{} could not be read; {read} were",
            repos.len(),
            if repos.len() == 1 { "y" } else { "ies" }
        );
    }
    // **A refusal is an exit code, not a remark.** Every repository refused and
    // exit 0 is indistinguishable from a history that is genuinely empty, which
    // is the failure this whole verb was rewritten to stop making.
    if read == 0 {
        bail!(
            "no reference history was read from any of the {} repositor{} looked at",
            repos.len(),
            if repos.len() == 1 { "y" } else { "ies" }
        );
    }
    Ok(())
}

/// The store directory of the **retired** filesystem backend, verbatim.
///
/// It was `gunnar_store::fs::STORE_DIR` until the storage restart amputated the packfile
/// engine that owned it. The constant is repeated here rather than revived
/// there because nothing *writes* this layout any longer: it is the name of a
/// format on old disks, and the only code left that has any business knowing it
/// is the one reader of those bytes.
const LEGACY_STORE_DIR: &str = "gunnar-store";

/// Whether `store_dir` really holds a legacy Arrow reflog.
///
/// Both halves are checked because either one alone is a complete history:
/// a repository compacted since its last push has everything in
/// `refs-archive/` and a live log holding only its schema, and one never
/// compacted has the lot in the live log and no archive at all.
///
/// The two names come from [`gunnar_core::arrow_log`], which is the module that
/// wrote them, rather than being spelled again here (one implementation).
fn legacy_reflog_present(store_dir: &std::path::Path) -> bool {
    store_dir.join(gunnar_core::arrow_log::REFS_LOG).is_file()
        || store_dir
            .join(gunnar_core::arrow_log::REFS_ARCHIVE_DIR)
            .is_dir()
}

/// What an operator is told when there is no history for this verb to read.
///
/// It names the engine **this process** runs, because "where is my ref history"
/// has a different answer per arm and a generic refusal would send half of the
/// readers to the wrong place.
fn refusal_explanation() -> String {
    let engine = match crate::storage::Engine::selected() {
        Ok(engine) => engine.name(),
        // `selected()` refuses an engine name this binary does not carry. That
        // is worth reporting as itself rather than swallowing into "unknown",
        // but it must not turn an explanation into a second failure.
        Err(_) => "<unselectable>",
    };
    format!(
        "`reflog` reads the retired filesystem backend's Arrow reflog —\n\
         `<git-dir>/{LEGACY_STORE_DIR}/{log}` and `{archive}/` beside it. That backend went with\n\
         the packfile engine (the storage restart) and nothing writes those bytes any more, so\n\
         a repository created since has none and this verb has nothing to read.\n\
         \n\
         This process runs the `{engine}` engine. Where reference history is now:\n\
         \n\
         * znippy — one Arrow IPC frame per push in the store's own `refs.log`. It is real and it\n\
         \x20 is complete, but `GitStore::ref_log()` is `pub(crate)`, and its rows carry no actor\n\
         \x20 and no previous value, so this verb's output is not merely re-pointable at it.\n\
         * gix    — none at all. `storage-git-gix` opens its ref store with `WriteReflog::Disable`,\n\
         \x20 deliberately: a git server has no use for a per-ref reflog file tree.\n\
         \n\
         Neither `GitOps` nor `GitServe` carries a ref-history accessor, and the `GitServe`\n\
         reading contract's ordering rule says one does not belong on the contract. Making this\n\
         verb answer again is an engine-shaped seam plus a decision about what it should\n\
         report — see `agentO-report.md`.",
        log = gunnar_core::arrow_log::REFS_LOG,
        archive = gunnar_core::arrow_log::REFS_ARCHIVE_DIR,
    )
}

/// The hash kind an existing repository was created with.
///
/// One key decides it, `extensions.objectformat`, and its absence means sha1,
/// which is the reading git gives `repositoryformatversion = 0`. Reading that
/// one key by hand is deliberate: linking `gix-config` to parse a file whose
/// every other key describes a worktree this server does not have is what put
/// `gix-config`, and through it the rest of the porcelain, into the appliance
/// build.
///
/// It is here because reading a repository's **history** must not open its
/// store or its catalog — both are locked by whatever server is serving them,
/// and dodging that lock is this verb's entire reason to exist. The hash kind
/// is the one thing a history reader needs that the log itself does not carry.
///
/// Restored verbatim from `gunnar_store::fs::stored_hash_kind`, which the storage restart
/// deleted with the backend around it. It is the only reader of this key left
/// in the workspace, so there is nothing for it to drift from.
fn stored_hash_kind(git_dir: &std::path::Path) -> Result<gix_hash::Kind> {
    const OBJECT_FORMAT_KEY: &str = "objectformat";

    let config_path = git_dir.join("config");
    let text = match std::fs::read_to_string(&config_path) {
        Ok(text) => text,
        // A repository with no config at all is a sha1 repository, which is
        // also how git reads it.
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(gix_hash::Kind::Sha1),
        Err(err) => {
            return Err(anyhow::Error::from(err))
                .with_context(|| format!("reading {}", config_path.display()))
        }
    };

    let mut in_extensions = false;
    for line in text.lines() {
        let line = line.trim();
        if let Some(header) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {
            in_extensions = header.trim().eq_ignore_ascii_case("extensions");
            continue;
        }
        if !in_extensions {
            continue;
        }
        let Some((key, value)) = line.split_once('=') else {
            continue;
        };
        // Git's config keys are case-insensitive; the value here is not.
        if !key.trim().eq_ignore_ascii_case(OBJECT_FORMAT_KEY) {
            continue;
        }
        return match value.trim() {
            "sha256" => Ok(gix_hash::Kind::Sha256),
            "sha1" => Ok(gix_hash::Kind::Sha1),
            other => bail!(
                "{} declares an object format gunnar does not know: {other:?}",
                config_path.display()
            ),
        };
    }
    Ok(gix_hash::Kind::Sha1)
}

/// Every bare repository under `repos_root`, as `(<ns>/<name>, git dir)`.
///
/// Walked off the filesystem rather than listed from the catalog, for the same
/// reason the rest of this verb avoids it: the catalog is a store and a running
/// server holds its lock. A directory holding a `HEAD` is the same test
/// `FsStore::open_or_init` uses to decide a repository is already there.
fn bare_repositories(repos_root: &std::path::Path) -> Result<Vec<(String, std::path::PathBuf)>> {
    fn walk(
        dir: &std::path::Path,
        prefix: &str,
        out: &mut Vec<(String, std::path::PathBuf)>,
    ) -> Result<()> {
        let entries = match std::fs::read_dir(dir) {
            Ok(entries) => entries,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(err) => return Err(err.into()),
        };
        for entry in entries {
            let entry = entry?;
            if !entry.file_type()?.is_dir() {
                continue;
            }
            let name = entry.file_name().to_string_lossy().into_owned();
            let path = entry.path();
            if let Some(stem) = name.strip_suffix(".git") {
                if path.join("HEAD").exists() {
                    out.push((format!("{prefix}{stem}"), path));
                    continue;
                }
            }
            // Namespaces nest (`team/sub`), so this recurses rather than
            // scanning one level.
            walk(&path, &format!("{prefix}{name}/"), out)?;
        }
        Ok(())
    }
    let mut out = Vec::new();
    walk(repos_root, "", &mut out)?;
    out.sort();
    Ok(out)
}

/// One side of a reference move, as an operator reads it.
///
/// `-` rather than an empty column: a creation and a deletion must not look
/// like a truncated line.
fn value(oid: Option<&gunnar_store::gix_hash::ObjectId>, symref: Option<&str>) -> String {
    match (oid, symref) {
        (Some(id), _) => id.to_string(),
        (None, Some(name)) => format!("ref: {name}"),
        (None, None) => "-".to_owned(),
    }
}

/// `gunnar attest …` — the transparency log over push attestations.
///
/// None of these verbs writes a record. They sign, check and prove over the
/// lines every push already appended, which is what makes them safe to run at
/// any time and impossible to use to change what the log says.
pub fn attest(stores: &ServerStores, command: gunnar_common::cli::AttestCommand) -> Result<()> {
    use gunnar_common::cli::AttestCommand;

    let log = stores.transparency();
    match command {
        AttestCommand::Head { builds } => {
            let log = if builds { stores.build_log() } else { log };
            let key = stores.transparency_key()?;
            let signed = log.issue_head(&key, gunnar_grpc::bus::now_unix_ms())?;
            println!("size      {}", signed.head.size);
            println!("root      {}", signed.head.root);
            println!("signed by {}", signed.key_fingerprint);
            println!("appended  {}", log.heads_path().display());
            if signed.head.size == 0 {
                // Said out loud rather than left to be inferred from `size 0`.
                // A signed empty head is a real claim — "this server had
                // accepted no push at this moment" — and an operator who issued
                // one by accident should notice.
                println!();
                println!(
                    "This head covers an EMPTY log: it says this server had accepted no push \
                     when it was signed."
                );
            }
            Ok(())
        }
        AttestCommand::Audit { builds } => {
            let log = if builds { stores.build_log() } else { log };
            let key = stores.transparency_key()?;
            let verdict = log.audit(key.public_key())?;
            println!("{verdict}");
            if verdict.is_consistent() {
                return Ok(());
            }
            // A non-zero exit, because this is the verb an operator wires into
            // a check. Printing the verdict and exiting 0 would make a
            // tampered log look like a passing job.
            anyhow::bail!("the transparency log did not audit clean")
        }
        AttestCommand::Key => {
            let key = stores.transparency_key()?;
            println!(
                "{}",
                key.public_key()
                    .to_openssh()
                    .context("rendering the transparency log's public key")?
            );
            Ok(())
        }
        AttestCommand::Prove { pack_hash } => {
            let Some(inclusion) = log.prove_pack(&pack_hash)? else {
                anyhow::bail!(
                    "no attestation in this log names pack {pack_hash}. A proof for a record \
                     the log does not hold is not an empty proof."
                )
            };
            println!("index {}", inclusion.index);
            println!("size  {}", inclusion.size);
            println!("leaf  {}", inclusion.leaf);
            println!("root  {}", inclusion.root);
            for (step, hash) in inclusion.proof.iter().enumerate() {
                println!("proof[{step}] {hash}");
            }
            Ok(())
        }
    }
}

/// `gunnar verify` — walk the chain backwards from a line of source.
///
/// **The exit code is the verdict.** Zero only when every link is proven; a
/// broken link and an unproven one both exit non-zero, and they print
/// differently. A command whose output an operator has to read to learn whether
/// it passed is a command nobody wires into anything.
pub fn verify(data_root: &std::path::Path, args: gunnar_common::cli::VerifyArgs) -> Result<()> {
    let stores = std::sync::Arc::new(crate::stores::ServerStores::open(data_root)?);
    let plane = crate::control_plane::CoreControlPlane::new(std::sync::Arc::clone(&stores));
    let chain = match (&args.artifact, &args.repo) {
        (Some(sha256), _) => {
            crate::verify::verify_artifact(&stores, &plane, sha256, args.line.as_deref())?
        }
        (None, Some(repo)) => {
            let (namespace, name) = repo.split_once('/').with_context(|| {
                format!(
                    "{repo:?} is not `<namespace>/<repo>`; every repository lives in a \
                     namespace (multi-tenancy)"
                )
            })?;
            crate::verify::verify(
                &stores,
                &plane,
                &gunnar_proto::RepoPath::new(namespace, name),
                &args.at,
                args.line.as_deref(),
            )?
        }
        // clap's `required_unless_present` makes this unreachable; it is here
        // because a parser guarantee is not a type guarantee.
        (None, None) => anyhow::bail!("name a repository, or an --artifact to start from"),
    };
    println!("{chain}");
    if chain.is_unbroken() {
        return Ok(());
    }
    if chain.is_broken() {
        anyhow::bail!("the chain is broken");
    }
    anyhow::bail!("the chain is incomplete")
}

/// `gunnar blame <ns>/<repo> <path> [--rev]` — who wrote each line.
///
/// A separate process from a running server for the reason `verify` is: opening
/// `ServerStores` takes redb's exclusive lock, which a live server holds.
pub fn blame(data_root: &std::path::Path, args: gunnar_common::cli::BlameArgs) -> Result<()> {
    let (namespace, name) = args.repo.split_once('/').with_context(|| {
        format!(
            "{:?} is not `<namespace>/<repo>`; every repository lives in a \
             namespace (multi-tenancy)",
            args.repo
        )
    })?;
    let rows = crate::blame::blame_at(
        data_root,
        &gunnar_proto::RepoPath::new(namespace, name),
        &args.rev,
        &args.path,
    )?;
    // `<short-oid> <author> <line>`, one line each — the shape `git blame`
    // readers expect, trimmed to what an auditor asked: who, and what.
    for row in &rows {
        let hex = row.commit.to_hex().to_string();
        let short = &hex[..hex.len().min(12)];
        println!("{short} {}\t{}", row.author, String::from_utf8_lossy(&row.line));
    }
    Ok(())
}

/// `gunnar search <ns>/<repo> <pattern> [--rev] [--max-*]` — a literal string
/// across the latest revision's files.
pub fn search(data_root: &std::path::Path, args: gunnar_common::cli::SearchArgs) -> Result<()> {
    let (namespace, name) = args.repo.split_once('/').with_context(|| {
        format!(
            "{:?} is not `<namespace>/<repo>`; every repository lives in a \
             namespace (multi-tenancy)",
            args.repo
        )
    })?;
    let limits = crate::search::Limits {
        max_file_bytes: args.max_file_bytes,
        max_matches: args.max_matches,
    };
    let hits = crate::search::search_at(
        data_root,
        &gunnar_proto::RepoPath::new(namespace, name),
        &args.rev,
        &args.pattern,
        limits,
    )?;
    for hit in &hits {
        println!(
            "{}:{}:{}",
            hit.path,
            hit.line_number,
            String::from_utf8_lossy(&hit.line)
        );
    }
    Ok(())
}

/// `gunnar build …` — build provenance (roadmap phase 3, links 5–6).
pub fn build(stores: &ServerStores, command: gunnar_common::cli::BuildCommand) -> Result<()> {
    use gunnar_common::cli::BuildCommand;

    let log = stores.build_log();
    match command {
        BuildCommand::Attest {
            repo,
            commit,
            recipe,
            toolchain,
            builder,
            network,
            materials,
            artifacts,
        } => {
            // **Digested here, from the files.** A digest supplied on the
            // command line is a digest the caller chose, and the whole point of
            // the record is that it names what was actually produced.
            let subject = |path: &std::path::Path| -> Result<gunnar_core::Subject> {
                Ok(gunnar_core::Subject {
                    name: path
                        .file_name()
                        .map(|n| n.to_string_lossy().into_owned())
                        .unwrap_or_else(|| path.display().to_string()),
                    sha256: gunnar_core::provenance::digest_file(path)
                        .with_context(|| format!("hashing {}", path.display()))?,
                })
            };
            let now = gunnar_grpc::bus::now_unix_ms();
            let record = gunnar_core::BuildRecord {
                builder,
                source_repo: repo,
                source_commit: commit,
                recipe,
                toolchain,
                network,
                materials: materials
                    .iter()
                    .map(|p| subject(p))
                    .collect::<Result<_>>()?,
                subjects: artifacts
                    .iter()
                    .map(|p| subject(p))
                    .collect::<Result<_>>()?,
                started_at_unix_ms: now,
                finished_at_unix_ms: now,
            };
            let key = stores.provenance_key()?;
            let signed = gunnar_core::provenance::sign_build(&key, record)
                .context("signing the build record")?;
            let mut line = serde_json::to_string(&signed)?;
            line.push('\n');
            let _framed = gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))?;
            use std::io::Write as _;
            let mut file = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(log.records_path())?;
            file.write_all(line.as_bytes())?;

            println!("recorded  {}", signed.record.digest());
            println!("builder   {}", signed.record.builder);
            println!(
                "source    {}@{}",
                signed.record.source_repo, signed.record.source_commit
            );
            println!("network   {}", signed.record.network);
            for subject in &signed.record.subjects {
                println!("produced  {} {}", subject.sha256, subject.name);
            }
            println!("signed by {}", signed.key_fingerprint);
            println!();
            println!(
                "No signed head covers it yet. `gunnar attest head --builds` is what turns this \
                 record into evidence somebody else can check."
            );
            Ok(())
        }
        BuildCommand::Sbom { lock, artifact } => {
            let text = std::fs::read_to_string(&lock)
                .with_context(|| format!("reading {}", lock.display()))?;
            let packages = gunnar_core::sbom::parse_lock(&text);
            anyhow::ensure!(
                !packages.is_empty(),
                "{} yielded no packages. An empty SBOM is not a small one — it is a document \
                 that says this artifact contains nothing, and that is never true.",
                lock.display()
            );
            let sha256 = gunnar_core::provenance::digest_file(&artifact)
                .with_context(|| format!("hashing {}", artifact.display()))?;
            let name = artifact
                .file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_else(|| artifact.display().to_string());
            // The serial is the artifact's own digest. A random uuid would make
            // two SBOMs of the same artifact different documents; this way the
            // document is a function of what it describes.
            print!(
                "{}",
                gunnar_core::sbom::cyclonedx(&packages, &name, &sha256, &sha256)
            );
            Ok(())
        }
        BuildCommand::Admit { artifact } => {
            let sha256 = gunnar_core::provenance::digest_file(&artifact)
                .with_context(|| format!("hashing {}", artifact.display()))?;
            match crate::registry::admit(stores, &sha256)? {
                Ok(admitted) => {
                    println!("admitted  {sha256}");
                    println!("built by  {}", admitted.record.builder);
                    println!(
                        "source    {}@{}",
                        admitted.record.source_repo, admitted.record.source_commit
                    );
                    println!("record    {} of the build log", admitted.index);
                    println!(
                        "head      size {} root {}",
                        admitted.head.size, admitted.head.root
                    );
                    Ok(())
                }
                Err(refusal) => {
                    // Printed to stdout and then failed, so the reason survives
                    // in a log that captures one stream and the exit code is
                    // still the verdict.
                    println!("refused   {sha256}");
                    println!("{refusal}");
                    anyhow::bail!("the registry refuses this artifact")
                }
            }
        }
        BuildCommand::Run {
            repo,
            at,
            workspace,
            builder,
            allow_network,
        } => {
            let (namespace, name) = repo
                .split_once('/')
                .with_context(|| format!("{repo:?} is not `<namespace>/<repo>`"))?;
            let repo_ref = gunnar_core::RepoRef::new(namespace, name)
                .context("the repository path is not a legal one")?;
            let commit = resolve_commit(stores, &repo_ref, &at)?;
            let built = crate::runner::build_commit(
                stores,
                &repo_ref,
                commit,
                &workspace,
                &builder,
                allow_network,
            )?;
            println!(
                "built     {}@{}",
                built.record.source_repo, built.record.source_commit
            );
            println!("recipe    {}", built.record.recipe);
            println!("toolchain {}", built.record.toolchain);
            println!("network   {}", built.record.network);
            for subject in &built.record.subjects {
                println!("produced  {} {}", subject.sha256, subject.name);
            }
            println!("workdir   {}", built.workdir.display());
            println!();
            println!(
                "`gunnar attest head --builds` is what turns this record into evidence \
                 somebody else can check."
            );
            Ok(())
        }
        BuildCommand::Watch {
            workspace,
            builder,
            refs,
            allow_network,
            max_builds,
        } => {
            anyhow::bail!(
                "`gunnar build watch` needs the event bus the serving process owns, and this \
                 command opens the stores on its own. Start the server with `--runner \
                 --runner-workspace {} --runner-builder {}`{}{}; the runner then shares the \
                 bus every push publishes on.\n\n\
                 A second process cannot subscribe to an in-process bus, and a runner that \
                 polled the log instead would build a push some time after it landed rather \
                 than because it landed.",
                workspace.display(),
                builder,
                if allow_network {
                    " --runner-allow-network"
                } else {
                    ""
                },
                match (refs.is_empty(), max_builds) {
                    (true, None) => String::new(),
                    _ => " (--ref/--max-builds are `watch` options; the served runner builds \
                         every branch)".to_string(),
                }
            )
        }
        BuildCommand::Image {
            artifact,
            entrypoint,
            builder,
        } => {
            let bytes = std::fs::read(&artifact)
                .with_context(|| format!("reading {}", artifact.display()))?;
            let artifact_digest = gunnar_core::provenance::digest_bytes(&bytes);

            // **The chain, one link at a time.** An image built from a binary
            // nobody attested to is an image whose provenance stops one link
            // short — and it would look exactly as green as one that does not.
            let admitted = match crate::registry::admit(stores, &artifact_digest)? {
                Ok(admitted) => admitted,
                Err(refusal) => {
                    anyhow::bail!("the binary this image would carry is not admitted: {refusal}")
                }
            };

            let registry = crate::oci::Registry::under(stores.root());
            let image = crate::oci::build_and_store(&registry, &bytes, &entrypoint)?;
            registry.put_blob(&image.manifest)?;

            // The image's own digest, attested in turn — same record type, same
            // key, same log. The material is the binary, which is what makes
            // this a link rather than a second, parallel claim.
            let now = gunnar_grpc::bus::now_unix_ms();
            let record = gunnar_core::BuildRecord {
                builder,
                source_repo: admitted.record.source_repo.clone(),
                source_commit: admitted.record.source_commit.clone(),
                recipe: format!("oci image of {entrypoint}"),
                toolchain: "gunnar oci".to_owned(),
                // No build ran: an image is assembled from bytes that already
                // exist. Recording `network: false` here is a fact and not a
                // sandbox claim.
                network: false,
                materials: vec![gunnar_core::Subject {
                    name: artifact
                        .file_name()
                        .map(|n| n.to_string_lossy().into_owned())
                        .unwrap_or_else(|| artifact.display().to_string()),
                    sha256: artifact_digest.clone(),
                }],
                subjects: vec![gunnar_core::Subject {
                    name: format!("{entrypoint} (oci manifest)"),
                    sha256: image.digest.trim_start_matches("sha256:").to_owned(),
                }],
                started_at_unix_ms: now,
                finished_at_unix_ms: now,
            };
            let key = stores.provenance_key()?;
            let signed = gunnar_core::provenance::sign_build(&key, record)
                .context("signing the image's build record")?;
            let mut line = serde_json::to_string(&signed)?;
            line.push('\n');
            let _framed = gunnar_core::txlog::sink::append_line(stores.build_log().records_path(), line.trim_end_matches('\n'))?;
            use std::io::Write as _;
            let mut file = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(stores.build_log().records_path())?;
            file.write_all(line.as_bytes())?;

            println!("image     {}", image.digest);
            println!("config    {}", image.config_digest);
            println!("layer     {}", image.layer_digest);
            println!("files     {}", image.files);
            println!("from      {}", artifact_digest);
            println!(
                "source    {}@{}",
                signed.record.source_repo, signed.record.source_commit
            );
            println!();
            println!(
                "Zero CVEs is this number: {} file(s) in the image's filesystem, on no base at \
                 all. Next: `gunnar attest head --builds`, then `gunnar registry push`.",
                image.files
            );
            Ok(())
        }
        BuildCommand::List => {
            let key = stores.provenance_key()?;
            let mut any = false;
            for line in log.lines_of()? {
                let signed: gunnar_core::SignedBuild =
                    serde_json::from_str(&line).context("decoding a build record")?;
                let verdict = match gunnar_core::provenance::verify_build(&signed, key.public_key())
                {
                    Ok(()) => "ok ".to_owned(),
                    Err(problem) => format!("BAD ({problem})"),
                };
                println!(
                    "{verdict} {} {}@{} {}",
                    signed.record.digest(),
                    signed.record.source_repo,
                    signed.record.source_commit,
                    signed.record.recipe
                );
                any = true;
            }
            if !any {
                // Said rather than left as blank output: an empty listing and a
                // listing that failed to read look the same on a terminal.
                println!("this server has attested to no builds");
            }
            Ok(())
        }
        BuildCommand::Key => {
            let key = stores.provenance_key()?;
            println!(
                "{}",
                key.public_key()
                    .to_openssh()
                    .context("rendering the provenance public key")?
            );
            Ok(())
        }
    }
}

/// `gunnar forge …` — issues, reviews and merges as signed events.
pub fn forge(stores: &ServerStores, command: gunnar_common::cli::ForgeCommand) -> Result<()> {
    use gunnar_common::cli::ForgeCommand;

    let log = stores.forge_log();

    /// Read every event, verifying nothing: the fold is over what was written,
    /// and verification is a separate question with its own answer.
    fn events(log: &crate::attest::TransparencyLog) -> Result<Vec<gunnar_core::SignedEvent>> {
        log.lines_of()?
            .iter()
            .map(|line| serde_json::from_str(line).context("decoding a forge event"))
            .collect()
    }

    fn append(
        stores: &ServerStores,
        identity: &std::path::Path,
        event: gunnar_core::ForgeEvent,
    ) -> Result<gunnar_core::SignedEvent> {
        let pem = std::fs::read_to_string(identity)
            .with_context(|| format!("reading {}", identity.display()))?;
        let key = gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .with_context(|| format!("{} is not an OpenSSH private key", identity.display()))?;
        // **The key must already be in the ACL table.** A signature by a key
        // gunnar does not know is not evidence to anybody reading this log
        // later — they would have nothing to check it against — so it is
        // refused here rather than stored and found unverifiable.
        let fingerprint = key
            .public_key()
            .fingerprint(gunnar_core::ssh_key::HashAlg::Sha256)
            .to_string();
        let known = stores
            .auth()
            .signing_keys()?
            .into_iter()
            .any(|row| row.fingerprint_sha256 == fingerprint);
        anyhow::ensure!(
            known,
            "{fingerprint} is not registered on this server, so nobody reading this event later \
             could check its signature. Register it with `gunnar key add` first."
        );
        let signed =
            gunnar_core::forge::sign_event(&key, event).context("signing the forge event")?;
        let mut line = serde_json::to_string(&signed)?;
        line.push('\n');
        let _framed = gunnar_core::txlog::sink::append_line(stores.forge_log().records_path(), line.trim_end_matches('\n'))?;
        use std::io::Write as _;
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(stores.forge_log().records_path())?;
        file.write_all(line.as_bytes())?;
        Ok(signed)
    }

    match command {
        ForgeCommand::Comment {
            repo,
            pull,
            body,
            actor,
            identity,
        } => {
            append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    pull,
                    actor,
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::Commented {
                        body,
                        path: String::new(),
                        line: 0,
                        blob_oid: String::new(),
                    },
                },
            )?;
            println!("commented on {repo}#{pull}");
            Ok(())
        }
        ForgeCommand::Approve {
            repo,
            pull,
            commit,
            actor,
            identity,
        } => {
            let signed = append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    pull,
                    actor: actor.clone(),
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::Approved {
                        commit: commit.clone(),
                    },
                },
            )?;
            println!("{actor} approved {commit} on {repo}#{pull}");
            println!("signed by {}", signed.key_fingerprint);
            Ok(())
        }
        ForgeCommand::Reject {
            repo,
            pull,
            commit,
            reason,
            actor,
            identity,
        } => {
            let signed = append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    pull,
                    actor: actor.clone(),
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::Rejected {
                        commit: commit.clone(),
                        reason,
                    },
                },
            )?;
            println!("{actor} withdrew their approval of {commit} on {repo}#{pull}");
            println!("signed by {}", signed.key_fingerprint);
            Ok(())
        }
        ForgeCommand::Show { repo, pull } => {
            let all = events(&log)?;
            let mine: Vec<gunnar_core::ForgeEvent> = all
                .iter()
                .filter(|e| e.event.repo == repo)
                .map(|e| e.event.clone())
                .collect();
            let state = gunnar_core::forge::fold(pull, &mine);
            // The REVIEW only. Whether `{repo}#{pull}` exists, what it is
            // called and whether it is open are `PullRequests`' answers, and
            // printing a second copy of them here would be a twin.
            println!("{repo}#{pull}  review");
            println!("comments  {}", state.comments);
            if state.approvals.is_empty() {
                println!("approvals none");
            }
            // **Each approval is re-verified against the ACL table**, here, when
            // it is read — not trusted because it is in the file. That is the
            // difference the whole design is about, and `PullRequests.Merge`
            // asks the same question before it moves a ref.
            let keys: Vec<(String, gunnar_core::ssh_key::PublicKey)> = stores
                .auth()
                .signing_keys()?
                .iter()
                .filter_map(|row| {
                    gunnar_core::ssh_key::PublicKey::from_openssh(&row.openssh)
                        .ok()
                        .map(|key| (row.fingerprint_sha256.clone(), key))
                })
                .collect();
            for (actor, commit) in &state.approvals {
                let signed = all.iter().rev().find(|e| {
                    e.event.repo == repo
                        && e.event.pull == pull
                        && e.event.actor == *actor
                        && matches!(&e.event.event, gunnar_core::Event::Approved { commit: c } if c == commit)
                });
                let verdict = match signed {
                    None => "NO EVENT".to_owned(),
                    Some(signed) => match keys
                        .iter()
                        .find(|(fingerprint, _)| *fingerprint == signed.key_fingerprint)
                    {
                        None => "UNKNOWN KEY".to_owned(),
                        Some((_, key)) => match gunnar_core::forge::verify_event(signed, key) {
                            Ok(()) => "verified".to_owned(),
                            Err(problem) => format!("BAD ({problem})"),
                        },
                    },
                };
                println!("approval  {actor} → {commit}  [{verdict}]");
            }
            Ok(())
        }
        ForgeCommand::Issue(issue) => forge_issue(stores, &log, issue),
    }
}

/// `gunnar forge issue …` — issues as signed events, folded from the one log.
fn forge_issue(
    stores: &ServerStores,
    log: &crate::attest::TransparencyLog,
    command: gunnar_common::cli::IssueCommand,
) -> Result<()> {
    use gunnar_common::cli::IssueCommand;

    /// Read every event, verifying nothing — the fold is over what was written.
    fn events(log: &crate::attest::TransparencyLog) -> Result<Vec<gunnar_core::SignedEvent>> {
        log.lines_of()?
            .iter()
            .map(|line| serde_json::from_str(line).context("decoding a forge event"))
            .collect()
    }

    /// One repository's events, in log order, for the fold.
    fn repo_events(
        log: &crate::attest::TransparencyLog,
        repo: &str,
    ) -> Result<Vec<gunnar_core::ForgeEvent>> {
        Ok(events(log)?
            .into_iter()
            .filter(|e| e.event.repo == repo)
            .map(|e| e.event)
            .collect())
    }

    /// Sign an event with the actor's own key, after checking the key is in the
    /// ACL table, and append it. The same door `gunnar forge comment` uses —
    /// one writer, one verification of what makes an event evidence (one implementation).
    fn append(
        stores: &ServerStores,
        identity: &std::path::Path,
        event: gunnar_core::ForgeEvent,
    ) -> Result<gunnar_core::SignedEvent> {
        let pem = std::fs::read_to_string(identity)
            .with_context(|| format!("reading {}", identity.display()))?;
        let key = gunnar_core::ssh_key::PrivateKey::from_openssh(&pem)
            .with_context(|| format!("{} is not an OpenSSH private key", identity.display()))?;
        let fingerprint = key
            .public_key()
            .fingerprint(gunnar_core::ssh_key::HashAlg::Sha256)
            .to_string();
        let known = stores
            .auth()
            .signing_keys()?
            .into_iter()
            .any(|row| row.fingerprint_sha256 == fingerprint);
        anyhow::ensure!(
            known,
            "{fingerprint} is not registered on this server, so nobody reading this event later \
             could check its signature. Register it with `gunnar key add` first."
        );
        let signed =
            gunnar_core::forge::sign_event(&key, event).context("signing the forge event")?;
        let mut line = serde_json::to_string(&signed)?;
        line.push('\n');
        let _framed = gunnar_core::txlog::sink::append_line(stores.forge_log().records_path(), line.trim_end_matches('\n'))?;
        use std::io::Write as _;
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(stores.forge_log().records_path())?;
        file.write_all(line.as_bytes())?;
        Ok(signed)
    }

    match command {
        IssueCommand::Open {
            repo,
            title,
            body,
            actor,
            identity,
        } => {
            append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    // An opening has no number yet: the fold assigns it.
                    pull: 0,
                    actor: actor.clone(),
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::IssueOpened {
                        title: title.clone(),
                        body,
                    },
                },
            )?;
            // Read back the number the fold now gives it: the Nth opening is
            // issue N, so the one just written is the last.
            let number = gunnar_core::forge::fold_issues(&repo_events(log, &repo)?)
                .last()
                .map(|i| i.number)
                .unwrap_or(0);
            println!("opened {repo}#{number}  {title}");
            println!("{actor} filed it, signed with their own key");
            Ok(())
        }
        IssueCommand::Comment {
            repo,
            number,
            body,
            actor,
            identity,
        } => {
            anyhow::ensure!(
                gunnar_core::forge::fold_issue(number, &repo_events(log, &repo)?)
                    .is_some_and(|i| i.open),
                "no OPEN issue #{number} in {repo}: a comment on a closed or absent issue is \
                 evidence about nothing"
            );
            append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    pull: number,
                    actor,
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::IssueCommented { body },
                },
            )?;
            println!("commented on {repo}#{number}");
            Ok(())
        }
        IssueCommand::Close {
            repo,
            number,
            reason,
            actor,
            identity,
        } => {
            anyhow::ensure!(
                gunnar_core::forge::fold_issue(number, &repo_events(log, &repo)?)
                    .is_some_and(|i| i.open),
                "no OPEN issue #{number} in {repo}: it is already closed, or was never opened"
            );
            append(
                stores,
                &identity,
                gunnar_core::ForgeEvent {
                    repo: repo.clone(),
                    pull: number,
                    actor: actor.clone(),
                    at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                    event: gunnar_core::Event::IssueClosed { reason },
                },
            )?;
            println!("{actor} closed {repo}#{number}");
            Ok(())
        }
        IssueCommand::List { repo } => {
            let issues = gunnar_core::forge::fold_issues(&repo_events(log, &repo)?);
            if issues.is_empty() {
                println!("{repo} has no issues");
                return Ok(());
            }
            for issue in issues {
                let state = if issue.open { "open  " } else { "closed" };
                println!(
                    "{repo}#{}  [{state}]  {}  ({} comment(s), by {})",
                    issue.number, issue.title, issue.comments, issue.author
                );
            }
            Ok(())
        }
        IssueCommand::Show { repo, number } => {
            let all = repo_events(log, &repo)?;
            let Some(issue) = gunnar_core::forge::fold_issue(number, &all) else {
                anyhow::bail!("no issue #{number} in {repo}: nobody has opened one");
            };
            let state = if issue.open { "open" } else { "closed" };
            println!("{repo}#{number}  [{state}]  {}", issue.title);
            println!("opened by {}", issue.author);
            if !issue.body.is_empty() {
                println!();
                println!("{}", issue.body);
            }
            println!();
            println!("comments  {}", issue.comments);
            Ok(())
        }
    }
}

/// A ref name or hex oid, as the commit it names.
///
/// The same two rules `Refs.Log` applies, spelled once: a hex oid of the
/// repository's own hash kind is taken as one, and anything else is validated
/// as a reference name and resolved against the snapshot.
fn resolve_commit(
    stores: &ServerStores,
    repo: &gunnar_core::RepoRef,
    rev: &str,
) -> Result<gix_hash::ObjectId> {
    let entry = stores
        .catalog()
        .get(repo)?
        .with_context(|| format!("no such repository: {repo}"))?;
    let store = stores.repo_store(&entry)?;
    let git = store.serve();
    if let Ok(oid) = gix_hash::ObjectId::from_hex(rev.as_bytes()) {
        anyhow::ensure!(
            oid.kind() == store.hash_kind(),
            "{rev} is a {} oid; this repository is {}",
            oid.kind(),
            store.hash_kind()
        );
        return Ok(oid);
    }
    let snapshot = gunnar_wire::RefSnapshot::read(git)?;
    let resolved = if rev == gunnar_store::HEAD {
        snapshot.head(git)?
    } else {
        snapshot.resolve(rev)?
    };
    resolved.with_context(|| format!("{rev} names no commit in {repo}"))
}

/// `gunnar registry …` — the artifact registry (roadmap phase 4).
pub fn registry(
    stores: &std::sync::Arc<ServerStores>,
    command: gunnar_common::cli::RegistryCommand,
) -> Result<()> {
    use gunnar_common::cli::RegistryCommand;

    let registry = crate::oci::Registry::under(stores.root());
    match command {
        RegistryCommand::Push { name, tag, digest } => {
            let manifest = registry
                .get_blob(&digest)?
                .with_context(|| format!("{digest} is not in this registry"))?;
            match registry.put_manifest(stores, &name, &tag, &manifest)? {
                Ok(landed) => {
                    println!(
                        "pushed {}:{} → {}",
                        landed.name, landed.reference, landed.digest
                    );
                    Ok(())
                }
                Err(refusal) => {
                    println!("refused {digest}");
                    println!("{refusal}");
                    anyhow::bail!("the registry refuses this manifest")
                }
            }
        }
        RegistryCommand::Resolve { name, tag } => {
            let digest = registry
                .resolve(&name, &tag)?
                .with_context(|| format!("{name}:{tag} is not in this registry"))?;
            println!("{digest}");
            Ok(())
        }
        RegistryCommand::Pull { name, tag, line } => {
            let digest = registry
                .resolve(&name, &tag)?
                .with_context(|| format!("{name}:{tag} is not in this registry"))?;
            // Read it back through the store, which verifies the blob still
            // hashes to its own name — a pull that skipped that would serve
            // whatever is on the disk under the right filename.
            let manifest = registry
                .get_blob(&digest)?
                .with_context(|| format!("{digest} is tagged and its manifest is not stored"))?;
            println!("{}", String::from_utf8_lossy(&manifest));
            println!();

            // The handle we were given, not a second one: redb takes an
            // exclusive lock on `control.redb`, so opening a second
            // `ServerStores` in the same process is `Database already open`.
            let plane = crate::control_plane::CoreControlPlane::new(std::sync::Arc::clone(stores));
            let chain = crate::verify::verify_artifact(
                stores,
                &plane,
                digest.trim_start_matches("sha256:"),
                line.as_deref(),
            )?;
            println!("{chain}");
            if chain.is_unbroken() {
                return Ok(());
            }
            if chain.is_broken() {
                anyhow::bail!("the chain behind this image is broken");
            }
            anyhow::bail!("the chain behind this image is incomplete")
        }
    }
}

/// **What a per-repository restore cannot bring back** (wiring tracer B 2026-10-08): a
/// repository archive carries the objects and the refs, never the control document's settings
/// — and a repository catalogued afresh by `archive-in` is NOT encrypted, whatever it was. The
/// unsafe direction, so it is said, with the one command that sets it again.
#[cfg(feature = "engine-znippy")]
pub(crate) fn encrypted_flag_note(path: &RepoRef) -> String {
    format!(
        "note: a repository archive does not carry the account's settings, so {path} is NOT \
         encrypted now, whatever it was. If it was, the account admin turns it on again: at the \
         console (the repository's Settings, \"Turn encryption on\"), or gunnar control repo \
         set-encrypted {path} on where the control plane is in reach; or restore the control \
         document as well"
    )
}

#[cfg(all(test, feature = "engine-znippy"))]
mod tests {
    use super::*;

    #[test]
    fn a_repository_restored_without_its_control_document_is_told_its_encryption_is_not_back() {
        let path = RepoRef::new("acme", "vault").expect("a legal path");
        let note = encrypted_flag_note(&path);
        assert!(note.contains("encrypted"), "{note}");
        assert!(note.contains("gunnar control repo set-encrypted acme/vault on"), "the one command that sets it again: {note}");
    }

    /// ★ An installed estate binds its control plane to the WireGuard overlay, so the account
    /// admin who restored there has no `gunnar control` in reach: the note also names the door
    /// they do have — the repository's Settings at the console ("Turn encryption on").
    #[test]
    fn the_note_names_the_consoles_switch_an_estate_admin_can_reach() {
        let path = RepoRef::new("acme", "vault").expect("a legal path");
        let note = encrypted_flag_note(&path);
        assert!(note.contains("console") && note.contains("Turn encryption on"), "the console's switch: {note}");
    }
}