gunnar-server 3.0.0

//! **The estate's repair, on the box** (5.4 resilience, owner 2026-10-10): the box
//! repairs itself where it can, through the ONE implementation of the verbs
//! (`gunnar-repair`). Two callers live here, both handing the same [`ServerEstate`]
//! to the crate:
//!
//! * the **boot check** ([`boot`]): this box came up with its data set dead or empty
//!   while a twin exists — it says `data lost; recharge from twin available`, and it
//!   recharges by itself ONLY when the estate opted into auto-fix (`--repair-auto`,
//!   OFF by default, like `--watch-primary`);
//! * the **pull channel** ([`pull_loop`]): the box REPORTS to monetize's `Fleet`,
//!   PULLS the orders queued for it, verifies each under the monetize key and runs it
//!   through `gunnar_repair::exec::execute`; the ack rides the next report. Monetize
//!   never calls in: no order reaches a box that does not fetch it.
//!
//! **Whose key runs a verb.** Every call is signed with the box's repair key
//! (`--repair-key`, else the twin's own `twin.key`): reading the twin's log needs a
//! member of group `twin`; a fresh main's `Bootstrap.Restore` takes only the keys it
//! was fed (`--restore-pubkey`, the installer's admin / monetize / console keys). A
//! twin's `twin.key` therefore recharges a main only when the installer fed it to the
//! re-laid main as a restore key; otherwise the front's console runs the recharge.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use gunnar_core::ssh_key::PrivateKey;
use gunnar_grpc::client_signer::ControlChannel;
use gunnar_repair::exec::{self, BootReport, BootStep, Estate};
#[cfg(feature = "fleet-pull")]
use gunnar_repair::fleet;
use gunnar_repair::proto::{ProtoFence, ProtoMain, ProtoTwin};
use gunnar_repair::{Fault, Refusal, RefusalCode};

use crate::stores::ServerStores;

/// The log target of everything here.
pub const TARGET: &str = "gunnar::repair";

/// The longest pause between reports on a door the feed named live (`--fleet-every-s`'s default).
pub const FED_EVERY: Duration = Duration::from_secs(60);

/// How the box repairs itself (`--repair-*`, `--fleet*`).
#[derive(Debug, Clone, Default)]
pub struct RepairConfig {
    /// **Auto-fix**: the boot check recharges by itself. The estate's opt-in, OFF by default.
    pub auto: bool,
    /// The repair key (OpenSSH private key). `None`: this box's `twin.key`, when it has one.
    pub key: Option<PathBuf>,
    /// The twin to recharge from. `None`: the twin the installer's re-admit named
    /// (`<data>/pairing/recharge_from`), if any.
    pub twin: Option<String>,
    /// The pull channel to monetize; `None` = off.
    pub fleet: Option<FleetConfig>,
}

/// Where and as whom this box pulls.
#[derive(Debug, Clone)]
pub struct FleetConfig {
    /// monetize's `Fleet` endpoint (`https://…`; `http://` only to a loopback address).
    pub url: String,
    /// PEM of the CA monetize's certificate is signed by; `None` for `https://` = the public roots.
    pub ca: Option<PathBuf>,
    /// The CA as fed ([`crate::feed::FEED_FLEET_CA`]), PEM text; used when [`Self::ca`] is `None`.
    pub ca_pem: Option<String>,
    /// Where to connect (`host:port`, [`crate::feed::FEED_FLEET_AT`]); `None` = the URL's own
    /// host. TLS verifies the URL's host either way.
    pub at: Option<String>,
    pub instance: String,
    /// `main` / `twin`; empty = what this box is ([`crate::twin::is_twin_root`]).
    pub role: String,
    /// The key orders are verified under; `None` = the monetize key in force
    /// ([`crate::feed::monetize_in_force`]), which the installer may feed later.
    pub monetize_pubkey: Option<[u8; 32]>,
    /// The longest pause between reports when monetize names none.
    pub every: Duration,
}

/// **How this box reaches the others**, every call signed with its repair key.
pub struct ServerEstate {
    pub key: PrivateKey,
    /// This box's own control plane, as another box dials it.
    pub own: String,
    /// What `restate_entitlements` reaches ([`pull_loop`] installs the box's own).
    pub entitlements: Entitlements,
}

/// ★ UW-4: the `Admin.Health` alarm a `grow_room` order raises.
pub const ROOM_WANTED: &str = "grow.room_wanted";

/// **What `restate_entitlements` reaches on this box**: its stores (does it serve?)
/// and, in a metered build, the entitlement slot the facts go through — the same
/// `Entitlement.Set` path, newer-wins, nothing deleted. `Default`: none, refused by name.
#[derive(Clone, Default)]
pub struct Entitlements {
    stores: Option<Arc<ServerStores>>,
    /// The box's own control plane, read in-process for `Report.product`.
    control: Option<Arc<dyn gunnar_grpc::ControlPlane>>,
    #[cfg(feature = "metered")]
    slot: Option<Arc<crate::entitlement::EntitlementSlot>>,
    /// This box's own grow machine, for a pulled `grow_go_ahead` (G16-d).
    grow: Option<crate::paid_growth::PaidGrowth>,
    /// `--repair-twin`: the twin a main waits for before it grows.
    twin: Option<String>,
    /// As whom the pull channel pulls ([`Entitlements::pulled_as`]).
    instance: String,
    role: String,
    monetize_key: Option<[u8; 32]>,
    /// S10: how this main mails a resend sign-in link it minted ([`crate::postman`]).
    #[cfg(feature = "fleet-pull")]
    postman: Option<Arc<dyn crate::postman::Postman>>,
}

impl Entitlements {
    pub fn of(stores: Arc<ServerStores>) -> Entitlements {
        Entitlements {
            stores: Some(stores),
            ..Entitlements::default()
        }
    }

    /// The control plane `Report.product` is read from.
    pub fn with_control(mut self, control: Arc<dyn gunnar_grpc::ControlPlane>) -> Entitlements {
        self.control = Some(control);
        self
    }

    /// **`Report.product`** (ONE monetize, G1 pull): JSON text of
    /// `monetize_product::pull::ProductReport`, on a main that serves; `None` otherwise.
    pub fn product(&self, now_unix_ms: u64) -> Option<String> {
        let (stores, control) = (self.stores.as_ref()?, self.control.as_ref()?);
        if health(stores, "").get("serving").map(String::as_str) != Some("yes") {
            return None;
        }
        let (promised, metered) = self.promised_and_door();
        let mut report =
            crate::product_report::read(control.as_ref(), promised, metered, now_unix_ms);
        report.accounts = crate::product_report::accounts_of(control.as_ref(), stores, &report);
        report.activity =
            crate::product_report::activity_of(control.as_ref(), &report, now_unix_ms);
        serde_json::to_string(&report)
            .map_err(|e| tracing::warn!(target: TARGET, "Report.product does not serialize: {e}"))
            .ok()
    }

    /// Each tenant's enforced caps, and the paid door as this build knows it.
    fn promised_and_door(
        &self,
    ) -> (
        BTreeMap<String, BTreeMap<String, u64>>,
        crate::product_report::wire::MeteredDoor,
    ) {
        use crate::product_report::wire::MeteredDoor;
        #[cfg(feature = "metered")]
        if let Some(slot) = &self.slot {
            let promised = match slot.list() {
                Ok(facts) => facts
                    .into_iter()
                    .map(|f| (f.namespace, f.caps.into_iter().collect()))
                    .collect(),
                Err(e) => {
                    tracing::warn!(target: TARGET, "Report.product: the entitlements could not be listed: {e}");
                    BTreeMap::new()
                }
            };
            let door = if slot.keyed() {
                MeteredDoor::Enforcing(format!("gunnar {}: a metered build enforcing monetize's signed facts, said by the box itself", env!("CARGO_PKG_VERSION")))
            } else {
                MeteredDoor::Unproven(format!("gunnar {}: a metered build with no monetize key yet, so it serves every tenant Free", env!("CARGO_PKG_VERSION")))
            };
            return (promised, door);
        }
        #[cfg(feature = "metered")]
        let door = MeteredDoor::Unproven(format!("gunnar {}: a metered build started without its entitlement slot (no --monetize-pubkey)", env!("CARGO_PKG_VERSION")));
        #[cfg(not(feature = "metered"))]
        let door = MeteredDoor::Absent(format!(
            "gunnar {}: a community build, no metered services compiled in",
            env!("CARGO_PKG_VERSION")
        ));
        (BTreeMap::new(), door)
    }

    /// The grow machine a pulled `grow_go_ahead` drives.
    pub fn with_grow(mut self, grow: crate::paid_growth::PaidGrowth) -> Entitlements {
        self.grow = Some(grow);
        self
    }

    /// The twin a main waits for (`--repair-twin`); `None` = the installer's re-admit, else the
    /// twin the door named this main ([`crate::pairing::door_twin`]).
    pub fn with_twin(mut self, twin: Option<String>) -> Entitlements {
        self.twin = twin.filter(|t| !t.trim().is_empty());
        self
    }

    /// How this main mails (`sign_in_again`); `None` = no relay, a resend refused by name.
    #[cfg(feature = "fleet-pull")]
    pub fn with_postman(
        mut self,
        postman: Option<Arc<dyn crate::postman::Postman>>,
    ) -> Entitlements {
        self.postman = postman;
        self
    }

    /// The estate (`--fleet-instance`), role and order key the pull channel runs as.
    pub fn pulled_as(
        mut self,
        instance: &str,
        role: &str,
        monetize_key: Option<[u8; 32]>,
    ) -> Entitlements {
        self.instance = instance.to_owned();
        self.role = role.to_owned();
        self.monetize_key = monetize_key;
        self
    }

    fn paid(&self) -> Result<&crate::paid_growth::PaidGrowth, Refusal> {
        self.grow
            .as_ref()
            .ok_or_else(|| crate::paid_growth::refuse(RefusalCode::BadOrder, "this box has no grow machine on its pull channel, so it grows no data set: grow_go_ahead refused"))
    }

    /// `grow_go_ahead` step 1: monetize's step verified for THIS estate and role.
    pub fn grow_check(
        &self,
        go_ahead: &str,
        data_gib: u64,
    ) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
        let key = self.monetize_key.or_else(crate::feed::monetize_in_force);
        self.paid()?.check(
            go_ahead,
            data_gib,
            &self.instance,
            key.as_ref(),
            self.role == "twin",
            crate::twin::now_ms(),
        )
    }

    /// The twin first: its `Bootstrap.GrowStatus` must say it holds this main's set.
    pub async fn twin_holds(
        &self,
        check: &gunnar_repair::exec::GrowCheck,
    ) -> Result<String, Refusal> {
        let paid = self.paid()?;
        let Some(addr) = self.twin.clone().or_else(|| {
            self.stores
                .as_ref()
                .and_then(|s| awaited(s).or_else(|| door_twin(s)))
        }) else {
            return Err(crate::paid_growth::refuse(
                RefusalCode::NoTwin,
                "the main grows only after its twin holds its set, and this box knows no twin address (--repair-twin, the installer's recharge_from, or a twin in the door's peer list): nothing moved",
            ));
        };
        let unreachable = |why: String| {
            crate::paid_growth::refuse(
                RefusalCode::Unreachable,
                format!("the twin at {addr} did not say what it holds ({why}): nothing moved"),
            )
        };
        let channel = dial(&addr).map_err(|f| unreachable(f.to_string()))?;
        let status = gunnar_proto::bootstrap_client::BootstrapClient::new(channel)
            .grow_status(gunnar_proto::GrowStatusRequest::default())
            .await
            .map_err(|s| unreachable(format!("{}: {}", s.code(), s.message())))?
            .into_inner();
        crate::paid_growth::twin_holds(&status, paid.members(), check.data_gib)
            .map(|said| format!("the twin at {addr}: {said}"))
    }

    pub fn grow_start(
        &self,
        check: &gunnar_repair::exec::GrowCheck,
        nonce: &str,
    ) -> Result<gunnar_repair::exec::GrowState, Refusal> {
        self.paid()?.start(check, nonce, crate::twin::now_ms())
    }

    pub async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
        match self.paid() {
            Ok(g) => g.settle(past_waiting).await,
            Err(r) => gunnar_repair::exec::GrowState::Failed(r.to_string()),
        }
    }

    pub fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
        self.paid()?.look_again(give_up)
    }

    #[cfg(feature = "metered")]
    pub fn with_slot(
        mut self,
        slot: Option<Arc<crate::entitlement::EntitlementSlot>>,
    ) -> Entitlements {
        self.slot = slot;
        self
    }

    /// **The snapshot** (`monetize_embed::Snapshot` as JSON) from the ONE monetize:
    /// only on a box that serves; verified whole (envelope and every fact) under the
    /// box's monetize key, else refused whole; applied newer-wins; nothing deleted.
    /// An empty snapshot is normal and asks nothing, so even a box without
    /// entitlements answers it ok.
    pub fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let Some(stores) = &self.stores else {
            return Err(bad(
                "this box holds no entitlements, so it cannot run restate_entitlements".into(),
            ));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no entitlements restated") });
        }
        let Some(facts) = snapshot
            .get("facts")
            .and_then(|f| f.as_array())
            .map(Vec::len)
        else {
            return Err(bad(
                "the snapshot carries no list of facts: not a monetize snapshot".into(),
            ));
        };
        #[cfg(feature = "metered")]
        if let Some(slot) = &self.slot {
            let snap: monetize_embed::signing::Snapshot = serde_json::from_value(snapshot.clone())
                .map_err(|e| bad(format!("the snapshot does not decode: {e}")))?;
            let done = slot.restate(&snap).map_err(bad)?;
            return Ok(restated(done.applied, done.held));
        }
        if facts == 0 {
            return Ok(restated(0, 0));
        }
        Err(bad(format!("this box enforces no entitlements (not a metered build, or no monetize key): the snapshot's {facts} entitlement facts are not restated")))
    }
}

impl Entitlements {
    /// **Order `entitlement`**: one signed fact (`monetize_embed::EntitlementFact` as
    /// JSON) from the ONE monetize — what `Entitlement.Set` took. Only on a box that
    /// serves; verified under the box's monetize key (a forgery moves nothing); newer
    /// wins, and a fact this box has moved past is acked as held, not refused, so
    /// monetize stops asking.
    pub fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let Some(stores) = &self.stores else {
            return Err(bad(
                "this box holds no entitlements, so it cannot run an entitlement order".into(),
            ));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): the entitlement is not applied") });
        }
        #[cfg(feature = "metered")]
        if let Some(slot) = &self.slot {
            use crate::entitlement::SetRefused;
            let fact: monetize_embed::EntitlementFact = serde_json::from_value(fact.clone())
                .map_err(|e| bad(format!("the body is not a signed entitlement fact: {e}")))?;
            let tenant = fact.tenant.0.clone();
            return match slot.push(&fact) {
                Ok(_) => Ok(format!("{tenant} is {:?}", fact.state)),
                Err(SetRefused::Replayed { .. } | SetRefused::Behind { .. }) => Ok(format!(
                    "{tenant}: this box already holds a fact as new or newer; nothing moved"
                )),
                Err(e) => Err(bad(e.to_string())),
            };
        }
        let tenant = fact.get("tenant").and_then(|t| t.as_str()).unwrap_or("?");
        Err(bad(format!("this box enforces no entitlements (not a metered build, or no slot): the fact for {tenant} is not applied")))
    }
}

impl Entitlements {
    /// **Order `add_principal`** (ONE monetize, G1 pull): a paid signup's account — the
    /// principal and its namespace, as `Access.AddPrincipal` makes them — on a main that
    /// serves. The ack is monetize-product's contract: `Ok(namespace)`, also on a repeat for
    /// the same `signup`; [`RefusalCode::Taken`] when the name is somebody else's (a
    /// principal, a namespace, or another signup's); anything else a final refusal in words.
    /// Only a `person` is made (G-D36). Which signup made which name is kept in the control
    /// store ([`SIGNUPS_REF`]), so the twin follows it and a recharged main still knows.
    pub fn add_principal(
        &self,
        username: &str,
        kind: &str,
        signup: &str,
    ) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
            return Err(bad(format!("this box makes no accounts (no control plane in reach), so it cannot run add_principal for {username}")));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no account made for {username}") });
        }
        if kind != "person" {
            return Err(bad(format!("{username} is asked as {kind:?}: a signup makes a person's account only; an org is a clan its admin makes")));
        }
        if gunnar_grpc::service::access::is_reserved_principal_name(username) {
            return Err(bad(format!("{username:?} is a reserved name (`console-*`, `kit-*`, `gunnar-ui`, `uitest-console`): no person may carry it")));
        }
        let store = stores.control_store().as_ref();
        let exists = |name: &str| -> Result<(bool, bool), Refusal> {
            let principal = stores
                .auth()
                .principal(name)
                .map_err(|e| bad(format!("the principals cannot be read: {e}")))?
                .is_some();
            let namespace = stores
                .catalog()
                .namespace(name)
                .map_err(|e| bad(format!("the namespaces cannot be read: {e}")))?
                .is_some();
            Ok((principal, namespace))
        };
        let made = signups::read(store).map_err(bad)?;
        match made.get(signup) {
            Some(name) if name == username => {
                // A repeat: monetize lost the ack. Made, or the record outran the account.
                if !exists(username)?.0 {
                    create(control.as_ref(), username, signup)
                        .map_err(|e| e.into_refusal(username))?;
                }
                return Ok(username.to_owned());
            }
            Some(name) if exists(name)?.0 => {
                return Err(bad(format!("signup {signup} already made the account {name}; it makes no second one ({username})")));
            }
            _ => {}
        }
        let claimed = made.iter().any(|(s, name)| name == username && s != signup);
        let (principal, namespace) = exists(username)?;
        if claimed || principal || namespace {
            return Err(taken(username));
        }
        if let Some(plane) = control.enterprise() {
            let verdict = plane.name_check("account", username, "").map_err(|e| {
                bad(format!(
                    "the name filter did not answer for {username}: {e}"
                ))
            })?;
            if !verdict.is_clean() {
                return Err(bad(format!(
                    "the name filter refuses {username:?} ({}): {}",
                    verdict.verdict, verdict.sentence
                )));
            }
        }
        // The record first: a crash between the two leaves a record a repeat completes,
        // never an account no signup owns.
        signups::put(store, signup, Some(username)).map_err(bad)?;
        if let Err(e) = create(control.as_ref(), username, signup) {
            if let Err(why) = signups::put(store, signup, None) {
                tracing::warn!(target: TARGET, "add_principal {username}: the signup record {signup} could not be taken back: {why}");
            }
            return Err(e.into_refusal(username));
        }
        Ok(username.to_owned())
    }
}

/// The control document of which signup made which account: `{signup: username}`.
pub const SIGNUPS_REF: &str = "refs/gunnar/signups";

impl Entitlements {
    /// **Order `sign_in_again`** (S10, owner 2026-10-11 05:53Z): a sold estate's buyer asked
    /// the shop for a NEW sign-in link; the shop queued this signed order and holds no secret.
    /// On a main that serves, with a mail relay: `account`'s principal, a person holding NO
    /// passkey yet (a link enrols the first one only), gets a fresh invite minted HERE — the
    /// `Access.CreateInvite` shape, its hash stored, living [`gunnar_repair::sign_in::ttl`] —
    /// and the link (the first link's shape) is mailed to `to` by this box. The ack says who
    /// was mailed, never the secret. A mail that does not go withdraws the invite.
    /// ★ UW-5 **Order `add_users`** (ONE monetize, G1 pull): "Add users", granted on the
    /// account `tenant` as the SELLER — the `Accounts.EnrollEmployee` rules
    /// ([`gunnar_grpc::service::accounts::users_asked`]), under `order` (a retry answers the
    /// first grant `unchanged`), no room check (monetize sold exactly the shortfall before the
    /// money moved). A new named user's invite is minted here and its secret rides the answer
    /// once (monetize mails it). The ack is monetize-product's `UsersAnswer` as JSON.
    pub fn add_users(
        &self,
        tenant: &str,
        ask: &serde_json::Value,
        order: &str,
    ) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
            return Err(bad(format!("this box adds no users (no control plane in reach), so it cannot run add_users for {tenant}")));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no users added to {tenant}") });
        }
        let count = ask
            .get("count")
            .and_then(|v| v.as_u64())
            .unwrap_or(1)
            .clamp(1, u64::from(u32::MAX)) as u32;
        let name = ask
            .get("name")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .trim()
            .to_owned();
        let named = count == 1;
        let request = gunnar_proto::EnrollEmployeeRequest {
            account: tenant.to_owned(),
            id: if named { name.clone() } else { String::new() },
            name: if named { name.clone() } else { String::new() },
            public_key: String::new(),
            room_bytes: ask.get("room_bytes").and_then(|v| v.as_u64()).unwrap_or(0),
            count,
            from_seat: false,
            dry_run: false,
            order: order.to_owned(),
            invite: named,
        };
        let employee = gunnar_grpc::service::accounts::users_asked(&request)
            .map_err(|s| bad(format!("add_users for {tenant}: {}", s.message())))?;
        for text in [employee.id.as_str(), employee.name.as_str()] {
            if !text.is_empty() && gunnar_grpc::service::access::is_reserved_principal_name(text) {
                return Err(bad(format!("{text:?} is a reserved name (`console-*`, `kit-*`, `gunnar-ui`, `uitest-console`): no user may carry it")));
            }
        }
        let answer = control
            .enroll_employee(tenant, &employee)
            .map_err(|e| bad(format!("add_users for {tenant}: {e}")))?;
        let mut invite_secret = String::new();
        if let (Some(principal), true, false) = (&answer.principal, named, answer.unchanged) {
            // The identity core's first-login link (IDENTITY-HOST): seven days.
            let link = stores
                .identity()
                .ok_or_else(|| "this box's identity core is not up yet".to_owned())
                .and_then(|identity| {
                    identity
                        .enrolment(
                            &principal.id,
                            "monetize:add_users",
                            Some(gunnar_grpc::service::access::INVITE_DEFAULT_MS / 1000),
                        )
                        .map_err(|refused| refused.to_string())
                });
            match link {
                Ok(made) => invite_secret = made.secret,
                Err(e) => {
                    tracing::warn!(target: TARGET, "add_users {tenant}: {} was added and its invite was refused: {e}", principal.id)
                }
            }
        }
        if !answer.unchanged {
            tracing::info!(
                target: gunnar_grpc::auth::AUDIT_TARGET,
                event = if named { "account.employee_enrolled" } else { "account.seats_added" },
                account = %tenant,
                employee = %answer.principal.as_ref().map(|p| p.id.as_str()).unwrap_or(""),
                seats = employee.seats,
                room_bytes = answer.room_bytes,
                order = %order,
                "an \"Add users\" order was delivered (add_users)"
            );
        }
        let out = serde_json::json!({
            "principal": answer.principal.as_ref().map(|p| p.id.clone()).unwrap_or_default(),
            "room_bytes": answer.room_bytes,
            "need_bytes": answer.need_bytes,
            "unused_bytes": answer.unused_bytes,
            "shortfall_bytes": answer.shortfall_bytes,
            "seats": answer.seats,
            "seats_open": answer.seats_open,
            "unchanged": answer.unchanged,
            "invite_secret": invite_secret,
        });
        Ok(out.to_string())
    }

    /// ★ UW-4 **Order `grow_room`**: room was sold past this set (owner: the customer never
    /// sees a cap — queued, resized later, the operator told). Kept as an `Admin.Health` alarm
    /// (`grow.room_wanted`) naming what is wanted and why, for this estate's operator; the
    /// set is resized by its growth. Acked with what was kept.
    pub fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
        let gib = need_bytes as f64 / (1u64 << 30) as f64;
        let words = format!("{gib:.1} GiB of stored room was committed past this set ({reason}) — queued: grow the data set");
        crate::feed::raise(ROOM_WANTED, &words);
        tracing::warn!(target: TARGET, event = ROOM_WANTED, need_bytes, "{words}");
        Ok(format!(
            "kept: {gib:.1} GiB wanted past the set; the operator is told ({ROOM_WANTED})"
        ))
    }

    /// ★ UW-11 **Order `purge_tenant`** (ONE monetize, G1 pull): the account `tenant` lapsed
    /// past retention and is deleted — `Accounts.Purge`'s own job on this main's control
    /// plane (lock first, then unlink; a machine's account is never purged). No `job`: the
    /// purge of the account already running is adopted, else one is started. A `job`: that
    /// job's state, refused when this box no longer knows it (monetize then starts again,
    /// never assumes the data went). The ack is monetize-product's `PurgeAnswer` as JSON.
    pub fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
        use gunnar_grpc::seam::{PurgeAccountOptions, PurgeState};
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
            return Err(bad(format!("this box purges no accounts (no control plane in reach), so it cannot run purge_tenant for {tenant}")));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): {tenant} is not purged here") });
        }
        let latest = control
            .purge_account_status(tenant)
            .map_err(|e| bad(format!("purge_tenant {tenant}: {e}")))?
            .0;
        let running = |s: PurgeState| matches!(s, PurgeState::Queued | PurgeState::Running);
        let found = match (job.trim(), latest) {
            ("", Some(j)) if running(j.state) => j,
            ("", _) => {
                let options = PurgeAccountOptions { principal_id: "monetize:nonpay".into(), reason: reason.to_owned(), remove_principal: false, expect_empty: false };
                let (j, _) = control.purge_account(tenant, options).map_err(|e| bad(format!("purge_tenant {tenant}: {e}")))?;
                tracing::info!(target: gunnar_grpc::auth::AUDIT_TARGET, event = "account.purge_ordered", account = %tenant, job = %j.job_id, reason = %reason, "a nonpay purge was ordered by monetize (purge_tenant)");
                j
            }
            (asked, Some(j)) if j.job_id == asked => j,
            (asked, _) => return Err(bad(format!("{tenant}: this box no longer reports purge job {asked}; nothing can say whether the data went"))),
        };
        let gib = found.bytes_reclaimed as f64 / (1u64 << 30) as f64;
        let out = serde_json::json!({
            "job": found.job_id,
            "running": running(found.state),
            "failed": matches!(found.state, PurgeState::Failed),
            "error": found.error.clone().unwrap_or_default(),
            "detail": format!("{} of {} stores, {gib:.1} GiB", found.stores_purged, found.stores_found),
            "bytes_reclaimed": found.bytes_reclaimed,
            "still_locked": !found.gate_released,
            "failures": found.failures,
        });
        Ok(out.to_string())
    }

    /// **`mail_relay`** (MAIL-RELAY 2026-10-28): the relay the ONE box fed this main again,
    /// kept sealed under the data root ([`crate::mail_relay::keep`]); the postman reads it over
    /// the lay's feed from the next mail.
    #[cfg(feature = "fleet-pull")]
    pub fn keep_mail_relay(
        &self,
        relay: &gunnar_repair::order::MailRelay,
    ) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let Some(stores) = &self.stores else {
            return Err(bad(format!(
                "this box has no data root in reach, so it keeps no mail relay (as {})",
                relay.smtp_from
            )));
        };
        crate::mail_relay::keep(stores.root(), relay).map_err(bad)
    }

    /// **A console setting the ONE box fed again** (a `refeed`, gunnar-repair's
    /// `keep_console_setting`): `value` stored under the generic setting `name`, as an
    /// administrator's `Admin.SetSetting` stores it. This server gives the name no meaning (it
    /// must stay generic: `tests/server_settings.rs`); it takes only a `console.*` name, the
    /// console's own namespace, so a refeed cannot write a setting of anything else. Only a
    /// serving main stores it; a twin answers that its main keeps it (the settings are control
    /// state, replicated to it). Audited on `gunnar::audit` (who: the ONE box's order).
    pub fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        if !name.starts_with("console.") {
            return Err(bad(format!(
                "a refeed keeps only a console setting (console.*), not {name}"
            )));
        }
        let Some(stores) = &self.stores else {
            return Err(bad(format!("this box has no data root in reach, so it keeps no console setting ({name} = {value})")));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Ok(format!("this box does not serve (serving={serving}): its main keeps {name}, and the setting replicates here"));
        }
        let previous = stores
            .settings()
            .set(name, value.as_bytes())
            .map_err(|e| bad(format!("the console setting {name} was not stored: {e}")))?;
        let was = previous
            .as_deref()
            .map(|b| String::from_utf8_lossy(b).into_owned())
            .unwrap_or_else(|| "(unset)".to_owned());
        tracing::warn!(
            target: gunnar_grpc::AUDIT_TARGET,
            event = "admin.setting_set",
            principal = "monetize (refeed order)",
            name = name,
            old = %was,
            new = value,
            "a server setting was written: the ONE box fed a console setting again"
        );
        Ok(format!(
            "{name}: {was} -> {value}; the console follows it within its recheck"
        ))
    }

    #[cfg(feature = "fleet-pull")]
    pub async fn sign_in_again(
        &self,
        account: &str,
        to: &str,
        host: &str,
        ttl_secs: u64,
    ) -> Result<String, Refusal> {
        let bad = |message: String| Refusal {
            code: RefusalCode::BadOrder,
            message,
        };
        let (Some(stores), Some(_)) = (&self.stores, &self.control) else {
            return Err(bad(format!("this box mints no sign-in links (no control plane in reach), so it cannot run sign_in_again for {account}")));
        };
        let serving = health(stores, "")
            .remove("serving")
            .unwrap_or_else(|| "no (a twin)".into());
        if serving != "yes" {
            return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no sign-in link minted for {account}") });
        }
        let Some(postman) = &self.postman else {
            return Err(bad(format!(
                "this box has no mail relay ({} and a token), so it mints no sign-in link for {account} it could not mail",
                crate::postman::FROM_ENV
            )));
        };
        postman.ready().map_err(|why| bad(format!("this box cannot mail ({why}), so it mints no sign-in link for {account} it could not mail")))?;
        gunnar_repair::sign_in::check(account, to, host)
            .map_err(|why| bad(format!("sign_in_again: {why}")))?;
        let principal = stores
            .auth()
            .principal_by_name(account)
            .map_err(|e| bad(format!("the principals cannot be read: {e}")))?
            .ok_or_else(|| {
                bad(format!(
                    "{account} is no account on this box: no sign-in link minted"
                ))
            })?;
        if principal.kind != gunnar_core::PrincipalKind::User {
            return Err(bad(format!(
                "{account} is not a person: a sign-in link enrols a person's passkey"
            )));
        }
        let passkeys = stores
            .auth()
            .web_credentials(Some(&principal.id))
            .map_err(|e| bad(format!("the passkeys cannot be read: {e}")))?;
        if !passkeys.is_empty() {
            return Err(bad(format!(
                "{account} already signs in with a passkey ({} held): a sign-in link only enrols the first one — sign in, or add a passkey from a signed-in device",
                passkeys.len()
            )));
        }
        let ttl = gunnar_repair::sign_in::ttl(ttl_secs);
        // The identity core's first-login link (IDENTITY-HOST): minted here,
        // kept only as its hash, withdrawn when the mail does not go.
        let identity = stores.identity().ok_or_else(|| {
            bad(format!(
                "this box's identity core is not up yet: no sign-in link for {account}; ask again"
            ))
        })?;
        let invite = identity
            .enrolment(
                &principal.id,
                "monetize:sign_in_again",
                Some(i64::try_from(ttl).unwrap_or(i64::MAX)),
            )
            .map_err(|e| bad(format!("the invite for {account} could not be made: {e}")))?;
        let secret = invite.secret.clone();
        let mail = gunnar_repair::sign_in::mail(host, account, to, &secret, ttl);
        drop(secret);
        if let Err(why) = postman.post(&mail).await {
            if let Err(e) = identity.withdraw_invite(&invite.invite_id, "monetize:sign_in_again") {
                tracing::warn!(target: TARGET, "sign_in_again {account}: the unmailed invite {} could not be withdrawn: {e}", invite.invite_id);
            }
            return Err(bad(format!("the sign-in link for {account} could not be mailed to {to} ({why}); the invite is withdrawn — ask again")));
        }
        tracing::info!(target: TARGET, account, invite = %invite.invite_id, ttl_secs = ttl, "sign_in_again: a new sign-in link was minted here and mailed to the order's address");
        Ok(format!(
            "a new sign-in link for {account} (invite {}, {}) was mailed to {to}",
            invite.invite_id,
            gunnar_repair::sign_in::ttl_words(ttl)
        ))
    }
}

fn taken(username: &str) -> Refusal {
    Refusal { code: RefusalCode::Taken, message: format!("{username} is somebody else's (a principal, a namespace or another signup holds it): choose another name") }
}

enum NotMade {
    Exists,
    Other(String),
}

impl NotMade {
    fn into_refusal(self, username: &str) -> Refusal {
        match self {
            NotMade::Exists => taken(username),
            NotMade::Other(why) => Refusal {
                code: RefusalCode::BadOrder,
                message: format!("the account {username} was not made: {why}"),
            },
        }
    }
}

/// The principal and its namespace, in one claim (`ControlPlane::add_principal`), and the
/// account's audit line.
fn create(
    control: &dyn gunnar_grpc::ControlPlane,
    username: &str,
    signup: &str,
) -> Result<(), NotMade> {
    use gunnar_grpc::ControlError;
    match control.add_principal(username, username, gunnar_proto::PrincipalKind::User) {
        Ok(_) => {}
        Err(ControlError::AlreadyExists(_)) => return Err(NotMade::Exists),
        Err(e) => return Err(NotMade::Other(e.to_string())),
    }
    tracing::info!(target: TARGET, username, signup, "add_principal: a signup's account was made");
    if let Some(plane) = control.enterprise() {
        if let Ok(Some(account)) = plane.account_of_principal(username) {
            let event = gunnar_grpc::enterprise::AccountAuditEvent {
                kind: "account.created".into(),
                actor: "monetize".into(),
                actor_machine: true,
                target: username.to_owned(),
                detail: format!("a signup at the shop ({signup}), order add_principal"),
                refused: false,
            };
            if let Err(e) = plane.account_audit_record(&account, &event) {
                tracing::warn!(target: TARGET, "add_principal {username}: the account's audit line was not written: {e}");
            }
        }
    }
    Ok(())
}

mod signups {
    use std::collections::BTreeMap;

    use gunnar_core::ControlStore;

    use super::SIGNUPS_REF;

    pub(super) fn read(store: &dyn ControlStore) -> Result<BTreeMap<String, String>, String> {
        Ok(load(store)?.1)
    }

    fn load(
        store: &dyn ControlStore,
    ) -> Result<(Option<gunnar_core::BlobId>, BTreeMap<String, String>), String> {
        let at = store
            .read_doc(SIGNUPS_REF)
            .map_err(|e| format!("{SIGNUPS_REF} cannot be read: {e}"))?;
        let Some(id) = at else {
            return Ok((None, BTreeMap::new()));
        };
        let bytes = store
            .get_blob(&id)
            .map_err(|e| format!("{SIGNUPS_REF} cannot be read: {e}"))?
            .ok_or_else(|| format!("{SIGNUPS_REF} points at {id}, which is not stored"))?;
        let map = serde_json::from_slice(&bytes)
            .map_err(|e| format!("{SIGNUPS_REF} does not decode: {e}"))?;
        Ok((Some(id), map))
    }

    /// Record (`Some`) or take back (`None`) what `signup` made; compare-and-swap, re-read on a lost race.
    pub(super) fn put(
        store: &dyn ControlStore,
        signup: &str,
        username: Option<&str>,
    ) -> Result<(), String> {
        for _ in 0..16 {
            let (at, mut map) = load(store)?;
            match username {
                Some(u) => map.insert(signup.to_owned(), u.to_owned()),
                None => map.remove(signup),
            };
            let bytes = serde_json::to_vec(&map).map_err(|e| e.to_string())?;
            let id = store
                .put_blob(&bytes)
                .map_err(|e| format!("{SIGNUPS_REF} cannot be written: {e}"))?;
            if store
                .commit_doc(SIGNUPS_REF, at, id)
                .map_err(|e| format!("{SIGNUPS_REF} cannot be written: {e}"))?
            {
                return Ok(());
            }
        }
        Err(format!("{SIGNUPS_REF}: 16 writers in a row moved it first"))
    }
}

fn restated(applied: usize, held: usize) -> String {
    format!("restated {applied} entitlement facts ({held} already held as new or newer)")
}

impl ServerEstate {
    /// The repair key named by `cfg`, else the FED one ([`crate::feed::FEED_REPAIR_KEY`]),
    /// else `<data>/twin.key`.
    pub fn new(
        root: &std::path::Path,
        cfg: &RepairConfig,
        own: String,
    ) -> Result<ServerEstate, String> {
        Self::with_fed(root, cfg, own, crate::feed::Fed::current().as_ref())
    }

    /// [`ServerEstate::new`] over the feed `fed` (the one in force, or a test's).
    pub fn with_fed(
        root: &std::path::Path,
        cfg: &RepairConfig,
        own: String,
        fed: Option<&crate::feed::Fed>,
    ) -> Result<ServerEstate, String> {
        let key = match (&cfg.key, fed.and_then(|f| f.repair_key.as_ref())) {
            (Some(path), _) => {
                let bytes = std::fs::read(path)
                    .map_err(|e| format!("the repair key {}: {e}", path.display()))?;
                PrivateKey::from_openssh(&bytes).map_err(|e| {
                    format!(
                        "the repair key {} is not an OpenSSH private key: {e}",
                        path.display()
                    )
                })?
            }
            (None, Some(k)) => PrivateKey::from_openssh(k.0.trim().as_bytes()).map_err(|e| {
                format!(
                    "the fed {} is not an OpenSSH private key: {e}",
                    crate::feed::FEED_REPAIR_KEY
                )
            })?,
            (None, None) => crate::twin::twin_private_key(root).map_err(|e| {
                format!(
                    "no --repair-key, no fed {}, and no twin key on this box ({e})",
                    crate::feed::FEED_REPAIR_KEY
                )
            })?,
        };
        Ok(ServerEstate {
            key,
            own,
            entitlements: Entitlements::default(),
        })
    }
}

/// **Is auto-fix on**: the medium's `--repair-auto`, or the estate's fed opt-in
/// ([`crate::feed::FEED_REPAIR_AUTO`]) in force now.
pub fn auto_on(cfg: &RepairConfig) -> bool {
    auto_with(cfg, crate::feed::Fed::current().as_ref())
}

/// Pure: [`auto_on`] over the feed `fed`.
pub fn auto_with(cfg: &RepairConfig, fed: Option<&crate::feed::Fed>) -> bool {
    cfg.auto || fed.is_some_and(|f| f.repair_auto)
}

/// What the serve armed the awaiting-main kick with ([`arm`]).
struct Armed {
    cfg: RepairConfig,
    own: String,
    rt: tokio::runtime::Handle,
}

fn armed() -> &'static std::sync::OnceLock<Armed> {
    static ARMED: std::sync::OnceLock<Armed> = std::sync::OnceLock::new();
    &ARMED
}

/// When the last awaiting-main kick started, and whether one is running.
fn kicked() -> &'static std::sync::Mutex<(Option<std::time::Instant>, bool)> {
    static KICKED: std::sync::Mutex<(Option<std::time::Instant>, bool)> =
        std::sync::Mutex::new((None, false));
    &KICKED
}

/// After a kick that did not finish the recharge, the next is no sooner than this.
pub const KICK_AGAIN_AFTER: Duration = Duration::from_secs(60);

/// **Arm the awaiting-main kick** (main rebuild, 2026-10-12): the serve hands the repair
/// configuration, this box's own address and its runtime. Once per process.
pub fn arm(cfg: RepairConfig, own: String, rt: tokio::runtime::Handle) {
    let _ = armed().set(Armed { cfg, own, rt });
}

/// Pure: **may an awaiting main kick its own recharge now?** Auto-fix on, none running,
/// and the last one at least [`KICK_AGAIN_AFTER`] ago.
pub fn kick_due(auto: bool, running: bool, last: Option<Duration>) -> bool {
    auto && !running && last.is_none_or(|ago| ago >= KICK_AGAIN_AFTER)
}

/// **A re-laid main awaits its twin** (pairing's `Hold::Awaiting`): with auto-fix on it
/// fills itself — the boot check's recharge, run now that the door has named the twin
/// (the boot check ran at start, before the feed did). `None`: nothing was started;
/// `Some(words)`: what was. The guard is the boot check's own: a fresh box with a twin
/// that can give it back (`gunnar_repair::rebuild::judge` → `Fill`); nothing is destroyed.
/// A kick that cannot start (no repair key) says why once and is due again at the next turn.
pub fn on_awaiting(stores: &Arc<ServerStores>, twin: &str) -> Option<String> {
    let a = armed().get()?;
    // A box-to-box call never leaves a box in the clear: with no client certificate for
    // another box's control plane, this box cannot dial its twin — the FRONT fills it then
    // (gunnar-front `autofill`, the console's key over the tunnel). Said once, not a WARN.
    if !may_dial(twin) {
        static SAID: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
        if auto_on(&a.cfg) && !SAID.swap(true, std::sync::atomic::Ordering::SeqCst) {
            tracing::info!(target: TARGET, twin, "auto-fix: this box holds no client certificate to dial its twin; the front fills it from the twin");
        }
        return None;
    }
    {
        let mut k = kicked().lock().unwrap_or_else(|p| p.into_inner());
        if !kick_due(auto_on(&a.cfg), k.1, k.0.map(|t| t.elapsed())) {
            return None;
        }
        *k = (Some(std::time::Instant::now()), true);
    }
    let estate = match ServerEstate::new(stores.root(), &a.cfg, a.own.clone()) {
        Ok(e) => e,
        Err(why) => {
            // Nothing started: the next turn may try again at once (a feed may bring the
            // key), and the same reason is said once, not on every turn.
            *kicked().lock().unwrap_or_else(|p| p.into_inner()) = (None, false);
            static SAID: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
            let mut said = SAID.lock().unwrap_or_else(|p| p.into_inner());
            if *said == why {
                return None;
            }
            said.clone_from(&why);
            tracing::warn!(target: TARGET, twin, "auto-fix: this re-laid main awaits its twin and cannot fill itself ({why}); an admin or a guardian order can");
            return Some(format!("auto-fix cannot run: {why}"));
        }
    };
    let (stores, twin_at) = (Arc::clone(stores), twin.to_owned());
    // A32-14, stays a task: one boot check of the auto-fix, async on the server's runtime; its report is the record.
    a.rt.spawn(async move {
        let dead = data_dead(&stores);
        let report = exec::boot_check(&estate, Some(&twin_at), dead, true, &|| false).await;
        match &report.recharge {
            Some(Ok(r)) => tracing::info!(target: TARGET, twin = %twin_at, head_seq = r.head_seq, "auto-fix: the re-laid main filled itself from its twin: {:?} — {}", r.phase, r.detail),
            Some(Err(why)) => tracing::error!(target: TARGET, twin = %twin_at, "auto-fix: the recharge from the twin was refused — {why}"),
            None => tracing::warn!(target: TARGET, twin = %twin_at, finding = ?report.finding, "auto-fix: the boot check found nothing it may recharge"),
        }
        kicked().lock().unwrap_or_else(|p| p.into_inner()).1 = false;
    });
    Some(format!(
        "auto-fix: filling this re-laid main from its twin at {twin}"
    ))
}

/// Can this box dial `at` for a repair call: on this box, or with a client certificate.
pub fn may_dial(at: &str) -> bool {
    gunnar_grpc::client_signer::is_loopback_target(at)
        || matches!(
            gunnar_grpc::client_signer::ClientTrust::from_env(),
            Ok(Some(_))
        )
}

/// A control-plane channel to `at`: mutual TLS when this box holds a client
/// certificate, else plain (which `plain_channel` allows only on this machine).
pub fn dial(at: &str) -> Result<ControlChannel, Fault> {
    use gunnar_grpc::client_signer as cs;
    let trust = cs::ClientTrust::from_env().map_err(|e| Fault::new("unauthenticated", e))?;
    match trust {
        Some(trust) => cs::mutual_channel(at, &trust, &cs::tls_domain_for(at)),
        None => cs::plain_channel(at),
    }
    .map_err(|e| Fault::new("unavailable", format!("{at}: {e}")))
}

impl Estate for ServerEstate {
    type Twin = ProtoTwin;
    type Main = ProtoMain;
    type Fence = ProtoFence;
    fn twin(&self, addr: &str) -> Result<ProtoTwin, Fault> {
        Ok(ProtoTwin {
            addr: addr.to_owned(),
            channel: dial(addr)?,
            key: self.key.clone(),
        })
    }
    fn main(&self, addr: &str) -> Result<ProtoMain, Fault> {
        Ok(ProtoMain {
            addr: addr.to_owned(),
            channel: dial(addr)?,
            key: self.key.clone(),
        })
    }
    fn fence(&self, addr: &str) -> Result<ProtoFence, Fault> {
        Ok(ProtoFence {
            addr: addr.to_owned(),
            channel: dial(addr)?,
            key: self.key.clone(),
        })
    }
    fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
        self.entitlements.restate(snapshot)
    }
    fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
        self.entitlements.entitle(fact)
    }
    fn add_principal(&self, username: &str, kind: &str, signup: &str) -> Result<String, Refusal> {
        self.entitlements.add_principal(username, kind, signup)
    }
    fn add_users(
        &self,
        tenant: &str,
        ask: &serde_json::Value,
        order: &str,
    ) -> Result<String, Refusal> {
        self.entitlements.add_users(tenant, ask, order)
    }
    fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
        self.entitlements.grow_room(need_bytes, reason)
    }
    fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
        self.entitlements.purge_tenant(tenant, reason, job)
    }
    #[cfg(feature = "fleet-pull")]
    async fn sign_in_again(
        &self,
        account: &str,
        to: &str,
        host: &str,
        ttl_secs: u64,
    ) -> Result<String, Refusal> {
        self.entitlements
            .sign_in_again(account, to, host, ttl_secs)
            .await
    }
    #[cfg(feature = "fleet-pull")]
    fn keep_mail_relay(&self, relay: &gunnar_repair::order::MailRelay) -> Result<String, Refusal> {
        self.entitlements.keep_mail_relay(relay)
    }
    fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
        self.entitlements.keep_console_setting(name, value)
    }
    fn grow_check(
        &self,
        go_ahead: &str,
        data_gib: u64,
    ) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
        self.entitlements.grow_check(go_ahead, data_gib)
    }
    async fn twin_holds(&self, check: &gunnar_repair::exec::GrowCheck) -> Result<String, Refusal> {
        self.entitlements.twin_holds(check).await
    }
    fn grow_start(
        &self,
        check: &gunnar_repair::exec::GrowCheck,
        nonce: &str,
    ) -> Result<gunnar_repair::exec::GrowState, Refusal> {
        self.entitlements.grow_start(check, nonce)
    }
    async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
        self.entitlements.grow_settle(past_waiting).await
    }
    fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
        self.entitlements.grow_look_again(give_up)
    }
    fn own(&self) -> String {
        self.own.clone()
    }
    fn now_unix_ms(&self) -> i64 {
        crate::twin::now_ms()
    }
}

/// This box's control plane as a peer dials it: an unspecified bind is this machine.
pub fn own_addr(bound: std::net::SocketAddr) -> String {
    if bound.ip().is_unspecified() {
        std::net::SocketAddr::from(([127, 0, 0, 1], bound.port())).to_string()
    } else {
        bound.to_string()
    }
}

#[cfg(feature = "wg-client")]
fn fresh(stores: &ServerStores) -> bool {
    crate::pairing::is_fresh(stores).unwrap_or(false)
}

#[cfg(not(feature = "wg-client"))]
fn fresh(stores: &ServerStores) -> bool {
    stores.auth().is_empty().unwrap_or(false)
        && stores
            .catalog()
            .list(None)
            .map(|l| l.is_empty())
            .unwrap_or(false)
}

#[cfg(feature = "wg-client")]
fn awaited(stores: &ServerStores) -> Option<String> {
    crate::pairing::recharge_awaited(stores)
}

#[cfg(not(feature = "wg-client"))]
fn awaited(_: &ServerStores) -> Option<String> {
    None
}

/// The twin the door named this main ([`crate::pairing::door_twin`]).
#[cfg(feature = "wg-client")]
fn door_twin(stores: &ServerStores) -> Option<String> {
    crate::pairing::door_twin(stores.root())
}

#[cfg(not(feature = "wg-client"))]
fn door_twin(_: &ServerStores) -> Option<String> {
    None
}

/// **Is this box's data set dead**: a restore it never finished, or a box with no
/// identity and no repository (a re-laid main, an emptied disk). Never on a twin.
pub fn data_dead(stores: &ServerStores) -> bool {
    !crate::twin::is_twin_root(stores.root())
        && (crate::twin::is_restoring(stores.root()) || fresh(stores))
}

/// `<data>/repair/hold`: the boot check found this box's data set lost with a twin
/// able to give it back (R-7b). It names that twin; while it stands and the box is
/// still dead, `Bootstrap.Enroll` takes no first admin and `Bootstrap.Status`
/// answers the twin in `recharge_from`.
pub const HOLD_REL: &str = "repair/hold";

/// **Hold this box** for a recharge from `twin` (R-7b). The boot check's act on
/// `Finding::DataLost`, whether or not auto-fix then recharges.
pub fn hold(stores: &ServerStores, twin: &str) -> anyhow::Result<()> {
    if let Some(dir) = stores.root().join(HOLD_REL).parent() {
        std::fs::create_dir_all(dir)?;
    }
    crate::durable::write_atomic_unframed(
        stores.root(),
        HOLD_REL,
        format!("{}\n", twin.trim()).as_bytes(),
    )
}

/// **The twin this box is held for**, if it is held. The hold lifts by itself once
/// the box holds a data set again (the recharge finished: the marker goes); the
/// admin's way out is removing the marker by hand.
pub fn held(stores: &ServerStores) -> Option<String> {
    let path = stores.root().join(HOLD_REL);
    let twin = std::fs::read_to_string(&path).ok()?.trim().to_owned();
    if !data_dead(stores) {
        if std::fs::remove_file(&path).is_ok() {
            tracing::info!(target: TARGET, twin = %twin, "the hold is lifted: this box holds its data set again");
        }
        return None;
    }
    (!twin.is_empty()).then_some(twin)
}

/// What `Bootstrap.Enroll` answers on a held box.
pub fn held_refusal(twin: &str) -> String {
    format!(
        "{}: this box is HELD for a recharge from the twin at {twin}. A first admin now would start a new, \
         empty estate and shut the window the recharge needs. Recharge it (the Twin panel, `gunnar control twin \
         replay`, or a guardian order); to start empty on purpose, remove <data>/{HOLD_REL} and enrol again",
        gunnar_repair::DATA_LOST_LINE
    )
}

/// The twin the boot check reviews: the configured one, else the one the installer's
/// re-admit named.
pub fn twin_of(stores: &ServerStores, cfg: &RepairConfig) -> Option<String> {
    cfg.twin
        .clone()
        .filter(|t| !t.trim().is_empty())
        .or_else(|| awaited(stores))
}

/// **The boot check** — once, after the control plane listens. A healthy box, a twin,
/// or a box with no twin to name says nothing.
pub async fn boot(stores: Arc<ServerStores>, cfg: RepairConfig, own: String) -> Option<BootReport> {
    if crate::twin::is_twin_root(stores.root()) {
        return None;
    }
    let dead = data_dead(&stores);
    let twin = twin_of(&stores, &cfg);
    if !dead || twin.is_none() {
        return None;
    }
    let estate = match ServerEstate::new(stores.root(), &cfg, own) {
        Ok(e) => e,
        Err(why) => {
            tracing::warn!(target: TARGET, twin = twin.as_deref().unwrap_or(""), "{}: this box cannot run it ({why}); an admin or a guardian order can", gunnar_repair::DATA_LOST_LINE);
            return None;
        }
    };
    let auto = auto_on(&cfg);
    let report = exec::boot_check(&estate, twin.as_deref(), dead, auto, &|| false).await;
    if let gunnar_repair::Finding::DataLost { twin_addr, .. } = &report.finding {
        match hold(&stores, twin_addr) {
            Ok(()) => {
                tracing::warn!(target: TARGET, twin = %twin_addr, "held: Bootstrap.Enroll takes no first admin until this box is recharged (or <data>/{HOLD_REL} is removed)")
            }
            Err(why) => {
                tracing::error!(target: TARGET, twin = %twin_addr, "the hold could not be written ({why}); the enrolment window stays open")
            }
        }
    }
    match &report.step {
        BootStep::Nothing => {}
        BootStep::Say { line } => tracing::warn!(target: TARGET, auto_fix = auto, "{line}"),
        BootStep::Recharge { line, .. } => match &report.recharge {
            Some(Ok(r)) => {
                tracing::info!(target: TARGET, head_seq = r.head_seq, "{line}: {:?} — {}", r.phase, r.detail)
            }
            Some(Err(why)) => tracing::error!(target: TARGET, "{line}: refused — {why}"),
            None => {}
        },
    }
    Some(report)
}

/// **The words the guardian reads** off this box's report.
pub fn health(stores: &ServerStores, own: &str) -> BTreeMap<String, String> {
    let root = stores.root();
    let mut h = BTreeMap::new();
    h.insert("serves_at".to_owned(), own.to_owned());
    if crate::twin::is_twin_root(root) {
        let (busy, total) = crate::twin::own_load();
        twin_words(&mut h, crate::twin::status(root).as_ref(), busy, total);
        return h;
    }
    let restoring = crate::twin::is_restoring(root);
    let dead = data_dead(stores);
    let awaited = awaited(stores).or_else(|| held(stores));
    let data = match (dead, restoring, awaited.is_some()) {
        (false, _, _) => "ok",
        (true, true, _) | (true, false, true) => "lost",
        (true, false, false) => "fresh",
    };
    h.insert("data".to_owned(), data.to_owned());
    // ★ main rebuild (R4): the main's own log head, so a guard can tell a main AHEAD of its
    // twin (entries only its disk holds) from one the twin can give back whole.
    if !dead {
        h.insert(
            "data_head".to_owned(),
            stores.txlog().head_seq().to_string(),
        );
    }
    // A primary fenced before a recharge (standby, no roster) answers and serves reads,
    // and takes no client write: the guardian reads `fenced`, never `yes`.
    let fenced = !dead
        && crate::mirror::ReplicationConfig::load(root)
            .map(|c| c.role.refuses_client_writes())
            .unwrap_or(false);
    h.insert(
        "serving".to_owned(),
        if dead {
            "no"
        } else if fenced {
            "fenced"
        } else {
            "yes"
        }
        .to_owned(),
    );
    if restoring {
        h.insert("recharge".to_owned(), "replaying".to_owned());
    }
    h
}

/// **A twin's own review** (G12), JSON text of [`gunnar_repair::TwinReview`]: what its
/// `Mirror.LogHead` says, reviewed by the box itself on every heartbeat. `None` on a main.
pub fn twin_review(stores: &ServerStores, own: &str, now_unix_ms: i64) -> Option<String> {
    if !crate::twin::is_twin_root(stores.root()) {
        return None;
    }
    use gunnar_repair::AsTwinHead;
    let head = crate::twin::own_head(stores).twin_head();
    serde_json::to_string(&gunnar_repair::review_twin(
        own,
        Ok(&head),
        None,
        None,
        now_unix_ms,
    ))
    .ok()
}

/// **Where this box pulls, flag or feed** (ONE monetize, M-D35): `--fleet` /
/// `--fleet-instance` on the medium, else the installer's feed (`fleet`, `fleet_instance`,
/// which the front offers every box). No precedence: both and different is refused by name
/// ([`crate::feed::FEED_MISMATCH`]), as for the monetize key. A fed door is https with no CA
/// file: its certificate is checked against the public roots, or the fed `fleet_ca`; it is
/// dialled at its name, or at the fed `fleet_at`.
pub fn resolve_fleet(
    flag: Option<FleetConfig>,
    fed: Option<&crate::feed::Fed>,
    every: Duration,
) -> Result<Option<FleetConfig>, String> {
    let named = resolve_fleet_named(flag, fed, every)?;
    if !PULLS {
        if let Some(f) = &named {
            return Err(format!(
                "{NO_FLEET_PULL} (named: {} for {})",
                f.url, f.instance
            ));
        }
    }
    Ok(named)
}

/// **AIRGAP B5** (owner rule 2026-10-25): whether this build carries the pull channel to the
/// vendor's monetize. Only estate and metered builds do (`fleet-pull`, named by every
/// gunnar-server row); gunnar-solo and gunnar-service do not.
pub const PULLS: bool = cfg!(feature = "fleet-pull");

/// The refusal a named or fed pull channel gets in a build without `fleet-pull`.
pub const NO_FLEET_PULL: &str =
    "this gunnar reports to no fleet: the pull channel to monetize needs gunnar-server's `fleet-pull` feature (estate and metered builds only)";

/// [`resolve_fleet`]'s reading of the flags and the feed, whatever the build.
pub fn resolve_fleet_named(
    flag: Option<FleetConfig>,
    fed: Option<&crate::feed::Fed>,
    every: Duration,
) -> Result<Option<FleetConfig>, String> {
    let (at, ca_pem) = (
        fed.and_then(|f| f.fleet_at.clone()),
        fed.and_then(|f| f.fleet_ca.clone()),
    );
    if let Some(why) = seam_refusal(at.as_deref(), ca_pem.is_some()) {
        return Err(why);
    }
    let fed = match fed.and_then(|f| f.fleet.clone()) {
        None => return Ok(flag),
        Some(url) => {
            let instance = fed.and_then(|f| f.fleet_instance.clone()).ok_or_else(|| {
                format!("the installer's door fed `{}` = {url} but no `{}`: the estate this box reports for", crate::feed::FEED_FLEET, crate::feed::FEED_FLEET_INSTANCE)
            })?;
            (url, instance)
        }
    };
    match flag {
        None => Ok(Some(FleetConfig { url: fed.0, ca: None, ca_pem, at, instance: fed.1, role: String::new(), monetize_pubkey: None, every })),
        Some(f) if f.url == fed.0 && f.instance == fed.1 => Ok(Some(f)),
        Some(f) => Err(format!(
            "REFUSED [{}] the medium names --fleet {} --fleet-instance {} and the installer's door fed `{}` = {} `{}` = {}; \
             two answers to where this box reports, and neither wins. Remove the flags or make them agree.",
            crate::feed::FEED_MISMATCH,
            f.url,
            f.instance,
            crate::feed::FEED_FLEET,
            fed.0,
            crate::feed::FEED_FLEET_INSTANCE,
            fed.1
        )),
    }
}

pub use crate::feed::seam_refusal;

/// **The pull channel the feed names LIVE** (a first boot: the front feeds `fleet` after
/// this process started). With a flag, a different fed door is an alarm
/// ([`crate::feed::FEED_MISMATCH`]) and the flag's channel stays; with none, the first fed
/// door is returned. `None` only if the feed never names one.
pub async fn fleet_when_fed(flag: Option<FleetConfig>, every: Duration) -> Option<FleetConfig> {
    if let Some(f) = flag {
        let mine = f.clone();
        nordisk_pairing::feedbus::on_published("fleet", move |feed| {
            if let Err(why) = resolve_fleet(
                Some(mine.clone()),
                Some(&crate::feed::Fed::from_feed(feed)),
                every,
            ) {
                crate::feed::raise(crate::feed::FEED_MISMATCH, &why);
            }
        });
        return Some(f);
    }
    let (tx, rx) = tokio::sync::oneshot::channel();
    let tx = std::sync::Mutex::new(Some(tx));
    nordisk_pairing::feedbus::on_published("fleet", move |feed| {
        match resolve_fleet(None, Some(&crate::feed::Fed::from_feed(feed)), every) {
            Ok(Some(f)) => {
                if let Some(tx) = tx.lock().unwrap_or_else(|p| p.into_inner()).take() {
                    tracing::info!(target: TARGET, fleet = %f.url, instance = %f.instance, "the installer fed the pull channel; pulling from now on");
                    let _ = tx.send(f);
                }
            }
            Ok(None) => {}
            Err(why) => tracing::warn!(target: TARGET, "{why}"),
        }
    });
    rx.await.ok()
}

/// **What a twin reports** (F4 x G12, the twin-health PULL — 5.4 G1-pull #25/#26): its
/// primary's word, and its lag and load, which the ONE monetize's twin autoscale reads off
/// the report (monetize `estate_order::twin::TwinReading::of_reports`, its `twin_state` /
/// `lag_entries` / `cpu_millicores_*` words) instead of dialing the estate's door. A twin that
/// has not polled yet reports no lag (no reading).
pub fn twin_words(
    h: &mut BTreeMap<String, String>,
    status: Option<&crate::twin::TwinStatus>,
    cpu_busy: Option<u64>,
    cpu_total: Option<u64>,
) {
    let primary = match status.map(|s| s.state) {
        Some(crate::twin::State::Unreachable) => "silent",
        Some(crate::twin::State::Diverged) => "diverged",
        Some(_) => "live",
        None => "unknown",
    };
    h.insert("primary".to_owned(), primary.to_owned());
    if let Some(s) = status {
        h.insert(TWIN_STATE.to_owned(), s.state.to_string());
        h.insert(
            TWIN_LAG_ENTRIES.to_owned(),
            s.primary_head_seq.saturating_sub(s.applied_seq).to_string(),
        );
    }
    if let (Some(busy), Some(total)) = (cpu_busy, cpu_total) {
        h.insert(TWIN_CPU_BUSY.to_owned(), busy.to_string());
        h.insert(TWIN_CPU_TOTAL.to_owned(), total.to_string());
    }
}

/// The twin's health words ([`twin_words`]; monetize-server `estate_order::twin::TwinReading::of_reports`).
pub const TWIN_STATE: &str = "twin_state";
pub const TWIN_LAG_ENTRIES: &str = "lag_entries";
pub const TWIN_CPU_BUSY: &str = "cpu_millicores_busy";
pub const TWIN_CPU_TOTAL: &str = "cpu_millicores_total";

/// The role this box pulls as.
pub fn role_of(stores: &ServerStores, fleet: &FleetConfig) -> Option<String> {
    nordisk_pairing::elect::pull_role(
        &fleet.role,
        crate::twin::is_twin_root(stores.root()),
        door_role_of(stores.root()).as_ref().map(|r| r.as_deref()),
    )
}

/// The role the installer's door elected for this box ([`nordisk_pairing::elect::pull_role`]'s `door`).
fn door_role_of(root: &std::path::Path) -> Option<Option<String>> {
    #[cfg(feature = "wg-client")]
    {
        crate::pairing::door_role(root)
    }
    #[cfg(not(feature = "wg-client"))]
    {
        let _ = root;
        None
    }
}

/// The channel to monetize's `Fleet`: dialled at [`FleetConfig::at`] when one is named, the
/// URL's host otherwise; for `https://` the certificate is verified under the URL's host,
/// against the CA file, else the fed CA, else the public roots.
#[cfg(feature = "fleet-pull")]
pub fn fleet_channel(fleet: &FleetConfig) -> Result<tonic::transport::Channel, String> {
    let origin: tonic::codegen::http::Uri = fleet
        .url
        .parse()
        .map_err(|e| format!("--fleet {}: {e}", fleet.url))?;
    let name = origin.host().unwrap_or("").to_owned();
    let dial = match fleet.at.as_deref() {
        Some(at) => format!("{}://{at}", origin.scheme_str().unwrap_or("https")),
        None => fleet.url.clone(),
    };
    let endpoint = tonic::transport::Channel::from_shared(dial.clone())
        .map_err(|e| format!("--fleet {} at {dial}: {e}", fleet.url))?
        .origin(origin)
        .connect_timeout(Duration::from_secs(10))
        .timeout(Duration::from_millis(fleet::LONG_POLL_MAX_MS + 10_000));
    let endpoint = if fleet.url.starts_with("https://") {
        gunnar_grpc::ensure_crypto_provider();
        let tls = match (fleet.ca.as_ref(), fleet.ca_pem.as_deref()) {
            (Some(ca), _) => {
                let pem =
                    std::fs::read(ca).map_err(|e| format!("--fleet-ca {}: {e}", ca.display()))?;
                tonic::transport::ClientTlsConfig::new()
                    .ca_certificate(tonic::transport::Certificate::from_pem(pem))
            }
            (None, Some(pem)) => tonic::transport::ClientTlsConfig::new()
                .ca_certificate(tonic::transport::Certificate::from_pem(pem)),
            // A fed door names no CA: the public roots (monetize.ignalina.dk).
            (None, None) => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
        };
        endpoint
            .tls_config(tls.domain_name(name))
            .map_err(|e| format!("--fleet {}: {e}", fleet.url))?
    } else {
        let host = dial
            .trim_start_matches("http://")
            .split(['/', ':'])
            .next()
            .unwrap_or("");
        let local = host == "localhost"
            || host
                .trim_matches(['[', ']'])
                .parse::<std::net::IpAddr>()
                .map(|ip| ip.is_loopback())
                .unwrap_or(false);
        if !local {
            return Err(format!("--fleet {} is plain http to another machine: orders are signed, but this box's reports would not be private; use https:// with --fleet-ca", fleet.url));
        }
        endpoint
    };
    Ok(endpoint.connect_lazy())
}

/// **How long after this box starts pulling a refusal as `unauthenticated` means "not
/// known YET"** (i54a run16): a box laid for an ATTACHED estate is let in by monetize only
/// once the estate's owner has opened its pin window (`POST /attach/installed`), which may
/// land after the box's first report. Inside this grace such a refusal is said at INFO and
/// asked again soon ([`NOT_YET_KNOWN_RETRY`]); after it, it is the WARN it always was. It
/// covers a whole install (monetize's attached-install window is as long).
pub const NOT_YET_KNOWN_GRACE: Duration = Duration::from_secs(45 * 60);
/// How soon a box asks again while monetize does not know it yet ([`NOT_YET_KNOWN_GRACE`]).
pub const NOT_YET_KNOWN_RETRY: Duration = Duration::from_secs(10);

/// Pure: **after a refused report or pull**, the pause before asking again and whether it is
/// said quietly (INFO, "not known yet") — `unauthenticated` inside [`NOT_YET_KNOWN_GRACE`]
/// since the channel started: [`NOT_YET_KNOWN_RETRY`] at most, quiet. Anything else: `pause`,
/// a WARN.
pub fn after_refusal(code: &str, since_start: Duration, pause: Duration) -> (Duration, bool) {
    if code == "unauthenticated" && since_start < NOT_YET_KNOWN_GRACE {
        (pause.min(NOT_YET_KNOWN_RETRY), true)
    } else {
        (pause, false)
    }
}

/// **The pull channel**: report, pull, verify, execute, ack — until `stop` turns true.
/// `entitlements` is what a `restate_entitlements` order reaches, with or without a repair key.
///
/// The pump is monetize-estate-side's (A28 M12): its [`fleet::Puller`] signs, checks and keeps
/// what it took, its [`fleet::TonicLink`] is the wire, and gunnar's executor is its
/// [`fleet::GunnarBox`]. The puller's calls block, so each runs on a blocking thread
/// ([`on_the_pump`]); this loop keeps the role wait, the pauses and the words.
#[cfg(feature = "fleet-pull")]
pub async fn pull_loop(
    stores: Arc<ServerStores>,
    fleet: FleetConfig,
    estate: Option<ServerEstate>,
    entitlements: Entitlements,
    own: String,
    mut stop: tokio::sync::watch::Receiver<bool>,
) {
    // ★ i54a run27 (2026-10-11): the twin was fed the pull channel BEFORE the door elected its
    // role, reported as `main` under its own key, and the ONE box — which pins one key per
    // role on its first report — never knew it. A box whose role is not elected yet waits.
    let mut said = false;
    let role = loop {
        if let Some(role) = role_of(&stores, &fleet) {
            break role;
        }
        if !said {
            tracing::info!(target: TARGET, "the pull channel waits for the installer's door to elect this box's role: it reports to monetize only as what it is");
            said = true;
        }
        tokio::select! {
            _ = tokio::time::sleep(ROLE_WAIT_EVERY) => {}
            _ = stop.changed() => {}
        }
        if *stop.borrow() {
            return;
        }
    };
    let me = fleet::Addressee {
        product: fleet::PRODUCT.to_owned(),
        instance: fleet.instance.clone(),
        role,
    };
    let entitlements = entitlements.pulled_as(&me.instance, &me.role, fleet.monetize_pubkey);
    let rt = tokio::runtime::Handle::current();
    let spent: Arc<dyn Fn() -> bool + Send + Sync> = {
        let stop = stop.clone();
        Arc::new(move || *stop.borrow())
    };
    match estate {
        Some(e) => {
            let boxed = fleet::GunnarBox::new(
                ServerEstate {
                    entitlements: entitlements.clone(),
                    ..e
                },
                rt,
            )
            .stopping(spent);
            pull_rounds(stores, fleet, me, Arc::new(boxed), entitlements, own, stop).await
        }
        None => {
            let boxed = fleet::GunnarBox::new(NoEstate(own.clone(), entitlements.clone()), rt)
                .stopping(spent);
            pull_rounds(stores, fleet, me, Arc::new(boxed), entitlements, own, stop).await
        }
    }
}

/// [`pull_loop`]'s rounds over one executor.
#[cfg(feature = "fleet-pull")]
async fn pull_rounds<E: Estate + Send + Sync + 'static>(
    stores: Arc<ServerStores>,
    fleet: FleetConfig,
    me: fleet::Addressee,
    boxed: Arc<fleet::GunnarBox<E>>,
    entitlements: Entitlements,
    own: String,
    mut stop: tokio::sync::watch::Receiver<bool>,
) {
    let puller = match fleet::Puller::at(stores.root(), me.clone(), env!("CARGO_PKG_VERSION")) {
        Ok(p) => p,
        Err(e) => {
            return tracing::error!(target: TARGET, "the pull channel is off: this box's fleet key or state could not be read: {e}")
        }
    };
    let channel = match fleet_channel(&fleet) {
        Ok(c) => c,
        Err(e) => return tracing::error!(target: TARGET, "the pull channel is off: {e}"),
    };
    let public: String = puller
        .box_public()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    tracing::info!(target: TARGET, url = %fleet.url, instance = %me.instance, role = %me.role, box_key = %public, "pulling repair orders from monetize; it never calls in");
    let mut pump = Some(Pump {
        puller,
        link: fleet::TonicLink::new(channel, tokio::runtime::Handle::current()),
    });
    let started = std::time::Instant::now();
    // RA3-6: panicked pump calls in a row; a call that answers resets it.
    let mut panics = 0u32;
    // After a panic: rebuild the puller from its kept state (the ALERT and the back-off are
    // said where it panicked).
    let rebuild = |pump: &mut Option<Pump>| {
        if let Some(p) = pump.as_mut() {
            match fleet::Puller::at(stores.root(), me.clone(), env!("CARGO_PKG_VERSION")) {
                Ok(fresh) => p.puller = fresh,
                Err(e) => {
                    tracing::error!(target: TARGET, "ALERT fleet.pump-rebuild the puller could not be rebuilt after the panic ({e}); the old one pulls again")
                }
            }
        }
    };
    loop {
        let now = || crate::twin::now_ms().max(0) as u64;
        let mut pause = fleet.every;
        let product = {
            let e = entitlements.clone();
            let at = now();
            tokio::task::spawn_blocking(move || e.product(at))
                .await
                .unwrap_or(None)
        };
        let own_review = twin_review(&stores, &own, now() as i64);
        let health = health(&stores, &own);
        let at = now();
        let reported = match on_the_pump(&mut pump, move |p: &mut Pump| {
            p.puller
                .report(&mut p.link, health, product, own_review, at)
        })
        .await
        {
            None => return,
            Some(Pumped::Ran(r)) => r,
            Some(Pumped::Panicked(why)) => {
                panics += 1;
                let pause = pump_backoff(panics);
                tracing::error!(target: TARGET, "{}", pump_panicked_line("report", panics, &why, pause));
                rebuild(&mut pump);
                tokio::select! {
                    _ = tokio::time::sleep(pause) => {}
                    _ = stop.changed() => {}
                }
                if *stop.borrow() {
                    return;
                }
                continue;
            }
        };
        match reported {
            Ok(next) if next > 0 => {
                pause = Duration::from_millis(next).clamp(
                    Duration::from_secs(1),
                    fleet.every.max(Duration::from_secs(1)),
                )
            }
            Ok(_) => {}
            Err(f) => {
                let (p, quiet) = after_refusal(&f.code, started.elapsed(), pause);
                pause = p;
                if quiet {
                    tracing::info!(target: TARGET, "monetize does not know this box yet (the estate's owner opens its window as the install feeds it); asking again in {} s: {f}", pause.as_secs());
                } else {
                    tracing::warn!(target: TARGET, "the report to monetize failed: {f}");
                }
            }
        }
        let key = fleet
            .monetize_pubkey
            .or_else(crate::feed::monetize_in_force);
        let (b, at) = (Arc::clone(&boxed), now());
        let pulled = match on_the_pump(&mut pump, move |p: &mut Pump| {
            p.puller.pull(
                &mut p.link,
                b.as_ref(),
                key.as_ref(),
                fleet::LONG_POLL_MAX_MS,
                at,
            )
        })
        .await
        {
            None => return,
            Some(Pumped::Ran(r)) => r,
            Some(Pumped::Panicked(why)) => {
                panics += 1;
                let pause = pump_backoff(panics);
                tracing::error!(target: TARGET, "{}", pump_panicked_line("pull", panics, &why, pause));
                rebuild(&mut pump);
                tokio::select! {
                    _ = tokio::time::sleep(pause) => {}
                    _ = stop.changed() => {}
                }
                if *stop.borrow() {
                    return;
                }
                continue;
            }
        };
        let mut ran = false;
        match pulled {
            Ok(p) => {
                panics = 0;
                for a in &p.acks {
                    ran = true;
                    if a.ok {
                        tracing::info!(target: TARGET, seq = a.seq, "monetize's order ran: {}", a.detail);
                    } else {
                        tracing::warn!(target: TARGET, seq = a.seq, "monetize's order was refused: {}", a.detail);
                    }
                }
                for f in &p.forged {
                    tracing::error!(target: TARGET, "an order that is NOT monetize's was offered and ignored: {f}");
                }
            }
            Err(f) if f.code == "no_key" => tracing::debug!(target: TARGET, "{}", f.message),
            Err(f) => {
                let (p, quiet) = after_refusal(&f.code, started.elapsed(), pause);
                pause = pause.min(p);
                if quiet {
                    tracing::info!(target: TARGET, "pulling orders: monetize does not know this box yet; asking again in {} s: {f}", pause.as_secs());
                } else {
                    tracing::warn!(target: TARGET, "pulling orders failed: {f}");
                }
            }
        }
        if *stop.borrow() {
            return;
        }
        if ran {
            continue;
        }
        tokio::select! {
            _ = tokio::time::sleep(pause) => {}
            _ = stop.changed() => {}
        }
        if *stop.borrow() {
            return;
        }
    }
}

/// The shared puller and its wire, handed to a blocking thread for each call.
#[cfg(feature = "fleet-pull")]
struct Pump {
    puller: fleet::Puller,
    link: fleet::TonicLink,
}

/// What one pump call came to ([`on_the_pump`]).
#[derive(Debug, PartialEq, Eq)]
pub enum Pumped<T> {
    /// It answered.
    Ran(T),
    /// It panicked (RA3-6): the words of the panic. The pump is kept; the loop rebuilds its
    /// puller from what it kept on disk, says an ALERT and pulls again after [`pump_backoff`].
    Panicked(String),
}

/// **One blocking pump call** off the runtime's workers ([`fleet::TonicLink`] blocks on the
/// runtime, and so does a [`fleet::GunnarBox`] running an order). A panic in the call (one bad
/// order, or a panic inside the executor while it runs one) is caught on that thread
/// ([`Pumped::Panicked`]) and the pump handed back (RA3-6: it used to turn the pull channel
/// off until the next start, said only as an error). `None` only when the blocking thread
/// itself was lost (the runtime is stopping).
pub async fn on_the_pump<P: Send + 'static, T: Send + 'static>(
    pump: &mut Option<P>,
    call: impl FnOnce(&mut P) -> T + Send + 'static,
) -> Option<Pumped<T>> {
    let mut p = pump.take()?;
    match tokio::task::spawn_blocking(move || {
        let out = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| call(&mut p)));
        (p, out)
    })
    .await
    {
        Ok((p, out)) => {
            *pump = Some(p);
            Some(match out {
                Ok(v) => Pumped::Ran(v),
                Err(panic) => Pumped::Panicked(panic_words(panic.as_ref())),
            })
        }
        Err(e) => {
            tracing::error!(target: TARGET, "ALERT fleet.pump-lost the pull channel is off: its blocking thread was lost ({e}); the runtime is stopping");
            None
        }
    }
}

/// The words of a caught panic.
fn panic_words(panic: &(dyn std::any::Any + Send)) -> String {
    panic
        .downcast_ref::<&str>()
        .map(|s| s.to_string())
        .or_else(|| panic.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "a panic with no words".into())
}

/// The first pause after a pump call panicked ([`pump_backoff`]).
pub const PUMP_BACKOFF_FIRST: Duration = Duration::from_secs(2);
/// The longest pause after pump calls that keep panicking.
pub const PUMP_BACKOFF_MOST: Duration = Duration::from_secs(5 * 60);

/// Pure: **the pause after the `n`th panic in a row** (1-based): 2 s, doubling, at most
/// [`PUMP_BACKOFF_MOST`]. A call that answers resets the row.
pub fn pump_backoff(n: u32) -> Duration {
    PUMP_BACKOFF_FIRST
        .saturating_mul(1u32 << n.saturating_sub(1).min(16))
        .min(PUMP_BACKOFF_MOST)
}

/// The ALERT line of a panicked pump call, by name (RA3-6).
pub fn pump_panicked_line(call: &str, n: u32, why: &str, pause: Duration) -> String {
    format!("ALERT fleet.pump-panicked {call}: the pump call panicked ({why}); panic {n} in a row: the puller is rebuilt from its kept state and pulls again in {} s", pause.as_secs())
}

/// How often a box whose role is not elected yet looks again ([`nordisk_pairing::elect::pull_role`]).
pub const ROLE_WAIT_EVERY: Duration = Duration::from_secs(2);

/// A box with no repair key still pulls: every order it cannot run is acked as refused,
/// by name, so the guardian learns why instead of waiting.
#[cfg(feature = "fleet-pull")]
struct NoEstate(String, Entitlements);

#[cfg(feature = "fleet-pull")]
impl Estate for NoEstate {
    type Twin = ProtoTwin;
    type Main = ProtoMain;
    type Fence = ProtoFence;
    fn twin(&self, _: &str) -> Result<ProtoTwin, Fault> {
        Err(Fault::new(
            "unauthenticated",
            "this box holds no repair key (--repair-key, or a twin key)",
        ))
    }
    fn main(&self, _: &str) -> Result<ProtoMain, Fault> {
        Err(Fault::new(
            "unauthenticated",
            "this box holds no repair key (--repair-key, or a twin key)",
        ))
    }
    fn fence(&self, _: &str) -> Result<ProtoFence, Fault> {
        Err(Fault::new(
            "unauthenticated",
            "this box holds no repair key (--repair-key, or a twin key)",
        ))
    }
    fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
        self.1.restate(snapshot)
    }
    fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
        self.1.entitle(fact)
    }
    fn add_principal(&self, username: &str, kind: &str, signup: &str) -> Result<String, Refusal> {
        self.1.add_principal(username, kind, signup)
    }
    fn add_users(
        &self,
        tenant: &str,
        ask: &serde_json::Value,
        order: &str,
    ) -> Result<String, Refusal> {
        self.1.add_users(tenant, ask, order)
    }
    fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
        self.1.grow_room(need_bytes, reason)
    }
    fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
        self.1.purge_tenant(tenant, reason, job)
    }
    async fn sign_in_again(
        &self,
        account: &str,
        to: &str,
        host: &str,
        ttl_secs: u64,
    ) -> Result<String, Refusal> {
        self.1.sign_in_again(account, to, host, ttl_secs).await
    }
    fn keep_mail_relay(&self, relay: &gunnar_repair::order::MailRelay) -> Result<String, Refusal> {
        self.1.keep_mail_relay(relay)
    }
    fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
        self.1.keep_console_setting(name, value)
    }
    fn grow_check(
        &self,
        go_ahead: &str,
        data_gib: u64,
    ) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
        self.1.grow_check(go_ahead, data_gib)
    }
    async fn twin_holds(&self, check: &gunnar_repair::exec::GrowCheck) -> Result<String, Refusal> {
        self.1.twin_holds(check).await
    }
    fn grow_start(
        &self,
        check: &gunnar_repair::exec::GrowCheck,
        nonce: &str,
    ) -> Result<gunnar_repair::exec::GrowState, Refusal> {
        self.1.grow_start(check, nonce)
    }
    async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
        self.1.grow_settle(past_waiting).await
    }
    fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
        self.1.grow_look_again(give_up)
    }
    fn own(&self) -> String {
        self.0.clone()
    }
    fn now_unix_ms(&self) -> i64 {
        crate::twin::now_ms()
    }
}

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

    /// ★ RA3-6 RED: **a pump call that panics keeps the pump, and the next call runs** — after
    /// M12 one panic in a pump call (a bad order, or the executor panicking while it runs one)
    /// turned the pull channel off until the next start, said only as an error.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_pump_that_panics_on_its_first_order_still_takes_the_second() {
        let mut pump = Some(0u32);
        let first = on_the_pump(&mut pump, |n: &mut u32| -> u32 {
            *n += 1;
            panic!("the first order is a bad one")
        })
        .await;
        assert!(
            first.is_some(),
            "a panic is an answer, not the end of the channel"
        );
        assert!(pump.is_some(), "the pump is handed back");
        let second = on_the_pump(&mut pump, |n: &mut u32| {
            *n += 1;
            *n
        })
        .await;
        assert!(second.is_some(), "the second order is taken");
        assert_eq!(
            first,
            Some(Pumped::Panicked("the first order is a bad one".into()))
        );
        assert_eq!(second, Some(Pumped::Ran(2)));
    }

    /// ★ RA3-6: **the back-off after panics in a row** doubles from 2 s and stops at 5 min, and
    /// the ALERT names the call and the row.
    #[test]
    fn a_panicking_pump_backs_off_and_says_an_alert() {
        assert_eq!(pump_backoff(1), Duration::from_secs(2));
        assert_eq!(pump_backoff(2), Duration::from_secs(4));
        assert_eq!(pump_backoff(4), Duration::from_secs(16));
        assert_eq!(pump_backoff(40), PUMP_BACKOFF_MOST);
        let line = pump_panicked_line("pull", 3, "boom", pump_backoff(3));
        assert!(
            line.starts_with("ALERT fleet.pump-panicked pull:")
                && line.contains("boom")
                && line.contains("in 8 s"),
            "{line}"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use gunnar_grpc::{ControlError, FirstBoot};

    /// ★ Main rebuild: auto-fix is the medium's flag OR the estate's fed opt-in; anything
    /// but exactly `true` fed is off.
    #[test]
    fn auto_fix_is_the_flag_or_the_fed_opt_in() {
        let off = RepairConfig::default();
        assert!(!auto_with(&off, None));
        assert!(auto_with(
            &RepairConfig {
                auto: true,
                ..RepairConfig::default()
            },
            None
        ));
        let feed = |v: &str| {
            crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
                crate::feed::FEED_REPAIR_AUTO.to_string(),
                v.to_string(),
            )]))
        };
        assert!(auto_with(&off, Some(&feed("true"))));
        for v in ["yes", "on", "TRUE", "1", ""] {
            assert!(!auto_with(&off, Some(&feed(v))), "{v:?} is not the opt-in");
        }
    }

    /// The awaiting-main kick: only with auto-fix, one at a time, and not again within a minute.
    #[test]
    fn the_awaiting_kick_is_due_once_a_minute_and_one_at_a_time() {
        assert!(kick_due(true, false, None));
        assert!(!kick_due(false, false, None), "auto-fix off");
        assert!(!kick_due(true, true, None), "one running");
        assert!(!kick_due(true, false, Some(Duration::from_secs(5))));
        assert!(kick_due(true, false, Some(KICK_AGAIN_AFTER)));
    }

    /// The repair key: the medium's file, else the FED key, else the twin's key — and a fed
    /// value that is no key is refused by name.
    #[test]
    fn the_repair_key_is_the_flag_then_the_fed_key_then_the_twin_key() {
        let d = tempfile::tempdir().unwrap();
        let none = RepairConfig::default();
        let e = ServerEstate::with_fed(d.path(), &none, "o".into(), None)
            .err()
            .expect("no key anywhere");
        assert!(e.contains(crate::feed::FEED_REPAIR_KEY), "{e}");
        let pem = String::from_utf8(gunnar_core::mint_ed25519().unwrap()).unwrap();
        let fed = crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
            crate::feed::FEED_REPAIR_KEY.to_string(),
            pem.clone(),
        )]));
        let k = ServerEstate::with_fed(d.path(), &none, "o".into(), Some(&fed))
            .expect("the fed key")
            .key;
        assert_eq!(
            k.public_key().to_openssh().unwrap(),
            PrivateKey::from_openssh(pem.as_bytes())
                .unwrap()
                .public_key()
                .to_openssh()
                .unwrap()
        );
        let other = String::from_utf8(gunnar_core::mint_ed25519().unwrap()).unwrap();
        let path = d.path().join("flag.key");
        std::fs::write(&path, &other).unwrap();
        let k = ServerEstate::with_fed(
            d.path(),
            &RepairConfig {
                key: Some(path),
                ..RepairConfig::default()
            },
            "o".into(),
            Some(&fed),
        )
        .expect("the flag")
        .key;
        assert_eq!(
            k.public_key().to_openssh().unwrap(),
            PrivateKey::from_openssh(other.as_bytes())
                .unwrap()
                .public_key()
                .to_openssh()
                .unwrap(),
            "the medium's flag wins"
        );
        let bad = crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
            crate::feed::FEED_REPAIR_KEY.to_string(),
            "not a key".to_string(),
        )]));
        let e = ServerEstate::with_fed(d.path(), &none, "o".into(), Some(&bad))
            .err()
            .expect("refused");
        assert!(e.contains("not an OpenSSH private key"), "{e}");
    }

    fn flag(url: &str, instance: &str) -> FleetConfig {
        FleetConfig {
            url: url.into(),
            ca: Some(PathBuf::from("/etc/one-ca.pem")),
            ca_pem: None,
            at: None,
            instance: instance.into(),
            role: "main".into(),
            monetize_pubkey: None,
            every: Duration::from_secs(30),
        }
    }

    fn fed(fleet: Option<&str>, instance: Option<&str>) -> crate::feed::Fed {
        crate::feed::Fed {
            fleet: fleet.map(Into::into),
            fleet_instance: instance.map(Into::into),
            ..crate::feed::Fed::default()
        }
    }

    /// ★ **AIRGAP B5: a build without `fleet-pull` reports to no fleet.** A named channel (the
    /// flags) and a fed one are refused by name; nothing named is still no channel.
    #[cfg(not(feature = "fleet-pull"))]
    #[test]
    fn a_build_without_fleet_pull_refuses_a_named_or_fed_channel_by_name() {
        let every = Duration::from_secs(60);
        assert!(resolve_fleet(None, None, every).unwrap().is_none());
        let err = resolve_fleet(
            Some(flag("https://one.mock.invalid:8443", "acme")),
            None,
            every,
        )
        .unwrap_err();
        assert!(
            err.contains("fleet-pull") && err.contains("one.mock.invalid"),
            "{err}"
        );
        let err = resolve_fleet(
            None,
            Some(&fed(Some("https://one.mock.invalid:8443"), Some("acme"))),
            every,
        )
        .unwrap_err();
        assert!(err.contains("fleet-pull"), "{err}");
    }

    /// With `fleet-pull`, the channel the flags or the feed name, as before.
    #[cfg(feature = "fleet-pull")]
    #[test]
    fn a_fleet_pull_build_pulls_where_it_is_told_as_before() {
        let every = Duration::from_secs(60);
        let f = resolve_fleet(
            Some(flag("https://one.mock.invalid:8443", "acme")),
            None,
            every,
        )
        .unwrap()
        .unwrap();
        assert_eq!(f.url, "https://one.mock.invalid:8443");
    }

    /// ONE monetize (BROKEN: "gunnar-front must hand fleet/fleet_instance/monetize_pubkey to
    /// main and twin"): a box with no `--fleet` pulls where the installer's feed says; a flag
    /// and an equal feed agree; a flag and a different feed are refused by name.
    #[test]
    fn the_pull_channel_is_the_flags_or_the_feeds_and_a_disagreement_is_refused() {
        let every = Duration::from_secs(60);
        assert!(
            resolve_fleet_named(None, None, every).unwrap().is_none(),
            "nothing named: no pull channel"
        );
        assert!(resolve_fleet_named(None, Some(&fed(None, None)), every)
            .unwrap()
            .is_none());

        let f = resolve_fleet_named(
            None,
            Some(&fed(
                Some("https://monetize.ignalina.dk:8443"),
                Some("acme"),
            )),
            every,
        )
        .unwrap()
        .expect("the feed names it");
        assert_eq!(
            (
                f.url.as_str(),
                f.instance.as_str(),
                f.role.as_str(),
                f.ca.is_none(),
                f.monetize_pubkey.is_none(),
                f.every
            ),
            (
                "https://monetize.ignalina.dk:8443",
                "acme",
                "",
                true,
                true,
                every
            ),
            "role = what this box is; key = the one in force"
        );

        let f = resolve_fleet_named(
            Some(flag("https://one:8443", "acme")),
            Some(&fed(Some("https://one:8443"), Some("acme"))),
            every,
        )
        .unwrap()
        .unwrap();
        assert_eq!(
            (f.ca.is_some(), f.role.as_str(), f.every),
            (true, "main", Duration::from_secs(30)),
            "equal: the medium's, with its CA and role"
        );
        let f = resolve_fleet_named(Some(flag("https://one:8443", "acme")), None, every)
            .unwrap()
            .unwrap();
        assert_eq!(f.url, "https://one:8443");

        for other in [
            fed(Some("https://two:8443"), Some("acme")),
            fed(Some("https://one:8443"), Some("globex")),
        ] {
            let e =
                resolve_fleet_named(Some(flag("https://one:8443", "acme")), Some(&other), every)
                    .expect_err("two answers to where this box reports");
            assert!(e.contains(crate::feed::FEED_MISMATCH), "{e}");
        }
        let e = resolve_fleet_named(None, Some(&fed(Some("https://one:8443"), None)), every)
            .expect_err("a door without the estate");
        assert!(e.contains("fleet_instance"), "{e}");
    }

    /// The fed door names no CA file: monetize's certificate is checked against the
    /// public roots, never not at all.
    #[cfg(feature = "fleet-pull")]
    #[tokio::test]
    async fn a_fed_https_door_without_a_ca_file_is_dialled_against_the_public_roots() {
        let f = FleetConfig {
            ca: None,
            ..flag("https://monetize.ignalina.dk:8443", "acme")
        };
        fleet_channel(&f).expect("public roots");
        let plain = FleetConfig {
            ca: None,
            ..flag("http://monetize.ignalina.dk:8080", "acme")
        };
        assert!(
            fleet_channel(&plain).is_err(),
            "plain http to another machine stays refused"
        );
    }

    /// One TLS server on loopback for `name` (its own self-signed certificate): the
    /// handshake's verdict and the name the client asked for, once. Returns the address,
    /// the certificate (PEM) and the verdict.
    #[cfg(feature = "fleet-pull")]
    fn tls_door(
        name: &str,
    ) -> (
        std::net::SocketAddr,
        String,
        std::sync::mpsc::Receiver<Result<String, String>>,
    ) {
        // A self-signed leaf, as the mock's ONE box serves (private-monetize-ops `bench_certificate`).
        let cert = rcgen::generate_simple_self_signed(vec![name.to_owned()]).expect("cert");
        let pem = cert.cert.pem();
        let der = cert.cert.der().clone();
        let key = rustls::pki_types::PrivateKeyDer::try_from(cert.signing_key.serialize_der())
            .expect("key");
        let cfg = rustls::ServerConfig::builder_with_provider(Arc::new(
            rustls::crypto::ring::default_provider(),
        ))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_no_client_auth()
        .with_single_cert(vec![der], key)
        .unwrap();
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let at = listener.local_addr().unwrap();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let Ok((mut tcp, _)) = listener.accept() else {
                return;
            };
            tcp.set_read_timeout(Some(Duration::from_secs(10))).ok();
            let mut conn = rustls::ServerConnection::new(Arc::new(cfg)).unwrap();
            let said = loop {
                if !conn.is_handshaking() {
                    break Ok(conn.server_name().unwrap_or("").to_owned());
                }
                if let Err(e) = conn.complete_io(&mut tcp) {
                    break Err(e.to_string());
                }
            };
            let _ = tx.send(said);
        });
        (at, pem, rx)
    }

    /// ★ **A fed connect address and CA for the pull channel** (i54a run12: every box's
    /// `fleet` = `https://monetize-test.ignalina.dk` answered "dns error" seven times a
    /// minute — a mock guest resolves nothing, and the mock's ONE box has a bench
    /// certificate). The installer may feed `fleet_at` (the address to dial) and
    /// `fleet_ca` (the PEM the door's certificate is checked against): the box dials
    /// THAT address, and TLS still verifies the URL's own name.
    #[cfg(feature = "fleet-pull")]
    #[tokio::test]
    async fn a_fed_connect_address_and_ca_are_dialled_while_tls_verifies_the_urls_name() {
        let (at, pem, said) = tls_door("one.test");
        let feed: BTreeMap<String, String> = [
            ("fleet", "https://one.test"),
            ("fleet_instance", "acme"),
            ("offer.front.fleet_at", &at.to_string()),
            ("fleet_ca", &pem),
        ]
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
        let f = resolve_fleet_named(
            None,
            Some(&crate::feed::Fed::from_feed(&feed)),
            Duration::from_secs(60),
        )
        .unwrap()
        .expect("the feed names a door");
        assert_eq!(f.url, "https://one.test", "the URL stays the door's name");
        let mut client =
            fleet::pb::fleet_client::FleetClient::new(fleet_channel(&f).expect("a channel"));
        let _ = tokio::time::timeout(
            Duration::from_secs(10),
            client.report(fleet::pb::FleetSigned::default()),
        )
        .await;
        let said = said
            .recv_timeout(Duration::from_secs(10))
            .expect("the box dialled the fed address");
        assert_eq!(
            said,
            Ok("one.test".to_owned()),
            "the handshake completed against the fed CA, under the URL's name"
        );
    }

    /// ★ ST-19 (STINKY 2026-10-11): **the mock's seam is honoured only as the mock feeds it.**
    /// `fleet_at` is a bench forward (a private, loopback or link-local address literal); one
    /// that names a public address or a name is refused by name, and so is a fed `fleet_ca`
    /// without such a `fleet_at` — a release box never re-roots its trust on a feed's word.
    #[test]
    fn the_fed_fleet_seam_is_refused_unless_it_is_a_bench_address() {
        let every = Duration::from_secs(60);
        let feed = |pairs: &[(&str, &str)]| {
            crate::feed::Fed::from_feed(
                &pairs
                    .iter()
                    .map(|(k, v)| (k.to_string(), v.to_string()))
                    .collect(),
            )
        };
        let base = [
            ("fleet", "https://monetize.ignalina.dk"),
            ("fleet_instance", "acme"),
        ];
        for at in [
            "10.0.2.2:40443",
            "127.0.0.1:443",
            "192.168.1.9:8443",
            "172.20.0.4:443",
            "169.254.1.1:443",
            "[::1]:443",
            "[fd00::2]:443",
            "100.64.0.9:443",
        ] {
            let f = resolve_fleet_named(
                None,
                Some(&feed(&[
                    base[0],
                    base[1],
                    ("fleet_at", at),
                    ("fleet_ca", "PEM"),
                ])),
                every,
            );
            assert!(
                f.as_ref()
                    .is_ok_and(|f| f.as_ref().is_some_and(|f| f.at.as_deref() == Some(at))),
                "a bench forward {at} is dialled: {f:?}"
            );
        }
        for at in [
            "203.0.113.7:443",
            "8.8.8.8:443",
            "monetize.example:443",
            "[2001:db8::1]:443",
            "10.0.2.2",
        ] {
            let e = resolve_fleet_named(
                None,
                Some(&feed(&[base[0], base[1], ("fleet_at", at)])),
                every,
            )
            .expect_err(at);
            assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
        }
        let e = resolve_fleet_named(
            None,
            Some(&feed(&[base[0], base[1], ("fleet_ca", "PEM")])),
            every,
        )
        .expect_err("a CA with no bench address");
        assert!(e.contains("[fleet-seam-bench-only]"), "{e}");
        let plain = resolve_fleet_named(None, Some(&feed(&base)), every)
            .unwrap()
            .expect("the public door");
        assert_eq!(
            (plain.at, plain.ca_pem),
            (None, None),
            "no seam: the name, under the public roots"
        );
    }

    /// ★ **A box monetize does not know YET asks again soon, quietly** (i54a run16: main and
    /// twin of an attached estate reported before its owner opened their pin window, were
    /// refused `unauthenticated`, backed off a full `fleet.every`, and missed the window's
    /// first minutes — WARN spam and ENT0's room never landed). Inside the grace: retry within
    /// 10 s, said at INFO; after it, or for any other refusal: the pause as before, a WARN.
    #[test]
    fn a_box_not_known_yet_asks_again_soon_and_quietly_inside_the_grace() {
        let every = Duration::from_secs(60);
        assert_eq!(
            after_refusal("unauthenticated", Duration::from_secs(30), every),
            (NOT_YET_KNOWN_RETRY, true)
        );
        assert_eq!(
            after_refusal(
                "unauthenticated",
                NOT_YET_KNOWN_GRACE - Duration::from_secs(1),
                every
            ),
            (NOT_YET_KNOWN_RETRY, true)
        );
        assert_eq!(
            after_refusal(
                "unauthenticated",
                Duration::from_secs(1),
                Duration::from_secs(3)
            ),
            (Duration::from_secs(3), true),
            "never longer than it would have paused"
        );
        assert_eq!(
            after_refusal(
                "unauthenticated",
                NOT_YET_KNOWN_GRACE + Duration::from_secs(1),
                every
            ),
            (every, false),
            "past the grace it is a WARN again"
        );
        assert_eq!(
            after_refusal("unavailable", Duration::from_secs(30), every),
            (every, false),
            "a door that is down is not 'not known yet'"
        );
        assert!(
            NOT_YET_KNOWN_GRACE >= Duration::from_secs(45 * 60),
            "covers a whole install"
        );
    }

    fn key_line() -> String {
        let pem = gunnar_core::mint_ed25519().expect("mint");
        PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap())
            .unwrap()
            .public_key()
            .to_openssh()
            .unwrap()
    }

    /// ★ R-7b (report §4.1): a main whose data set is lost, with a twin able to give it
    /// back, HOLDS — its open bootstrap window takes no first admin (that would make a
    /// new, empty estate and shut the window `Bootstrap.Restore` needs) — and says what
    /// it awaits. The way out by hand: remove the marker.
    #[test]
    fn a_held_box_takes_no_first_admin_and_says_its_twin_until_the_marker_goes() {
        let d = tempfile::tempdir().unwrap();
        let stores = Arc::new(ServerStores::open(d.path()).unwrap());
        let plane = crate::CoreControlPlane::new(Arc::clone(&stores));
        assert_eq!(held(&stores), None, "nothing holds a box nobody held");

        hold(&stores, "10.13.8.3:50051").expect("hold");
        assert_eq!(held(&stores).as_deref(), Some("10.13.8.3:50051"));
        assert_eq!(
            plane.bootstrap_state().unwrap().recharge_from,
            "10.13.8.3:50051",
            "Bootstrap.Status says what it awaits"
        );
        assert_eq!(
            health(&stores, "x").get("data").map(String::as_str),
            Some("lost")
        );

        let err = plane
            .enroll_first_identity("alice", &key_line())
            .expect_err("a held box takes no first admin");
        assert!(
            matches!(err, ControlError::FailedPrecondition(_)),
            "{err:?}"
        );
        let said = err.to_string();
        for words in [gunnar_repair::DATA_LOST_LINE, "10.13.8.3:50051", HOLD_REL] {
            assert!(said.contains(words), "`{words}` missing from: {said}");
        }
        assert!(
            stores.auth().bootstrap_window_open().unwrap(),
            "the window stays open for the recharge"
        );

        std::fs::remove_file(d.path().join(HOLD_REL)).expect("the admin removes the marker");
        assert_eq!(held(&stores), None);
        assert_eq!(plane.bootstrap_state().unwrap().recharge_from, "");
        plane
            .enroll_first_identity("alice", &key_line())
            .expect("the box enrols once the marker is gone");
    }

    /// G12 (o-g11-split): monetize's twin autoscale reads the twin's lag off the newest
    /// report's `twin_review` and its load off the twin box's health words. A twin says both
    /// itself, on every heartbeat; a main says neither of its own.
    #[test]
    fn a_twin_reviews_itself_with_its_lag_and_says_its_cores() {
        let d = tempfile::tempdir().unwrap();
        std::fs::write(d.path().join(crate::twin::TWIN_CONFIG), b"{}").unwrap();
        let stores = ServerStores::open(d.path()).unwrap();
        let idle: gunnar_repair::TwinReview = serde_json::from_str(
            &twin_review(&stores, "10.0.0.2:50051", 1_000)
                .expect("a twin reviews itself before its first poll"),
        )
        .unwrap();
        assert_eq!(idle.state, "idle", "nothing polled yet: said, not guessed");

        let status = crate::twin::TwinStatus {
            applied_seq: 5,
            primary_head_seq: 12,
            state: crate::twin::State::Following,
            why: String::new(),
            last_answer_unix_ms: 1_500,
            last_level_unix_ms: 0,
        };
        crate::twin::set_status(stores.root(), status);
        let r: gunnar_repair::TwinReview =
            serde_json::from_str(&twin_review(&stores, "10.0.0.2:50051", 2_000).expect("review"))
                .unwrap();
        assert_eq!(
            (r.applied_seq, r.primary_head_seq, r.state.as_str()),
            (5, 12, "following"),
            "{r:?}"
        );
        assert_eq!(r.twin_addr, "10.0.0.2:50051");
        assert_eq!(r.reviewed_unix_ms, 2_000);
        assert_eq!(
            r.would_lose.unreplicated, 7,
            "the lag, as monetize reads it"
        );

        let h = health(&stores, "10.0.0.2:50051");
        let word = |k: &str| {
            h.get(k)
                .unwrap_or_else(|| panic!("`{k}` missing from {h:?}"))
                .parse::<u64>()
                .unwrap_or_else(|e| panic!("`{k}` is not a u64: {e}"))
        };
        assert!(word("cpu_millicores_total") >= 1_000, "{h:?}");
        let _ = word("cpu_millicores_busy");

        let m = tempfile::tempdir().unwrap();
        let main = ServerStores::open(m.path()).unwrap();
        assert_eq!(twin_review(&main, "10.0.0.1:50051", 2_000), None);
        assert!(!health(&main, "10.0.0.1:50051").contains_key("cpu_millicores_busy"));
    }

    /// The hold lifts by itself once the box holds a data set again (a recharge
    /// finished): the marker goes, nothing is held.
    #[test]
    fn the_hold_lifts_once_the_box_holds_data_again() {
        let d = tempfile::tempdir().unwrap();
        let stores = ServerStores::open(d.path()).unwrap();
        hold(&stores, "t:1").expect("hold");
        assert!(held(&stores).is_some());
        stores
            .catalog()
            .create_namespace("alice")
            .expect("namespace");
        stores
            .create_repo(
                gunnar_core::NewRepo::new(gunnar_core::RepoRef::new("alice", "one").unwrap()),
                false,
            )
            .expect("repo");
        assert!(!data_dead(&stores));
        assert_eq!(held(&stores), None, "a box with data is held by nothing");
        assert!(!d.path().join(HOLD_REL).exists(), "and the marker is gone");
    }

    /// G16-d glue: a box whose pull channel has no grow machine refuses a paid step by
    /// name; a main with no twin address does not grow before a twin it cannot see.
    #[tokio::test]
    async fn a_paid_step_needs_this_boxs_grow_machine_and_a_main_needs_its_twins_address() {
        let r = Entitlements::default().grow_check("{}", 120).unwrap_err();
        assert!(
            r.code == RefusalCode::BadOrder && r.message.contains("no grow machine"),
            "{r}"
        );

        let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
            crate::grow::world::NoEngine,
        )));
        let e = Entitlements::default()
            .with_grow(paid)
            .with_twin(Some("  ".into()));
        let c = gunnar_repair::exec::GrowCheck {
            twin: false,
            data_gib: 120,
            reference: "estate/acme/change-2".into(),
            carried: false,
        };
        let r = e.twin_holds(&c).await.unwrap_err();
        assert!(
            r.code == RefusalCode::NoTwin
                && r.message.contains("--repair-twin")
                && r.message.contains("nothing moved"),
            "{r}"
        );
        assert!(
            r.message.contains("peer list"),
            "the refusal names the door's peer list too: {r}"
        );
    }

    /// ★ **ASK-twin** (resilience-gunnar 21:10): a sold main fed neither `--repair-twin` nor
    /// `recharge_from` asks the twin the door named it — not `no_twin`.
    #[cfg(feature = "wg-client")]
    #[tokio::test]
    async fn a_main_asks_the_twin_the_door_named_when_nothing_else_names_one() {
        use nordisk_pairing::join::{Assigned, JoinReply, Peer};
        use nordisk_pairing::Role;
        let d = tempfile::tempdir().unwrap();
        let stores = Arc::new(ServerStores::open(d.path()).unwrap());
        let lo = std::net::Ipv4Addr::LOCALHOST;
        let twin = Peer {
            name: "twin".into(),
            role: Role::Twin,
            tunnel_pubkey: String::new(),
            overlay: lo,
            endpoint: None,
            utility: Some(lo),
        };
        let reply = JoinReply {
            estate: "gunnar.rs".into(),
            me: Assigned {
                name: "main".into(),
                overlay: nordisk_pairing::OVERLAY_BOX_A,
            },
            cert_pem: None,
            role: Some(Role::Primary),
            primary_overlay: None,
            primary_utility: None,
            peers: vec![twin],
            feed: BTreeMap::new(),
        };
        crate::fed_at_rest::write(
            d.path(),
            crate::pairing::JOINED_REL,
            &reply.render().unwrap(),
        )
        .unwrap();

        let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
            crate::grow::world::NoEngine,
        )));
        let e = Entitlements::of(stores).with_grow(paid).with_twin(None);
        let c = gunnar_repair::exec::GrowCheck {
            twin: false,
            data_gib: 120,
            reference: "estate/acme/change-2".into(),
            carried: false,
        };
        let said = match e.twin_holds(&c).await {
            Ok(said) => said,
            Err(r) => {
                assert_ne!(r.code, RefusalCode::NoTwin, "the door named a twin: {r}");
                r.message
            }
        };
        assert!(
            said.contains("127.0.0.1:50051"),
            "the twin the door named is the one asked: {said}"
        );
    }

    /// The pull channel's estate, role and key reach the step's verification: a twin
    /// checks as the twin (its disk holds main's whole set).
    #[cfg(feature = "metered")]
    #[test]
    fn the_pull_channels_estate_role_and_key_reach_the_paid_steps_check() {
        use crate::paid_growth::tests::signed::{shop_pub, step, SHOP};
        let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
            crate::grow::world::NoEngine,
        )));
        let go = step(SHOP, "acme", 120, crate::twin::now_ms() + 60_000);
        let e = Entitlements::default().with_grow(paid.clone()).pulled_as(
            "acme",
            "twin",
            Some(shop_pub()),
        );
        let c = e.grow_check(&go, 480).expect("the twin's own size");
        assert!(c.twin && !c.carried && c.data_gib == 480, "{c:?}");
        let r = Entitlements::default()
            .with_grow(paid)
            .pulled_as("globex", "main", Some(shop_pub()))
            .grow_check(&go, 120)
            .unwrap_err();
        assert!(
            r.code == RefusalCode::BadOrder && r.message.contains("globex"),
            "{r}"
        );
    }
}

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

    /// ★ F4 x G12: **a twin REPORTS its lag and load** — what the one monetize used to dial the
    /// front's `/twin/health` door for (a dial-in, gone under the 5.4 pull rule).
    #[test]
    fn a_twin_reports_its_lag_and_load_for_the_pull() {
        let s = crate::twin::TwinStatus {
            applied_seq: 60,
            primary_head_seq: 100,
            state: crate::twin::State::Following,
            why: String::new(),
            last_answer_unix_ms: 1,
            last_level_unix_ms: 0,
        };
        let mut h = BTreeMap::new();
        twin_words(&mut h, Some(&s), Some(420), Some(2000));
        assert_eq!(h["primary"], "live");
        assert_eq!(h[TWIN_STATE], "following");
        assert_eq!(h[TWIN_LAG_ENTRIES], "40");
        assert_eq!(
            (h[TWIN_CPU_BUSY].as_str(), h[TWIN_CPU_TOTAL].as_str()),
            ("420", "2000")
        );
        let mut none = BTreeMap::new();
        twin_words(&mut none, None, None, Some(2000));
        assert_eq!(none["primary"], "unknown");
        assert!(
            !none.contains_key(TWIN_LAG_ENTRIES) && !none.contains_key(TWIN_CPU_BUSY),
            "no poll yet, no reading"
        );
    }
}