fix(lease): wire the poison cooldown and the stand-down notice to their callers

crates/batten/src/land.rs

@@ -1459,6 +1459,45 @@ impl Basis {
     }
 }
 
+/// Did THIS clone just redden the metered matrix on a base of its own?
+///
+/// The predicate behind the poison cooldown (CLOUD-1797). It is the same cell
+/// [`progress_of`] answers `Stop` for, narrowed by the basis, and each part of
+/// it is load-bearing:
+///
+/// - **`Wait` and `Violation`** is the wait's OWN refusal. Keying on a red
+///   reading alone would fire on a wait that succeeded while a stale red sat in
+///   `seen` — the defect `progress_of`'s own comment records one cell below.
+/// - **`Red`, never `Pending` or `None`.** A could-not-look is not a poisoning,
+///   and a cooldown placed on a network blip holds a clone back over nothing.
+/// - **`Own`, never `Borrowed`.** On a borrowed base the red may be the
+///   speculated base's fault, so charging it here would let a neighbour's bad
+///   tree evict an innocent agent — the asymmetry `charge_the_lap` already
+///   refuses for the same reason.
+///
+/// Pure, and free-standing rather than a branch inside the lap, so the cell can
+/// be decided in a test instead of only through a git resolution and a
+/// filesystem write.
+//MUTANT-SUITE crates/batten/tests/it/lease_lifecycle.rs
+//MUTANT poison-never-recorded|s@    matches!(@    !matches!(@|a_red_wait_on_this_clones_own_base_is_what_poisons_it
+#[must_use]
+pub const fn poisons_this_clone(
+    step: Step,
+    code: crate::exit::ExitCode,
+    seen: Option<TapVerdict>,
+    basis: Basis,
+) -> bool {
+    matches!(
+        (step, code, seen, basis),
+        (
+            Step::Wait,
+            crate::exit::ExitCode::Violation,
+            Some(TapVerdict::Red),
+            Basis::Own
+        )
+    )
+}
+
 #[must_use]
 pub const fn progress_of(
     step: Step,

crates/batten/src/lease.rs

@@ -2309,6 +2309,67 @@ pub fn notice(body: &Body, from: &str) -> Body {
     }
 }
 
+/// Should `ours` ask a holder writing `theirs` to stand down?
+///
+/// The predicate behind the only sender of [`notice`] (CLOUD-1798). Strictly
+/// newer, and **every uncertain reading answers `false`**:
+///
+/// - **An equal build is not out-ranked.** Two clones on the same version asking
+///   each other to stand down is a request loop with no newer party to win it.
+/// - **An unreadable version on either side is not evidence of anything.** A
+///   body minted by a build whose version this one cannot parse might be newer;
+///   reading it as older would let an old clone evict the whole fleet, which is
+///   the inversion [`Body::writer`]'s own doc refuses when it says the field
+///   decides nothing on its own.
+///
+/// A pre-release or build suffix is cut before the comparison rather than
+/// ordered, because the question here is "is the holder behind us" and no
+/// decision in this repository turns on `-rc.1` ordering. Saying so is cheaper
+/// than implementing an ordering nothing asks for.
+#[must_use]
+pub fn outranks(ours: &str, theirs: &str) -> bool {
+    fn triple(version: &str) -> Option<(u64, u64, u64)> {
+        let core = version.trim().split(['-', '+']).next().unwrap_or_default();
+        let mut parts = core.split('.');
+        let mut next = || parts.next()?.parse::<u64>().ok();
+        let (major, minor, patch) = (next()?, next()?, next()?);
+        // A FOURTH SEGMENT IS NOT A VERSION THIS UNDERSTANDS. Reading `1.2.3.4`
+        // as `1.2.3` would rank two distinct builds equal.
+        parts.next().is_none().then_some((major, minor, patch))
+    }
+    match (triple(ours), triple(theirs)) {
+        (Some(ours), Some(theirs)) => ours > theirs,
+        _ => false,
+    }
+}
+
+/// Is it worth `asker` leaving a stand-down request on this body?
+///
+/// The three guards of the notice sender as one pure predicate, so the decision
+/// is testable without a ref, a CAS or a wire (CLOUD-1798). Each clause is a
+/// distinct way to get this wrong:
+///
+/// 1. **Strictly newer** — [`outranks`], which refuses an equal build and every
+///    unreadable one.
+/// 2. **Not already asked** — idempotence. Without it every lap of every waiter
+///    rewrites the ref and the holder's own heartbeat CAS fails against a stream
+///    of cosmetic updates, evicting by contention rather than by asking.
+/// 3. **Never self-addressed** — the mirror of
+///    [`stand_down_requested`]'s own guard, one step earlier: that one stops a
+///    holder honouring a notice it minted onto itself, this one stops it being
+///    written.
+///
+/// A released or expired body is not excluded here and does not need to be: the
+/// caller only reaches this on `Turn::Wait`, which those states never produce.
+//MUTANT-SUITE crates/batten/tests/it/lease_lifecycle.rs
+//MUTANT notice-ignores-the-writer|s@    outranks(\&writer_version(), \&body.writer)@    true@|a_waiter_asks_only_a_strictly_older_holder_to_stand_down
+#[must_use]
+pub fn worth_asking(body: &Body, asker: &str) -> bool {
+    outranks(&writer_version(), &body.writer)
+        && body.stand_down.trim().is_empty()
+        && body.holder != asker
+}
+
 /// Has a peer asked this holder to stand down?
 ///
 /// A pure reading over a body already in hand, so the beat pays no extra fetch
@@ -2398,6 +2459,62 @@ impl Local {
         Ok(minted)
     }
 
+    /// Record that THIS clone's head reddened the metered matrix on `trunk`.
+    ///
+    /// The written value is a trunk POSITION, never an instant, and that is the
+    /// whole reason [`cooling`] can decide without a clock: the cooldown lapses
+    /// when the trunk moves off this sha or when the pool goes idle, both of them
+    /// observable events (CLOUD-1784's decomposition, CLOUD-1797's wiring).
+    ///
+    /// Beside `holder` in the same per-clone directory, because it is the same
+    /// kind of fact — something this clone knows about itself that must outlive
+    /// the process, since the `land` that reddened CI and the `acquire` that must
+    /// be held back are different processes.
+    ///
+    /// # THE RECORD NAMES ITS OWN REF, and that is not decoration
+    ///
+    /// It stores `<tracking-ref>\t<sha>`, so the reader resolves the ref the
+    /// WRITER used rather than a constant of its own. A reader that assumed
+    /// `origin/main` would compare this clone's poisoned sha against an unrelated
+    /// ref on any other base, never match, and answer "not cooling" — a
+    /// fail-open arm inside a mechanism that exists to fail closed, which is the
+    /// shape CLOUD-1792 was filed over. Naming the ref costs one field and
+    /// removes the assumption entirely.
+    ///
+    /// # Errors
+    ///
+    /// A directory or file this clone cannot write. **Never swallowed**: a poison
+    /// that failed to record reads as an unpoisoned clone on the next acquire,
+    /// which is the defect this exists to close.
+    pub fn poison(&self, tracking: &str, at: &str) -> Result<()> {
+        std::fs::create_dir_all(&self.dir)?;
+        std::fs::write(self.dir.join("poisoned"), format!("{tracking}\t{at}\n"))?;
+        Ok(())
+    }
+
+    /// The ref and position this clone last poisoned, if it has poisoned one.
+    ///
+    /// `None` for an absent, empty or malformed record, and for one this clone
+    /// cannot read. **Unreadable reads as unpoisoned on purpose** — the opposite
+    /// of [`Local::holder`]'s rule, and for the opposite reason: a holder id that
+    /// defaulted would let two clones claim one lease, while a cooldown that
+    /// defaulted to ON would hold a clone back over a file it cannot read, which
+    /// is the single-agent stranding [`cooling`]'s idle-pool clause exists to
+    /// prevent.
+    ///
+    /// A record missing its tab is malformed rather than a bare sha, because a
+    /// bare sha is what the FIRST spelling of this wrote and reading it loosely
+    /// would resurrect the assumption the tab exists to remove.
+    #[must_use]
+    pub fn poisoned(&self) -> Option<(String, String)> {
+        let record = std::fs::read_to_string(self.dir.join("poisoned")).ok()?;
+        let (tracking, at) = record.trim().split_once('\t')?;
+        if tracking.is_empty() || at.is_empty() {
+            return None;
+        }
+        Some((tracking.to_owned(), at.to_owned()))
+    }
+
     /// How long `token` has been what this clone sees under `name`, on OUR clock.
     ///
     /// **Expiry alone is not safe to steal on**, which is the whole reason this

crates/batten/src/lib.rs

@@ -7401,6 +7401,9 @@ fn run_land_lap(
             // different world than the one that answered (CLOUD-1306).
             let basis = land::Basis::of(bet.live());
             note_the_poison(step, code, basis, &mut bet);
+            if land::poisons_this_clone(step, code, seen, basis) {
+                note_the_cooldown(root, reference, out, err)?;
+            }
             match land::progress_of(step, code, seen, basis) {
                 land::Progress::Proceed => {}
                 // `None` is not merged, or nobody could say. Either way this is
@@ -8052,6 +8055,77 @@ fn note_the_poison(
     }
 }
 
+/// Record the poison cooldown when THIS branch's own head reddened the matrix.
+///
+/// The mirror of [`note_the_poison`], and deliberately not folded into it: that
+/// one is about somebody ELSE's base and lives in the `Bet`, which dies with the
+/// process. This one is about THIS clone and must outlive the process, because
+/// the `land` that reddened CI and the `acquire` that must be held back are
+/// different runs (CLOUD-1797).
+///
+/// # The cell, and why each part of it is load-bearing
+///
+/// `(Wait, Violation, Red, Own)` — the same cell [`land::progress_of`] answers
+/// `Stop` for, narrowed by the basis:
+///
+/// - **`Wait` and `Violation`** is the wait's own refusal. Keying on a red
+///   reading alone would fire on a wait that SUCCEEDED while a stale red sat in
+///   `seen`, which is the defect `progress_of`'s own comment records.
+/// - **`Red`, never `Pending` or `None`.** Could-not-look is not a poisoning, and
+///   a cooldown placed on a network blip would hold this clone back over nothing.
+/// - **`Own`, never `Borrowed`.** On a borrowed base the red may be the
+///   speculated base's fault, and charging it to this clone would let a
+///   neighbour's bad tree evict an innocent agent — the same asymmetry
+///   `charge_the_lap` already refuses.
+///
+/// # What it records
+///
+/// The trunk POSITION, never an instant. [`lease::cooling`] lapses the cooldown
+/// when the trunk moves off that sha or when the pool goes idle, so the whole
+/// mechanism is two observable events and no clock (CLOUD-1784).
+///
+/// # Why a failure here reports rather than returns
+///
+/// The lap is already stopping — this cell is `Progress::Stop`. Failing the land
+/// over a bookkeeping write would replace a reported stop with an unreported one
+/// and tell the operator less. It is loud on `err` because a cooldown that did
+/// not record reads as an unpoisoned clone on the next acquire, which is exactly
+/// the hole this closes.
+///
+/// **The cell is [`land::poisons_this_clone`], asked by the caller**, for the
+/// reason [`land::progress_of`] is a free function rather than a branch inside
+/// this loop: the decision is the part worth testing, and a decision reachable
+/// only through a git resolution and a filesystem write is a decision nothing
+/// tests.
+fn note_the_cooldown(
+    root: &Path,
+    reference: &str,
+    out: &mut dyn Write,
+    err: &mut dyn Write,
+) -> Result<()> {
+    let tracking = land::tracking_ref(reference);
+    let Ok(Some(trunk)) = git::resolve_ref(root, &tracking) else {
+        writeln!(
+            err,
+            "::error:: land: {tracking} would not read, so the poison cooldown has no base to name"
+        )?;
+        return Ok(());
+    };
+    match lease_identity(root).and_then(|(local, _)| {
+        local
+            .poison(&tracking, &trunk)
+            .map_err(|why| format!("cannot record the poison cooldown: {why}"))
+    }) {
+        // Pointer-only: the base, never the record.
+        Ok(()) => writeln!(
+            out,
+            "land: cooling on {trunk} — this head reddened the matrix"
+        )?,
+        Err(why) => writeln!(err, "::error:: land: {why}")?,
+    }
+    Ok(())
+}
+
 /// Refresh the trunk's remote-tracking ref before a speculation reads it.
 ///
 /// **EVERY SPECULATION READ OF THE TRUNK WAS ONE LAP BEHIND** (CLOUD-1620's
@@ -10798,6 +10872,19 @@ fn run_lease_acquire(
             match &observed {
                 lease::Observed::Held { body, .. } => {
                     writeln!(out, "lease: held by {}", body.holder)?;
+                    // THE ONLY SENDER OF A STAND-DOWN NOTICE (CLOUD-1798). The
+                    // receiving half — `stand_down_requested`, `Beat::StandDown`,
+                    // `honour_the_notice` — landed wired and this arm is what was
+                    // missing, so `stand_down` was empty on every lease in the
+                    // fleet.
+                    //
+                    // IT DOES NOT CHANGE THE WAIT. The waiter still answers
+                    // `Violation`, still does not take the lease and still does
+                    // not retry: asking is a REQUEST the holder may decline,
+                    // which is what makes it the eviction path that is safe
+                    // under a false suspicion. `turn`'s steal arms are the
+                    // unsafe ones and they are unchanged.
+                    ask_the_holder_to_stand_down(terms, &observed, body, &holder, now, out, err)?;
                     Ok(ExitCode::Violation)
                 }
                 // **A REF THAT IS NOT A LEASE IS A WAIT, NOT AN INTERNAL ERROR**
@@ -10833,6 +10920,16 @@ fn run_lease_acquire(
             }
         }
         lease::Turn::Take(why) => {
+            // THE POISON COOLDOWN, ASKED AFTER THE TURN AND BEFORE THE CAS
+            // (CLOUD-1797). `turn` decides whose turn it is; this decides whether
+            // THIS clone may take a turn it is otherwise owed, and the order is
+            // what keeps the two separable — folding it into `turn` would put a
+            // local file inside a decision every other clone must be able to
+            // reproduce from the ref alone.
+            if let Some(reason) = still_cooling(root, &local, &observed) {
+                writeln!(out, "lease: {reason}")?;
+                return Ok(ExitCode::Violation);
+            }
             let body = lease::claim(terms, &holder, branch, &head, now);
             match lease::cas(terms, &observed, &body, now) {
                 Ok(lease::Outcome::Applied) => {
@@ -10857,6 +10954,73 @@ fn run_lease_acquire(
     }
 }
 
+/// Leave a stand-down request on a holder this clone out-ranks, if it should.
+///
+/// **Three guards, and each is a way to get this wrong:**
+///
+/// - **Strictly newer only** ([`lease::outranks`]). A waiter asking an equal or
+///   NEWER holder to stand down inverts the design and would let an old clone
+///   evict the fleet.
+/// - **Idempotent.** A body already carrying a request is left alone. Without
+///   this, every lap of every waiter rewrites the ref, and the holder's own
+///   heartbeat CAS fails against a stream of cosmetic updates — a request that
+///   bullies the holder off the lease by contention rather than by asking.
+/// - **Never self-addressed.** Guarded here as well as in
+///   [`lease::stand_down_requested`], because the reader's guard protects the
+///   holder from a notice it minted onto itself and this one stops it being
+///   written at all.
+///
+/// # Why a lost CAS is silent
+///
+/// Somebody else moved the ref — the holder beat, released, or another waiter
+/// asked first. All three make the request moot or already-made, and none is
+/// this clone's problem. Retrying would be the tight spin `run_lease_acquire`'s
+/// own CAS arm refuses for the same reason.
+fn ask_the_holder_to_stand_down(
+    terms: &lease::Terms,
+    observed: &lease::Observed,
+    body: &lease::Body,
+    holder: &str,
+    now: i64,
+    out: &mut dyn Write,
+    err: &mut dyn Write,
+) -> Result<()> {
+    if !lease::worth_asking(body, holder) {
+        return Ok(());
+    }
+    match lease::cas(terms, observed, &lease::notice(body, holder), now) {
+        // Pointer-only: who was asked and which build, never the body.
+        Ok(lease::Outcome::Applied) => writeln!(
+            out,
+            "lease: asked {} to stand down — it writes {} and this clone writes {}",
+            body.holder,
+            body.writer,
+            lease::writer_version()
+        )?,
+        Ok(lease::Outcome::Rejected { .. }) => {}
+        Err(reason) => writeln!(err, "::error:: lease: the stand-down request: {reason}")?,
+    }
+    Ok(())
+}
+
+/// Why this clone may not take a turn it is otherwise owed, or `None`.
+///
+/// The reading half of [`note_the_cooldown`], and the whole of the decision is
+/// [`lease::cooling`] — this resolves the two inputs and renders the reason.
+///
+/// **It resolves the ref the WRITER recorded**, never a constant of its own. A
+/// ref that will not read answers `None`: a could-not-look is not evidence this
+/// clone is still poisoned, and holding it back on one would strand a clone over
+/// a ref that was renamed or pruned.
+fn still_cooling(root: &Path, local: &lease::Local, observed: &lease::Observed) -> Option<String> {
+    let (tracking, at) = local.poisoned()?;
+    let trunk = git::resolve_ref(root, &tracking).ok().flatten()?;
+    // Pointer-only: the ref and the position, never the lease body.
+    lease::cooling(Some(&at), &trunk, observed).then(|| {
+        format!("cooling — this clone reddened the matrix on {tracking} at {at}, which is still trunk, and the pool is not idle")
+    })
+}
+
 /// `lease renew`: extend this clone's lease by one term.
 fn run_lease_renew(
     root: &Path,