monetize-estate-side 0.3.3

//! **`ShopDoor`: a sold estate's signed shop-door client** (feature `sold-door`; P4a, A22 W9; over `Fleet.Ask` since P3-FLEET).
//!
//! A sold estate runs NO monetize of its own (G16): its boxes pull from the ONE monetize
//! ([`crate::pump`]), and its front calls the ONE monetize (`Fleet.Ask`, H16) for what only the
//! shop can do for it — every call signed by the estate's client key (the fed
//! `secret.shop_client_key`), nothing listening on the front:
//!
//! | call | what |
//! |---|---|
//! | `TakeMaterial` | the installer material the shop's pairing left, then `done` |
//! | `Recharge` | one LOST box laid again with a fresh seed, then `done` |
//! | `Upgrade` | one standing box booted once from the release's keep medium (data kept), then `done` |
//! | `UpdateTake` / `UpdateReport` | the update the operator opened, pulled box by box (the shared update step's [`Shop`]) |
//!
//! Moved from Holger's `holger-estate-front/src/shop_door.rs` (Holger first); Gunnar's
//! `gunnar-front/src/shop_door.rs` carried the same `unhex32`, `from_status` and `blocking` and
//! moves in T16 ([`ShopDoor::from_settings`] and [`ShopDoor::with_seam`] are its constructors).
//! What a front does with the answers (keeping the material in its primary's slot, re-admitting a
//! box) stays the product's.
//!
//! **Thin on purpose** (TASKS-4-v2 T16, A28 H16): the SoldEstate → Fleet merge replaces the
//! transport ([`Wire`]) and nothing else.
//!
//! **P3-FLEET (plan §8, decided 2026-10-24): the five calls ride `Fleet.Ask`.** [`FleetWire`] asks
//! each as a `Fleet.Ask` kind (`upgrade`, `recharge`, `material`, `update_take`,
//! `update_report`; monetize-embed `fleet::ask::FRONT`), a `FleetCall` of `role = front` signed
//! by the estate's client key, the old door's answer decoded from `FleetAskAnswer.answer`. A door
//! fed its `product` ([`FEED_SHOP_DOOR`]) or told it ([`ShopDoor::or_over_fleet`]) takes it.
//! **H16: the old `SoldEstate` door and its overlap are gone** (the ONE box served both doors
//! for one release, R3+R4): there is no fallback, and a door that knows no product to sign as
//! refuses every call by name ([`NoProduct`]) — every live front says its product.

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

use ed25519_dalek::Signer as _;
use monetize_embed::fleet::{
    ask as fleet_ask, call as fleet_call, fleet_call_message, Ask, FleetCall, FrontAsk,
};
use monetize_embed::signing::{estate_call_message, EstateCall};
use monetize_proto as pb;
use nordisk_pairing::frontside::update::{LiveApp, Shop, Take};

pub use crate::unhex32;

/// The fed settings (`monetize-shop-pairing` `FEED_SHOP_DOOR`): TOML `door`, `company`,
/// `shop_pubkey`, `client_key_secret`.
pub const FEED_SHOP_DOOR: &str = "shop_door.toml";
/// The secret the client key is fed as when the settings name none (64 hex, its seed).
pub const CLIENT_KEY_SECRET: &str = "shop_client_key";

pub const CALL_UPGRADE: &str = "upgrade";
pub const CALL_RECHARGE: &str = "recharge";
pub const CALL_TAKE_MATERIAL: &str = "take-material";
pub const CALL_UPDATE_TAKE: &str = "update-take";
pub const CALL_UPDATE_REPORT: &str = "update-report";

/// An in-place upgrade waits on the shop's whole keep boot (import, clone, stop, the medium's
/// 45 min, give-back, start): two hours, the ONE constant the shop holds its Upgrade call by
/// (monetize `KEEP_BOOT_WITHIN`; A7 row 9, Gunnar b74ca801). A front that gave up first counted a
/// slow keep boot as lost and asked for a Recharge while the shop still worked.
pub const UPGRADE_WITHIN: Duration = monetize_embed::fleet::KEEP_BOOT_WITHIN;
/// A Recharge waits up to an hour (the shop lays a lost box within 30 min). Holger's bound;
/// Gunnar's front waits 30 min today (T16 takes this one).
pub const RECHARGE_WITHIN: Duration = Duration::from_secs(60 * 60);
/// Every other call (a take, a report, the material).
pub const READ_WITHIN: Duration = Duration::from_secs(30);

/// **How an Upgrade is asked: started, then polled** (REL-T2-1, 2026-10-29). One ask waits at
/// most `ask` for its answer; an ask that was cut (no answer from the shop: the connection
/// dropped, or `ask` passed) or answered "already running" is asked again every `every`, until
/// the shop answers or `within` has passed. The shop's keep boot runs on whether or not the
/// front is still on the line.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Polling {
    pub ask: Duration,
    pub every: Duration,
    pub within: Duration,
}

/// **The tag of a call that was cut**: no answer came from the shop (the connection dropped
/// mid-call, the ask's own bound passed, or the door could not be reached), so the shop may still
/// be working on it. Said as `unavailable: [shop-cut] <why>`.
pub const SHOP_CUT: &str = "[shop-cut]";

/// How often every call's connection pings the ONE box (HTTP/2 PING), and how long a ping may
/// go unanswered: the edge closes a connection silent for 120 s (edda
/// `estate-front::web::IDLE_TIMEOUT`), and a call waiting on the shop sends nothing else.
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(20);
pub const KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);

impl Polling {
    /// The front's own: each ask well under the ONE box edge's 120 s idle cut (edda
    /// `estate-front::web::IDLE_TIMEOUT`), polled every 15 s, for the shop's whole keep boot.
    pub const UPGRADE: Polling = Polling {
        ask: Duration::from_secs(60),
        every: Duration::from_secs(15),
        within: UPGRADE_WITHIN,
    };
}

/// The fed keys of the ONE box's pull channel the seam is read from (the boxes' own).
pub const FEED_FLEET: &str = "fleet";
pub const FEED_FLEET_AT: &str = nordisk_pairing::fleet::FEED_FLEET_AT;
pub const FEED_FLEET_CA: &str = nordisk_pairing::fleet::FEED_FLEET_CA;
/// The estate's name on the pull channel, fed beside the door (monetize-shop-pairing's
/// `FEED_FLEET_INSTANCE`); the door's `company` when none is fed.
pub const FEED_FLEET_INSTANCE: &str = "fleet_instance";
/// The role a front asks as on `Fleet`.
pub const FRONT_ROLE: &str = "front";

/// **Where the shop is, and as whom this estate calls it.**
#[derive(Clone)]
pub struct ShopDoor {
    /// `https://monetize.<apex>`.
    pub door: String,
    pub company: String,
    client: [u8; 32],
    clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
    /// **The seam** (row 16, Gunnar 7c3feb61): where the door is DIALLED (`fleet_at`,
    /// `host:port`) and the CA its certificate is verified under (`fleet_ca`), fed for the ONE
    /// box's pull channel. The TLS name stays the door's host. None: the door's host, under the
    /// public roots.
    pub seam: Option<Seam>,
    wire: Arc<dyn Wire>,
    /// Whether the calls ride `Fleet.Ask` ([`FleetWire`]).
    on_fleet: bool,
    /// How an Upgrade is asked ([`Polling::UPGRADE`]; a test's shop polls faster).
    polling: Polling,
}

/// The dialled address and the CA of the shop door ([`ShopDoor::seam`]).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Seam {
    pub at: String,
    pub ca_pem: Option<String>,
}

/// Where one call goes: the door, and its seam when fed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Dial {
    pub door: String,
    pub seam: Option<Seam>,
}

/// **The transport the calls ride** — what the SoldEstate → Fleet merge swaps. Each call is the
/// signed request as monetize-proto shapes it, answered within `within`; an error is the shop's
/// refusal by name (`refused: …`) or worth asking again (`unavailable: …`).
pub trait Wire: Send + Sync {
    fn upgrade(
        &self,
        d: &Dial,
        r: pb::EstateUpgradeRequest,
        within: Duration,
    ) -> Result<pb::EstateUpgradeResponse, String>;
    fn recharge(
        &self,
        d: &Dial,
        r: pb::EstateRechargeRequest,
        within: Duration,
    ) -> Result<pb::EstateRechargeResponse, String>;
    fn take_material(
        &self,
        d: &Dial,
        r: pb::EstateTakeMaterialRequest,
        within: Duration,
    ) -> Result<pb::EstateTakeMaterialResponse, String>;
    fn update_take(
        &self,
        d: &Dial,
        r: pb::EstateUpdateTakeRequest,
        within: Duration,
    ) -> Result<pb::EstateUpdateTakeResponse, String>;
    fn update_report(
        &self,
        d: &Dial,
        r: pb::EstateUpdateReportRequest,
        within: Duration,
    ) -> Result<pb::EstateUpdateReportResponse, String>;
    /// The product and the estate the calls are signed as on `Fleet` ([`FleetWire`]); none for
    /// any other wire.
    fn fleet_as(&self) -> Option<(String, String)> {
        None
    }
}

/// Pure: the host of an `http(s)://host[:port]/…` URL.
fn host_of(url: &str) -> &str {
    let rest = url.split_once("://").map(|(_, r)| r).unwrap_or(url);
    let authority = rest.split('/').next().unwrap_or("");
    authority
        .rsplit_once(':')
        .map(|(h, _)| h)
        .unwrap_or(authority)
}

/// Pure: **the seam of the shop door `door`** from the feed: `fleet_at` (and `fleet_ca`) when
/// the fed `fleet` is a door on the same host — the ONE box the boxes already dial there.
pub fn seam_of(door: &str, feed: &BTreeMap<String, String>) -> Option<Seam> {
    let get = |k: &str| feed.get(k).map(|v| v.trim()).filter(|v| !v.is_empty());
    let at = get(FEED_FLEET_AT)?;
    let fleet = get(FEED_FLEET)?;
    (host_of(fleet) == host_of(door) && !host_of(door).is_empty()).then(|| Seam {
        at: at.to_string(),
        ca_pem: get(FEED_FLEET_CA).map(str::to_string),
    })
}

impl std::fmt::Debug for ShopDoor {
    /// Never the client key.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ShopDoor")
            .field("door", &self.door)
            .field("company", &self.company)
            .field("at", &self.seam.as_ref().map(|s| s.at.as_str()))
            .finish_non_exhaustive()
    }
}

impl ShopDoor {
    /// **From the fed settings** (`shop_door.toml`'s `door`, `company`, `shop_pubkey`,
    /// `client_key_secret`) and the fed client key itself (64 hex). The shop's key, when named,
    /// must be a key. No seam: [`ShopDoor::with_seam`].
    pub fn from_settings(settings: &toml::Table, client_key_hex: &str) -> Result<ShopDoor, String> {
        ShopDoor::settings(settings, client_key_hex, true)
    }

    /// [`ShopDoor::from_settings`]; `shop_key` says whether a named `shop_pubkey` must be a key
    /// (Gunnar's front checks it; Holger's front verifies nothing under it and never did).
    fn settings(
        settings: &toml::Table,
        client_key_hex: &str,
        shop_key: bool,
    ) -> Result<ShopDoor, String> {
        let get = |k: &str| {
            settings
                .get(k)
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|v| !v.is_empty())
                .map(str::to_string)
        };
        let door = get("door")
            .ok_or("`door` is not set: the shop's monetize door (`https://monetize.<apex>`)")?;
        if !(door.starts_with("https://") || door.starts_with("http://")) {
            return Err(format!("`door` {door:?} is not an http(s) URL"));
        }
        let company =
            get("company").ok_or("`company` is not set: this estate's name at the shop")?;
        if let Some(k) = get("shop_pubkey").filter(|_| shop_key) {
            monetize_embed::VerifyingKey::from_bytes(&unhex32(&k, "`shop_pubkey`")?)
                .map_err(|e| format!("`shop_pubkey` is not a key: {e}"))?;
        }
        let secret = get("client_key_secret").unwrap_or_else(|| CLIENT_KEY_SECRET.to_string());
        let client = unhex32(
            client_key_hex,
            &format!("the client key (the fed secret `{secret}`)"),
        )?;
        let mut d = ShopDoor {
            door: door.trim_end_matches('/').to_string(),
            company,
            client,
            clock: None,
            seam: None,
            wire: Arc::new(NoProduct),
            on_fleet: false,
            polling: Polling::UPGRADE,
        };
        if let Some(product) = get("product") {
            let instance = d.company.clone();
            d = d.over_fleet(&product, &instance);
        }
        Ok(d)
    }

    /// **From the front's feed**; `None` when it names no shop door (an open or a site front). The
    /// seam is the ONE box's pull channel's, taken when its `fleet` is the same host
    /// ([`seam_of`]) and only as a bench feeds it (nordisk-pairing `fleet::seam_refusal`).
    pub fn from_feed(feed: &BTreeMap<String, String>) -> Option<Result<ShopDoor, String>> {
        let text = feed.get(FEED_SHOP_DOOR)?;
        Some((|| {
            let t: toml::Table =
                toml::from_str(text).map_err(|e| format!("{FEED_SHOP_DOOR}: {e}"))?;
            let secret = t
                .get("client_key_secret")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|v| !v.is_empty())
                .unwrap_or(CLIENT_KEY_SECRET)
                .to_string();
            let door = t
                .get("door")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .unwrap_or_default()
                .to_string();
            if door.is_empty() {
                return Err("`door` is not set: the shop's monetize door".into());
            }
            let key = feed
                .get(&format!("secret.{secret}"))
                .ok_or_else(|| format!("the client key (the fed secret `{secret}`) is not fed"))?;
            let mut d = ShopDoor::settings(&t, key, false)?;
            // A7 rows 16 and 23 (Gunnar 7c3feb61, 8134ec6f): the fed seam only as a bench's.
            let fed = |k: &str| feed.get(k).map(|v| v.trim()).filter(|v| !v.is_empty());
            if let Some(why) = nordisk_pairing::fleet::seam_refusal(
                fed(FEED_FLEET_AT),
                fed(FEED_FLEET_CA).is_some(),
            ) {
                return Err(why);
            }
            d.seam = seam_of(&d.door, feed);
            // The pull channel's name for this estate, when fed (an operator site's label).
            if let Some(instance) = fed(FEED_FLEET_INSTANCE).filter(|_| d.on_fleet) {
                let product = d.fleet_product().unwrap_or_default();
                d = d.over_fleet(&product, instance);
            }
            Ok(d)
        })())
    }

    /// **The bench's seam, named directly** (`fleet_at`, `fleet_ca`), honoured only as
    /// nordisk-pairing `fleet::seam_refusal` allows — a bench `ip:port`, its CA beside it;
    /// anything else refused by name.
    pub fn with_seam(mut self, at: Option<&str>, ca_pem: Option<&str>) -> Result<ShopDoor, String> {
        let at = at.map(str::trim).filter(|a| !a.is_empty());
        let ca = ca_pem.map(str::trim).filter(|c| !c.is_empty());
        if let Some(why) = nordisk_pairing::fleet::seam_refusal(at, ca.is_some()) {
            return Err(why);
        }
        self.seam = at.map(|a| Seam {
            at: a.to_string(),
            ca_pem: ca.map(str::to_string),
        });
        Ok(self)
    }

    /// Sign calls at `clock`'s time (a test's shop runs on its own clock).
    pub fn with_clock(mut self, clock: Arc<dyn Fn() -> i64 + Send + Sync>) -> ShopDoor {
        self.clock = Some(clock);
        self
    }

    /// Ask an Upgrade as `polling` says (a test's shop polls faster than [`Polling::UPGRADE`]).
    pub fn with_polling(mut self, polling: Polling) -> ShopDoor {
        self.polling = polling;
        self
    }

    /// The calls ride `wire` instead of `Fleet.Ask` (a test's shop).
    pub fn with_wire(mut self, wire: Arc<dyn Wire>) -> ShopDoor {
        self.wire = wire;
        self.on_fleet = false;
        self
    }

    /// **The calls ride `Fleet.Ask`** (plan §8), signed as `product`'s estate `instance`'s
    /// front, by the estate's client key.
    pub fn over_fleet(mut self, product: &str, instance: &str) -> ShopDoor {
        self.wire = Arc::new(FleetWire {
            product: product.trim().to_string(),
            instance: instance.trim().to_string(),
            client: self.client,
        });
        self.on_fleet = true;
        self
    }

    /// **Over `Fleet.Ask` as `product`, unless the feed already named one** — what a product's
    /// front calls with its own product name, so a front fed before the fold (no `product` in
    /// its `shop_door.toml`) moves to Fleet too.
    pub fn or_over_fleet(self, product: &str) -> ShopDoor {
        if self.on_fleet {
            return self;
        }
        let instance = self.company.clone();
        self.over_fleet(product, &instance)
    }

    /// Whether the calls ride `Fleet.Ask`.
    pub fn on_fleet(&self) -> bool {
        self.on_fleet
    }

    /// The product the calls are signed as on `Fleet`, when they ride it.
    pub fn fleet_product(&self) -> Option<String> {
        self.on_fleet
            .then(|| self.wire.fleet_as().map(|(p, _)| p))
            .flatten()
    }

    /// The estate's name on `Fleet`, when the calls ride it.
    pub fn fleet_instance(&self) -> Option<String> {
        self.on_fleet
            .then(|| self.wire.fleet_as().map(|(_, i)| i))
            .flatten()
    }

    /// The shop's name as a person reads it: `ignalina.dk` of `https://monetize.ignalina.dk`.
    pub fn shop_host(&self) -> String {
        let host = host_of(&self.door);
        host.strip_prefix("monetize.").unwrap_or(host).to_string()
    }

    /// Where every call goes.
    pub fn dial(&self) -> Dial {
        Dial {
            door: self.door.clone(),
            seam: self.seam.clone(),
        }
    }

    /// The signed call, as monetize-embed's `EstateCall` canonical JSON.
    pub fn signed(
        &self,
        call: &str,
        members: &[String],
        gib: u64,
        carries: &str,
    ) -> pb::EstateCallSignature {
        let at = match &self.clock {
            Some(c) => c(),
            None => std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |d| d.as_millis() as i64),
        };
        let c = EstateCall {
            v: 1,
            call: call.into(),
            estate: self.company.clone(),
            members: members.to_vec(),
            gib,
            at_unix_ms: at,
            carries: carries.to_string(),
        };
        let signature = ed25519_dalek::SigningKey::from_bytes(&self.client)
            .sign(&estate_call_message(&c))
            .to_bytes()
            .to_vec();
        pb::EstateCallSignature {
            at_unix_ms: at,
            signature,
        }
    }

    /// **`SoldEstate.Upgrade`**: boot this estate's standing `role` once from the release's keep
    /// medium, or, with `done`, record that it runs the release.
    ///
    /// **Started, then polled** (REL-T2-1, [`Polling`]): each ask waits at most `polling.ask`;
    /// one that was cut ([`SHOP_CUT`]: the ONE box's edge closed the silent connection, prod
    /// 2026-10-29) or answered "already running" ([`monetize_embed::fleet::UPGRADE_RUNNING`]) is
    /// asked again, freshly signed, until the shop answers. The shop's keep boot runs on without
    /// the front, and it records the release on the box, so its "already runs" after a cut is the
    /// keep boot this front started: answered `upgraded`, for the front to re-admit the box. A
    /// refusal by name and the shop's own failure come back at once, never asked again.
    pub fn upgrade(&self, role: &str, done: bool) -> Result<pb::EstateUpgradeResponse, String> {
        let p = self.polling;
        let started = std::time::Instant::now();
        let mut unanswered: Option<String> = None;
        loop {
            let req = pb::EstateUpgradeRequest {
                company: self.company.clone(),
                role: role.into(),
                done,
                signed: Some(self.signed(CALL_UPGRADE, &[role.to_string()], u64::from(done), "")),
            };
            match self.wire.upgrade(&self.dial(), req, p.ask) {
                Ok(mut a) => {
                    if let Some(why) = unanswered.as_ref().filter(|_| !done && !a.upgraded) {
                        a.upgraded = true;
                        a.words = format!(
                            "the shop keep-booted the {role} while the front polled it ({why}); {}",
                            a.words
                        );
                    }
                    return Ok(a);
                }
                Err(e) if still_working(&e) => {
                    if started.elapsed() + p.every >= p.within {
                        return Err(format!(
                            "{e} — no answer from the shop within {} min",
                            p.within.as_secs().div_ceil(60)
                        ));
                    }
                    unanswered = Some(e.trim_start_matches("unavailable: ").to_string());
                    std::thread::sleep(p.every);
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// **`SoldEstate.Recharge`**: lay this estate's LOST `role` again with `seed` as its
    /// user_data, or, with `done`, give its release clone back (the box is re-admitted).
    pub fn recharge_box(
        &self,
        role: &str,
        seed: &str,
        done: bool,
    ) -> Result<pb::EstateRechargeResponse, String> {
        let carries = if done { "" } else { seed };
        let req = pb::EstateRechargeRequest {
            company: self.company.clone(),
            role: role.into(),
            seed: carries.into(),
            done,
            signed: Some(self.signed(CALL_RECHARGE, &[role.to_string()], u64::from(done), carries)),
        };
        self.wire.recharge(&self.dial(), req, RECHARGE_WITHIN)
    }

    /// **`SoldEstate.TakeMaterial`**: the installer material the pairing left, or, with `done`,
    /// that it is kept (the shop forgets it at the hand-off).
    pub fn take_material(&self, done: bool) -> Result<pb::EstateTakeMaterialResponse, String> {
        let req = pb::EstateTakeMaterialRequest {
            company: self.company.clone(),
            done,
            signed: Some(self.signed(CALL_TAKE_MATERIAL, &[], u64::from(done), "")),
        };
        self.wire.take_material(&self.dial(), req, READ_WITHIN)
    }

    /// **`SoldEstate.UpdateTake`**: the update the operator opened, as the shop hands it out.
    pub fn update_take(&self) -> Result<pb::EstateUpdateTakeResponse, String> {
        let req = pb::EstateUpdateTakeRequest {
            company: self.company.clone(),
            signed: Some(self.signed(CALL_UPDATE_TAKE, &[], 0, "")),
        };
        self.wire.update_take(&self.dial(), req, READ_WITHIN)
    }

    /// **`SoldEstate.UpdateReport`**: `role`'s step, `outcome` and `words` signed together
    /// (`carries` = `<outcome>\n<words>`).
    pub fn update_report(
        &self,
        role: &str,
        outcome: &str,
        words: &str,
    ) -> Result<pb::EstateUpdateReportResponse, String> {
        let carries = format!("{outcome}\n{words}");
        let req = pb::EstateUpdateReportRequest {
            company: self.company.clone(),
            role: role.into(),
            signed: Some(self.signed(CALL_UPDATE_REPORT, &[role.to_string()], 0, &carries)),
            carries,
        };
        self.wire.update_report(&self.dial(), req, READ_WITHIN)
    }
}

/// Pure: **an Upgrade ask the shop is still working on**: cut before its answer ([`SHOP_CUT`]),
/// or answered "already running" by the shop.
pub fn still_working(e: &str) -> bool {
    e.starts_with("unavailable: ")
        && (e.contains(SHOP_CUT) || e.contains(monetize_embed::fleet::UPGRADE_RUNNING))
}

/// Pure: **the shop's live app as the update step takes it**.
pub fn live_app_of(l: pb::EstateLiveApp) -> LiveApp {
    LiveApp {
        base: l.base,
        app: l.app,
        app_len: l.app_len,
        sig: l.sig,
        url: l.url,
        sha256: l.sha256,
        release: l.release,
        release_sig: l.release_sig,
        env: l.env,
        release_url: l.release_url,
        release_sha256: l.release_sha256,
        release_len: l.release_len,
    }
}

/// Pure: **the shop's take as the update step takes it**.
pub fn take_of(r: pb::EstateUpdateTakeResponse) -> Take {
    Take {
        open: r.open,
        release: r.release,
        live: r.live,
        role: r.role,
        app: r.app.map(live_app_of),
        front: r.front.map(live_app_of),
        words: r.words,
    }
}

/// **The ONE box's update door** (`SoldEstate.UpdateTake` / `UpdateReport`) as the shared update
/// step (nordisk-pairing `frontside::update`) pulls it.
impl Shop for ShopDoor {
    fn take(&self) -> Result<Take, String> {
        self.update_take().map(take_of)
    }

    fn report(&self, role: &str, outcome: &str, words: &str) -> Result<String, String> {
        self.update_report(role, outcome, words).map(|r| r.words)
    }
}

/// Pure: the address a call to `door` is dialled at — `scheme://<at>` over the seam, the door
/// itself otherwise.
pub fn dialled(door: &str, seam: Option<&Seam>) -> String {
    match seam {
        Some(s) => format!(
            "{}://{}",
            door.split_once("://").map(|(sc, _)| sc).unwrap_or("https"),
            s.at.trim()
        ),
        None => door.to_string(),
    }
}

/// ★ **H16: the wire of a door that knows no product** — the old `SoldEstate` door is gone, so
/// a front that does not say which product it is (its `shop_door.toml` `product`, or
/// [`ShopDoor::or_over_fleet`]) calls nothing: every call is refused by name.
#[derive(Clone, Copy, Debug, Default)]
pub struct NoProduct;

/// The refusal [`NoProduct`] answers every call with.
pub const NO_PRODUCT: &str = "refused: this front's shop door names no product, and the ONE box is called over Fleet only (the old SoldEstate door is gone): feed `product` in shop_door.toml, or call or_over_fleet";

impl Wire for NoProduct {
    fn upgrade(
        &self,
        _: &Dial,
        _: pb::EstateUpgradeRequest,
        _: Duration,
    ) -> Result<pb::EstateUpgradeResponse, String> {
        Err(NO_PRODUCT.into())
    }
    fn recharge(
        &self,
        _: &Dial,
        _: pb::EstateRechargeRequest,
        _: Duration,
    ) -> Result<pb::EstateRechargeResponse, String> {
        Err(NO_PRODUCT.into())
    }
    fn take_material(
        &self,
        _: &Dial,
        _: pb::EstateTakeMaterialRequest,
        _: Duration,
    ) -> Result<pb::EstateTakeMaterialResponse, String> {
        Err(NO_PRODUCT.into())
    }
    fn update_take(
        &self,
        _: &Dial,
        _: pb::EstateUpdateTakeRequest,
        _: Duration,
    ) -> Result<pb::EstateUpdateTakeResponse, String> {
        Err(NO_PRODUCT.into())
    }
    fn update_report(
        &self,
        _: &Dial,
        _: pb::EstateUpdateReportRequest,
        _: Duration,
    ) -> Result<pb::EstateUpdateReportResponse, String> {
        Err(NO_PRODUCT.into())
    }
}

/// The ONE box's door, dialled as [`Dial`] says: TLS named as the door, under the seam's CA or
/// the public roots; plain http only for a loopback test's door.
async fn channel(d: Dial, within: Duration) -> Result<tonic::transport::Channel, String> {
    let door = d.door.as_str();
    let dial = dialled(door, d.seam.as_ref());
    let origin: tonic::codegen::http::Uri = door
        .parse()
        .map_err(|e| format!("the shop's door {door:?}: {e}"))?;
    let mut ep = tonic::transport::Endpoint::from_shared(dial.clone())
        .map_err(|e| format!("the shop's door {door:?} at {dial}: {e}"))?
        .origin(origin)
        .connect_timeout(Duration::from_secs(10))
        .timeout(within)
        // REL-T2-1: a call waiting on the shop sends nothing else; the ONE box's edge closes a silent connection.
        .http2_keep_alive_interval(KEEPALIVE_EVERY)
        .keep_alive_timeout(KEEPALIVE_TIMEOUT)
        .keep_alive_while_idle(true)
        .tcp_keepalive(Some(KEEPALIVE_EVERY));
    if door.starts_with("https://") {
        let tls = match d.seam.as_ref().and_then(|s| s.ca_pem.as_deref()) {
            Some(pem) => tonic::transport::ClientTlsConfig::new()
                .ca_certificate(tonic::transport::Certificate::from_pem(pem)),
            None => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
        };
        ep = ep
            .tls_config(tls.domain_name(host_of(door).to_string()))
            .map_err(|e| format!("the shop's door {door:?}: {e}"))?;
    }
    ep.connect().await.map_err(|e| {
        format!("unavailable: {SHOP_CUT} the shop's door {door} (at {dial}) does not answer: {e}")
    })
}

/// The shop's refusal by name; anything else is worth asking again.
pub fn from_status(s: tonic::Status) -> String {
    use tonic::Code;
    match s.code() {
        Code::Unauthenticated
        | Code::FailedPrecondition
        | Code::InvalidArgument
        | Code::PermissionDenied
        | Code::NotFound => format!("refused: {}", s.message()),
        _ => format!("unavailable: {}", s.message()),
    }
}

/// A call to the shop on its own thread and runtime: safe from any caller (a jera job's
/// blocking thread, a runtime's worker).
pub fn blocking<T: Send + 'static>(
    f: impl std::future::Future<Output = Result<T, String>> + Send + 'static,
) -> Result<T, String> {
    // A32-14, stays a task: one shop call on a runtime of its own, joined at once (safe from a jera job's blocking thread).
    std::thread::spawn(move || {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|e| format!("a runtime for the shop's door: {e}"))?
            .block_on(f)
    })
    .join()
    .unwrap_or_else(|_| Err("the call to the shop panicked".into()))
}

/// **The Fleet door** (plan §8, P3-FLEET): each call a `Fleet.Ask` of the front's kind
/// (monetize-embed `fleet::ask::FRONT`), body [`FrontAsk`], a `FleetCall` of `role = front`
/// signed by the estate's client key at the call's own signed time; the answer is the call's
/// message, decoded from `FleetAskAnswer.answer`. (H16: no fallback to the old door.)
#[derive(Clone)]
pub struct FleetWire {
    pub product: String,
    pub instance: String,
    client: [u8; 32],
}

impl std::fmt::Debug for FleetWire {
    /// Never the client key.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FleetWire")
            .field("product", &self.product)
            .field("instance", &self.instance)
            .finish_non_exhaustive()
    }
}

/// Why a Fleet ask did not come back: the door's refusal (its gRPC status), or no door.
enum Asked {
    Status(tonic::Status),
    Other(String),
}

impl FleetWire {
    /// Pure: **the signed `Fleet.Ask`** of `kind` with `body`, at `at_ms`.
    pub fn signed(&self, kind: &str, body: &FrontAsk, at_ms: u64) -> pb::FleetSigned {
        let body = serde_json::to_string(&Ask {
            kind: kind.into(),
            body: serde_json::to_value(body).unwrap_or_default(),
        })
        .unwrap_or_default();
        let c = FleetCall {
            product: self.product.clone(),
            instance: self.instance.clone(),
            role: FRONT_ROLE.into(),
            call: fleet_call::ASK.into(),
            body,
            at_unix_ms: at_ms,
        };
        let signature = ed25519_dalek::SigningKey::from_bytes(&self.client)
            .sign(&fleet_call_message(&c))
            .to_bytes()
            .to_vec();
        pb::FleetSigned {
            product: c.product,
            instance: c.instance,
            role: c.role,
            call: c.call,
            body: c.body,
            at_unix_ms: c.at_unix_ms,
            signature,
            box_key: Vec::new(),
        }
    }

    fn ask<M: prost::Message + Default>(
        &self,
        d: &Dial,
        kind: &str,
        body: FrontAsk,
        signed: Option<&pb::EstateCallSignature>,
        within: Duration,
    ) -> Result<M, Asked> {
        let at = signed.map_or(0, |s| s.at_unix_ms.max(0) as u64);
        let req = self.signed(kind, &body, at);
        let d = d.clone();
        let answer = blocking(async move {
            let mut c = pb::fleet_client::FleetClient::new(channel(d, within).await?);
            Ok(c.ask(req).await.map(|r| r.into_inner().answer))
        })
        .map_err(Asked::Other)?
        .map_err(Asked::Status)?;
        M::decode(answer.as_slice()).map_err(|e| {
            Asked::Other(format!(
                "unavailable: the ONE box's answer to {kind} is not its message: {e}"
            ))
        })
    }

    /// One call over Fleet.
    fn call<M: prost::Message + Default>(
        &self,
        d: &Dial,
        kind: &str,
        body: FrontAsk,
        signed: Option<&pb::EstateCallSignature>,
        within: Duration,
    ) -> Result<M, String> {
        match self.ask(d, kind, body, signed, within) {
            Ok(m) => Ok(m),
            // A status the shop never sent (the connection dropped, the ask's bound passed): cut.
            Err(Asked::Status(s)) if std::error::Error::source(&s).is_some() => {
                Err(format!("unavailable: {SHOP_CUT} {}", s.message()))
            }
            Err(Asked::Status(s)) => Err(from_status(s)),
            Err(Asked::Other(e)) => Err(e),
        }
    }
}

impl Wire for FleetWire {
    fn upgrade(
        &self,
        d: &Dial,
        r: pb::EstateUpgradeRequest,
        within: Duration,
    ) -> Result<pb::EstateUpgradeResponse, String> {
        let body = FrontAsk {
            role: r.role.clone(),
            done: r.done,
            ..Default::default()
        };
        self.call(d, fleet_ask::UPGRADE, body, r.signed.as_ref(), within)
    }
    fn recharge(
        &self,
        d: &Dial,
        r: pb::EstateRechargeRequest,
        within: Duration,
    ) -> Result<pb::EstateRechargeResponse, String> {
        let body = FrontAsk {
            role: r.role.clone(),
            done: r.done,
            seed: r.seed.clone(),
            ..Default::default()
        };
        self.call(d, fleet_ask::RECHARGE, body, r.signed.as_ref(), within)
    }
    fn take_material(
        &self,
        d: &Dial,
        r: pb::EstateTakeMaterialRequest,
        within: Duration,
    ) -> Result<pb::EstateTakeMaterialResponse, String> {
        let body = FrontAsk {
            done: r.done,
            ..Default::default()
        };
        self.call(d, fleet_ask::MATERIAL, body, r.signed.as_ref(), within)
    }
    fn update_take(
        &self,
        d: &Dial,
        r: pb::EstateUpdateTakeRequest,
        within: Duration,
    ) -> Result<pb::EstateUpdateTakeResponse, String> {
        self.call(
            d,
            fleet_ask::UPDATE_TAKE,
            FrontAsk::default(),
            r.signed.as_ref(),
            within,
        )
    }
    fn update_report(
        &self,
        d: &Dial,
        r: pb::EstateUpdateReportRequest,
        within: Duration,
    ) -> Result<pb::EstateUpdateReportResponse, String> {
        let body = FrontAsk {
            role: r.role.clone(),
            carries: r.carries.clone(),
            ..Default::default()
        };
        self.call(d, fleet_ask::UPDATE_REPORT, body, r.signed.as_ref(), within)
    }
    fn fleet_as(&self) -> Option<(String, String)> {
        Some((self.product.clone(), self.instance.clone()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    fn feed(door: &str) -> BTreeMap<String, String> {
        BTreeMap::from([
            (FEED_SHOP_DOOR.to_string(), format!("door = \"{door}\"\ncompany = \"acme\"\nshop_pubkey = \"{}\"\nclient_key_secret = \"shop_client_key\"\n", hex_of(&ed25519_dalek::SigningKey::from_bytes(&[0x11; 32]).verifying_key().to_bytes()))),
            ("secret.shop_client_key".to_string(), "22".repeat(32)),
        ])
    }

    fn hex_of(b: &[u8]) -> String {
        crate::hex(b)
    }

    /// ★ Row 16 (Gunnar 7c3feb61): the door is dialled over the fed seam when the fed `fleet`
    /// is the same ONE box — at `fleet_at`, verified under `fleet_ca`, named as the door — and
    /// at its own host otherwise.
    #[test]
    fn the_shop_door_dials_the_fed_seam_of_the_same_one_box() {
        let mut f = feed("https://monetize-test.ignalina.dk");
        assert!(
            ShopDoor::from_feed(&f).unwrap().unwrap().seam.is_none(),
            "no seam fed: the door's host"
        );
        f.insert(
            FEED_FLEET.into(),
            "https://monetize-test.ignalina.dk".into(),
        );
        f.insert(FEED_FLEET_AT.into(), "10.13.12.5:443".into());
        f.insert(
            FEED_FLEET_CA.into(),
            "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n".into(),
        );
        let d = ShopDoor::from_feed(&f).unwrap().unwrap();
        let seam = d.seam.clone().expect("the seam of the same box");
        assert_eq!(seam.at, "10.13.12.5:443");
        assert!(seam.ca_pem.is_some());
        assert_eq!(dialled(&d.door, d.seam.as_ref()), "https://10.13.12.5:443");
        assert_eq!(dialled(&d.door, None), "https://monetize-test.ignalina.dk");
        f.insert(FEED_FLEET.into(), "https://monetize.other.example".into());
        assert!(
            ShopDoor::from_feed(&f).unwrap().unwrap().seam.is_none(),
            "another box's seam is never this door's"
        );
    }

    /// ★ A7 rows 16 and 23 (Gunnar 7c3feb61, 8134ec6f): the fed seam is honoured only as a bench
    /// feeds it (nordisk-pairing `fleet::seam_refusal`): a public `fleet_at`, a name, or a
    /// `fleet_ca` with no `fleet_at` refuses the door by name; a bench `ip:port` is dialled. The
    /// same rule holds for a seam named directly (`with_seam`, Gunnar's constructor).
    #[test]
    fn a_seam_that_is_not_a_benchs_refuses_the_door_by_name() {
        let mut f = feed("https://monetize-test.ignalina.dk");
        f.insert(
            FEED_FLEET.into(),
            "https://monetize-test.ignalina.dk".into(),
        );
        for at in ["81.27.107.63:443", "monetize-test.ignalina.dk:443"] {
            f.insert(FEED_FLEET_AT.into(), at.into());
            let e = ShopDoor::from_feed(&f).unwrap().unwrap_err();
            assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
        }
        f.remove(FEED_FLEET_AT);
        f.insert(
            FEED_FLEET_CA.into(),
            "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n".into(),
        );
        assert!(
            ShopDoor::from_feed(&f)
                .unwrap()
                .unwrap_err()
                .contains("[fleet-seam-bench-only]"),
            "a CA with no bench address"
        );
        f.insert(FEED_FLEET_AT.into(), "10.0.2.2:35219".into());
        let d = ShopDoor::from_feed(&f).unwrap().unwrap();
        assert_eq!(
            d.seam.clone().map(|s| s.at),
            Some("10.0.2.2:35219".to_string())
        );
        assert!(d
            .clone()
            .with_seam(Some("81.27.107.63:443"), None)
            .unwrap_err()
            .contains("[fleet-seam-bench-only]"));
        assert_eq!(
            d.with_seam(Some(" 10.0.2.2:1 "), Some("PEM")).unwrap().seam,
            Some(Seam {
                at: "10.0.2.2:1".into(),
                ca_pem: Some("PEM".into())
            })
        );
    }

    /// ★ A7 row 9 (Gunnar b74ca801): an Upgrade waits on the shop's whole keep boot, the ONE
    /// constant the shop holds its call by (monetize `KEEP_BOOT_WITHIN`, 2 h).
    #[test]
    fn an_upgrade_waits_on_the_shops_whole_keep_boot() {
        assert_eq!(UPGRADE_WITHIN, Duration::from_secs(2 * 3600));
        assert_eq!(
            (RECHARGE_WITHIN, READ_WITHIN),
            (Duration::from_secs(3600), Duration::from_secs(30))
        );
    }

    #[test]
    fn the_fed_door_signs_its_calls_the_way_the_shop_verifies_them() {
        let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk/"))
            .unwrap()
            .unwrap()
            .with_clock(Arc::new(|| 1_000));
        assert_eq!(
            (d.door.as_str(), d.company.as_str(), d.shop_host().as_str()),
            ("https://monetize.ignalina.dk", "acme", "ignalina.dk")
        );
        let s = d.signed(CALL_TAKE_MATERIAL, &[], 0, "");
        let c = EstateCall {
            v: 1,
            call: CALL_TAKE_MATERIAL.into(),
            estate: "acme".into(),
            members: vec![],
            gib: 0,
            at_unix_ms: 1_000,
            carries: String::new(),
        };
        let vk = ed25519_dalek::SigningKey::from_bytes(&[0x22; 32]).verifying_key();
        monetize_embed::signing::verify_estate_call(&c, &s.signature, &vk, 1_000)
            .expect("the shop verifies the call under the estate's client key");
        assert!(
            ShopDoor::from_feed(&BTreeMap::new()).is_none(),
            "no shop door fed: none"
        );
        let mut no_key = feed("https://x");
        no_key.remove("secret.shop_client_key");
        assert!(ShopDoor::from_feed(&no_key)
            .unwrap()
            .unwrap_err()
            .contains("not fed"));
        assert!(
            !format!("{d:?}").contains("22"),
            "the client key is never printed"
        );
    }

    /// ★ The front's feed is taken as the shop feeds it (monetize-shop-pairing `front_feed`): the
    /// `shop_pubkey` beside the door is the boxes' monetize key, and the front verifies nothing
    /// under it, so a value that is no curve point does not refuse the door (Holger's
    /// `shop_pairing_contract` feeds `ab`×32). Gunnar's `from_settings` still checks it.
    #[test]
    fn the_fed_door_does_not_refuse_a_shop_key_it_never_verifies_under() {
        let mut f = feed("https://monetize.ignalina.dk");
        f.insert(FEED_SHOP_DOOR.into(), format!("door = \"https://monetize.ignalina.dk\"\ncompany = \"acme\"\nshop_pubkey = \"{}\"\n", "ab".repeat(32)));
        assert_eq!(ShopDoor::from_feed(&f).unwrap().unwrap().company, "acme");
        let t: toml::Table = toml::from_str(&f[FEED_SHOP_DOOR]).unwrap();
        assert!(ShopDoor::from_settings(&t, &"22".repeat(32))
            .unwrap_err()
            .contains("`shop_pubkey` is not a key"));
    }

    /// ★ Gunnar's constructor (`from_settings`): a door that is no URL, no company, a shop key
    /// that is no key and a client key that is not 64 hex are each refused by name.
    #[test]
    fn the_settings_are_checked_by_name() {
        let t = |s: &str| -> toml::Table { toml::from_str(s).unwrap() };
        let key = "22".repeat(32);
        assert!(ShopDoor::from_settings(&t("company = \"acme\""), &key)
            .unwrap_err()
            .contains("`door` is not set"));
        assert!(
            ShopDoor::from_settings(&t("door = \"monetize.x\"\ncompany = \"acme\""), &key)
                .unwrap_err()
                .contains("is not an http(s) URL")
        );
        assert!(ShopDoor::from_settings(&t("door = \"https://m.x\""), &key)
            .unwrap_err()
            .contains("`company` is not set"));
        assert!(ShopDoor::from_settings(
            &t("door = \"https://m.x\"\ncompany = \"acme\"\nshop_pubkey = \"12\""),
            &key
        )
        .unwrap_err()
        .contains("`shop_pubkey` is not 64 hex"));
        assert!(ShopDoor::from_settings(
            &t("door = \"https://m.x\"\ncompany = \"acme\"\nclient_key_secret = \"k\""),
            "12"
        )
        .unwrap_err()
        .contains("the client key (the fed secret `k`) is not 64 hex"));
        assert_eq!(
            ShopDoor::from_settings(&t("door = \"https://m.x/\"\ncompany = \" acme \""), &key)
                .unwrap()
                .dial(),
            Dial {
                door: "https://m.x".into(),
                seam: None
            }
        );
    }

    /// A shop on a wire of its own: every request it was handed, and its answers.
    #[derive(Default)]
    struct Heard(Mutex<Vec<(String, Duration, String)>>);

    impl Wire for Heard {
        fn upgrade(
            &self,
            d: &Dial,
            r: pb::EstateUpgradeRequest,
            within: Duration,
        ) -> Result<pb::EstateUpgradeResponse, String> {
            self.0.lock().unwrap().push((
                format!("upgrade {} done={}", r.role, r.done),
                within,
                d.door.clone(),
            ));
            Ok(pb::EstateUpgradeResponse {
                words: "booted".into(),
                ..Default::default()
            })
        }
        fn recharge(
            &self,
            _: &Dial,
            r: pb::EstateRechargeRequest,
            within: Duration,
        ) -> Result<pb::EstateRechargeResponse, String> {
            self.0.lock().unwrap().push((
                format!("recharge {} seed={}", r.role, r.seed),
                within,
                String::new(),
            ));
            Err("refused: acme is not delivered".into())
        }
        fn take_material(
            &self,
            _: &Dial,
            r: pb::EstateTakeMaterialRequest,
            within: Duration,
        ) -> Result<pb::EstateTakeMaterialResponse, String> {
            self.0.lock().unwrap().push((
                format!("take-material done={}", r.done),
                within,
                String::new(),
            ));
            Ok(pb::EstateTakeMaterialResponse {
                bundle: b"kept".to_vec(),
                words: "handed".into(),
            })
        }
        fn update_take(
            &self,
            _: &Dial,
            _: pb::EstateUpdateTakeRequest,
            within: Duration,
        ) -> Result<pb::EstateUpdateTakeResponse, String> {
            self.0
                .lock()
                .unwrap()
                .push(("update-take".into(), within, String::new()));
            Ok(pb::EstateUpdateTakeResponse {
                open: true,
                release: "03".into(),
                live: true,
                role: "twin".into(),
                app: Some(pb::EstateLiveApp {
                    base: "b".into(),
                    app: "a".into(),
                    app_len: 3,
                    env: vec!["korp.env.X=1".into()],
                    ..Default::default()
                }),
                front: None,
                words: "your turn".into(),
            })
        }
        fn update_report(
            &self,
            _: &Dial,
            r: pb::EstateUpdateReportRequest,
            within: Duration,
        ) -> Result<pb::EstateUpdateReportResponse, String> {
            self.0.lock().unwrap().push((
                format!("update-report {} {:?}", r.role, r.carries),
                within,
                String::new(),
            ));
            Ok(pb::EstateUpdateReportResponse {
                words: "recorded".into(),
            })
        }
    }

    /// ★ **The transport is swappable** (the SoldEstate → Fleet merge, A28 H16): every call rides
    /// the door's [`Wire`] with its own bound and the shop's words come back as they are; the
    /// update step's [`Shop`] is the same two calls.
    #[test]
    fn every_call_rides_the_doors_wire_with_its_own_bound() {
        let heard = Arc::new(Heard::default());
        let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
            .unwrap()
            .unwrap()
            .with_wire(heard.clone());
        assert_eq!(d.upgrade("twin", false).unwrap().words, "booted");
        assert_eq!(
            d.recharge_box("main", "SEED", false).unwrap_err(),
            "refused: acme is not delivered"
        );
        assert_eq!(d.take_material(true).unwrap().bundle, b"kept");
        let t = Shop::take(&d).unwrap();
        assert_eq!(
            (t.open, t.role.as_str(), t.words.as_str()),
            (true, "twin", "your turn")
        );
        assert_eq!(t.app.unwrap().env, vec!["korp.env.X=1".to_string()]);
        assert_eq!(
            Shop::report(&d, "twin", "done:live", "the twin runs app a").unwrap(),
            "recorded"
        );
        let h = heard.0.lock().unwrap();
        assert_eq!(
            h.iter()
                .map(|(c, w, _)| (c.as_str(), *w))
                .collect::<Vec<_>>(),
            vec![
                ("upgrade twin done=false", Polling::UPGRADE.ask),
                ("recharge main seed=SEED", RECHARGE_WITHIN),
                ("take-material done=true", READ_WITHIN),
                ("update-take", READ_WITHIN),
                (
                    "update-report twin \"done:live\\nthe twin runs app a\"",
                    READ_WITHIN
                ),
            ]
        );
        assert_eq!(h[0].2, "https://monetize.ignalina.dk");
    }

    /// ★ P3-FLEET (plan §8): a door fed its `product` rides `Fleet.Ask` as the fed
    /// `fleet_instance`'s front; a door fed before the fold (no `product`) calls nothing until its
    /// product's front says which product it is ([`ShopDoor::or_over_fleet`]) — H16: the old door
    /// is gone, so it is refused by name; a test's wire is never Fleet.
    #[test]
    fn a_door_fed_its_product_rides_fleet_and_an_older_feed_is_moved_by_its_product() {
        let mut f = feed("https://monetize.ignalina.dk");
        let old = ShopDoor::from_feed(&f).unwrap().unwrap();
        assert!(
            !old.on_fleet() && old.fleet_product().is_none(),
            "no product fed"
        );
        assert_eq!(
            old.update_take().unwrap_err(),
            NO_PRODUCT,
            "H16: no old door to fall back to"
        );
        let moved = old.clone().or_over_fleet("holger");
        assert_eq!(
            (
                moved.on_fleet(),
                moved.fleet_product().as_deref(),
                moved.fleet_instance().as_deref()
            ),
            (true, Some("holger"), Some("acme"))
        );
        f.insert(
            FEED_SHOP_DOOR.into(),
            format!("{}product = \"gunnar\"\n", f[FEED_SHOP_DOOR]),
        );
        let fed = ShopDoor::from_feed(&f).unwrap().unwrap();
        assert_eq!(
            (
                fed.fleet_product().as_deref(),
                fed.fleet_instance().as_deref()
            ),
            (Some("gunnar"), Some("acme"))
        );
        assert_eq!(
            fed.clone()
                .or_over_fleet("holger")
                .fleet_product()
                .as_deref(),
            Some("gunnar"),
            "the fed product wins"
        );
        f.insert(FEED_FLEET_INSTANCE.into(), "gunnar-rs".into());
        assert_eq!(
            ShopDoor::from_feed(&f)
                .unwrap()
                .unwrap()
                .fleet_instance()
                .as_deref(),
            Some("gunnar-rs"),
            "an operator site's label"
        );
        assert!(!fed.with_wire(Arc::new(Heard::default())).on_fleet());
        let t: toml::Table =
            toml::from_str("door = \"https://m.x\"\ncompany = \"acme\"\nproduct = \"gunnar\"")
                .unwrap();
        assert!(
            ShopDoor::from_settings(&t, &"22".repeat(32))
                .unwrap()
                .on_fleet(),
            "Gunnar's constructor reads it too"
        );
    }

    /// ★ The front's ask is a `FleetCall` of `role = front` the ONE box verifies under the
    /// estate's client key (the same key as the old door's `EstateCall`), the body the old
    /// request's fields; the key is never printed.
    #[test]
    fn the_fleet_ask_is_signed_as_the_front_by_the_estates_client_key() {
        let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
            .unwrap()
            .unwrap();
        let w = FleetWire {
            product: "gunnar".into(),
            instance: "acme".into(),
            client: [0x22; 32],
        };
        let s = w.signed(
            fleet_ask::RECHARGE,
            &FrontAsk {
                role: "main".into(),
                seed: "SEED".into(),
                ..Default::default()
            },
            1_000,
        );
        assert_eq!(
            (
                s.product.as_str(),
                s.instance.as_str(),
                s.role.as_str(),
                s.call.as_str(),
                s.at_unix_ms
            ),
            ("gunnar", "acme", "front", "ask", 1_000)
        );
        assert!(s.box_key.is_empty(), "a front pins nothing");
        let c = FleetCall {
            product: s.product.clone(),
            instance: s.instance.clone(),
            role: s.role.clone(),
            call: s.call.clone(),
            body: s.body.clone(),
            at_unix_ms: s.at_unix_ms,
        };
        let vk = ed25519_dalek::SigningKey::from_bytes(&[0x22; 32]).verifying_key();
        monetize_embed::fleet::verify_fleet_call(&c, &s.signature, &vk, 1_000)
            .expect("the ONE box verifies the ask under the estate's client key");
        let a: Ask = serde_json::from_str(&s.body).unwrap();
        assert_eq!(a.kind, "recharge");
        assert_eq!(
            serde_json::from_value::<FrontAsk>(a.body).unwrap(),
            FrontAsk {
                role: "main".into(),
                seed: "SEED".into(),
                ..Default::default()
            }
        );
        assert!(
            !format!("{w:?}").contains("22") && !format!("{d:?}").contains("2222"),
            "the client key is never printed"
        );
    }

    /// A shop whose Upgrade answers are scripted, one per ask, in order.
    struct Scripted(
        Mutex<Vec<Result<pb::EstateUpgradeResponse, String>>>,
        Mutex<Vec<Duration>>,
    );

    impl Scripted {
        fn new(answers: Vec<Result<pb::EstateUpgradeResponse, String>>) -> Arc<Scripted> {
            Arc::new(Scripted(Mutex::new(answers), Mutex::default()))
        }
        fn asks(&self) -> Vec<Duration> {
            self.1.lock().unwrap().clone()
        }
    }

    impl Wire for Scripted {
        fn upgrade(
            &self,
            _: &Dial,
            _: pb::EstateUpgradeRequest,
            within: Duration,
        ) -> Result<pb::EstateUpgradeResponse, String> {
            self.1.lock().unwrap().push(within);
            let mut a = self.0.lock().unwrap();
            assert!(!a.is_empty(), "asked once too often");
            a.remove(0)
        }
        fn recharge(
            &self,
            _: &Dial,
            _: pb::EstateRechargeRequest,
            _: Duration,
        ) -> Result<pb::EstateRechargeResponse, String> {
            unreachable!()
        }
        fn take_material(
            &self,
            _: &Dial,
            _: pb::EstateTakeMaterialRequest,
            _: Duration,
        ) -> Result<pb::EstateTakeMaterialResponse, String> {
            unreachable!()
        }
        fn update_take(
            &self,
            _: &Dial,
            _: pb::EstateUpdateTakeRequest,
            _: Duration,
        ) -> Result<pb::EstateUpdateTakeResponse, String> {
            unreachable!()
        }
        fn update_report(
            &self,
            _: &Dial,
            _: pb::EstateUpdateReportRequest,
            _: Duration,
        ) -> Result<pb::EstateUpdateReportResponse, String> {
            unreachable!()
        }
    }

    const FAST: Polling = Polling {
        ask: Duration::from_secs(60),
        every: Duration::from_millis(1),
        within: Duration::from_secs(30),
    };

    /// ★ **REL-T2-1 (prod 2026-10-29, test2.gunnar.rs and holger.rs)**: the ONE box's edge cut the
    /// front's Upgrade after ~4 min ("unavailable: transport error") while the shop's keep boot ran
    /// on to the end, and the update was recorded failed. The Upgrade is started, then polled: a
    /// cut ask and an "already running" answer are asked again, and the shop's "already runs"
    /// after a cut is the keep boot this front started — `upgraded`, so the front re-admits it.
    #[test]
    fn an_upgrade_cut_mid_call_is_polled_to_the_shops_answer_never_recorded_failed() {
        let running = |host: &str| {
            Err(format!(
                "unavailable: an upgrade of {host} {}",
                monetize_embed::fleet::UPGRADE_RUNNING
            ))
        };
        let shop = Scripted::new(vec![
            Err(format!("unavailable: {SHOP_CUT} transport error")),
            running("acme.example"),
            Err(format!(
                "unavailable: {SHOP_CUT} the shop's door https://m.x does not answer"
            )),
            running("acme.example"),
            Ok(pb::EstateUpgradeResponse {
                upgraded: false,
                words: "acme.example's twin already runs release 03: nothing to upgrade".into(),
                ..Default::default()
            }),
        ]);
        let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
            .unwrap()
            .unwrap()
            .with_wire(shop.clone())
            .with_polling(FAST);
        let up = d
            .upgrade("twin", false)
            .expect("a cut ask is never the update's failure");
        assert!(
            up.upgraded,
            "the keep boot the cut ask started is re-admitted by the front: {up:?}"
        );
        assert!(up.words.contains("already runs release 03"), "{}", up.words);
        assert_eq!(
            shop.asks(),
            vec![FAST.ask; 5],
            "every ask bounded by the poll's own bound, never the whole keep boot"
        );

        // Asked at once, not cut: the shop's own "already runs" is nothing to re-admit.
        let shop = Scripted::new(vec![Ok(pb::EstateUpgradeResponse {
            upgraded: false,
            words: "already runs".into(),
            ..Default::default()
        })]);
        let d = d.with_wire(shop.clone());
        assert!(!d.upgrade("twin", false).unwrap().upgraded);
        // A refusal by name, and the shop's own failure (its keep boot failed: CODE RED on its
        // side), are answered at once: never asked again, no second keep boot.
        for said in [
            "refused: no update is open on the twin",
            "unavailable: the provider: the medium left in",
        ] {
            let shop = Scripted::new(vec![
                Err(format!("unavailable: {SHOP_CUT} transport error")),
                Err(said.to_string()),
            ]);
            assert_eq!(
                d.clone()
                    .with_wire(shop.clone())
                    .upgrade("twin", false)
                    .unwrap_err(),
                said
            );
            assert_eq!(shop.asks().len(), 2);
        }
        // A shop that never answers within the poll's bound: the cut is said, by name.
        let shop = Scripted::new(
            (0..1000)
                .map(|_| Err(format!("unavailable: {SHOP_CUT} transport error")))
                .collect(),
        );
        let e = d
            .with_wire(shop)
            .with_polling(Polling {
                within: Duration::from_millis(30),
                ..FAST
            })
            .upgrade("twin", false)
            .unwrap_err();
        assert!(
            e.starts_with("unavailable: ")
                && e.contains("transport error")
                && e.contains("no answer from the shop within"),
            "{e}"
        );
    }

    /// ★ REL-T2-1: the front's own poll keeps each ask under the ONE box edge's 120 s idle cut
    /// (edda `estate-front::web::IDLE_TIMEOUT`), and every call's connection pings the edge well
    /// inside it (a Recharge waits up to an hour on one call).
    #[test]
    fn every_ask_stays_under_the_edges_idle_cut() {
        let edge_idle = Duration::from_secs(120);
        assert!(Polling::UPGRADE.ask < edge_idle && Polling::UPGRADE.within == UPGRADE_WITHIN);
        assert!(KEEPALIVE_EVERY < edge_idle / 2 && KEEPALIVE_TIMEOUT < KEEPALIVE_EVERY);
    }

    #[test]
    fn a_refusal_is_said_by_name_and_anything_else_is_worth_asking_again() {
        assert_eq!(
            from_status(tonic::Status::failed_precondition("[update-out-of-turn] x")),
            "refused: [update-out-of-turn] x"
        );
        assert_eq!(
            from_status(tonic::Status::unauthenticated("bad sig")),
            "refused: bad sig"
        );
        assert_eq!(
            from_status(tonic::Status::unavailable("down")),
            "unavailable: down"
        );
        assert_eq!(blocking(async { Ok::<_, String>(7) }), Ok(7));
    }
}