holger-fleet 2.0.0

//! **The order executor**: one verified order → one of holger's own verbs → one ack.
//!
//! The host hands a [`HolgerBox`] (holger-server-lib's `fleet_box`, a test's fake). Every
//! verb defaults to a refusal BY NAME, so a box that cannot do one says why instead of
//! the guardian waiting. The money is checked HERE as well as in the box: an entitlement,
//! a snapshot and a growth step are verified under the monetize key before the box is
//! asked to move anything.

use std::collections::BTreeMap;

use monetize_embed::fleet::{GrowGoAhead, OrderAck};
use monetize_product::pull::{AddPrincipal, GrowRoom, PurgeAnswer, PurgeTenant};
use monetize_product::EntitlementFact;

use crate::order::Order;

pub use crate::order::Recharge as RechargeAsk;

/// 512-byte sectors in a GiB.
pub const SECTORS_PER_GIB: u64 = 2_097_152;

/// What an entitlement did on the box.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Applied {
    /// Stored: the words say what.
    New(String),
    /// The box already holds this fact or a newer one: nothing moved.
    Held(String),
}

/// Where the grow machine stands, as a pulled growth reads it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GrowState {
    Running(String),
    /// `/data` is off; the machine waits for the members to come back larger.
    WaitingMembers(String),
    Done(String),
    Failed(String),
}

/// **`Fleet.Ask{grow_set}`**: monetize sizes THIS box's own data disks.
pub trait GrowSet {
    fn grow_set(&mut self, gib: u64) -> Result<String, String>;
}

/// A caller with no channel to ask on.
pub struct NoAsk;
impl GrowSet for NoAsk {
    fn grow_set(&mut self, _: u64) -> Result<String, String> {
        Err("this box has no pull channel to ask monetize for its disks".into())
    }
}

/// **Holger's verbs, as an order reaches them.** Sync: the executor runs on a jera job's
/// blocking thread.
pub trait HolgerBox {
    /// This box serves as the primary (not a twin, not awaiting its recharge).
    fn serving(&self) -> bool;
    /// The box's health words for `Fleet.Report` (`serving`, `data`, `serves_at`, `twin`, `epoch`).
    fn health(&self) -> BTreeMap<String, String> {
        BTreeMap::new()
    }
    /// `monetize_product::pull::ProductReport` as JSON — the main that serves sends it.
    fn product_report(&self) -> Option<String> {
        None
    }
    /// A twin reviews itself on every report (G12): its review as JSON.
    fn own_review(&self) -> Option<String> {
        None
    }
    /// Store one VERIFIED fact, newer wins.
    fn entitle(&self, fact: &EntitlementFact) -> Result<Applied, String> {
        Err(format!("this box holds no entitlements, so the fact for {} is not applied", fact.tenant.0))
    }
    /// Purge one repository (or answer its purge's state).
    fn purge(&self, p: &PurgeTenant) -> Result<PurgeAnswer, String> {
        Err(format!("this box purges no repositories, so {} is not purged", p.tenant))
    }
    /// **A sold estate's resend sign-in link** (Gunnar's S10): mint a fresh first-login invite
    /// for the order's account and mail it to the order's address. Default: this box mints none.
    fn sign_in_again(&self, s: &crate::order::SignInAgain) -> Result<String, String> {
        let _ = (&s.to, &s.host, s.ttl_secs);
        Err(format!("this box mints no sign-in links, so it cannot run sign_in_again for {}", s.account))
    }
    /// **Keep the estate keys fed again** (monetize MAIL-RELAY + NIGHT-PUBREPO 2026-10-28): kept
    /// beside the feed the box was laid with, which they go over. Default: this box keeps none.
    fn keep_refeed(&self, r: &crate::order::Refeed) -> Result<String, String> {
        Err(format!("this box keeps no estate keys fed again, so {} is not kept", r.keys.keys().cloned().collect::<Vec<_>>().join(", ")))
    }
    /// A paid signup's repository (`a.username`), bound to `a.signup` (idempotent); answers the
    /// namespace made. A name bound to another signup is refused starting `taken`.
    fn add_principal(&self, a: &AddPrincipal) -> Result<String, String> {
        Err(format!("this box makes no repositories for a signup, so {} is not made", a.username))
    }
    /// Keep the room wanted past the set and tell the operator.
    fn want_room(&self, g: &GrowRoom) -> Result<String, String> {
        Err(format!("this box grows no data set, so {} bytes wanted cannot be kept", g.need_bytes))
    }
    /// Do the box's data disks already carry `data_gib` a disk?
    fn grow_carried(&self, data_gib: u64) -> Result<bool, String> {
        Err(format!("this box grows no data set, so it cannot grow to {data_gib} GiB"))
    }
    /// The twin first (a main only): Ok once the twin holds main's whole set.
    fn twin_holds(&self, data_gib: u64) -> Result<String, String> {
        Err(format!("this box cannot see whether its twin holds {data_gib} GiB a disk"))
    }
    /// Start the growth to `data_gib` a disk under a one-time `nonce` (the step was verified).
    fn grow_start(&self, data_gib: u64, nonce: &str) -> Result<GrowState, String> {
        let _ = (data_gib, nonce);
        Err("this box grows no data set".into())
    }
    /// Wait for the grow machine to stop moving (past `waiting-members` when told).
    fn grow_settle(&self, past_waiting: bool) -> GrowState {
        let _ = past_waiting;
        GrowState::Failed("this box grows no data set".into())
    }
    /// Look at the members again (`give_up`: mount back what is there).
    fn grow_look_again(&self, give_up: bool) -> Result<(), String> {
        let _ = give_up;
        Err("this box grows no data set".into())
    }
    /// Fill `r.main` from `r.twin` (holger's recharge lane).
    fn recharge(&self, r: &RechargeAsk) -> Result<String, String> {
        Err(format!("this box runs no recharge lane, so {} is not filled from {}", r.main, r.twin))
    }
    /// Fence the old primary at `primary` (this box itself when it names it): it refuses
    /// every write from now on. Refused by name by default.
    fn fence(&self, primary: &str, epoch: u64) -> Result<String, String> {
        let _ = epoch;
        Err(format!("this box cannot fence the primary at {primary}: the guardian fences it at the cloud (the recharge's `fenced` proof)"))
    }
    /// Certify the twin at `twin` for this primary.
    fn adopt(&self, twin: &str, principal: &str) -> Result<String, String> {
        let _ = principal;
        Err(format!("this box cannot certify the twin at {twin}: the front's certification does"))
    }
    /// Take the twin role: follow `primary` (a FRESH box only, exactly once).
    fn become_twin(&self, primary: &str) -> Result<String, String> {
        Err(format!("this box cannot become the twin of {primary}"))
    }
    /// Review the twin (on a twin: itself), as JSON.
    fn review(&self) -> Result<serde_json::Value, String> {
        Err("this box cannot review a twin".into())
    }
}

/// What the executor knows about the order's context: the shared pump's own (P4a).
pub use monetize_estate_side::pump::Ctx;

fn said(seq: u64, r: Result<String, String>) -> OrderAck {
    match r {
        Ok(detail) => OrderAck { seq, ok: true, detail, receipt_json: String::new() },
        Err(detail) => OrderAck { seq, ok: false, detail, receipt_json: String::new() },
    }
}

/// **Run one verified order** and say what came of it — the ack the next Report carries.
/// Never panics, never retries: the guardian reads the ack.
pub fn execute<B: HolgerBox + ?Sized>(b: &B, ctx: &Ctx, seq: u64, order: &Order, ask: &mut dyn GrowSet) -> OrderAck {
    let not_serving = || Err::<String, String>("this box does not serve as the primary (a twin, or awaiting its recharge): nothing applied".into());
    match order {
        Order::Entitlement(fact) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            if let Err(e) = monetize_embed::signing::verify_fact(fact, ctx.key) {
                return said(seq, Err(format!("the fact for {} does not verify under this box's monetize key ({e}): nothing moved", fact.tenant.0)));
            }
            said(seq, b.entitle(fact).map(|a| match a {
                Applied::New(w) | Applied::Held(w) => w,
            }))
        }
        Order::RestateEntitlements(snap) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            if let Err(e) = monetize_embed::signing::verify_snapshot(snap, ctx.key) {
                return said(seq, Err(format!("the snapshot does not verify under this box's monetize key ({e}): nothing restated")));
            }
            let (mut new, mut held) = (0usize, 0usize);
            for fact in &snap.facts {
                match b.entitle(fact) {
                    Ok(Applied::New(_)) => new += 1,
                    Ok(Applied::Held(_)) => held += 1,
                    Err(e) => return said(seq, Err(format!("restated {new} before {} was refused: {e}", fact.tenant.0))),
                }
            }
            said(seq, Ok(format!("restated {new} entitlement fact(s) ({held} already held as new or newer); nothing deleted")))
        }
        Order::PurgeTenant(p) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            match b.purge(p) {
                Ok(answer) => OrderAck { seq, ok: true, detail: serde_json::to_string(&answer).unwrap_or_default(), receipt_json: String::new() },
                Err(e) => said(seq, Err(e)),
            }
        }
        Order::GrowRoom(g) => said(seq, b.want_room(g)),
        Order::AddPrincipal(a) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            said(seq, b.add_principal(a))
        }
        Order::SignInAgain(s) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            said(seq, b.sign_in_again(s))
        }
        // Any box the order is for keeps it (the pump checked the role): a twin that serves later
        // mails through it too.
        Order::Refeed(r) => said(seq, b.keep_refeed(r)),
        Order::GrowGoAhead(g) => grow(b, ctx, seq, g, ask),
        Order::RechargeFromTwin(r) => {
            // Fence before recharge (Gunnar's order): the old primary the order names is
            // fenced first, and only a fence that held lets the recharge run beside it. A
            // guardian that fenced it AT THE CLOUD sends its proof (`fenced`) instead. With
            // neither, the box's recharge lane refuses `not_fenced` (A7 row 4: the shared
            // `nordisk_twin::fence::fence_gate`), so a bare order never makes two writers.
            let fenced = match r.fence.as_deref().map(str::trim) {
                Some(old) if old == r.main.trim() => {
                    return said(seq, Err(format!("order {seq} names the main {} as the primary to fence: a recharge never fences the box it fills", r.main)));
                }
                Some(old) => match b.fence(old, r.epoch) {
                    Ok(_) => Some(old.to_string()),
                    Err(e) => return said(seq, Err(format!("order {seq} fences {old} first and that fence did not hold ({e}): nothing was recharged"))),
                },
                None => None,
            };
            let prefix = match (&fenced, r.fenced.as_deref()) {
                (Some(f), _) => format!("fenced {f}; "),
                (None, Some(p)) => format!("cloud-fenced ({p}); "),
                (None, None) => String::new(),
            };
            said(seq, b.recharge(r).map(|w| format!("{prefix}{w}")).map_err(|e| format!("{prefix}{e}")))
        }
        Order::FencePrimary { primary, epoch } => said(seq, b.fence(primary, *epoch)),
        Order::AdoptTwin { twin, principal } => said(seq, b.adopt(twin, principal)),
        Order::BecomeTwin { primary } => said(seq, b.become_twin(primary)),
        Order::ReviewTwin => match b.review() {
            Ok(v) => OrderAck { seq, ok: true, detail: v.get("verdict").and_then(|w| w.as_str()).unwrap_or("reviewed").to_string(), receipt_json: v.to_string() },
            Err(e) => said(seq, Err(e)),
        },
    }
}

/// **A pulled `grow_go_ahead`** (G16-d): verify monetize's step for THIS estate and role,
/// the twin first, drain and unmount, ask monetize to size this box's own disks, look
/// again, remount. A growth already at size is acked ok with nothing moved.
fn grow<B: HolgerBox + ?Sized>(b: &B, ctx: &Ctx, seq: u64, g: &GrowGoAhead, ask: &mut dyn GrowSet) -> OrderAck {
    let fail = |w: String| said(seq, Err(w));
    let step = match monetize_embed::signing::verify_estate_growth(g.go_ahead.as_bytes(), ctx.key) {
        Ok(s) => s,
        Err(e) => return fail(format!("the paid step does not verify under this box's monetize key ({e}): nothing moved")),
    };
    if step.estate != ctx.me.instance {
        return fail(format!("the paid step is estate {}'s; this box pulls as {}: nothing moved", step.estate, ctx.me.instance));
    }
    if step.until_unix_ms <= ctx.now_unix_ms as i64 {
        return fail(format!("the paid step was good until {} (unix ms): its period is over, nothing moved", step.until_unix_ms));
    }
    let twin = ctx.me.role == "twin";
    let main_gib = step.main_data_gib_per_member;
    if !twin && g.data_gib != main_gib {
        return fail(format!("the order sizes main's disks to {} GiB, the paid step to {main_gib} GiB: nothing moved", g.data_gib));
    }
    if twin && g.data_gib < main_gib {
        return fail(format!("the order sizes the twin's disk to {} GiB, under main's {main_gib} GiB a disk: nothing moved", g.data_gib));
    }
    match b.grow_carried(g.data_gib) {
        Ok(true) => return said(seq, Ok(format!("already {} GiB a data disk: nothing moved", g.data_gib))),
        Ok(false) => {}
        Err(e) => return fail(e),
    }
    if !twin {
        if let Err(e) = b.twin_holds(g.data_gib) {
            return fail(e);
        }
    }
    match b.grow_start(g.data_gib, &format!("pull-{seq}-{}", step.reference)) {
        Ok(GrowState::Done(d)) => return said(seq, Ok(d)),
        Ok(GrowState::Failed(f)) => return fail(f),
        Ok(_) => {}
        Err(e) => return fail(e),
    }
    match b.grow_settle(false) {
        GrowState::WaitingMembers(_) => {}
        GrowState::Done(d) => return said(seq, Ok(d)),
        GrowState::Failed(f) | GrowState::Running(f) => return fail(f),
    }
    let asked = ask.grow_set(g.data_gib);
    if let Err(e) = b.grow_look_again(asked.is_err()) {
        return fail(format!("the grow machine would not look again: {e}"));
    }
    let end = b.grow_settle(true);
    match (asked, end) {
        (Ok(words), GrowState::Done(d)) => said(seq, Ok(format!("{d} (monetize: {words})"))),
        (Ok(words), GrowState::Failed(f) | GrowState::Running(f) | GrowState::WaitingMembers(f)) => fail(format!("{f} (monetize: {words})")),
        (Err(e), GrowState::Done(s) | GrowState::Failed(s) | GrowState::Running(s) | GrowState::WaitingMembers(s)) => {
            fail(format!("monetize did not size this box's disks ({e}); /data: {s}"))
        }
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::order::tests::{me, monetize_key, MONETIZE_SEED};
    use ed25519_dalek::Signer as _;
    use monetize_product::{State, TenantId};
    use std::cell::RefCell;

    /// A Holger box in memory: what it holds and every verb it was asked.
    #[derive(Default)]
    pub struct Fake {
        pub twin: bool,
        pub facts: RefCell<BTreeMap<String, EntitlementFact>>,
        pub did: RefCell<Vec<String>>,
        pub carried: bool,
        pub settle: RefCell<Vec<GrowState>>,
    }

    impl HolgerBox for Fake {
        fn serving(&self) -> bool {
            !self.twin
        }
        fn entitle(&self, fact: &EntitlementFact) -> Result<Applied, String> {
            let mut f = self.facts.borrow_mut();
            if f.get(&fact.tenant.0).is_some_and(|h| h.issued_unix_ms >= fact.issued_unix_ms) {
                return Ok(Applied::Held(format!("{}: held", fact.tenant.0)));
            }
            f.insert(fact.tenant.0.clone(), fact.clone());
            Ok(Applied::New(format!("{} is {:?}", fact.tenant.0, fact.state)))
        }
        fn purge(&self, p: &PurgeTenant) -> Result<PurgeAnswer, String> {
            self.did.borrow_mut().push(format!("purge {}", p.tenant));
            Ok(PurgeAnswer { job: format!("purge:{}", p.tenant), detail: "3 artifacts".into(), ..Default::default() })
        }
        fn want_room(&self, g: &GrowRoom) -> Result<String, String> {
            self.did.borrow_mut().push(format!("room {}", g.need_bytes));
            Ok("kept".into())
        }
        fn sign_in_again(&self, s: &crate::order::SignInAgain) -> Result<String, String> {
            self.did.borrow_mut().push(format!("sign-in {} {} {} {}", s.account, s.to, s.host, s.ttl_secs));
            Ok(format!("a new sign-in link for {} was mailed to {}", s.account, s.to))
        }
        fn keep_refeed(&self, r: &crate::order::Refeed) -> Result<String, String> {
            self.did.borrow_mut().push(format!("refeed {}", r.keys.keys().cloned().collect::<Vec<_>>().join(",")));
            Ok(format!("kept {}", r.keys.keys().cloned().collect::<Vec<_>>().join(", ")))
        }
        fn add_principal(&self, a: &AddPrincipal) -> Result<String, String> {
            self.did.borrow_mut().push(format!("signup {} {}", a.username, a.signup));
            if a.username == "theirs" { Err("taken: theirs is another signup's".into()) } else { Ok(a.username.clone()) }
        }
        fn grow_carried(&self, _: u64) -> Result<bool, String> {
            Ok(self.carried)
        }
        fn twin_holds(&self, gib: u64) -> Result<String, String> {
            self.did.borrow_mut().push(format!("twin_holds {gib}"));
            Ok("the twin holds it".into())
        }
        fn grow_start(&self, gib: u64, nonce: &str) -> Result<GrowState, String> {
            self.did.borrow_mut().push(format!("start {gib} {nonce}"));
            Ok(GrowState::Running("draining".into()))
        }
        fn grow_settle(&self, past: bool) -> GrowState {
            self.did.borrow_mut().push(format!("settle {past}"));
            let mut s = self.settle.borrow_mut();
            if s.is_empty() {
                if past { GrowState::Done("grown".into()) } else { GrowState::WaitingMembers("waiting-members".into()) }
            } else {
                s.remove(0)
            }
        }
        fn grow_look_again(&self, give_up: bool) -> Result<(), String> {
            self.did.borrow_mut().push(format!("look_again {give_up}"));
            Ok(())
        }
        fn recharge(&self, r: &RechargeAsk) -> Result<String, String> {
            self.did.borrow_mut().push(format!("recharge {} from {} lost={}", r.main, r.twin, r.accept_lost));
            Ok("recharged".into())
        }
        fn fence(&self, primary: &str, epoch: u64) -> Result<String, String> {
            self.did.borrow_mut().push(format!("fence {primary} {epoch}"));
            if primary.starts_with("stuck") { Err(format!("{primary} answers nothing")) } else { Ok(format!("{primary} is fenced at epoch {epoch}")) }
        }
        fn adopt(&self, twin: &str, principal: &str) -> Result<String, String> {
            self.did.borrow_mut().push(format!("adopt {twin} {principal}"));
            Ok("follows".into())
        }
        fn review(&self) -> Result<serde_json::Value, String> {
            Ok(serde_json::json!({"verdict": "the twin is whole", "complete": true}))
        }
        fn become_twin(&self, primary: &str) -> Result<String, String> {
            self.did.borrow_mut().push(format!("become_twin {primary}"));
            Ok("elected".into())
        }
    }

    pub fn fact(repo: &str, issued: u64, signer: [u8; 32]) -> EntitlementFact {
        let mut f = EntitlementFact {
            tenant: TenantId(repo.into()),
            plan: "o-1".into(),
            state: State::Paid,
            paid_until_unix_ms: Some(99_000_000),
            caps: Default::default(),
            source: "payment:stripe:o-1".into(),
            signature: Vec::new(),
            issued_unix_ms: (issued != 0).then_some(issued),
            issued_signature: Vec::new(),
        };
        let key = ed25519_dalek::SigningKey::from_bytes(&signer);
        f.signature = key.sign(&monetize_embed::signing::fact_message(&f)).to_bytes().to_vec();
        if issued != 0 {
            f.issued_signature = key.sign(&monetize_embed::signing::fact_message_issued(&f)).to_bytes().to_vec();
        }
        f
    }

    fn run(b: &dyn HolgerBox, role: &str, o: Order) -> OrderAck {
        let (m, k) = (me(role), monetize_key());
        execute(b, &Ctx { me: &m, key: &k, now_unix_ms: 10_000 }, 5, &o, &mut NoAsk)
    }

    struct Asked(Vec<u64>, bool);
    impl GrowSet for Asked {
        fn grow_set(&mut self, gib: u64) -> Result<String, String> {
            self.0.push(gib);
            if self.1 { Ok("resized".into()) } else { Err("no answer".into()) }
        }
    }

    fn growth(estate: &str, gib: u64, until: i64) -> String {
        let mut g = monetize_embed::signing::EstateGrowth { v: 1, signer: "shop".into(), estate: estate.into(), main_data_gib_per_member: gib, until_unix_ms: until, reference: "o-7".into(), issued_unix_ms: 1, signature: String::new() };
        let sig = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&monetize_embed::signing::estate_growth_message(&g));
        g.signature = monetize_embed::signing::base64_encode(&sig.to_bytes());
        serde_json::to_string(&g).unwrap()
    }

    #[test]
    fn entitlement_is_applied_newer_wins_and_a_forgery_moves_nothing() {
        let b = Fake::default();
        let a = run(&b, "main", Order::Entitlement(fact("libs", 5, MONETIZE_SEED)));
        assert!(a.ok, "{a:?}");
        assert!(b.facts.borrow().contains_key("libs"));
        let again = run(&b, "main", Order::Entitlement(fact("libs", 4, MONETIZE_SEED)));
        assert!(again.ok && again.detail.contains("held"), "an older fact is held, not refused: {again:?}");
        let forged = run(&b, "main", Order::Entitlement(fact("other", 6, [9u8; 32])));
        assert!(!forged.ok && forged.detail.contains("does not verify"), "{forged:?}");
        assert!(!b.facts.borrow().contains_key("other"));
        let twin = Fake { twin: true, ..Default::default() };
        let refused = run(&twin, "twin", Order::Entitlement(fact("libs", 5, MONETIZE_SEED)));
        assert!(!refused.ok && refused.detail.contains("does not serve"), "{refused:?}");
    }

    /// ★ **A resend sign-in link reaches the box that serves** (A16, Gunnar's S10): the order's
    /// words go to the host as they are; a twin mints nothing; a box that cannot mint says so.
    #[test]
    fn a_sign_in_again_reaches_the_serving_box_and_a_twin_mints_nothing() {
        let s = crate::order::SignInAgain { account: "acme".into(), to: "buyer@mock.invalid".into(), host: "acme.holger.mock.invalid".into(), ttl_secs: 86_400 };
        let b = Fake::default();
        let a = run(&b, "main", Order::SignInAgain(s.clone()));
        assert!(a.ok && a.detail.contains("mailed to buyer@mock.invalid"), "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["sign-in acme buyer@mock.invalid acme.holger.mock.invalid 86400"]);
        let twin = Fake { twin: true, ..Default::default() };
        let refused = run(&twin, "twin", Order::SignInAgain(s.clone()));
        assert!(!refused.ok && refused.detail.contains("does not serve") && twin.did.borrow().is_empty(), "{refused:?}");
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                true
            }
        }
        let none = run(&Bare, "main", Order::SignInAgain(s));
        assert!(!none.ok && none.detail.contains("mints no sign-in link"), "{none:?}");
    }

    /// ★ **MAIL-RELAY: a refeed reaches the box's keep**; a box that keeps none says so.
    #[test]
    fn a_refeed_reaches_the_boxes_keep_and_a_box_that_keeps_none_refuses() {
        let r = crate::order::Refeed { keys: [("secret.smtp_url", "smtps://u:tok@relay.mock.invalid:465"), ("secret.smtp_from", "holger@mock.invalid"), ("public_repositories", "crates")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect() };
        let b = Fake::default();
        let a = run(&b, "main", Order::Refeed(r.clone()));
        assert!(a.ok && a.detail.contains("public_repositories") && !a.detail.contains("tok@"), "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["refeed public_repositories,secret.smtp_from,secret.smtp_url"]);
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                true
            }
        }
        let none = run(&Bare, "main", Order::Refeed(r));
        assert!(!none.ok && none.detail.contains("keeps no estate keys"), "{none:?}");
    }

    #[test]
    fn restate_entitlements_verifies_the_snapshot_and_counts_new_and_held() {
        let b = Fake::default();
        run(&b, "main", Order::Entitlement(fact("a", 5, MONETIZE_SEED)));
        let facts = vec![fact("a", 5, MONETIZE_SEED), fact("b", 5, MONETIZE_SEED)];
        let mut snap = monetize_embed::Snapshot { issued_unix_ms: 7, facts, signature: Vec::new() };
        snap.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&monetize_embed::signing::snapshot_message(snap.issued_unix_ms, &snap.facts)).to_bytes().to_vec();
        let a = run(&b, "main", Order::RestateEntitlements(snap.clone()));
        assert!(a.ok && a.detail.contains("restated 1") && a.detail.contains("1 already held"), "{a:?}");
        snap.signature = vec![0; 64];
        let bad = run(&b, "main", Order::RestateEntitlements(snap));
        assert!(!bad.ok && bad.detail.contains("does not verify"), "{bad:?}");
    }

    #[test]
    fn purge_tenant_answers_monetizes_purge_answer() {
        let b = Fake::default();
        let a = run(&b, "main", Order::PurgeTenant(PurgeTenant { tenant: "libs".into(), reason: "nonpay".into(), job: String::new() }));
        assert!(a.ok, "{a:?}");
        let answer: PurgeAnswer = serde_json::from_str(&a.detail).unwrap();
        assert_eq!(answer.job, "purge:libs");
        assert_eq!(b.did.borrow().as_slice(), ["purge libs"]);
    }

    #[test]
    fn add_principal_makes_a_repository_on_a_plan_and_says_taken_by_name() {
        let b = Fake::default();
        let a = run(&b, "main", Order::AddPrincipal(AddPrincipal { username: "crates".into(), kind: "org".into(), signup: "s-1".into() }));
        assert!(a.ok && a.detail == "crates", "the namespace made is the repository: {a:?}");
        let t = run(&b, "main", Order::AddPrincipal(AddPrincipal { username: "theirs".into(), kind: "org".into(), signup: "s-2".into() }));
        assert!(!t.ok && t.detail.starts_with("taken"), "{t:?}");
        let twin = Fake { twin: true, ..Default::default() };
        assert!(!run(&twin, "twin", Order::AddPrincipal(AddPrincipal { username: "crates".into(), kind: "org".into(), signup: "s-1".into() })).ok, "only the main that serves");
    }

    #[test]
    fn grow_room_is_kept_for_the_operator() {
        let b = Fake::default();
        let a = run(&b, "main", Order::GrowRoom(GrowRoom { need_bytes: 1 << 30, reason: "libs".into() }));
        assert!(a.ok, "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["room 1073741824"]);
    }

    #[test]
    fn grow_go_ahead_runs_twin_first_asks_monetize_and_remounts() {
        let b = Fake::default();
        let (m, k) = (me("main"), monetize_key());
        let mut ask = Asked(Vec::new(), true);
        let o = Order::GrowGoAhead(GrowGoAhead { go_ahead: growth("acme", 120, 99_000), data_gib: 120 });
        let a = execute(&b, &Ctx { me: &m, key: &k, now_unix_ms: 10_000 }, 6, &o, &mut ask);
        assert!(a.ok, "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["twin_holds 120", "start 120 pull-6-o-7", "settle false", "look_again false", "settle true"]);
        assert_eq!(ask.0, [120]);

        // Monetize does not answer: /data is mounted back and the ack says why.
        let b = Fake::default();
        let mut deaf = Asked(Vec::new(), false);
        b.settle.borrow_mut().extend([GrowState::WaitingMembers("w".into()), GrowState::Failed("mounted back".into())]);
        let a = execute(&b, &Ctx { me: &m, key: &k, now_unix_ms: 10_000 }, 7, &o, &mut deaf);
        assert!(!a.ok && a.detail.contains("no answer") && a.detail.contains("mounted back"), "{a:?}");
        assert!(b.did.borrow().contains(&"look_again true".to_string()));
    }

    #[test]
    fn grow_go_ahead_refuses_another_estate_a_lapsed_step_and_a_forgery_and_skips_a_done_one() {
        let b = Fake::default();
        for (go, why) in [(growth("other", 120, 99_000), "estate other"), (growth("acme", 120, 5_000), "period is over"), (growth("acme", 120, 99_000).replace("\"o-7\"", "\"o-8\""), "does not verify")] {
            let a = run(&b, "main", Order::GrowGoAhead(GrowGoAhead { go_ahead: go, data_gib: 120 }));
            assert!(!a.ok && a.detail.contains(why), "{why}: {a:?}");
        }
        assert!(b.did.borrow().is_empty(), "nothing moved");
        let done = Fake { carried: true, ..Default::default() };
        let a = run(&done, "main", Order::GrowGoAhead(GrowGoAhead { go_ahead: growth("acme", 120, 99_000), data_gib: 120 }));
        assert!(a.ok && a.detail.contains("nothing moved"), "{a:?}");
        let twin = Fake { twin: true, ..Default::default() };
        let a = run(&twin, "twin", Order::GrowGoAhead(GrowGoAhead { go_ahead: growth("acme", 120, 99_000), data_gib: 240 }));
        assert!(a.detail.contains("no pull channel"), "{a:?}");
        assert!(!twin.did.borrow().iter().any(|d| d.starts_with("twin_holds")), "the twin grows first, asking no twin: {a:?}");
    }

    #[test]
    fn recharge_from_twin_runs_the_lane_on_a_cloud_fence_proof() {
        let b = Fake::default();
        let r = RechargeAsk { main: "main:1".into(), twin: "twin:1".into(), accept_lost: true, fenced: Some("upcloud: gone".into()), ..Default::default() };
        let a = run(&b, "twin", Order::RechargeFromTwin(r));
        assert!(a.ok && a.detail.starts_with("cloud-fenced"), "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["recharge main:1 from twin:1 lost=true"]);
    }

    /// ★ **Fence before recharge, by the pulled order** (M2, Gunnar's pattern): the old
    /// primary the order names is fenced first, and only a fence that held lets the
    /// recharge run; a fence that did not hold runs nothing; the main it fills is never
    /// the box it fences.
    #[test]
    fn recharge_from_twin_fences_the_old_primary_first() {
        let b = Fake::default();
        let r = RechargeAsk { main: "main:2".into(), twin: "twin:1".into(), accept_lost: false, epoch: 4, fence: Some("old:1".into()), ..Default::default() };
        let a = run(&b, "twin", Order::RechargeFromTwin(r.clone()));
        assert!(a.ok && a.detail.starts_with("fenced old:1; "), "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["fence old:1 4", "recharge main:2 from twin:1 lost=false"]);

        let stuck = Fake::default();
        let a = run(&stuck, "twin", Order::RechargeFromTwin(RechargeAsk { fence: Some("stuck:1".into()), ..r.clone() }));
        assert!(!a.ok && a.detail.contains("stuck:1 answers nothing") && a.detail.contains("nothing was recharged"), "{a:?}");
        assert_eq!(stuck.did.borrow().as_slice(), ["fence stuck:1 4"], "no recharge beside an unfenced primary");

        let same = Fake::default();
        let a = run(&same, "twin", Order::RechargeFromTwin(RechargeAsk { fence: Some("main:2".into()), ..r }));
        assert!(!a.ok && a.detail.contains("never fences the box it fills"), "{a:?}");
        assert!(same.did.borrow().is_empty(), "nothing ran");
    }

    #[test]
    fn become_twin_runs_holgers_election() {
        let b = Fake { twin: true, ..Default::default() };
        assert!(run(&b, "twin", Order::BecomeTwin { primary: "main:1".into() }).ok);
        assert_eq!(b.did.borrow().as_slice(), ["become_twin main:1"]);
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                false
            }
        }
        let none = run(&Bare, "twin", Order::BecomeTwin { primary: "main:1".into() });
        assert!(!none.ok && none.detail.contains("cannot become"), "the default refuses by name: {none:?}");
    }

    #[test]
    fn fence_primary_runs_the_boxes_fence_adopt_and_review_run() {
        let b = Fake::default();
        let a = run(&b, "twin", Order::FencePrimary { primary: "old:1".into(), epoch: 3 });
        assert!(a.ok && b.did.borrow().contains(&"fence old:1 3".to_string()), "{a:?}");
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                false
            }
        }
        let none = run(&Bare, "twin", Order::FencePrimary { primary: "old:1".into(), epoch: 3 });
        assert!(!none.ok && none.detail.contains("cannot fence"), "the default refuses by name: {none:?}");
        let a = run(&b, "main", Order::AdoptTwin { twin: "twin:1".into(), principal: "twin".into() });
        assert!(a.ok && b.did.borrow().contains(&"adopt twin:1 twin".to_string()), "{a:?}");
        let a = run(&b, "twin", Order::ReviewTwin);
        assert!(a.ok && a.detail == "the twin is whole" && a.receipt_json.contains("complete"), "{a:?}");
    }
}