monetize_estate_side/pump/

puller.rs

//! **The puller**: report, pull, verify, run, ack — the box's whole side of
//! `monetize.v1.Fleet`, over any [`Link`] (the gRPC client in [`super::grpc`], an in-process
//! guardian in the tests), running each order through the product's [`PumpBox`].
//!
//! Moved from Holger's `holger-fleet/src/puller.rs` (P4a). What was Holger's there and is the
//! product's here: the order kinds and their bodies ([`PumpBox::run`]), the health words.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use monetize_embed::fleet::{ask, call, kind, order_message, Ask, FleetCall, OrderAck, OrdersAfter, Report, SignedOrder, LONG_POLL_MAX_MS};
use monetize_embed::VerifyingKey;
use serde::{Deserialize, Serialize};

use super::key::BoxKey;

/// What the puller remembers across restarts.
pub const STATE_REL: &str = "fleet/pulled.json";

/// Who this box is on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Addressee {
    pub product: String,
    pub instance: String,
    /// `main` | `twin`.
    pub role: String,
}

/// Why a call failed: monetize's word (`unauthenticated`, `unavailable`, …) and its sentence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fault {
    pub code: String,
    pub message: String,
}

impl Fault {
    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Fault {
        Fault { code: code.into(), message: message.into() }
    }
}

impl std::fmt::Display for Fault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code, self.message)
    }
}

/// One signed call as it goes on the wire.
#[derive(Debug, Clone)]
pub struct Signed {
    pub call: FleetCall,
    pub signature: Vec<u8>,
    /// The box's public key, sent until monetize has heard the box once (its pin).
    pub box_key: Vec<u8>,
}

/// **What an ask is answered**: the case it opened (a remedy), monetize's words, and a page
/// to open (empty for every ask a box makes today).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Asked {
    pub case_id: String,
    pub words: String,
    pub url: String,
}

/// **The three calls**, over whatever carries them.
pub trait Link {
    /// Answers monetize's `next_report_ms`.
    fn report(&mut self, s: Signed) -> Result<u64, Fault>;
    fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault>;
    fn ask(&mut self, s: Signed) -> Result<Asked, Fault>;
    /// [`Link::ask`] with the answer's BYTES (`FleetAskAnswer.answer`): the file a `fetch` ask
    /// pulls (MONETIZE-FIX). A link that carries no bytes answers none.
    fn ask_bytes(&mut self, s: Signed) -> Result<(Asked, Vec<u8>), Fault> {
        self.ask(s).map(|a| (a, Vec::new()))
    }
}

/// **A `Fleet.Ask` from inside a running order** (a pulled growth's `grow_set`), signed by this
/// box, on the link the order came on.
pub trait Asker {
    fn ask(&mut self, a: &Ask) -> Result<Asked, Fault>;
    /// [`Asker::ask`] with the answer's bytes ([`Link::ask_bytes`]).
    fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
        self.ask(a).map(|x| (x, Vec::new()))
    }
}

/// **A pulled growth's disk resize** (`Fleet.Ask{grow_set}`, every member, `gib` a disk): monetize's
/// words, or the case it opened when it said none.
pub fn grow_set(asker: &mut dyn Asker, gib: u64) -> Result<String, String> {
    let a = Ask { kind: ask::GROW_SET.into(), body: serde_json::json!({ "members": [], "gib": gib }) };
    let Asked { case_id: case, words, .. } = asker.ask(&a).map_err(|f| f.to_string())?;
    Ok(if words.is_empty() { case } else { words })
}

/// **The file a product order carries, pulled** (MONETIZE-FIX 2026-11-01: the ONE box never
/// pushes a file into an estate): `Fleet.Ask{fetch}` on the order's own link. The ONE box answers
/// only a file an order queued for this box names, checked against its sha256 there; its size
/// is checked here before it is handed to the product's executor.
pub fn fetch(asker: &mut dyn Asker, f: &monetize_embed::fleet::Fetch) -> Result<Vec<u8>, String> {
    let a = Ask { kind: ask::FETCH.into(), body: serde_json::json!({ "id": f.id }) };
    let (_, bytes) = asker.ask_bytes(&a).map_err(|e| e.to_string())?;
    if bytes.len() as u64 != f.bytes {
        return Err(format!("the file pulled for {} is {} bytes, and the order names {}", f.param, bytes.len(), f.bytes));
    }
    Ok(bytes)
}

/// What the product's executor knows about the order's context.
pub struct Ctx<'a> {
    pub me: &'a Addressee,
    /// The monetize key the order verified under (a fact or snapshot inside it is verified again).
    pub key: &'a VerifyingKey,
    pub now_unix_ms: u64,
}

/// **The product's half of the pump**: what its box reports, and what it does with an order.
pub trait PumpBox {
    /// 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
    }
    /// **Run one order that passed [`check`]** (signed by monetize, for this box, newer, in its
    /// window): decode its body, run it through the product's own verbs, and answer its ack. A
    /// kind or body the product does not take is an ack not-ok, by name. `ask` is
    /// `Fleet.Ask` on the order's own link ([`grow_set`]).
    fn run(&self, ctx: &Ctx, order: &SignedOrder, ask: &mut dyn Asker) -> OrderAck;
}

/// How far ahead of this box's clock an order may be dated (Gunnar's order check: five minutes).
/// ONE constant with monetize-embed's `fleet::verify_order` (T16, A28 M11).
pub use monetize_embed::fleet::FUTURE_SKEW_MS;

/// What [`check`] made of one wire order, before its body is read.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Checked {
    /// Not monetize's: never run, never acked, the sequence does not move.
    Forged(String),
    /// Monetize's, but not for this box to run now: acked not-ok with these words.
    Refused(String),
    /// Monetize's, for this box, newer than the last one taken, in its window: run it.
    Fresh,
}

/// **The box's check of one order** (the order's body is the product's): signed by monetize
/// (monetize-embed [`order_message`] under `key`), then for this box, newer than the last order
/// taken, inside its window (`not_after` 0 = none), dated no more than [`FUTURE_SKEW_MS`] ahead
/// of this box's clock. An order that does not verify is FORGED —
/// never run, never acked, and it never moves the sequence (a forged high seq must not silence
/// the real ones). Everything else wrong with it is a REFUSAL, acked by name.
pub fn check(o: &SignedOrder, me: &Addressee, key: &VerifyingKey, last_seq: u64, now_unix_ms: u64) -> Checked {
    // STRICT (P4c, A28 M11): no small-order key or `R`, as Gunnar's own check always was.
    let signed = ed25519_dalek::Signature::from_slice(&o.signature).map(|s| key.verify_strict(&order_message(o), &s).is_ok()).unwrap_or(false);
    if !signed {
        return Checked::Forged(format!("order {} ({}) is not signed by this product's monetize key", o.seq, o.kind));
    }
    if o.product != me.product || o.instance != me.instance || o.role != me.role {
        return Checked::Refused(format!("order {} is for {}/{}/{}, and this box is {}/{}/{}", o.seq, o.product, o.instance, o.role, me.product, me.instance, me.role));
    }
    if o.seq <= last_seq {
        return Checked::Refused(format!("order {} is not newer than {last_seq}, the last one this box took: a replay", o.seq));
    }
    if o.not_after_unix_ms != 0 && now_unix_ms > o.not_after_unix_ms {
        return Checked::Refused(format!("order {} expired {} s ago", o.seq, (now_unix_ms - o.not_after_unix_ms) / 1000));
    }
    if o.issued_unix_ms > now_unix_ms.saturating_add(FUTURE_SKEW_MS) {
        return Checked::Refused(format!("order {} is dated {} s in the future; whose clock is wrong?", o.seq, (o.issued_unix_ms - now_unix_ms) / 1000));
    }
    Checked::Fresh
}

/// What is kept across restarts.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct Pulled {
    /// The last order TAKEN (persisted before it runs: an order runs at most once).
    pub last_seq: u64,
    /// Acks not yet carried by a Report monetize answered.
    pub pending: Vec<OrderAck>,
    /// A review a `review_twin` order made: it rides the next Report only.
    pub twin_review: Option<String>,
}

/// What one pull did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Pull {
    pub acks: Vec<OrderAck>,
    pub forged: Vec<String>,
}

pub struct Puller {
    pub me: Addressee,
    pub build: String,
    key: BoxKey,
    state: Option<PathBuf>,
    pub pulled: Pulled,
    heard: bool,
}

impl Puller {
    /// A puller that remembers nothing on disk (tests).
    pub fn new(me: Addressee, key: BoxKey, build: impl Into<String>) -> Puller {
        Puller { me, build: build.into(), key, state: None, pulled: Pulled::default(), heard: false }
    }

    /// The box's puller over its data root: its key (minted on first use) and what it took.
    pub fn at(root: &Path, me: Addressee, build: impl Into<String>) -> std::io::Result<Puller> {
        let key = BoxKey::load_or_mint(root)?;
        let state = root.join(STATE_REL);
        let pulled = match std::fs::read(&state) {
            Ok(bytes) => serde_json::from_slice(&bytes).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, format!("{}: {e}", state.display())))?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Pulled::default(),
            Err(e) => return Err(e),
        };
        Ok(Puller { me, build: build.into(), key, state: Some(state), pulled, heard: false })
    }

    pub fn box_public(&self) -> [u8; 32] {
        self.key.public()
    }

    fn save(&self) -> Result<(), Fault> {
        let Some(path) = &self.state else { return Ok(()) };
        let bytes = serde_json::to_vec(&self.pulled).map_err(|e| Fault::new("internal", e.to_string()))?;
        crate::write_atomic(path, &bytes, false).map_err(|e| Fault::new("io", format!("{}: {e}", path.display())))
    }

    fn signed(&self, which: &str, body: String, now_unix_ms: u64) -> Signed {
        let c = FleetCall { product: self.me.product.clone(), instance: self.me.instance.clone(), role: self.me.role.clone(), call: which.to_owned(), body, at_unix_ms: now_unix_ms };
        let signature = self.key.sign(&c);
        Signed { call: c, signature, box_key: if self.heard { Vec::new() } else { self.key.public().to_vec() } }
    }

    /// **Report** this box's health, the product's own part (the main that serves), its own
    /// twin review (a twin, every report) and every pending ack. The acks leave the queue
    /// once monetize answered. Returns monetize's `next_report_ms`.
    pub fn report(&mut self, link: &mut dyn Link, health: BTreeMap<String, String>, product: Option<String>, own_review: Option<String>, now_unix_ms: u64) -> Result<u64, Fault> {
        let sent = self.pulled.pending.len();
        let report = Report {
            build: self.build.clone(),
            health,
            acked: self.pulled.pending.clone(),
            twin_review: own_review.or_else(|| self.pulled.twin_review.clone()),
            product,
            measured_unix_ms: now_unix_ms,
        };
        let body = serde_json::to_string(&report).map_err(|e| Fault::new("internal", e.to_string()))?;
        let next = link.report(self.signed(call::REPORT, body, now_unix_ms))?;
        self.heard = true;
        self.pulled.pending.drain(..sent);
        self.pulled.twin_review = None;
        self.save()?;
        Ok(next)
    }

    /// **Pull** the orders after the last one taken, held up to `wait_ms` (≤ 25 s); [`check`]
    /// each under `monetize_pubkey`, run it through `b` and queue its ack. Without a key nothing
    /// is pulled: nothing could be verified. A `review_twin` that ran rides the next report.
    pub fn pull(&mut self, link: &mut dyn Link, b: &dyn PumpBox, monetize_pubkey: Option<&[u8; 32]>, wait_ms: u64, now_unix_ms: u64) -> Result<Pull, Fault> {
        let Some(raw) = monetize_pubkey else {
            return Err(Fault::new("no_key", "this box has no monetize key yet, so no order could be verified: none is pulled"));
        };
        let key = VerifyingKey::from_bytes(raw).map_err(|e| Fault::new("no_key", format!("the monetize key is not an Ed25519 point: {e}")))?;
        let body = serde_json::to_string(&OrdersAfter { after: self.pulled.last_seq, wait_ms: wait_ms.min(LONG_POLL_MAX_MS) }).map_err(|e| Fault::new("internal", e.to_string()))?;
        let mut orders = link.orders(self.signed(call::ORDERS, body, now_unix_ms))?;
        self.heard = true;
        orders.sort_by_key(|o| o.seq);
        let mut out = Pull::default();
        for o in orders {
            let checked = check(&o, &self.me, &key, self.pulled.last_seq, now_unix_ms);
            if let Checked::Forged(why) = checked {
                out.forged.push(why);
                continue;
            }
            if o.seq <= self.pulled.last_seq {
                continue;
            }
            self.pulled.last_seq = o.seq;
            self.save()?;
            let ack = match checked {
                Checked::Fresh => {
                    let me = self.me.clone();
                    let ctx = Ctx { me: &me, key: &key, now_unix_ms };
                    let ack = {
                        let mut via = AskVia { puller: &mut *self, link: &mut *link, now_unix_ms };
                        b.run(&ctx, &o, &mut via)
                    };
                    if o.kind == kind::REVIEW_TWIN && ack.ok {
                        self.pulled.twin_review = Some(ack.receipt_json.clone());
                    }
                    ack
                }
                Checked::Refused(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
                Checked::Forged(_) => unreachable!("forged orders were skipped"),
            };
            self.pulled.pending.push(ack.clone());
            self.save()?;
            out.acks.push(ack);
        }
        Ok(out)
    }

    /// **Ask** monetize (`grow_set`, or a remedy: it opens a case, never approves one).
    pub fn ask(&mut self, link: &mut dyn Link, a: &Ask, now_unix_ms: u64) -> Result<Asked, Fault> {
        let body = serde_json::to_string(a).map_err(|e| Fault::new("internal", e.to_string()))?;
        let answer = link.ask(self.signed(call::ASK, body, now_unix_ms))?;
        self.heard = true;
        Ok(answer)
    }

    /// [`Puller::ask`] with the answer's bytes ([`Link::ask_bytes`]).
    pub fn ask_bytes(&mut self, link: &mut dyn Link, a: &Ask, now_unix_ms: u64) -> Result<(Asked, Vec<u8>), Fault> {
        let body = serde_json::to_string(a).map_err(|e| Fault::new("internal", e.to_string()))?;
        let answer = link.ask_bytes(self.signed(call::ASK, body, now_unix_ms))?;
        self.heard = true;
        Ok(answer)
    }
}

/// An order's `Fleet.Ask`, signed by this box, on the link the order came on.
struct AskVia<'a> {
    puller: &'a mut Puller,
    link: &'a mut dyn Link,
    now_unix_ms: u64,
}

impl Asker for AskVia<'_> {
    fn ask(&mut self, a: &Ask) -> Result<Asked, Fault> {
        self.puller.ask(self.link, a, self.now_unix_ms)
    }
    fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
        self.puller.ask_bytes(self.link, a, self.now_unix_ms)
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use ed25519_dalek::Signer as _;
    use std::cell::RefCell;

    pub const MONETIZE_SEED: [u8; 32] = [7u8; 32];

    pub fn me(role: &str) -> Addressee {
        Addressee { product: "a-product".into(), instance: "acme".into(), role: role.into() }
    }

    pub fn monetize_key() -> VerifyingKey {
        ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).verifying_key()
    }

    fn mkey() -> [u8; 32] {
        monetize_key().to_bytes()
    }

    /// One order for `role`, signed by `seed`.
    pub fn signed(seq: u64, role: &str, kind: &str, body: &str, seed: [u8; 32]) -> SignedOrder {
        let mut o = SignedOrder { seq, product: "a-product".into(), instance: "acme".into(), role: role.into(), kind: kind.into(), body: body.into(), issued_unix_ms: 9_000, not_after_unix_ms: 0, signature: Vec::new() };
        o.signature = ed25519_dalek::SigningKey::from_bytes(&seed).sign(&order_message(&o)).to_bytes().to_vec();
        o
    }

    /// Monetize in memory: a queue, and every call it heard.
    #[derive(Default)]
    pub struct Guardian {
        pub queue: Vec<SignedOrder>,
        pub heard: Vec<Signed>,
    }

    impl Link for Guardian {
        fn report(&mut self, s: Signed) -> Result<u64, Fault> {
            let r: Report = serde_json::from_str(&s.call.body).unwrap();
            self.queue.retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
            self.heard.push(s);
            Ok(30_000)
        }
        fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
            let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
            self.heard.push(s);
            Ok(self.queue.iter().filter(|o| o.seq > after.after).cloned().collect())
        }
        fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
            let _: Ask = serde_json::from_str(&s.call.body).unwrap();
            self.heard.push(s);
            Ok(Asked { case_id: "c-1".into(), words: "resized".into(), url: String::new() })
        }
    }

    /// A product box: it runs `note` (its words) and `grow` (asks `grow_set`), reviews as a
    /// twin, and refuses any other kind by name.
    #[derive(Default)]
    pub struct Fake {
        pub ran: RefCell<Vec<(u64, String)>>,
    }

    impl PumpBox for Fake {
        fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
            self.ran.borrow_mut().push((o.seq, o.kind.clone()));
            let said = match o.kind.as_str() {
                "note" => Ok(format!("noted for {}", ctx.me.role)),
                "grow" => grow_set(ask, 64),
                kind::REVIEW_TWIN => return OrderAck { seq: o.seq, ok: true, detail: "reviewed".into(), receipt_json: r#"{"complete":"yes"}"#.into() },
                other => Err(format!("order {}: this product does not know the order kind {other:?}", o.seq)),
            };
            match said {
                Ok(detail) => OrderAck { seq: o.seq, ok: true, detail, receipt_json: String::new() },
                Err(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
            }
        }
    }

    #[test]
    fn an_order_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
        let b = Fake::default();
        let mut g = Guardian::default();
        g.queue.push(signed(4, "main", "note", "{}", MONETIZE_SEED));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
        assert_eq!(pulled.acks.len(), 1);
        assert!(pulled.acks[0].ok, "{pulled:?}");
        assert_eq!(pulled.acks[0].detail, "noted for main");
        p.report(&mut g, BTreeMap::from([("serving".into(), "yes".into())]), Some("{}".into()), None, 10_001).unwrap();
        let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
        assert_eq!(r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(4, true)]);
        assert_eq!((r.product.as_deref(), r.build.as_str()), (Some("{}"), "test"));
        assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
        assert!(p.pull(&mut g, &b, Some(&mkey()), 0, 10_002).unwrap().acks.is_empty(), "nothing runs twice");
        assert_eq!(b.ran.borrow().len(), 1);
    }

    /// ★ **The one order check is STRICT Ed25519** (P4c, A28 M11: Gunnar's gunnar-repair copy
    /// always was, `verify_strict`). Under a small-order key (the identity point) the all-zero
    /// `R = identity, S = 0` signature satisfies the plain equation for ANY message; a key fed
    /// wrong must not turn every order into monetize's. A real key still verifies.
    #[test]
    fn a_weak_key_signs_no_order_and_a_real_one_still_does() {
        let mut identity = [0u8; 32];
        identity[0] = 1;
        let weak = VerifyingKey::from_bytes(&identity).expect("the identity point decompresses");
        let mut o = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
        o.signature = { let mut s = vec![0u8; 64]; s[0] = 1; s };
        assert!(matches!(check(&o, &me("main"), &weak, 0, 9_500), Checked::Forged(_)), "a weak key verifies nothing");
        let real = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
        assert_eq!(check(&real, &me("main"), &monetize_key(), 0, 9_500), Checked::Fresh);
    }

    /// ★ **An order dated more than five minutes ahead of this box's clock is refused** (P4c,
    /// A28 M11: Gunnar's order check always refused it). Five minutes of skew still runs.
    #[test]
    fn an_order_dated_ahead_of_the_box_clock_is_refused_by_name() {
        let mut ahead = signed(3, "main", "note", "{}", MONETIZE_SEED);
        ahead.issued_unix_ms = 9_000 + FUTURE_SKEW_MS + 1;
        ahead.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(&ahead)).to_bytes().to_vec();
        match check(&ahead, &me("main"), &monetize_key(), 0, 9_000) {
            Checked::Refused(why) => assert!(why.contains("in the future"), "{why}"),
            other => panic!("{other:?}"),
        }
        let mut skew = ahead.clone();
        skew.issued_unix_ms = 9_000 + FUTURE_SKEW_MS;
        skew.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(&skew)).to_bytes().to_vec();
        assert_eq!(check(&skew, &me("main"), &monetize_key(), 0, 9_000), Checked::Fresh);
    }

    #[test]
    fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
        let b = Fake::default();
        let mut g = Guardian::default();
        g.queue.push(signed(u64::MAX, "main", "note", "{}", [9u8; 32]));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
        assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
        assert_eq!(p.pulled.last_seq, 0);
        assert!(b.ran.borrow().is_empty());
    }

    /// ★ The check before the body: another box's order, an expired one and a replay are acked
    /// not-ok by name and never reach the product; a kind the product does not take is ITS
    /// refusal, acked by name too.
    #[test]
    fn another_boxs_an_expired_or_an_unknown_order_is_refused_by_name() {
        let k = monetize_key();
        let twin = signed(5, "twin", "note", "{}", MONETIZE_SEED);
        assert!(matches!(check(&twin, &me("main"), &k, 0, 10_000), Checked::Refused(w) if w.contains("for a-product/acme/twin")));
        let mut late = signed(5, "main", "note", "{}", MONETIZE_SEED);
        late.not_after_unix_ms = 1_000;
        let late = { let mut l = late.clone(); l.signature = signed_like(&l); l };
        assert!(matches!(check(&late, &me("main"), &k, 0, 61_000), Checked::Refused(w) if w.contains("expired 60 s ago")));
        assert!(matches!(check(&signed(5, "main", "note", "{}", MONETIZE_SEED), &me("main"), &k, 5, 1), Checked::Refused(w) if w.contains("a replay")));
        assert_eq!(check(&signed(6, "main", "note", "{}", MONETIZE_SEED), &me("main"), &k, 5, 1), Checked::Fresh);
        let b = Fake::default();
        let mut g = Guardian::default();
        g.queue.push(twin);
        g.queue.push(signed(7, "main", "reboot_the_moon", "{}", MONETIZE_SEED));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
        assert_eq!(pulled.acks.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(5, false), (7, false)]);
        assert!(pulled.acks[1].detail.contains("does not know the order kind \"reboot_the_moon\""), "{pulled:?}");
        assert_eq!(*b.ran.borrow(), vec![(7, "reboot_the_moon".to_string())], "only the box's own order reaches the product");
    }

    fn signed_like(o: &SignedOrder) -> Vec<u8> {
        ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(o)).to_bytes().to_vec()
    }

    #[test]
    fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        assert_eq!(p.pull(&mut g, &Fake::default(), None, 0, 10_000).unwrap_err().code, "no_key");
        assert!(g.heard.is_empty());
    }

    #[test]
    fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
        assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
        assert!(g.heard[1].box_key.is_empty());
    }

    #[test]
    fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
        g.queue.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
        p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
        p.report(&mut g, BTreeMap::new(), None, Some(r#"{"own":1}"#.into()), 13).unwrap();
        let said: Vec<Option<String>> = g.heard.iter().filter(|s| s.call.call == "report").map(|s| serde_json::from_str::<Report>(&s.call.body).unwrap().twin_review).collect();
        assert!(said[0].as_deref().is_some_and(|r| r.contains("complete")), "{said:?}");
        assert_eq!(said[1], None);
        assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
    }

    /// ★ An order's `Fleet.Ask` goes on the order's own link, signed by this box.
    #[test]
    fn an_order_asks_monetize_on_its_own_link_signed_by_the_box() {
        let mut g = Guardian::default();
        g.queue.push(signed(2, "main", "grow", "{}", MONETIZE_SEED));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000).unwrap();
        assert_eq!(pulled.acks[0].detail, "resized");
        let asked = g.heard.iter().find(|s| s.call.call == "ask").expect("asked");
        let a: Ask = serde_json::from_str(&asked.call.body).unwrap();
        assert_eq!((a.kind.as_str(), a.body["gib"].as_u64()), ("grow_set", Some(64)));
        let vk = VerifyingKey::from_bytes(&p.box_public()).unwrap();
        monetize_embed::fleet::verify_fleet_call(&asked.call, &asked.signature, &vk, 10_000).expect("the ask is the box's");
    }

    #[test]
    fn the_puller_remembers_what_it_took_across_a_restart() {
        let dir = tempfile::tempdir().unwrap();
        let mut g = Guardian::default();
        g.queue.push(signed(6, "main", "note", "{}", MONETIZE_SEED));
        let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
        let key = p.box_public();
        p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000).unwrap();
        drop(p);
        let p = Puller::at(dir.path(), me("main"), "test").unwrap();
        assert_eq!((p.box_public(), p.pulled.last_seq, p.pulled.pending.len()), (key, 6, 1));
    }

    /// ★ MONETIZE-FIX: **a product order's file is pulled with a `fetch` ask**, naming the file;
    /// bytes of another size than the order names are refused, never handed on.
    #[test]
    fn a_product_orders_file_is_pulled_by_a_fetch_ask_and_checked() {
        struct Shop(Vec<u8>, Vec<Ask>);
        impl Asker for Shop {
            fn ask(&mut self, _: &Ask) -> Result<Asked, Fault> {
                unreachable!("a fetch asks for its bytes")
            }
            fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
                self.1.push(a.clone());
                Ok((Asked::default(), self.0.clone()))
            }
        }
        let f = monetize_embed::fleet::Fetch { id: "ab".repeat(32), bytes: 3, param: "bundle".into() };
        let mut shop = Shop(b"abc".to_vec(), Vec::new());
        assert_eq!(fetch(&mut shop, &f).unwrap(), b"abc");
        assert_eq!((shop.1[0].kind.as_str(), shop.1[0].body["id"].as_str()), (ask::FETCH, Some(f.id.as_str())));
        let mut short = Shop(b"ab".to_vec(), Vec::new());
        assert!(fetch(&mut short, &f).unwrap_err().contains("names 3"));
    }
}