open-standards/eoc/eoc-rs/crates/eoc-cascade/src/lib.rs

//! EOC cascade — the four-stage memoizing pipeline.
//!
//! `Cascade::resolve` walks `cache → kv → graph → neural` in order. The
//! first stage to return `Some(Response)` wins.
//!
//! # What a resolve is charged
//!
//! The rule, as the owner set it: a live counter's figure is `Measured`, and
//! nothing that was not read from one may be. The cascade keeps it by using
//! its meter only where the meter can resolve the work.
//!
//! **A resolve the neural stage answers is read.** The meter is read once the
//! three deterministic tiers have missed, and again after the neural stage
//! and the memoizing write: two reads, one span. That span is combined with
//! the cost the stage reported (`cascade_cost`), and the total is `Measured`
//! only when every micro-joule of it was read from a live counter. The three
//! misses are spent before the opening read, so they are not in the figure.
//! They are small against what the span holds of the meter itself, its
//! closing read's own wake-up: on an M3 Ultra the three together took 220 ns
//! and 0.95 µJ, and a span over no work read 12 to 29 µJ. So the measured
//! figure is the neural stage's energy, the memoizing write's and the meter's
//! own closing cost: over the neural stage's energy, never under it.
//!
//! **A resolve a deterministic tier answers (cache, kv, graph) reads no
//! meter**, and is never `Measured`. A read of a counter costs more than the
//! hit it would bracket, and what such a span reads is the read. On an M3
//! Ultra (macOS 27.0.1, an optimized build), with this process's own energy
//! counter, whose every read first sees the reading thread leave its core: a
//! resolve answered by the cache took 164 µs at the median with a span around
//! it, 0.38 µs without, and the span read 13 to 28 µJ, 14 µJ on average over
//! 10,000 hits, as a span around a neural stage that does nothing did (14 to
//! 19 µJ on average), where the hit itself is calibrated at 1.2 to 1.7 µJ.
//! That figure is the meter's own cost, not the hit's; tagged `Measured` and
//! sealed in the hit's receipt it was a true reading booked to the wrong
//! work. So a hit is charged a figure calibrated on this host instead
//! ([`HitCalibration`]): the first time a tier answers, the cascade repeats
//! that lookup back to back inside one span, enough times that the span is
//! large against the meter's own cost, and keeps the span's reading per hit.
//! Every later hit on that tier is charged that figure, with no read. It is
//! `Estimated`: it was read by this counter, on this host, but over a batch,
//! not over this resolve. A per-operation figure for work shorter than the
//! meter can resolve is sound only as the mean of many operations in one
//! span, and a mean is not a reading of any one of them.
//!
//! With no live counter — the cascade was given none, a read failed, or the
//! counter did not advance — there is no calibration and no span, and every
//! resolve keeps the cost its stage reported, figure and tag.

#![forbid(unsafe_code)]

use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use eoc_cache::{LruCache, Stage};
use eoc_core::{JouleCost, JouleSource, Query, Response};
use eoc_graph::GraphStage;
use eoc_kv::KvStage;
use eoc_meter::{JouleCounter, StubCounter};
use eoc_neural::NeuralStage;

/// How many times the meter's own cost a calibration span must read before
/// its reading is taken as the batch's: the meter's share of the per-hit
/// figure is then a fiftieth at most.
const SPAN_TO_EMPTY: u64 = 50;

/// The least a calibration span must read, in micro-joules, whatever the
/// meter's own cost. The per-hit figure is the span's whole micro-joules over
/// the batch, so a span of 200 µJ keeps the counter's own rounding under
/// half a percent of it.
const SMALLEST_SPAN_MICROJOULES: u64 = 200;

/// How much larger each calibration batch is than the one before it.
const BATCH_GROWTH: u64 = 8;

/// The wall time the second batch aims at: the first batch is one lookup,
/// timed, and the second is as many as fill this, so a lookup of well under
/// a microsecond does not climb to a readable batch eight at a time.
const SECOND_BATCH: Duration = Duration::from_micros(100);

/// The wall time after which a calibration starts no further batch. It keeps
/// the largest span it read by then, whatever that span's size against the
/// meter's own cost ([`HitCalibration`] reports both).
const CALIBRATION_CAP: Duration = Duration::from_millis(40);

/// How long a counter may hold over back-to-back reads before a calibration
/// takes it for one that is not live.
const HELD_AFTER: Duration = Duration::from_millis(2);

/// The largest batch a calibration runs, whatever the clock says.
const LARGEST_BATCH: u64 = 1 << 22;

/// What one hit on a deterministic tier costs on this host: the cascade's
/// meter read over a batch of that tier's lookup, divided by the batch.
///
/// Each lookup in the batch is what a resolve answered by that tier does once
/// the cheaper tiers have missed: the tier's own lookup of the query, and the
/// minting of the response the cascade returns for it. The misses of the
/// cheaper tiers before it and the memoizing write after it are not in the
/// figure. Nor is the meter's own cost taken out of it: the span holds its
/// closing read, as every span does, so the figure is over the batch's energy
/// by at most `empty_span_microjoules` in `span_microjoules`.
///
/// The figure is that of the query the tier first answered (or the one
/// [`Cascade::calibrate`] was given), charged to every hit on the tier: a
/// lookup of another key, or a longer scan, costs what it costs, and is
/// charged this. What the calibration itself spent, its spans and its
/// reads, is charged to no resolve.
///
/// The counter reads its whole domain over the batch. On this process's own
/// counter that is every thread of the process, so work on other threads
/// during the batch is in the figure; on a whole-machine counter, so is the
/// machine's. Calibrate on a quiet process, read the record, and call
/// [`Cascade::calibrate`] again if it was not.
///
/// The figure is also that of the core the batch ran on, and of this build.
/// On an M3 Ultra (macOS 27.0.1, optimized, this process's counter): 1.2 to
/// 1.7 µJ a cache hit, 2.6 to 3.4 µJ a KV hit and 2.9 to 3.4 µJ a graph hit
/// on a performance core, from 224 to 4,032 lookups in a span of 0.7 to
/// 5.9 mJ, 51 to 351 times the 12 to 29 µJ a span over no work read,
/// calibrated in 0.85 to 2.4 ms; a batch that landed on an efficiency core
/// read a fifth of that (253 nJ a cache hit). An unoptimized build's cache
/// hit read 22 µJ, most of it the receipt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HitCalibration {
    /// The span's reading per lookup, in nanojoules, to the nearest one.
    pub per_hit_nanojoules: u64,
    /// How many times the lookup ran inside the span.
    pub batch: u64,
    /// What the meter read over the batch, in micro-joules: one span, two
    /// reads.
    pub span_microjoules: u64,
    /// What the meter read over no work at all, in micro-joules: the first
    /// advance it showed across back-to-back reads. It is the meter's own
    /// cost in a span, and the batch was grown until its span read fifty
    /// times this, or until the time allowed ran out.
    pub empty_span_microjoules: u64,
    /// The wall time the calibration took, its reads and every batch.
    pub took: Duration,
}

/// Why a tier has no [`HitCalibration`]. Its hits keep the cost the stage
/// reported.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NoCalibration {
    /// The neural stage is not calibrated: a resolve it answers is read.
    NotADeterministicTier,
    /// The tier has not answered yet.
    NotYet,
    /// A resolve is calibrating the tier now.
    InProgress,
    /// Switched off for this tier ([`Cascade::without_hit_calibration`]).
    Disabled,
    /// The cascade was given no meter: it holds the stub.
    NoMeter,
    /// This target has no monotonic clock to bound the calibration by.
    NoClock,
    /// A read of the meter failed.
    ReadFailed,
    /// The counter did not advance: over back-to-back reads, or over any
    /// batch.
    CounterHeld,
    /// The lookup stopped answering while it was repeated.
    LookupStoppedAnswering,
}

impl std::fmt::Display for NoCalibration {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::NotADeterministicTier => "the neural stage is read, not calibrated",
            Self::NotYet => "the tier has not answered yet",
            Self::InProgress => "the tier is being calibrated",
            Self::Disabled => "calibration is switched off for the tier",
            Self::NoMeter => "the cascade was given no meter",
            Self::NoClock => "no monotonic clock on this target",
            Self::ReadFailed => "a read of the meter failed",
            Self::CounterHeld => "the counter did not advance",
            Self::LookupStoppedAnswering => "the lookup stopped answering while it was repeated",
        })
    }
}

/// A deterministic tier: one whose hits are calibrated, not read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Tier {
    Cache,
    Kv,
    Graph,
}

impl Tier {
    fn of(stage: eoc_core::Stage) -> Option<Self> {
        match stage {
            eoc_core::Stage::Cache => Some(Self::Cache),
            eoc_core::Stage::Kv => Some(Self::Kv),
            eoc_core::Stage::Graph => Some(Self::Graph),
            eoc_core::Stage::Neural => None,
        }
    }

    fn stage(self) -> eoc_core::Stage {
        match self {
            Self::Cache => eoc_core::Stage::Cache,
            Self::Kv => eoc_core::Stage::Kv,
            Self::Graph => eoc_core::Stage::Graph,
        }
    }
}

/// What the cascade holds about one deterministic tier's hits.
#[derive(Debug)]
struct TierHits {
    calibration: Result<HitCalibration, NoCalibration>,
    /// The nanojoules calibrated hits have accrued that no charge has
    /// carried yet: under 1,000 after every charge.
    carry_nanojoules: u64,
}

impl TierHits {
    fn new() -> Self {
        Self { calibration: Err(NoCalibration::NotYet), carry_nanojoules: 0 }
    }

    /// The whole micro-joules one more hit is charged. A hit accrues the
    /// tier's per-hit nanojoules; what does not make a micro-joule is carried
    /// to the next hit. So the micro-joules charged over any run of hits sum
    /// to the nanojoules those hits accrued, to within the one micro-joule
    /// still carried — never a round-up of each hit, and never a string of
    /// zeros for a figure under a micro-joule. Nothing where the tier has no
    /// calibration.
    fn charge(&mut self) -> u64 {
        let Ok(calibration) = self.calibration else {
            return 0;
        };
        let accrued = self.carry_nanojoules.saturating_add(calibration.per_hit_nanojoules);
        self.carry_nanojoules = accrued % 1_000;
        accrued / 1_000
    }
}

/// A monotonic clock: the time since some fixed instant, or `None` where the
/// target has none.
type Clock = Arc<dyn Fn() -> Option<Duration> + Send + Sync>;

/// The time since this process first asked.
#[cfg(not(target_arch = "wasm32"))]
fn monotonic() -> Option<Duration> {
    static FIRST: std::sync::OnceLock<std::time::Instant> = std::sync::OnceLock::new();
    Some(FIRST.get_or_init(std::time::Instant::now).elapsed())
}

/// `wasm32-unknown-unknown` has no clock (`Instant::now` panics there), and
/// no energy counter either: nothing is calibrated.
#[cfg(target_arch = "wasm32")]
fn monotonic() -> Option<Duration> {
    None
}

/// The four-stage cascade.
pub struct Cascade {
    cache: Arc<LruCache>,
    kv: Arc<KvStage>,
    graph: Arc<GraphStage>,
    neural: Arc<NeuralStage>,
    meter: Arc<dyn JouleCounter>,
    /// Whether a meter was given ([`Self::with_meter`]): without one, nothing
    /// is calibrated and no read is spent finding the stub does not move.
    metered: bool,
    /// Whether to memoize neural/graph/kv answers back into the cache.
    memoize: bool,
    /// What a hit costs on each deterministic tier: cache, kv, graph.
    hits: [Mutex<TierHits>; 3],
    clock: Clock,
}

impl Cascade {
    /// Construct a cascade. The meter defaults to `StubCounter`.
    pub fn new(
        cache: Arc<LruCache>,
        kv: Arc<KvStage>,
        graph: Arc<GraphStage>,
        neural: Arc<NeuralStage>,
    ) -> Self {
        Self {
            cache,
            kv,
            graph,
            neural,
            meter: Arc::new(StubCounter),
            metered: false,
            memoize: true,
            hits: [Mutex::new(TierHits::new()), Mutex::new(TierHits::new()), Mutex::new(TierHits::new())],
            clock: Arc::new(monotonic),
        }
    }

    /// Swap in a different joule counter.
    ///
    /// The counter is read twice around the neural stage of a resolve that
    /// reaches it, and around the batches that calibrate each deterministic
    /// tier the first time it answers ([`HitCalibration`]); never around a
    /// hit.
    pub fn with_meter(mut self, meter: Arc<dyn JouleCounter>) -> Self {
        self.meter = meter;
        self.metered = true;
        self
    }

    /// Disable memoization (each query re-walks the full cascade).
    pub fn without_memoization(mut self) -> Self {
        self.memoize = false;
        self
    }

    /// Never calibrate `stage`'s hits: they keep the cost the stage reports.
    ///
    /// A calibration repeats the tier's lookup of one query, up to the time
    /// allowed (tens of milliseconds). That is safe for the stages this
    /// workspace ships: the cache's lookup only refreshes the entry's
    /// recency, which the hit before it already did; the KV stage's reads its
    /// backend ([`eoc_kv::KvBackend::get`], and `embeddings` for a similarity
    /// match); the graph stage's scans its triples. A KV backend whose `get`
    /// is not a plain read — one that bills, rate-limits or counts each call
    /// — should not be repeated, and its tier is switched off here. No effect
    /// for the neural stage, which is never calibrated.
    pub fn without_hit_calibration(self, stage: eoc_core::Stage) -> Self {
        if let Some(tier) = Tier::of(stage) {
            self.hits(tier).calibration = Err(NoCalibration::Disabled);
        }
        self
    }

    /// Borrow the cache (for warmup / wiring).
    pub fn cache(&self) -> &LruCache {
        &self.cache
    }
    /// Borrow the kv stage.
    pub fn kv(&self) -> &KvStage {
        &self.kv
    }
    /// Borrow the graph stage.
    pub fn graph(&self) -> &GraphStage {
        &self.graph
    }
    /// Borrow the neural stage.
    pub fn neural(&self) -> &NeuralStage {
        &self.neural
    }
    /// Borrow the joule meter.
    pub fn meter(&self) -> &dyn JouleCounter {
        self.meter.as_ref()
    }

    /// What a hit on `stage` is charged on this host, or why it is charged
    /// nothing beyond the stage's own cost.
    pub fn calibration(&self, stage: eoc_core::Stage) -> Result<HitCalibration, NoCalibration> {
        match Tier::of(stage) {
            Some(tier) => self.hits(tier).calibration,
            None => Err(NoCalibration::NotADeterministicTier),
        }
    }

    /// Calibrate, now, the deterministic tier that answers `q`, whether or
    /// not it was calibrated before: the figure replaces the one held, and
    /// the next hit on that tier is charged it. For a host whose state has
    /// changed since the first hit, or to calibrate before the first resolve
    /// rather than inside it.
    ///
    /// Returns the stage that answered and its calibration, or why there is
    /// none; `None` when no deterministic tier answers `q`. Nothing is
    /// memoized and no resolve is charged. A tier switched off by
    /// [`Self::without_hit_calibration`] stays off.
    pub async fn calibrate(&self, q: &Query) -> Option<(eoc_core::Stage, Result<HitCalibration, NoCalibration>)> {
        let mut answering = None;
        for tier in [Tier::Cache, Tier::Kv, Tier::Graph] {
            if self.lookup(tier, q).await.is_some() {
                answering = Some(tier);
                break;
            }
        }
        let tier = answering?;
        let held = self.hits(tier).calibration;
        if held == Err(NoCalibration::Disabled) {
            return Some((tier.stage(), held));
        }
        let calibration = self.calibrate_tier(tier, q).await;
        self.hits(tier).calibration = calibration;
        Some((tier.stage(), calibration))
    }

    /// Resolve a query through the cascade.
    ///
    /// On a miss at one stage, falls through to the next. The neural
    /// stage always answers, so this function always returns `Some` if
    /// the neural stage's backend is well-behaved. We still return
    /// `Option<Response>` for symmetry with the `Stage` trait.
    ///
    /// The meter is read only when the walk reaches the neural stage; a
    /// resolve a deterministic tier answers reads none (see the crate
    /// documentation), apart from the one that first calibrates its tier.
    pub async fn resolve(&self, q: Query) -> Response {
        // Walk the deterministic tiers in order. None of them is read.
        if let Some(r) = self.cache.try_resolve(&q).await {
            return self.hit(Tier::Cache, &q, r).await;
        }
        if let Some(r) = self.kv.try_resolve(&q).await {
            self.maybe_memoize(&q, &r);
            return self.hit(Tier::Kv, &q, r).await;
        }
        if let Some(r) = self.graph.try_resolve(&q).await {
            self.maybe_memoize(&q, &r);
            return self.hit(Tier::Graph, &q, r).await;
        }

        // All three missed: only now is the meter read, so the span is the
        // neural stage's, and the misses, spent before it, are not in it.
        let start = self.meter.read_microjoules().ok();
        // Neural always answers.
        let r = self
            .neural
            .try_resolve(&q)
            .await
            .expect("neural stage must always return Some");
        self.maybe_memoize(&q, &r);

        // Re-derive the response with the cascade-attributed joule cost: the
        // stage's own cost, combined with the meter's span when it is live.
        attach_cascade_cost(r, start, self.meter.as_ref())
    }

    fn maybe_memoize(&self, _q: &Query, r: &Response) {
        if self.memoize {
            // Memoize as a *cache* entry so future hits report `Stage::Cache`
            // and `JouleCost::zero()`, an estimated zero: no counter is read
            // for a hit. The original receipt is no longer valid (different
            // stage + cost) — we mint a new one.
            let cached =
                Response::new(r.query_id, r.payload.clone(), eoc_core::Stage::Cache, JouleCost::zero());
            self.cache.insert_response(r.query_id, cached);
        }
    }

    fn hits(&self, tier: Tier) -> MutexGuard<'_, TierHits> {
        // The state is two plain figures, whole at every point a holder
        // could have panicked.
        self.hits[tier as usize].lock().unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// `tier`'s own lookup of `q`.
    async fn lookup(&self, tier: Tier, q: &Query) -> Option<Response> {
        match tier {
            Tier::Cache => self.cache.try_resolve(q).await,
            Tier::Kv => self.kv.try_resolve(q).await,
            Tier::Graph => self.graph.try_resolve(q).await,
        }
    }

    /// The response for a resolve `tier` answered with `r`: charged the
    /// tier's calibrated share on top of the stage's own cost, an estimate,
    /// with no read of the meter. The first hit on a tier calibrates it.
    async fn hit(&self, tier: Tier, q: &Query, r: Response) -> Response {
        let first = {
            let mut hits = self.hits(tier);
            let first = hits.calibration == Err(NoCalibration::NotYet);
            if first {
                // Claimed before the lock is dropped, so a resolve on
                // another thread does not calibrate the tier a second time:
                // it is charged the stage's own cost until this one is done.
                hits.calibration = Err(NoCalibration::InProgress);
            }
            first
        };
        if first {
            let calibration = self.calibrate_tier(tier, q).await;
            self.hits(tier).calibration = calibration;
        }
        let share = self.hits(tier).charge();
        // Rebuild the response so the receipt covers the final joule cost.
        Response::new(r.query_id, r.payload, r.stage, hit_cost(r.joule_cost, share))
    }

    /// What one hit on `tier` costs on this host, from one span of the meter
    /// around a batch of `tier`'s lookup of `q`.
    ///
    /// First the meter's own cost: back-to-back reads until the counter
    /// moves. A counter that holds for [`HELD_AFTER`] is not live. Then the
    /// batch, grown until its span reads [`SPAN_TO_EMPTY`] times that (and
    /// [`SMALLEST_SPAN_MICROJOULES`] at least), each batch inside one span
    /// of its own; no further batch is started once [`CALIBRATION_CAP`] of
    /// wall time has gone, or once another would not fit in what is left of
    /// it, and the largest span read by then is kept. The stage's lookups
    /// here do not yield, so the batch runs on the calling thread between
    /// the span's two reads.
    async fn calibrate_tier(&self, tier: Tier, q: &Query) -> Result<HitCalibration, NoCalibration> {
        if !self.metered {
            return Err(NoCalibration::NoMeter);
        }
        let clock = self.clock.as_ref();
        let started = clock().ok_or(NoCalibration::NoClock)?;
        let since = |then: Duration| clock().map_or(Duration::MAX, |now| now.saturating_sub(then));
        let meter = self.meter.as_ref();
        let read = || meter.read_microjoules().map_err(|_| NoCalibration::ReadFailed);

        let base = read()?;
        let empty = loop {
            let now = read()?;
            if now > base {
                break now - base;
            }
            if since(started) >= HELD_AFTER {
                return Err(NoCalibration::CounterHeld);
            }
        };
        let enough = empty.saturating_mul(SPAN_TO_EMPTY).max(SMALLEST_SPAN_MICROJOULES);

        let mut batch = 1u64;
        let mut largest = None;
        loop {
            let open = read()?;
            let began = clock().ok_or(NoCalibration::NoClock)?;
            for _ in 0..batch {
                let Some(r) = self.lookup(tier, q).await else {
                    return Err(NoCalibration::LookupStoppedAnswering);
                };
                // The response the cascade mints for a hit, as `hit` does.
                std::hint::black_box(Response::new(r.query_id, r.payload, r.stage, r.joule_cost));
            }
            let lookups = since(began);
            let close = read()?;
            if close > open {
                largest = Some((batch, close - open));
                if close - open >= enough {
                    break;
                }
            }
            let each = (lookups.as_nanos() / u128::from(batch)).max(1);
            let fit = CALIBRATION_CAP.saturating_sub(since(started)).as_nanos() / each;
            let second = u64::try_from(SECOND_BATCH.as_nanos() / each).unwrap_or(u64::MAX);
            let wanted = batch.saturating_mul(BATCH_GROWTH).max(second);
            let next = wanted.min(u64::try_from(fit).unwrap_or(u64::MAX)).min(LARGEST_BATCH);
            if next <= batch {
                break;
            }
            batch = next;
        }
        let (batch, span) = largest.ok_or(NoCalibration::CounterHeld)?;
        let per_hit = (u128::from(span) * 1_000 + u128::from(batch) / 2) / u128::from(batch);
        Ok(HitCalibration {
            per_hit_nanojoules: u64::try_from(per_hit).unwrap_or(u64::MAX),
            batch,
            span_microjoules: span,
            empty_span_microjoules: empty,
            took: since(started),
        })
    }
}

/// The response for a resolve the neural stage answered: its cost combined
/// with the span the cascade's meter read around the stage, from `start`,
/// read before it, to a second read taken here. No second read is taken
/// where the first failed: no span could come of it.
fn attach_cascade_cost(
    response: Response,
    start: Option<u64>,
    meter: &dyn JouleCounter,
) -> Response {
    let end = start.and_then(|_| meter.read_microjoules().ok());
    let joule_cost = cascade_cost(response.joule_cost, live_span(start, end));
    // Rebuild the response so the receipt covers the final joule cost.
    Response::new(response.query_id, response.payload, response.stage, joule_cost)
}

/// What the cascade's meter read around the neural stage of one resolve: the
/// delta between its two reads, only when both succeeded and the counter
/// advanced. A counter that read but did not move — the stub, a frozen rail
/// — measured nothing, and a failed read leaves no span; neither is a figure.
///
/// The first read is taken after the cache, the KV stage and the graph have
/// missed, and the second after the neural stage and the memoizing write. So
/// the span holds the neural stage, that write, and the meter's own closing
/// cost (on this process's counter, the wake-up of the read that closes it);
/// it does not hold the three misses, which were spent before it opened.
fn live_span(start: Option<u64>, end: Option<u64>) -> Option<u64> {
    match (start, end) {
        (Some(s), Some(e)) if e > s => Some(e - s),
        _ => None,
    }
}

/// The joule cost of one resolve the neural stage answered, from the stage's
/// own cost and the cascade meter's live span around the stage, if there was
/// one.
///
/// It is `Measured` only when every micro-joule of it was read from a live
/// counter; any micro-joule that was estimated makes the whole an estimate.
///
/// - No live span: the cascade read nothing, adds nothing, and the stage's
///   cost stands as the stage reported it.
/// - A live span and a measured stage: the stage read a counter on this host
///   during the resolve, so inside the span. On the cascade's own counter the
///   span already holds that reading, and adding the two would count it
///   twice. The cost is the larger of the two, measured: exact when the stage
///   read the cascade's counter, a lower bound if it read another.
/// - A live span and an estimated stage: the estimate is energy no counter
///   read, possibly spent elsewhere (a remote backend), so it is added to the
///   span and the total is an estimate. An estimated zero, which is what a
///   stage reports when it has no figure of its own, adds nothing: the cost
///   is the span's reading alone, and measured.
///
/// A measured figure is what the counter read from the opening read to the
/// closing one: the neural stage, the memoizing write and the meter's own
/// closing cost. It leaves out the three deterministic misses before the
/// opening read, which are smaller than that closing cost, so it is not
/// under the neural stage's energy.
///
/// This rule is never applied to a resolve a deterministic tier answered.
/// It once was, and its last case then read: an estimated zero on a live
/// span, a cache hit, is the span's reading, measured. With a counter whose
/// read costs more than the hit, that reading was the meter's own cost (on
/// an M3 Ultra, 14 µJ on average over 10,000 cache hits, each bracketed by
/// two reads taking 164 µs between them, where the hit is 0.38 µs and about
/// 1.5 µJ), sealed as the hit's. A hit is charged by [`hit_cost`] instead,
/// and no meter is read for it.
fn cascade_cost(stage: JouleCost, span: Option<u64>) -> JouleCost {
    let Some(span) = span else {
        return stage;
    };
    match stage.source {
        JouleSource::Measured => JouleCost::measured(stage.microjoules.max(span)),
        JouleSource::Estimated if stage.microjoules == 0 => JouleCost::measured(span),
        JouleSource::Estimated => JouleCost::estimated(stage.microjoules.saturating_add(span)),
    }
}

/// The joule cost of one resolve a deterministic tier answered: the cost the
/// stage reported plus the tier's calibrated share for this hit, in whole
/// micro-joules (nothing, where the tier has no calibration). Always an
/// estimate.
///
/// No counter was read for this resolve, so none of it is measured. Not the
/// calibrated share: it was read on this host, by the cascade's own counter,
/// but over a batch of the lookup, and a mean is not a reading of this hit.
/// And not a figure the stage's response carries tagged measured: a cached
/// response replays what was read for an earlier resolve, not this one.
///
/// The stage's own figure is added, not replaced, for the reason
/// [`cascade_cost`] adds an estimate to a span: it may be energy the counter
/// never saw. Where it is the stage's guess at the very lookup the
/// calibration read (the reference KV and graph stages report a fixed figure
/// for theirs), the two overlap; the cascade cannot tell, and an estimate
/// that is over says so by its tag.
fn hit_cost(stage: JouleCost, share_microjoules: u64) -> JouleCost {
    JouleCost::estimated(stage.microjoules.saturating_add(share_microjoules))
}

#[cfg(test)]
mod tests {
    use super::*;
    use eoc_cache::LruCache;
    use eoc_core::Stage as StageKind;
    use eoc_graph::{GraphStage, Triple};
    use eoc_kv::{KvBackend, KvStage, MemoryKvBackend};
    use eoc_neural::{EchoBackend, NeuralStage};
    use std::sync::atomic::{AtomicU64, Ordering};

    impl Cascade {
        /// Bound calibrations by a scripted clock instead of this host's.
        fn with_clock(mut self, clock: Clock) -> Self {
            self.clock = clock;
            self
        }
    }

    fn fixture() -> Cascade {
        let cache = Arc::new(LruCache::new(64));
        let kv_backend = Box::new(MemoryKvBackend::new());
        kv_backend.put("known key", b"known value".to_vec());
        let kv = Arc::new(KvStage::new(kv_backend));
        let graph = Arc::new(GraphStage::new());
        graph.insert(Triple::new("Mars", "fourth planet from", "the Sun"));
        let neural = Arc::new(NeuralStage::new(Box::new(
            EchoBackend::new().with_cost(50_000_000),
        )));
        Cascade::new(cache, kv, graph, neural)
    }

    /// A query each deterministic tier of [`fixture`] answers, once the
    /// cache has been handed its own.
    fn answered_by(cascade: &Cascade, stage: StageKind) -> Query {
        match stage {
            StageKind::Cache => {
                let q = Query::new("seen before");
                cascade.cache().insert(&q, "answer");
                q
            }
            StageKind::Kv => Query::new("known key"),
            StageKind::Graph => Query::new("Mars is the fourth planet from what?"),
            StageKind::Neural => Query::new("totally novel question"),
        }
    }

    /// What each deterministic stage of [`fixture`] reports for a hit.
    fn own_cost(stage: StageKind) -> u64 {
        match stage {
            StageKind::Cache => 0,
            StageKind::Kv => 50,
            StageKind::Graph => 500,
            StageKind::Neural => 50_000_000,
        }
    }

    const TIERS: [StageKind; 3] = [StageKind::Cache, StageKind::Kv, StageKind::Graph];

    #[tokio::test]
    async fn cascade_uses_kv_when_it_can() {
        let cascade = fixture();
        let r = cascade.resolve(Query::new("known key")).await;
        assert_eq!(r.stage, StageKind::Kv);
        assert_eq!(r.payload, "known value");
    }

    #[tokio::test]
    async fn cascade_uses_graph_when_kv_misses() {
        let cascade = fixture();
        let r = cascade
            .resolve(Query::new("Mars is the fourth planet from what?"))
            .await;
        assert_eq!(r.stage, StageKind::Graph);
        assert_eq!(r.payload, "the Sun");
    }

    #[tokio::test]
    async fn cascade_falls_through_to_neural() {
        let cascade = fixture();
        let r = cascade.resolve(Query::new("totally novel question")).await;
        assert_eq!(r.stage, StageKind::Neural);
        assert!(r.payload.contains("totally novel question"));
        assert!(r.joule_cost.microjoules > 0);
    }

    #[tokio::test]
    async fn cascade_memoizes_into_cache() {
        let cascade = fixture();
        let q = Query::new("totally novel question");
        let r1 = cascade.resolve(q.clone()).await;
        assert_eq!(r1.stage, StageKind::Neural);
        // Same query the second time round should hit the cache.
        let r2 = cascade.resolve(q).await;
        assert_eq!(r2.stage, StageKind::Cache);
    }

    /// A counter that reads the scripted values in turn, then the last one
    /// again; `None` is a failed read. One instance can be shared by the
    /// cascade and a stage, as one host counter would be. It counts its
    /// reads, and keeps a clock that only its reads move, a millisecond each,
    /// so a calibration against it ends by the script and not by this host's
    /// time. A lookup takes none of that clock's time: a script must end a
    /// calibration by its reads — the span it scripts is enough, the counter
    /// holds from the start, or a read fails.
    struct Scripted {
        reads: std::sync::Mutex<std::collections::VecDeque<Option<u64>>>,
        taken: AtomicU64,
    }

    impl Scripted {
        fn new(reads: &[Option<u64>]) -> Arc<Self> {
            Arc::new(Self {
                reads: std::sync::Mutex::new(reads.iter().copied().collect()),
                taken: AtomicU64::new(0),
            })
        }

        /// How many reads have been taken.
        fn taken(&self) -> u64 {
            self.taken.load(Ordering::SeqCst)
        }

        fn clock(self: &Arc<Self>) -> Clock {
            let scripted = self.clone();
            Arc::new(move || Some(Duration::from_millis(scripted.taken())))
        }
    }

    impl JouleCounter for Scripted {
        fn read_microjoules(&self) -> eoc_core::Result<u64> {
            self.taken.fetch_add(1, Ordering::SeqCst);
            let mut r = self.reads.lock().unwrap();
            let next = if r.len() > 1 {
                r.pop_front().unwrap()
            } else {
                r[0]
            };
            next.ok_or(eoc_core::Error::NoCounter)
        }
        fn name(&self) -> &'static str {
            "scripted"
        }
    }

    /// A script that calibrates a tier in four reads: the counter moves 10 µJ
    /// over no work, so a span of 500 µJ is enough, and the first batch, one
    /// lookup, reads 700. Every hit is then charged 700 µJ.
    const CALIBRATES_AT_700: [Option<u64>; 4] = [Some(1_000), Some(1_010), Some(2_000), Some(2_700)];

    /// A neural backend that meters itself the way the local backends do: it
    /// reads its counter before and after, and reports the delta as measured
    /// when the counter advanced.
    struct SelfMetered {
        meter: Arc<dyn JouleCounter>,
    }

    #[async_trait::async_trait]
    impl eoc_neural::NeuralBackend for SelfMetered {
        async fn infer(&self, q: &Query) -> Response {
            let start = self.meter.read_microjoules().ok();
            let end = self.meter.read_microjoules().ok();
            let cost = match (start, end) {
                (Some(s), Some(e)) if e > s => JouleCost::measured(e - s),
                _ => JouleCost::estimated(1),
            };
            Response::new(q.id, "metered".to_string(), StageKind::Neural, cost)
        }
    }

    /// [`fixture`]'s stages around a given neural backend, metered by a
    /// scripted counter and bounded by its clock.
    fn metered(meter: &Arc<Scripted>, neural: Box<dyn eoc_neural::NeuralBackend>) -> Cascade {
        let kv_backend = Box::new(MemoryKvBackend::new());
        kv_backend.put("known key", b"known value".to_vec());
        let graph = GraphStage::new();
        graph.insert(Triple::new("Mars", "fourth planet from", "the Sun"));
        Cascade::new(
            Arc::new(LruCache::new(8)),
            Arc::new(KvStage::new(kv_backend)),
            Arc::new(graph),
            Arc::new(NeuralStage::new(neural)),
        )
        .with_meter(meter.clone())
        .with_clock(meter.clock())
    }

    /// No meter is read for a cache hit, so it claims no measurement: not a
    /// memoized answer, not an inserted one.
    #[tokio::test]
    async fn a_cache_hit_claims_no_measurement() {
        let cascade = fixture();
        let q = Query::new("totally novel question");
        cascade.resolve(q.clone()).await;
        let memoized = cascade.resolve(q).await;
        assert_eq!(memoized.stage, StageKind::Cache);
        assert_eq!(memoized.joule_cost, JouleCost::estimated(0));

        let q = Query::new("inserted by hand");
        cascade.cache().insert(&q, "answer");
        let inserted = cascade.resolve(q).await;
        assert_eq!(inserted.stage, StageKind::Cache);
        assert_eq!(inserted.joule_cost, JouleCost::estimated(0));
    }

    /// A live counter advanced 500 µJ over the neural stage, and the stage
    /// reported a 50 J estimate. The total holds an estimate, so it is one.
    #[tokio::test]
    async fn an_estimate_plus_a_live_span_is_an_estimate() {
        let meter = Scripted::new(&[Some(1_000), Some(1_500)]);
        let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(50_000_000)));
        let r = cascade.resolve(Query::new("totally novel question")).await;
        assert_eq!(r.stage, StageKind::Neural);
        assert_eq!(r.joule_cost, JouleCost::estimated(50_000_500));
        assert!(r.receipt.verify(&r.query_id, &r.stage, &r.joule_cost, &r.payload));
    }

    /// The stage metered itself on the cascade's own counter: it read 300 µJ
    /// (1_100 -> 1_400) inside the cascade's 500 µJ span (1_000 -> 1_500). The
    /// span already holds those 300 µJ, so the resolve cost 500 µJ, measured,
    /// never 800.
    #[tokio::test]
    async fn a_stage_metered_on_the_cascade_counter_is_counted_once() {
        let meter = Scripted::new(&[Some(1_000), Some(1_100), Some(1_400), Some(1_500)]);
        let cascade = metered(&meter, Box::new(SelfMetered { meter: meter.clone() }));
        let r = cascade.resolve(Query::new("totally novel question")).await;
        assert_eq!(r.stage, StageKind::Neural);
        assert_eq!(r.joule_cost, JouleCost::measured(500));
    }

    /// A counter that reads but does not move, or fails a read, measured
    /// nothing: the stage's own cost stands, figure and tag. A failed first
    /// read is not followed by a second.
    #[tokio::test]
    async fn a_span_that_is_not_live_adds_nothing() {
        for (reads, taken) in [(vec![Some(7_000)], 2), (vec![None, Some(9_000)], 1), (vec![Some(9_000), None], 2)] {
            let meter = Scripted::new(&reads);
            let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(42)));
            let r = cascade.resolve(Query::new("totally novel question")).await;
            assert_eq!(r.joule_cost, JouleCost::estimated(42), "reads {reads:?}");
            assert_eq!(meter.taken(), taken, "reads {reads:?}");
        }
    }

    /// Every branch of the rule for a resolve the neural stage answered, on
    /// the figures alone.
    #[test]
    fn the_cost_is_measured_only_when_every_microjoule_was_read() {
        let e = JouleCost::estimated;
        let m = JouleCost::measured;
        // No live span: the stage's cost, untouched.
        assert_eq!(cascade_cost(e(40), None), e(40));
        assert_eq!(cascade_cost(m(40), None), m(40));
        assert_eq!(cascade_cost(e(0), None), e(0));
        // A live span and an estimate: summed, and an estimate.
        assert_eq!(cascade_cost(e(40), Some(10)), e(50));
        // A live span and a measurement: the larger reading, never the sum.
        assert_eq!(cascade_cost(m(40), Some(100)), m(100));
        assert_eq!(cascade_cost(m(400), Some(100)), m(400));
        // A live span and an estimated zero: the span's reading.
        assert_eq!(cascade_cost(e(0), Some(100)), m(100));
        // A span is live only when the counter advanced.
        assert_eq!(live_span(Some(5), Some(5)), None);
        assert_eq!(live_span(Some(6), Some(5)), None);
        assert_eq!(live_span(None, Some(5)), None);
        assert_eq!(live_span(Some(5), None), None);
        assert_eq!(live_span(Some(5), Some(8)), Some(3));
    }

    /// Every branch of the rule for a resolve a deterministic tier answered:
    /// the stage's figure plus the calibrated share, and an estimate whatever
    /// the stage's own tag was.
    #[test]
    fn a_hits_cost_is_an_estimate_whatever_it_is_made_of() {
        let e = JouleCost::estimated;
        let m = JouleCost::measured;
        assert_eq!(hit_cost(e(0), 0), e(0), "no calibration, a cache hit");
        assert_eq!(hit_cost(e(50), 0), e(50), "no calibration, a KV hit");
        assert_eq!(hit_cost(e(0), 3), e(3), "the calibrated share alone");
        assert_eq!(hit_cost(e(50), 3), e(53), "the stage's own estimate is added, not replaced");
        assert_eq!(hit_cost(m(9), 0), e(9), "a replayed reading is not a reading of this resolve");
        assert_eq!(hit_cost(m(9), 3), e(12));
        assert_eq!(hit_cost(e(u64::MAX), 3), e(u64::MAX));
    }

    /// The carry, on the figures alone: after any number of hits the
    /// micro-joules charged are the whole micro-joules in the nanojoules
    /// accrued, the rest is carried, and the carry stays under a micro-joule.
    #[test]
    fn the_carry_conserves_every_nanojoule() {
        for per_hit in [0u64, 1, 137, 999, 1_000, 1_001, 2_345, 123_456_789] {
            let mut hits = TierHits::new();
            hits.calibration = Ok(HitCalibration {
                per_hit_nanojoules: per_hit,
                batch: 1,
                span_microjoules: 1,
                empty_span_microjoules: 1,
                took: Duration::ZERO,
            });
            let mut charged = 0u64;
            for n in 1..=5_000u64 {
                charged += hits.charge();
                assert_eq!(charged, n * per_hit / 1_000, "{per_hit} nJ a hit, after {n} hits");
                assert_eq!(hits.carry_nanojoules, n * per_hit % 1_000, "{per_hit} nJ a hit, after {n} hits");
            }
        }
        // No calibration, no charge, and the carry is left as it stands.
        let mut hits = TierHits::new();
        hits.carry_nanojoules = 600;
        for why in [NoCalibration::NotYet, NoCalibration::InProgress, NoCalibration::CounterHeld] {
            hits.calibration = Err(why);
            assert_eq!((hits.charge(), hits.carry_nanojoules), (0, 600), "{why:?}");
        }
    }

    /// A host in miniature, for calibrations that must come out the same on
    /// every run: one cumulative energy count and one clock, which a read of
    /// the meter and a KV lookup move, each by a fixed amount, and nothing
    /// else does. It is the cascade's meter, its clock and its KV backend,
    /// and it keeps what it saw in order. Only the KV tier can be calibrated
    /// against it: a cache or graph lookup moves neither count.
    struct Host {
        nanojoules: AtomicU64,
        nanos: AtomicU64,
        /// What a read of the meter costs: nanojoules, nanoseconds.
        read: (u64, u64),
        /// What a KV lookup costs: nanojoules, nanoseconds.
        get: (AtomicU64, u64),
        /// Once this many reads of the meter have been taken the count
        /// holds, whatever is spent.
        holds_after: Option<u64>,
        store: MemoryKvBackend,
        seen: Mutex<Vec<Seen>>,
    }

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    enum Seen {
        /// A read of the meter, and the micro-joules it returned.
        Read(u64),
        /// A KV lookup.
        Get,
        /// The neural stage.
        Infer,
    }

    impl Host {
        /// A host whose meter costs 10 µJ and 80 µs a read, and whose KV
        /// lookup costs `get_nanojoules` and `get_nanos`.
        fn new(get_nanojoules: u64, get_nanos: u64) -> Arc<Self> {
            Self::holding_after(get_nanojoules, get_nanos, None)
        }

        fn holding_after(get_nanojoules: u64, get_nanos: u64, holds_after: Option<u64>) -> Arc<Self> {
            let store = MemoryKvBackend::new();
            store.put("known key", b"known value".to_vec());
            Arc::new(Self {
                nanojoules: AtomicU64::new(1_000_000),
                nanos: AtomicU64::new(0),
                read: (10_000, 80_000),
                get: (AtomicU64::new(get_nanojoules), get_nanos),
                holds_after,
                store,
                seen: Mutex::new(Vec::new()),
            })
        }

        fn spend(&self, nanojoules: u64, nanos: u64) {
            if self.holds_after.is_none_or(|n| self.reads() < n) {
                self.nanojoules.fetch_add(nanojoules, Ordering::SeqCst);
            }
            self.nanos.fetch_add(nanos, Ordering::SeqCst);
        }

        fn elapsed(&self) -> Duration {
            Duration::from_nanos(self.nanos.load(Ordering::SeqCst))
        }

        fn clock(self: &Arc<Self>) -> Clock {
            let host = self.clone();
            Arc::new(move || Some(host.elapsed()))
        }

        fn seen(&self) -> Vec<Seen> {
            self.seen.lock().unwrap().clone()
        }

        fn count(&self, which: impl Fn(&Seen) -> bool) -> u64 {
            self.seen.lock().unwrap().iter().filter(|s| which(s)).count() as u64
        }

        fn reads(&self) -> u64 {
            self.count(|s| matches!(s, Seen::Read(_)))
        }

        fn gets(&self) -> u64 {
            self.count(|s| *s == Seen::Get)
        }

        /// A cascade on this host that does not memoize, so a KV hit stays a
        /// KV hit and the cache never answers, around a neural stage that
        /// reports `neural`.
        fn cascade(self: &Arc<Self>, neural: JouleCost) -> Cascade {
            Cascade::new(
                Arc::new(LruCache::new(8)),
                Arc::new(KvStage::new(Box::new(HostKv(self.clone())))),
                Arc::new(GraphStage::new()),
                Arc::new(NeuralStage::new(Box::new(HostNeural(self.clone(), neural)))),
            )
            .with_meter(Arc::new(HostMeter(self.clone())))
            .with_clock(self.clock())
            .without_memoization()
        }
    }

    struct HostMeter(Arc<Host>);

    impl JouleCounter for HostMeter {
        /// The read's own cost is spent before the count is taken, so a span
        /// holds its closing read, as a span on a real counter does.
        fn read_microjoules(&self) -> eoc_core::Result<u64> {
            self.0.spend(self.0.read.0, self.0.read.1);
            let microjoules = self.0.nanojoules.load(Ordering::SeqCst) / 1_000;
            self.0.seen.lock().unwrap().push(Seen::Read(microjoules));
            Ok(microjoules)
        }
        fn name(&self) -> &'static str {
            "host"
        }
    }

    struct HostKv(Arc<Host>);

    impl KvBackend for HostKv {
        fn get(&self, key: &str) -> Option<Vec<u8>> {
            self.0.seen.lock().unwrap().push(Seen::Get);
            self.0.spend(self.0.get.0.load(Ordering::SeqCst), self.0.get.1);
            self.0.store.get(key)
        }
        fn put(&self, key: &str, value: Vec<u8>) {
            self.0.store.put(key, value)
        }
        fn embeddings(&self) -> Vec<(Vec<f32>, String)> {
            self.0.store.embeddings()
        }
        fn put_embedding(&self, embedding: Vec<f32>, key: String) {
            self.0.store.put_embedding(embedding, key)
        }
    }

    struct HostNeural(Arc<Host>, JouleCost);

    #[async_trait::async_trait]
    impl eoc_neural::NeuralBackend for HostNeural {
        async fn infer(&self, q: &Query) -> Response {
            self.0.seen.lock().unwrap().push(Seen::Infer);
            Response::new(q.id, "inferred".to_string(), StageKind::Neural, self.1)
        }
    }

    /// The micro-joules of a KV hit's cost that are the calibrated share:
    /// the rest is the 50 µJ the KV stage reports for itself.
    fn share(r: &Response) -> u64 {
        assert_eq!((r.stage, r.joule_cost.source), (StageKind::Kv, JouleSource::Estimated), "{}", r.joule_cost);
        r.joule_cost.microjoules - own_cost(StageKind::Kv)
    }

    /// The first hit on a tier calibrates it, and after that no hit on the
    /// tier reads the meter: not one read in a thousand hits, on any of the
    /// three tiers. Every hit is an estimate.
    #[tokio::test]
    async fn a_hit_reads_no_meter_once_its_tier_is_calibrated() {
        for stage in TIERS {
            let meter = Scripted::new(&CALIBRATES_AT_700);
            let cascade = metered(&meter, Box::new(EchoBackend::new())).without_memoization();
            let q = answered_by(&cascade, stage);
            assert_eq!(cascade.calibration(stage), Err(NoCalibration::NotYet));
            let first = cascade.resolve(q.clone()).await;
            assert_eq!(first.stage, stage);
            assert_eq!(meter.taken(), 4, "{stage}: the calibration's reads");
            assert_eq!(cascade.calibration(stage).map(|c| c.per_hit_nanojoules), Ok(700_000), "{stage}");
            for _ in 0..1_000 {
                let r = cascade.resolve(q.clone()).await;
                assert_eq!(r.stage, stage);
                assert_eq!(r.joule_cost, JouleCost::estimated(own_cost(stage) + 700), "{stage}");
                assert!(r.receipt.verify(&r.query_id, &r.stage, &r.joule_cost, &r.payload));
            }
            assert_eq!(first.joule_cost, JouleCost::estimated(own_cost(stage) + 700), "{stage}: the first hit");
            assert_eq!(meter.taken(), 4, "{stage}: a hit read the meter");
            // The other tiers have not answered, and were not calibrated.
            for other in TIERS.into_iter().filter(|other| *other != stage) {
                assert_eq!(cascade.calibration(other), Err(NoCalibration::NotYet), "{stage}: {other}");
            }
        }

        // The same on a host whose every read and lookup is counted.
        let host = Host::new(137, 150);
        let cascade = host.cascade(JouleCost::estimated(42));
        let q = Query::new("known key");
        cascade.resolve(q.clone()).await;
        cascade.calibration(StageKind::Kv).expect("the KV tier is calibrated");
        let (reads, gets) = (host.reads(), host.gets());
        assert!(reads >= 4, "{reads} reads calibrated the tier");
        for _ in 0..1_000 {
            cascade.resolve(q.clone()).await;
        }
        assert_eq!(host.reads(), reads, "a hit read the meter");
        assert_eq!(host.gets(), gets + 1_000, "a hit is one lookup");
    }

    /// A resolve the neural stage answers reads the meter exactly twice: once
    /// after the deterministic tiers have missed, once after the stage. What
    /// the KV lookup spent before the first read is not in the span.
    #[tokio::test]
    async fn a_neural_resolve_reads_the_meter_twice_and_only_after_the_walk() {
        // A miss in the KV stage here spends 2 mJ, forty times the 50 µJ
        // span the two reads then bracket.
        let host = Host::new(2_000_000, 150);
        let cascade = host.cascade(JouleCost::zero());
        let before = host.nanojoules.load(Ordering::SeqCst) / 1_000;
        let r = cascade.resolve(Query::new("nothing knows this")).await;
        let seen = host.seen();
        let &[Seen::Get, Seen::Read(open), Seen::Infer, Seen::Read(close)] = seen.as_slice() else {
            panic!("the walk, a read, the neural stage, a read: {seen:?}");
        };
        assert_eq!(r.stage, StageKind::Neural);
        assert_eq!(r.joule_cost, JouleCost::measured(close - open));
        assert_eq!(close - open, 10, "the span holds its closing read's 10 µJ and nothing of the walk");
        assert_eq!(open - before, 2_010, "the miss's 2 mJ and the opening read were spent before the span");

        // And an estimating stage: its estimate plus the span, an estimate.
        let host = Host::new(2_000_000, 150);
        let r = host.cascade(JouleCost::estimated(42)).resolve(Query::new("nothing knows this")).await;
        assert_eq!(r.joule_cost, JouleCost::estimated(52));
        assert_eq!(host.reads(), 2);
    }

    /// What a calibration records is what it did: the batch is the number of
    /// lookups between the two reads of its last span, the span is those two
    /// reads' delta, the empty span is the counter's first advance over no
    /// work, and the per-hit figure is the span over the batch. The batch
    /// was grown until its span read fifty times the empty one.
    #[tokio::test]
    async fn a_calibration_records_the_batch_it_ran_and_the_span_it_read() {
        for get_nanojoules in [137, 2_345, 40_000] {
            let host = Host::new(get_nanojoules, 150);
            let cascade = host.cascade(JouleCost::estimated(42));
            cascade.resolve(Query::new("known key")).await;
            let c = cascade.calibration(StageKind::Kv).expect("calibrated");
            let seen = host.seen();
            let reads: Vec<usize> = (0..seen.len()).filter(|i| matches!(seen[*i], Seen::Read(_))).collect();
            let at = |i: usize| match seen[i] {
                Seen::Read(microjoules) => microjoules,
                other => panic!("{other:?} is not a read"),
            };
            // The hit's own lookup comes first, then the two reads that
            // found the meter's own cost, with nothing between them.
            assert_eq!(seen[0], Seen::Get);
            assert_eq!((reads[0], reads[1]), (1, 2), "{seen:?}");
            assert_eq!(c.empty_span_microjoules, at(reads[1]) - at(reads[0]));
            assert_eq!(c.empty_span_microjoules, 10);
            // The last span: its two reads, and the lookups between them.
            let (open, close) = (reads[reads.len() - 2], reads[reads.len() - 1]);
            assert_eq!(close, seen.len() - 1, "nothing follows the closing read");
            assert_eq!(c.batch, (close - open - 1) as u64, "{get_nanojoules} nJ: the batch is the lookups in the span");
            assert!(seen[open + 1..close].iter().all(|s| *s == Seen::Get));
            assert_eq!(c.span_microjoules, at(close) - at(open), "{get_nanojoules} nJ");
            assert_eq!(c.per_hit_nanojoules, (c.span_microjoules * 1_000 + c.batch / 2) / c.batch);
            assert!(c.span_microjoules >= SPAN_TO_EMPTY * c.empty_span_microjoules, "{c:?}");
            // All the host's time but the hit's own lookup, which came first.
            assert_eq!(c.took, host.elapsed() - Duration::from_nanos(150));
            // Every span is a pair of reads: the empty one, then each batch.
            assert_eq!(reads.len() % 2, 0, "{seen:?}");
            // The figure is the lookup's own, over by the meter's share of
            // the span and no more: 10 µJ in a span of 500 at least.
            let over = c.per_hit_nanojoules as f64 / get_nanojoules as f64;
            assert!((1.0..=1.03).contains(&over), "{get_nanojoules} nJ a lookup calibrated as {c:?}");
        }
    }

    /// Integer conservation. Over any number of hits the micro-joules
    /// charged sum to the hits' calibrated nanojoules, to within the one
    /// micro-joule still carried: for a figure under a micro-joule, where
    /// most hits are charged nothing and some one, and for a figure over it.
    #[tokio::test]
    async fn the_microjoules_charged_sum_to_the_calibrated_total() {
        const HITS: u64 = 10_000;
        for get_nanojoules in [137u64, 2_345] {
            let host = Host::new(get_nanojoules, 150);
            let cascade = host.cascade(JouleCost::estimated(42));
            let q = Query::new("known key");
            let mut charges = Vec::new();
            for _ in 0..HITS {
                charges.push(share(&cascade.resolve(q.clone()).await));
            }
            let per_hit = cascade.calibration(StageKind::Kv).expect("calibrated").per_hit_nanojoules;
            assert_eq!(per_hit < 1_000, get_nanojoules < 1_000, "{per_hit} nJ calibrated for {get_nanojoules}");
            let mut charged = 0u64;
            for (n, charge) in charges.iter().enumerate() {
                charged += charge;
                let accrued = (n as u64 + 1) * per_hit;
                assert_eq!(charged, accrued / 1_000, "{per_hit} nJ a hit, after {} hits", n + 1);
            }
            let total = (HITS * per_hit + 500) / 1_000;
            assert!(charged.abs_diff(total) <= 1, "{charged} µJ charged for {total} µJ calibrated");
            // Neither every hit rounded up nor a string of zeros.
            let (least, most) = (per_hit / 1_000, per_hit.div_ceil(1_000));
            assert!(charges.iter().all(|c| *c == least || *c == most), "{per_hit} nJ a hit");
            assert!(charges.contains(&least) && charges.contains(&most), "{per_hit} nJ a hit");
            assert_ne!(charged, HITS * most, "every hit was rounded up");
            assert_ne!(charged, HITS * least, "every hit was rounded down");
        }
    }

    /// The carry is one count for the tier, whichever thread a hit runs on:
    /// eight threads' hits together are charged the calibrated total.
    #[test]
    fn hits_on_many_threads_are_charged_the_calibrated_total() {
        const THREADS: u64 = 8;
        const HITS: u64 = 500;
        /// The micro-joules `n` hits are charged, on a runtime of this
        /// thread's own.
        fn block_on(cascade: &Cascade, n: u64) -> u64 {
            let runtime = tokio::runtime::Builder::new_current_thread().build().unwrap();
            runtime.block_on(async {
                let mut charged = 0;
                for _ in 0..n {
                    charged += share(&cascade.resolve(Query::new("known key")).await);
                }
                charged
            })
        }
        let host = Host::new(2_345, 150);
        let cascade = Arc::new(host.cascade(JouleCost::estimated(42)));
        // One hit first, so the tier is calibrated before the threads start.
        let first = block_on(&cascade, 1);
        let per_hit = cascade.calibration(StageKind::Kv).expect("calibrated").per_hit_nanojoules;
        let threads: Vec<_> = (0..THREADS)
            .map(|_| {
                let cascade = cascade.clone();
                std::thread::spawn(move || block_on(&cascade, HITS))
            })
            .collect();
        let charged: u64 = first + threads.into_iter().map(|t| t.join().unwrap()).sum::<u64>();
        assert_eq!(charged, (THREADS * HITS + 1) * per_hit / 1_000);
    }

    /// A lookup that is slow is repeated only as far as the time allowed: no
    /// batch is started that would not fit, and the largest span read by
    /// then is kept, short of fifty empty spans as it is. A lookup slower
    /// than the whole allowance is repeated once.
    #[tokio::test]
    async fn a_slow_lookup_is_repeated_only_as_far_as_the_time_allowed() {
        // 1 ms a lookup: 40 fit.
        let host = Host::new(2_000, 1_000_000);
        let cascade = host.cascade(JouleCost::estimated(42));
        cascade.resolve(Query::new("known key")).await;
        let c = cascade.calibration(StageKind::Kv).expect("a live span is a calibration, however short");
        assert!(c.took <= CALIBRATION_CAP, "{c:?}");
        assert!(host.gets() <= 1 + 40, "{} lookups of 1 ms", host.gets());
        assert!(c.batch > 8 && c.batch <= 40, "{c:?}");
        assert!(c.span_microjoules < SPAN_TO_EMPTY * c.empty_span_microjoules, "{c:?}");
        assert_eq!(c.per_hit_nanojoules, (c.span_microjoules * 1_000 + c.batch / 2) / c.batch);

        // 100 ms a lookup: one, in one span.
        let host = Host::new(2_000, 100_000_000);
        let cascade = host.cascade(JouleCost::estimated(42));
        let r = cascade.resolve(Query::new("known key")).await;
        let c = cascade.calibration(StageKind::Kv).expect("calibrated");
        assert_eq!((c.batch, host.gets()), (1, 2), "the hit's lookup and one more");
        assert_eq!(c.span_microjoules, 12, "the lookup's 2 µJ and the closing read's 10");
        assert_eq!(r.joule_cost, JouleCost::estimated(50 + 12));
    }

    /// With no live counter a hit is what its stage reported, and so is a
    /// neural resolve: the cascade given no meter at all, a counter whose
    /// read fails, one that holds from the start, and one that holds once
    /// the batches begin. The attempt is made once: later hits read nothing.
    #[tokio::test]
    async fn with_no_live_counter_every_resolve_keeps_its_stages_cost() {
        // No meter given: nothing is read, and nothing is calibrated.
        let cascade = fixture().without_memoization();
        for stage in TIERS {
            let q = answered_by(&cascade, stage);
            for _ in 0..3 {
                let r = cascade.resolve(q.clone()).await;
                assert_eq!((r.stage, r.joule_cost), (stage, JouleCost::estimated(own_cost(stage))));
            }
            assert_eq!(cascade.calibration(stage), Err(NoCalibration::NoMeter));
        }
        let r = cascade.resolve(Query::new("totally novel question")).await;
        assert_eq!(r.joule_cost, JouleCost::estimated(50_000_000));

        // A failed read (one read spent), and a counter that holds over
        // back-to-back reads (given up after 2 ms of them).
        for (reads, why, spent) in [
            (vec![None], NoCalibration::ReadFailed, 1),
            (vec![Some(7_000), None], NoCalibration::ReadFailed, 2),
            (vec![Some(7_000)], NoCalibration::CounterHeld, 2),
        ] {
            for stage in TIERS {
                let meter = Scripted::new(&reads);
                let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(42))).without_memoization();
                let q = answered_by(&cascade, stage);
                for _ in 0..3 {
                    let r = cascade.resolve(q.clone()).await;
                    assert_eq!((r.stage, r.joule_cost), (stage, JouleCost::estimated(own_cost(stage))), "{reads:?}");
                }
                assert_eq!(cascade.calibration(stage), Err(why), "{stage} on {reads:?}");
                assert_eq!(meter.taken(), spent, "{stage} on {reads:?}: the one attempt's reads");
            }
        }

        // A counter that moved over no work and then held over every batch:
        // the batches stop at the time allowed (the last of them fits in it,
        // its two reads of 80 µs come on top, and the hit's own lookup of
        // 20 µs came before), with no span to keep.
        let host = Host::holding_after(2_000, 20_000, Some(2));
        let cascade = host.cascade(JouleCost::estimated(42));
        let r = cascade.resolve(Query::new("known key")).await;
        assert_eq!(r.joule_cost, JouleCost::estimated(50));
        assert_eq!(cascade.calibration(StageKind::Kv), Err(NoCalibration::CounterHeld));
        assert!(host.elapsed() <= CALIBRATION_CAP + Duration::from_micros(160 + 20), "{:?}", host.elapsed());
        assert!(host.elapsed() > CALIBRATION_CAP / 2, "the batches stopped early: {:?}", host.elapsed());
        let reads = host.reads();
        for _ in 0..3 {
            let r = cascade.resolve(Query::new("known key")).await;
            assert_eq!(r.joule_cost, JouleCost::estimated(50));
        }
        assert_eq!(host.reads(), reads, "a hit read the meter");
        let r = cascade.resolve(Query::new("nothing knows this")).await;
        assert_eq!(r.joule_cost, JouleCost::estimated(42), "the neural stage's own cost");
        assert_eq!(host.reads(), reads + 2);
    }

    /// The old behaviour is gone: a resolve a deterministic tier answers is
    /// never measured. Not on a live counter, where a span around the hit
    /// would have read (1_000 -> 1_010 here, once `measured(10)`); not the
    /// hit that calibrates its tier; not where the calibration failed; and
    /// not a cached response that carries a measured figure of its own.
    #[tokio::test]
    async fn a_hit_is_never_measured() {
        let scripts: [&[Option<u64>]; 5] = [
            &CALIBRATES_AT_700,
            &[Some(1_000), Some(1_010), Some(2_000), None],
            &[Some(2_000), Some(2_250), None],
            &[Some(7_000)],
            &[None],
        ];
        for script in scripts {
            for stage in TIERS {
                let meter = Scripted::new(script);
                let cascade = metered(&meter, Box::new(EchoBackend::new())).without_memoization();
                let q = answered_by(&cascade, stage);
                for hit in 0..5 {
                    let r = cascade.resolve(q.clone()).await;
                    assert_eq!(r.stage, stage);
                    let cost = r.joule_cost;
                    assert_eq!(cost.source, JouleSource::Estimated, "{stage} hit {hit} on {script:?}: {cost}");
                }
            }
        }

        // A memoized answer, on a live counter: the neural resolve is read,
        // the hit that follows is not.
        let meter = Scripted::new(&[Some(1_000), Some(1_500), Some(2_000), Some(2_010), Some(3_000), Some(3_700)]);
        let cascade = metered(&meter, Box::new(EchoBackend::new().with_cost(0)));
        let q = Query::new("totally novel question");
        assert_eq!(cascade.resolve(q.clone()).await.joule_cost, JouleCost::measured(500));
        let hit = cascade.resolve(q).await;
        assert_eq!((hit.stage, hit.joule_cost), (StageKind::Cache, JouleCost::estimated(700)));

        // A cached response that carries a measured figure: what was read
        // for an earlier resolve is not a reading of this one.
        for cascade in [fixture(), metered(&Scripted::new(&CALIBRATES_AT_700), Box::new(EchoBackend::new()))] {
            let q = Query::new("answered and read once");
            let read_once = Response::new(q.id, "answer".to_string(), StageKind::Cache, JouleCost::measured(9));
            cascade.cache().insert_response(q.id, read_once);
            let r = cascade.resolve(q).await;
            assert_eq!(r.joule_cost.source, JouleSource::Estimated, "{}", r.joule_cost);
            assert!(r.joule_cost.microjoules >= 9);
            assert!(r.receipt.verify(&r.query_id, &r.stage, &r.joule_cost, &r.payload));
        }
    }

    /// While one resolve calibrates a tier, another that the tier answers is
    /// charged the stage's own cost and reads nothing; and a tier switched
    /// off is never calibrated, by a hit or by `calibrate`.
    #[tokio::test]
    async fn a_tier_being_calibrated_or_switched_off_charges_its_stages_cost() {
        let host = Host::new(137, 150);
        let cascade = host.cascade(JouleCost::estimated(42));
        cascade.hits(Tier::Kv).calibration = Err(NoCalibration::InProgress);
        let r = cascade.resolve(Query::new("known key")).await;
        assert_eq!(r.joule_cost, JouleCost::estimated(50));
        assert_eq!((host.reads(), cascade.calibration(StageKind::Kv)), (0, Err(NoCalibration::InProgress)));

        let host = Host::new(137, 150);
        let cascade = host.cascade(JouleCost::estimated(42)).without_hit_calibration(StageKind::Kv);
        let q = Query::new("known key");
        assert_eq!(cascade.resolve(q.clone()).await.joule_cost, JouleCost::estimated(50));
        assert_eq!(cascade.calibrate(&q).await, Some((StageKind::Kv, Err(NoCalibration::Disabled))));
        assert_eq!(cascade.resolve(q).await.joule_cost, JouleCost::estimated(50));
        assert_eq!(host.reads(), 0);
        // The neural stage is read, never calibrated, and cannot be switched.
        let cascade = cascade.without_hit_calibration(StageKind::Neural);
        assert_eq!(cascade.calibration(StageKind::Neural), Err(NoCalibration::NotADeterministicTier));
    }

    /// `calibrate` calibrates the tier that answers a query without a
    /// resolve, and again whenever it is asked: the new figure replaces the
    /// old, and the nanojoules carried under the old one are not dropped.
    #[tokio::test]
    async fn a_tier_can_be_calibrated_again() {
        let host = Host::new(137, 150);
        let cascade = host.cascade(JouleCost::estimated(42));
        let q = Query::new("known key");
        assert_eq!(cascade.calibrate(&Query::new("nothing knows this")).await, None);
        assert_eq!(host.reads(), 0, "no tier answered, so nothing was read");

        // Before any resolve: the first hit then finds the tier calibrated.
        let (stage, first) = cascade.calibrate(&q).await.expect("the KV stage answers");
        let first = first.expect("calibrated");
        assert_eq!(stage, StageKind::Kv);
        assert_eq!(cascade.calibration(StageKind::Kv), Ok(first));
        let reads = host.reads();
        let mut charged = 0;
        for _ in 0..7 {
            charged += share(&cascade.resolve(q.clone()).await);
        }
        assert_eq!(host.reads(), reads, "a hit read the meter");
        assert_eq!(charged, 7 * first.per_hit_nanojoules / 1_000);

        // The lookup now costs seventeen times as much.
        host.get.0.store(2_345, Ordering::SeqCst);
        let (_, second) = cascade.calibrate(&q).await.expect("the KV stage answers");
        let second = second.expect("calibrated");
        assert!(second.per_hit_nanojoules > 10 * first.per_hit_nanojoules, "{first:?} then {second:?}");
        assert_eq!(cascade.calibration(StageKind::Kv), Ok(second));
        for _ in 0..7 {
            charged += share(&cascade.resolve(q.clone()).await);
        }
        assert_eq!(charged, (7 * first.per_hit_nanojoules + 7 * second.per_hit_nanojoules) / 1_000);
    }

    /// A resolve and a calibration can be spawned onto a runtime's other
    /// threads: no lock is held across a stage's lookup.
    #[test]
    fn a_resolve_can_move_between_threads() {
        fn sendable<T: Send>(_: &T) {}
        let cascade = fixture();
        let q = Query::new("known key");
        sendable(&cascade.resolve(q.clone()));
        sendable(&cascade.calibrate(&q));
    }
}