gunnar-server 3.0.0

//! **`Mirror.Sync` against a real upstream: enqueue, fetch, apply, and read the
//! objects back.**
//!
//! The verb answered `UNIMPLEMENTED` until now, so there is no round-trip to
//! regression-test and every assertion here is on **applied output**: a
//! repository that was empty holding the upstream's commit object, its
//! reference at the upstream's tip, and a job whose progress pair moved from
//! `(0, N)` to `(N, N)` while it did.
//!
//! # What each half proves, and why neither alone is enough
//!
//! * **SSH** (`a_mirror_sync_over_ssh_…`) runs against a whole real gunnar:
//!   `Server::start` on a loopback russh listener, the real key table, the real
//!   `WireTransport`, the real `upload_pack` over a real znippy store. The
//!   client half is `gunnar_client::connect_upload_pack` — the fetch-side twin
//!   of the dialler replication pushes with. Nothing about that path is
//!   a fixture, which is the only way to know the two ends of `gunnar-wire`
//!   actually meet.
//! * **Smart HTTP** (`a_mirror_sync_over_smart_http_…`) runs against a loopback
//!   HTTP/1.1 listener in this file that drives the **same**
//!   `UploadPack::serve_session(V2, StatelessRpc, …)` gunnar's own
//!   `gunnar-http` drives, plus the two routes (`GET /info/refs`,
//!   `POST /git-upload-pack`) it serves. What is written here is the HTTP
//!   *framing* and nothing else — and framing is precisely what the adapter
//!   under test does, so a fixture that reimplemented the protocol would be
//!   testing itself.
//!
//!   It is written here rather than reached through `gunnar-http` for one
//!   reason: this file has to serve a **401**, and refusing a request before it
//!   reaches the git seam is not something that crate can be asked to do
//!   without inventing an auth mode nothing ships.
//!
//! # Both hash kinds, on purpose
//!
//! The git wire protocol's rule is that nothing may assume sha1, and a mirror is the one verb
//! where the assumption would be invisible: a sha1 client against a sha256
//! remote does not fail, it negotiates an empty answer, and the operator is
//! left with a command that reports success and copied nothing.
//! `fetch_pack::greet` refuses the mismatch by name, and these tests run the
//! whole path at each kind so the refusal is never the thing being exercised.
//!
//! # The credential is asserted ON THE BYTES
//!
//! `a_private_http_upstream_…` stands up an upstream that answers 401 without
//! the token and serves with it. That proves the token is *sent*. What matters
//! as much is that it is never seen anywhere else, so the test taps the tracing
//! subscriber, collects every RPC response and every error string, and asserts
//! the token's bytes appear in **none** of them. A `has_credential` flag that
//! is right while the value leaks through an error message is the failure this
//! shape exists to catch.
#![cfg(feature = "mirror-http")] // A32-18: the airgap default carries no HTTP client; the appliance rows and CI name `mirror-http`.

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use gix_object::Kind;
#[cfg(feature = "replication")]
use gunnar_core::model::{Grant, Level, PrincipalKind, Visibility};
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_grpc::seam::ControlPlane as _;
use gunnar_proto::{MirrorCredentialKind, MirrorJobState, RepoPath};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_store::conformance::pack::PackBuilder;
use gunnar_store::git_storage_trait::{GitOps, GitServe};

type Oid = gix_hash::ObjectId;

// ════════════════════════════════════════════════════════════════════════════
// Fixtures: a repository with real objects in it
// ════════════════════════════════════════════════════════════════════════════

fn compute(hash: gix_hash::Kind, kind: Kind, content: &[u8]) -> Oid {
    gix_object::compute_hash(hash, kind, content).expect("hashing an object")
}

fn tree_bytes(entries: &[(&str, &str, Oid)]) -> Vec<u8> {
    let mut body = Vec::new();
    for (mode, name, oid) in entries {
        body.extend_from_slice(mode.as_bytes());
        body.push(b' ');
        body.extend_from_slice(name.as_bytes());
        body.push(0);
        body.extend_from_slice(oid.as_bytes());
    }
    body
}

fn commit_bytes(tree: Oid, message: &str) -> Vec<u8> {
    let who = "Alice <alice@example.com> 1700000000 +0000";
    let mut body = format!("tree {}\n", tree.to_hex());
    body.push_str(&format!("author {who}\n"));
    body.push_str(&format!("committer {who}\n\n"));
    body.push_str(message);
    body.push('\n');
    body.into_bytes()
}

/// One commit, one tree, one blob — three objects — under `refs/heads/main`.
///
/// Returns the commit id and the blob id: the commit is what the reference must
/// end up at on the mirror, and the blob is what proves the whole closure was
/// copied rather than only the tip.
///
/// Gated on `replication` because its one caller is `SshUpstream::start`, which
/// is itself `#[cfg(feature = "replication")]`. Ungated it is `dead_code` on a
/// default build — the one warning `cargo check -p gunnar-server --all-targets`
/// still emitted on 2026-09-12. `plant_a_commit_of` below stays ungated: it has
/// callers on both sides of the feature.
#[cfg(feature = "replication")]
fn plant_a_commit(git: &dyn GitServe, hash: gix_hash::Kind) -> (Oid, Oid) {
    plant_a_commit_of(git, hash, b"the quick brown fox\n".to_vec())
}

/// **Incompressible bytes**, from a linear congruential generator so the fixture
/// is deterministic and pulls in no dependency.
///
/// Incompressible matters: the packfile is deflated, and a blob of zeroes would
/// produce a pack of a few hundred bytes however many bytes went in. See
/// [`the_progress_pair_moves_from_zero_to_n_while_the_sync_runs`] for why one
/// fixture needs a pack that does not fit in a single sideband packet.
fn noise(bytes: usize) -> Vec<u8> {
    let mut state: u64 = 0x2545_F491_4F6C_DD1D;
    (0..bytes)
        .map(|_| {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (state >> 33) as u8
        })
        .collect()
}

/// [`plant_a_commit`] over a caller-chosen blob.
fn plant_a_commit_of(git: &dyn GitServe, hash: gix_hash::Kind, blob: Vec<u8>) -> (Oid, Oid) {
    let blob_id = compute(hash, Kind::Blob, &blob);
    let tree = tree_bytes(&[("100644", "fox.txt", blob_id)]);
    let tree_id = compute(hash, Kind::Tree, &tree);
    let commit = commit_bytes(tree_id, "first");
    let commit_id = compute(hash, Kind::Commit, &commit);

    let mut pack = PackBuilder::new(hash);
    for (kind, content) in [
        (Kind::Blob, blob),
        (Kind::Tree, tree),
        (Kind::Commit, commit),
    ] {
        pack.object(kind, content);
    }
    let fixture = pack.build().expect("building the pack");
    git.put_pack(&fixture.bytes).expect("storing the pack");
    git.update_ref("refs/heads/main", None, Some(commit_id.as_bytes()))
        .expect("pointing main at the commit");
    (commit_id, blob_id)
}

/// The oid `name` resolves to in a store, or `None` when it holds no such
/// reference. The applied output every assertion below is really about.
fn tip(git: &dyn GitServe, name: &str) -> Option<Oid> {
    GitOps::refs(git)
        .expect("reading refs")
        .into_iter()
        .find(|row| row.name == name)
        .and_then(|row| row.oid)
        .map(|raw| {
            gix_hash::oid::try_from_bytes(&raw)
                .expect("a stored oid decodes")
                .to_owned()
        })
}

/// **A downstream box**: stores, an empty repository, and a control plane over
/// them. No listener — a mirror is a client, and what it needs is a place to
/// put what it fetched.
struct Downstream {
    _dir: tempfile::TempDir,
    stores: Arc<ServerStores>,
    plane: CoreControlPlane,
    path: RepoPath,
    repo: gunnar_core::model::Repo,
}

impl Downstream {
    /// Every upstream in this file is on loopback, which a mirror refuses to
    /// dial unless the box says otherwise — so the fixture says otherwise, the
    /// same way an operator would: by creating the file under the data root.
    /// `open_strict` is the same box without it, which is the shipped default.
    fn open(hash: HashKind) -> Downstream {
        let down = Downstream::open_strict(hash);
        std::fs::write(
            down.stores
                .root()
                .join(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
            b"the upstreams in this test are on 127.0.0.1\n",
        )
        .expect("allowing private upstreams on this box");
        down
    }

    fn open_strict(hash: HashKind) -> Downstream {
        let dir = tempfile::tempdir().expect("a temp dir");
        let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
        stores
            .catalog()
            .create_namespace("mirror")
            .expect("namespace");
        let repo_ref = RepoRef::new("mirror", "copy").expect("a legal repo path");
        let repo = stores
            .create_repo(NewRepo::new(repo_ref).hash_kind(hash), false)
            .expect("create the local repository");
        // Materialise it, so the mirror is fetching into a store that exists —
        // the same thing `mirror_add`'s `self.open(path)` guarantees.
        stores.repo_store(&repo).expect("open the local store");
        let plane = CoreControlPlane::new(Arc::clone(&stores));
        Downstream {
            _dir: dir,
            stores,
            plane,
            path: RepoPath::new("mirror", "copy"),
            repo,
        }
    }

    fn git(&self) -> Arc<gunnar_server::storage::OpenRepo> {
        self.stores.repo_store(&self.repo).expect("open the store")
    }

    /// Poll `Mirror.Status` until the job settles, failing rather than hanging.
    ///
    /// A bounded wait around an asynchronous verb, not a fixed sleep: the
    /// assertions after it decide the test and this only turns "it never
    /// finished" into a message. It also **records every progress pair it
    /// sees**, because the pair is only meaningful while the job runs and a
    /// reading taken after the finish would prove nothing.
    fn settle(&self) -> (gunnar_proto::MirrorJob, Vec<(bool, u64, u64)>) {
        let deadline = Instant::now() + Duration::from_secs(120);
        let mut seen: Vec<(bool, u64, u64)> = Vec::new();
        while Instant::now() < deadline {
            let job = self.job();
            let reading = (job.has_progress, job.progress_done, job.progress_total);
            if seen.last() != Some(&reading) {
                seen.push(reading);
            }
            if job.state != MirrorJobState::Queued as i32
                && job.state != MirrorJobState::Running as i32
            {
                return (job, seen);
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        panic!("the mirror sync never settled");
    }

    fn job(&self) -> gunnar_proto::MirrorJob {
        self.plane
            .mirror_status(None, Some(&self.path))
            .expect("mirror_status is implemented")
            .into_iter()
            .next()
            .expect("the registration is there")
            .job
            .expect("a job that was just enqueued is still known")
    }
}

// ════════════════════════════════════════════════════════════════════════════
// The SSH upstream: a whole real gunnar on a loopback russh listener
// ════════════════════════════════════════════════════════════════════════════

/// A running gunnar serving one repository over SSH, with one key registered
/// against it.
#[cfg(feature = "replication")]
struct SshUpstream {
    _dir: tempfile::TempDir,
    runtime: tokio::runtime::Runtime,
    server: Option<gunnar_server::Server>,
    key_pem: String,
    url: String,
    head: Oid,
    blob: Oid,
}

#[cfg(feature = "replication")]
impl SshUpstream {
    fn start(hash: HashKind) -> SshUpstream {
        let dir = tempfile::tempdir().expect("a temp dir");
        let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
        stores
            .catalog()
            .create_namespace("upstream")
            .expect("namespace");
        let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
        let repo = stores
            .catalog()
            .create(
                NewRepo::new(repo_ref.clone())
                    .hash_kind(hash)
                    // **Private**, so the fetch has to authenticate. A public
                    // repository would be served to an anonymous caller and the
                    // whole SSH credential path would go untested while looking
                    // green.
                    .visibility(Visibility::Private),
            )
            .expect("create the upstream repository");
        let store = stores.repo_store(&repo).expect("materialise it");
        let (head, blob) = plant_a_commit(store.serve(), store.hash_kind());

        // A real ed25519 key, minted in process. No `ssh-keygen` — no
        // shelling out.
        let key_pem = String::from_utf8(gunnar_core::mint_ed25519().expect("mint a key"))
            .expect("an OpenSSH private key is utf-8 PEM");
        let public = gunnar_core::ssh_key::PrivateKey::from_openssh(&key_pem)
            .expect("the minted key parses")
            .public_key()
            .to_openssh()
            .expect("render the public half");
        stores
            .auth()
            .add_principal(
                gunnar_core::Principal::new("mirrorer", "mirrorer", PrincipalKind::User).unwrap(),
            )
            .expect("add the principal");
        stores
            .auth()
            .register_key("mirrorer", &public, None)
            .expect("register the key");
        stores
            .auth()
            .grant(Grant::repo("mirrorer", repo_ref, Level::Read))
            .expect("grant read");

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .expect("a tokio runtime");
        let config = gunnar_server::ServeConfig {
            no_grpc: true,
            // The upstream must not serve this repository to a caller with no
            // key at all, or the credential is never exercised.
            allow_anonymous: false,
            ..gunnar_server::ServeConfig::ephemeral()
        };
        let server = runtime
            .block_on(gunnar_server::Server::start(Arc::clone(&stores), config))
            .expect("the upstream server starts");
        let port = server.ssh_addr().port();
        SshUpstream {
            _dir: dir,
            runtime,
            server: Some(server),
            key_pem,
            url: format!("ssh://mirrorer@127.0.0.1:{port}/upstream/origin.git"),
            head,
            blob,
        }
    }
}

#[cfg(feature = "replication")]
impl Drop for SshUpstream {
    fn drop(&mut self) {
        if let Some(server) = self.server.take() {
            let _ = self.runtime.block_on(server.shutdown());
        }
    }
}

// ════════════════════════════════════════════════════════════════════════════
// The smart-HTTP upstream: the two routes, and a 401 when a token is required
// ════════════════════════════════════════════════════════════════════════════

/// A loopback HTTP/1.1 listener serving `git-upload-pack` for one repository.
///
/// **The git half is not written here.** `advertise` and `serve_session` are
/// `gunnar-wire`'s own, called exactly as `gunnar-http` calls them
/// (`Session::StatelessRpc`, so no advertisement is written into a POST
/// response). What this file contributes is the request line, the headers and
/// the `Content-Length` — which is the part the adapter under test has to get
/// right, and the part a fixture can legitimately own.
struct HttpUpstream {
    addr: SocketAddr,
    /// Set when the listener should stop; the accept loop checks it between
    /// connections.
    stop: Arc<AtomicBool>,
    /// Requests that arrived with no acceptable `Authorization` header and were
    /// answered 401. Applied output for "the credential was actually needed".
    refused: Arc<AtomicU64>,
    /// Every `Authorization` header value the listener saw, so a test can
    /// assert the token really did travel — the other half of proving it does
    /// not leak.
    seen_auth: Arc<Mutex<Vec<String>>>,
    /// **The rendezvous that makes the mid-transfer moment observable.**
    ///
    /// When set, the listener writes every byte of a POST response EXCEPT the
    /// last few and then waits for the test to release it. By that point the
    /// client has read the packfile header — which is what publishes
    /// `objects_total` — and is blocked on the rest, so a poller has all the
    /// time it needs to see `(0, N)` while the job is unambiguously RUNNING.
    ///
    /// A sleep would have been the other way to write this and would have been
    /// a flake in disguise: a fetch of three objects over loopback finishes in
    /// about a millisecond, and a poller that happened to miss the window
    /// reports the same red as a progress field that was never published.
    /// (MEASURED: it did, on the first run of this file.)
    ///
    /// Held on the struct as well as moved into the listener thread so the
    /// fixture's own shape says the gate belongs to this upstream; the thread
    /// is what actually waits on it.
    #[allow(dead_code)]
    hold: Option<Arc<AtomicBool>>,
    /// **A dishonest upstream**: `(from, to)`, applied to every response body
    /// this listener writes.
    ///
    /// The two must be the SAME LENGTH, because a pkt-line carries its own
    /// four-hex length and a substitution that changed it would break the
    /// framing rather than test anything. That constraint is what picks the
    /// name in `an_upstream_may_not_name_a_reference_gunnar_would_not_accept`:
    /// `refs/heads/main` and `refs/heads/../x` are both fifteen bytes.
    ///
    /// Rewriting the real `upload_pack` output rather than hand-writing a
    /// protocol answer is deliberate: everything except the one field under
    /// test stays the shipped implementation, so the fetch really does
    /// negotiate, really does receive a pack, and really does reach `apply`.
    #[allow(dead_code)]
    rename: Option<(String, String)>,
}

impl HttpUpstream {
    /// `require` is the exact header value the upstream will accept; `None`
    /// serves anybody.
    fn start(
        stores: Arc<ServerStores>,
        repo: gunnar_core::model::Repo,
        require: Option<String>,
    ) -> HttpUpstream {
        HttpUpstream::start_held(stores, repo, require, None)
    }

    /// The same, answering with `to` everywhere the honest answer said `from`.
    /// See [`HttpUpstream::rename`].
    fn start_renaming(
        stores: Arc<ServerStores>,
        repo: gunnar_core::model::Repo,
        from: &str,
        to: &str,
    ) -> HttpUpstream {
        assert_eq!(
            from.len(),
            to.len(),
            "a pkt-line carries its own length; the substitution must not change it"
        );
        HttpUpstream::start_with(
            stores,
            repo,
            None,
            None,
            Some((from.to_owned(), to.to_owned())),
        )
    }

    /// The same, with the mid-transfer rendezvous armed. See
    /// [`HttpUpstream::hold`].
    fn start_held(
        stores: Arc<ServerStores>,
        repo: gunnar_core::model::Repo,
        require: Option<String>,
        hold: Option<Arc<AtomicBool>>,
    ) -> HttpUpstream {
        HttpUpstream::start_with(stores, repo, require, hold, None)
    }

    fn start_with(
        stores: Arc<ServerStores>,
        repo: gunnar_core::model::Repo,
        require: Option<String>,
        hold: Option<Arc<AtomicBool>>,
        rename: Option<(String, String)>,
    ) -> HttpUpstream {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind a loopback port");
        listener
            .set_nonblocking(false)
            .expect("a blocking listener");
        let addr = listener.local_addr().expect("the bound address");
        let stop = Arc::new(AtomicBool::new(false));
        let refused = Arc::new(AtomicU64::new(0));
        let seen_auth = Arc::new(Mutex::new(Vec::new()));

        let worker = HttpUpstream {
            addr,
            stop: Arc::clone(&stop),
            refused: Arc::clone(&refused),
            seen_auth: Arc::clone(&seen_auth),
            hold: hold.clone(),
            rename: rename.clone(),
        };
        std::thread::spawn(move || {
            for connection in listener.incoming() {
                if stop.load(Ordering::SeqCst) {
                    return;
                }
                let Ok(mut connection) = connection else {
                    return;
                };
                let store = stores.repo_store(&repo).expect("open the upstream store");
                let _ = serve_one_http(
                    &mut connection,
                    store.serve(),
                    store.hash_kind(),
                    require.as_deref(),
                    &refused,
                    &seen_auth,
                    hold.as_deref(),
                    rename.as_ref(),
                );
            }
        });
        worker
    }

    fn base(&self) -> String {
        format!("http://{}/upstream/origin.git", self.addr)
    }
}

impl Drop for HttpUpstream {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::SeqCst);
        // Unblock the accept.
        let _ = std::net::TcpStream::connect(self.addr);
    }
}

/// One HTTP request, answered.
///
/// Eight arguments and an `allow` rather than a struct: every one of them is a
/// fixture knob a single test turns on, and folding them into a options type
/// would put a second shape between a test and the listener it is configuring.
#[allow(clippy::too_many_arguments)]
fn serve_one_http(
    connection: &mut std::net::TcpStream,
    git: &dyn GitServe,
    hash: gix_hash::Kind,
    require: Option<&str>,
    refused: &AtomicU64,
    seen_auth: &Mutex<Vec<String>>,
    hold: Option<&AtomicBool>,
    rename: Option<&(String, String)>,
) -> std::io::Result<()> {
    let mut reader = BufReader::new(connection.try_clone()?);
    let mut line = String::new();
    reader.read_line(&mut line)?;
    let mut parts = line.split_whitespace();
    let method = parts.next().unwrap_or_default().to_owned();
    let target = parts.next().unwrap_or_default().to_owned();

    let mut length = 0usize;
    let mut authorization: Option<String> = None;
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header)? == 0 {
            break;
        }
        let header = header.trim_end();
        if header.is_empty() {
            break;
        }
        // Header NAMES are case-insensitive (RFC 9110 §5.1), and ureq 3 sends them
        // lower-case (`content-length`, `authorization`) where ureq 2 sent them
        // as written: the fixture reads them as a server must (2026-10-03).
        let (name, value) = header.split_once(':').unwrap_or((header, ""));
        if name.eq_ignore_ascii_case("content-length") {
            length = value.trim().parse().unwrap_or(0);
        }
        if name.eq_ignore_ascii_case("authorization") {
            authorization = Some(value.trim().to_owned());
        }
    }
    if let Some(value) = &authorization {
        seen_auth.lock().unwrap().push(value.clone());
    }

    // **The 401 comes before the git seam**, which is what makes the credential
    // load-bearing rather than decorative: without it there is no
    // advertisement to read, not an empty one.
    if let Some(want) = require {
        if authorization.as_deref() != Some(want) {
            refused.fetch_add(1, Ordering::SeqCst);
            connection.write_all(
                b"HTTP/1.1 401 Unauthorized\r\n\
                  WWW-Authenticate: Basic realm=\"gunnar\"\r\n\
                  Content-Length: 0\r\n\
                  Connection: close\r\n\r\n",
            )?;
            return connection.flush();
        }
    }

    let mut body = vec![0u8; length];
    reader.read_exact(&mut body)?;
    // **Only a `command=fetch` response carries a packfile**, and the hold is
    // about the packfile header. Holding the `ls-refs` answer instead would
    // block the client before the fetch had even been asked for — which is a
    // deadlock, and is exactly what the first version of this fixture did.
    let is_fetch = body
        .windows(b"command=fetch".len())
        .any(|window| window == b"command=fetch");

    let upload = gunnar_wire::UploadPack::new(git, hash);
    let mut out = Vec::new();
    let (content_type, outcome) = if method == "GET" && target.contains("/info/refs") {
        (
            "application/x-git-upload-pack-advertisement",
            upload.advertise(gunnar_wire::ProtocolVersion::V2, &mut out),
        )
    } else {
        let mut input = std::io::Cursor::new(body);
        (
            "application/x-git-upload-pack-result",
            upload
                .serve_session(
                    gunnar_wire::ProtocolVersion::V2,
                    gunnar_wire::Session::StatelessRpc,
                    &mut input,
                    &mut out,
                )
                .map(|_| ()),
        )
    };
    if let Err(err) = outcome {
        let body = format!("upload-pack failed: {err}");
        connection.write_all(
            format!(
                "HTTP/1.1 500 Internal Server Error\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            )
            .as_bytes(),
        )?;
        return connection.flush();
    }
    // **The one field a dishonest upstream gets to choose**, substituted in the
    // real answer. Equal lengths, so every pkt-line header stays correct.
    if let Some((from, to)) = rename {
        let (from, to) = (from.as_bytes(), to.as_bytes());
        let mut at = 0;
        while at + from.len() <= out.len() {
            if &out[at..at + from.len()] == from {
                out[at..at + from.len()].copy_from_slice(to);
                at += from.len();
            } else {
                at += 1;
            }
        }
    }
    connection.write_all(
        format!(
            "HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\n\
             Cache-Control: no-cache\r\nConnection: close\r\n\r\n",
            out.len()
        )
        .as_bytes(),
    )?;
    // **Held mid-body when a test asked for it.** The tail is small enough that
    // everything before it — the acknowledgments, the `packfile` section header
    // and the 12-byte pack header the client reads `objects_total` out of — has
    // already gone, and large enough that the client cannot finish without it.
    match (hold, is_fetch && out.len() > 32) {
        (Some(gate), true) => {
            let cut = out.len() - 24;
            connection.write_all(&out[..cut])?;
            connection.flush()?;
            while !gate.load(Ordering::SeqCst) {
                std::thread::yield_now();
            }
            connection.write_all(&out[cut..])?;
        }
        _ => connection.write_all(&out)?,
    }
    connection.flush()
}

/// An upstream repository with content, on its own stores, for the HTTP tests.
fn http_upstream_stores(
    hash: HashKind,
) -> (
    tempfile::TempDir,
    Arc<ServerStores>,
    gunnar_core::model::Repo,
    Oid,
    Oid,
) {
    http_upstream_stores_of(hash, b"the quick brown fox\n".to_vec())
}

/// **An upstream whose pack is wide rather than deep**: one commit, one tree,
/// and `blobs` distinct files under it.
///
/// The shared fixture is three objects, which is the right size for every
/// assertion about *what* was copied and the wrong size for every assertion
/// about the copying being watchable: on a three-object pack a store that
/// published once per pack and a store that published once per object are
/// indistinguishable. Returns the tip and the number of objects the fetched
/// pack will carry — `blobs + 2`, the closure of the one reference.
///
/// The names are zero-padded so the tree's entries are in the order git
/// requires, and the contents are distinct so no two blobs collapse into one
/// object.
fn http_upstream_stores_wide(
    hash: HashKind,
    blobs: usize,
) -> (
    tempfile::TempDir,
    Arc<ServerStores>,
    gunnar_core::model::Repo,
    Oid,
    u64,
) {
    let dir = tempfile::tempdir().expect("a temp dir");
    let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
    stores
        .catalog()
        .create_namespace("upstream")
        .expect("namespace");
    let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
    let repo = stores
        .catalog()
        .create(NewRepo::new(repo_ref).hash_kind(hash))
        .expect("create the upstream repository");
    let store = stores.repo_store(&repo).expect("materialise it");
    let git = store.serve();
    let hash_kind = store.hash_kind();

    let mut pack = PackBuilder::new(hash_kind);
    let mut entries: Vec<(&'static str, String, Oid)> = Vec::with_capacity(blobs);
    for n in 0..blobs {
        let content = format!("file number {n}\n").into_bytes();
        let id = compute(hash_kind, Kind::Blob, &content);
        pack.object(Kind::Blob, content);
        entries.push(("100644", format!("f{n:05}"), id));
    }
    let tree = tree_bytes(
        &entries
            .iter()
            .map(|(mode, name, id)| (*mode, name.as_str(), *id))
            .collect::<Vec<_>>(),
    );
    let tree_id = compute(hash_kind, Kind::Tree, &tree);
    let commit = commit_bytes(tree_id, "wide");
    let commit_id = compute(hash_kind, Kind::Commit, &commit);
    pack.object(Kind::Tree, tree);
    pack.object(Kind::Commit, commit);

    let fixture = pack.build().expect("building the wide pack");
    git.put_pack(&fixture.bytes).expect("storing the pack");
    git.update_ref("refs/heads/main", None, Some(commit_id.as_bytes()))
        .expect("pointing main at the commit");

    (dir, stores, repo, commit_id, blobs as u64 + 2)
}

/// The same, over a caller-chosen blob.
fn http_upstream_stores_of(
    hash: HashKind,
    blob: Vec<u8>,
) -> (
    tempfile::TempDir,
    Arc<ServerStores>,
    gunnar_core::model::Repo,
    Oid,
    Oid,
) {
    let dir = tempfile::tempdir().expect("a temp dir");
    let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
    stores
        .catalog()
        .create_namespace("upstream")
        .expect("namespace");
    let repo_ref = RepoRef::new("upstream", "origin").expect("a legal repo path");
    let repo = stores
        .catalog()
        .create(NewRepo::new(repo_ref).hash_kind(hash))
        .expect("create the upstream repository");
    let store = stores.repo_store(&repo).expect("materialise it");
    let (head, blob_id) = plant_a_commit_of(store.serve(), store.hash_kind(), blob);
    (dir, stores, repo, head, blob_id)
}

// ════════════════════════════════════════════════════════════════════════════
// The assertions every transport shares
// ════════════════════════════════════════════════════════════════════════════

/// **The whole property**, asserted once so the SSH and HTTP tests cannot drift
/// apart about what "mirrored" means.
///
/// It is deliberately not "the job says DONE": a control plane that marked a
/// job DONE without fetching anything would pass that, and every one of these
/// is something it could not fake.
fn assert_mirrored(
    down: &Downstream,
    job: &gunnar_proto::MirrorJob,
    progress: &[(bool, u64, u64)],
    head: Oid,
    blob: Oid,
) {
    assert_eq!(
        job.state,
        MirrorJobState::Done as i32,
        "the sync failed: {}",
        job.error
    );

    // ── the reference moved, to the upstream's own tip ───────────────────────
    let store = down.git();
    assert_eq!(
        tip(store.serve(), "refs/heads/main"),
        Some(head),
        "the mirror's branch is not at the upstream's commit"
    );

    // ── and the OBJECTS came with it ─────────────────────────────────────────
    //
    // A reference pointing at an object the store does not hold is exactly the
    // failure a mirror that applied refs without a pack would produce, and it
    // reads as success at every other seam.
    for (what, oid) in [("the commit", head), ("the blob", blob)] {
        assert!(
            GitOps::extents(store.serve(), &[oid.as_bytes()])
                .expect("asking the store for an object")
                .first()
                .is_some_and(Option::is_some),
            "{what} the reference needs is not in the mirror's store"
        );
    }

    // ── three objects, and the job says so ───────────────────────────────────
    assert_eq!(
        job.objects, 3,
        "the fixture is one commit, one tree and one blob: {job:?}"
    );
    assert!(job.bytes > 0, "a pack of three objects has a size: {job:?}");
    assert_eq!(
        job.updates.len(),
        1,
        "one reference moved, and the job must name it: {:?}",
        job.updates
    );
    assert_eq!(job.updates[0].ref_name, "refs/heads/main");
    assert_eq!(job.updates[0].new_oid, head.to_string());
    assert_eq!(
        job.updates[0].old_oid,
        "0".repeat(head.to_string().len()),
        "a branch that did not exist here before is a CREATE: all-zeroes old"
    );

    // ── the progress pair ends at (N, N) ────────────────────────────────────
    assert_eq!(
        (job.has_progress, job.progress_done, job.progress_total),
        (true, 3, 3),
        "a finished sync reports every object it took: {job:?}"
    );
    // …and **every reading a poller ever saw carried the right denominator**.
    // A transfer of three objects over loopback is about a millisecond, so
    // whether the poller catches the `(0, N)` instant is luck — that claim is
    // made deterministically by
    // `the_progress_pair_moves_from_zero_to_n_while_the_sync_runs`, which holds
    // the upstream mid-body. What every run CAN assert is that the denominator
    // is never invented: a reading of `(true, d, t)` with `t != 3` would be a
    // total that came from somewhere other than the packfile header.
    for reading in progress {
        assert!(
            !reading.0 || (reading.2 == 3 && reading.1 <= 3),
            "a progress reading with a denominator the packfile never declared: \
             {reading:?} in {progress:?}"
        );
    }
}

/// Register the upstream on the downstream and set its credential.
fn register(down: &Downstream, url: &str, kind: MirrorCredentialKind, secret: &str, user: &str) {
    down.plane
        .mirror_add(&down.path, url, false, false)
        .expect("register the mirror");
    if kind != MirrorCredentialKind::Unspecified {
        let config = down
            .plane
            .mirror_set_credential(&down.path, kind, secret, user)
            .expect("store the credential");
        assert_eq!(
            config.credential, kind as i32,
            "the registration must say WHICH kind is stored"
        );
    }
}

// ════════════════════════════════════════════════════════════════════════════
// SSH
// ════════════════════════════════════════════════════════════════════════════

#[cfg(feature = "replication")]
fn mirror_over_ssh(hash: HashKind) {
    let upstream = SshUpstream::start(hash);
    let down = Downstream::open(hash);
    register(
        &down,
        &upstream.url,
        MirrorCredentialKind::SshKey,
        &upstream.key_pem,
        "",
    );

    let accepted = down
        .plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let job = accepted.job.expect("Mirror.Sync answers with a job");
    assert!(
        job.state == MirrorJobState::Queued as i32 || job.state == MirrorJobState::Running as i32,
        "Mirror.Sync must return with the job still live, not after fetching: {job:?}"
    );
    assert!(job.job_id.starts_with("mirror-"), "{job:?}");

    let (done, progress) = down.settle();
    assert_eq!(
        done.job_id, job.job_id,
        "the finished job must be the one that was accepted"
    );
    assert_mirrored(&down, &done, &progress, upstream.head, upstream.blob);
}

/// **A box that says so refuses an upstream host key nobody put there.**
///
/// The mirror's SSH dial was hardcoded trust-on-first-use — the policy was a
/// literal in `dial_ssh` and there was no way to say otherwise — so the first
/// sync trusted whatever answered on the upstream's address, on the one path
/// that is about to offer that address a private key. Replication's own
/// targets have carried a `strict_host_key` knob all along, and its comment
/// says on "is the right setting for anything that outlives an afternoon".
///
/// Every other assertion in this file's SSH half is the TOFU case, which still
/// works: `a_mirror_sync_over_ssh_copies_a_sha256_repository` runs against the
/// same upstream with no `mirror_strict_host_key` file and mirrors it. So this
/// is a rule the box chooses, not a mirror that stopped working.
///
/// MEASURED before the knob existed: the sync reached DONE against an upstream
/// whose host key had never been seen, and `mirror_known_hosts` was created by
/// the fetch itself.
#[cfg(feature = "replication")]
#[test]
fn a_box_with_strict_host_keys_refuses_an_upstream_it_has_never_seen() {
    let upstream = SshUpstream::start(HashKind::Sha256);
    let down = Downstream::open(HashKind::Sha256);
    let known_hosts = down
        .stores
        .root()
        .join(gunnar_server::mirror_fetch::MIRROR_KNOWN_HOSTS);
    assert!(
        !known_hosts.exists(),
        "the fixture must start with nothing pinned, or strictness proves nothing"
    );
    std::fs::write(
        down.stores
            .root()
            .join(gunnar_server::mirror_fetch::MIRROR_STRICT_HOST_KEY),
        b"this box does not learn host keys from the network\n",
    )
    .expect("asking this box to refuse an unknown host key");

    register(
        &down,
        &upstream.url,
        MirrorCredentialKind::SshKey,
        &upstream.key_pem,
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();

    assert_eq!(
        job.state,
        MirrorJobState::Failed as i32,
        "a box that refuses unknown host keys mirrored from one anyway: {job:?}"
    );
    assert!(
        job.error.contains("host key") || job.error.contains("known_hosts"),
        "the refusal must name what it refused: {}",
        job.error
    );
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        None,
        "nothing was fetched"
    );
    assert!(
        !known_hosts.exists(),
        "a strict dial must LEARN NOTHING; `{}` was written by the fetch",
        known_hosts.display()
    );
}

/// **sha256**, the kind gunnar creates by default.
#[cfg(feature = "replication")]
#[test]
fn a_mirror_sync_over_ssh_copies_a_sha256_repository() {
    mirror_over_ssh(HashKind::Sha256);
}

/// **sha1**, and it is not a formality: `fetch_pack` refuses an `object-format`
/// mismatch by name, so a client that quietly ran at one width against a remote
/// of the other would negotiate an EMPTY answer and report success.
#[cfg(feature = "replication")]
#[test]
fn a_mirror_sync_over_ssh_copies_a_sha1_repository() {
    mirror_over_ssh(HashKind::Sha1);
}

// ════════════════════════════════════════════════════════════════════════════
// Smart HTTP
// ════════════════════════════════════════════════════════════════════════════

fn mirror_over_http(hash: HashKind) {
    let (_dir, stores, repo, head, blob) = http_upstream_stores(hash);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(hash);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );

    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (done, progress) = down.settle();
    assert_mirrored(&down, &done, &progress, head, blob);
}

/// **The adapter, against the same `upload_pack` gunnar-http drives.**
#[test]
fn a_mirror_sync_over_smart_http_copies_a_sha256_repository() {
    mirror_over_http(HashKind::Sha256);
}

#[test]
fn a_mirror_sync_over_smart_http_copies_a_sha1_repository() {
    mirror_over_http(HashKind::Sha1);
}

/// **A private upstream: 401 without the token, served with it, and the token
/// appears in nothing anybody can read.**
#[test]
fn a_private_http_upstream_needs_the_token_and_never_shows_it() {
    const TOKEN: &str = "ghp_MirrorSyncNeverLeaksThis";
    let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(
        Arc::clone(&stores),
        repo.clone(),
        Some(format!("Bearer {TOKEN}")),
    );
    let down = Downstream::open(HashKind::Sha256);

    // ── RED HALF: with no credential the upstream refuses, and the job FAILS ──
    //
    // Without this, the green half below would pass against a mirror that
    // ignored the credential entirely and against an upstream that never
    // checked it.
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane.mirror_sync(&down.path).expect("accepted");
    let (refused, _) = down.settle();
    assert_eq!(
        refused.state,
        MirrorJobState::Failed as i32,
        "a private upstream must refuse an anonymous mirror, not serve it an empty answer"
    );
    assert!(
        refused.error.contains("401"),
        "the failure must name what the upstream said: {}",
        refused.error
    );
    assert!(
        upstream.refused.load(Ordering::SeqCst) >= 1,
        "the upstream never actually refused anything, so this proves nothing"
    );
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        None,
        "a refused sync must move no reference"
    );

    // ── GREEN HALF: the same mirror, with the token ──────────────────────────
    down.plane
        .mirror_set_credential(&down.path, MirrorCredentialKind::Token, TOKEN, "")
        .expect("store the token");
    down.plane.mirror_sync(&down.path).expect("accepted");
    let (done, progress) = down.settle();
    assert_mirrored(&down, &done, &progress, head, blob);

    // The token really did travel — otherwise the "never shows it" assertions
    // below would be about a token nothing ever sent.
    let seen = upstream.seen_auth.lock().unwrap().clone();
    assert!(
        seen.iter().any(|value| value == &format!("Bearer {TOKEN}")),
        "the upstream never received the token: {seen:?}"
    );

    // ── AND IT IS IN NONE OF THE BYTES ANYBODY CAN READ ─────────────────────
    //
    // Every surface a caller or an operator reaches: the status response, the
    // finished job, the failed job from the red half, and the registration.
    let mut surfaces: Vec<String> = vec![
        format!(
            "{:?}",
            down.plane.mirror_status(None, Some(&down.path)).unwrap()
        ),
        format!("{done:?}"),
        format!("{refused:?}"),
        format!("{:?}", down.plane.mirror_status(None, None).unwrap()),
    ];
    // …and the one RPC that CARRIES it, read back: `SetCredential`'s own
    // response. If any RPC were going to hand a secret back it would be this
    // one.
    surfaces.push(format!(
        "{:?}",
        down.plane
            .mirror_set_credential(&down.path, MirrorCredentialKind::Token, TOKEN, "")
            .unwrap()
    ));
    for surface in &surfaces {
        assert!(
            !surface.contains(TOKEN),
            "the token appears in something a caller can read:\n{surface}"
        );
    }

    // …and forgetting it is a real forget, not a flag flip: the next sync is
    // refused again.
    down.plane
        .mirror_set_credential(&down.path, MirrorCredentialKind::Unspecified, "", "")
        .expect("forget the token");
    down.plane.mirror_sync(&down.path).expect("accepted");
    let (after, _) = down.settle();
    assert_eq!(
        after.state,
        MirrorJobState::Failed as i32,
        "a forgotten credential must actually be gone: {after:?}"
    );
}

/// **`(0, N)` is readable WHILE the fetch is still running, and only then does
/// it become `(N, N)`.**
///
/// The claim the queue exists to make: a fetch of a large repository is minutes
/// of network with a console watching, and a job that sat at RUNNING and then
/// jumped to DONE would tell that console nothing. So the denominator has to be
/// published at the *start* of the transfer — which is what
/// `FetchProgress::objects_total` is for, fired the instant the packfile header
/// arrives — and not computed at the end when nobody needs it.
///
/// Asserted against an upstream **held mid-body**, not against a poll that
/// happened to be quick: a three-object fetch over loopback finishes in about a
/// millisecond, and a timing-dependent version of this test is a flake that
/// reads exactly like the bug.
#[test]
fn the_progress_pair_moves_from_zero_to_n_while_the_sync_runs() {
    // **A pack that does not fit in one sideband packet.** The hold below cuts
    // the response short, and a pkt-line is only readable once it is COMPLETE:
    // with a three-hundred-byte pack the whole transfer is one packet, so
    // withholding its tail blocks the client BEFORE the packfile header instead
    // of after it, and nothing is ever published. (MEASURED: that is exactly
    // what the first version of this test did — a 60-second timeout with
    // `has_progress: false`.) 200 KiB of incompressible noise is three sideband
    // packets, so the header is in the part that has already gone.
    let (_dir, stores, repo, head, blob) =
        http_upstream_stores_of(HashKind::Sha256, noise(200 * 1024));
    let release = Arc::new(AtomicBool::new(false));
    let upstream = HttpUpstream::start_held(stores, repo, None, Some(Arc::clone(&release)));
    let down = Downstream::open(HashKind::Sha256);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );

    down.plane.mirror_sync(&down.path).expect("accepted");

    // The upstream is holding the tail of the pack, so this cannot pass by
    // being fast.
    let deadline = Instant::now() + Duration::from_secs(60);
    loop {
        assert!(
            Instant::now() < deadline,
            "the sync never published a denominator: {:?}",
            down.job()
        );
        let job = down.job();
        if job.has_progress {
            assert_eq!(
                (job.progress_done, job.progress_total),
                (0, 3),
                "the FIRST reading must be (0, N): the total is known from the packfile \
                 header, the numerator is not known until the ingest returns"
            );
            assert_eq!(
                job.state,
                MirrorJobState::Running as i32,
                "progress must be readable WHILE the job runs, which is the only time \
                 anybody needs it: {job:?}"
            );
            break;
        }
        assert_ne!(
            job.state,
            MirrorJobState::Done as i32,
            "the job finished without ever publishing a denominator: {job:?}"
        );
        std::thread::yield_now();
    }

    release.store(true, Ordering::SeqCst);
    let (done, progress) = down.settle();
    assert_mirrored(&down, &done, &progress, head, blob);
}

/// **The numerator MOVES**: more than two distinct `done` values, monotone,
/// ending at `N`.
///
/// # 🔴 The red this was written against
///
/// Until 2026-09-08 this test recorded exactly two readings for a 302-object
/// pack:
///
/// ```text
/// the count has 2 distinct value(s) over a pack of 302 objects: [0, 302]
/// ```
///
/// `(0, N)` from `FetchProgress::objects_total` when the packfile header
/// arrived, `(N, N)` when the ingest returned, and nothing in between — which
/// is what `mirror_queue`'s module doc said in as many words and what a console
/// rendered as a bar that jumped from empty to full. Everything else about the
/// sync was already green: the reference moved, the objects landed, the job
/// reported the right total. Storing a pack correctly and *saying how far you
/// have got while you store it* are different claims, and only one of them was
/// being made.
///
/// It records 304 readings now — `0` from the header, `1..=302` one per object
/// as the engine's pack walk splits them, and `302` again from `finish` —
/// 303 distinct values where there were two.
///
/// # Why the series is recorded at a sink and not polled off the job
///
/// [`Reporter::progress`] overwrites — deliberately; its own doc says a
/// progress that accumulated would be a second counter to keep right — so the
/// job record holds the latest reading and never the sequence. Polling it would
/// make this test a race between a sampler thread and a pack walk that takes
/// microseconds, and a flaky version of this test reads exactly like the bug it
/// is for. `sync_once_watching` hands the sink the same values the `Reporter`
/// is given, in the same order, on the same thread, so what is recorded here is
/// what the job saw.
///
/// # 302 objects, and the real path under them
///
/// The whole production path runs: the real `MirrorQueue` worker, the real
/// `mirror_fetch`, the real `PushIngest`, the real storage engine, a real
/// upload-pack on the other end of a loopback socket. The only thing this test
/// supplies is a second listener.
#[test]
fn the_object_count_moves_while_the_pack_lands() {
    /// Every publish, in order.
    struct Series(Mutex<Vec<(u64, u64)>>);

    impl gunnar_store::git_storage_trait::PackProgress for Series {
        fn indexed(&self, done: u64, total: u64) {
            self.0
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push((done, total));
        }
    }

    let (_dir, up_stores, up_repo, head, objects) =
        http_upstream_stores_wide(HashKind::Sha256, 300);
    let upstream = HttpUpstream::start(up_stores, up_repo, None);
    let down = Downstream::open(HashKind::Sha256);

    let series = Arc::new(Series(Mutex::new(Vec::new())));
    let sink = Arc::clone(&series);
    let stores = Arc::clone(&down.stores);
    // The real queue, so the `Reporter` the sink shadows is a real one, taken
    // from a real job by the real worker.
    let queue = gunnar_server::mirror_queue::MirrorQueue::new(Arc::new(move |order, report| {
        gunnar_server::mirror_fetch::sync_once_watching(
            &stores,
            order,
            &gunnar_server::mirror_fetch::Credential::None,
            report,
            Some(&*sink),
        )
    }));
    let order = gunnar_server::mirror_queue::MirrorOrder {
        path: "mirror/copy".to_owned(),
        upstream: upstream.base(),
        prune: false,
    };
    queue.enqueue(order);

    let deadline = Instant::now() + Duration::from_secs(120);
    let job = loop {
        assert!(Instant::now() < deadline, "the mirror sync never settled");
        let job = queue.status("mirror/copy").expect("the job is still known");
        if job.state != gunnar_server::job_queue::JobState::Queued
            && job.state != gunnar_server::job_queue::JobState::Running
        {
            break job;
        }
        std::thread::sleep(Duration::from_millis(5));
    };
    queue.stop();
    assert_eq!(
        job.state,
        gunnar_server::job_queue::JobState::Done,
        "the sync failed: {:?}",
        job.error
    );

    // The mirror is real before the series is judged: a sync that published a
    // beautiful count and copied nothing must fail here, not there.
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        Some(head),
        "the mirror's branch is not at the upstream's commit"
    );
    assert_eq!(job.objects, objects, "the pack's own object count");

    let seen = series.0.lock().unwrap_or_else(|p| p.into_inner()).clone();
    let dones: Vec<u64> = seen.iter().map(|(done, _)| *done).collect();
    assert!(!seen.is_empty(), "the ingest published nothing at all");
    for (done, total) in &seen {
        assert_eq!(
            *total, objects,
            "the denominator moved mid-transfer: a reading declared {total} for a pack \
             of {objects}"
        );
        assert!(
            *done <= objects,
            "a reading claims {done} of {objects} objects are in"
        );
    }
    assert!(
        seen.windows(2).all(|w| w[0].0 <= w[1].0),
        "the count went backwards: {dones:?}"
    );
    assert_eq!(
        seen.last().map(|(done, _)| *done),
        Some(objects),
        "the last reading is not ({objects}, {objects}); a bar that stops short of the \
         end reads exactly like a transfer that hung"
    );
    let distinct = {
        let mut values = dones.clone();
        values.dedup();
        values.len()
    };
    assert!(
        distinct > 2,
        "the count has {distinct} distinct value(s) over a pack of {objects} objects: \
         {:?}. Two is `(0, N)` and `(N, N)` — the shape this test exists to make \
         unshippable.",
        if dones.len() > 8 {
            &dones[..8]
        } else {
            &dones[..]
        }
    );

    // …and the job carries the last of them, so what a console polls is the
    // same number the sink recorded and not a second bookkeeping.
    assert_eq!(
        job.progress,
        Some((objects, objects)),
        "the job's own progress pair must end where the series does"
    );
}

// ════════════════════════════════════════════════════════════════════════════
// What an upstream may call a reference
// ════════════════════════════════════════════════════════════════════════════

/// **A reference name from an upstream is untrusted input, and goes through the
/// same rules a pushed name does.**
///
/// The upstream here is the real `upload_pack`, serving a real repository, with
/// `refs/heads/main` rewritten to `refs/heads/../x` in every response body —
/// fifteen bytes for fifteen, so the pkt-line framing is untouched and the
/// negotiation, the pack and the ingest are all the shipped ones. Only the name
/// is a lie, which is exactly the shape a hostile or broken upstream has.
///
/// `fetch_pack::mirror` is the only filter such a name met before this: a
/// prefix test, which `refs/heads/../x` passes.
///
/// MEASURED before the check existed: the fetch succeeded, the job reached
/// DONE, and `refs/heads/../x` was in this repository's ref listing.
#[test]
fn an_upstream_may_not_name_a_reference_gunnar_would_not_accept() {
    let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start_renaming(stores, repo, "refs/heads/main", "refs/heads/../x");
    let down = Downstream::open(HashKind::Sha256);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );

    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();

    assert_eq!(
        job.state,
        MirrorJobState::Failed as i32,
        "a name with `..` in it was accepted from an upstream: {job:?}"
    );
    assert!(
        job.error.contains("will not write") && job.error.contains(".."),
        "the refusal must say which rule the name broke: {}",
        job.error
    );
    let held = GitOps::refs(down.git().serve()).expect("reading refs");
    assert!(
        held.iter().all(|row| !row.name.contains("..")),
        "a reference the upstream named is in this store: {:?}",
        held.iter().map(|row| &row.name).collect::<Vec<_>>()
    );
}

// ════════════════════════════════════════════════════════════════════════════
// Where a mirror may be pointed
// ════════════════════════════════════════════════════════════════════════════

/// **`Mirror.Add` refuses a destination only the server can reach**, on the
/// call that made it rather than four minutes later on a job.
///
/// The URL here is the loopback one the rest of this file mirrors from happily
/// — the difference is the box, not the address: this `Downstream` is the
/// shipped default, with no `mirror_allow_private_destinations` file. That is
/// the whole finding: `Mirror.Add` needs Admin, every principal is
/// intrinsically Admin in its own namespace, and the only check on the URL was
/// that it was not empty.
///
/// MEASURED before the guard existed: `mirror_add` returned `Ok` and the
/// registration was stored, for `http://169.254.169.254/latest/meta-data/` as
/// readily as for anything else.
#[test]
fn a_mirror_may_not_be_pointed_at_an_address_only_the_server_can_reach() {
    let down = Downstream::open_strict(HashKind::Sha256);
    for url in [
        "http://127.0.0.1:9418/ns/repo.git",
        "http://169.254.169.254/latest/meta-data/",
        "https://10.0.0.5/ns/repo.git",
        "ssh://git@192.168.1.1/ns/repo.git",
        "ssh://git@[::1]:2222/ns/repo.git",
        "git@10.1.2.3:ns/repo.git",
    ] {
        let err = down
            .plane
            .mirror_add(&down.path, url, false, false)
            .expect_err("a destination the caller cannot reach on their own");
        match err {
            gunnar_grpc::seam::ControlError::InvalidArgument(why) => assert!(
                why.contains("network position")
                    && why.contains(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
                "the refusal must name the address and the way to allow it: {why}"
            ),
            other => panic!("{url} must be INVALID_ARGUMENT, not {other:?}"),
        }
        assert!(
            down.plane
                .mirror_status(None, Some(&down.path))
                .expect("status")
                .is_empty(),
            "{url} was refused and still written down"
        );
    }

    // …and a public destination is still accepted, so the guard is a rule and
    // not a wall. Nothing is dialled by `Mirror.Add`, so this makes no request.
    down.plane
        .mirror_add(
            &down.path,
            "https://93.184.216.34/ns/repo.git",
            false,
            false,
        )
        .expect("a routable destination registers");
}

// ════════════════════════════════════════════════════════════════════════════
// Branch protection, over the batch a mirror applies
// ════════════════════════════════════════════════════════════════════════════

/// Put a protection rule on `refs/heads/main`, through the verb an operator
/// uses. `Policy.SetProtection` needs **Admin**; `Mirror.Sync` needs **Write**,
/// which is the whole asymmetry these two tests are about.
fn protect_main(down: &Downstream, allowed: Vec<String>) {
    down.plane
        .set_protection(
            &down.path,
            gunnar_proto::BranchProtection {
                ref_pattern: "refs/heads/main".to_owned(),
                allow_force_push: false,
                allow_delete: false,
                allowed_principal_ids: allowed,
                ..Default::default()
            },
        )
        .expect("the protection rule is stored");
}

/// **A mirror sync may not force-push over a protected branch.**
///
/// The downstream's `refs/heads/main` is at a commit the upstream has never
/// heard of, so the upstream's tip is a genuine non-fast-forward onto it — the
/// update a push would get `non-fast-forward` for, from the rule that is
/// configured here. Every assertion is on applied output: the reference the
/// store holds afterwards, which is what a bypass would have moved.
///
/// MEASURED before the gate existed: the job reached DONE and
/// `refs/heads/main` held the upstream's commit, with `Policy.GetProtection`
/// still reporting `allow_force_push: false` on the same reference.
#[test]
fn a_mirror_sync_does_not_force_push_over_a_protected_branch() {
    let (_dir, stores, repo, upstream_head, _blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);

    // A local history the upstream does not share: same shape, different blob,
    // therefore a different commit with no ancestry in common.
    let store = down.git();
    let (local_head, _) = plant_a_commit_of(
        store.serve(),
        store.hash_kind(),
        b"a history of our own\n".to_vec(),
    );
    assert_ne!(local_head, upstream_head, "the fixture must diverge");
    protect_main(&down, Vec::new());

    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();

    assert_eq!(
        job.state,
        MirrorJobState::Failed as i32,
        "a mirror that force-pushes over branch protection is the third independent way past \
         the gate; the job must FAIL: {job:?}"
    );
    assert!(
        job.error.contains("non-fast-forward") && job.error.contains("refs/heads/main"),
        "the refusal must name what it refused and why: {}",
        job.error
    );
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        Some(local_head),
        "nothing was moved: the protected reference still holds the commit it held"
    );
}

/// **A rule that names its principals refuses the sync**, because the caller's
/// identity does not cross the control-plane seam on this verb and a gate that
/// cannot decide must not allow.
///
/// The upstream's tip here would be an ordinary CREATE — nothing about it is a
/// force-push — so the only thing that can refuse it is the undecidable
/// `allowed_principals`, and the message has to say so rather than blame
/// ancestry.
#[test]
fn a_rule_that_names_its_principals_refuses_a_mirror_sync() {
    let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    protect_main(&down, vec!["alice".to_owned()]);

    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();

    assert_eq!(
        job.state,
        MirrorJobState::Failed as i32,
        "a rule this path cannot evaluate must refuse, not skip: {job:?}"
    );
    assert!(
        job.error.contains("carries no principal"),
        "the refusal must say WHY it could not decide, not invent an ancestry reason: {}",
        job.error
    );
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        None,
        "nothing was moved"
    );
}

/// **The gate is not a blanket refusal.** The same protected repository, with
/// the same rule, mirrors normally when the upstream's tip is a fast-forward —
/// which is what keeps the two tests above from being satisfied by a mirror
/// that simply stopped working.
#[test]
fn a_protected_repository_still_mirrors_a_fast_forward() {
    let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    protect_main(&down, Vec::new());

    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (done, progress) = down.settle();
    assert_mirrored(&down, &done, &progress, head, blob);
}

// ════════════════════════════════════════════════════════════════════════════
// The queue's own contract, at the door
// ════════════════════════════════════════════════════════════════════════════

/// **Two clicks are one job**, asserted on the id and on the *applied output* —
/// how many objects the upstream was actually asked for. A queue that
/// deduplicated its bookkeeping while running the fetch twice would pass an
/// id comparison.
#[test]
fn two_syncs_of_one_repository_are_one_job() {
    let (_dir, stores, repo, head, blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );

    let first = down.plane.mirror_sync(&down.path).unwrap().job.unwrap();
    let second = down.plane.mirror_sync(&down.path).unwrap().job.unwrap();
    assert_eq!(
        first.job_id, second.job_id,
        "an impatient second click must join the job that is already there, not start another"
    );

    let (done, progress) = down.settle();
    assert_eq!(done.job_id, first.job_id);
    assert_mirrored(&down, &done, &progress, head, blob);
}

/// A sync of a repository nobody registered is `NOT_FOUND` **by name**, and it
/// is refused on the request that made it rather than becoming a job that fails
/// later. Those are two different things to tell an operator and only one of
/// them is their mistake.
#[test]
fn a_sync_of_an_unregistered_repository_is_not_found() {
    let down = Downstream::open(HashKind::Sha256);
    let err = down
        .plane
        .mirror_sync(&down.path)
        .expect_err("nothing was registered");
    match err {
        gunnar_grpc::seam::ControlError::NotFound(why) => assert!(
            why.contains("no mirror registered for mirror/copy"),
            "the refusal must name the repository: {why}"
        ),
        other => panic!("an unregistered mirror must be NOT_FOUND, not {other:?}"),
    }
}

/// **`Mirror.Remove` takes the sealed credential with it.**
///
/// The registration was removed and the secret was not: the ssh private key or
/// forge token stayed in the sealed store for the life of the box, with nothing
/// on the control plane naming it or able to name it. No RPC lists secrets and
/// no RPC reads one back, so it was un-listable, un-rotatable, and still
/// decryptable by anything that reaches the KEK — while the operator who
/// removed the mirror because the credential leaked had been told it was gone.
///
/// Asserted on the sealed store itself, which is the only place the fact lives.
///
/// MEASURED before the fix: `SecretStore::get("mirror/mirror/copy")` returned
/// the token's bytes after `Mirror.Remove` returned `Ok`.
#[test]
fn removing_a_mirror_deletes_its_sealed_credential() {
    const TOKEN: &str = "ghp_ThisMustNotOutliveTheRegistration";
    let down = Downstream::open(HashKind::Sha256);
    register(
        &down,
        "https://93.184.216.34/ns/repo.git",
        MirrorCredentialKind::Token,
        TOKEN,
        "",
    );

    let repo_ref = RepoRef::new("mirror", "copy").expect("a legal repo path");
    let key = gunnar_core::credential_secret_key(&repo_ref);
    let secrets = down.stores.secrets().expect("the sealed store");
    assert_eq!(
        gunnar_store::SecretStore::get(&secrets, &key).expect("reading the sealed store"),
        Some(TOKEN.as_bytes().to_vec()),
        "the fixture must actually have sealed something, or the assertion below is vacuous"
    );

    down.plane
        .mirror_remove(&down.path)
        .expect("remove the mirror");

    assert_eq!(
        gunnar_store::SecretStore::get(&secrets, &key).expect("reading the sealed store"),
        None,
        "the credential outlived the registration that was the only thing naming it"
    );
    assert!(
        down.plane
            .mirror_status(None, Some(&down.path))
            .expect("status")
            .is_empty(),
        "and the registration is gone"
    );
}

/// **The wrong kind of credential is refused, by name, before anything is
/// dialled.**
///
/// The failure this prevents is the silent one: an SSH key against an
/// `https://` remote cannot ride in an HTTP request, so the fetch would go out
/// anonymous and come back 401 — against an operator who can see a credential
/// is stored and will therefore rotate a key that was never the problem.
#[test]
fn a_credential_of_the_wrong_kind_is_refused_by_name() {
    let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    let key = String::from_utf8(gunnar_core::mint_ed25519().expect("mint a key"))
        .expect("an OpenSSH private key is utf-8 PEM");
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::SshKey,
        &key,
        "",
    );

    down.plane.mirror_sync(&down.path).expect("accepted");
    let (job, _) = down.settle();
    assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
    assert!(
        job.error.contains("smart http") && job.error.contains("ssh private key"),
        "the refusal must name BOTH the transport and the credential it cannot carry: {}",
        job.error
    );
    assert!(
        !job.error.contains("PRIVATE KEY"),
        "the refusal quoted the key: {}",
        job.error
    );
}

/// **A credential value with no kind is refused**, because the alternative is
/// that "set this token" and "forget my key" are the same request read two
/// ways — and one of those readings silently destroys a stored credential.
#[test]
fn a_credential_write_that_does_not_say_which_way_it_means_is_refused() {
    let down = Downstream::open(HashKind::Sha256);
    down.plane
        .mirror_add(&down.path, "https://example.test/ns/repo.git", false, false)
        .expect("register");

    let err = down
        .plane
        .mirror_set_credential(&down.path, MirrorCredentialKind::Unspecified, "a-token", "")
        .expect_err("a value with no kind");
    assert!(
        matches!(err, gunnar_grpc::seam::ControlError::InvalidArgument(_)),
        "{err:?}"
    );

    let err = down
        .plane
        .mirror_set_credential(&down.path, MirrorCredentialKind::Token, "", "")
        .expect_err("a kind with no value");
    assert!(
        matches!(err, gunnar_grpc::seam::ControlError::InvalidArgument(_)),
        "{err:?}"
    );
}

// ════════════════════════════════════════════════════════════════════════════
// A mirror sync counts against the namespace's room (2026-10-02)
// ════════════════════════════════════════════════════════════════════════════

/// The refusal a capped gate composes — the plan gate's own sentence, the one
/// a push past the same ceiling is told.
const CAP_WORDS: &str = "plan team would pass its pack quota with this push (cap 1 bytes): the \
                         transfer was stopped and nothing was stored";

/// A gate that holds the downstream's namespace to a fixed ceiling, or says it
/// is already full.
struct Capped {
    ceiling: u64,
    full: bool,
}

impl gunnar_ssh::PlanGate for Capped {
    fn name(&self) -> &'static str {
        "capped-for-the-test"
    }
    fn verdict(&self, _: &str, _: gunnar_ssh::PlanAction) -> gunnar_ssh::PlanVerdict {
        gunnar_ssh::PlanVerdict::Allow
    }
    fn pack_quota(&self, namespace: &str) -> gunnar_ssh::PlanVerdict {
        if self.full && namespace == "mirror" {
            gunnar_ssh::PlanVerdict::Refuse(CAP_WORDS.to_owned())
        } else {
            gunnar_ssh::PlanVerdict::Allow
        }
    }
    fn pack_ceiling(&self, namespace: &str) -> Option<gunnar_ssh::PackCeiling> {
        (namespace == "mirror").then(|| gunnar_ssh::PackCeiling {
            bytes: self.ceiling,
            refusal: CAP_WORDS.to_owned(),
        })
    }
}

fn capped_sync(ceiling: u64, full: bool) -> (Downstream, gunnar_proto::MirrorJob, Oid) {
    let (_dir, stores, repo, head, _blob) =
        http_upstream_stores_of(HashKind::Sha256, noise(512 * 1024));
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    // The same chain the front doors hold a push to — installed where a
    // mirror's worker reads it.
    down.stores.install_plan_gates(vec![
        Arc::new(Capped { ceiling, full }) as Arc<dyn gunnar_ssh::PlanGate>
    ]);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();
    (down, job, head)
}

/// ★ **A sync bigger than the namespace's room stops cleanly at the cap.** The
/// fetch is cut mid-pack at the ceiling, the job fails with the gate's own
/// sentence, and the mirror holds nothing of it — no reference, no object.
/// Before this, a mirror fetch was the one write on the box no ceiling held:
/// an import landed its whole upstream however far past the cap it went.
#[test]
fn a_mirror_sync_past_the_namespaces_room_stops_at_the_cap_and_lands_nothing() {
    let ceiling = 64 * 1024;
    let (down, job, head) = capped_sync(ceiling, false);
    assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
    assert!(
        job.error.contains(CAP_WORDS),
        "the job says {:?}",
        job.error
    );
    assert_eq!(
        job.refused_reason,
        gunnar_proto::MirrorRefusal::PackCap as i32,
        "the cap stop is a FIELD, not only gunnar's prose: {job:?}"
    );
    assert!(
        job.bytes <= ceiling + 64 * 1024,
        "the fetch took {} bytes off the wire past a {ceiling}-byte ceiling",
        job.bytes
    );
    let store = down.git();
    assert_eq!(
        tip(store.serve(), "refs/heads/main"),
        None,
        "a cut sync moved a ref"
    );
    assert!(
        GitOps::extents(store.serve(), &[head.as_bytes()])
            .expect("asking the store")
            .first()
            .is_none_or(Option::is_none),
        "a cut sync stored the commit"
    );
}

/// A namespace already at its cap fetches nothing at all — refused before the
/// dial, in the gate's words.
#[test]
fn a_mirror_sync_into_a_full_namespace_fetches_nothing() {
    let (down, job, _) = capped_sync(u64::MAX, true);
    assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
    assert!(job.error.contains(CAP_WORDS), "{:?}", job.error);
    assert_eq!(
        job.refused_reason,
        gunnar_proto::MirrorRefusal::PackCap as i32,
        "{job:?}"
    );
    assert_eq!(job.bytes, 0, "a full namespace took bytes off the wire");
    assert_eq!(tip(down.git().serve(), "refs/heads/main"), None);
}

/// A ceiling the upstream fits under changes nothing.
#[test]
fn a_mirror_sync_under_the_ceiling_lands_whole() {
    let (down, job, head) = capped_sync(64 * 1024 * 1024, false);
    assert_eq!(job.state, MirrorJobState::Done as i32, "{}", job.error);
    assert_eq!(
        job.refused_reason,
        gunnar_proto::MirrorRefusal::Unspecified as i32
    );
    assert_eq!(tip(down.git().serve(), "refs/heads/main"), Some(head));
}

/// A gate whose plan refuses every write — a suspended namespace.
struct Suspended;

const SUSPENDED_WORDS: &str = "plan team is suspended: writes are refused until it is paid";

impl gunnar_ssh::PlanGate for Suspended {
    fn name(&self) -> &'static str {
        "suspended-for-the-test"
    }
    fn verdict(&self, namespace: &str, _: gunnar_ssh::PlanAction) -> gunnar_ssh::PlanVerdict {
        if namespace == "mirror" {
            gunnar_ssh::PlanVerdict::Refuse(SUSPENDED_WORDS.to_owned())
        } else {
            gunnar_ssh::PlanVerdict::Allow
        }
    }
    fn pack_quota(&self, _: &str) -> gunnar_ssh::PlanVerdict {
        gunnar_ssh::PlanVerdict::Allow
    }
    fn pack_ceiling(&self, _: &str) -> Option<gunnar_ssh::PackCeiling> {
        None
    }
}

/// ★ **A suspended namespace's sync is refused by its plan, and the job says
/// so as a field** (2026-10-04): `MIRROR_REFUSAL_PLAN`, beside the gate's own
/// sentence — not the cap, and not a fetch that failed.
#[test]
fn a_mirror_sync_into_a_suspended_namespace_is_refused_by_its_plan() {
    let (_dir, stores, repo, _head, _blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    down.stores
        .install_plan_gates(vec![Arc::new(Suspended) as Arc<dyn gunnar_ssh::PlanGate>]);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .mirror_sync(&down.path)
        .expect("the sync is accepted");
    let (job, _) = down.settle();
    assert_eq!(job.state, MirrorJobState::Failed as i32, "{job:?}");
    assert!(job.error.contains(SUSPENDED_WORDS), "{:?}", job.error);
    assert_eq!(
        job.refused_reason,
        gunnar_proto::MirrorRefusal::Plan as i32,
        "{job:?}"
    );
    assert_eq!(job.bytes, 0);
}

// ════════════════════════════════════════════════════════════════════════════
// Mirror.Add never takes over a repository that exists (2026-10-04)
// ════════════════════════════════════════════════════════════════════════════

/// `Mirror.Add` through the real gRPC service, as the account's owner.
fn add_as_owner(
    down: &Downstream,
    url: &str,
    create_repo: bool,
    adopt: bool,
) -> Result<gunnar_proto::AddMirrorResponse, tonic::Status> {
    use gunnar_proto::mirror_server::Mirror as _;
    let _ = down.stores.auth().add_principal(gunnar_core::Principal {
        id: "mirror".to_owned(),
        name: "mirror".to_owned(),
        kind: gunnar_core::PrincipalKind::User,
        created: 0,
    });
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(CoreControlPlane::new(Arc::clone(&down.stores)));
    let service = gunnar_grpc::MirrorService::new(gunnar_grpc::ServiceContext::new(
        plane,
        Arc::new(gunnar_grpc::EventBus::new(16)),
        gunnar_grpc::ServerConfig::default(),
    ));
    let mut request = tonic::Request::new(gunnar_proto::AddMirrorRequest {
        repo: Some(down.path.clone()),
        remote_url: url.to_owned(),
        prune: true,
        create_repo,
        adopt,
    });
    request.extensions_mut().insert(gunnar_grpc::Caller {
        principal_id: "mirror".to_owned(),
        fingerprint: "SHA256:owner".to_owned(),
        machine: false,
        authenticated: true,
    });
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime")
        .block_on(service.add(request))
        .map(tonic::Response::into_inner)
}

/// ★ **An import whose name collides does not turn somebody's repository into
/// a mirror.** `Mirror.Add{create_repo}` over an existing repository that is
/// not a mirror is refused by name and nothing is registered — a pruning
/// sync would otherwise have moved its branches to the upstream's. `adopt`
/// says it on purpose; re-adding an existing MIRROR stays a no-op.
#[test]
fn a_mirror_that_would_create_its_repository_does_not_take_over_one_that_exists() {
    let down = Downstream::open(HashKind::Sha256);
    let url = "https://127.0.0.1:9/upstream.git";

    let refused = add_as_owner(&down, url, true, false)
        .expect_err("an existing, non-mirror repository is not taken over");
    assert_eq!(
        refused.code(),
        tonic::Code::FailedPrecondition,
        "{refused:?}"
    );
    assert!(
        refused.message().contains("mirror/copy"),
        "names it: {refused:?}"
    );
    assert!(
        down.plane
            .mirror_status(None, Some(&down.path))
            .expect("mirror_status")
            .is_empty(),
        "a refused add registered the mirror anyway"
    );

    add_as_owner(&down, url, true, true).expect("adopt takes it over on purpose");
    add_as_owner(&down, url, true, false).expect("re-adding an existing mirror is unchanged");
    assert_eq!(
        down.plane
            .mirror_status(None, Some(&down.path))
            .expect("mirror_status")
            .len(),
        1
    );
}

// ════════════════════════════════════════════════════════════════════════════
// A mirror that creates its repository takes the upstream's object format
// ════════════════════════════════════════════════════════════════════════════

/// The downstream box with NO repository yet: `Mirror.Add { create_repo }`
/// registers the import, and its first sync makes the repository at the
/// upstream's format (2026-10-18) — which is what every console import does.
/// Returns the box, the finished first sync and the progress it showed.
type Imported = (
    Downstream,
    Oid,
    Oid,
    gunnar_proto::MirrorJob,
    Vec<(bool, u64, u64)>,
);

fn import_creating(hash_of_upstream: HashKind) -> Imported {
    let (_dir, stores, repo, head, blob) = http_upstream_stores(hash_of_upstream);
    // Leaked on purpose for the test's life: the upstream must outlive the sync.
    let upstream = Box::leak(Box::new(HttpUpstream::start(stores, repo, None)));
    let dir = tempfile::tempdir().expect("a temp dir");
    let stores = Arc::new(ServerStores::open(dir.path().join("data")).expect("stores"));
    stores
        .catalog()
        .create_namespace("mirror")
        .expect("namespace");
    std::fs::write(
        stores
            .root()
            .join(gunnar_server::mirror_fetch::MIRROR_ALLOW_PRIVATE),
        b"the upstreams in this test are on 127.0.0.1\n",
    )
    .expect("allowing private upstreams on this box");
    let plane = CoreControlPlane::new(Arc::clone(&stores));
    let path = RepoPath::new("mirror", "copy");
    let repo_ref = RepoRef::new("mirror", "copy").expect("a legal path");
    plane
        .mirror_add(&path, &upstream.base(), false, true)
        .expect("register the import");
    assert!(
        stores
            .catalog()
            .get(&repo_ref)
            .expect("the catalog answers")
            .is_none(),
        "Mirror.Add made the repository before the upstream said its format"
    );
    plane.mirror_sync(&path).expect("the sync is accepted");
    let deadline = Instant::now() + Duration::from_secs(120);
    let mut seen: Vec<(bool, u64, u64)> = Vec::new();
    let job = loop {
        assert!(Instant::now() < deadline, "the import never settled");
        let job = plane
            .mirror_status(None, Some(&path))
            .expect("mirror_status")
            .into_iter()
            .next()
            .and_then(|m| m.job)
            .expect("the job");
        let reading = (job.has_progress, job.progress_done, job.progress_total);
        if seen.last() != Some(&reading) {
            seen.push(reading);
        }
        if job.state != MirrorJobState::Queued as i32 && job.state != MirrorJobState::Running as i32
        {
            break job;
        }
        std::thread::sleep(Duration::from_millis(5));
    };
    let repo = stores
        .catalog()
        .get(&repo_ref)
        .expect("the catalog answers")
        .unwrap_or_else(|| panic!("the first sync made no repository: {}", job.error));
    let down = Downstream {
        _dir: dir,
        stores,
        plane,
        path,
        repo,
    };
    (down, head, blob, job, seen)
}

/// ★ **A SHA-1 upstream — every repository on GitHub — mirrors into a
/// repository the import creates.** Before 2026-10-02 every such repository
/// was created at sha256 whatever the remote was, so the first sync of any
/// sha1 upstream failed with "this repository is object-format=sha256, but the
/// remote advertised sha1" — measured in the console's uitest, importing a
/// stock-git repository served by `git http-backend`.
#[test]
fn a_mirror_that_creates_its_repository_takes_a_sha1_upstreams_format() {
    let (down, head, blob, done, progress) = import_creating(HashKind::Sha1);
    assert_eq!(
        down.repo.hash_kind,
        HashKind::Sha1,
        "the repository was created at the wrong format"
    );
    assert_mirrored(&down, &done, &progress, head, blob);
}

/// …and a sha256 upstream still lands at sha256.
#[test]
fn a_mirror_that_creates_its_repository_takes_a_sha256_upstreams_format() {
    let (down, head, blob, done, progress) = import_creating(HashKind::Sha256);
    assert_eq!(down.repo.hash_kind, HashKind::Sha256);
    assert_mirrored(&down, &done, &progress, head, blob);
}

// ════════════════════════════════════════════════════════════════════════════
// An encrypted repository is never fetched into
// ════════════════════════════════════════════════════════════════════════════

/// The Pack frames the transaction log holds (what the twin would replay).
fn pack_frames(stores: &ServerStores) -> usize {
    let mut n = 0;
    stores
        .txlog()
        .scan_meta(|_, kind, _| {
            if gunnar_core::txlog::Kind::from_u16(kind) == Some(gunnar_core::txlog::Kind::Pack) {
                n += 1;
            }
        })
        .expect("scan the transaction log");
    n
}

/// **A mirror whose repository is encrypted lands NOTHING** (wiring tracer B
/// 2026-10-08). A mirror's pack is an upstream's plaintext and `PushIngest`
/// runs no encrypted gate on it, so the worker refuses the sync before the
/// dial: the job FAILS saying so, no object is stored, no Pack frame reaches
/// the twin, the reference does not move. The door refuses making such a
/// mirror (`Mirror.Add`, `Repos.SetEncrypted`); this is the registration that
/// predates the rule, or a flag set below the door.
#[test]
fn a_mirror_never_lands_plaintext_in_an_encrypted_repository() {
    let (_dir, stores, repo, _head, blob) = http_upstream_stores(HashKind::Sha256);
    let upstream = HttpUpstream::start(stores, repo, None);
    let down = Downstream::open(HashKind::Sha256);
    register(
        &down,
        &upstream.base(),
        MirrorCredentialKind::Unspecified,
        "",
        "",
    );
    down.plane
        .set_encrypted(&down.path, true)
        .expect("the flag, set below the door");
    let frames_before = pack_frames(&down.stores);

    match down.plane.mirror_sync(&down.path) {
        // Refused at enqueue: nothing ran.
        Err(err) => assert!(format!("{err}").contains("encrypted"), "{err}"),
        // Or enqueued and refused by the worker, before the dial.
        Ok(_) => {
            let (job, _) = down.settle();
            assert_eq!(
                job.state,
                MirrorJobState::Failed as i32,
                "the sync must FAIL: {job:?}"
            );
            assert!(
                job.error.contains("encrypted"),
                "the job says why: {}",
                job.error
            );
        }
    }
    assert_eq!(
        tip(down.git().serve(), "refs/heads/main"),
        None,
        "no reference moved"
    );
    assert!(
        !GitOps::extents(down.git().serve(), &[blob.as_bytes()])
            .expect("asking the store")
            .first()
            .is_some_and(Option::is_some),
        "the upstream's plaintext blob is in the encrypted repository's store"
    );
    assert_eq!(
        pack_frames(&down.stores),
        frames_before,
        "a pack was framed for the twin"
    );
}