gunnar-server 3.0.0

//! The twin over a scripted primary, in one process: the gap, a restart
//! mid-range, a torn tail, a hole in retention, and the restore that is the
//! only proof that counts. Every row is on the test matrix.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use gunnar_core::txlog::{Entry, Log, Meta};
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_store::conformance::pack::PackBuilder;

use gix_object::Kind as ObjKind;
use gunnar_store::SecretStore as _;

use super::*;
use crate::stores::ServerStores;

const REPO: &str = "ns/app";
const SECRET_KEY: &str = "twin-test/sealed";
const LFS_OID: &str = "4e07408562bedb8b60ce05c1decfe3ad16b72230967de01f640b7e4729b49fce";

/// What the primary was told, so the comparison names things.
struct Facts {
    oids: Vec<Vec<u8>>,
    doc_names: Vec<String>,
    blob_ids: Vec<BlobId>,
    lfs_bytes: Vec<u8>,
}

/// **The second namespace, so "namespaces come back" can fail.** A box with one namespace
/// cannot tell a replayed catalog from a catalog the restore happened to recreate.
const NS2: &str = "team";
/// The group the grant hangs off (punkt 9): a principal of its own, a membership edge, and a
/// namespace grant — three different shapes in the identity table, none of them a user row.
const GROUP: &str = "reviewers";
/// The branch `ns/app` is moved to, so a default branch that was NOT replayed reads as `main`.
const DEFAULT_BRANCH: &str = "refs/heads/trunk";

/// A fresh ed25519 public-key line, for a key row that is a real parse rather than a string.
fn a_public_key_line() -> String {
    let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
        .unwrap();
    key.public_key().to_openssh().unwrap()
}

/// **Wait until this box's packs are indexed, because a snapshot needs the whole graph.**
///
/// `snapshot::produce` walks reachability, and reachability refuses while a pack is still being
/// resolved: *"index in progress for this repository: pack 1 0% (0/1 objects, resolve) —
/// reachability needs the whole graph; retry when the index row reaches 100 %"*. That refusal is
/// CORRECT — a real `Mirror.SnapshotFetch` arriving mid-resolve gets it and retries — but a test
/// that pushes a pack and produces a snapshot on the next line was reading a race: on an idle box
/// the resolver had always finished first.
///
/// MEASURED 2026-09-16, under five concurrent lanes and a KVM run: two twin tests failed here in
/// one suite run and in no other, and the message was this refusal, not an assertion. A flake in
/// the twin's most important path is worse than the bug it hides, so the test waits for the
/// condition it actually depends on instead of for the machine to be quiet.
///
/// Bounded, and it says what it was waiting for rather than hanging.
fn wait_until_indexed(stores: &ServerStores, what: &str) {
    // 🔴 The progress rows below only see a pack once its drain has STARTED it: a pack still
    // queued for the background drain reads as `active: false, 0/0`, i.e. "finished". cov13 lost
    // that race at load 25 on dwarf (2026-10-16): this returned, and the snapshot fetch then met
    // "index in progress … pack 5 0% (0/0 objects, walk)". So the queue is emptied first through
    // the engine's own `index_now` (it takes the absorb gate, so it waits for a drain already
    // running and absorbs what is still pending); after it nothing is queued and the rows are
    // the truth.
    for row in stores.catalog().list(None).expect("the catalog") {
        let repo = stores
            .catalog()
            .get(&RepoRef::new(&row.namespace, &row.name).expect("a catalogued name"))
            .expect("the catalog")
            .expect("listed, so catalogued");
        stores
            .repo_store(&repo)
            .expect("its store")
            .serve()
            .index_now()
            .expect("index the pending packs");
    }
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
    loop {
        let unfinished: Vec<String> = stores
            .engine()
            .index_progress()
            .into_iter()
            .filter(|(root, _)| root.starts_with(stores.root()))
            .filter(|(_, p)| p.active || p.objects_done < p.objects_total)
            .map(|(root, p)| format!("{} {}/{}", root.display(), p.objects_done, p.objects_total))
            .collect();
        if unfinished.is_empty() {
            return;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "{what}: the index never reached 100 % in 60 s, still resolving: {unfinished:?}"
        );
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
}

fn one_blob_pack(hash: gix_hash::Kind, content: &[u8]) -> (Vec<u8>, Vec<u8>) {
    let id = gix_object::compute_hash(hash, ObjKind::Blob, content).expect("hash");
    let mut pb = PackBuilder::new(hash);
    pb.object(ObjKind::Blob, content.to_vec());
    let fixture = pb.build().expect("pack");
    (fixture.bytes, id.as_bytes().to_vec())
}

/// A primary that was told a little of everything: two packs, a ref set, a
/// lost CAS, a plain ref batch, HEAD, a principal (blob + doc), a secret,
/// an LFS object.
fn primary(dir: &std::path::Path) -> (Arc<ServerStores>, Facts) {
    let stores = Arc::new(ServerStores::open(dir).expect("primary stores"));
    stores.catalog().create_namespace("ns").expect("namespace");
    let repo = stores
        .create_repo(
            NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
            false,
        )
        .expect("repo");
    let store = stores.repo_store(&repo).expect("store");
    let git = store.serve();
    let hash = store.hash_kind();
    let (p1, o1) = one_blob_pack(hash, b"first\n");
    let (p2, o2) = one_blob_pack(hash, &vec![b'x'; 3 << 20]);
    git.put_pack(&p1).expect("pack 1");
    git.update_ref("refs/heads/main", None, Some(&o1))
        .expect("main");
    // The lost race: the same create again. Logged with its claim, refused
    // by the store — and refused the same way on the twin.
    assert!(git.update_ref("refs/heads/main", None, Some(&o2)).is_err());
    git.put_pack(&p2).expect("pack 2");
    git.put_refs(&[gunnar_store::git_storage_trait::RefUpdate::set(
        "refs/tags/v1",
        hex::encode(&o2),
    )])
    .expect("tag");
    // A second branch, so the default branch this box carries is NOT the one `create_repo`
    // hands out: a `default_branch` check against `refs/heads/main` could not fail.
    git.put_refs(&[gunnar_store::git_storage_trait::RefUpdate::set(
        DEFAULT_BRANCH,
        hex::encode(&o1),
    )])
    .expect("trunk");
    git.set_head(DEFAULT_BRANCH).expect("HEAD");

    stores
        .auth()
        .add_principal(Principal::new("alice", "Alice", PrincipalKind::User).unwrap())
        .expect("principal");
    stores
        .auth()
        .register_key("alice", &a_public_key_line(), None)
        .expect("alice's key");
    stores
        .secrets()
        .expect("secrets")
        .put(SECRET_KEY, b"the sealed thing")
        .expect("secret");

    // ── the configuration a box IS, beyond its objects (lane TWIN-COMPLETENESS) ──
    //
    // Every one of these was replicated before this fixture wrote it; none of them was
    // COMPARED, because the fixture never made one. A restore that lost the issuer, the host
    // key, a group, a grant, a namespace, a visibility or a default branch passed
    // `assert_same` — the test could not fail for the things its name promised.

    // The SSH host key: a restored box that mints a new one is a box every client refuses
    // with a changed-fingerprint warning.
    gunnar_core::ensure_host_key(&stores.secrets().expect("secrets")).expect("host key");
    // The certificate issuer: the private key is a secret (M5), the certificate a control
    // document (M4). A restored box that mints a NEW issuer is a box every pinned client
    // certificate no longer chains to.
    stores
        .pki()
        .expect("pki")
        .create("gunnar-twin-test-ca", 365)
        .expect("issuer");
    // A second namespace, a group principal, a membership edge and a namespace grant.
    stores
        .catalog()
        .create_namespace(NS2)
        .expect("second namespace");
    stores
        .auth()
        .add_principal(Principal::new(GROUP, "Reviewers", PrincipalKind::Group).unwrap())
        .expect("group");
    stores
        .auth()
        .add_to_group(GROUP, "alice")
        .expect("membership");
    stores
        .auth()
        .grant(gunnar_core::Grant::namespace(
            GROUP,
            NS2,
            gunnar_core::Level::Write,
        ))
        .expect("grant");
    // The repository row's own settings.
    stores
        .catalog()
        .set_visibility(
            &RepoRef::new("ns", "app").unwrap(),
            gunnar_core::Visibility::Public,
        )
        .expect("visibility");
    // Both halves, as `ControlPlane::set_default_branch` does them: the catalog row is the
    // logged document, and `HEAD` is a projection of it every box derives on open (see
    // `ServerStores::point_head` — deliberately UNLOGGED). A fixture that moved only the row
    // left the primary's HEAD at `main` and every twin's at `trunk`, which is the derivation
    // working, not a defect.
    let moved = stores
        .catalog()
        .set_default_branch(&RepoRef::new("ns", "app").unwrap(), DEFAULT_BRANCH)
        .expect("default branch");
    stores
        .repoint_head(&moved)
        .expect("HEAD follows the default branch");
    // The entitlement FACT per namespace: a control document of its own, and the thing T1d
    // in the KVM drill could not make the keyless medium produce.
    #[cfg(feature = "metered")]
    {
        let ents = gunnar_core::EntitlementStore::new(Arc::clone(stores.control_store()))
            .expect("entitlements");
        ents.set(gunnar_core::Entitlement {
            namespace: NS2.into(),
            plan: "10gb".into(),
            state: gunnar_core::EntitlementState::Paid,
            paid_until_unix_ms: Some(1_790_812_800_000),
            caps: std::collections::BTreeMap::from([("pack_bytes".to_string(), 10u64 << 30)]),
            source: "payment:invoice:twin-drill".into(),
            signature: vec![7; 64],
            issued_unix_ms: Some(1_789_000_000_000),
            issued_signature: vec![8; 64],
            accepted_at_unix_ms: 1,
        })
        .expect("entitlement fact");
    }

    // A signed-log line (an entitlement fact rides here) and a replaced
    // document: Append and File frames through the sink. **Every** framed file, not one:
    // `FRAMED_FILES` is the set a snapshot carries and a restore must reproduce, and a file
    // the fixture never wrote is a file the comparison cannot miss.
    for rel in gunnar_core::txlog::FRAMED_FILES {
        // A framed file may live in a SUBDIRECTORY since lane T11 added the audit trail
        // (`log/audit-chain.jsonl`); the appender does not make one.
        if let Some(parent) = dir.join(rel).parent() {
            std::fs::create_dir_all(parent).unwrap_or_else(|e| panic!("{rel}: {e}"));
        }
        // The audit trail is not a free-form JSONL log: its lines are hash-chained entries and
        // `AuditChain::open` refuses anything else. So the fixture writes a REAL decision
        // through the real appender — which is also the only way this file's other tests can
        // compare a restored trail rather than a blob.
        if *rel == gunnar_core::txlog::AUDIT_REL {
            let chain = gunnar_core::AuditChain::open_framed(dir.join(rel))
                .unwrap_or_else(|e| panic!("{rel}: {e}"));
            chain
                .append(
                    gunnar_core::AuditEvent {
                        method: "Access.Grant",
                        principal: "alice",
                        fingerprint: "SHA256:fixture",
                        scope: "server",
                        level: "admin",
                        auth_mode: "required",
                        decision: "allow",
                        reason: "",
                    },
                    1,
                )
                .unwrap_or_else(|e| panic!("{rel}: {e}"));
            continue;
        }
        if rel.ends_with(".jsonl") {
            crate::attest::AttestationLog::at(dir.join(rel))
                .append_line(&format!("{{\"log\":\"{rel}\",\"entitlement\":{{\"tenant\":\"alice\",\"plan\":\"team\"}}}}"))
                .unwrap_or_else(|e| panic!("{rel}: {e}"));
        } else {
            crate::durable::write_atomic(dir, rel, format!("{{\"doc\":\"{rel}\"}}").as_bytes())
                .unwrap_or_else(|e| panic!("{rel}: {e}"));
        }
    }

    let lfs_bytes = vec![7u8; 2 << 20];
    {
        use gunnar_lfs::LfsStore as _;
        use std::io::Write as _;
        let lfs = crate::txlog_lfs::LoggedLfs::new(
            gunnar_lfs::FsLfsStore::open_or_init(store.git_dir()).unwrap(),
            Arc::clone(stores.txlog()),
            REPO,
        );
        let oid = gunnar_lfs::Oid::parse(LFS_OID).unwrap();
        let mut w = lfs.begin_write(&oid, lfs_bytes.len() as u64).unwrap();
        w.write_all(&lfs_bytes).unwrap();
        w.commit().unwrap();
    }

    let log = stores.txlog();
    let entries = log.range(log.first_seq(), log.head_seq()).unwrap();
    let mut doc_names = Vec::new();
    let mut blob_ids = Vec::new();
    for e in &entries {
        match e.meta() {
            Ok(Meta::ControlDoc { name, .. }) => doc_names.push(name),
            Ok(Meta::ControlBlob { id }) => blob_ids.push(BlobId::from_hex(&id).unwrap()),
            _ => {}
        }
    }
    doc_names.sort();
    doc_names.dedup();
    let facts = Facts {
        oids: vec![o1, o2],
        doc_names,
        blob_ids,
        lfs_bytes,
    };
    (stores, facts)
}

fn refs_of(stores: &ServerStores) -> Vec<(String, Option<Vec<u8>>, Option<String>)> {
    let repo = stores
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .expect("catalogued");
    let store = stores.repo_store(&repo).unwrap();
    let mut rows: Vec<_> = store
        .serve()
        .refs()
        .unwrap()
        .into_iter()
        .map(|r| (r.name, r.oid, r.symref_target))
        .collect();
    rows.sort();
    rows
}

/// Byte for byte, name for name: what a restore has to reproduce.
fn assert_same(primary: &ServerStores, other: &ServerStores, facts: &Facts, who: &str) {
    assert_eq!(refs_of(primary), refs_of(other), "{who}: refs");
    let head_p = primary
        .repo_store(
            &primary
                .catalog()
                .get(&RepoRef::new("ns", "app").unwrap())
                .unwrap()
                .unwrap(),
        )
        .unwrap();
    let head_o = other
        .repo_store(
            &other
                .catalog()
                .get(&RepoRef::new("ns", "app").unwrap())
                .unwrap()
                .unwrap(),
        )
        .unwrap();
    assert_eq!(
        head_p.serve().head().unwrap().map(|r| r.symref_target),
        head_o.serve().head().unwrap().map(|r| r.symref_target),
        "{who}: HEAD"
    );
    for oid in &facts.oids {
        assert!(
            head_o.serve().has(oid).unwrap(),
            "{who}: object {} missing",
            hex::encode(oid)
        );
        assert_eq!(
            head_p.serve().read(oid).unwrap(),
            head_o.serve().read(oid).unwrap(),
            "{who}: object {} differs",
            hex::encode(oid)
        );
    }
    for name in &facts.doc_names {
        assert_eq!(
            primary.raw_control_store().read_doc(name).unwrap(),
            other.raw_control_store().read_doc(name).unwrap(),
            "{who}: document {name}"
        );
    }
    for id in &facts.blob_ids {
        assert_eq!(
            primary.raw_control_store().get_blob(id).unwrap(),
            other.raw_control_store().get_blob(id).unwrap(),
            "{who}: blob {}",
            hex::encode(id.as_bytes())
        );
    }
    assert_eq!(
        primary.raw_secrets().get(SECRET_KEY).unwrap(),
        other.raw_secrets().get(SECRET_KEY).unwrap(),
        "{who}: secret"
    );
    assert!(
        other.raw_secrets().get(SECRET_KEY).unwrap().is_some(),
        "{who}: the secret is there"
    );
    {
        use gunnar_lfs::LfsStore as _;
        let lfs = gunnar_lfs::FsLfsStore::open_or_init(head_o.git_dir()).unwrap();
        let oid = gunnar_lfs::Oid::parse(LFS_OID).unwrap();
        assert_eq!(
            lfs.size_of(&oid).unwrap(),
            Some(facts.lfs_bytes.len() as u64),
            "{who}: lfs"
        );
    }
    for name in ROOT_FILES {
        let a = std::fs::read(primary.root().join(name)).unwrap_or_default();
        let b = std::fs::read(other.root().join(name)).unwrap_or_default();
        assert!(!a.is_empty(), "{who}: the primary has {name}");
        assert_eq!(a, b, "{who}: {name}");
    }
    assert_same_configuration(primary, other, who);
    // The twin's log IS the primary's, number for number — nothing the
    // apply path did wrote a frame of its own.
    assert_eq!(
        primary.txlog().head_seq(),
        other.txlog().head_seq(),
        "{who}: log head"
    );
}

/// **What makes the box ITSELF**, class by class, each failing on its own name (lane
/// TWIN-COMPLETENESS). The owner's question was not "are the repositories back" — it was
/// "is it the same box". Objects, refs, LFS and the documents-by-frame-name are
/// [`assert_same`]'s half; this is the other one, and it is the half nothing checked.
///
/// Every assertion here is reached through a READ-ONLY path on `other`
/// ([`ServerStores::secrets_unframed`], [`ServerStores::pki_readonly`]), never
/// `ensure_host_key` or `Pki::create`: a check that MINTS what it is looking for cannot fail
/// for its absence, which is the shape of test this lane was told to stop writing.
fn assert_same_configuration(primary: &ServerStores, other: &ServerStores, who: &str) {
    // **The SEALED bytes, not the plaintext**, and that is not laziness: `assert_same` is
    // also called on a box in the middle of a restore, which holds the fresh box's KEK in
    // memory and the replayed one on disk — a sealed READ there opens nothing, by design
    // ([`restore_finish`]: "the box restarts"). Every caller of this compares two boxes
    // whose sealed bytes are the same generation. The plaintext, after the re-seal has moved
    // it under a KEK of the restored box's own, is
    // [`a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories`]'s job.
    let read_secret = |s: &ServerStores, key: &str| -> Option<Vec<u8>> {
        s.raw_secrets().get(key).expect("secret read")
    };

    // ── the SSH host key (M5) ───────────────────────────────────────────────
    let hk_p =
        read_secret(primary, gunnar_store::SSH_HOST_KEY).expect("the primary has a host key");
    assert_eq!(
        read_secret(other, gunnar_store::SSH_HOST_KEY).as_deref(),
        Some(&hk_p[..]),
        "{who}: the SSH host key. A box that comes back with a different one is a box every \
         client refuses with a changed-fingerprint warning"
    );

    // ── the certificate issuer: the key (M5) and the certificate (M4) ───────
    assert!(
        read_secret(primary, gunnar_store::ISSUER_PRIVATE_KEY).is_some(),
        "{who}: the primary has an issuer key"
    );
    assert_eq!(
        read_secret(other, gunnar_store::ISSUER_PRIVATE_KEY),
        read_secret(primary, gunnar_store::ISSUER_PRIVATE_KEY),
        "{who}: the issuer PRIVATE key"
    );
    let issuer_p = primary
        .pki_readonly()
        .expect("pki")
        .issuer()
        .expect("issuer doc")
        .expect("the primary has an issuer");
    let issuer_o = other
        .pki_readonly()
        .expect("pki")
        .issuer()
        .expect("issuer doc");
    let issuer_o = issuer_o.unwrap_or_else(|| {
        panic!("{who}: no issuer at all — every pinned client certificate is now unchainable")
    });
    assert_eq!(
        issuer_o.cert_pem, issuer_p.cert_pem,
        "{who}: the issuer CERTIFICATE"
    );
    assert_eq!(
        issuer_o.fingerprint, issuer_p.fingerprint,
        "{who}: the issuer fingerprint"
    );
    assert_eq!(
        other.pki_readonly().unwrap().trusted_cas().unwrap().len(),
        primary.pki_readonly().unwrap().trusted_cas().unwrap().len(),
        "{who}: the trusted-CA list"
    );

    // ── the identity table beyond the user rows (M4) ────────────────────────
    let principals = |s: &ServerStores| {
        let mut rows: Vec<_> = s
            .auth()
            .principals()
            .unwrap()
            .into_iter()
            .map(|p| (p.id, p.kind))
            .collect();
        rows.sort();
        rows
    };
    assert_eq!(
        principals(other),
        principals(primary),
        "{who}: principals (users, machines AND groups)"
    );
    let keys = |s: &ServerStores| {
        let mut rows: Vec<_> = s
            .auth()
            .keys()
            .unwrap()
            .into_iter()
            .map(|k| (k.fingerprint_sha256, k.principal_id))
            .collect();
        rows.sort();
        rows
    };
    assert_eq!(keys(other), keys(primary), "{who}: public keys");
    assert!(
        !keys(primary).is_empty(),
        "{who}: the primary holds a key at all"
    );
    let grants = |s: &ServerStores| {
        let mut rows: Vec<_> = s
            .auth()
            .grants(None)
            .unwrap()
            .into_iter()
            .map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
            .collect();
        rows.sort();
        rows
    };
    assert_eq!(grants(other), grants(primary), "{who}: grants");
    assert!(
        grants(primary).iter().any(|g| g.starts_with(GROUP)),
        "{who}: the primary holds the group's grant"
    );
    assert_eq!(
        other.auth().groups_of("alice").unwrap(),
        primary.auth().groups_of("alice").unwrap(),
        "{who}: group membership"
    );
    assert_eq!(
        primary.auth().groups_of("alice").unwrap(),
        vec![GROUP.to_string()],
        "{who}: the primary's membership edge"
    );

    // ── the catalog: namespaces, repository rows, visibility, default branch (M4) ──
    let namespaces = |s: &ServerStores| {
        let mut rows: Vec<String> = s
            .catalog()
            .namespaces()
            .unwrap()
            .into_iter()
            .map(|n| n.name)
            .collect();
        rows.sort();
        rows
    };
    assert_eq!(namespaces(other), namespaces(primary), "{who}: namespaces");
    assert!(
        namespaces(primary).iter().any(|n| n == NS2),
        "{who}: the primary holds the second namespace"
    );
    let repos = |s: &ServerStores| {
        let mut rows: Vec<String> = s
            .catalog()
            .list(None)
            .unwrap()
            .into_iter()
            .map(|r| {
                format!(
                    "{}/{} vis={:?} head={} hash={:?}",
                    r.namespace, r.name, r.visibility, r.default_branch, r.hash_kind
                )
            })
            .collect();
        rows.sort();
        rows
    };
    assert_eq!(
        repos(other),
        repos(primary),
        "{who}: catalog rows (visibility, default branch, hash kind)"
    );
    assert!(
        repos(primary)
            .iter()
            .any(|r| r.contains("vis=Public") && r.contains(&format!("head={DEFAULT_BRANCH}"))),
        "{who}: the primary's row carries the visibility and default branch this asserts"
    );

    // ── the entitlement facts, per namespace (M4) ───────────────────────────
    #[cfg(feature = "metered")]
    {
        let facts = |s: &ServerStores| {
            gunnar_core::EntitlementStore::new(Arc::clone(s.raw_control_store()))
                .unwrap()
                .all()
                .unwrap()
                .into_iter()
                .map(|f| {
                    format!(
                        "{}={}@{} caps={:?} sig={}",
                        f.namespace,
                        f.state,
                        f.plan,
                        f.caps,
                        hex::encode(&f.signature)
                    )
                })
                .collect::<Vec<_>>()
        };
        assert_eq!(facts(other), facts(primary), "{who}: entitlement facts");
        assert!(
            !facts(primary).is_empty(),
            "{who}: the primary holds an entitlement fact at all"
        );
    }

    // ── every framed file, byte for byte (M7, M8) ───────────────────────────
    for rel in gunnar_core::txlog::FRAMED_FILES {
        let a = std::fs::read(primary.root().join(rel)).ok();
        let b = std::fs::read(other.root().join(rel)).ok();
        assert!(a.is_some(), "{who}: the primary has {rel}");
        assert_eq!(b, a, "{who}: {rel}");
    }

    // ── the whole control database, not just the names that were framed ─────
    let (ea, eb) = (
        primary.control_db().export().unwrap(),
        other.control_db().export().unwrap(),
    );
    assert_eq!(eb.docs, ea.docs, "{who}: every control document pointer");
    let live = |e: &gunnar_core::control::ControlExport| {
        e.secrets
            .iter()
            .filter(|(_, v)| v.is_some())
            .map(|(k, _)| k.clone())
            .collect::<Vec<_>>()
    };
    assert_eq!(live(&eb), live(&ea), "{who}: which secrets exist");
}

/// A twin's stores: `twin.json` first, so the box opens follower-only and
/// frames nothing of its own.
fn open_twin(dir: &std::path::Path) -> Arc<ServerStores> {
    TwinConfig {
        primary: "primary.test:50051".into(),
        joined_at_unix_ms: 0,
    }
    .save(dir)
    .unwrap();
    Arc::new(ServerStores::open(dir).unwrap())
}

// ── the store-only twin (OWNER RULING 2026-09-22) ───────────────────────────

/// **Recharge a new main from `twin`**: a fresh data root, the twin's kept base (when it has
/// one) and then its log, through the same `restore_from` `gunnar control twin replay` drives.
/// A store-only twin's data is readable ONLY this way — so every "the twin holds X" check reads
/// the box recharged from it.
fn recharged(twin: &Arc<ServerStores>) -> (tempfile::TempDir, Arc<ServerStores>) {
    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let head = twin.txlog().head_seq();
    assert_eq!(
        restore_from(
            Box::new(LocalSource {
                stores: Arc::clone(twin),
                retained_from: 0
            }),
            Arc::clone(&fresh),
            ""
        )
        .unwrap(),
        head,
        "the recharge reaches the twin's head"
    );
    (fd, fresh)
}

/// Every file under a data root that is a repository store or anything derived from one:
/// the archive (`objects.pack`), its oid tail (`objects.tail`), the commit graph
/// (`objects.graph`), the explode pool (`objects.exploded*`), a pack index (`*.idx`), an LFS
/// object — all of it lives under `repos/`.
fn derived_on(root: &std::path::Path) -> Vec<String> {
    files_under(root)
        .into_iter()
        .filter(|p| {
            // The log's own segment index is the log, the one store a twin keeps.
            !p.starts_with("txlog/")
                && (p.starts_with(&format!("{}/", crate::stores::REPOS_DIR))
                    || p.contains("objects.")
                    || p.ends_with(".idx")
                    || p.contains("/lfs/"))
        })
        .collect()
}

/// **The twin only stores**: no repository store and nothing derived from one on its data
/// root, and its log is the primary's, frame for frame.
fn assert_store_only(primary: &ServerStores, twin: &ServerStores, who: &str) {
    let derived = derived_on(twin.root());
    assert!(
        derived.is_empty(),
        "{who}: a store-only twin built {} derived file(s): {derived:?}",
        derived.len()
    );
    let (p, t) = (primary.txlog(), twin.txlog());
    assert_eq!(
        t.head_seq(),
        p.head_seq(),
        "{who}: the twin's log is at the primary's head"
    );
    for seq in t.first_seq().max(p.first_seq())..=t.head_seq() {
        let (a, b) = (p.read(seq).unwrap(), t.read(seq).unwrap());
        assert!(
            a.kind == b.kind && a.meta == b.meta && a.blob == b.blob,
            "{who}: seq {seq} differs between the primary's log and the twin's"
        );
    }
}

/// **A twin that builds a derived index fails here** (ruling 2026-09-22: "the twin shouldnt
/// do indexing or processing, only receive and save data"). The fixture pushes two packs, a
/// ref set, HEAD, LFS and the whole control plane; the twin follows; its data root must hold
/// no repository store, no archive, no oid tail, no graph, no explode pool, no pack index and
/// no LFS object — and every frame, pack bytes included, must be in its log byte for byte.
/// Neutralise [`take_raw`]'s store-only branch and this goes RED on the first `repos/` file.
#[test]
fn a_store_only_twin_builds_no_derived_index_and_keeps_every_frame_byte_for_byte() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    assert!(
        !derived_on(primary.root()).is_empty(),
        "the fixture must build derived files on the PRIMARY, or this test cannot fail"
    );
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    let head = primary.txlog().head_seq();
    assert_eq!(f.step().unwrap(), Progress::Applied { count: head, head });
    assert_store_only(&primary, &twin, "store-only twin");
    // The control plane it gates its own door with IS applied: the identity table and the KEK.
    assert!(
        twin.auth().principal("alice").unwrap().is_some(),
        "the twin holds the primary's principals"
    );
    assert_eq!(kek_of(&twin), kek_of(&primary), "the twin holds the KEK");
    // A restart re-takes the head entry: still nothing derived.
    drop(f);
    drop(twin);
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    assert_eq!(f.step().unwrap(), Progress::CaughtUp { head });
    assert_store_only(&primary, &twin, "store-only twin after a restart");
    row(
        "a store-only twin builds no derived index and keeps every frame byte for byte",
        true,
        &format!("{head} frames, 0 derived files"),
    );
}

/// **The recharge builds on the NEW main what the twin never built.** A main recharged from a
/// store-only twin holds every repository store, the archive's oid tail and every object —
/// read through the engine, so an index that was not rebuilt fails on the first `has` — and is
/// the primary in every class [`assert_same`] checks.
#[test]
fn a_recharge_from_a_store_only_twin_rebuilds_every_index_on_the_new_main() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert!(derived_on(twin.root()).is_empty(), "the twin built nothing");
    let (_fd, main) = recharged(&twin);
    let derived = derived_on(main.root());
    assert!(
        derived.iter().any(|p| p.ends_with("objects.tail")),
        "the recharge built the archive's oid tail on the new main: {derived:?}"
    );
    assert!(
        derived.iter().any(|p| p.contains("/lfs/")),
        "the recharge put the LFS object on the new main: {derived:?}"
    );
    assert_same(
        &primary,
        &main,
        &facts,
        "main recharged from a store-only twin",
    );
    row("a recharge from a store-only twin rebuilds every index on the new main and matches the primary", true, &format!("{} derived files on main, 0 on the twin", derived.len()));
}

/// The per-root files a restore must reproduce byte for byte.
const ROOT_FILES: &[&str] = &[
    "attestations.jsonl",
    "webhooks.json",
    "control.git/secrets.kek",
];

fn row(check: &str, ok: bool, detail: &str) {
    crate::functional_status("twin", check, ok, detail);
}

#[test]
fn a_twin_that_joins_late_fetches_everything_in_order_and_matches() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    assert!(
        head >= 9,
        "the primary was told at least nine things: {head}"
    );
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    let p = f.step().unwrap();
    assert_eq!(p, Progress::Applied { count: head, head });
    assert_eq!(f.step().unwrap(), Progress::CaughtUp { head });
    let s = status(twin.root()).unwrap();
    assert!(s.caught_up(), "{s:?}");
    assert_store_only(&primary, &twin, "late twin");
    assert_same(
        &primary,
        &recharged(&twin).1,
        &facts,
        "main recharged from a late twin",
    );
    row(
        "a twin that joins late fetches every entry in order and matches the primary",
        true,
        &format!("{head} entries"),
    );
}

/// A source that fails after `allow` fetches — the network going away
/// mid-range.
struct Flaky {
    inner: Arc<Log>,
    allow: usize,
    seen: AtomicUsize,
}

impl LogSource for Flaky {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        self.inner.head()
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        if self.seen.fetch_add(1, Ordering::SeqCst) >= self.allow {
            return Err(anyhow::anyhow!("connection reset"));
        }
        self.inner.fetch(from, to, max_bytes)
    }
}

#[test]
fn a_restart_mid_range_resumes_at_the_twins_own_head() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let applied_before = {
        let twin = open_twin(td.path());
        let f = Follower::new(
            Arc::clone(&twin),
            Box::new(Flaky {
                inner: Arc::clone(primary.txlog()),
                allow: 3,
                seen: AtomicUsize::new(0),
            }),
        )
        .with_fetch_bytes(1);
        let err = f.step().unwrap_err();
        assert!(err.to_string().contains("connection reset"), "{err}");
        let got = twin.txlog().head_seq();
        assert_eq!(got, 3, "one entry per fetch, three fetches allowed");
        got
    };
    // "Restart": fresh stores over the same directory, a fresh follower.
    let twin = open_twin(td.path());
    assert_eq!(twin.txlog().head_seq(), applied_before);
    let f =
        Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog()))).with_fetch_bytes(1);
    assert_eq!(
        f.step().unwrap(),
        Progress::Applied {
            count: head - applied_before,
            head
        }
    );
    assert_store_only(&primary, &twin, "restarted twin");
    assert_same(
        &primary,
        &recharged(&twin).1,
        &facts,
        "main recharged from a restarted twin",
    );
    row(
        "a twin restarted mid-range resumes at its own head and ends identical",
        true,
        &format!("cut at {applied_before} of {head}"),
    );
}

#[test]
fn a_torn_tail_on_the_twin_is_cut_on_open_and_fetched_again() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    {
        let twin = open_twin(td.path());
        let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
        f.step().unwrap();
        assert_eq!(twin.txlog().head_seq(), head);
    }
    // Tear the last frame: the crash between the payload and the fsync.
    let seg = std::fs::read_dir(td.path().join(crate::stores::TXLOG_DIR))
        .unwrap()
        .filter_map(|e| e.ok())
        .map(|e| e.path())
        .find(|p| p.extension().is_some_and(|x| x == "log"))
        .expect("one segment");
    let len = std::fs::metadata(&seg).unwrap().len();
    let f = std::fs::OpenOptions::new().write(true).open(&seg).unwrap();
    f.set_len(len - 3).unwrap();
    drop(f);
    let twin = open_twin(td.path());
    assert_eq!(twin.txlog().head_seq(), head - 1, "the torn frame is gone");
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    assert_eq!(f.step().unwrap(), Progress::Applied { count: 1, head });
    assert_store_only(&primary, &twin, "torn twin");
    assert_same(
        &primary,
        &recharged(&twin).1,
        &facts,
        "main recharged from a torn twin",
    );
    row(
        "a torn tail on the twin is cut on open and the entry fetched again",
        true,
        &format!("seq {head} re-applied"),
    );
}

/// A primary whose retention has moved past what the twin needs.
struct Retained {
    first: u64,
    head: u64,
}

impl LogSource for Retained {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Ok((self.first, self.head))
    }
    fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
        panic!("a twin with a hole must not fetch (asked for {from})")
    }
}

#[test]
fn a_hole_in_retention_stops_the_twin_and_says_snapshot_needed() {
    let td = tempfile::tempdir().unwrap();
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Retained { first: 5, head: 10 }));
    assert_eq!(
        f.step().unwrap(),
        Progress::SnapshotNeeded {
            wanted: 1,
            first: 5
        }
    );
    assert_eq!(twin.txlog().head_seq(), 0, "nothing appended past a hole");
    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::SnapshotNeeded);
    assert!(!s.caught_up());
    assert!(s.why.contains("retains from 5"), "{}", s.why);
    let stop = AtomicBool::new(false);
    // `run` returns on its own on a snapshot-needed — it does not spin.
    f.run(Duration::from_millis(1), &stop);
    row(
        "a hole in the primary's retention stops the twin with snapshot-needed and appends nothing",
        true,
        "first 5, twin at 0",
    );
}

#[test]
fn a_restore_from_the_twin_rebuilds_every_object_and_every_document() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    // The primary is gone. The twin's log is what there is.
    // (Kept open only to be COMPARED against: a store-only twin holds no stores to compare.)
    let twin_log = Arc::clone(twin.txlog());
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    // The fresh box wrote documents of its own before the restore (its
    // first-start seed): the fresh-box rule must not let them win.
    fresh
        .catalog()
        .create_namespace("fresh-seed")
        .expect("fresh seed");
    assert_eq!(
        restore(Arc::clone(&twin_log), Arc::clone(&fresh), "").unwrap(),
        head
    );
    assert_same(&primary, &fresh, &facts, "restored primary");
    // The restored box restarts (it holds the fresh KEK in memory) and opens
    // as the primary: its sealed secret opens under the replayed KEK.
    drop(fresh);
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert!(!fresh.is_twin() && !fresh.txlog().is_follower());
    assert_eq!(
        fresh.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
        Some(&b"the sealed thing"[..])
    );
    // The open re-sealed under its own KEK (a Kek frame and one Secret frame per sealed value).
    let head = fresh.txlog().head_seq();
    // And the restored primary carries on: its next mutation is `head + 1`.
    let repo = fresh
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = fresh.repo_store(&repo).unwrap();
    let (p3, o3) = one_blob_pack(store.hash_kind(), b"after the restore\n");
    store.serve().put_pack(&p3).unwrap();
    store
        .serve()
        .update_ref("refs/heads/next", None, Some(&o3))
        .unwrap();
    assert_eq!(fresh.txlog().head_seq(), head + 2);
    row("a restore from the twin rebuilds every object and every document, and the restored primary continues the sequence", true, &format!("{head} entries replayed"));
}

/// **The owner's question, as a test** (lane TWIN-COMPLETENESS): after the restart — the one
/// that re-seals every secret under a KEK of the restored box's own — is the box ITSELF, or
/// is it a box with the repositories and none of what made it that box?
///
/// Every assertion here is read through the WORKING path, after the re-seal has moved the
/// bytes: the sealed store, `Pki` as the server builds it, the identity table, the catalog.
/// Nothing here mints, so each one fails for the thing it is named after — a box that came
/// back with a new host key, a new issuer, no groups or no entitlement fact fails on that
/// line and no other.
#[test]
fn a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());

    // What the box WAS, read the way a client sees it.
    let host_fp = gunnar_core::ensure_host_key(&primary.secrets().unwrap())
        .unwrap()
        .fingerprint()
        .to_string();
    let issuer = primary
        .pki_readonly()
        .unwrap()
        .issuer()
        .unwrap()
        .expect("issuer");
    let issuer_key = primary
        .secrets()
        .unwrap()
        .get(gunnar_store::ISSUER_PRIVATE_KEY)
        .unwrap()
        .expect("issuer key");
    let grants: Vec<String> = primary
        .auth()
        .grants(None)
        .unwrap()
        .iter()
        .map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
        .collect();
    let namespaces: Vec<String> = primary
        .catalog()
        .namespaces()
        .unwrap()
        .into_iter()
        .map(|n| n.name)
        .collect();
    let rows: Vec<String> = primary
        .catalog()
        .list(None)
        .unwrap()
        .into_iter()
        .map(|r| {
            format!(
                "{}/{} {:?} {}",
                r.namespace, r.name, r.visibility, r.default_branch
            )
        })
        .collect();
    let framed: Vec<Option<Vec<u8>>> = gunnar_core::txlog::FRAMED_FILES
        .iter()
        .map(|rel| std::fs::read(pd.path().join(rel)).ok())
        .collect();
    #[cfg(feature = "metered")]
    let entitlements: Vec<String> =
        gunnar_core::EntitlementStore::new(Arc::clone(primary.control_store()))
            .unwrap()
            .all()
            .unwrap()
            .into_iter()
            .map(|f| {
                format!(
                    "{}={}@{} {:?} {}",
                    f.namespace,
                    f.state,
                    f.plan,
                    f.caps,
                    hex::encode(&f.signature)
                )
            })
            .collect();

    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let primary_kek = kek_of(&primary);
    drop(primary);

    // A fresh box on blank disks, restored from the twin, and RESTARTED — the re-seal runs on
    // that open, so every read below is through a KEK the dead primary never held.
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    drop(fresh);
    let back = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_ne!(kek_of(&back), primary_kek, "the restored box owns its KEK");

    // ── the SSH host key: no client sees a changed fingerprint ──────────────
    let back_fp = gunnar_core::ensure_host_key(&back.secrets().unwrap())
        .unwrap()
        .fingerprint()
        .to_string();
    assert_eq!(
        back_fp, host_fp,
        "the restored box serves a DIFFERENT SSH host key; every client would refuse it with a \
         changed-fingerprint warning"
    );

    // ── the issuer: the certificate, and a key that still signs under it ────
    let back_issuer = back
        .pki_readonly()
        .unwrap()
        .issuer()
        .unwrap()
        .expect("the restored box has an issuer");
    assert_eq!(
        back_issuer.cert_pem, issuer.cert_pem,
        "the restored box minted a NEW issuer certificate"
    );
    assert_eq!(
        back_issuer.fingerprint, issuer.fingerprint,
        "the issuer fingerprint moved"
    );
    assert_eq!(
        back.secrets()
            .unwrap()
            .get(gunnar_store::ISSUER_PRIVATE_KEY)
            .unwrap(),
        Some(issuer_key),
        "the issuer private key did not survive the re-seal"
    );
    // And it can still ISSUE: a sealed key that cannot be used is a key that is not back.
    let issued = back
        .pki()
        .unwrap()
        .issue_with_server_key("someone", 30)
        .expect("the restored box can still mint a certificate");
    assert!(!issued.cert_pem.is_empty());

    // ── the control plane's own state ───────────────────────────────────────
    assert_eq!(
        back.auth()
            .grants(None)
            .unwrap()
            .iter()
            .map(|g| format!("{}@{:?}={:?}", g.principal, g.scope, g.level))
            .collect::<Vec<_>>(),
        grants,
        "grants"
    );
    assert_eq!(
        back.auth().groups_of("alice").unwrap(),
        vec![GROUP.to_string()],
        "group membership"
    );
    assert_eq!(
        back.catalog()
            .namespaces()
            .unwrap()
            .into_iter()
            .map(|n| n.name)
            .collect::<Vec<_>>(),
        namespaces,
        "namespaces"
    );
    assert_eq!(
        back.catalog()
            .list(None)
            .unwrap()
            .into_iter()
            .map(|r| format!(
                "{}/{} {:?} {}",
                r.namespace, r.name, r.visibility, r.default_branch
            ))
            .collect::<Vec<_>>(),
        rows,
        "catalog rows: visibility and default branch"
    );
    #[cfg(feature = "metered")]
    assert_eq!(
        gunnar_core::EntitlementStore::new(Arc::clone(back.control_store()))
            .unwrap()
            .all()
            .unwrap()
            .into_iter()
            .map(|f| format!(
                "{}={}@{} {:?} {}",
                f.namespace,
                f.state,
                f.plan,
                f.caps,
                hex::encode(&f.signature)
            ))
            .collect::<Vec<_>>(),
        entitlements,
        "entitlement facts: a restored box that forgets who has paid is the worst version of this"
    );
    assert_eq!(
        gunnar_core::txlog::FRAMED_FILES
            .iter()
            .map(|rel| std::fs::read(fd.path().join(rel)).ok())
            .collect::<Vec<_>>(),
        framed,
        "the signed append-only logs and whole documents"
    );

    row(
        "a restored box comes back as the SAME box: same SSH host key, same issuer certificate and a key that still mints, same grants, groups, namespaces, visibility, default branches, entitlement facts and signed logs",
        true,
        &format!("host key {host_fp}, issuer {}", issuer.fingerprint),
    );
}

/// **The restore onto DIFFERENT hardware** (lane TWIN-COMPLETENESS): the data set left a
/// machine with four members and arrives at one with two, which may hold less than what is
/// coming. The honest outcomes are "it fits and restores whole" or "it refuses by name, before
/// it writes anything, saying how much is short" — never a stream that fills the volume and
/// stops with the tail missing.
#[test]
fn a_restore_onto_a_data_set_too_small_for_it_refuses_before_the_first_byte_and_says_how_much_is_short(
) {
    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let (total, _used) = volume_bytes(fd.path());
    assert!(
        total > 0,
        "this test needs a measurable filesystem under the temp dir"
    );

    // The service's order: `restore_begin` (which resets this box's own log to empty), then the
    // room check on the first chunk, then the first `restore_apply`.
    assert_eq!(restore_begin(&fresh).unwrap(), 1);
    assert_eq!(
        fresh.txlog().head_seq(),
        0,
        "a box that has begun a restore has an empty log"
    );

    // A stream announcing more than the whole volume cannot fit however empty the box is.
    let err = room_for_restore(fd.path(), total).unwrap_err().to_string();
    assert!(err.contains("NOT ENOUGH ROOM"), "{err}");
    assert!(
        err.contains("SHORT BY"),
        "the refusal must say how much is short: {err}"
    );
    assert!(
        err.contains("SMALLER data set"),
        "the refusal must name the cause: {err}"
    );
    // Nothing was written by asking, and the box stays resumable rather than half-restored.
    assert_eq!(fresh.txlog().head_seq(), 0, "the room check writes nothing");
    assert!(
        is_restoring(fd.path()),
        "the refused box stays restoring: the same replay into a bigger set starts at 1"
    );

    // A stream that fits is not refused, and a sender that did not say is not guessed at.
    room_for_restore(fd.path(), 1 << 10).expect("a kilobyte fits");
    room_for_restore(fd.path(), 0).expect("0 = not announced, and a guess is not a measurement");
    // An unmeasurable root is not "full": a `statvfs` that failed must not refuse a restore.
    room_for_restore(
        std::path::Path::new("/proc/self/no-such-place"),
        u64::MAX / 4,
    )
    .expect("unmeasurable is not full");

    // The rule itself, spelled: twice the log plus the box's reserve.
    let need_for = |b: u64| b * RESTORE_BYTES_PER_LOG_BYTE;
    assert!(
        need_for(total) > total,
        "a log the size of the volume cannot fit twice over"
    );

    row(
        "a restore onto a data set too small for it refuses before the first byte lands and says how much is short",
        true,
        &format!("volume {total} B, refused a stream announcing {total} B of log"),
    );
}

#[test]
fn fetch_range_gives_whole_entries_never_fewer_than_one_and_names_the_edges() {
    let dir = tempfile::tempdir().unwrap();
    let log = Log::open(dir.path()).unwrap();
    for i in 0..5u8 {
        log.append(
            Kind::Pack,
            &Meta::Pack {
                repo: "a/b".into(),
                hash: "sha256".into(),
            },
            &vec![i; 1000],
        )
        .unwrap();
    }
    let got = fetch_range(&log, 1, 0, 1).unwrap();
    assert_eq!(got.len(), 1, "one entry crosses the byte cap");
    let got = fetch_range(&log, 2, 4, 2500).unwrap();
    assert_eq!(got.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![2, 3, 4]);
    let got = fetch_range(&log, 1, 0, 0).unwrap();
    assert_eq!(got.len(), 5);
    assert!(matches!(
        fetch_range(&log, 6, 0, 0),
        Err(txlog::Error::BeyondHead { .. })
    ));
    assert!(
        matches!(
            fetch_range(&log, 0, 0, 0),
            Err(txlog::Error::NotRetained { .. }) | Err(txlog::Error::BeyondHead { .. })
        ) || fetch_range(&log, 0, 0, 0).is_err()
    );
}

/// Every file under `root`, relative.
fn files_under(root: &std::path::Path) -> Vec<String> {
    fn walk(base: &std::path::Path, dir: &std::path::Path, out: &mut Vec<String>) {
        for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
            let p = e.path();
            if p.is_dir() {
                walk(base, &p, out);
            } else {
                out.push(p.strip_prefix(base).unwrap().to_string_lossy().into_owned());
            }
        }
    }
    let mut out = Vec::new();
    walk(root, root, &mut out);
    out
}

#[test]
fn no_cache_path_ever_enters_the_txlog_and_none_lands_on_the_twin() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    // The primary grows its caches: every object read back (the engine's read
    // cache), the start-up disk probe, a temporary.
    let repo = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&repo).unwrap();
    for oid in &facts.oids {
        store.serve().read(oid).unwrap();
    }
    std::fs::create_dir_all(pd.path().join(crate::disktest::DIR_NAME)).unwrap();
    std::fs::write(
        pd.path().join(crate::disktest::DIR_NAME).join("probe"),
        b"x",
    )
    .unwrap();
    // A cache write offered to the sink is refused by name, not framed.
    assert!(
        gunnar_core::txlog::sink::replace_file(&pd.path().join("gc/scratch.tmp"), b"x", false)
            .is_err()
    );
    assert!(
        gunnar_core::txlog::sink::append_line(&pd.path().join(".korp-disktest/log"), "x").is_err()
    );

    let log = primary.txlog();
    let entries = log.range(log.first_seq(), log.head_seq()).unwrap();
    let mut named = 0;
    for e in &entries {
        let path = match e.meta().unwrap() {
            Meta::Append { log } => log,
            Meta::File { path, .. } => path,
            _ => continue,
        };
        named += 1;
        assert!(
            !gunnar_core::txlog::is_cache_path(&path),
            "seq {}: {path} is a cache path in the log",
            e.seq
        );
    }
    assert!(
        named >= 2,
        "the primary framed its signed line and its document ({named})"
    );

    // A frame that names a cache path is refused on apply, whoever sent it.
    let twin = open_twin(td.path());
    let bad = Entry {
        seq: 1,
        kind: Kind::Append as u16,
        meta: Meta::Append {
            log: ".korp-disktest/probe".into(),
        }
        .to_bytes(),
        blob: b"x".to_vec(),
    };
    assert!(apply(&twin, &bad).is_err());

    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let caches: Vec<String> = files_under(td.path())
        .into_iter()
        .filter(|p| gunnar_core::txlog::is_cache_path(p))
        .collect();
    assert!(caches.is_empty(), "cache files on the twin: {caches:?}");
    let primary_caches = files_under(pd.path())
        .into_iter()
        .filter(|p| gunnar_core::txlog::is_cache_path(p))
        .count();
    row(
        "no cache path ever enters the txlog, and none lands on the twin",
        true,
        &format!(
            "{} frames, {named} path-named; primary holds {primary_caches} cache files, twin 0",
            entries.len()
        ),
    );
}

/// Records every `from` a fetch asks for.
struct Recording {
    inner: Arc<Log>,
    froms: Arc<std::sync::Mutex<Vec<u64>>>,
}

impl LogSource for Recording {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        self.inner.head()
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        self.froms.lock().unwrap().push(from);
        self.inner.fetch(from, to, max_bytes)
    }
}

/// A primary that answers a range starting one PAST what was asked — the gap
/// arriving without the entry that fills it.
struct Skipping {
    inner: Arc<Log>,
}

impl LogSource for Skipping {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        self.inner.head()
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        self.inner.fetch(from + 1, to, max_bytes)
    }
}

#[test]
fn a_twin_one_gap_behind_fetches_the_gap_before_applying() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let h = twin.txlog().head_seq();
    // The primary moves on while the twin is away: two more transactions.
    let repo = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&repo).unwrap();
    store
        .serve()
        .update_ref("refs/heads/gap-1", None, Some(&facts.oids[0]))
        .unwrap();
    store
        .serve()
        .update_ref("refs/heads/gap-2", None, Some(&facts.oids[1]))
        .unwrap();
    assert_eq!(primary.txlog().head_seq(), h + 2);

    // Offered h+2 without h+1: refused before anything is appended or applied.
    let err = Follower::new(
        Arc::clone(&twin),
        Box::new(Skipping {
            inner: Arc::clone(primary.txlog()),
        }),
    )
    .step()
    .unwrap_err();
    assert!(
        err.to_string()
            .contains(&format!("where {} was due", h + 1)),
        "{err}"
    );
    assert_eq!(twin.txlog().head_seq(), h, "nothing appended past the gap");

    // The honest loop asks for the gap first, then applies in order.
    let froms = Arc::new(std::sync::Mutex::new(Vec::new()));
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(Recording {
            inner: Arc::clone(primary.txlog()),
            froms: Arc::clone(&froms),
        }),
    )
    .with_fetch_bytes(1);
    assert_eq!(
        f.step().unwrap(),
        Progress::Applied {
            count: 2,
            head: h + 2
        }
    );
    assert_eq!(
        froms.lock().unwrap().first().copied(),
        Some(h + 1),
        "the first fetch is the gap"
    );
    assert_store_only(&primary, &twin, "a twin that fetched its gap");
    assert_eq!(
        refs_of(&primary),
        refs_of(&recharged(&twin).1),
        "the gap's refs are on the main recharged from the twin"
    );
    row(
        "a twin one gap behind fetches the gap before applying, and refuses a range that skips it",
        true,
        &format!("gap {}..={}", h + 1, h + 2),
    );
}

#[test]
fn become_twin_takes_a_fresh_box_only_and_the_box_reopens_as_a_twin() {
    let dir = tempfile::tempdir().unwrap();
    {
        let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
        assert!(
            stores.txlog().head_seq() >= 1,
            "a primary frames its KEK first"
        );
        let key = become_twin(&stores, "10.0.3.2:50051").unwrap();
        assert!(key.starts_with("ssh-ed25519 "), "{key}");
        assert_eq!(
            stores.txlog().head_seq(),
            0,
            "the fresh box gave up its own numbers"
        );
        assert!(stores.txlog().is_follower());
        // The fresh box wrote its own identity mirror at open; a twin keeps none (C7, the KVM scan).
        let meta = stores.slow().path_of(crate::slow::META_DIR);
        assert!(
            !meta.join(crate::stores::IDENTITY_MIRROR).exists(),
            "no identity mirror on a twin"
        );
        assert!(
            stores.is_twin(),
            "marked a twin at runtime, not only at the next open"
        );
        stores.mirror_identity();
        assert!(
            !meta.join(crate::stores::IDENTITY_MIRROR).exists(),
            "a marked twin writes no mirror"
        );
        assert_eq!(
            become_twin(&stores, "10.0.3.2:50051").unwrap(),
            key,
            "idempotent for the same primary"
        );
        assert!(
            become_twin(&stores, "elsewhere:50051").is_err(),
            "a twin is never re-pointed"
        );
    }
    let stores = ServerStores::open(dir.path()).unwrap();
    assert!(
        stores.is_twin() && stores.txlog().is_follower() && stores.browse().load(Ordering::SeqCst)
    );
    assert!(
        stores.catalog().create_namespace("local").is_err(),
        "a twin originates nothing"
    );

    let busy = tempfile::tempdir().unwrap();
    let stores = ServerStores::open(busy.path()).unwrap();
    stores.catalog().create_namespace("ns").unwrap();
    stores
        .create_repo(
            NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
            false,
        )
        .unwrap();
    assert!(
        become_twin(&stores, "10.0.3.2:50051").is_err(),
        "a box with repositories is not fresh"
    );
    row("Bootstrap.BecomeTwin takes a fresh box only; it reopens follower-only, frozen and originating nothing", true, "");
}

/// ★ **The WireGuard-only upgrade drill (2026-10-05, KVM mock)**: test2's twin
/// was joined on main's UTILITY address; upgraded in place (`/data` kept), the
/// door names the same main by its OVERLAY address and the twin refused it 335
/// times ("a twin is never re-pointed"). The same primary moves onto its tunnel;
/// nothing else moves, and `become_twin` still refuses any other primary.
#[test]
fn a_twin_joined_on_its_primarys_utility_address_moves_onto_the_same_boxs_tunnel_and_nowhere_else()
{
    let dir = tempfile::tempdir().unwrap();
    let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
    let (utility, overlay) = ("10.13.12.118:50051", "10.200.0.2:50051");
    assert!(
        !onto_tunnel(&stores, utility, overlay).unwrap(),
        "not a twin: nothing to move"
    );
    let key = become_twin(&stores, utility).unwrap();
    assert!(
        become_twin(&stores, overlay).is_err(),
        "the refusal the drill met"
    );

    assert!(
        !onto_tunnel(&stores, "10.13.12.119:50051", overlay).unwrap(),
        "another box's utility address: not moved"
    );
    assert_eq!(
        TwinConfig::load(dir.path()).unwrap().unwrap().primary,
        utility
    );

    assert!(
        onto_tunnel(&stores, utility, overlay).unwrap(),
        "the same primary, over its tunnel"
    );
    assert_eq!(
        TwinConfig::load(dir.path()).unwrap().unwrap().primary,
        overlay
    );
    assert_eq!(
        become_twin(&stores, overlay).unwrap(),
        key,
        "then idempotent, the same key"
    );
    assert!(
        !onto_tunnel(&stores, utility, overlay).unwrap(),
        "moved once"
    );
    assert!(
        become_twin(&stores, "10.200.0.9:50051").is_err(),
        "and never re-pointed at another primary"
    );
    row("a twin joined on its primary's utility address moves onto that same box's tunnel, and nowhere else", true, overlay);
}

#[test]
fn a_restore_that_received_nothing_does_not_finish_and_the_box_stays_resumable() {
    let fd = tempfile::tempdir().unwrap();
    {
        let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
        assert_eq!(restore_begin(&fresh).unwrap(), 1);
        // The stream ended before its first entry — a fetch that failed on the twin's side.
        let err = restore_finish(&fresh, 0, "").unwrap_err();
        assert!(err.to_string().contains("nothing was restored"), "{err}");
        assert!(
            is_restoring(fd.path()),
            "the marker stays, so a resent stream resumes"
        );
    }
    // A box still restoring reopens follower-only and takes the twin's numbers from 1.
    let again = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert!(
        again.txlog().is_follower(),
        "a box still restoring originates nothing"
    );
    assert_eq!(restore_begin(&again).unwrap(), 1);
    // The resent replay then finishes, and the box is no longer restoring.
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    assert_eq!(
        restore(Arc::clone(primary.txlog()), Arc::clone(&again), "").unwrap(),
        head
    );
    assert!(!is_restoring(fd.path()));
    row("a restore stream that carried nothing does not finish; the box stays restoring and a resent replay completes", true, &format!("{head} entries on the resend"));
}

#[test]
fn a_short_stream_below_the_announced_head_does_not_finish_and_the_box_resumes() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let log = primary.txlog();
    let (first, head) = (log.first_seq(), log.head_seq());
    assert!(
        first == 1 && head >= 3,
        "the primary's log is long enough to stop short: {first}..={head}"
    );
    let entries = log.range(first, head).unwrap();
    let short = head - 1;

    let fd = tempfile::tempdir().unwrap();
    {
        let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
        assert_eq!(restore_begin(&fresh).unwrap(), 1);
        // The sender's fetch broke after `short` entries and the stream still ended cleanly.
        for e in entries.iter().take(short as usize) {
            restore_apply(&fresh, e).unwrap();
        }
        let err = restore_finish(&fresh, head, "").unwrap_err();
        assert!(err.to_string().contains("short stream"), "{err}");
        assert!(is_restoring(fd.path()), "a short restore keeps the marker");
        assert_eq!(fresh.txlog().head_seq(), short);
    }
    // Reopened: still follower-only, and the resend continues at the short head + 1.
    let again = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert!(
        again.txlog().is_follower(),
        "a box still restoring originates nothing"
    );
    assert_eq!(restore_begin(&again).unwrap(), short + 1);
    for e in entries.iter().skip(short as usize) {
        restore_apply(&again, e).unwrap();
    }
    assert_eq!(restore_finish(&again, head, "").unwrap(), head);
    assert!(!is_restoring(fd.path()));
    row("a restore stream that ends cleanly below the announced head does not finish; the resend from head + 1 does", true, &format!("short {short} of {head}"));
}

#[test]
fn an_entry_whose_apply_failed_is_applied_again_before_anything_newer_not_skipped() {
    let pd = tempfile::tempdir().unwrap();
    let sd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());

    // A source that is the primary's log plus one Refs frame pointing at an object the twin does
    // not have yet — the shape the KVM drill met when a pushed pack was not framed.
    let src = Arc::new(Log::open(sd.path()).unwrap());
    let plog = primary.txlog();
    for e in plog.range(plog.first_seq(), plog.head_seq()).unwrap() {
        src.append_at(e.seq, e.kind, &e.meta, &e.blob).unwrap();
    }
    let hash = {
        let repo = primary
            .catalog()
            .get(&RepoRef::new("ns", "app").unwrap())
            .unwrap()
            .unwrap();
        primary.repo_store(&repo).unwrap().hash_kind()
    };
    let (late_pack, late_oid) = one_blob_pack(hash, b"arrives after its ref\n");
    let edit = RefEdit {
        name: "refs/heads/late".into(),
        old: None,
        new: Some(hex::encode(&late_oid)),
        peeled: None,
        symref: None,
    };
    let dangling = src
        .append(
            Kind::Refs,
            &Meta::Refs {
                repo: REPO.into(),
                atomic: false,
                edits: vec![edit],
            },
            &[],
        )
        .unwrap();

    // A store-only twin (ruling 2026-09-22) applies no ref, so a dangling one costs it nothing:
    // the frame is KEPT and the twin is level. The apply — and its retry — happens where the
    // indexes are built, on the box being recharged, which is what the rest of this test drives.
    {
        let tw = tempfile::tempdir().unwrap();
        let twin = open_twin(tw.path());
        assert_eq!(
            Follower::new(Arc::clone(&twin), Box::new(Arc::clone(&src)))
                .step()
                .unwrap(),
            Progress::Applied {
                count: dangling,
                head: dangling
            }
        );
        assert!(
            derived_on(twin.root()).is_empty(),
            "the twin stored the dangling ref and built nothing"
        );
    }
    let recharging = Arc::new(ServerStores::open(td.path()).unwrap());
    restore_begin(&recharging).unwrap();
    let f = Follower::new(Arc::clone(&recharging), Box::new(Arc::clone(&src)));
    let err = f.step().unwrap_err();
    assert!(format!("{err:#}").contains("refs/heads/late"), "{err:#}");
    assert_eq!(
        recharging.txlog().head_seq(),
        dangling,
        "durable first: the entry is in the box's log"
    );
    let late = |s: &ServerStores| {
        refs_of(s)
            .into_iter()
            .find(|(n, _, _)| n == "refs/heads/late")
            .and_then(|(_, o, _)| o)
    };
    assert_eq!(late(&recharging), None, "…and not in its stores");

    // Nothing new on the source. The object now reaches the box's store.
    let repo = recharging
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    recharging
        .repo_store(&repo)
        .unwrap()
        .unlogged()
        .put_pack(&late_pack)
        .unwrap();

    // The next poll re-applies the head entry instead of resuming past it.
    assert_eq!(f.step().unwrap(), Progress::CaughtUp { head: dangling });
    assert_eq!(
        late(&recharging).as_deref(),
        Some(late_oid.as_slice()),
        "the failed entry landed on the re-apply"
    );
    row(
        "an entry whose apply failed is applied again before anything newer, never skipped",
        true,
        &format!("seq {dangling}"),
    );
}

/// The LFS object `LFS_OID` in `ns/app` on `stores`, if present.
fn lfs_size(stores: &ServerStores) -> Option<u64> {
    use gunnar_lfs::LfsStore as _;
    let dir = stores
        .repo_dir(&RepoRef::new("ns", "app").unwrap())
        .unwrap();
    gunnar_lfs::FsLfsStore::open_or_init(&dir)
        .unwrap()
        .size_of(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
        .unwrap()
}

#[test]
fn a_deleted_lfs_object_and_a_purged_repository_are_gone_on_the_twin_and_a_restore_does_not_resurrect_them(
) {
    use gunnar_lfs::LfsStore as _;
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let doomed_ref = RepoRef::new("ns", "doomed").unwrap();
    let doomed = primary
        .create_repo(
            NewRepo::new(doomed_ref.clone()).hash_kind(HashKind::Sha256),
            false,
        )
        .unwrap();
    {
        let store = primary.repo_store(&doomed).unwrap();
        let (p, o) = one_blob_pack(store.hash_kind(), b"soon gone\n");
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref("refs/heads/main", None, Some(&o))
            .unwrap();
    }
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    f.step().unwrap();
    // A store-only twin holds both in its LOG; a main recharged from it holds them in stores.
    assert!(
        derived_on(twin.root()).is_empty(),
        "the twin builds no store"
    );
    {
        let (_fd, main) = recharged(&twin);
        assert!(
            lfs_size(&main).is_some(),
            "the twin holds the LFS object before the delete"
        );
        assert!(
            main.repo_dir(&doomed_ref).unwrap().exists(),
            "the twin holds the doomed store before the purge"
        );
    }

    // The primary deletes the LFS object and purges the repository, the way `repo delete --purge` does.
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let app_store = primary.repo_store(&app).unwrap();
    let lfs = crate::txlog_lfs::LoggedLfs::new(
        gunnar_lfs::FsLfsStore::open_or_init(app_store.git_dir()).unwrap(),
        Arc::clone(primary.txlog()),
        REPO,
    );
    assert!(lfs
        .delete(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
        .unwrap());
    assert!(
        !lfs.delete(&gunnar_lfs::Oid::parse(LFS_OID).unwrap())
            .unwrap(),
        "a second delete finds nothing and frames nothing"
    );
    primary.catalog().delete(&doomed_ref).unwrap();
    primary.purge_repo(&doomed_ref).unwrap();
    let kinds: Vec<u16> = {
        let log = primary.txlog();
        log.range(log.first_seq(), log.head_seq())
            .unwrap()
            .iter()
            .map(|e| e.kind)
            .collect()
    };
    assert_eq!(
        kinds
            .iter()
            .filter(|k| **k == Kind::LfsDelete as u16)
            .count(),
        1,
        "one LfsDelete frame"
    );
    assert_eq!(
        kinds
            .iter()
            .filter(|k| **k == Kind::RepoStore as u16)
            .count(),
        1,
        "one RepoStore purge frame"
    );
    assert_eq!(lfs_size(&primary), None);

    // After the watermark the twin has neither.
    let head = primary.txlog().head_seq();
    f.step().unwrap();
    assert_eq!(twin.txlog().head_seq(), head);
    assert!(
        derived_on(twin.root()).is_empty(),
        "the twin still builds no store"
    );
    assert!(
        twin.catalog().get(&doomed_ref).unwrap().is_none(),
        "the catalog is control plane: applied on the twin"
    );

    // A restore from the twin replays the pack and the ref, then the delete and the purge: nothing comes back.
    drop(primary);
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(
        restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap(),
        head
    );
    assert_eq!(
        lfs_size(&fresh),
        None,
        "a restore does not resurrect a deleted LFS object"
    );
    assert!(
        !fresh.repo_dir(&doomed_ref).unwrap().exists(),
        "a restore does not resurrect a purged store"
    );
    assert!(fresh.catalog().get(&doomed_ref).unwrap().is_none());
    assert!(
        fresh
            .catalog()
            .get(&RepoRef::new("ns", "app").unwrap())
            .unwrap()
            .is_some(),
        "the live repository is back"
    );
    row("an LFS delete and a repository purge on the primary are applied on the twin in order, and a restore does not resurrect either", true, &format!("head {head}"));
}

fn kek_of(stores: &ServerStores) -> Vec<u8> {
    std::fs::read(gunnar_core::seal::kek_path(stores.control_dir())).unwrap()
}

#[test]
fn a_restored_box_reseals_every_secret_under_a_kek_of_its_own_and_its_twin_follows_the_new_key() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let gd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let primary_kek = kek_of(&primary);
    let twin = open_twin(td.path());
    let follow = |from: &Arc<ServerStores>| {
        Follower::new(Arc::clone(&twin), Box::new(Arc::clone(from.txlog())))
            .step()
            .unwrap()
    };
    follow(&primary);
    drop(primary);

    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let head = restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    assert_eq!(
        kek_of(&fresh),
        primary_kek,
        "before its restart the restored box holds the replayed KEK"
    );
    assert!(reseal_pending(fd.path()));
    drop(fresh);

    // The restart re-seals.
    let restored = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert!(
        !reseal_pending(fd.path()),
        "the marker goes once the re-seal is done"
    );
    let own = kek_of(&restored);
    assert_ne!(own, primary_kek, "the restored box owns its KEK");
    let old = gunnar_core::Kek::from_bytes(primary_kek.clone().try_into().unwrap());
    let sealed = restored
        .raw_secrets()
        .get(SECRET_KEY)
        .unwrap()
        .expect("sealed value");
    assert!(
        gunnar_core::unseal(&old, SECRET_KEY, &sealed).is_err(),
        "the dead primary's KEK no longer opens the restored box's secret"
    );
    assert_eq!(
        restored
            .secrets()
            .unwrap()
            .get(SECRET_KEY)
            .unwrap()
            .as_deref(),
        Some(&b"the sealed thing"[..])
    );
    let frames = restored
        .txlog()
        .range(head + 1, restored.txlog().head_seq())
        .unwrap();
    assert_eq!(
        frames.first().map(|e| e.kind),
        Some(Kind::Kek as u16),
        "the new KEK is a frame first"
    );
    assert!(
        frames
            .iter()
            .any(|e| matches!(e.meta(), Ok(Meta::Secret { ref key, .. }) if key == SECRET_KEY)),
        "each re-sealed value is a frame"
    );
    // A second open does not re-seal again.
    let h = restored.txlog().head_seq();
    drop(restored);
    let restored = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(restored.txlog().head_seq(), h);
    assert_eq!(kek_of(&restored), own);

    // The twin follows the restored primary onto the new key …
    follow(&restored);
    assert_eq!(kek_of(&twin), own, "the twin holds the restored box's KEK");
    assert_eq!(
        twin.raw_secrets().get(SECRET_KEY).unwrap(),
        restored.raw_secrets().get(SECRET_KEY).unwrap()
    );
    // … and a restore from it opens every secret (and re-seals once more, under that box's own).
    let again = Arc::new(ServerStores::open(gd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&again), "").unwrap();
    drop(again);
    let again = ServerStores::open(gd.path()).unwrap();
    assert_eq!(
        again.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
        Some(&b"the sealed thing"[..])
    );
    assert_ne!(kek_of(&again), own);
    row("a restored box re-seals every secret under a KEK of its own; the old KEK opens nothing, the twin follows the new key and a restore from it opens every secret", true, &format!("head {head}"));
}

/// The state a snapshot must reproduce: the source's documents, its live secrets (sealed bytes),
/// the refs of `ns/app`, the KEK, every framed file, the LFS object.
fn same_state(a: &ServerStores, b: &ServerStores, who: &str) {
    let (ea, eb) = (
        a.control_db().export().unwrap(),
        b.control_db().export().unwrap(),
    );
    for (name, id) in &ea.docs {
        assert_eq!(eb.docs.get(name), Some(id), "{who}: document {name}");
    }
    let live = |e: &gunnar_core::control::ControlExport| {
        e.secrets
            .iter()
            .filter(|(_, v)| v.is_some())
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect::<Vec<_>>()
    };
    assert_eq!(live(&ea), live(&eb), "{who}: live secrets");
    assert_eq!(refs_of(a), refs_of(b), "{who}: refs");
    assert_eq!(kek_of(a), kek_of(b), "{who}: KEK");
    for rel in gunnar_core::txlog::FRAMED_FILES {
        assert_eq!(
            std::fs::read(a.root().join(rel)).ok(),
            std::fs::read(b.root().join(rel)).ok(),
            "{who}: {rel}"
        );
    }
    assert_eq!(lfs_size(a), lfs_size(b), "{who}: LFS");
}

#[test]
fn a_twin_behind_a_retention_hole_catches_up_via_snapshot_and_then_follows() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, facts) = primary(pd.path());
    let doomed_ref = RepoRef::new("ns", "doomed").unwrap();
    let doomed = primary
        .create_repo(
            NewRepo::new(doomed_ref.clone()).hash_kind(HashKind::Sha256),
            false,
        )
        .unwrap();
    {
        let store = primary.repo_store(&doomed).unwrap();
        let (p, o) = one_blob_pack(store.hash_kind(), b"the twin saw this one\n");
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref("refs/heads/main", None, Some(&o))
            .unwrap();
    }
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert!(
        !twin.repo_dir(&doomed_ref).unwrap().exists(),
        "a store-only twin opens no repository store"
    );

    // The twin is away. The primary deletes a tag, purges a repository, deletes a secret, adds a
    // principal and a ref — and retention drops everything the twin has not seen.
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let app_store = primary.repo_store(&app).unwrap();
    app_store
        .serve()
        .put_refs(&[gunnar_store::git_storage_trait::RefUpdate {
            name: "refs/tags/v1".into(),
            target: None,
            peeled: None,
            symref_target: None,
        }])
        .unwrap();
    primary.catalog().delete(&doomed_ref).unwrap();
    primary.purge_repo(&doomed_ref).unwrap();
    primary.secrets().unwrap().delete(SECRET_KEY).unwrap();
    primary
        .auth()
        .add_principal(Principal::new("bob", "Bob", PrincipalKind::User).unwrap())
        .unwrap();
    app_store
        .serve()
        .update_ref("refs/heads/after-hole", None, Some(&facts.oids[1]))
        .unwrap();
    let hole_at = primary.txlog().head_seq();

    // A bare range source cannot close the hole: the twin stops, as before.
    let bare = Follower::new(
        Arc::clone(&twin),
        Box::new(Retained {
            first: hole_at,
            head: hole_at,
        }),
    );
    assert!(matches!(
        bare.step().unwrap(),
        Progress::SnapshotNeeded { .. }
    ));

    // A source that can snapshot closes it: the whole state at the primary's head, installed.
    // The source produces the snapshot, so its packs must be indexed first — see
    // [`wait_until_indexed`].
    wait_until_indexed(
        &primary,
        "before the twin fetches a snapshot from the primary",
    );
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: hole_at,
        }),
    );
    assert_eq!(
        f.step().unwrap(),
        Progress::SnapshotInstalled { at: hole_at }
    );
    assert_eq!(
        (twin.txlog().first_seq(), twin.txlog().head_seq()),
        (hole_at + 1, hole_at),
        "the twin's log resumes after the snapshot"
    );
    assert!(!snapshot::installing(td.path()));
    // Store-only (ruling 2026-09-22): the twin installed the snapshot's control plane and KEPT
    // the rest as bytes — no store, nothing derived — and a main recharged from it is the
    // primary.
    assert!(
        derived_on(twin.root()).is_empty(),
        "a store-only twin installs no repository from a snapshot"
    );
    assert_eq!(
        snapshot::base_at(twin.root()).unwrap(),
        Some(hole_at),
        "the twin keeps the snapshot as its base"
    );
    assert!(twin.catalog().get(&doomed_ref).unwrap().is_none());
    assert!(
        twin.auth().principal("bob").unwrap().is_some(),
        "a principal added inside the hole is there"
    );
    assert!(
        twin.raw_secrets().get(SECRET_KEY).unwrap().is_none(),
        "a secret deleted inside the hole is gone"
    );
    {
        let (_d, main) = recharged(&twin);
        same_state(
            &primary,
            &main,
            "main recharged from the twin after the snapshot",
        );
        assert!(
            !main.repo_dir(&doomed_ref).unwrap().exists(),
            "a store purged inside the hole is gone"
        );
        assert!(
            refs_of(&main).iter().all(|(n, _, _)| n != "refs/tags/v1"),
            "a tag deleted inside the hole is gone"
        );
    }

    // … and then it follows.
    app_store
        .serve()
        .update_ref("refs/heads/after-snapshot", None, Some(&facts.oids[0]))
        .unwrap();
    primary
        .secrets()
        .unwrap()
        .put(SECRET_KEY, b"again")
        .unwrap();
    assert!(matches!(
        f.step().unwrap(),
        Progress::Applied { count: 2, .. }
    ));
    assert_eq!(twin.txlog().head_seq(), primary.txlog().head_seq());
    assert!(
        derived_on(twin.root()).is_empty(),
        "still nothing derived on the twin"
    );

    // A restore from that twin — whose log no longer starts at 1 — closes the same hole with the
    // twin's KEPT snapshot, then replays its log.
    let head = primary.txlog().head_seq();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(
        restore_from(
            Box::new(LocalSource {
                stores: Arc::clone(&twin),
                retained_from: 0
            }),
            Arc::clone(&fresh),
            ""
        )
        .unwrap(),
        head
    );
    same_state(&primary, &fresh, "restored from a snapshotted twin");
    assert!(!fresh.repo_dir(&doomed_ref).unwrap().exists());
    drop(fresh);
    let fresh = ServerStores::open(fd.path()).unwrap();
    assert_eq!(
        fresh.secrets().unwrap().get(SECRET_KEY).unwrap().as_deref(),
        Some(&b"again"[..])
    );
    row("a twin behind a retention hole catches up via a snapshot and then follows; a restore from it closes the hole the same way", true, &format!("snapshot at {hole_at}, head {head}"));
}

/// ★ **A named sealed slot rides every way a secret does** (F4, the owner 2026-10-09): the
/// twin copies it sealed, a snapshot carries it, and a main recharged from the twin has it back
/// — owner and value — re-sealed under its own KEK. No slot code of the twin's own: the slot is a
/// `slot/<name>` secret written through the sealed, logged store.
#[cfg(feature = "metered")]
#[test]
fn a_sealed_slot_rides_the_twin_a_snapshot_and_a_recharge() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let sd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let material = b"ca.key, installer.cert, link.toml, ledger.toml".to_vec();
    assert!(
        primary
            .sealed_slot_put("installer-material", "front-certify", &material)
            .unwrap(),
        "created"
    );
    assert!(
        !primary
            .sealed_slot_put("installer-material", "front-certify", &material)
            .unwrap(),
        "replaced, not created"
    );
    assert_eq!(
        primary.sealed_slot_get("installer-material").unwrap(),
        Some(("front-certify".to_string(), material.clone()))
    );
    // Sealed at rest: the raw row is not the value, and the KEK opens it only under its own key.
    let raw = primary
        .raw_secrets()
        .get("slot/installer-material")
        .unwrap()
        .expect("the row");
    assert!(
        !raw.windows(material.len())
            .any(|w| w == material.as_slice()),
        "the row is sealed"
    );
    let frames = primary
        .txlog()
        .range(1, primary.txlog().head_seq())
        .unwrap();
    assert!(frames.iter().any(|e| matches!(e.meta(), Ok(Meta::Secret { ref key, .. }) if key == "slot/installer-material")), "a Secret frame");

    // The twin copies it, sealed.
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert_eq!(
        twin.raw_secrets().get("slot/installer-material").unwrap(),
        Some(raw.clone()),
        "the twin holds the sealed row"
    );

    // A snapshot carries it: a twin that catches up through a snapshot holds it too.
    wait_until_indexed(&primary, "before the snapshot");
    let snap_twin = open_twin(sd.path());
    let at = primary.txlog().head_seq();
    let f = Follower::new(
        Arc::clone(&snap_twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: at,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
    assert!(
        snap_twin
            .raw_secrets()
            .get("slot/installer-material")
            .unwrap()
            .is_some(),
        "the snapshot carried the slot"
    );

    // A recharge restores it: main recharged from the twin reads it back, re-sealed under its own KEK.
    drop(primary);
    let (fd, main) = recharged(&twin);
    drop(main);
    let main = ServerStores::open(fd.path()).unwrap();
    assert_eq!(
        main.sealed_slot_get("installer-material").unwrap(),
        Some(("front-certify".to_string(), material.clone())),
        "the recharged main has the slot"
    );
    assert_ne!(
        main.raw_secrets().get("slot/installer-material").unwrap(),
        Some(raw),
        "re-sealed under the recharged box's own KEK"
    );
    let (_sd2, from_snap) = recharged(&snap_twin);
    drop(from_snap);
    row("a named sealed slot rides the twin (sealed), a snapshot and a recharge (owner and value, re-sealed)", true, "slot installer-material");
}

/// ★ **S9: a twin answers `Sealed.Get` for the primary's slot** (F4, 2026-10-10). A sold front
/// that lost its own disk and its main reads the installer material from the OTHER box, which
/// may be the twin. The twin opened with a KEK of its own before it ever followed; the
/// primary's KEK arrives as the log's first frame. The running twin must open the primary's
/// sealed row from that moment on — not only after a restart — and keep doing so after one,
/// and a later replace on the primary must be what it answers.
#[cfg(feature = "metered")]
#[test]
fn a_running_twin_answers_a_sealed_slot_get_with_the_primarys_value() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let material = b"ca.key, installer.cert, link.toml, ledger.toml".to_vec();
    primary
        .sealed_slot_put("installer-material", "front-certify", &material)
        .unwrap();

    let twin = open_twin(td.path());
    let own = twin.kek_bytes();
    assert_ne!(
        own,
        primary.kek_bytes(),
        "the twin opened under a KEK of its own"
    );
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert_eq!(
        twin.kek_bytes(),
        primary.kek_bytes(),
        "the running twin holds the primary's KEK once it is framed"
    );
    assert_eq!(
        twin.sealed_slot_get("installer-material").unwrap(),
        Some(("front-certify".to_string(), material.clone())),
        "the running twin answers the slot"
    );

    let renewed = b"renewed ca.key, installer.cert, link.toml, ledger.toml".to_vec();
    primary
        .sealed_slot_put("installer-material", "front-certify", &renewed)
        .unwrap();
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert_eq!(
        twin.sealed_slot_get("installer-material").unwrap(),
        Some(("front-certify".to_string(), renewed.clone())),
        "the replace"
    );

    drop(twin);
    let twin = Arc::new(ServerStores::open(td.path()).unwrap());
    assert!(twin.is_twin());
    assert_eq!(
        twin.sealed_slot_get("installer-material").unwrap(),
        Some(("front-certify".to_string(), renewed)),
        "after a restart"
    );
    row("S9: a running twin answers Sealed.Get for the primary's slot, from the KEK frame on and after a restart", true, "slot installer-material");
}

// ── AUDIT-TWIN 2026-09-16 ───────────────────────────────────────────────────

/// **`Mirror.LogFetch`'s byte ceiling does not bound a range of empty-blob entries.**
///
/// `max_bytes` counts `e.blob.len()` and nothing else, but the majority of a busy forge's
/// frames carry NO blob at all — every `Refs`, `ControlDoc`, `LfsDelete` and `RepoStore` is
/// meta only. So the `bytes >= cap` break is never reached and [`fetch_range`] reads the whole
/// requested range into one `Vec<Entry>`, whatever the caller asked for; the gRPC layer then
/// builds a second whole copy of it (`service::mirror::chunks_of`). A twin catching up over a
/// ref-heavy history — or any member of group `twin` asking `LogFetch{from: 1, to: 0,
/// max_bytes: 1}` — makes the PRIMARY allocate its entire log.
#[test]
fn a_fetch_of_entries_that_carry_no_blob_is_still_bounded_by_the_byte_ceiling() {
    let dir = tempfile::tempdir().unwrap();
    let log = Log::open(dir.path()).unwrap();
    // 2 000 ref frames, no blob between them — the shape of a mirror repository's history.
    let n = 2_000u64;
    for i in 0..n {
        let edit = RefEdit {
            name: format!("refs/heads/b{i:06}"),
            old: None,
            new: Some("00".repeat(32)),
            peeled: None,
            symref: None,
        };
        log.append(
            Kind::Refs,
            &Meta::Refs {
                repo: REPO.into(),
                atomic: false,
                edits: vec![edit],
            },
            &[],
        )
        .unwrap();
    }
    assert_eq!(log.head_seq(), n);

    // One byte asked for. The rule is "whole entries, never fewer than one" — not "all of them".
    let got = fetch_range(&log, 1, 0, 1).unwrap();
    assert_eq!(
        got.len(),
        1,
        "a one-byte ceiling bought {} of {n} meta-only entries",
        got.len()
    );

    // And a real ceiling stops within a few entries of it, counting what actually crosses the
    // wire: a `Refs` meta is the whole transaction when the blob is empty.
    let meta_len = log.read(1).unwrap().meta.len() as u64;
    let got = fetch_range(&log, 1, 0, meta_len * 10).unwrap();
    assert!(
        got.len() <= 11,
        "a {}-byte ceiling bought {} of {n} meta-only entries",
        meta_len * 10,
        got.len()
    );
    row("Mirror.LogFetch bounds a range of blob-less entries by the byte ceiling the caller asked for", true, &format!("{n} meta-only entries, meta {meta_len} B"));
}

/// **A restore that met a frame this build cannot apply finishes as if it had applied it.**
///
/// `Kind::from_u16` answers `None` for a kind a newer primary wrote, and [`apply`] then returns
/// [`Applied::Unknown`] — right for a FOLLOWING twin (its log keeps the frame byte for byte, so
/// nothing is lost), and wrong for a RESTORE: the fresh box stops being a follower at
/// [`restore_finish`] and starts ORIGINATING from that head, with its stores missing whatever
/// those frames carried and `RestoreResponse` saying "restored to seq N (N applied, 0 lost)".
///
/// The scenario is the one the twin exists for: the primary runs a newer build, it dies, and
/// the operator restores from the twin onto the ISO he has on a stick — an older one.
#[test]
fn a_restore_that_meets_a_kind_this_build_cannot_apply_refuses_to_finish_by_name() {
    let pd = tempfile::tempdir().unwrap();
    let sd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());

    // The twin's log: the primary's, plus one frame of a kind no build in this tree knows.
    let src = Arc::new(Log::open(sd.path()).unwrap());
    let plog = primary.txlog();
    for e in plog.range(plog.first_seq(), plog.head_seq()).unwrap() {
        src.append_at(e.seq, e.kind, &e.meta, &e.blob).unwrap();
    }
    let newer = plog.head_seq() + 1;
    src.append_at(newer, 99, b"{\"Future\":{}}", b"what a later build wrote")
        .unwrap();

    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let err = restore(Arc::clone(&src), Arc::clone(&fresh), "").unwrap_err();
    let say = format!("{err:#}");
    assert!(
        say.contains("99"),
        "the refusal names the kind it could not apply: {say}"
    );
    assert!(
        is_restoring(fd.path()),
        "a box that could not apply every frame stays restoring"
    );
    // Durable first, so a re-run under a build that KNOWS kind 99 finishes this restore rather
    // than starting it over: the frame is the box's head, byte for byte, and `restore_begin`
    // re-applies the head before it takes anything newer.
    let log = fresh.txlog();
    assert_eq!(
        log.head_seq(),
        newer,
        "the frame that stopped the restore is the box's head"
    );
    let held = log.read(newer).unwrap();
    assert_eq!(
        (held.kind, held.blob.as_slice()),
        (99u16, b"what a later build wrote".as_slice())
    );
    // And the refusal repeats — every attempt with this build is told the same thing by name.
    let again = restore_begin(&fresh).unwrap_err();
    assert!(format!("{again:#}").contains("99"), "{again:#}");
    row(
        "a restore that meets a frame kind this build cannot apply refuses to finish, by name",
        true,
        &format!("kind 99 at seq {newer}"),
    );
}

/// **A snapshot item of a kind this build cannot install is dropped, silently and for ever.**
///
/// The following path may keep an unknown frame and go on, because the twin's log still holds
/// it. A SNAPSHOT item is in no log at all: its `seq` is 0, `restart_after(at)` moves the twin's
/// log past everything the snapshot stands for, and the only copy of the bytes is a staging file
/// [`snapshot::install`] deletes on its way out. `install` matched `(kind, _)` and threw
/// `apply_raw`'s `Applied::Unknown` away — so a twin taking a snapshot from a newer primary
/// reported itself caught up while a whole class of its state had been dropped, and a restore
/// from that twin could never get it back.
#[test]
fn a_snapshot_item_of_an_unknown_kind_stops_the_install_instead_of_being_dropped() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());

    // The primary's own snapshot, with one item a newer build would have written in it —
    // injected just before the `SnapshotAt` terminator, where a real newer producer puts its own.
    let mut stager = snapshot::Stager::begin(td.path()).unwrap();
    let mut injected = false;
    wait_until_indexed(&primary, "before producing a snapshot");
    snapshot::produce(&primary, &mut |p| {
        if p.kind == Kind::SnapshotAt as u16 && !injected {
            injected = true;
            stager.feed(snapshot::Piece {
                kind: 99,
                meta: b"{\"Future\":{}}",
                blob_len: 4,
                blob: b"data",
                last: true,
            })?;
        }
        stager.feed(p)
    })
    .unwrap();
    assert!(
        injected,
        "the producer emitted a SnapshotAt terminator to inject in front of"
    );
    let staged = stager.finish().unwrap();

    let err = snapshot::install(&twin, &staged).unwrap_err();
    let say = format!("{err:#}");
    assert!(
        say.contains("99"),
        "the refusal names the kind it could not install: {say}"
    );
    assert!(
        snapshot::installing(td.path()),
        "the install marker stays, so this half-installed twin asks for a whole snapshot again \
         instead of following on from a state it cannot vouch for"
    );
    row(
        "a snapshot item of a kind this build cannot install stops the install, by name",
        true,
        "kind 99 before the terminator",
    );
}

/// **The other half of the stall deadline: a sink that gives up gives the primary back.**
///
/// [`snapshot::produce`] holds `Log::freeze()` for the whole stream, so every framed write on
/// the primary — every push, every ref update, every control-plane mutation — is stopped while
/// a snapshot is on the wire. `Mirror.SnapshotFetch` cuts a twin off that stops taking pieces
/// (`service::mirror::SNAPSHOT_STALL_TIMEOUT`); this is the guarantee that doing so is worth
/// anything: the refusal comes back out of `produce` — never swallowed by `emit_pack`'s
/// `io::Error` round trip — and the freeze goes with it.
#[test]
fn a_sink_that_refuses_a_piece_ends_the_snapshot_and_the_primary_writes_again_at_once() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let before = primary.txlog().head_seq();

    let mut seen = 0usize;
    wait_until_indexed(&primary, "before producing a snapshot");
    let err = snapshot::produce(&primary, &mut |_p| {
        seen += 1;
        if seen == 2 {
            anyhow::bail!("the twin took no snapshot piece for 120 s");
        }
        Ok(())
    })
    .unwrap_err();
    assert!(
        format!("{err:#}").contains("took no snapshot piece"),
        "the sink's own words survive: {err:#}"
    );

    // The freeze is gone with the producer: the next framed write goes through, here and now.
    let (pack, _oid) = one_blob_pack(
        gix_hash::Kind::Sha256,
        b"after the snapshot was abandoned\n",
    );
    let seq = primary
        .txlog()
        .append(
            Kind::Pack,
            &Meta::Pack {
                repo: REPO.into(),
                hash: "sha256".into(),
            },
            &pack,
        )
        .expect("the log is writable the moment the snapshot gives up");
    assert!(seq > before);
    row(
        "a snapshot whose sink refuses a piece ends there and the primary's writes resume",
        true,
        &format!("refused at piece {seen}"),
    );
}

/// **A framed write during a slow snapshot is not held for the stream** (R5.1-A, 2026-10-07).
///
/// [`snapshot::produce`] held `Log::freeze()` for the WHOLE stream: every framed mutation on the
/// primary — every push, every control write, every audited RPC — waited in the gate until the
/// twin had received the last piece, which on a real estate is every repository's pack. The
/// freeze is for a consistent CUT (the head, the control database, the framed files, every
/// repository's refs); the bytes can stream after it, unfrozen, from the cut.
///
/// A sink that takes 40 ms per piece makes the stream last seconds; once its first piece is out,
/// another thread creates a namespace (control frames and a document move). That write must be
/// back in under a second — and the snapshot must still be the state at its `at`: its document
/// pointers are exactly the primary's from before the write, and `at` is below the write's frames.
#[test]
fn a_framed_write_during_a_slow_snapshot_is_not_held_for_the_stream() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    wait_until_indexed(&primary, "before producing a snapshot");
    let docs_before: std::collections::BTreeMap<String, String> = primary
        .control_db()
        .export()
        .unwrap()
        .docs
        .iter()
        .map(|(n, id)| (n.clone(), hex::encode(id.as_bytes())))
        .collect();

    let (started_tx, started_rx) = std::sync::mpsc::channel::<()>();
    let writer = {
        let primary = Arc::clone(&primary);
        std::thread::spawn(move || {
            started_rx.recv().unwrap();
            let t = std::time::Instant::now();
            primary
                .catalog()
                .create_namespace("written-during-the-snapshot")
                .expect("the write goes through");
            (t.elapsed(), primary.txlog().head_seq())
        })
    };
    let mut docs_in_snapshot: std::collections::BTreeMap<String, String> = Default::default();
    let mut pieces = 0u64;
    let t0 = std::time::Instant::now();
    let at = snapshot::produce(&primary, &mut |p| {
        pieces += 1;
        if pieces == 1 {
            let _ = started_tx.send(());
        }
        if p.kind == Kind::ControlDoc as u16 && !p.meta.is_empty() {
            if let Ok(Meta::ControlDoc { name, new, .. }) = Meta::from_bytes(p.meta) {
                docs_in_snapshot.insert(name, new);
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(40));
        Ok(())
    })
    .unwrap();
    let streamed = t0.elapsed();
    let (write_took, head_after) = writer.join().unwrap();
    let line = format!("{pieces} pieces streamed in {streamed:?}; the write during it took {write_took:?}; at {at}, head after {head_after}");
    eprintln!("{line}");
    assert!(
        streamed >= std::time::Duration::from_millis(1500),
        "the stream is slow enough to tell: {line}"
    );
    assert!(
        write_took < std::time::Duration::from_secs(1),
        "a framed write waited for the snapshot's stream: {line}"
    );
    assert!(
        at < head_after,
        "the write's frames come after the cut: {line}"
    );
    assert_eq!(
        docs_in_snapshot, docs_before,
        "the snapshot is the state at its cut, not a mix with the write: {line}"
    );
    row(
        "a framed write during a slow snapshot is not held for the stream",
        true,
        &line,
    );
}

/// **A restore stream that broke mid-snapshot leaves its half in the next attempt's snapshot.**
///
/// `Bootstrap.Restore` stages snapshot pieces in a process-global `RESTORE_STAGER`
/// ([`snapshot::restore_piece`]). The slot is only ever emptied by an error inside `feed` or by
/// the terminator — so when the STREAM breaks (the twin's fetch dies, the network drops, the
/// sender is killed) the half that arrived stays in the slot, and the resent stream's pieces are
/// appended to it. `install` then runs over the union of two snapshots.
///
/// It is not loud, because the union is mostly the same data twice. It is wrong where the two
/// differ: `install` builds `carried` and `named` from every item it is holding, so a framed
/// file (or a secret) that the SOURCE deleted between the attempts is "carried" by the stale
/// half — kept, with its stale bytes, on a box whose whole contract is that it holds exactly
/// what the source holds. "Installed whole, not merged" is the rule, and a leftover stager
/// merges.
#[test]
fn a_snapshot_stream_that_broke_does_not_leave_its_half_in_the_next_ones_install() {
    let pd = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(restore_begin(&fresh).unwrap(), 1);

    // Attempt 1: the sender dies after the `webhooks.json` item and before the terminator.
    let mut broke = false;
    wait_until_indexed(&primary, "before producing a snapshot");
    let cut = snapshot::produce(&primary, &mut |p| {
        if broke {
            return Ok(());
        }
        snapshot::restore_piece(&fresh, p)?;
        if matches!(Meta::from_bytes(p.meta), Ok(Meta::File { ref path, .. }) if path == "webhooks.json")
            && p.last
        {
            broke = true;
        }
        Ok(())
    });
    assert!(
        cut.is_ok() && broke,
        "the producer reached webhooks.json before the stream was cut"
    );

    // The source no longer has that file at all.
    std::fs::remove_file(primary.root().join("webhooks.json")).unwrap();

    // Attempt 2: the operator resends. A whole snapshot, of a source without webhooks.json.
    assert_eq!(restore_begin(&fresh).unwrap(), 1);
    let mut at = None;
    wait_until_indexed(&primary, "before producing a snapshot");
    snapshot::produce(&primary, &mut |p| {
        if let Some(seen) = snapshot::restore_piece(&fresh, p)? {
            at = Some(seen);
        }
        Ok(())
    })
    .unwrap();
    assert!(
        at.is_some(),
        "the second stream's terminator installed a snapshot"
    );

    assert!(
        !fd.path().join("webhooks.json").exists(),
        "the restored box kept a framed file the source deleted — the broken stream's half was \
         installed beside the whole one"
    );
    row(
        "a snapshot stream that broke leaves nothing of itself in the next attempt's install",
        true,
        "webhooks.json, deleted between the attempts",
    );
}

/// **A crash mid-install must not be followed on from, whatever the source can do.**
///
/// `Follower::step` reads the install marker, tries a snapshot to repair the half-installed
/// stores, and then — when the source answered `Ok(false)`, "I cannot make one" — falls through
/// to the ordinary fetch-and-apply loop. The `snapshot-needed` refusal below it is guarded by
/// the RETENTION test (`first > wanted`), which is false whenever the source retains everything.
/// So a source that cannot snapshot and has no hole let the twin append and apply entries onto
/// stores that `install` had already wiped, with the marker still on the disk and nothing said.
#[test]
fn a_twin_that_crashed_mid_install_refuses_to_follow_on_from_half_a_snapshot() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());

    // The state a power cut inside `install` leaves: the marker, and stores that are half the
    // snapshot. The source is a bare log — it retains everything and cannot make a snapshot.
    crate::durable::write_atomic_unframed(td.path(), snapshot::INSTALLING, b"7\n").unwrap();
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));

    let err = f.step().unwrap_err();
    let say = format!("{err:#}");
    assert!(
        say.contains("snapshot"),
        "the refusal names the half-installed snapshot: {say}"
    );
    assert_eq!(
        twin.txlog().head_seq(),
        0,
        "and not one entry was applied on top of it"
    );
    assert!(
        snapshot::installing(td.path()),
        "the marker stays until a whole snapshot replaces it"
    );
    row(
        "a twin that crashed mid-install refuses to follow on from half a snapshot",
        true,
        "a source that cannot snapshot, no retention hole",
    );
}
/// **A fresh box that held a secret of its OWN before the restore** (lane TWIN-COMPLETENESS).
///
/// MEASURED: this bricked the box. The fresh box's secrets are sealed under the KEK it minted
/// at first open; the replay's `Kek` frame replaces that key on disk, and a row the replay had
/// no frame for then opened under NEITHER key. `reseal_after_restore` met it with a `?` and the
/// restored box REFUSED TO OPEN — every time, for ever — with
/// *"opening fresh-only/key with the replayed KEK: … wrong key-encrypting key"*. A box that had
/// been restored successfully and could not start.
///
/// Both halves are asserted here: `prepare_replay` drops the fresh box's own secrets with its
/// own numbers, so the row never reaches the re-seal; and the re-seal drops such a row loudly
/// instead of refusing, so a box restored by an older build still opens.
#[test]
fn a_fresh_boxs_own_secret_does_not_brick_the_box_it_is_restored_into() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let host_fp = gunnar_core::ensure_host_key(&primary.secrets().unwrap())
        .unwrap()
        .fingerprint()
        .to_string();
    drop(primary);

    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    // Two kinds at once: one the replay WILL overwrite (the host key this box minted at its own
    // first open) and one it never will.
    gunnar_core::ensure_host_key(&fresh.secrets().unwrap()).unwrap();
    fresh
        .secrets()
        .unwrap()
        .put("fresh-only/key", b"the fresh box's own")
        .unwrap();
    assert!(fresh.raw_secrets().get("fresh-only/key").unwrap().is_some());

    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    // Given up with the numbers: the replay wrote the primary's rows into an empty table.
    assert_eq!(
        fresh.raw_secrets().get("fresh-only/key").unwrap(),
        None,
        "the fresh box's own secret went with its own numbers"
    );
    drop(fresh);

    // The restart re-seals — and OPENS.
    let back =
        ServerStores::open(fd.path()).expect("a restored box opens; it used to refuse for ever");
    assert_eq!(
        gunnar_core::ensure_host_key(&back.secrets().unwrap())
            .unwrap()
            .fingerprint()
            .to_string(),
        host_fp,
        "the primary's host key, not the fresh box's"
    );
    assert_eq!(back.raw_secrets().get("fresh-only/key").unwrap(), None);

    // The other half: a box whose fresh row reached the re-seal anyway (an older build's
    // restore) drops it and opens, rather than bricking.
    let gd = tempfile::tempdir().unwrap();
    let again = Arc::new(ServerStores::open(gd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&again), "").unwrap();
    // Put it back AFTER the replay, sealed under a key that is now gone: exactly the shape
    // `prepare_replay` used to leave behind.
    let dead = gunnar_core::Kek::generate().unwrap();
    again
        .raw_secrets()
        .put(
            "stranded/key",
            &gunnar_core::seal(&dead, "stranded/key", b"unreadable").unwrap(),
        )
        .unwrap();
    drop(again);
    let opened = ServerStores::open(gd.path()).expect("a stranded row is dropped, not a brick");
    assert_eq!(
        opened.raw_secrets().get("stranded/key").unwrap(),
        None,
        "the unopenable row is dropped by name"
    );
    assert_eq!(
        gunnar_core::ensure_host_key(&opened.secrets().unwrap())
            .unwrap()
            .fingerprint()
            .to_string(),
        host_fp,
        "and everything the twin's log carried is still there"
    );

    row("a fresh box's own secret does not brick the box it is restored into: given up with its numbers, and dropped by name if it reaches the re-seal anyway", true, &format!("host key {host_fp}"));
}

/// ★ **STINK S1 — a follower that cannot start is started again, and says so.**
///
/// The pre-loop failures (`twin.json`, `twin.key`, the channel) used to kill
/// the follower thread for the life of the process. Nothing retried it, and
/// nothing said it had happened: `status()` stayed `None`, so the twin's state
/// word on the heartbeat was the empty string and the only visible number was
/// a lag that never moved. Because `flushing` will not unmount a data set whose
/// twin is behind, that one dead thread makes EVERY growth the estate will ever
/// order fail — while the shop keeps selling.
///
/// Measured here with the failure an operator actually hits: a box told it is a
/// twin whose `twin.key` is not there. Before the supervisor this asserted at
/// the first line; now the attempt count climbs and the status is RED with the
/// reason in it.
#[test]
fn a_follower_that_cannot_start_is_started_again_and_the_status_says_it_is_not_running() {
    let dir = tempfile::tempdir().unwrap();
    let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
    // A twin by configuration, with the credential missing: `run()` fails
    // before it ever polls, every time.
    TwinConfig {
        primary: "127.0.0.1:1".into(),
        joined_at_unix_ms: now_ms(),
    }
    .save(dir.path())
    .unwrap();
    assert!(!dir.path().join(TWIN_KEY).exists());

    start_follower_every(Arc::clone(&stores), Duration::from_millis(10));
    // A28 M18: the follower is a duty, listed while it runs (a panic in it is said, and it runs on).
    assert!(
        nordisk_pairing::duty::find(TWIN_FOLLOWER).is_some(),
        "the follower is a listed duty"
    );
    let started = std::time::Instant::now();
    let mut seen = None;
    while started.elapsed() < Duration::from_secs(10) {
        if let Some(s) = status(stores.root()) {
            if s.why.contains("attempt 3") {
                seen = Some(s);
                break;
            }
        }
        std::thread::sleep(Duration::from_millis(5));
    }
    stop_follower();

    let s = seen.unwrap_or_else(|| {
        panic!(
            "the follower never reached a third attempt; status: {:?}",
            status(stores.root())
        )
    });
    assert_eq!(
        s.state,
        State::Unreachable,
        "a follower that is not running is RED, not a lag number: {s:?}"
    );
    assert!(s.why.contains("is NOT running"), "{}", s.why);
    assert!(
        s.why.contains(TWIN_KEY),
        "the status names what stopped it: {}",
        s.why
    );
    // The state word is what the heartbeat carries to the primary, and
    // `heartbeat_lines` prints anything that is not `ok` in upper case.
    assert_eq!(measure_report(&stores).state, "unreachable");
    row("a follower that cannot start is supervised: retried for ever, and RED in the status with the reason (STINK S1)", true, &s.why);
}

/// ★ **A primary that was RE-IMAGED reads as a healthy pairing from both ends.**
///
/// The twin is the replica for the worst case: when main dies it is what main
/// is fed from. So the pairing going quiet is not a degraded twin — it is a
/// lie, and the restore that leans on it fails at the worst possible moment.
///
/// This is that lie in its purest form, and it is a thing this estate does: a
/// main is re-imaged (or the address is re-pointed at another set), so the box
/// at the other end answers a head far BELOW what the twin has applied.
///
/// MEASURED on `agent/stink-leaks` before the lineage check, with exactly the
/// numbers below:
///
/// * `step()` returned `Ok(Progress::CaughtUp { head })` — `wanted <=
///   primary_head` is false, so the snapshot branch, the snapshot-needed
///   refusal and the fetch loop were all skipped, and the poll fell out of the
///   bottom of `step` setting `State::Following` with an EMPTY `why`.
/// * On the primary's side the lag is `head_seq.saturating_sub(applied)`, which
///   SATURATES TO 0 — so `Registry::behind()` returns `None` and the heartbeat
///   reads `ok`.
///
/// Both ends green, for ever, while not one entry would ever be applied again
/// and the twin's contents belong to a box that no longer exists.
///
/// To see it fail before the fix, neutralise the lineage check in
/// `Follower::step` (`if false && !installing && primary_head < own`): the
/// first assertion below reports `CaughtUp` instead of `Diverged`.
#[test]
fn a_primary_whose_head_is_below_the_twins_is_a_different_log_and_the_twin_stops_instead_of_reading_ok(
) {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "the twin is level with the primary it grew up with"
    );

    // The primary is re-imaged: same address, a fresh log, three things told to
    // it since. Nothing about the wire changes — it answers `LogHead` happily.
    let reimaged = Follower::new(Arc::clone(&twin), Box::new(Retained { first: 1, head: 3 }));
    let p = reimaged.step().unwrap();
    assert_eq!(
        p,
        Progress::Diverged { own: head, primary_head: 3 },
        "a head below this twin's is a DIFFERENT log; before the lineage check this was CaughtUp {{ head: {head} }}"
    );
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "nothing is applied on top of a foreign log"
    );

    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Diverged, "{s:?}");
    assert!(
        !s.caught_up(),
        "a diverged twin is never `caught_up`: {s:?}"
    );
    assert!(
        s.why.contains(&"head 3".to_string()) && s.why.contains(&head.to_string()),
        "the refusal carries BOTH numbers: {}",
        s.why
    );
    assert_eq!(
        measure_report(&twin).state,
        "diverged",
        "the word the heartbeat carries to the primary"
    );
    // And the state word is not `ok`-shaped, so `heartbeat_lines`' `loud()`
    // prints it in upper case.
    assert_ne!(State::Diverged.to_string(), "ok");

    row("a primary whose head is below the twin's is a different log: the twin stops with `diverged` instead of reporting a healthy pairing for ever", true, &format!("applied {head}, the box at the other end answers 3"));
}

/// ★ **The stop stays stopped.** `Progress::Diverged` must come out of
/// `Follower::run` the way `SnapshotNeeded` does — an operator decides whether
/// to re-join, and re-running the same poll would write the same refusal every
/// two seconds for ever.
///
/// Neutralise by deleting the `Progress::Diverged` arm of `run`'s match (it
/// falls into `Ok(_) => {}`), or by neutralising the lineage check in `step`:
/// the loop then polls for ever and this fails on the deadline below. It is run
/// on a thread precisely so the neutralised form FAILS rather than hanging the
/// suite — a test that cannot go red is not a test.
#[test]
fn a_diverged_twin_stops_polling_rather_than_refuse_the_same_thing_for_ever() {
    /// Never set: the only thing that may end this loop is the divergence.
    static NEVER: AtomicBool = AtomicBool::new(false);

    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();

    let following = Arc::clone(&twin);
    let running = std::thread::spawn(move || {
        Follower::new(following, Box::new(Retained { first: 1, head: 3 }))
            .run(Duration::from_millis(20), &NEVER)
    });
    let started = std::time::Instant::now();
    while !running.is_finished() && started.elapsed() < Duration::from_secs(5) {
        std::thread::sleep(Duration::from_millis(20));
    }
    assert!(
        running.is_finished(),
        "run() is still polling a primary it has already refused as a different log; a stop that does not stop writes the same refusal for ever"
    );
    running.join().unwrap();
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "and nothing was applied while it went round"
    );
    row(
        "a diverged twin's loop returns instead of writing the same refusal every poll for ever",
        true,
        &format!("run() returned in {:?}", started.elapsed()),
    );
}

/// A source that answers `head` and then dies on the fetch — the twin's volume
/// filling, a pack that will not come across, a link that drops mid-range.
struct HeadOnly {
    head: u64,
}

impl LogSource for HeadOnly {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Ok((1, self.head))
    }
    fn fetch(&self, _from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
        Err(anyhow!("No space left on device (os error 28)"))
    }
}

/// ★ **A poll that reached the primary and then failed used to leave the status
/// saying the twin was current.**
///
/// Only the `head()` call set a status on failure. Every path after it — a
/// fetch that dies mid-range, `append_at` on a full volume, an `apply` that
/// will not go in — returned through `?`, and `Follower::run` logged one `warn`
/// and polled again. THE STATUS WAS NEVER TOUCHED, so it still said whatever
/// the last good poll left: `following`, `caught_up`, an empty `why` — and
/// `measure_report` carried `following` to the primary on the very next poll,
/// which succeeds, because the wire is fine.
///
/// That is the shape a twin's life actually ends in: its data volume fills. It
/// read as a healthy, current copy from both ends for as long as the box stayed
/// up.
///
/// MEASURED here with the twin one entry behind and the fetch answering ENOSPC.
///
/// To see it fail before the fix, delete the `if let (Err(err), false)` block at
/// the top of `Follower::step`: the state below is `Following` and `caught_up()`
/// is true, over a twin that just failed to copy anything.
#[test]
fn a_poll_that_failed_after_reaching_the_primary_no_longer_leaves_the_status_saying_current() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());
    // A good poll first: the status this test is about is the one it leaves.
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert!(
        status(twin.root()).unwrap().caught_up(),
        "level before the failure"
    );

    // The primary moves on; the twin can no longer take it.
    let f = Follower::new(Arc::clone(&twin), Box::new(HeadOnly { head: head + 5 }));
    let err = format!("{:#}", f.step().expect_err("the fetch fails"));
    assert!(err.contains("os error 28"), "{err}");

    let s = status(twin.root()).unwrap();
    assert_eq!(
        s.state,
        State::Stalled,
        "a twin that could not take what it was sent is RED, not `following`: {s:?}"
    );
    assert!(!s.caught_up(), "and never `caught_up`: {s:?}");
    assert!(
        s.why.contains("NOT a current copy") && s.why.contains("os error 28"),
        "the reason is carried, not only logged: {}",
        s.why
    );
    assert_eq!(
        measure_report(&twin).state,
        "stalled",
        "the word the heartbeat carries to the primary"
    );
    row("a poll that reached the primary and then failed says `stalled` in the status instead of leaving the last good poll's `following` standing", true, &s.why);
}

/// A source that answers head and a log identity of its choosing — a box at the
/// other end of an address, which is all a twin ever really has.
struct Named {
    first: u64,
    head: u64,
    id: Option<&'static str>,
}

impl LogSource for Named {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Ok((self.first, self.head))
    }
    fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
        Ok((self.first, self.head, self.id.map(str::to_owned)))
    }
    fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
        panic!("a twin that has refused its source must not fetch (asked for {from})")
    }
}

/// ★ **A RE-IMAGED MAIN THAT HAS ALREADY WRITTEN PAST THE TWIN'S HEAD.**
///
/// This is the one that destroys the only copy, and no number can see it.
///
/// `State::Diverged` on a lower head catches a re-imaged main caught EARLY. Give
/// that same main an afternoon — a fresh install at a familiar address, taking
/// pushes — and it passes the twin's head. From that moment every number agrees:
/// `first <= wanted`, `wanted <= primary_head`, the gap rule is satisfied, the
/// fetch loop runs. The twin appends a stranger's frames at its own next
/// sequence numbers, on top of the last surviving copy of its own history, and
/// reports `following`, `caught_up`, `ok`.
///
/// MEASURED here with a twin at head 9+ and a stranger at head 100 — which is
/// what `2. the numbers agree` below asserts, so the test states the trap as a
/// fact rather than describing it.
///
/// The fix is that a log now has a NAME (`twin::LOG_ID`) and the answer carries
/// it. To see this fail before, neutralise the identity check in
/// `Follower::step_inner` (`if let Some(theirs) = None::<&str> { … }`, or delete
/// the block): case 3 applies 100 entries from a box the twin has never met.
#[test]
fn a_primary_that_is_a_different_log_is_refused_even_when_every_number_agrees() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());
    // 1. The twin follows its real primary and learns which log it is.
    let real = mint_log_id(pd.path()).unwrap();
    Follower::new(
        Arc::clone(&twin),
        Box::new(LocalNamed {
            stores: Arc::clone(&primary),
            id: real.clone(),
        }),
    )
    .step()
    .unwrap();
    assert_eq!(twin.txlog().head_seq(), head);
    assert_eq!(
        log_id(td.path()).as_deref(),
        Some(real.as_str()),
        "the twin pinned its primary's log identity on the first poll"
    );

    // 2. The numbers agree — this is the trap, asserted, not described.
    let stranger = Named {
        first: 1,
        head: 100,
        id: Some("0000ffff0000ffff0000ffff0000ffff"),
    };
    let (first, their_head, _) = stranger.head_and_identity().unwrap();
    let wanted = twin.txlog().head_seq() + 1;
    assert!(first <= wanted && wanted <= their_head, "every arithmetic test the twin had passes: first {first} <= wanted {wanted} <= head {their_head}");

    // 3. And it is refused anyway, because it is not this log.
    let f = Follower::new(Arc::clone(&twin), Box::new(stranger));
    let p = f.step().unwrap();
    assert_eq!(
        p,
        Progress::Diverged {
            own: head,
            primary_head: 100
        },
        "a different log is refused however good its numbers look"
    );
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "not one frame of a stranger's history was appended"
    );

    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Diverged, "{s:?}");
    assert!(
        s.why.contains(&real) && s.why.contains("0000ffff0000ffff0000ffff0000ffff"),
        "the refusal names BOTH logs: {}",
        s.why
    );
    assert!(
        s.why.contains("restore a primary FROM this twin"),
        "and says what the way out is, which is never a retry: {}",
        s.why
    );
    assert_eq!(
        log_id(td.path()).as_deref(),
        Some(real.as_str()),
        "and the twin still knows which log it holds"
    );
    row("a primary that is a DIFFERENT LOG is refused even when every sequence number agrees — the re-imaged main that has already written past the twin's head", true, &s.why);
}

/// A source that answers with a log identity AND a generation: over a real
/// primary's log, or scripted numbers it must never be asked to fetch.
struct Generational {
    stores: Option<Arc<ServerStores>>,
    scripted: (u64, u64),
    id: String,
    generation: u64,
}

impl LogSource for Generational {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Ok(match &self.stores {
            Some(s) => (s.txlog().first_seq(), s.txlog().head_seq()),
            None => self.scripted,
        })
    }
    fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
        let (first, head) = self.head()?;
        Ok((first, head, Some(self.id.clone())))
    }
    fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
        let (first, head, id) = self.head_and_identity()?;
        Ok((first, head, id, self.generation))
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        let stores = self.stores.as_ref().unwrap_or_else(|| {
            panic!("a twin that has refused its source must not fetch (asked for {from})")
        });
        Ok(fetch_range(stores.txlog(), from, to, max_bytes)?)
    }
}

/// Stinky row 18 (2026-10-10): **the old main that comes back after a restore
/// is fenced off by its generation.**
///
/// A restored main KEEPS the log id and the host key (both on purpose, see
/// [`LOG_ID`]), so the OLD main, if it ever comes back, is the same name and the
/// same key — and once it has taken writes past the twin's head every number
/// agrees too. The twin spliced its history onto the restored one. A restore
/// now stamps a newer generation; the twin pins the highest it has followed and
/// refuses a lower one by name.
#[test]
fn an_old_main_that_comes_back_after_a_restore_is_fenced_off_by_its_generation() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let real = mint_log_id(pd.path()).unwrap();
    let head = primary.txlog().head_seq();
    let old_generation = log_generation(pd.path()).unwrap();

    let twin = open_twin(td.path());
    let following = |stores: &Arc<ServerStores>, generation| Generational {
        stores: Some(Arc::clone(stores)),
        scripted: (0, 0),
        id: real.clone(),
        generation,
    };
    Follower::new(
        Arc::clone(&twin),
        Box::new(following(&primary, old_generation)),
    )
    .step()
    .unwrap();
    assert_eq!(twin.txlog().head_seq(), head);

    // The main dies; a fresh box is restored from the twin.
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    restore_begin(&fresh).unwrap();
    for seq in 1..=head {
        restore_apply(&fresh, &twin.txlog().read(seq).unwrap()).unwrap();
    }
    restore_finish(&fresh, head, &real).unwrap();
    let restored = log_generation(fd.path()).unwrap();
    assert!(
        restored > old_generation,
        "a restore makes a newer generation of the same log: {restored} after {old_generation}"
    );

    // The twin follows the restored main, untouched, and pins its generation.
    let p = Follower::new(Arc::clone(&twin), Box::new(following(&fresh, restored)))
        .step()
        .unwrap();
    assert!(matches!(p, Progress::CaughtUp { .. }), "{p:?}");
    assert_eq!(pinned_generation(td.path()).unwrap(), restored);

    // The old main comes back: same id, written past the twin's head.
    let old = Generational {
        stores: None,
        scripted: (1, head + 50),
        id: real.clone(),
        generation: old_generation,
    };
    let p = Follower::new(Arc::clone(&twin), Box::new(old))
        .step()
        .unwrap();
    assert_eq!(
        p,
        Progress::Diverged {
            own: head,
            primary_head: head + 50
        },
        "the old main is refused however good its name and numbers are"
    );
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "not one of its frames was appended"
    );
    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Diverged, "{s:?}");
    assert!(
        s.why.contains("generation"),
        "the refusal names the generations: {}",
        s.why
    );
    assert_eq!(
        pinned_generation(td.path()).unwrap(),
        restored,
        "and the pin does not move back"
    );
}

/// Stinky row 18: the generation is inside the primary's signature, so an
/// answer cannot be edited to look older or newer than its box.
#[test]
fn a_primarys_generation_is_inside_its_signature() {
    let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
        .unwrap();
    let line = key.public_key().to_openssh().unwrap();
    let id = "1234abcd1234abcd1234abcd1234abcd";
    let sig = gunnar_grpc::auth::sign_twin_answer_at(&key, "n1", id, 1, 9, 7).unwrap();
    assert!(gunnar_grpc::auth::verify_twin_answer_at(
        &line, &sig, "n1", id, 1, 9, 7
    ));
    assert!(
        !gunnar_grpc::auth::verify_twin_answer_at(&line, &sig, "n1", id, 1, 9, 8),
        "a generation edited in flight"
    );
    assert!(
        !gunnar_grpc::auth::verify_twin_answer_at(&line, &sig, "n1", id, 1, 9, 0),
        "or dropped"
    );
    let v1 = gunnar_grpc::auth::sign_twin_answer(&key, "n1", id, 1, 9).unwrap();
    assert!(
        gunnar_grpc::auth::verify_twin_answer_at(&line, &v1, "n1", id, 1, 9, 0),
        "generation 0 is the answer every older primary signs"
    );
}

/// A restored main's signed `LogHead` answer at `generation`, made the way the
/// restored box makes it: with the host key the restore replayed, which is the
/// key `of` answers with too.
fn restored_answer(
    of: &ServerStores,
    log_id: &str,
    generation: u64,
) -> gunnar_proto::LogHeadResponse {
    let (first_seq, head_seq) = (of.txlog().first_seq(), of.txlog().head_seq());
    let (answer_key, answer_sig) =
        sign_head_answer(of, "feedc0de", log_id, first_seq, head_seq, generation);
    assert!(
        !answer_sig.is_empty(),
        "the fixture's box signs its answers"
    );
    gunnar_proto::LogHeadResponse {
        log_id: log_id.to_owned(),
        first_seq,
        head_seq,
        generation,
        answer_key,
        answer_sig,
        ..Default::default()
    }
}

/// Stinky row 18, **the twin's half of the main-side fence** (2026-10-10): a
/// twin keeps the newest generation of its log it has followed, AS ITS MAIN
/// SIGNED IT, and relays that on every poll. Only a signed answer with a
/// generation is kept, and the kept one only ever moves up.
#[test]
fn a_twin_keeps_the_newest_signed_generation_it_has_followed_to_relay_it() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let real = mint_log_id(pd.path()).unwrap();
    assert_eq!(
        generation_proof(td.path()).unwrap(),
        None,
        "a twin that has never followed a restored main relays nothing"
    );

    let unsigned = gunnar_proto::LogHeadResponse {
        log_id: real.clone(),
        generation: 5,
        ..Default::default()
    };
    remember_generation_proof(td.path(), "n", &unsigned).unwrap();
    assert_eq!(
        generation_proof(td.path()).unwrap(),
        None,
        "an answer nobody signed proves nothing to anybody"
    );
    remember_generation_proof(td.path(), "feedc0de", &restored_answer(&primary, &real, 0)).unwrap();
    assert_eq!(
        generation_proof(td.path()).unwrap(),
        None,
        "generation 0 is a main never restored: nothing to fence with"
    );

    let newer = restored_answer(&primary, &real, 2_000);
    remember_generation_proof(td.path(), "feedc0de", &newer).unwrap();
    let kept = generation_proof(td.path())
        .unwrap()
        .expect("the restored main's answer is kept");
    assert_eq!(
        kept,
        gunnar_proto::GenerationProof {
            answer_key: newer.answer_key.clone(),
            answer_sig: newer.answer_sig.clone(),
            nonce: "feedc0de".into(),
            log_id: real.clone(),
            first_seq: newer.first_seq,
            head_seq: newer.head_seq,
            generation: 2_000,
        },
        "exactly the fields its signature covers, and the key that made it"
    );
    assert!(gunnar_grpc::auth::verify_twin_answer_at(
        &kept.answer_key,
        &kept.answer_sig,
        &kept.nonce,
        &kept.log_id,
        kept.first_seq,
        kept.head_seq,
        kept.generation
    ));

    remember_generation_proof(
        td.path(),
        "feedc0de",
        &restored_answer(&primary, &real, 1_000),
    )
    .unwrap();
    assert_eq!(
        generation_proof(td.path()).unwrap().map(|p| p.generation),
        Some(2_000),
        "an older generation never replaces a newer proof"
    );
}

/// Stinky row 18, **the main's half** (2026-10-10). Row 18 fenced the TWIN: it
/// refuses an old main come back after a restore. The old main itself went on
/// taking writes — split brain with every client that still reached it. Now a
/// box shown a proof of a NEWER generation of its OWN log, signed by its OWN host
/// key (only the restored main can make one: the restore replayed that key),
/// fences itself: every write is refused, `Mirror.LogFetch` is refused, an alarm
/// says so on `Admin.Health`, and the fence survives a restart. Anything that is
/// not such a proof is ignored.
#[test]
fn an_old_main_shown_a_newer_generation_of_its_own_log_fences_itself_for_good() {
    use gunnar_grpc::seam::ControlPlane as _;
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let real = mint_log_id(pd.path()).unwrap();
    let own = log_generation(pd.path()).unwrap();
    let head = primary.txlog().head_seq();
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));

    // What a twin relays once it has followed the restored main.
    let newer = own + 2_000;
    remember_generation_proof(
        td.path(),
        "feedc0de",
        &restored_answer(&primary, &real, newer),
    )
    .unwrap();
    let proof = generation_proof(td.path())
        .unwrap()
        .expect("the twin relays the restored main's answer");

    // Not proofs: its own (or an older) generation, another log, a key that is
    // not this box's, an answer edited after it was signed.
    let stranger =
        gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
            .unwrap();
    let not_proofs = [
        gunnar_proto::GenerationProof {
            generation: own,
            ..proof.clone()
        },
        gunnar_proto::GenerationProof {
            log_id: "0000ffff0000ffff0000ffff0000ffff".into(),
            ..proof.clone()
        },
        gunnar_proto::GenerationProof {
            answer_key: stranger.public_key().to_openssh().unwrap(),
            answer_sig: gunnar_grpc::auth::sign_twin_answer_at(
                &stranger,
                &proof.nonce,
                &real,
                proof.first_seq,
                proof.head_seq,
                newer,
            )
            .unwrap(),
            ..proof.clone()
        },
        gunnar_proto::GenerationProof {
            generation: newer + 1,
            ..proof.clone()
        },
    ];
    for (n, p) in not_proofs.iter().enumerate() {
        plane.fence_by_generation(p).unwrap();
        assert!(
            !is_fenced(pd.path()),
            "not-a-proof #{n} fenced the box: {p:?}"
        );
    }
    primary
        .catalog()
        .create_namespace("still-mine")
        .expect("a box shown nothing genuine keeps its writes");
    let head = primary.txlog().head_seq().max(head);

    // The genuine one.
    plane.fence_by_generation(&proof).unwrap();
    assert!(
        is_fenced(pd.path()),
        "the old main is fenced by a proof of a newer generation of its own log"
    );
    assert!(
        primary.catalog().create_namespace("split-brain").is_err(),
        "a fenced box originates nothing"
    );
    assert_eq!(
        primary.txlog().head_seq(),
        head,
        "not one frame after the fence"
    );
    assert!(
        plane.log_fetch(1, head, u64::MAX).is_err(),
        "and nothing is fetched from it: its history ends where the restore began"
    );
    let alarm = plane
        .health_alarms()
        .into_iter()
        .find(|a| a.event == "twin.fenced_by_generation")
        .expect("Admin.Health says the box is fenced");
    assert!(
        alarm.detail.contains(&newer.to_string()) && alarm.detail.contains(LOG_FENCED),
        "it names the newer generation and the file: {}",
        alarm.detail
    );
    let answer = plane
        .log_head_signed("abc")
        .expect("it still answers LogHead, so a twin refuses it by its generation");
    assert_eq!(answer.generation, own);
    drop(plane);
    drop(primary);

    let back = Arc::new(ServerStores::open(pd.path()).unwrap());
    assert!(
        back.catalog().create_namespace("after-restart").is_err(),
        "the fence survives a restart"
    );
}

/// Stinky row 18, **the other side** (OWNER 20:20Z): a main merely CUT OFF from
/// its twin is no old main. It keeps serving writes however long the twin is
/// gone (no lease, no quorum: nothing on the main waits on its twin), it is not
/// fenced and raises no fence alarm, and when the twin reaches it again the twin
/// recharges what it missed from the main. Only a repair that fenced the old
/// main at the provider first may promote a new one (monetize-guardian's
/// `no_promotion_without_a_confirmed_fence_of_the_old_main`).
#[test]
fn a_main_cut_off_from_its_twin_keeps_taking_writes_and_the_twin_catches_up_when_back() {
    use gunnar_grpc::seam::ControlPlane as _;
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let real = mint_log_id(pd.path()).unwrap();
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
    let twin = open_twin(td.path());
    let following = || LocalNamed {
        stores: Arc::clone(&primary),
        id: real.clone(),
    };
    Follower::new(Arc::clone(&twin), Box::new(following()))
        .step()
        .unwrap();
    let level = primary.txlog().head_seq();
    assert_eq!(twin.txlog().head_seq(), level);

    // Cut off: the twin's polls reach nothing, again and again.
    for _ in 0..3 {
        Follower::new(Arc::clone(&twin), Box::new(Dead))
            .step()
            .expect_err("the twin cannot reach its main");
    }
    assert_eq!(status(twin.root()).unwrap().state, State::Unreachable);

    // The main alone takes every write it is given.
    for n in 0..3 {
        primary
            .catalog()
            .create_namespace(&format!("alone-{n}"))
            .expect("a main cut off from its twin keeps its writes");
    }
    primary
        .auth()
        .add_principal(Principal::new("bob", "Bob", PrincipalKind::User).unwrap())
        .expect("and its identities");
    let head = primary.txlog().head_seq();
    assert!(
        head > level,
        "the writes are logged for the twin to fetch later: {head} after {level}"
    );
    assert!(!is_fenced(pd.path()), "a twin's silence fences nothing");
    assert!(
        plane
            .health_alarms()
            .iter()
            .all(|a| a.event != "twin.fenced_by_generation"),
        "and raises no fence alarm"
    );

    // The twin is back: it recharges what it missed from the main.
    let f = Follower::new(Arc::clone(&twin), Box::new(following()));
    let mut caught = false;
    for _ in 0..10 {
        if f.step().unwrap() == (Progress::CaughtUp { head }) {
            caught = true;
            break;
        }
    }
    assert!(
        caught,
        "the twin catches up to {head}: {:?}",
        status(twin.root()).unwrap()
    );
    let names: Vec<String> = twin
        .catalog()
        .namespaces()
        .unwrap()
        .into_iter()
        .map(|n| n.name)
        .collect();
    assert!(
        (0..3).all(|n| names.contains(&format!("alone-{n}"))),
        "{names:?}"
    );
    primary
        .catalog()
        .create_namespace("after-the-twin-is-back")
        .expect("and the main never stopped");
}

/// A [`LocalSource`] that also answers with a log identity.
struct LocalNamed {
    stores: Arc<ServerStores>,
    id: String,
}

impl LogSource for LocalNamed {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        let log = self.stores.txlog();
        Ok((log.first_seq(), log.head_seq()))
    }
    fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
        let (first, head) = self.head()?;
        Ok((first, head, Some(self.id.clone())))
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        Ok(fetch_range(self.stores.txlog(), from, to, max_bytes)?)
    }
}

/// ★ **A main restored from this twin KEEPS the log's identity, so the twin
/// keeps following it.**
///
/// The design question, asserted rather than asserted-about. The id names the
/// HISTORY: a restore replays the twin's log into a fresh root frame for frame
/// at the same sequence numbers, so the restored box IS that history and the
/// twin that fed it must not need re-joining — the hour after an estate has
/// just lost its main is the worst possible hour to have no twin, and
/// `become_twin` takes a FRESH box, so re-joining would mean wiping the twin
/// and re-seeding it from the box it had just rescued.
///
/// The other half is here too: a box that mints its own id — a fresh install —
/// gets a DIFFERENT one, which is what the previous test refuses.
///
/// To see it fail before, delete the `pin_log_id` call in `restore_finish`: the
/// restored box comes up with no identity, and the assertion that it carries the
/// twin's fails.
#[test]
fn a_main_restored_from_a_twin_keeps_the_logs_identity_and_a_fresh_box_mints_a_different_one() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let real = mint_log_id(pd.path()).unwrap();
    let head = primary.txlog().head_seq();

    // The twin follows and learns the id.
    let twin = open_twin(td.path());
    Follower::new(
        Arc::clone(&twin),
        Box::new(LocalNamed {
            stores: Arc::clone(&primary),
            id: real.clone(),
        }),
    )
    .step()
    .unwrap();
    assert_eq!(log_id(td.path()).as_deref(), Some(real.as_str()));

    // The main dies; a fresh box is restored from the twin. It mints an id of
    // its own at first open — and gives it up the moment it takes another box's
    // log, exactly as it gives up its own numbers and its own secrets.
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let own_before = log_id(fd.path()).expect("a fresh box mints its own identity at first open");
    assert_ne!(
        own_before, real,
        "…and it is not the one it is about to be handed"
    );
    restore_begin(&fresh).unwrap();
    assert_eq!(
        log_id(fd.path()),
        None,
        "a box taking another box's log gives up its own log's name with its numbers"
    );
    let mut next = 1;
    while next <= head {
        let e = twin.txlog().read(next).unwrap();
        restore_apply(&fresh, &e).unwrap();
        next += 1;
    }
    // The stream announced the twin's identity, as `control twin replay` does.
    assert_eq!(restore_finish(&fresh, head, &real).unwrap(), head);

    assert_eq!(
        log_id(fd.path()).as_deref(),
        Some(real.as_str()),
        "the restored main IS the history it was built from"
    );

    // …so the twin follows it without being touched: same id, no Diverged.
    let p = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalNamed {
            stores: Arc::clone(&fresh),
            id: real.clone(),
        }),
    )
    .step()
    .unwrap();
    assert!(
        matches!(p, Progress::CaughtUp { .. }),
        "the twin follows the box it was restored into: {p:?}"
    );
    assert_eq!(status(twin.root()).unwrap().state, State::Following);
    row("a main restored from a twin KEEPS the log's identity, so the twin keeps following it without a re-join; a fresh box mints a different one", true, &format!("log {real}"));
}

/// ★ **A box that becomes a twin gives up the log identity it minted for
/// itself.**
///
/// Without this the very first poll would compare the id this box minted at its
/// own first open against the primary's, find them different, and the twin would
/// refuse itself for ever — a join that can never succeed. The clearing sits
/// beside the two it already does (its numbers, its own secrets), because it is
/// the same act: a fresh box giving up a history it is about to stop having.
///
/// To see it fail before, delete the `forget_log_id` call in `prepare_replay`.
#[test]
fn a_fresh_box_that_becomes_a_twin_gives_up_the_log_identity_it_minted_for_itself() {
    let dir = tempfile::tempdir().unwrap();
    let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
    let own = log_id(dir.path()).expect("a fresh primary mints one at first open");

    become_twin(&stores, "primary.test:50051").unwrap();
    assert_eq!(
        log_id(dir.path()),
        None,
        "it is a twin now: the name of the log it holds is its primary's to give"
    );

    // And the first poll takes the primary's, which is a different value.
    let source = Named {
        first: 1,
        head: 0,
        id: Some("aaaabbbbccccddddeeeeffff00001111"),
    };
    Follower::new(Arc::clone(&stores), Box::new(source))
        .step()
        .unwrap();
    assert_eq!(
        log_id(dir.path()).as_deref(),
        Some("aaaabbbbccccddddeeeeffff00001111")
    );
    assert_ne!(own, "aaaabbbbccccddddeeeeffff00001111");
    row("a fresh box that becomes a twin gives up the log identity it minted for itself, so its first poll can take its primary's", true, &format!("minted {own}, follows aaaabbbb…"));
}

/// ★ **An unnamed primary is neither trusted nor refused.**
///
/// A primary older than this change sends no identity. Inventing a verdict
/// either way would be wrong: refusing would break every estate on the day it
/// upgraded its twin first, and pretending the id matched would be a lie in the
/// file whose whole job is to not lie. The numbers are all there is, exactly as
/// before, and the twin pins nothing it was not told.
#[test]
fn a_primary_that_carries_no_identity_is_followed_on_its_numbers_and_pins_nothing() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());

    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    assert!(matches!(f.step().unwrap(), Progress::Applied { .. }));
    assert_eq!(twin.txlog().head_seq(), head, "it followed");
    assert_eq!(
        log_id(td.path()),
        None,
        "and pinned nothing, because it was told nothing"
    );
    row("a primary that carries no log identity is followed on its numbers and the twin pins nothing — the upgrade path, neither trusted nor refused", true, "no id offered, none recorded");
}

// ── the primary proving itself to the twin ──────────────────────────────────

/// Sign an answer the way a primary does, so the test drives the real
/// verification and not a copy of it.
fn signed_answer(
    key: &gunnar_core::ssh_key::PrivateKey,
    nonce: &str,
    id: &str,
    first: u64,
    head: u64,
) -> gunnar_proto::LogHeadResponse {
    gunnar_proto::LogHeadResponse {
        first_seq: first,
        head_seq: head,
        log_id: id.to_owned(),
        answer_key: key.public_key().to_openssh().unwrap(),
        answer_sig: gunnar_grpc::auth::sign_twin_answer(key, nonce, id, first, head).unwrap(),
        ..Default::default()
    }
}

/// ★ **The twin pins the key its primary answers with, and refuses every answer
/// that is not signed by it.**
///
/// Before this the twin dialled `client_signer::plain_channel` — a bare
/// `http://` endpoint with no TLS — and verified NOTHING about who answered.
/// The credential on every call proved the TWIN to the primary; not one byte
/// went the other way, so the trust was one-directional and whatever won the
/// address won the twin.
///
/// Trust on first use, because there is no earlier moment: `Bootstrap.BecomeTwin`
/// is answered by the TWIN and the primary is not in that conversation, so at
/// join there is nothing to pin but an address. From the first poll on, this is
/// `known_hosts`.
///
/// This drives the real verifier and the real pin against real signatures.
/// `GrpcSource::check_answer`, which composes them, needs a live channel to
/// reach; what it is made of is here, and its own three branches are stated in
/// its doc comment.
///
/// To see it fail before, make `gunnar_grpc::auth::verify_twin_answer` return
/// `true` unconditionally: cases 2, 3, 4 and 5 all stop refusing. Neutralising
/// `pin_primary_key`'s `ensure!` instead reddens case 2's second half alone.
#[test]
fn a_twin_pins_the_key_its_primary_answers_with_and_refuses_anything_else() {
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let primary_key =
        gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
            .unwrap();
    let line = primary_key.public_key().to_openssh().unwrap();
    let id = "1234abcd1234abcd1234abcd1234abcd";

    // 1. First poll: nothing pinned, a signature that verifies against the key
    //    offered → it is written down.
    assert_eq!(pinned_primary_key(root), None);
    let answer = signed_answer(&primary_key, "n1", id, 1, 9);
    assert!(gunnar_grpc::auth::verify_twin_answer(
        &line,
        &answer.answer_sig,
        "n1",
        id,
        1,
        9
    ));
    pin_primary_key(root, &answer.answer_key).unwrap();
    assert_eq!(pinned_primary_key(root).as_deref(), Some(line.as_str()));

    // 2. Another box at the same address. The whole point.
    let impostor =
        gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
            .unwrap();
    let theirs = signed_answer(&impostor, "n2", id, 1, 9);
    assert!(
        !gunnar_grpc::auth::verify_twin_answer(&line, &theirs.answer_sig, "n2", id, 1, 9),
        "a perfectly well-formed answer from a DIFFERENT box must not verify against the pin"
    );
    let err = format!(
        "{:#}",
        pin_primary_key(root, &theirs.answer_key).unwrap_err()
    );
    assert!(err.contains("already follows"), "{err}");
    assert_eq!(
        pinned_primary_key(root).as_deref(),
        Some(line.as_str()),
        "and the pin does not move"
    );

    // 3. The real primary's answer to an EARLIER poll, replayed now. The nonce
    //    is inside the signature, so it is not this poll's answer.
    let old = signed_answer(&primary_key, "n1", id, 1, 9);
    assert!(
        !gunnar_grpc::auth::verify_twin_answer(&line, &old.answer_sig, "n3", id, 1, 9),
        "a recorded answer must not satisfy a later poll"
    );

    // 4. The real primary's signature with the numbers edited in flight.
    assert!(
        !gunnar_grpc::auth::verify_twin_answer(&line, &answer.answer_sig, "n1", id, 1, 10),
        "the head is inside the signature"
    );
    assert!(
        !gunnar_grpc::auth::verify_twin_answer(
            &line,
            &answer.answer_sig,
            "n1",
            "deadbeefdeadbeefdeadbeefdeadbeef",
            1,
            9
        ),
        "so is the log identity"
    );

    // 5. No signature at all, which is what every answer was before this lane.
    assert!(!gunnar_grpc::auth::verify_twin_answer(
        &line, "", "n1", id, 1, 9
    ));

    row("a twin pins the key its primary answers with and refuses another box, a replayed answer, edited numbers and an unsigned one", true, &format!("pinned {}", &line[..line.len().min(40)]));
}

/// ★ **A primary signs its answers, out of its own host key, and the signature
/// is over what the twin acts on.**
///
/// The end-to-end half of the above: the real `sign_head_answer` on a real
/// `ServerStores`, verified by the real verifier. The host key is the signing
/// key on purpose — it is replayed by a restore, so a main restored from this
/// twin keeps verifying and the twin keeps following it, matching what
/// `LOG_ID` does for the history; a fresh box at the same address mints its own
/// and is refused.
///
/// To see it fail before, make `sign_head_answer` return `(String::new(),
/// String::new())`: the first assertion finds no signature.
#[test]
fn a_primary_signs_its_answers_with_the_host_key_a_restore_replays() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let id = mint_log_id(pd.path()).unwrap();
    let head = primary.txlog().head_seq();

    let (key, sig) = sign_head_answer(&primary, "a-nonce", &id, 1, head, 0);
    assert!(
        !sig.is_empty() && !key.is_empty(),
        "a primary asked with a nonce signs its answer"
    );
    assert!(
        gunnar_grpc::auth::verify_twin_answer(&key, &sig, "a-nonce", &id, 1, head),
        "and it verifies against the key it offered"
    );

    // It IS the host key — the value a restore replays.
    let host = gunnar_core::ensure_host_key(&primary.secrets().unwrap()).unwrap();
    assert_eq!(
        key.trim(),
        host.public_openssh().unwrap().trim(),
        "the answering key is this box's SSH host key"
    );

    // No nonce, nothing to be fresh about, nothing signed: a caller that is not
    // a twin's poll gets the same answer it always got.
    let (k, s) = sign_head_answer(&primary, "", &id, 1, head, 0);
    assert!(
        k.is_empty() && s.is_empty(),
        "an unnonced caller is not a pairing and is not signed to"
    );
    row("a primary signs its LogHead answers with the SSH host key a restore replays, over the nonce, the log identity and the numbers", true, &format!("head {head}"));
}

/// **A fresh box's own audit trail goes with its own numbers, and the restored chain
/// verifies** (lane T11, 2026-09-21).
///
/// The audit trail is a replicated, hash-chained document now, so the frames arriving from the
/// twin are `Append`s of the DEAD primary's chain from its own seq 0. A fresh box that had
/// written even one line of its own would take them onto the end of it, and the result would
/// not verify — entry 0's `prev` is GENESIS and here it would follow somebody else's tip. A
/// box that cannot open its trail serves WITHOUT one, so the restore would have destroyed the
/// audit history it exists to carry. The same reasoning as `forget_own_secrets` and
/// `forget_log_id`, one document along.
///
/// **Fails before the fix**: delete the `forget_own_audit_trail(stores)?;` line from
/// `prepare_replay` — the restored chain then holds the fresh box's line first and
/// `AuditChain::open` refuses it as broken.
#[test]
fn a_fresh_boxs_own_audit_trail_goes_with_its_own_numbers() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();

    let decision = |method: &'static str| gunnar_core::AuditEvent {
        method,
        principal: "alice",
        fingerprint: "SHA256:abc",
        scope: "server",
        level: "admin",
        auth_mode: "required",
        decision: "allow",
        reason: "",
    };

    let (primary, _facts) = primary(pd.path());
    // The primary's own trail, framed — three decisions it will be asked about afterwards.
    let ptrail = primary.audit_path();
    let chain =
        gunnar_core::AuditChain::open_framed(&ptrail).expect("the primary's trail is framable");
    for m in ["Repos.Create", "Access.Grant", "Access.Revoke"] {
        chain.append(decision(m), 1).expect("append");
    }
    let primary_tip = chain.tip();
    let primary_entries = chain.verify().expect("the primary's chain holds");
    drop(chain);

    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    drop(primary);

    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    // The fresh box's OWN line, written before anybody told it to take another box's log.
    let ftrail = fresh.audit_path();
    let own =
        gunnar_core::AuditChain::open_framed(&ftrail).expect("a fresh primary's trail is framable");
    own.append(decision("Bootstrap.Enroll"), 1).expect("append");
    let own_tip = own.tip();
    drop(own);
    assert_ne!(
        own_tip, primary_tip,
        "the two boxes must really have different histories"
    );

    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    drop(fresh);

    let back = ServerStores::open(fd.path()).expect("a restored box opens");
    let trail = back.audit_path();
    let restored = gunnar_core::AuditChain::open(&trail).expect(
        "the restored trail opens — it would not if the fresh box's own line were still under it",
    );
    assert_eq!(
        restored.verify().expect("the restored chain holds"),
        primary_entries
    );
    assert_eq!(
        restored.tip(),
        primary_tip,
        "the dead primary's own tip, not a splice"
    );
    let entries = restored.entries().expect("entries");
    assert_eq!(
        entries[0].prev,
        gunnar_core::audit::GENESIS,
        "the chain starts where the primary's did"
    );
    assert!(
        entries.iter().all(|e| e.method != "Bootstrap.Enroll"),
        "the fresh box's own decision is not in the primary's trail: {entries:?}"
    );

    // And the restored box CONTINUES it: the history is one chain across the disaster.
    let next = restored.append(decision("Repos.List"), 2).expect("append");
    assert_eq!(next.prev, primary_tip);

    row(
        "a fresh box's own audit trail goes with its own numbers: the restored chain is the primary's, from GENESIS, and the restored box continues it",
        true,
        &format!("{primary_entries} entries, tip {primary_tip}"),
    );
}

/// **A restored box whose copy of the audit trail is missing when it first opens rebuilds it
/// from its own log** (lane T11, 2026-09-21).
///
/// **What this test proves, and what it does not.** It proves the rebuild: the file is deleted
/// between the replay and the reopen, and the box comes back with the primary's whole chain and
/// continues it. It does NOT reproduce T3's KVM `T1v`/`T2v` failure, and this comment used to
/// say it did. That failure had a different cause: the `txlog-restoring` unlink was lost in the
/// hard stop, the box reopened still restoring, and in that state this rebuild does not run at
/// all. `twin.restoring_at_open` and T3's `durable::remove_durable` are that fix. Deleting the
/// file here is a stand-in for "the copy on disk disagrees with the log" — a case the rebuild
/// defends against but that nothing has yet been measured to cause on the restore path.
///
/// **Fails before the fix**: delete the `crate::twin::rematerialise_audit_trail(&stores, just_restored)` call
/// from `ServerStores::open` — the restored box opens an EMPTY chain and its first own decision
/// is seq 0 with `prev` = GENESIS.
#[test]
fn a_restored_box_whose_off_stripe_trail_was_lost_rebuilds_it_from_its_own_log() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let d = |method: &'static str| gunnar_core::AuditEvent {
        method,
        principal: "bench",
        fingerprint: "SHA256:x",
        scope: "server",
        level: "admin",
        auth_mode: "required",
        decision: "allow",
        reason: "",
    };

    let (primary, _facts) = primary(pd.path());
    let chain = gunnar_core::AuditChain::open_framed(primary.audit_path()).unwrap();
    for m in [
        "Access.AddPrincipal",
        "Access.Grant",
        "Repos.SetVisibility",
        "Admin.AuditTrail",
    ] {
        chain.append(d(m), 1).unwrap();
    }
    let (primary_entries, primary_tip) = (chain.verify().unwrap(), chain.tip());
    drop(chain);
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    drop(primary);

    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let trail = fresh.audit_path();
    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    assert!(
        trail.exists(),
        "the replay wrote the trail — this test is about losing it AFTER"
    );
    drop(fresh);

    // The guest is stopped hard; the off-stripe copy does not survive. The log (on /data) does.
    std::fs::remove_file(&trail).unwrap();

    let back = ServerStores::open(fd.path()).expect("a restored box opens");
    let path = back.audit_path();
    assert_eq!(path, trail, "the replicated trail, not the box's own");
    // Opened the way `serve` opens it for the gate.
    let gate =
        gunnar_core::AuditChain::open_framed(&path).expect("the gate opens the rebuilt trail");
    assert_eq!(
        gate.verify().unwrap(),
        primary_entries,
        "every entry the twin held came back"
    );
    assert_eq!(
        gate.tip(),
        primary_tip,
        "the dead primary's own tip, not a fresh chain"
    );
    let next = gate.append(d("Access.ListPrincipals"), 2).unwrap();
    assert_eq!(next.prev, primary_tip, "and the restored box continues it");
    assert_eq!(next.seq, primary_entries as u64);

    row(
        "a restored box whose off-stripe audit trail was lost after the replay rebuilds it from its own log before the gate opens it",
        true,
        &format!("{primary_entries} entries, tip {primary_tip}"),
    );
}

/// **The rebuild never deletes from an audit trail, and a snapshot's whole copy supersedes the
/// lines before it** (lane T11) — the two rules `rebuild_audit_trail` keeps besides "rebuild
/// what was lost".
///
/// **Fails before the fix** in two named ways: drop the `Rebuilt::Longer` early return and the
/// longer file is rewritten SHORTER (a line nobody can account for, deleted); drop the
/// `lines.clear()` in the `File` arm and the snapshot's document is appended to the lines it
/// replaced, so the rebuilt trail holds entries twice and does not verify.
#[test]
fn the_audit_rebuild_never_deletes_and_a_snapshot_supersedes() {
    use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
    let dir = tempfile::tempdir().unwrap();
    let log = Log::open(dir.path().join("txlog")).unwrap();
    // Three real entries, made by a real chain so they verify.
    let scratch = dir.path().join("scratch.jsonl");
    let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
    let d = |method: &'static str| gunnar_core::AuditEvent {
        method,
        principal: "p",
        fingerprint: "",
        scope: "server",
        level: "read",
        auth_mode: "required",
        decision: "allow",
        reason: "",
    };
    for m in ["A.One", "A.Two", "A.Three", "A.Four"] {
        chain.append(d(m), 1).unwrap();
    }
    let all: Vec<String> = std::fs::read_to_string(&scratch)
        .unwrap()
        .lines()
        .map(str::to_owned)
        .collect();
    let app = |line: &str| {
        log.append(
            Kind::Append,
            &Meta::Append {
                log: AUDIT_REL.into(),
            },
            line.as_bytes(),
        )
        .unwrap()
    };
    for l in &all[..3] {
        app(l);
    }

    // A file already holding MORE than the log (a fourth line): left exactly as found.
    let path = dir.path().join("audit-chain.jsonl");
    std::fs::write(
        &path,
        all.iter().map(|l| format!("{l}\n")).collect::<String>(),
    )
    .unwrap();
    let before = std::fs::read(&path).unwrap();
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Longer { file: 4, log: 3 }
    );
    assert_eq!(
        std::fs::read(&path).unwrap(),
        before,
        "nothing is deleted from an audit trail"
    );

    // A snapshot's whole document (the first two lines) supersedes the three before it; the
    // fourth line arrives after it. The rebuilt trail is those three lines, once each.
    log.append(
        Kind::File,
        &Meta::File {
            path: AUDIT_REL.into(),
            deleted: false,
        },
        format!("{}\n{}\n", all[0], all[1]).as_bytes(),
    )
    .unwrap();
    app(&all[2]);
    app(&all[3]);
    std::fs::remove_file(&path).unwrap();
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Rebuilt { had: 0, now: 4 }
    );
    assert_eq!(
        gunnar_core::AuditChain::open(&path)
            .unwrap()
            .verify()
            .unwrap(),
        4
    );
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Intact(4),
        "a second pass changes nothing"
    );
}

/// A primary whose disks were just destroyed: it does not answer at all.
struct Dead;

impl LogSource for Dead {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Err(anyhow!(
            "transport error: Connection refused (os error 111)"
        ))
    }
    fn fetch(&self, from: u64, _to: u64, _max: u64) -> anyhow::Result<Vec<Entry>> {
        panic!("a twin that cannot reach its primary must not fetch (asked for {from})")
    }
}

/// ★ **(d) The numbers survive `unreachable` — and a stall records the answer
/// it actually got.**
///
/// MEASURED under KVM by T3: a twin level at 221 whose primary had just been
/// destroyed reported "a primary head it last saw at **0**". The head() error
/// branch wrote `primary_head_seq: 0`, so the one number an operator must read
/// before deciding to restore was wrong at the one moment it was read. The
/// primary saying nothing NOW is not the primary having had nothing THEN.
///
/// The second half is the same bug the other way round: a poll that REACHED the
/// primary and then failed recorded the previous poll's head and time, not the
/// answer it had just been given, so a twin stalling behind a primary still
/// taking writes under-reported how far behind it was.
///
/// To see it fail before, put back `primary_head_seq: 0` in the head() error
/// branch of `Follower::step_inner` (the first assertion reads 0); for the second
/// half, make `step` ignore `heard` (`let heard = None::<(u64, i64)>;`).
#[test]
fn a_twin_keeps_the_primarys_last_head_and_time_when_the_primary_stops_answering() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    let twin = open_twin(td.path());

    // Level with a live primary.
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let level = status(twin.root()).unwrap();
    assert!(level.caught_up(), "{level:?}");
    assert!(
        level.last_answer_unix_ms > 0 && level.last_level_unix_ms == level.last_answer_unix_ms,
        "level at that answer: {level:?}"
    );

    // The primary's disks are destroyed.
    Follower::new(Arc::clone(&twin), Box::new(Dead))
        .step()
        .expect_err("nothing answers");
    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Unreachable, "{s:?}");
    assert_eq!(
        s.primary_head_seq, head,
        "the head the primary last reported survives its silence — T3 read 0 here: {s:?}"
    );
    assert_eq!(s.applied_seq, head);
    assert_eq!(
        s.last_answer_unix_ms, level.last_answer_unix_ms,
        "and so does WHEN it said it"
    );
    assert_eq!(
        s.last_level_unix_ms, s.last_answer_unix_ms,
        "so the twin can still say it was level at its last word: {s:?}"
    );

    // A different history: a primary that answers, then cannot be taken from.
    std::thread::sleep(Duration::from_millis(5));
    let f = Follower::new(Arc::clone(&twin), Box::new(HeadOnly { head: head + 5 }));
    f.step().expect_err("the fetch fails");
    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Stalled, "{s:?}");
    assert_eq!(
        s.primary_head_seq,
        head + 5,
        "a stall records the head the primary JUST reported, not the last poll's: {s:?}"
    );
    assert!(
        s.last_answer_unix_ms > level.last_answer_unix_ms,
        "and the time of that answer: {s:?}"
    );
    assert!(
        s.last_level_unix_ms < s.last_answer_unix_ms,
        "and it was NOT level at it — the restore guard's signal: {s:?}"
    );

    row(
        "a twin keeps its primary's last head and time through `unreachable`, and a stall records the answer it actually got — the numbers a restore decision is made on",
        true,
        &format!("level at {head}; unreachable reads {head}, not 0"),
    );
}

/// **A restored box takes writes again, and a box that is STILL restoring says so on the wire.**
///
/// MEASURED under KVM 2026-09-21 (lane T3, medium `3560df65…`): after `twin replay` and a
/// restart, `control group add` on the restored primary answered `txlog: this box is a twin (or
/// restoring); it originates no transactions`. The finish's unlink of `txlog-restoring` had not
/// reached the disk before the stop (`durable::remove_durable`). A page-cache loss cannot be
/// staged in a unit test, so this pins the two halves that can be: the finished box writes,
/// and a box whose marker IS present raises the `Admin.Health` alarm instead of being silent.
#[test]
fn a_restored_box_writes_and_one_still_restoring_raises_an_alarm() {
    use gunnar_grpc::ControlPlane as _;
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    drop(primary);
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    drop(fresh);
    assert!(
        !fd.path().join(RESTORING).exists(),
        "a finished restore leaves no marker"
    );
    let back = Arc::new(ServerStores::open(fd.path()).unwrap());
    let head = back.txlog().head_seq();
    back.catalog()
        .create_namespace("after-restore")
        .expect("a restored box must take a write");
    assert!(
        back.txlog().head_seq() > head,
        "the write must be a transaction the twin can follow"
    );
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&back));
    assert!(plane
        .health_alarms()
        .iter()
        .all(|a| a.event != "twin.restoring_at_open"));
    drop(plane);
    drop(back);

    // The KVM state: the marker survived the restart.
    crate::durable::write_atomic_unframed(fd.path(), RESTORING, b"restoring\n").unwrap();
    let stuck = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert!(
        stuck.catalog().create_namespace("refused").is_err(),
        "a restoring box originates nothing"
    );
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&stuck));
    assert!(
        plane
            .health_alarms()
            .iter()
            .any(|a| a.event == "twin.restoring_at_open"),
        "a box stuck mid-restore must say so on Admin.Health, not only refuse writes"
    );
}

/// Five real, chained audit lines — made by a real chain so they verify.
fn five_chained_lines(dir: &std::path::Path) -> Vec<String> {
    let scratch = dir.join("five.jsonl");
    let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
    let d = |method: &'static str| gunnar_core::AuditEvent {
        method,
        principal: "p",
        fingerprint: "",
        scope: "server",
        level: "read",
        auth_mode: "required",
        decision: "allow",
        reason: "",
    };
    for m in ["A.One", "A.Two", "A.Three", "A.Four", "A.Five"] {
        chain.append(d(m), 1).unwrap();
    }
    std::fs::read_to_string(&scratch)
        .unwrap()
        .lines()
        .map(str::to_owned)
        .collect()
}

/// **A primary that crashed and lost the tail of its audit trail rebuilds it on the next start
/// — not only after a restore** (T17, 2026-09-21; owner's decision: check on EVERY start).
///
/// The trail's bytes live on the boot disk's P3 (`/data/log` → `/serialdat/log`), the
/// transaction log on the striped `/data`. After a hard stop XFS log recovery runs on both, and
/// the file can come back SHORTER than the log's audit frames, or with a torn last line. A
/// primary that then appends continues a shorter chain: its `prev` no longer matches the
/// twin's copy, and the twin's trail stops verifying from that entry on.
///
/// **Fails before the fix**: at `624a62e` the rebuild runs only when `reseal_pending` (the
/// first open after a restore); a crashed primary opens the two-line file, and the gate's next
/// append is seq 2 with the tip of line 2 — a fork of the chain the twin holds.
#[test]
fn a_primary_that_lost_its_audit_tail_in_a_crash_rebuilds_it_on_the_next_start() {
    // (kept, torn): 4 of 5 is EXACTLY the window T11 named — `AuditChain::append` syncs the
    // frame, then writes the line; a crash between the two leaves the log one entry ahead.
    for (kept, torn) in [(4usize, false), (2, false), (2, true)] {
        let dir = tempfile::tempdir().unwrap();
        let d = |method: &'static str| gunnar_core::AuditEvent {
            method,
            principal: "bench",
            fingerprint: "SHA256:x",
            scope: "server",
            level: "admin",
            auth_mode: "required",
            decision: "allow",
            reason: "",
        };
        let stores = ServerStores::open(dir.path()).expect("a primary opens");
        let path = stores.audit_path();
        let chain = gunnar_core::AuditChain::open_framed(&path).expect("the replicated trail");
        for m in [
            "Access.AddPrincipal",
            "Access.Grant",
            "Repos.SetVisibility",
            "Admin.AuditTrail",
            "Access.Revoke",
        ] {
            chain.append(d(m), 1).unwrap();
        }
        let (entries, tip) = (chain.verify().unwrap(), chain.tip());
        drop(chain);
        drop(stores);

        // The hard stop: the file comes back with `kept` whole lines (and, torn, half the next).
        let text = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = text.lines().collect();
        let mut short: String = lines[..kept].iter().map(|l| format!("{l}\n")).collect();
        if torn {
            short.push_str(&lines[kept][..lines[kept].len() / 2]);
        }
        std::fs::write(&path, short).unwrap();

        let back = ServerStores::open(dir.path()).expect("the primary starts again");
        assert_eq!(back.audit_path(), path);
        let gate =
            gunnar_core::AuditChain::open_framed(&path).expect("the gate opens the rebuilt trail");
        assert_eq!(
            gate.verify().unwrap(),
            entries,
            "kept={kept} torn={torn}: every entry the log carries is back"
        );
        assert_eq!(
            gate.tip(),
            tip,
            "kept={kept} torn={torn}: the same tip the twin holds"
        );
        let next = gate.append(d("Access.ListPrincipals"), 2).unwrap();
        assert_eq!(
            (next.seq, next.prev.as_str()),
            (entries as u64, tip.as_str()),
            "kept={kept} torn={torn}: and it continues that chain"
        );
        row(
            &format!("a primary whose audit trail lost its last {} line(s){} in a crash rebuilds it from its log on the next start, before the gate appends", 5 - kept, if torn { " (torn last line)" } else { "" }),
            true,
            &format!("{entries} entries, tip {tip}"),
        );
    }
}

/// **The every-start check never shortens a trail the log only partly holds** (T17): after the
/// operator's retention door (`Log::retain_from`) the log carries only a WINDOW of the trail,
/// and a box upgraded from before T11 has lines no frame ever carried. Rebuilding "the log's
/// view" there would delete the history the log no longer has. The rebuild APPENDS what the
/// file is missing past its own tip, linked by `prev`, and deletes nothing but a torn line.
///
/// **Fails before the fix**: `624a62e` rewrites the file to exactly the log's view whenever it
/// is not a prefix — the retained window of three lines replaces the file's four, and the
/// first two entries are gone (and the trail no longer opens: seq 2 is its first line).
#[test]
fn the_every_start_audit_check_appends_the_lost_tail_and_never_shortens_a_windowed_trail() {
    use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
    let dir = tempfile::tempdir().unwrap();
    let all = five_chained_lines(dir.path());
    let path = dir.path().join("audit-chain.jsonl");
    let write = |n: usize| {
        std::fs::write(
            &path,
            all[..n]
                .iter()
                .map(|l| format!("{l}\n"))
                .collect::<String>(),
        )
        .unwrap()
    };

    // A retention window: frames for lines 0..5, then retention drops the segment holding 0 and 1.
    let log = Log::open_with(dir.path().join("txlog"), 1).unwrap();
    for l in &all {
        log.append(
            Kind::Append,
            &Meta::Append {
                log: AUDIT_REL.into(),
            },
            l.as_bytes(),
        )
        .unwrap();
    }
    log.retain_from(3).unwrap();
    assert!(
        log.first_seq() > 1,
        "the window really opened: first {}",
        log.first_seq()
    );

    write(5);
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Intact(5),
        "whole, windowed: untouched"
    );
    write(4);
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Rebuilt { had: 4, now: 5 }
    );
    assert_eq!(
        gunnar_core::AuditChain::open(&path)
            .unwrap()
            .verify()
            .unwrap(),
        5,
        "the lost line is back and the head kept"
    );

    // An upgraded box: lines 0 and 1 were written before the trail was framed; 2..5 are frames.
    let d2 = tempfile::tempdir().unwrap();
    let log = Log::open(d2.path().join("txlog")).unwrap();
    for l in &all[2..] {
        log.append(
            Kind::Append,
            &Meta::Append {
                log: AUDIT_REL.into(),
            },
            l.as_bytes(),
        )
        .unwrap();
    }
    write(3);
    assert_eq!(
        rebuild_audit_trail(&log, &path).unwrap(),
        Rebuilt::Rebuilt { had: 3, now: 5 }
    );
    assert_eq!(
        gunnar_core::AuditChain::open(&path)
            .unwrap()
            .verify()
            .unwrap(),
        5,
        "the unframed history kept, the tail appended"
    );
    // A file whose tip the log contradicts is left as found: nothing is guessed.
    std::fs::write(&path, format!("{}\n{}\n{}\n", all[0], all[1], all[3])).unwrap();
    let before = std::fs::read(&path).unwrap();
    assert!(
        !matches!(
            rebuild_audit_trail(&log, &path),
            Ok(Rebuilt::Rebuilt { .. })
        ),
        "nothing rebuilt over a contradiction"
    );
    assert_eq!(
        std::fs::read(&path).unwrap(),
        before,
        "a diverged trail is never rewritten"
    );
}

/// **What the every-start audit check costs a boot** (T17) — MEASURED, not asserted: run with
/// `cargo test --release -p gunnar-server --lib -- --ignored --nocapture audit_check_boot_cost`.
/// An empty log; a log of 100 000 frames of which 10 000 are audit lines; a log of 100 000
/// audit lines. Each with the file whole (the every-boot case) and with its last 10 lines lost.
/// Also the per-frame `locate_blob` walk `624a62e` used, for the comparison. Warm page cache.
#[test]
#[ignore = "a measurement: run in release with --nocapture"]
fn audit_check_boot_cost() {
    use gunnar_core::txlog::{Kind, Meta, AUDIT_REL};
    let d = |method: &'static str| gunnar_core::AuditEvent {
        method,
        principal: "bench",
        fingerprint: "SHA256:x",
        scope: "server",
        level: "admin",
        auth_mode: "required",
        decision: "allow",
        reason: "",
    };
    let time = |label: &str, log: &Log, path: &std::path::Path, lost: usize, full: &[u8]| {
        let keep: usize = full
            .split(|b| *b == b'\n')
            .filter(|l| !l.is_empty())
            .count()
            .saturating_sub(lost);
        let short: Vec<u8> = full
            .split_inclusive(|b| *b == b'\n')
            .take(keep)
            .flatten()
            .copied()
            .collect();
        std::fs::write(path, &short).unwrap();
        let t = std::time::Instant::now();
        let r = rebuild_audit_trail(log, path).unwrap();
        let every = t.elapsed();
        let t = std::time::Instant::now();
        let mut n = 0u64;
        for seq in log.first_seq()..=log.head_seq() {
            let at = log.locate_blob(seq).unwrap();
            n += at.blob_len.min(1);
        }
        let old = t.elapsed();
        eprintln!("AUDIT-BOOT-COST {label:<44} lost {lost:>2}: check {:>9.3} ms  ({r:?})   | 624a62e per-frame locate_blob walk {:>9.3} ms ({n} frames)", every.as_secs_f64() * 1e3, old.as_secs_f64() * 1e3);
    };
    for (label, frames, every_nth_audit) in [
        ("empty log", 0usize, 1usize),
        ("100 000 frames, 10 000 audit", 100_000, 10),
        ("100 000 frames, all audit", 100_000, 1),
    ] {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path().join("txlog")).unwrap();
        let scratch = dir.path().join("scratch.jsonl");
        let chain = gunnar_core::AuditChain::open(&scratch).unwrap();
        let mut audits = 0usize;
        for i in 0..frames {
            if i % every_nth_audit == 0 {
                chain.append(d("Access.Grant"), i as i64).unwrap();
                audits += 1;
            }
        }
        drop(chain);
        let full = std::fs::read(&scratch).unwrap_or_default();
        let mut lines = full.split(|b| *b == b'\n').filter(|l| !l.is_empty());
        let other = vec![b'x'; 200];
        for i in 0..frames {
            if i % every_nth_audit == 0 {
                log.append(
                    Kind::Append,
                    &Meta::Append {
                        log: AUDIT_REL.into(),
                    },
                    lines.next().unwrap(),
                )
                .unwrap();
            } else {
                log.append(
                    Kind::Append,
                    &Meta::Append {
                        log: "attestations.jsonl".into(),
                    },
                    &other,
                )
                .unwrap();
            }
        }
        assert_eq!(log.head_seq() as usize, frames);
        let path = dir.path().join("audit-chain.jsonl");
        time(label, &log, &path, 0, &full);
        if audits > 0 {
            time(label, &log, &path, 10, &full);
        }
    }
}

/// **Two twins in one process keep two statuses** (T17, 2026-09-21). The status was one
/// process-wide slot, so a test's follower read — and, through `last_level_unix_ms`, inherited
/// — whatever a test running beside it had just written. T15 saw
/// `a_twin_keeps_the_primarys_last_head_and_time_when_the_primary_stops_answering` fail once in
/// a full run and pass 5/5 alone. This drives the interleaving deterministically: A level,
/// then B unreachable with no answer ever; A must still read A.
///
/// **Fails before the fix**: key every slot by one path (`status_slots` ignoring `root`) and
/// A reads B's `unreachable`, applied 0.
#[test]
fn two_twins_in_one_process_keep_two_statuses() {
    let pd = tempfile::tempdir().unwrap();
    let (ad, bd) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
    let (primary, _facts) = primary(pd.path());
    let a = open_twin(ad.path());
    let b = open_twin(bd.path());
    Follower::new(Arc::clone(&a), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let level = status(a.root()).expect("A polled");
    assert!(level.caught_up(), "A is level: {level:?}");
    assert!(Follower::new(Arc::clone(&b), Box::new(Dead))
        .step()
        .is_err());
    let still = status(a.root()).expect("A still has its own status");
    assert_eq!(still, level, "B's poll must not touch A's status");
    let bs = status(b.root()).expect("B has its own");
    assert_eq!(
        (bs.state, bs.last_level_unix_ms),
        (State::Unreachable, 0),
        "and B inherits nothing from A: {bs:?}"
    );
}

/// **Between "Restore finished" and the restart, a twin's poll is refused BY NAME** (T17,
/// 2026-09-21; T15's TD8 host-key WARN).
///
/// MEASURED under KVM: 0.4 s after "Restore finished; the box must restart", before the
/// restart, the box logged `this box cannot read its host key … wrong key-encrypting key`
/// once. A twin's `Mirror.LogHead` (the only caller that sends a nonce) arrived in the window:
/// the replayed host key is sealed under the dead primary's KEK, this process still holds the
/// fresh box's, and the re-seal that reconciles them runs on the NEXT open. So the box
/// answered UNSIGNED, a pinned twin refused that as an impostor's answer, and the only trace
/// was a WARN that reads like a broken box.
///
/// DECIDED: in that window — `txlog-restoring` present (mid-replay) or `txlog-reseal` present
/// (finished, restart owed) — the box is not yet the primary it will be, and it says so:
/// `Unavailable`, naming the restart, with no key read and no WARN. The twin records that
/// sentence as its `why` and polls again; after the restart the same poll is answered and
/// signed. An UNNONCED `LogHead` (an operator, `twin replay`) is still answered: it asks for
/// numbers, not for proof of identity.
///
/// **Fails before the fix**: at `693e96c` the nonced poll is `Ok`, unsigned (`answer_key`
/// empty), and the WARN is the only thing that happened.
#[test]
fn a_twins_poll_between_restore_finished_and_the_restart_is_refused_by_name() {
    use gunnar_grpc::seam::ControlPlane as _;
    let (pd, td, fd) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    drop(primary);

    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    // The same process, after "Restore finished; the box must restart".
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
    let err = plane
        .log_head_signed("a-twins-nonce")
        .expect_err("a twin's poll in the restart window must be refused");
    let text = err.to_string();
    assert!(
        matches!(err, gunnar_grpc::seam::ControlError::Unavailable(_)),
        "Unavailable, so the twin polls again: {text}"
    );
    assert!(text.contains("restart"), "and it names the restart: {text}");
    let plain = plane
        .log_head_signed("")
        .expect("an unnonced LogHead is still answered");
    assert_eq!(plain.head_seq, fresh.txlog().head_seq());
    drop(plane);
    drop(fresh);

    // The restart: the re-seal runs, and the same poll is answered.
    let back = Arc::new(ServerStores::open(fd.path()).unwrap());
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&back));
    plane
        .log_head_signed("a-twins-nonce")
        .expect("after the restart the twin's poll is answered");
    row("a twin's poll between restore-finished and the restart is refused by name, and answered after it", true, &text);
}

/// **Does the primary keep what its twin has not fetched?** (T17, 2026-09-22, the owner's
/// question) — proved end to end on a real primary's log, not a scripted source.
///
/// 1. **A restart drops nothing.** The twin is held back while the primary takes six more
///    transactions, then the primary process goes away and comes back (the log reopened from
///    disk). The twin resumes and gets every one of them, contiguous, and the log still starts
///    at 1: nothing but the operator's retention door (`Mirror.LogRetain` →
///    `CoreControlPlane::log_retain` → `Log::retain_from`) ever deletes a frame.
/// 2. **The one door that drops un-fetched frames is loud and has a resync.** The twin is held
///    back again, the operator retains from past the head, and the twin's next step:
///    * from a source that CANNOT snapshot: `Progress::SnapshotNeeded` and the status word
///      `snapshot-needed`, naming where the primary retains from and what the twin needs —
///      never a skip past the hole;
///    * from a source that CAN (the production gRPC source can): it installs the primary's
///      whole state at the head and follows on from there.
#[test]
fn a_primary_keeps_what_its_twin_has_not_fetched_and_the_one_door_that_drops_it_is_loud() {
    use gunnar_grpc::seam::ControlPlane as _;
    let (pd, td) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
    let (primary, facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    let level = twin.txlog().head_seq();

    // ── 1. held back, then the primary restarts ────────────────────────────────
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    for i in 0..6 {
        primary
            .repo_store(&app)
            .unwrap()
            .serve()
            .update_ref(&format!("refs/heads/held-{i}"), None, Some(&facts.oids[0]))
            .unwrap();
    }
    let before = primary.txlog().head_seq();
    assert!(before >= level + 6);
    drop(app);
    drop(primary);
    let primary = Arc::new(ServerStores::open(pd.path()).expect("the primary comes back"));
    // The reopen may frame something of its own (MEASURED: one frame, 88 → 89); it never
    // drops one.
    assert_eq!(
        primary.txlog().first_seq(),
        1,
        "a restart keeps every frame"
    );
    assert!(
        primary.txlog().head_seq() >= before,
        "and loses none of the held-back ones"
    );
    let before = primary.txlog().head_seq();
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    assert_eq!(
        f.step().unwrap(),
        Progress::Applied {
            count: before - level,
            head: before
        },
        "every held-back entry, none skipped"
    );
    assert_eq!(twin.txlog().head_seq(), before);
    assert!(
        derived_on(twin.root()).is_empty(),
        "store-only: nothing derived on the twin"
    );
    assert_eq!(
        refs_of(&recharged(&twin).1),
        refs_of(&primary),
        "and a main recharged from the twin is the primary"
    );

    // ── 2. held back again, and the operator's retention door drops the lot ─────
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    primary
        .repo_store(&app)
        .unwrap()
        .serve()
        .update_ref("refs/heads/in-the-hole", None, Some(&facts.oids[1]))
        .unwrap();
    primary
        .secrets()
        .unwrap()
        .put(SECRET_KEY, b"in the hole")
        .unwrap();
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
    let after = plane.log_retain(0).expect("the operator's door").first_seq;
    let head = primary.txlog().head_seq();
    assert!(
        after > before + 1,
        "retention really dropped frames the twin had not fetched: first {after}, twin at {before}"
    );

    // A source that cannot snapshot: a loud, named stop — never a skip.
    let bare = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    assert!(
        matches!(bare.step().unwrap(), Progress::SnapshotNeeded { .. }),
        "a hole is a stop, not a skip"
    );
    let s = status(twin.root()).expect("the twin says so");
    assert_eq!(s.state, State::SnapshotNeeded);
    assert!(
        s.why.contains(&format!("retains from {after}"))
            && s.why.contains(&format!("needs {} next", before + 1)),
        "{}",
        s.why
    );
    assert_eq!(
        twin.txlog().head_seq(),
        before,
        "nothing was applied past the hole"
    );

    // A source that can snapshot (as the production gRPC source can): the resync.
    wait_until_indexed(
        &primary,
        "before the twin fetches a snapshot across the retention hole",
    );
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at: head });
    assert_eq!(
        refs_of(&recharged(&twin).1),
        refs_of(&primary),
        "the snapshot carries what the hole held (read through a recharge)"
    );
    assert_eq!(
        twin.raw_secrets().get(SECRET_KEY).unwrap().as_deref(),
        primary.raw_secrets().get(SECRET_KEY).unwrap().as_deref()
    );
    primary
        .repo_store(&app)
        .unwrap()
        .serve()
        .update_ref("refs/heads/after-the-hole", None, Some(&facts.oids[0]))
        .unwrap();
    assert!(
        matches!(f.step().unwrap(), Progress::Applied { count: 1, .. }),
        "and it follows on"
    );
    row(
        "a primary keeps every frame its twin has not fetched across a restart; the operator's retention door is the one way to drop them, and the twin then stops loudly (snapshot-needed) or resyncs by snapshot",
        true,
        &format!("held back {} across a restart, all applied; retention to {after} with the twin at {before}", before - level),
    );
}

/// **`txlog.size_over_cap`: raised over the cap, clear under it, and nothing is deleted**
/// (2026-09-22, the owner's ruling). Driven through `Admin.Health`'s own source
/// (`CoreControlPlane::health_alarms` reads the same functions) on a real primary's log.
///
/// **Fails before the fix**: there was no cap and no alarm; the log grew until the disk was full
/// and `Admin.Health` said nothing about it.
#[test]
fn the_txlog_size_alarm_is_raised_over_the_cap_clear_under_it_and_deletes_nothing() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let log = primary.txlog();
    let (bytes, first, head) = (log.bytes(), log.first_seq(), log.head_seq());
    assert!(bytes > 0);

    let over = txlog_size_alarm(bytes, bytes - 1, "a test cap")
        .expect("one byte over the cap is an alarm");
    assert_eq!(over.event, TXLOG_SIZE_ALARM);
    assert!(
        over.detail.contains(&bytes.to_string())
            && over.detail.contains("Nothing has been deleted"),
        "{}",
        over.detail
    );
    assert!(
        txlog_size_alarm(bytes, bytes, "a test cap").is_none(),
        "at the cap: clear"
    );
    assert!(
        txlog_size_alarm(bytes, bytes * 2, "a test cap").is_none(),
        "under it: clear"
    );
    assert_eq!(
        (log.bytes(), log.first_seq(), log.head_seq()),
        (bytes, first, head),
        "the alarm touched nothing"
    );

    // The cap: the variable in bytes; unset, a quarter of the volume; garbage, said out loud.
    assert_eq!(
        txlog_alarm_cap(Some("1000"), 1 << 40).unwrap().unwrap().0,
        1000
    );
    assert_eq!(txlog_alarm_cap(None, 400).unwrap().unwrap().0, 100);
    assert_eq!(
        txlog_alarm_cap(Some(" "), 400).unwrap().unwrap().0,
        100,
        "empty = unset"
    );
    assert_eq!(
        txlog_alarm_cap(None, 0).unwrap(),
        None,
        "an unmeasurable volume has no default"
    );
    assert!(txlog_alarm_cap(Some("64GiB"), 400)
        .unwrap_err()
        .contains("NOT watched"));

    // And through the health source: this primary's log is far under a quarter of its volume.
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
    use gunnar_grpc::seam::ControlPlane as _;
    assert!(
        plane
            .health_alarms()
            .iter()
            .all(|a| a.event != TXLOG_SIZE_ALARM),
        "{:?}",
        plane.health_alarms()
    );
    assert_eq!(
        (log.bytes(), log.first_seq()),
        (bytes, first),
        "and asking deleted nothing"
    );
}

// ── what store-only saves, measured on the TD8 workload (ruling 2026-09-22) ──

/// Whole-process CPU (user + system, every thread — the engine's resolver included), in ms,
/// from `/proc/self/stat` (fields 14 and 15, clock ticks of 10 ms).
fn process_cpu_ms() -> u64 {
    let stat = std::fs::read_to_string("/proc/self/stat").unwrap();
    let rest = &stat[stat.rfind(')').unwrap() + 2..];
    let f: Vec<&str> = rest.split_whitespace().collect();
    // `rest` starts at field 3 (state), so utime (14) is index 11 and stime (15) index 12.
    (f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap()) * 10
}

/// Bytes a data root holds on disk (allocated blocks), and how many of them are the log.
fn disk_bytes(root: &std::path::Path) -> (u64, u64) {
    use std::os::unix::fs::MetadataExt as _;
    let (mut all, mut log) = (0u64, 0u64);
    for rel in files_under(root) {
        let b = std::fs::metadata(root.join(&rel))
            .map(|m| m.blocks() * 512)
            .unwrap_or(0);
        all += b;
        if rel.starts_with("txlog/") {
            log += b;
        }
    }
    (all, log)
}

/// **The TD8 workload, in process**: growbench `seed_repos` (the repositories TD8 pushes to
/// main — `mixed`: 1 KiB, 100 KiB, chunk−1, chunk+1, 3×chunk, 3 MiB, 20 MiB, 700 KiB; `big`:
/// 8 × 24 MiB; incompressible bytes, chunk = the RAID0 chunk of 512 KiB) and TD8 step 5's
/// control plane (a user with a key, group `twin-reviewers` with a member, namespace
/// `rehearsal`, a namespace grant, a repository made public). One pack per repository, one
/// ref per file, as a push of each repository lands.
fn td8_primary(dir: &std::path::Path) -> Arc<ServerStores> {
    let stores = Arc::new(ServerStores::open(dir).expect("primary"));
    stores.catalog().create_namespace("bench").unwrap();
    let c: usize = 512 << 10;
    let mut seed: u64 = 0x7d08_0001;
    let mut bytes = |n: usize| -> Vec<u8> {
        let mut v = vec![0u8; n];
        for chunk in v.chunks_mut(8) {
            seed ^= seed << 13;
            seed ^= seed >> 7;
            seed ^= seed << 17;
            chunk.copy_from_slice(&seed.to_le_bytes()[..chunk.len()]);
        }
        v
    };
    let sets: [(&str, Vec<usize>); 2] = [
        (
            "mixed",
            vec![
                1024,
                100 << 10,
                c - 1,
                c + 1,
                3 * c,
                3 << 20,
                20 << 20,
                700 << 10,
            ],
        ),
        ("big", vec![24 << 20; 8]),
    ];
    for (name, sizes) in sets {
        let repo = stores
            .create_repo(
                NewRepo::new(RepoRef::new("bench", name).unwrap()).hash_kind(HashKind::Sha256),
                false,
            )
            .unwrap();
        let store = stores.repo_store(&repo).unwrap();
        let hash = store.hash_kind();
        let mut pb = PackBuilder::new(hash);
        let mut oids = Vec::new();
        for n in sizes {
            let content = bytes(n);
            oids.push(
                gix_object::compute_hash(hash, ObjKind::Blob, &content)
                    .unwrap()
                    .as_bytes()
                    .to_vec(),
            );
            pb.object(ObjKind::Blob, content);
        }
        store.serve().put_pack(&pb.build().unwrap().bytes).unwrap();
        for (i, o) in oids.iter().enumerate() {
            store
                .serve()
                .update_ref(&format!("refs/heads/f{i}"), None, Some(o))
                .unwrap();
        }
    }
    let a = stores.auth();
    a.add_principal(Principal::new("carol", "carol", PrincipalKind::User).unwrap())
        .unwrap();
    a.register_key("carol", &a_public_key_line(), None).unwrap();
    a.add_principal(
        Principal::new("twin-reviewers", "twin-reviewers", PrincipalKind::Group).unwrap(),
    )
    .unwrap();
    a.add_to_group("twin-reviewers", "carol").unwrap();
    stores.catalog().create_namespace("rehearsal").unwrap();
    a.grant(gunnar_core::Grant::namespace(
        "twin-reviewers",
        "rehearsal",
        gunnar_core::Level::Write,
    ))
    .unwrap();
    stores
        .catalog()
        .set_visibility(
            &RepoRef::new("bench", "mixed").unwrap(),
            gunnar_core::Visibility::Public,
        )
        .unwrap();
    stores
}

/// **What a twin spends per frame, before and after "the twin only stores"**, on the TD8
/// workload. The same frames, read once into memory, are taken by two twins:
///
/// * BEFORE — `append_at` then [`apply`] for every frame: exactly master `fea5fd5`'s
///   `Follower::step_inner` body, plus the wait for the engine's resolver to finish the packs
///   it was handed (the resolver is part of an apply's cost, it just runs on its own thread);
/// * AFTER — `append_at` then [`take`]: this branch's follower.
///
/// CPU is the whole process (every thread) from `/proc/self/stat`; disk is allocated blocks
/// under each twin's data root. Run it:
/// `cargo test -p gunnar-server --lib twin_cost_per_frame -- --ignored --nocapture`
#[test]
#[ignore = "a measurement, not a check: run by name with --ignored --nocapture"]
fn twin_cost_per_frame_on_the_td8_workload() {
    let pd = tempfile::tempdir().unwrap();
    let primary = td8_primary(pd.path());
    let log = primary.txlog();
    let frames = log.range(log.first_seq(), log.head_seq()).unwrap();
    let data = frames.iter().filter(|e| !twin_applies(e.kind)).count();
    let blob: u64 = frames.iter().map(|e| e.blob.len() as u64).sum();

    let run = |label: &str, store_only: bool| -> (u64, u64, u64) {
        let td = tempfile::tempdir().unwrap();
        let twin = open_twin(td.path());
        let t0 = process_cpu_ms();
        let w0 = std::time::Instant::now();
        for e in &frames {
            twin.txlog()
                .append_at(e.seq, e.kind, &e.meta, &e.blob)
                .unwrap();
            if store_only {
                take(&twin, e).unwrap()
            } else {
                apply(&twin, e).unwrap()
            };
        }
        if !store_only {
            wait_until_indexed(&twin, "the BEFORE twin's resolver");
        }
        let cpu = process_cpu_ms() - t0;
        let wall = w0.elapsed().as_millis() as u64;
        let (disk, in_log) = disk_bytes(twin.root());
        println!(
            "TWIN-COST {label}: frames={} cpu_ms={cpu} cpu_us_per_frame={} wall_ms={wall} disk_bytes={disk} disk_bytes_per_frame={} of_which_log={in_log} derived_files={}",
            frames.len(),
            cpu * 1000 / frames.len() as u64,
            disk / frames.len() as u64,
            derived_on(twin.root()).len()
        );
        (cpu, disk, wall)
    };
    let before = run("BEFORE (apply every frame, master fea5fd5)", false);
    let after = run("AFTER  (store-only, this branch)", true);
    println!(
        "TWIN-COST workload: {} frames ({} data frames kept, {} control frames applied), {} blob bytes; \
         CPU {} ms -> {} ms ({:.1}% less); disk {} -> {} bytes ({:.1}% less)",
        frames.len(),
        data,
        frames.len() - data,
        blob,
        before.0,
        after.0,
        100.0 * (before.0.saturating_sub(after.0)) as f64 / before.0.max(1) as f64,
        before.1,
        after.1,
        100.0 * (before.1.saturating_sub(after.1)) as f64 / before.1.max(1) as f64
    );
    assert!(after.1 < before.1, "store-only must use less disk");
}

/// **The follower's runtime is dropped where blocking is allowed.**
///
/// MEASURED 2026-09-23 on the live gunnar.rs twin: the box printed "following
/// the primary" and "this twin's client certificate is good for 364 more
/// day(s)", then
///
/// ```text
/// gunnar PANIC (exit 101): panicked at tokio/src/runtime/blocking/shutdown.rs:51:21:
/// Cannot drop a runtime in a context where blocking is not allowed.
/// Kernel panic - not syncing: Attempted to kill init! exitcode=0x00006500
/// ```
///
/// gunnar is PID 1 on the appliance, so the panic took the kernel down with it.
/// This test is that panic: with `GrpcSource`'s runtime held as a plain
/// `tokio::runtime::Runtime` it panics here exactly as it did on the box, and
/// it passes only because `OffThreadRuntime` moves the drop onto a thread of
/// its own.
///
/// It builds the runtime directly rather than a whole `GrpcSource`, because
/// `GrpcSource::connect` dials a primary and the defect is entirely in how the
/// runtime is HELD — not in anything on the wire.
#[tokio::test]
async fn the_followers_runtime_is_dropped_off_the_async_thread() {
    let inner = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("a current-thread runtime");
    let held = super::OffThreadRuntime::new(inner);
    // Inside `#[tokio::test]` this IS an async context: a plain Runtime dropped
    // here is the panic the twin died of.
    drop(held);
}

/// **The shape this replaced, kept as a test, so the defect cannot come back
/// unnoticed.**
///
/// A plain `tokio::runtime::Runtime` dropped in an async context panics with
/// the exact line the live twin died on. If a later change ever holds the
/// follower's runtime plainly again, the test above stops being the one that
/// passes and this one stops being the one that panics — and both say why.
#[tokio::test]
#[should_panic(expected = "Cannot drop a runtime in a context where blocking is not allowed")]
async fn a_plain_runtime_dropped_in_an_async_context_is_the_panic_the_twin_died_of() {
    let inner = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    drop(inner);
}

/// Fix 2026-09-24: the twin prefers the overlay and falls back to the
/// utility network when the tunnel's door does not answer.
#[test]
fn the_twin_dials_the_first_replication_host_that_answers() {
    let utility = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = utility.local_addr().unwrap().port();
    let tunnel_first = vec!["127.0.0.2".to_string(), "127.0.0.1".to_string()];
    assert_eq!(
        pick_replication_dial(&tunnel_first, port, Duration::from_millis(500)),
        format!("127.0.0.1:{port}"),
        "tunnel down ⇒ utility"
    );
    let tunnel = std::net::TcpListener::bind(("127.0.0.2", port)).unwrap();
    assert_eq!(
        pick_replication_dial(&tunnel_first, port, Duration::from_millis(500)),
        format!("127.0.0.2:{port}"),
        "tunnel up ⇒ the overlay first"
    );
    drop((tunnel, utility));
    assert_eq!(
        pick_replication_dial(&tunnel_first, port, Duration::from_millis(200)),
        format!("127.0.0.2:{port}"),
        "none answers ⇒ the first, and the follower retries"
    );
}

/// Bytes that do not compress: a pack of them carries them (nearly) verbatim, so a
/// window of the pack is a needle the segment files can be searched for.
fn noise(seed: u64, n: usize) -> Vec<u8> {
    let mut x = seed
        .wrapping_mul(6364136223846793005)
        .wrapping_add(1442695040888963407);
    (0..n)
        .map(|_| {
            x = x
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (x >> 33) as u8
        })
        .collect()
}

fn segment_files_hold(root: &std::path::Path, needle: &[u8]) -> bool {
    let dir = root.join(crate::stores::TXLOG_DIR);
    std::fs::read_dir(&dir).unwrap().any(|e| {
        let p = e.unwrap().path();
        p.extension().and_then(|x| x.to_str()) == Some("log")
            && std::fs::read(&p)
                .unwrap()
                .windows(needle.len())
                .any(|w| w == needle)
    })
}

/// ★ **g-nonpay (owner ruling 2026-09-29): an account deleted for non-payment is gone
/// from main AND from the twin — the stores, and the log that is the twin's copy.**
///
/// `Accounts.Purge` emptied the disk on main and left every pack and LFS object the
/// account ever pushed in the transaction log: on main until retention trimmed it, and
/// on the twin for ever, because the twin keeps the log. The purge now frames an
/// `Erase` and erases the account's frames in place on main; the twin, storing that
/// frame, erases its own. A recharge from the twin still rebuilds every OTHER account.
#[test]
fn an_account_purge_erases_the_accounts_bytes_from_the_log_on_main_and_on_the_twin() {
    use crate::account_purge::{purge_account, PurgeRequest};
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    // ns/app gets a pack whose bytes can be searched for.
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let (secret_pack, secret_oid) = {
        let store = primary.repo_store(&app).unwrap();
        let (p, o) = one_blob_pack(store.hash_kind(), &noise(1, 64 << 10));
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref("refs/heads/secret", None, Some(&o))
            .unwrap();
        (p, o)
    };
    let needle = secret_pack[secret_pack.len() / 2..secret_pack.len() / 2 + 64].to_vec();
    // Somebody else, who keeps everything.
    primary.catalog().create_namespace("keeper").unwrap();
    let keeper_ref = RepoRef::new("keeper", "lib").unwrap();
    let keeper = primary
        .create_repo(
            NewRepo::new(keeper_ref.clone()).hash_kind(HashKind::Sha256),
            false,
        )
        .unwrap();
    let (keeper_pack, keeper_oid) = {
        let store = primary.repo_store(&keeper).unwrap();
        let (p, o) = one_blob_pack(store.hash_kind(), &noise(2, 64 << 10));
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref("refs/heads/main", None, Some(&o))
            .unwrap();
        (p, o)
    };
    let keeper_needle = keeper_pack[keeper_pack.len() / 2..keeper_pack.len() / 2 + 64].to_vec();

    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    f.step().unwrap();
    assert!(
        segment_files_hold(pd.path(), &needle) && segment_files_hold(td.path(), &needle),
        "the fixture: both logs hold ns's pack"
    );

    let report = purge_account(
        &primary,
        PurgeRequest {
            account: "ns".into(),
            principal_id: "monetize".into(),
            reason: "nonpay: lapsed".into(),
            remove_principal: false,
            expect_empty: false,
        },
        &mut |_| {},
    )
    .unwrap();
    assert!(report.failures.is_empty(), "{report:?}");
    assert!(
        report.log_frames_erased >= 3,
        "the packs, the refs and the LFS object of ns/app: {report:?}"
    );

    // MAIN: the store and the log.
    assert!(!primary
        .repo_dir(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .exists());
    assert!(
        !segment_files_hold(pd.path(), &needle),
        "main's log no longer holds the account's pack"
    );
    assert!(
        segment_files_hold(pd.path(), &keeper_needle),
        "and still holds everybody else's"
    );

    // THE TWIN: it stores the Erase frame and erases its own log.
    f.step().unwrap();
    assert_eq!(twin.txlog().head_seq(), primary.txlog().head_seq());
    assert!(
        !segment_files_hold(td.path(), &needle),
        "the twin's log no longer holds the account's pack"
    );
    assert!(
        segment_files_hold(td.path(), &keeper_needle),
        "the twin still holds everybody else's"
    );
    let erased = twin
        .txlog()
        .range(1, twin.txlog().head_seq())
        .unwrap()
        .iter()
        .filter(|e| e.kind == Kind::Redacted as u16)
        .count();
    assert!(
        erased >= 3,
        "the twin's frames of ns/app read as erased: {erased}"
    );

    // A recharge from the twin: the other account is whole, the erased one is not there.
    let (_fd, main) = recharged(&twin);
    assert!(
        !main
            .repo_dir(&RepoRef::new("ns", "app").unwrap())
            .unwrap()
            .exists(),
        "nothing of ns comes back"
    );
    let k = main
        .catalog()
        .get(&keeper_ref)
        .unwrap()
        .expect("keeper/lib is restored");
    let store = main.repo_store(&k).unwrap();
    let refs: Vec<(String, Option<Vec<u8>>)> = store
        .serve()
        .refs()
        .unwrap()
        .into_iter()
        .map(|r| (r.name, r.oid))
        .collect();
    assert!(
        refs.contains(&("refs/heads/main".to_string(), Some(keeper_oid.clone()))),
        "keeper's ref is restored: {refs:?}"
    );
    let _ = secret_oid;
    row("an account purge erases the account's frames from the log on main and on the twin; a recharge still restores every other account", true, &format!("{} frame(s), {} bytes", report.log_frames_erased, report.log_bytes_erased));
}

/// **M1 (release-3 test2 mock 2026-09-29): a primary growing its data set answers the twin's
/// `Mirror.LogFetch` with a NAMED, retryable refusal.** Its log was released for the remount and
/// /data was off: every fetch answered `Internal: transaction log: txlog: No such file or
/// directory (os error 2)` — 7/min `WARN a control-plane call was refused … grpc_status=13` on
/// main, and the twin's poll loop took it as a fault. The answer is `Unavailable` carrying
/// [`gunnar_grpc::PRIMARY_GROWING`], before the log is touched.
#[test]
fn a_primary_growing_its_set_answers_log_fetch_unavailable_by_name() {
    use gunnar_grpc::ControlPlane as _;
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    assert!(
        primary.txlog().head_seq() >= 1,
        "the fixture wrote something"
    );
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
    primary.txlog().release().unwrap();
    // The unmount: the log's files are gone from under it.
    let dir = primary.txlog().dir().to_path_buf();
    let aside = dir.with_extension("unmounted");
    std::fs::rename(&dir, &aside).unwrap();
    match plane.log_fetch(1, 1, 1 << 20) {
        Err(gunnar_grpc::ControlError::Unavailable(why)) => {
            assert!(why.contains(gunnar_grpc::PRIMARY_GROWING), "{why}")
        }
        other => panic!(
            "a growing primary answers Unavailable {}, not {other:?}",
            gunnar_grpc::PRIMARY_GROWING
        ),
    }
    std::fs::rename(&aside, &dir).unwrap();
    primary.txlog().reacquire().unwrap();
    assert!(
        plane.log_fetch(1, 1, 1 << 20).is_ok(),
        "after the remount the log reads again"
    );
}

/// **Round 4's fresh UpCloud mockup: a primary growing its data set answers `Usage.Read` with the
/// same NAMED, retryable refusal.** /data was unmounted for the remount and the read answered
/// `Internal: resolving the directory for nordisk/gunnar: path escape: /data/repos the repository
/// root does not resolve` — a 502 at the console, which no caller waits out. Like `LogFetch`
/// above: `Unavailable` carrying [`gunnar_grpc::PRIMARY_GROWING`], before /data is touched.
#[test]
fn a_primary_growing_its_set_answers_usage_unavailable_by_name() {
    use gunnar_grpc::ControlPlane as _;
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&primary));
    assert!(
        plane.usage(None).is_ok(),
        "a serving primary reads its usage"
    );
    primary.txlog().release().unwrap();
    match plane.usage(None) {
        Err(gunnar_grpc::ControlError::Unavailable(why)) => {
            assert!(why.contains(gunnar_grpc::PRIMARY_GROWING), "{why}")
        }
        other => panic!(
            "a growing primary answers Usage.Read Unavailable {}, not {other:?}",
            gunnar_grpc::PRIMARY_GROWING
        ),
    }
    primary.txlog().reacquire().unwrap();
    assert!(
        plane.usage(None).is_ok(),
        "after the remount the usage reads again"
    );
}

/// A primary that is growing for `growing` fetches, then serves.
struct GrowingPrimary {
    log: Arc<Log>,
    growing: AtomicUsize,
}

impl LogSource for GrowingPrimary {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        self.log.head()
    }
    fn alternatives(&self) -> bool {
        true
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        if self.growing.load(Ordering::SeqCst) > 0 {
            self.growing.fetch_sub(1, Ordering::SeqCst);
            anyhow::bail!(
                "status: 'Service unavailable', message: \"{} this box's data set is being grown\"",
                gunnar_grpc::PRIMARY_GROWING
            );
        }
        self.log.fetch(from, to, max_bytes)
    }
}

/// **M2 (the same mock): the twin WAITS for a growing primary, quietly, and its loop never
/// returns on it.** Every refused fetch was `WARN poll failed; retrying` every 2 s (7/min), and
/// after five the loop handed back to the supervisor for a re-dial, which said `ERROR the twin's
/// poll loop returned without being asked to stop`. A growing primary is said ONCE at INFO, the
/// twin backs off, is not re-dialled, and follows on when the primary serves again.
#[test]
fn the_twin_waits_for_a_growing_primary_quietly_and_never_returns_for_it() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    let head = primary.txlog().head_seq();
    let source = GrowingPrimary {
        log: Arc::clone(primary.txlog()),
        growing: AtomicUsize::new(8),
    };
    let follower = Follower::new(Arc::clone(&twin), Box::new(source));

    #[derive(Clone, Default)]
    struct Heard(Arc<std::sync::Mutex<Vec<u8>>>);
    impl std::io::Write for Heard {
        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(b);
            Ok(b.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    let heard = Heard::default();
    let h = heard.clone();
    let sub = tracing_subscriber::fmt()
        .with_writer(move || h.clone())
        .with_ansi(false)
        .finish();
    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let (s2, t2) = (Arc::clone(&stop), Arc::clone(&twin));
    let stopper = std::thread::spawn(move || {
        let t0 = std::time::Instant::now();
        while t2.txlog().head_seq() < head && t0.elapsed() < std::time::Duration::from_secs(20) {
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        s2.store(true, Ordering::SeqCst);
    });
    tracing::subscriber::with_default(sub, || {
        follower.run(std::time::Duration::from_millis(10), &stop)
    });
    let stopped_first = stop.load(Ordering::SeqCst);
    stopper.join().unwrap();
    let said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
    assert!(
        stopped_first,
        "the loop returned before it was asked to stop:\n{said}"
    );
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "the twin followed on after the growth:\n{said}"
    );
    assert!(
        !said.contains(" WARN ") && !said.contains(" ERROR "),
        "a growing primary is no WARN and no ERROR:\n{said}"
    );
    assert_eq!(
        said.matches("is growing its data set").count(),
        1,
        "said once, at INFO:\n{said}"
    );
}

/// **M2, the supervisor's ERROR**: a poll loop that hands back to have the primary re-dialled
/// on its other address says so in what it returns, so the supervisor never calls it "returned
/// without being asked to stop" (ERROR on the mock, attempt=3).
#[test]
fn a_poll_loop_that_hands_back_for_a_redial_says_so() {
    struct Down;
    impl LogSource for Down {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            anyhow::bail!("status: 'Unknown error', message: \"transport error\"")
        }
        fn alternatives(&self) -> bool {
            true
        }
        fn fetch(&self, _: u64, _: u64, _: u64) -> anyhow::Result<Vec<Entry>> {
            anyhow::bail!("status: 'Unknown error', message: \"transport error\"")
        }
    }
    let td = tempfile::tempdir().unwrap();
    let twin = open_twin(td.path());
    let never = std::sync::atomic::AtomicBool::new(false);
    assert_eq!(
        Follower::new(twin, Box::new(Down)).run(std::time::Duration::from_millis(1), &never),
        Ended::Redial
    );
}

/// **T1 (estate-box check 2, release-3)**: a twin that has just applied what its
/// primary had asks the head AGAIN AT ONCE, not a poll later. The primary's view of
/// the twin (the dual heartbeat) is the report carried on each `LogHead`, taken
/// BEFORE that poll's fetch — so a twin that fetched and then slept its 2 s poll
/// was reported `lag N` for 2 s after it was level, and a push the twin had in
/// hand read as a lagging twin (MEASURED: `applied 280 of 288 (lag 8), heard
/// 1954 ms ago` 10 s after the push). The re-poll after progress carries the new
/// position; a CAUGHT-UP poll then waits the interval as before.
#[test]
fn a_twin_that_applied_asks_again_at_once_so_its_report_is_level() {
    struct Counting {
        inner: Arc<Log>,
        heads: Arc<AtomicUsize>,
    }
    impl LogSource for Counting {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.heads.fetch_add(1, Ordering::SeqCst);
            self.inner.head()
        }
        fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
            self.inner.fetch(from, to, max_bytes)
        }
    }
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let head = primary.txlog().head_seq();
    assert!(head > 0);
    let twin = open_twin(td.path());
    let heads = Arc::new(AtomicUsize::new(0));
    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(Counting {
            inner: Arc::clone(primary.txlog()),
            heads: Arc::clone(&heads),
        }),
    );
    // A long poll: anything the test sees inside it was NOT the poll's doing.
    let poll = std::time::Duration::from_secs(30);
    let (s2, t0) = (Arc::clone(&stop), std::time::Instant::now());
    let runner = std::thread::spawn(move || f.run(poll, &s2));
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
    while heads.load(Ordering::SeqCst) < 2 && std::time::Instant::now() < deadline {
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
    let asked = heads.load(Ordering::SeqCst);
    stop.store(true, Ordering::SeqCst);
    runner.join().unwrap();
    assert_eq!(twin.txlog().head_seq(), head, "the twin applied everything");
    assert!(asked >= 2, "after applying {head} entries the twin asked the head {asked} time(s) in {:?}: its report stays a poll ({poll:?}) behind", t0.elapsed());
    assert!(
        t0.elapsed() < std::time::Duration::from_secs(20),
        "the stop was heard inside the poll, not after it ({:?})",
        t0.elapsed()
    );
}

/// **T1, the silence (estate-box check 2)**: a primary whose connection went
/// quiet — vm1's relay restarted under the tunnel, the twin's TCP segments lost
/// and retransmitted on an exponential back-off — held the twin's `LogHead` for
/// 8 s (13 s on test2, 57 s once), and nothing bounded it: no deadline on the
/// call, the same stale connection reused. MEASURED: main heard the twin's
/// LogHead at 02:48:41.0 and next at 02:48:49.5, a push landing in between.
/// A `LogHead` now answers within [`HEAD_WITHIN`] or fails as STALLED.
#[test]
fn a_primary_that_accepts_and_never_answers_fails_the_head_within_its_deadline() {
    let quiet = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let at = quiet.local_addr().unwrap().to_string();
    // Accepts (the kernel does) and never says a word.
    let _hold = std::thread::spawn(move || {
        let mut held = Vec::new();
        for c in quiet.incoming() {
            held.push(c);
        }
    });
    let td = tempfile::tempdir().unwrap();
    let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
        .unwrap();
    let (tx, rx) = std::sync::mpsc::channel();
    let root = td.path().to_path_buf();
    std::thread::spawn(move || {
        let source = GrpcSource::connect(&at, key, &root).unwrap();
        let t = std::time::Instant::now();
        let r = source.head().map(|_| ()).map_err(|e| format!("{e:#}"));
        let _ = tx.send((r, t.elapsed()));
    });
    let (r, took) = rx
        .recv_timeout(HEAD_WITHIN + std::time::Duration::from_secs(5))
        .expect("the head call never returned: no deadline");
    let err = r.expect_err("a primary that says nothing is not an answer");
    assert!(err.contains(STALLED), "{err}");
    assert!(
        took < HEAD_WITHIN + std::time::Duration::from_secs(1),
        "{took:?}"
    );
}

/// ...and the loop does not wait for [`FAILOVER_AFTER`] such polls: a STALLED
/// call hands back AT ONCE for a re-dial (a fresh connection, the first
/// address that answers), because the one it has is the thing that is broken.
#[test]
fn a_stalled_poll_hands_back_for_a_redial_at_once() {
    struct Stalled(std::sync::atomic::AtomicU32);
    impl LogSource for Stalled {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            anyhow::bail!("{STALLED}: the primary said nothing within 3 s")
        }
        fn fetch(&self, _: u64, _: u64, _: u64) -> anyhow::Result<Vec<Entry>> {
            unreachable!()
        }
    }
    let td = tempfile::tempdir().unwrap();
    let twin = open_twin(td.path());
    let never = std::sync::atomic::AtomicBool::new(false);
    let source = Arc::new(Stalled(std::sync::atomic::AtomicU32::new(0)));
    struct Shared(Arc<Stalled>);
    impl LogSource for Shared {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.0.head()
        }
        fn fetch(&self, a: u64, b: u64, c: u64) -> anyhow::Result<Vec<Entry>> {
            self.0.fetch(a, b, c)
        }
    }
    let _ = never;
    // Bounded: a loop that never hands back is stopped after 5 s and says Stopped.
    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let s2 = Arc::clone(&stop);
    std::thread::spawn(move || {
        std::thread::sleep(std::time::Duration::from_secs(5));
        s2.store(true, std::sync::atomic::Ordering::SeqCst);
    });
    let t = std::time::Instant::now();
    assert_eq!(
        Follower::new(twin, Box::new(Shared(Arc::clone(&source))))
            .run(std::time::Duration::from_millis(50), &stop),
        Ended::Redial
    );
    assert_eq!(
        source.0.load(std::sync::atomic::Ordering::SeqCst),
        1,
        "one stalled call is enough"
    );
    assert!(
        t.elapsed() < std::time::Duration::from_secs(1),
        "{:?}",
        t.elapsed()
    );
}

/// ★ estate-box3 plain6 R1 (MEASURED 2026-09-30, the recharged main): the
/// first Restore stream announced the twin's head 1078 and broke at seq 21
/// (refused, the box stayed restoring); the second broke at 46 (refused); the
/// THIRD carried nothing — the twin's fetch failed before its first entry, so
/// no chunk announced a head or a log id — and `restore_finish(0, "")` saw a
/// box holding 46 entries and FINISHED it: the main restarted at seq 46 of
/// 1078, minted a log id of its own, the twin stopped (DIVERGED, "two
/// different histories") and the probe repository was gone. What a restore
/// was once told it is restoring (the head, the log id) holds for every
/// resend until it finishes.
#[test]
fn a_resend_that_carries_nothing_does_not_finish_a_restore_an_earlier_stream_announced_longer() {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let log = primary.txlog();
    let (first, head) = (log.first_seq(), log.head_seq());
    assert!(first == 1 && head >= 3, "{first}..={head}");
    let entries = log.range(first, head).unwrap();
    let short = head - 1;
    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(restore_begin(&fresh).unwrap(), 1);
    for e in entries.iter().take(short as usize) {
        restore_apply(&fresh, e).unwrap();
    }
    // The first stream announced the head and the source's log id, then broke.
    assert!(restore_finish(&fresh, head, "the-twins-log")
        .unwrap_err()
        .to_string()
        .contains("short stream"));
    drop(fresh);
    // The resend: the sender's fetch failed before its first entry — nothing announced.
    let again = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(restore_begin(&again).unwrap(), short + 1);
    let err = restore_finish(&again, 0, "")
        .expect_err("an empty resend must not finish a restore announced to reach further");
    assert!(err.to_string().contains("short stream"), "{err}");
    assert!(is_restoring(fd.path()), "the box stays restoring");
    // The resend that carries the rest finishes it, as the log it was announced to be.
    for e in entries.iter().skip(short as usize) {
        restore_apply(&again, e).unwrap();
    }
    assert_eq!(restore_finish(&again, 0, "").unwrap(), head);
    assert_eq!(
        std::fs::read_to_string(fd.path().join(LOG_ID))
            .map(|s| s.trim().to_string())
            .ok()
            .as_deref(),
        Some("the-twins-log"),
        "the box took the announced log's identity"
    );
}

// ── streamed entries (twin-streaming, 2026-10-06) ───────────────────────────

/// A bare primary log of Pack frames (a twin stores them and applies nothing of them):
/// `sizes` are the blobs' lengths, each blob a pattern of its own sequence number.
fn bare_primary(dir: &std::path::Path, sizes: &[usize]) -> Arc<Log> {
    let log = Arc::new(Log::open(dir.join("txlog")).unwrap());
    for (i, &n) in sizes.iter().enumerate() {
        let blob: Vec<u8> = (0..n).map(|b| ((b * 31 + i * 7) % 251) as u8).collect();
        log.append(
            gunnar_core::txlog::Kind::Pack,
            &Meta::Pack {
                repo: REPO.into(),
                hash: "sha1".into(),
            },
            &blob,
        )
        .unwrap();
    }
    log
}

/// The bytes of this box's open segment(s) — what a half frame left behind would show in.
fn txlog_bytes_on_disk(stores: &ServerStores) -> u64 {
    std::fs::read_dir(stores.root().join(crate::stores::TXLOG_DIR))
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|x| x == "log"))
        .map(|e| e.metadata().unwrap().len())
        .sum()
}

/// Every frame of `twin` is the primary's, byte for byte, up to `through`.
fn assert_frames_equal(primary: &Log, twin: &ServerStores, through: u64) {
    assert_eq!(twin.txlog().head_seq(), through);
    for seq in 1..=through {
        assert_eq!(
            twin.txlog().read(seq).unwrap(),
            primary.read(seq).unwrap(),
            "seq {seq}"
        );
    }
}

/// A primary whose wire breaks INSIDE an entry: it streams off the real log and cuts the
/// connection after `cut_after` blob bytes of the first entry larger than that, `cuts` times.
struct CutMidEntry {
    inner: Arc<Log>,
    cut_after: u64,
    cuts: AtomicUsize,
}

struct Cutting<'a> {
    into: &'a mut dyn FrameSink,
    cut_after: u64,
    sent: u64,
}

impl FrameSink for Cutting<'_> {
    fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
        self.sent = 0;
        self.into.begin(seq, kind, meta, blob_len)
    }
    fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
        if self.sent + bytes.len() as u64 > self.cut_after {
            let keep = (self.cut_after - self.sent) as usize;
            self.into.piece(&bytes[..keep])?;
            anyhow::bail!(
                "connection reset inside an entry, {} blob bytes in",
                self.cut_after
            );
        }
        self.sent += bytes.len() as u64;
        self.into.piece(bytes)
    }
    fn end(&mut self) -> anyhow::Result<()> {
        self.into.end()
    }
}

impl LogSource for CutMidEntry {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        self.inner.head()
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        self.inner.fetch(from, to, max_bytes)
    }
    fn fetch_into(
        &self,
        from: u64,
        to: u64,
        max_bytes: u64,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        if self.cuts.load(Ordering::SeqCst) == 0 {
            return self.inner.fetch_into(from, to, max_bytes, into);
        }
        self.cuts.fetch_sub(1, Ordering::SeqCst);
        stream_range(
            &self.inner,
            from,
            to,
            max_bytes,
            &mut Cutting {
                into,
                cut_after: self.cut_after,
                sent: 0,
            },
        )
        .map(drop)
    }
}

/// **A cut inside an entry leaves nothing of it, and the next poll takes it whole** — the resume.
/// The wire breaks 2.5 MiB into a 5 MiB entry: the entries before it are durable and taken, the
/// half entry is NOT in the log and not on the disk (the segment ends exactly where the last
/// whole frame does), the status says stalled; the next poll asks again from the twin's own head
/// and the copy ends byte-identical to the primary's.
#[test]
fn a_cut_inside_an_entry_leaves_nothing_of_it_and_the_next_poll_takes_it_whole() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let primary = bare_primary(pd.path(), &[100, 5 << 20, 3, 2 << 20]);
    let twin = open_twin(td.path());
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(CutMidEntry {
            inner: Arc::clone(&primary),
            cut_after: (5 << 20) / 2,
            cuts: AtomicUsize::new(1),
        }),
    );
    let err = f.step().unwrap_err();
    assert!(
        format!("{err:#}").contains("connection reset inside an entry"),
        "{err:#}"
    );
    assert_frames_equal(&primary, &twin, 1);
    let whole_frames = {
        let at = twin.txlog().locate_blob(1).unwrap();
        at.blob_offset + at.blob_len + 4
    };
    assert_eq!(
        txlog_bytes_on_disk(&twin),
        whole_frames,
        "not one byte of the cut entry is on the twin's disk"
    );
    assert_eq!(status(twin.root()).unwrap().state, State::Stalled);
    // The next poll: the same source, its wire whole again.
    assert_eq!(f.step().unwrap(), Progress::Applied { count: 3, head: 4 });
    assert_frames_equal(&primary, &twin, 4);
    // And the copy survives a restart byte for byte (no half frame hiding behind the index).
    drop(f);
    drop(twin);
    let twin = open_twin(td.path());
    assert_frames_equal(&primary, &twin, 4);
    row(
        "a cut inside an entry leaves nothing of it; the next poll takes it whole",
        true,
        "cut 2.5 of 5 MiB",
    );
}

/// **An entry whose bytes rotted on the primary is never appended on the twin.** The primary
/// checks the payload crc before the entry's LAST piece leaves; the twin commits only on the
/// last piece. So the twin holds everything before the rotten entry, nothing of it, and stops
/// — stalled, by name — rather than keep a copy the primary itself would refuse to read.
#[test]
fn an_entry_that_fails_its_crc_on_the_primary_is_never_appended_on_the_twin() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let primary = bare_primary(pd.path(), &[100, 3 << 20, 7]);
    let at = primary.locate_blob(2).unwrap();
    {
        use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
        let mut file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&at.path)
            .unwrap();
        let off = at.blob_offset + (2 << 20) + 17;
        file.seek(SeekFrom::Start(off)).unwrap();
        let mut b = [0u8; 1];
        file.read_exact(&mut b).unwrap();
        file.seek(SeekFrom::Start(off)).unwrap();
        file.write_all(&[b[0] ^ 0x5A]).unwrap();
    }
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(&primary)));
    let err = f.step().unwrap_err();
    assert!(format!("{err:#}").contains("payload crc"), "{err:#}");
    assert_frames_equal(&primary, &twin, 1);
    let at1 = twin.txlog().locate_blob(1).unwrap();
    assert_eq!(
        txlog_bytes_on_disk(&twin),
        at1.blob_offset + at1.blob_len + 4,
        "nothing of the rotten entry is on the twin's disk"
    );
    let s = status(twin.root()).unwrap();
    assert_eq!(s.state, State::Stalled);
    assert!(s.why.contains("payload crc"), "{}", s.why);
    row(
        "an entry that fails its crc on the primary is never appended on the twin",
        true,
        "seq 2 of 3 rotted",
    );
}

/// What one chunk sequence leaves in a fresh twin's log, through the follower's own reader of the
/// wire ([`ChunkFeed`] into the follower's [`Appender`]): `Ok(head)` or the refusal.
fn feed_twin(twin: &ServerStores, chunks: &[gunnar_proto::LogChunk]) -> Result<u64, String> {
    let reapply = AtomicBool::new(false);
    let marked = std::sync::atomic::AtomicU64::new(0);
    let group = twin.is_twin().then(Group::new);
    let mut into = Appender {
        stores: twin,
        reapply: &reapply,
        open: None,
        next: twin.txlog().head_seq() + 1,
        count: 0,
        group,
        marked: &marked,
    };
    let mut feed = ChunkFeed::default();
    let r = chunks
        .iter()
        .try_for_each(|c| feed.take(c, &mut into))
        .and_then(|()| feed.finish());
    // As the follower's step: whatever arrived whole is made durable, however the feed ended.
    let flushed = into.flush();
    drop(into);
    let r = r.and(flushed);
    r.map(|()| twin.txlog().head_seq())
        .map_err(|e| format!("{e:#}"))
}

fn piece(seq: u64, meta: &[u8], blob_len: u64, blob: &[u8], last: bool) -> gunnar_proto::LogChunk {
    gunnar_proto::LogChunk {
        seq,
        kind: gunnar_core::txlog::Kind::Pack as u32,
        meta: meta.to_vec(),
        blob_len,
        blob: blob.to_vec(),
        last,
        restore_head: 0,
        restore_bytes: 0,
        log_id: String::new(),
        offset: None,
        entry_crc: None,
        snapshot_item: None,
        snapshot_id: String::new(),
    }
}

/// **The wire's protocol, held by the twin, and every refusal leaves nothing.** A short entry
/// (`last` before its declared length), an overlong one (more than declared), a stream that ends
/// inside an entry, a stream that moves to the next `seq` with one unfinished, and a gap: each is
/// refused by name, and the twin's log and its disk are exactly what they were. A well-formed
/// stream of an entry in three pieces and one with no blob is taken.
#[test]
fn the_twin_refuses_a_short_overlong_cut_or_jumbled_entry_and_keeps_nothing_of_it() {
    let td = tempfile::tempdir().unwrap();
    let twin = open_twin(td.path());
    let m = Meta::Pack {
        repo: REPO.into(),
        hash: "sha1".into(),
    }
    .to_bytes();
    let empty_disk = txlog_bytes_on_disk(&twin);
    let refused: [(&str, Vec<gunnar_proto::LogChunk>, &str); 5] = [
        (
            "short",
            vec![
                piece(1, &m, 10, b"abcd", false),
                piece(1, &[], 10, b"ef", true),
            ],
            "6 of 10 blob bytes arrived",
        ),
        (
            "overlong",
            vec![
                piece(1, &m, 4, b"abc", false),
                piece(1, &[], 4, b"de", true),
            ],
            "more than the 4 blob bytes",
        ),
        (
            "cut",
            vec![piece(1, &m, 10, b"abcd", false)],
            "the stream ended inside seq 1",
        ),
        (
            "jumbled",
            vec![
                piece(1, &m, 10, b"abcd", false),
                piece(2, &m, 1, b"x", true),
            ],
            "moved to seq 2 with seq 1 unfinished",
        ),
        (
            "gap",
            vec![piece(2, &m, 1, b"x", true)],
            "sent seq 2 where 1 was due",
        ),
    ];
    for (what, chunks, says) in refused {
        let err = feed_twin(&twin, &chunks).expect_err(what);
        assert!(err.contains(says), "{what}: {err}");
        assert_eq!(twin.txlog().head_seq(), 0, "{what}: nothing in the log");
        assert_eq!(
            txlog_bytes_on_disk(&twin),
            empty_disk,
            "{what}: nothing on the disk"
        );
    }
    let good = [
        piece(1, &m, 6, b"ab", false),
        piece(1, &[], 6, b"cd", false),
        piece(1, &[], 6, b"ef", true),
        piece(2, &m, 0, b"", true),
    ];
    assert_eq!(feed_twin(&twin, &good), Ok(2));
    let e = twin.txlog().read(1).unwrap();
    assert_eq!(
        (e.blob.as_slice(), e.meta.as_slice()),
        (&b"abcdef"[..], m.as_slice())
    );
    assert!(twin.txlog().read(2).unwrap().blob.is_empty());
    row(
        "the twin refuses a short, overlong, cut or jumbled entry and keeps nothing of it",
        true,
        "5 refusals, 2 entries taken",
    );
}

/// Stinky row 20 (2026-10-10): `Gather` grew each blob by whatever arrived —
/// `reserve_for` bounds only the first allocation — so a peer that declared a
/// small entry and kept sending, or sent one huge range, held all of it in
/// memory. Bounded per entry, in all, and by the length the entry declared.
#[test]
fn a_gather_holds_no_more_than_its_bounds_or_an_entry_declared() {
    let mut g = Gather::bounded(10, 25);
    g.begin(1, 1, b"", 8).unwrap();
    assert!(
        g.piece(&[7; 9]).is_err(),
        "more than the 8 bytes the entry declared"
    );

    let mut g = Gather::bounded(10, 25);
    assert!(
        g.begin(1, 1, b"", 11).is_err(),
        "an entry past the per-entry bound is refused at its start"
    );

    let mut g = Gather::bounded(10, 25);
    for seq in 1..=2 {
        g.begin(seq, 1, b"", 10).unwrap();
        g.piece(&[1; 10]).unwrap();
        g.end().unwrap();
    }
    assert!(
        g.begin(3, 1, b"", 10).is_err(),
        "a third entry of 10 is past the 25 in all"
    );
    assert_eq!(g.out.len(), 2);

    let mut g = Gather::default();
    assert!(
        g.begin(1, 1, b"", GATHER_MAX_ENTRY + 1).is_err(),
        "the default is bounded too"
    );
}

/// **`stream_range` is `fetch_range`, streamed** — the same entries under the same bounds (bytes
/// and count, never fewer than one) and the same refusals, the pieces adding up to the blobs.
#[test]
fn stream_range_answers_what_fetch_range_answers_in_pieces() {
    let pd = tempfile::tempdir().unwrap();
    let primary = bare_primary(pd.path(), &[10, (5 << 20) + 3, 0, 300, 2 << 20]);
    for (from, to, max) in [
        (1, 0, 0),
        (1, 0, 1),
        (2, 4, 1 << 20),
        (3, 5, 64 << 20),
        (5, 5, 1),
        (2, 0, 6 << 20),
    ] {
        let whole = fetch_range(&primary, from, to, max).unwrap();
        let mut got = Gather::default();
        let n = stream_range(&primary, from, to, max, &mut got).unwrap();
        assert_eq!(n as usize, got.out.len());
        assert_eq!(got.out, whole, "{from}..={to} max {max}");
    }
    for (from, to) in [(6, 0), (9, 9)] {
        assert!(
            matches!(
                check_range(&primary, from, to),
                Err(txlog::Error::BeyondHead { .. })
            ),
            "{from}"
        );
        assert!(stream_range(&primary, from, to, 0, &mut Gather::default()).is_err());
    }
    // Retention past the head: the one segment is closed and dropped, the numbers kept.
    primary.retain_from(6).unwrap();
    assert_eq!(primary.first_seq(), 6);
    assert!(matches!(
        check_range(&primary, 1, 0),
        Err(txlog::Error::NotRetained { seq: 1, first: 6 })
    ));
    assert!(matches!(
        check_range(&primary, 5, 5),
        Err(txlog::Error::NotRetained { .. })
    ));
}

/// **The twin says its memory once, and again only when its peak climbs a step** — bounded
/// memory is one line, a climbing one is a line per [`SAY_MEMORY_STEP`], never a line per poll.
#[test]
fn a_twin_says_its_memory_once_and_again_only_when_the_peak_climbs_a_step() {
    assert!(memory_is_news(0, 1));
    assert!(!memory_is_news(100 << 20, 100 << 20));
    assert!(!memory_is_news(
        100 << 20,
        (100 << 20) + SAY_MEMORY_STEP - 1
    ));
    assert!(memory_is_news(100 << 20, (100 << 20) + SAY_MEMORY_STEP));
    let (peak, now) = own_memory().expect("this test runs on Linux");
    assert!(peak >= now && now > 0, "{peak} {now}");
}

// ── cov-13 (2026-10-06): the log is bounded in BYTES, and a silent twin is resynced ──

/// ★ **cov-13: a SILENT twin is not waited for once the log is over its byte budget — the
/// log is trimmed to the budget, the twin is owed a full resync, and when it answers it
/// catches up through a snapshot with NOTHING lost.**
///
/// MEASURED on the UpCloud mockup: with 512 MiB pushes the log held 4.3 GB on /data (the
/// window is 64 ENTRIES, and nothing at all was trimmed while the twin was silent), so every
/// pushed byte cost the set twice. Here: a twin follows, goes silent (its VM died — heard an
/// hour ago), main takes four ~300 KiB pushes; the retention pass under a 256 KiB budget
/// trims to it and records the debt. The twin comes back, finds the primary retaining from
/// past its own head, installs a snapshot of main's stores — and a main recharged from it IS
/// the primary (nothing lost on either box); a pass after it polls again settles the debt.
#[test]
fn cov13_a_silent_twin_past_the_byte_budget_is_owed_a_resync_and_catches_up_through_a_snapshot_with_nothing_lost(
) {
    use gunnar_grpc::heartbeat::{Bounds, Registry};
    use gunnar_grpc::retention::{pass_with, Policy};
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
    let followed = twin.txlog().head_seq();
    assert_eq!(followed, primary.txlog().head_seq());

    // The primary's view of it: heard ONCE, an hour ago — silent.
    let reg = Arc::new(Registry::new(Bounds {
        silent_after_ms: 60_000,
        behind_polls: 5,
    }));
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(crate::CoreControlPlane::new(Arc::clone(&primary)));
    let ctx = gunnar_grpc::ServiceContext::new(
        plane,
        Arc::new(gunnar_grpc::EventBus::new(16)),
        gunnar_grpc::ServerConfig::default(),
    )
    .with_heartbeats(Arc::clone(&reg));
    let an_hour_ago = estate_clock::now_unix_ms() - 3_600_000;
    reg.record(
        "gunnar-twin",
        "10.13.8.3:40000",
        &gunnar_proto::TwinReport {
            applied_seq: followed,
            disk_total_bytes: 64 << 30,
            ..Default::default()
        },
        followed,
        an_hour_ago,
    );

    // Main keeps taking pushes while the twin is gone.
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&app).unwrap();
    for i in 0..4u64 {
        let (p, o) = one_blob_pack(store.hash_kind(), &noise(0x9E37_79B9 + i, 300 << 10));
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref(&format!("refs/heads/while-silent-{i}"), None, Some(&o))
            .unwrap();
    }
    let budget = 256u64 << 10;
    let before = primary.txlog().bytes();
    assert!(
        before > 2 * budget,
        "the log is well over its budget: {before}"
    );

    let policy = Policy {
        window: 64,
        budget_percent: 0,
        budget_max_bytes: budget,
        silent_grace_ms: 600_000,
    };
    let line = pass_with(&ctx, &policy, false)
        .expect("a silent twin past its grace, the log over its budget: trimmed");
    let first = primary.txlog().first_seq();
    assert!(
        primary.txlog().bytes() <= budget,
        "the log is within its budget after the pass: {} > {budget}: {line}",
        primary.txlog().bytes()
    );
    assert!(first > followed + 1, "the log no longer holds what the twin needs next (first {first}, the twin at {followed}): {line}");
    let owed = reg.owed();
    assert_eq!(
        owed.twins.get("gunnar-twin").map(|o| o.resync_from),
        Some(first),
        "the twin is owed a full resync: {owed:?}"
    );
    assert!(
        std::fs::read(pd.path().join(TWIN_OWED_FILE)).is_ok_and(|b| !b.is_empty()),
        "the debt is on the primary's disk"
    );

    // The twin's VM is back: it finds the hole and takes a snapshot of main's stores.
    wait_until_indexed(
        &primary,
        "before the twin fetches a snapshot from the primary",
    );
    let head = primary.txlog().head_seq();
    assert_eq!(
        f.step().unwrap(),
        Progress::SnapshotInstalled { at: head },
        "the twin behind the trimmed log took a full resync"
    );
    assert_eq!(twin.txlog().head_seq(), head);
    // Nothing lost: a main recharged from the resynced twin is the primary, refs and all.
    {
        let (_d, main) = recharged(&twin);
        same_state(
            &primary,
            &main,
            "main recharged from the twin after its resync",
        );
        for i in 0..4 {
            assert!(
                refs_of(&main)
                    .iter()
                    .any(|(n, _, _)| n == &format!("refs/heads/while-silent-{i}")),
                "the push made while the twin was silent is on it: {i}"
            );
        }
    }
    // Heard again at the primary's head: the next pass settles the debt.
    reg.record(
        "gunnar-twin",
        "10.13.8.3:40000",
        &gunnar_proto::TwinReport {
            applied_seq: head,
            disk_total_bytes: 64 << 30,
            ..Default::default()
        },
        head,
        estate_clock::now_unix_ms(),
    );
    let _ = pass_with(&ctx, &policy, false);
    assert!(
        reg.owed().twins.is_empty(),
        "the resync is paid: {:?}",
        reg.owed()
    );
    row("a silent twin past the byte budget is trimmed past, owed a resync, and catches up through a snapshot with nothing lost", true, &format!("log {before} -> {} bytes (budget {budget}), snapshot at {head}", primary.txlog().bytes()));
}

/// ★ **cov-13: a twin the primary refuses as too small for main's grown set waits in
/// `recovering` — appending nothing, polling on — and follows once it may.**
#[test]
fn cov13_a_twin_refused_as_too_small_waits_in_recovering_and_appends_nothing() {
    struct Undersized {
        inner: LocalSource,
        refuse: std::sync::atomic::AtomicBool,
    }
    impl LogSource for Undersized {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.inner.head()
        }
        fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
            if self.refuse.load(Ordering::SeqCst) {
                anyhow::bail!("status: FailedPrecondition, message: \"{} this twin's data volume holds 20.0 GiB and main's data set grew to 30.0 GiB while it was silent\"", gunnar_grpc::TWIN_TOO_SMALL);
            }
            self.inner.fetch(from, to, max_bytes)
        }
        fn fetch_into(
            &self,
            from: u64,
            to: u64,
            max_bytes: u64,
            into: &mut dyn FrameSink,
        ) -> anyhow::Result<()> {
            if self.refuse.load(Ordering::SeqCst) {
                anyhow::bail!("status: FailedPrecondition, message: \"{} this twin's data volume holds 20.0 GiB and main's data set grew to 30.0 GiB while it was silent\"", gunnar_grpc::TWIN_TOO_SMALL);
            }
            self.inner.fetch_into(from, to, max_bytes, into)
        }
    }
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(Undersized {
            inner: LocalSource {
                stores: Arc::clone(&primary),
                retained_from: 0,
            },
            refuse: std::sync::atomic::AtomicBool::new(true),
        }),
    );
    assert!(f.step().is_err(), "the primary refused the fetch");
    let st = status(twin.root()).expect("a status");
    assert_eq!(st.state.to_string(), "recovering", "{st:?}");
    assert!(
        st.why.starts_with(gunnar_grpc::TWIN_TOO_SMALL) && st.why.contains("30.0 GiB"),
        "{}",
        st.why
    );
    assert_eq!(
        twin.txlog().head_seq(),
        0,
        "nothing was appended into the too-small volume"
    );
}

// ── restore-large-entry (2026-10-07): an entry larger than one call, staged ─────────────────

/// Every `LogChunk` the twin's `Mirror.LogFetch` sends for `from..=to`, its first entry from byte
/// `offset` — through the real control plane, so the chunks are the ones the wire carries.
fn fetched(
    source: &Arc<ServerStores>,
    from: u64,
    offset: u64,
    to: u64,
) -> Vec<gunnar_proto::LogChunk> {
    use gunnar_grpc::ControlPlane as _;
    let plane = crate::CoreControlPlane::new(Arc::clone(source));
    let produce = plane
        .log_fetch_stream_at(from, offset, to, 0)
        .expect("the fetch is served");
    let mut out = Vec::new();
    produce(&mut |c| {
        out.push(c);
        Ok(())
    })
    .expect("the range streams");
    out
}

/// **One call of the restore lane, straight into the box's intake**: from where the box is
/// (its head + 1, at the byte of the entry it holds part of), the twin's log, until `budget`
/// blob bytes went in — then the stream ends. `Ok(head)` when the restore finished.
fn intake_call(
    main: &Arc<ServerStores>,
    twin: &Arc<ServerStores>,
    budget: u64,
) -> Result<u64, String> {
    let head = twin.txlog().head_seq();
    let mut intake = intake::Intake::open(main, "the-lane")
        .map_err(|e| format!("{e:#}"))?
        .expect("no other stream holds the box");
    let next = intake.next();
    let offset = match intake::partial(main.root()) {
        Some((seq, bytes)) if seq == next => bytes,
        _ => 0,
    };
    let (mut cursor, mut sent) = (next, 0u64);
    while cursor <= head {
        let chunks = fetched(twin, cursor, if cursor == next { offset } else { 0 }, head);
        let mut highest = cursor - 1;
        for mut c in chunks {
            c.restore_head = head;
            let (seq, last) = (c.seq, c.last);
            sent += c.blob.len() as u64;
            intake.take(&c).map_err(|e| e.to_string())?;
            highest = highest.max(seq);
            if sent >= budget && !(last && seq == head) {
                return Err(intake
                    .end_of_stream()
                    .expect_err("the budget ended the stream inside the log")
                    .to_string());
            }
        }
        cursor = highest + 1;
    }
    intake.end_of_stream().map_err(|e| e.to_string())?;
    drop(intake);
    restore_finish(main, head, "").map_err(|e| format!("{e:#}"))
}

/// ★ **The recharge-equality test, through an intake that cuts a large entry three times.** The
/// fixture primary — every class [`assert_same`] checks — plus one 12 MiB pack of noise; a
/// store-only twin follows it; the new main is recharged from the twin by a lane that carries
/// 3 MiB per call. The big entry is cut inside at least three times, each call resumes it at the
/// byte the box holds, its staging is gone the moment it is applied, and the recharged main is
/// the primary in every class — the same check the in-process recharge passes.
#[test]
fn a_recharge_whose_lane_cuts_a_large_entry_three_times_matches_the_primary() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, mut facts) = primary(pd.path());
    {
        let repo = primary
            .catalog()
            .get(&RepoRef::new("ns", "app").unwrap())
            .unwrap()
            .unwrap();
        let store = primary.repo_store(&repo).unwrap();
        let (pack, oid) = one_blob_pack(store.hash_kind(), &noise(5, 12 << 20));
        store.serve().put_pack(&pack).expect("the big pack");
        store
            .serve()
            .update_ref("refs/heads/big", None, Some(&oid))
            .expect("big");
        facts.oids.push(oid);
    }
    let twin = open_twin(td.path());
    let f = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
    while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
    let big = (1..=twin.txlog().head_seq())
        .find(|s| twin.txlog().read(*s).unwrap().blob.len() >= 12 << 20)
        .expect("the big frame");

    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    let mut inside = Vec::new();
    let mut finished = None;
    for _ in 0..16 {
        match intake_call(&main, &twin, 3 << 20) {
            Ok(head) => {
                finished = Some(head);
                break;
            }
            Err(said) => {
                if main.txlog().head_seq() + 1 == big {
                    assert!(
                        said.contains(&format!("resumes seq {big} at byte")),
                        "{said}"
                    );
                    inside.push(
                        intake::partial(fd.path())
                            .expect("the box holds part of the big entry")
                            .1,
                    );
                } else if main.txlog().head_seq() >= big {
                    assert!(
                        !fd.path().join(intake::PARTIAL_DIR).exists(),
                        "the big entry is applied and its staging is still on the volume"
                    );
                }
            }
        }
    }
    assert_eq!(
        finished,
        Some(twin.txlog().head_seq()),
        "the recharge never finished: the box held {inside:?} of seq {big}"
    );
    assert!(
        inside.len() >= 3 && inside.windows(2).all(|w| w[1] > w[0]),
        "cut inside three times or more, each call further in: {inside:?}"
    );
    assert!(
        !fd.path().join(intake::PARTIAL_DIR).exists(),
        "nothing staged is left on the volume"
    );
    assert_same(
        &primary,
        &main,
        &facts,
        "main recharged through a lane that cut a large entry",
    );
    row("a recharge whose lane cuts a large entry three times resumes inside it and matches the primary", true, &format!("seq {big} held {inside:?} across calls"));
}

/// **What a crash leaves in the staging area is never taken on trust.** A staging area for an
/// entry the box already applied (a crash between the apply and the removal) goes when the next
/// stream opens; one whose file is SHORTER than its record says was made durable (a filesystem
/// that lied) goes too — the entry starts again at byte 0; one with an uncounted tail is cut back
/// to its count.
#[test]
fn a_staging_area_a_crash_left_is_cut_back_or_dropped_when_the_next_stream_opens() {
    let pd = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let primary = Arc::clone(primary.txlog());
    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    let stage = |seq: u64, blob_len: u64, durable: u64, on_disk: usize| {
        let dir = fd.path().join(intake::PARTIAL_DIR);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("entry.rec"), format!("gunnar-restore-partial v1\nseq {seq}\nkind 1\nblob_len {blob_len}\ndurable {durable}\nmeta 7b7d\n")).unwrap();
        std::fs::write(dir.join("entry.blob"), vec![9u8; on_disk]).unwrap();
    };
    {
        let mut i = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
        for e in primary.range(1, 2).unwrap() {
            restore_apply(&main, &e).unwrap();
        }
        drop(i.end_of_stream());
    }
    // Applied already: seq 2 is the head.
    stage(2, 200, 50, 50);
    let i = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
    assert_eq!(i.next(), 3);
    assert!(
        !fd.path().join(intake::PARTIAL_DIR).exists(),
        "a staging area for an applied entry goes"
    );
    drop(i);
    // Shorter than its record: not trusted.
    stage(3, 300, 120, 80);
    drop(intake::Intake::open(&main, "the-lane").unwrap().unwrap());
    assert_eq!(
        intake::partial(fd.path()),
        None,
        "a file shorter than its durable count is dropped"
    );
    // An uncounted tail: cut back to the count.
    stage(3, 300, 120, 170);
    drop(intake::Intake::open(&main, "the-lane").unwrap().unwrap());
    assert_eq!(intake::partial(fd.path()), Some((3, 120)));
    assert_eq!(
        std::fs::metadata(fd.path().join(intake::PARTIAL_DIR).join("entry.blob"))
            .unwrap()
            .len(),
        120,
        "the tail nobody counted is cut"
    );
    // Only one stream at a time holds the box.
    let held = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
    assert!(
        intake::Intake::open(&main, "the-lane").unwrap().is_none(),
        "a second stream waits for the first"
    );
    drop(held);
    assert!(intake::Intake::open(&main, "the-lane").unwrap().is_some());
}

/// **`Mirror.LogFetch` from an offset** (`from_offset`, restore-large-entry): the first entry's
/// pieces start at the asked byte and every piece names its offset, contiguous to the blob's
/// end; the last carries the entry's payload crc — the one its frame holds; an offset AT the end
/// is one empty last piece (with the crc); an offset PAST it sends the entry whole from 0.
#[test]
fn log_fetch_from_an_offset_starts_the_first_entry_there_and_vouches_with_its_crc() {
    let pd = tempfile::tempdir().unwrap();
    let primary = Arc::new(ServerStores::open(pd.path()).unwrap());
    let blob = noise(3, (5 << 20) + 17);
    let s1 = primary
        .txlog()
        .append(
            Kind::Pack,
            &Meta::Pack {
                repo: REPO.into(),
                hash: "sha1".into(),
            },
            &blob,
        )
        .unwrap();
    let s2 = primary
        .txlog()
        .append(
            Kind::Pack,
            &Meta::Pack {
                repo: REPO.into(),
                hash: "sha1".into(),
            },
            b"tail",
        )
        .unwrap();
    let e = primary.txlog().read(s1).unwrap();
    let mut crc = gunnar_core::txlog::Crc32c::new();
    crc.update(&e.meta);
    crc.update(&e.blob);
    let crc = crc.finish();
    let first = |chunks: &[gunnar_proto::LogChunk]| {
        chunks
            .iter()
            .filter(|c| c.seq == s1)
            .cloned()
            .collect::<Vec<_>>()
    };

    let from = 3 * (1 << 20) + 5;
    let c = first(&fetched(&primary, s1, from, s2));
    assert_eq!(
        c[0].offset,
        Some(from),
        "the first piece starts at the asked byte"
    );
    let mut at = from;
    for p in &c {
        assert_eq!(p.offset, Some(at), "contiguous");
        at += p.blob.len() as u64;
    }
    assert_eq!(at, blob.len() as u64);
    assert_eq!(
        c.iter().flat_map(|p| p.blob.clone()).collect::<Vec<_>>(),
        blob[from as usize..].to_vec(),
        "the bytes from the offset, and only those"
    );
    assert!(
        c.last().unwrap().last && c.last().unwrap().entry_crc == Some(crc),
        "the last piece vouches with the frame's crc"
    );
    assert!(c[..c.len() - 1]
        .iter()
        .all(|p| p.entry_crc.is_none() && !p.last));
    let whole = fetched(&primary, s1, from, s2);
    assert_eq!(
        whole.iter().find(|p| p.seq == s2).unwrap().offset,
        Some(0),
        "the next entry starts whole"
    );
    assert!(whole
        .iter()
        .rfind(|p| p.seq == s2)
        .unwrap()
        .entry_crc
        .is_some());

    let at_end = first(&fetched(&primary, s1, blob.len() as u64, s1));
    assert_eq!(at_end.len(), 1);
    assert!(
        at_end[0].blob.is_empty()
            && at_end[0].last
            && at_end[0].offset == Some(blob.len() as u64)
            && at_end[0].entry_crc == Some(crc),
        "{:?}",
        at_end[0].offset
    );

    let past = first(&fetched(&primary, s1, blob.len() as u64 + 1, s1));
    assert_eq!(
        past[0].offset,
        Some(0),
        "an offset past the blob is not this entry as the asker knew it: whole from 0"
    );
    assert_eq!(past.iter().map(|p| p.blob.len()).sum::<usize>(), blob.len());
}

// ── restore-after-snapshot (2026-10-07): a twin's base, resumed across calls ─────────────

/// Every `LogChunk` the twin's `Mirror.SnapshotFetch` sends from item `from_item`, byte
/// `from_offset` on, for base `expect_id` — through the real control plane. A base that changed
/// is asked for again from its start (as the restore lane does).
fn snapshot_fetched(
    source: &Arc<ServerStores>,
    from_item: u64,
    from_offset: u64,
    expect_id: &str,
) -> Vec<gunnar_proto::LogChunk> {
    use gunnar_grpc::ControlPlane as _;
    let plane = crate::CoreControlPlane::new(Arc::clone(source));
    let mut out = Vec::new();
    match plane.snapshot_stream_from(from_item, from_offset, expect_id, &mut |c| {
        out.push(c);
        Ok(())
    }) {
        Ok(_) => out,
        Err(err) if err.to_string().contains(snapshot::SNAPSHOT_CHANGED) => {
            snapshot_fetched(source, 0, 0, "")
        }
        Err(err) => panic!("SnapshotFetch: {err}"),
    }
}

/// **One call of the restore lane after a snapshot, straight into the box's intake**: the
/// twin's base from where the box holds it, then the log from the twin's `first_seq`, until
/// `budget` blob bytes went in. `Ok(head)` when the restore finished.
fn snapshot_call(
    main: &Arc<ServerStores>,
    twin: &Arc<ServerStores>,
    budget: u64,
) -> Result<u64, String> {
    let head = twin.txlog().head_seq();
    let first = twin.txlog().first_seq();
    let mut intake = intake::Intake::open(main, "the-lane")
        .map_err(|e| format!("{e:#}"))?
        .expect("no other stream holds the box");
    let mut sent = 0u64;
    if main.txlog().head_seq() + 1 < first {
        let held = snapshot::held_snapshot(main.root());
        let (item, offset, id) =
            held.map(|h| (h.items, h.bytes, h.id))
                .unwrap_or((0, 0, String::new()));
        for mut c in snapshot_fetched(twin, item, offset, &id) {
            c.restore_head = head;
            let terminator = c.kind == Kind::SnapshotAt as u32 && c.last;
            sent += c.blob.len() as u64;
            intake.take(&c).map_err(|e| e.to_string())?;
            if sent >= budget && !terminator {
                return Err(intake
                    .end_of_stream()
                    .expect_err("the budget ended the stream inside the base")
                    .to_string());
            }
        }
    }
    let next = intake.next();
    let offset = match intake::partial(main.root()) {
        Some((seq, bytes)) if seq == next => bytes,
        _ => 0,
    };
    let mut cursor = next;
    while cursor <= head {
        let chunks = fetched(twin, cursor, if cursor == next { offset } else { 0 }, head);
        let mut highest = cursor - 1;
        for mut c in chunks {
            c.restore_head = head;
            let (seq, last) = (c.seq, c.last);
            sent += c.blob.len() as u64;
            intake.take(&c).map_err(|e| e.to_string())?;
            highest = highest.max(seq);
            if sent >= budget && !(last && seq == head) {
                return Err(intake
                    .end_of_stream()
                    .expect_err("the budget ended the stream inside the log")
                    .to_string());
            }
        }
        cursor = highest + 1;
    }
    intake.end_of_stream().map_err(|e| e.to_string())?;
    drop(intake);
    restore_finish(main, head, "").map_err(|e| format!("{e:#}"))
}

/// The fixture primary with a 12 MiB pack of noise, and a twin that took a SNAPSHOT of it (its
/// retention past the twin) and then followed the primary past the base.
fn snapshotted(
    pd: &std::path::Path,
    td: &std::path::Path,
) -> (Arc<ServerStores>, Facts, Arc<ServerStores>, u64) {
    let (primary, mut facts) = primary(pd);
    {
        let repo = primary
            .catalog()
            .get(&RepoRef::new("ns", "app").unwrap())
            .unwrap()
            .unwrap();
        let store = primary.repo_store(&repo).unwrap();
        let (pack, oid) = one_blob_pack(store.hash_kind(), &noise(5, 12 << 20));
        store.serve().put_pack(&pack).expect("the big pack");
        store
            .serve()
            .update_ref("refs/heads/big", None, Some(&oid))
            .expect("big");
        facts.oids.push(oid);
    }
    wait_until_indexed(&primary, "before the twin takes a snapshot");
    let twin = open_twin(td);
    let at = primary.txlog().head_seq();
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: at,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
    primary
        .auth()
        .add_principal(Principal::new("carol", "Carol", PrincipalKind::User).unwrap())
        .unwrap();
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
    assert_eq!(
        (twin.txlog().first_seq(), twin.txlog().head_seq()),
        (at + 1, primary.txlog().head_seq())
    );
    (primary, facts, twin, at)
}

/// ★ **The recharge-equality test, from a twin whose log starts after its snapshot**, through an
/// intake that cuts the base three times or more. The fixture primary — every class
/// [`assert_same`] checks — plus a 12 MiB pack; the twin takes a snapshot of it and follows past
/// it; the new main is filled by a lane that carries 3 MiB per call: the base resumed at the
/// item and byte the box holds, installed at its terminator, then the log. The main is the
/// primary in every class, and nothing staged is left on its volume.
#[test]
fn a_recharge_from_a_snapshotted_twin_resumes_its_base_across_calls_and_matches_the_primary() {
    let (pd, td, fd) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let (primary, facts, twin, at) = snapshotted(pd.path(), td.path());
    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    let mut held = Vec::new();
    let mut finished = None;
    for _ in 0..24 {
        match snapshot_call(&main, &twin, 3 << 20) {
            Ok(head) => {
                finished = Some(head);
                break;
            }
            Err(said) => {
                if let Some(h) = snapshot::held_snapshot(fd.path()) {
                    assert!(
                        said.contains("inside the twin's snapshot")
                            && said.contains("resumes there"),
                        "{said}"
                    );
                    held.push((h.items, h.bytes));
                }
            }
        }
    }
    assert_eq!(
        finished,
        Some(twin.txlog().head_seq()),
        "the recharge never finished: the box held {held:?} of the base"
    );
    assert!(
        held.len() >= 3 && held.windows(2).all(|w| w[1] > w[0]),
        "the base cut inside three times or more, each call further in: {held:?}"
    );
    assert!(
        !fd.path().join(snapshot::RESTORE_SNAPSHOT).exists(),
        "nothing staged is left on the volume"
    );
    assert_eq!(
        main.txlog().first_seq(),
        at + 1,
        "the main's log starts after the base, as the twin's does"
    );
    assert_same(
        &primary,
        &main,
        &facts,
        "main recharged from a snapshotted twin through a lane that cut its base",
    );
    assert!(
        main.auth().principal("carol").unwrap().is_some(),
        "a principal from the log after the base is there"
    );
    row(
        "a recharge from a snapshotted twin resumes its base across calls and matches the primary",
        true,
        &format!("base at {at} held {held:?} across calls"),
    );
}

/// **Which packs a cut carries beside the reachable export**: per repository, the `Pack` frames
/// after its last `Refs` frame at or below the cut — a push caught between its two frames; a pack
/// whose refs are already framed is reachable and is not carried twice; nothing after the cut.
#[test]
fn a_cut_carries_the_packs_of_pushes_caught_between_their_frames() {
    use gunnar_core::txlog::Kind;
    let pack = |r: &str| Meta::Pack {
        repo: r.into(),
        hash: "sha256".into(),
    };
    let refs = |r: &str| Meta::Refs {
        repo: r.into(),
        atomic: false,
        edits: vec![RefEdit {
            name: "refs/heads/main".into(),
            old: None,
            new: Some("ab".repeat(32)),
            peeled: None,
            symref: None,
        }],
    };
    let (p, f) = (Kind::Pack as u16, Kind::Refs as u16);
    let frames = vec![
        (10, p, pack("ns/a")),
        (11, f, refs("ns/a")),
        (12, p, pack("ns/a")),
        (13, p, pack("ns/b")),
        (14, f, refs("ns/b")),
        (15, p, pack("ns/b")),
        (16, p, pack("ns/c")),
        (17, f, refs("ns/c")),
    ];
    let open = snapshot::pending_packs(&frames, 16);
    assert_eq!(
        open.get("ns/a"),
        Some(&vec![12]),
        "a's second push: its pack in, its refs not yet"
    );
    assert_eq!(open.get("ns/b"), Some(&vec![15]));
    assert_eq!(
        open.get("ns/c"),
        Some(&vec![16]),
        "c's refs are AFTER the cut (17)"
    );
    assert!(
        !snapshot::pending_packs(&frames, 17).contains_key("ns/c"),
        "a cut after c's refs carries nothing of c's"
    );
}

/// **Two pushes into ONE repository straddling the cut** (R6 residual, 5.3.3): push A frames its
/// pack, push B frames its pack, A frames its refs — and the cut falls there, before B's refs. A
/// `Refs` frame does not say which pack it closes, so the first refs must not close the repository's
/// every open pack: every pack since the repository last had no push open is carried (a carried
/// pack whose objects are reachable already costs bytes, never correctness). A refs frame that
/// only deletes closes no push.
#[test]
fn two_pushes_into_one_repo_straddling_the_cut_both_carry_their_packs() {
    use gunnar_core::txlog::Kind;
    let pack = |r: &str| Meta::Pack {
        repo: r.into(),
        hash: "sha256".into(),
    };
    let edit = |new: Option<String>| RefEdit {
        name: "refs/heads/x".into(),
        old: None,
        new,
        peeled: None,
        symref: None,
    };
    let refs = |r: &str, new: Option<String>| Meta::Refs {
        repo: r.into(),
        atomic: false,
        edits: vec![edit(new)],
    };
    let (p, f) = (Kind::Pack as u16, Kind::Refs as u16);
    let oid = Some("cd".repeat(32));
    // A's pack, B's pack, A's refs | cut | B's refs.
    let frames = vec![
        (20, p, pack("ns/a")),
        (21, p, pack("ns/a")),
        (22, f, refs("ns/a", oid.clone())),
        (23, f, refs("ns/a", oid.clone())),
    ];
    assert_eq!(
        snapshot::pending_packs(&frames, 22).get("ns/a"),
        Some(&vec![20, 21]),
        "B's pack (21) must be carried; A's may be"
    );
    assert!(
        !snapshot::pending_packs(&frames, 23).contains_key("ns/a"),
        "after both refs nothing is open"
    );
    // A refs frame that only deletes closes no push: the pack framed before it stays carried.
    let frames = vec![(30, p, pack("ns/d")), (31, f, refs("ns/d", None))];
    assert_eq!(
        snapshot::pending_packs(&frames, 31).get("ns/d"),
        Some(&vec![30])
    );
}

/// ★ **R6 residual, end to end**: two pushes into the SAME repository, the cut between the first
/// push's refs and the second's. The recharge from the twin must finish and serve BOTH pushed
/// objects (the first `Refs` frame closed every open pack of the repository, so the second push's
/// pack was not carried and its refs named an object the base never had).
#[test]
fn two_pushes_into_one_repo_straddling_a_snapshot_cut_still_recharge_the_main() {
    let (pd, td, fd) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let (primary, _facts) = primary(pd.path());
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&app).unwrap();
    let (pack_a, oid_a) = one_blob_pack(store.hash_kind(), b"push A, framed first\n");
    let (pack_b, oid_b) = one_blob_pack(store.hash_kind(), b"push B, framed second\n");
    store.serve().put_pack(&pack_a).expect("A's pack");
    store.serve().put_pack(&pack_b).expect("B's pack");
    store
        .serve()
        .update_ref("refs/heads/pushed-a", None, Some(&oid_a))
        .expect("A's ref");
    wait_until_indexed(&primary, "before the cut between A's refs and B's");
    let twin = open_twin(td.path());
    let at = primary.txlog().head_seq();
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: at,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
    store
        .serve()
        .update_ref("refs/heads/pushed-b", None, Some(&oid_b))
        .expect("B's ref, after the cut");
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    let mut said = Vec::new();
    let mut finished = None;
    for _ in 0..8 {
        match snapshot_call(&main, &twin, 1 << 40) {
            Ok(head) => {
                finished = Some(head);
                break;
            }
            Err(e) => said.push(e),
        }
    }
    assert_eq!(
        finished,
        Some(twin.txlog().head_seq()),
        "the recharge never finished: {said:?}"
    );
    let restored = main
        .repo_store(
            &main
                .catalog()
                .get(&RepoRef::new("ns", "app").unwrap())
                .unwrap()
                .unwrap(),
        )
        .unwrap();
    for (oid, r) in [
        (&oid_a, "refs/heads/pushed-a"),
        (&oid_b, "refs/heads/pushed-b"),
    ] {
        assert!(
            restored.serve().read(oid).unwrap().is_some(),
            "{r}'s object is on the recharged main"
        );
        assert!(
            restored.serve().refs().unwrap().iter().any(|x| x.name == r),
            "and {r} names it"
        );
    }
}

/// ★ **R6, the estate-box gate on dwarf (kit 39d46ea, gunnar 35730a8, 2026-10-09): a cut BETWEEN a
/// push's pack and its refs.** A push is two frames — the `Pack` (framed inside the ingest, under
/// its own mutation guard) and, after the policy gates, the `Refs`. The twin re-snapshotted the
/// old main at 4103 between the two: the snapshot exported each repository's objects as the
/// reachability over its ref TIPS at the cut, and the pushed objects were not reachable from any
/// tip yet. Entry 4104 — the refs — then named an object the base never had: the recharged main
/// re-applied it on every call, "refs/heads/main would point at …, which this repository does not
/// have", for twenty minutes, and never said it was restored.
///
/// The cut here is taken exactly there: the pack in, the snapshot, the ref after. The recharge
/// from the twin must finish and serve the pushed object.
#[test]
fn a_snapshot_cut_between_a_pushs_pack_and_its_refs_still_recharges_the_main() {
    let (pd, td, fd) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let (primary, _facts) = primary(pd.path());
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&app).unwrap();
    // The push's first frame: its pack, in the store, reachable from no ref yet.
    let (pack, oid) = one_blob_pack(store.hash_kind(), b"pushed just before the twin's cut\n");
    store.serve().put_pack(&pack).expect("the pushed pack");
    wait_until_indexed(
        &primary,
        "before the twin takes its snapshot between the pack and the refs",
    );
    // The twin is behind a retention hole and takes a snapshot of the primary NOW.
    let twin = open_twin(td.path());
    let at = primary.txlog().head_seq();
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: at,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
    // The push's second frame, after the cut: the ref that names the pushed object.
    store
        .serve()
        .update_ref("refs/heads/pushed", None, Some(&oid))
        .expect("the push's ref");
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: 0,
        }),
    );
    while !matches!(f.step().unwrap(), Progress::CaughtUp { .. }) {}
    assert_eq!(
        twin.txlog().head_seq(),
        primary.txlog().head_seq(),
        "the twin followed past the refs"
    );
    // main is lost; a fresh main is recharged from the twin.
    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    let mut said = Vec::new();
    let mut finished = None;
    for _ in 0..8 {
        match snapshot_call(&main, &twin, 1 << 40) {
            Ok(head) => {
                finished = Some(head);
                break;
            }
            Err(e) => said.push(e),
        }
    }
    assert_eq!(
        finished,
        Some(twin.txlog().head_seq()),
        "the recharge never finished: {said:?}"
    );
    let restored = main
        .repo_store(
            &main
                .catalog()
                .get(&RepoRef::new("ns", "app").unwrap())
                .unwrap()
                .unwrap(),
        )
        .unwrap();
    assert!(
        restored.serve().read(&oid).unwrap().is_some(),
        "the pushed object is on the recharged main"
    );
    assert!(
        restored
            .serve()
            .refs()
            .unwrap()
            .iter()
            .any(|r| r.name == "refs/heads/pushed"),
        "and the push's ref names it"
    );
}

/// **The twin's base, served from a position** (`SnapshotFetchRequest`): every piece names its
/// item, offset and base; the bytes from the asked offset only; each item's last piece the crc
/// the twin kept (crc32c over meta ‖ blob); the base's id stable across calls and refused by name
/// when the asker holds part of another one.
#[test]
fn a_twins_base_is_served_from_an_item_and_byte_with_its_crcs_and_its_id() {
    let (pd, td) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
    let (_primary, _facts, twin, _at) = snapshotted(pd.path(), td.path());
    let whole = snapshot_fetched(&twin, 0, 0, "");
    let id = whole[0].snapshot_id.clone();
    assert!(
        !id.is_empty() && whole.iter().all(|c| c.seq == 0 && c.snapshot_id == id),
        "every piece names the base"
    );
    assert_eq!(
        snapshot::base_id(twin.root()).unwrap().as_deref(),
        Some(id.as_str()),
        "and it is the base's id, stable"
    );
    // Every item put back together matches the crc on its last piece.
    let mut crc = gunnar_core::txlog::Crc32c::new();
    let mut items = 0;
    for c in &whole {
        if c.offset == Some(0) {
            crc = gunnar_core::txlog::Crc32c::new();
            crc.update(&c.meta);
        }
        crc.update(&c.blob);
        if c.last {
            assert_eq!(
                c.entry_crc,
                Some(crc.finish()),
                "item {:?}: the crc the twin kept",
                c.snapshot_item
            );
            items += 1;
        }
    }
    // The big item, from a byte inside it: only the bytes from there, contiguous, its crc at the end.
    let (big, big_len) = whole
        .iter()
        .filter(|c| c.last)
        .map(|c| (c.snapshot_item.unwrap(), c.blob_len))
        .max_by_key(|(_, l)| *l)
        .unwrap();
    let from = 5 * (1 << 20) + 7;
    let tail = snapshot_fetched(&twin, big, from, &id);
    assert_eq!(
        (tail[0].snapshot_item, tail[0].offset),
        (Some(big), Some(from)),
        "it starts where it was asked"
    );
    assert!(
        !tail[0].meta.is_empty(),
        "a resumed item carries its meta again"
    );
    let bytes: u64 = tail
        .iter()
        .filter(|c| c.snapshot_item == Some(big))
        .map(|c| c.blob.len() as u64)
        .sum();
    assert_eq!(
        bytes,
        big_len - from,
        "the bytes from the offset, and only those"
    );
    assert_eq!(
        tail.iter().filter(|c| c.last).count() as u64,
        items - big,
        "then every item after it"
    );
    // A base the asker does not hold is refused by name rather than spliced.
    use gunnar_grpc::ControlPlane as _;
    let plane = crate::CoreControlPlane::new(Arc::clone(&twin));
    let err = plane
        .snapshot_stream_from(big, from, "another-base", &mut |_| Ok(()))
        .unwrap_err();
    assert!(
        err.to_string().contains(snapshot::SNAPSHOT_CHANGED),
        "{err}"
    );
    // And LogHead says the base: what lets the restore lane tell it from lost entries.
    let head = plane.log_head().unwrap();
    assert_eq!(
        (head.snapshot_at, head.first_seq),
        (
            snapshot::base_at(twin.root()).unwrap().unwrap(),
            snapshot::base_at(twin.root()).unwrap().unwrap() + 1
        )
    );
    assert_eq!(head.snapshot_bytes, snapshot::base_bytes(twin.root()));
}

/// **What a crash leaves of a staged base is never taken on trust.** Staged part-way and
/// committed, then: an uncounted tail is cut back to the count; a file SHORTER than its count is
/// dropped (the base starts again); and an item whose bytes do not match the twin's crc is
/// refused by name — nothing installed, the staging dropped.
#[test]
fn a_staged_base_a_crash_left_is_cut_back_or_dropped_and_a_bad_item_is_refused_by_its_crc() {
    let (pd, td, fd) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let (_primary, _facts, twin, _at) = snapshotted(pd.path(), td.path());
    let main = Arc::new(ServerStores::open(fd.path()).unwrap());
    // Two calls of 3 MiB: part of the base staged, durably.
    for _ in 0..2 {
        assert!(
            snapshot_call(&main, &twin, 3 << 20).is_err(),
            "the budget ends the call inside the base"
        );
    }
    let held = snapshot::held_snapshot(fd.path()).expect("part of the base is staged");
    assert!(held.items > 0 || held.bytes > 0, "{held:?}");
    let blobs = fd.path().join(snapshot::RESTORE_SNAPSHOT).join("blobs");
    let durable = std::fs::metadata(&blobs).unwrap().len();
    // An uncounted tail (a crash between a write and its commit): cut back.
    {
        use std::io::Write as _;
        std::fs::OpenOptions::new()
            .append(true)
            .open(&blobs)
            .unwrap()
            .write_all(&[7u8; 4321])
            .unwrap();
    }
    let s = snapshot::RestoreSnapshot::resume(fd.path())
        .unwrap()
        .expect("resumed");
    assert_eq!(
        s.position(),
        (held.items, held.bytes),
        "it holds what it made durable — no more"
    );
    drop(s);
    assert_eq!(
        std::fs::metadata(&blobs).unwrap().len(),
        durable,
        "the tail nobody counted is cut"
    );
    // A file shorter than its count: not trusted.
    std::fs::OpenOptions::new()
        .write(true)
        .open(&blobs)
        .unwrap()
        .set_len(durable - 1)
        .unwrap();
    assert!(
        snapshot::RestoreSnapshot::resume(fd.path())
            .unwrap()
            .is_none(),
        "a staging shorter than its record is dropped"
    );
    assert!(snapshot::held_snapshot(fd.path()).is_none());
    // An item that does not match the twin's crc: refused by name, the staging dropped.
    let whole = snapshot_fetched(&twin, 0, 0, "");
    let mut intake = intake::Intake::open(&main, "the-lane").unwrap().unwrap();
    let mut refused = None;
    for mut c in whole {
        if c.last && c.entry_crc.is_some() {
            c.entry_crc = c.entry_crc.map(|x| x ^ 1);
        }
        if let Err(e) = intake.take(&c) {
            refused = Some(e.to_string());
            break;
        }
    }
    let refused = refused.expect("an item whose crc does not match is refused");
    assert!(
        refused.contains("takes it again from its first item"),
        "{refused}"
    );
    drop(intake);
    assert!(
        snapshot::held_snapshot(fd.path()).is_none(),
        "nothing of it is kept"
    );
    assert_eq!(main.txlog().head_seq(), 0, "and nothing was installed");
}

// ★ twin-resync-space (2026-10-06): a full resync on a twin the size of main's set.
mod resync_space;

// ── the restore marker and the stream's log id (stinky rows 4 and 7, 2026-10-10) ──

/// A fresh box holding every entry of a primary's log, restoring, with its root.
fn a_fully_streamed_restore() -> (tempfile::TempDir, tempfile::TempDir, Arc<ServerStores>, u64) {
    let pd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let log = primary.txlog();
    let (first, head) = (log.first_seq(), log.head_seq());
    let entries = log.range(first, head).unwrap();
    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    assert_eq!(restore_begin(&fresh).unwrap(), 1);
    for e in &entries {
        restore_apply(&fresh, e).unwrap();
    }
    (pd, fd, fresh, head)
}

/// ★ **A corrupt restore marker is no "nothing announced"** (stinky row 4).
/// `restore_announced` read the marker with `unwrap_or_default()` and the head
/// with `parse().ok().unwrap_or(0)`, so a marker it could not read or parse
/// meant "no stream ever announced anything", and the restore finished on
/// whatever head it had, as whatever log — a twin silently short.
#[test]
fn a_corrupt_restore_marker_refuses_to_finish_the_restore() {
    let (_pd, fd, fresh, _head) = a_fully_streamed_restore();
    std::fs::write(fd.path().join(RESTORE_ANNOUNCED), "l0g-1d 1078\n").unwrap();
    let finished = restore_finish(&fresh, 0, "");
    assert!(
        finished.is_err(),
        "a restore finished past a marker it could not parse: {finished:?}"
    );
    assert!(is_restoring(fd.path()), "the box stays restoring");
}

/// …and neither is a marker that cannot be READ.
#[test]
fn an_unreadable_restore_marker_refuses_to_finish_the_restore() {
    let (_pd, fd, fresh, _head) = a_fully_streamed_restore();
    std::fs::create_dir(fd.path().join(RESTORE_ANNOUNCED)).unwrap();
    let finished = restore_finish(&fresh, 0, "");
    assert!(
        finished.is_err(),
        "a restore finished past a marker it could not read: {finished:?}"
    );
    assert!(is_restoring(fd.path()), "the box stays restoring");
}

/// ★ **A restore over the network names its log** (stinky row 7). The lineage
/// check let an empty id through on either side, so a stream from another
/// history was taken whenever the sequence numbers happened to fit.
#[test]
fn a_network_restore_that_names_no_log_is_refused() {
    use gunnar_grpc::ControlPlane as _;
    let (_pd, fd, fresh, head) = a_fully_streamed_restore();
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
    let finished = plane.restore_finish(head, "");
    assert!(
        finished.is_err(),
        "a streamed restore that named no log finished: {finished:?}"
    );
    assert!(is_restoring(fd.path()), "the box stays restoring");
    assert_eq!(plane.restore_finish(head, "the-twins-log").unwrap(), head);
}

/// …and a restore whose earlier stream left no log id behind is refused too.
#[test]
fn a_network_restore_over_a_marker_with_no_log_id_is_refused() {
    use gunnar_grpc::ControlPlane as _;
    let (_pd, fd, fresh, head) = a_fully_streamed_restore();
    std::fs::write(fd.path().join(RESTORE_ANNOUNCED), format!("{head} \n")).unwrap();
    let plane = crate::control_plane::CoreControlPlane::new(Arc::clone(&fresh));
    let finished = plane.restore_finish(head, "the-twins-log");
    assert!(
        finished.is_err(),
        "a streamed restore finished over a marker that named no log: {finished:?}"
    );
    assert!(is_restoring(fd.path()), "the box stays restoring");
}

// ── a KEK or key renamed into place is durable only once its directory is (stinky row 9) ──

/// ★ **A KEK whose rename was never made durable is no KEK written.**
/// `write_kek` and `write_key_file` renamed the key into place and then
/// `let _ = d.sync_all()`: a power cut could take the name back while the
/// caller went on as if the key were there.
#[test]
fn a_kek_whose_directory_cannot_be_flushed_is_not_written() {
    let d = tempfile::tempdir().unwrap();
    crate::durable::dir_sync_seam::fail_from_now();
    assert!(
        write_kek(d.path(), &[7u8; 32]).is_err(),
        "write_kek reported a KEK whose rename was never made durable"
    );
    assert!(
        write_key_file(d.path(), "k", &[7u8; 32]).is_err(),
        "write_key_file reported a key whose rename was never made durable"
    );
}

/// …and **the re-seal marker goes only once the new KEK's rename is durable**:
/// with the directory unflushable the re-seal fails and stays owed, rather
/// than dropping `txlog-reseal` over a key a power cut can take back.
#[test]
fn a_reseal_whose_kek_rename_cannot_be_flushed_stays_owed() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let fd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    drop(primary);
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    restore(Arc::clone(twin.txlog()), Arc::clone(&fresh), "").unwrap();
    assert!(reseal_pending(fd.path()));
    drop(fresh);

    crate::durable::dir_sync_seam::fail_from_now();
    let reopened = ServerStores::open(fd.path());
    assert!(
        reopened.is_err(),
        "a re-seal whose KEK rename was never flushed reported the box open"
    );
    assert!(
        reseal_pending(fd.path()),
        "the re-seal marker went before the new KEK was durable"
    );
}

// ── last seen (owner, 2026-10-21): the stamps reach the twin ───────────────

/// ★ **A stamp on the primary is on the twin after a flush** — as ONE frame for
/// the whole flush (none when nothing moved, none per call); the twin keeps it
/// across its own restart, and a main recharged from the twin has it.
#[test]
fn a_last_seen_stamp_on_the_primary_is_on_the_twin_after_a_flush() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    let twin = open_twin(td.path());
    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    primary.activity().flush().unwrap();
    let before = primary.txlog().head_seq();

    primary.activity().signed_in("kim", 1_000_000);
    for i in 0..100 {
        primary.activity().operated("boss", 2_000_000 + i);
    }
    assert_eq!(
        primary.txlog().head_seq(),
        before,
        "a stamp frames nothing by itself"
    );
    assert!(primary.activity().flush().unwrap());
    assert_eq!(
        primary.txlog().head_seq(),
        before + 1,
        "one frame for the whole flush"
    );
    assert!(!primary.activity().flush().unwrap());
    assert_eq!(
        primary.txlog().head_seq(),
        before + 1,
        "a flush with nothing moved frames nothing"
    );

    Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
        .step()
        .unwrap();
    assert_eq!(
        twin.activity().seen("kim").signed_in,
        Some(1_000_000),
        "the twin has kim's sign-in"
    );
    assert_eq!(
        twin.activity().seen("boss").operated,
        Some(2_000_000),
        "the twin has boss's operation"
    );

    drop(twin);
    let twin = open_twin(td.path());
    assert_eq!(
        twin.activity().seen("kim").signed_in,
        Some(1_000_000),
        "the twin keeps it across a restart"
    );
    let (_fd, main) = recharged(&twin);
    assert_eq!(
        main.activity().seen("boss").operated,
        Some(2_000_000),
        "a main recharged from the twin has it"
    );
    row(
        "a last-seen stamp on the primary is on the twin after a flush, one frame per flush",
        true,
        "kim, boss",
    );
}

/// ★ **A twin that joins after the stamps has them from the snapshot.**
#[test]
fn a_twin_joined_after_last_seen_stamps_has_them_from_the_snapshot() {
    let pd = tempfile::tempdir().unwrap();
    let sd = tempfile::tempdir().unwrap();
    let (primary, _facts) = primary(pd.path());
    primary.activity().signed_in("kim", 3_000_000);
    primary.activity().operated("kim", 4_000_000);
    primary.activity().flush().unwrap();
    wait_until_indexed(&primary, "before the snapshot");
    let snap_twin = open_twin(sd.path());
    let at = primary.txlog().head_seq();
    let f = Follower::new(
        Arc::clone(&snap_twin),
        Box::new(LocalSource {
            stores: Arc::clone(&primary),
            retained_from: at,
        }),
    );
    assert_eq!(f.step().unwrap(), Progress::SnapshotInstalled { at });
    assert_eq!(
        (
            snap_twin.activity().seen("kim").signed_in,
            snap_twin.activity().seen("kim").operated
        ),
        (Some(3_000_000), Some(4_000_000)),
        "the snapshot carried the stamps"
    );
    drop(f);
    drop(snap_twin);
    let snap_twin = open_twin(sd.path());
    assert_eq!(
        snap_twin.activity().seen("kim").operated,
        Some(4_000_000),
        "and the twin keeps them"
    );
    row(
        "a twin joined after last-seen stamps has them from the snapshot",
        true,
        "kim",
    );
}