car-sync 0.34.0

//! Execution lease + fencing (slice B5 of
//! `docs/proposals/multi-device-sync.md`, §"Deep dive: the execution-lease /
//! fencing protocol").
//!
//! **What this slice delivers: DETERMINISTIC LEDGER CONVERGENCE + a durable
//! idempotency oracle — NOT exactly-once execution.** The exactly-once
//! *execution* gate is B6's **dispatch fence**: a dispatch-time linearizable
//! "am I still epoch N?" read plus a durable non-fenced idempotency read
//! ([`crate::fold::SyncState::committed_run`]) **before** the external side
//! effect. That is necessary, not an optimization; the fold below gates the
//! *ledger*, not the side effect. Do not read "the fold fences the write" as
//! "the fold makes execution safe" — it does not.
//!
//! Replicating *state* is a leaderless CRDT fold — safe with any number of
//! concurrent writers (that is B1–B4). Performing *actions* is not: a tool
//! call has an external side effect and is not idempotent, so if two sites
//! both hold the agent and the same scheduled trigger fires, the email goes
//! out twice. B5 adds the two mechanisms the proposal specs for that:
//!
//! 1. **A separate [`LeaseCoordinator`] (liveness).** The crisp principle,
//!    from the proposal: *the lease is for liveness; only the fence is for
//!    safety.* A lease demands **linearizable compare-and-swap** — exactly
//!    one holder per agent at a time — which the eventually-consistent
//!    [`crate::relay::Relay`] structurally **cannot** provide (an `FsRelay`
//!    over a synced folder gives no consensus; two sites could each append "I
//!    take it" and both believe they won). So the lease lives behind its
//!    **own trait**, never bolted onto `Relay`. The proposal's data/control
//!    split: the relay stays a dumb, E2E-ciphertext ordered log for *content*;
//!    the lease register holds only non-sensitive metadata (`agent_id`,
//!    `holder`, `epoch`, `expiry`), so a coordinator can serialize it in
//!    cleartext without breaching E2E. [`InMemoryLeaseCoordinator`] is the
//!    honest in-process reference — `Arc<Mutex>` CAS *is* genuinely
//!    linearizable within one process (exactly as `InMemoryRelay` is the
//!    honest single-process relay). A distributed backend (Cosmos `if_match`,
//!    a single-writer daemon, Postgres advisory locks) is B6; the trait is its
//!    contract.
//!
//! 2. **Fencing as a fold property, over two views.** The `epoch` a successful
//!    acquire mints is the **monotone fencing token** (never reused). Since
//!    `car-sync` is the state layer, B5 realizes fencing as a **deterministic
//!    fold rule** over the leased [`crate::oplog::Surface::Intent`] surface,
//!    yielding two per-agent views ([`crate::fold::IntentAgent`]):
//!    - **`committed_runs` — the durable idempotency ORACLE.** Keep-all,
//!      **fence-independent**: a commit is a permanent fact, recorded whatever
//!      its epoch and never cleared by a later epoch or by compaction.
//!      [`crate::fold::SyncState::committed_run`] is the correct "did this run
//!      already execute?" lookup — the read a B6 dispatch fence performs before
//!      any external write.
//!    - **`runs` — the "who holds now" view.** Per-agent epoch fencing applies
//!      to *pending* intents (a stale zombie holder's pending is fenced,
//!      order-independently, without a wall-clock race — fencing beats HLC),
//!      while committed/failed records are **terminal-immune** (never reverted,
//!      never cleared). Read via [`crate::fold::SyncState::intent`]. This is
//!      NOT the idempotency oracle — a fenced pending is absent here even when
//!      the run committed under a prior epoch.
//!
//!    Two partitioned sites that both believe they hold the lease both write
//!    intents; after convergence the ledger converges deterministically (the
//!    higher-epoch pending wins `runs`; committed facts survive in
//!    `committed_runs`).
//!
//! **Idempotency ties B7.** A scheduled/triggered run's id is content-derived
//! (`car_proto::deterministic_run_id`), so two sites computing the same
//! occurrence produce the **same `run_id`**. An [`Intent`] keys on that
//! `run_id`, so a lease holder and a just-failed-over holder proposing the
//! *same* logical run collapse to one committed-oracle record.
//!
//! **Honesty (the proposal's tier-3 residual).** For an external resource that
//! doesn't understand CAR's tokens, exactly-once is impossible without that
//! resource's cooperation. B5 does not close that: it converges the ledger and
//! provides the durable committed-oracle read; the dispatch-time fence + the
//! oracle-read-before-effect (B6) are what make execution single-shot in the
//! common cases, with a documented bounded residual — not a false exactly-once
//! claim, and NOT something the fold alone provides.

use crate::oplog::WallClock;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::fmt;
use std::sync::{Arc, Mutex};

/// A granted lease — the proposal's `Lease { agent_id, holder, epoch,
/// expires_at }`. `epoch` is the **fencing token**: strictly monotone per
/// agent, never reused, minted on every successful [`LeaseCoordinator::acquire`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Lease {
    pub agent_id: String,
    /// The device that holds it.
    pub holder: String,
    /// Monotone fencing token (never reused). Threaded into every leased
    /// [`Intent`] this holder writes; the fold fences anything below the
    /// agent's max.
    pub epoch: u64,
    /// **Server-authoritative** expiry (coordinator wall-clock ms). Dodges
    /// client skew — the proposal's requirement. Expiry only enables
    /// *stealing*; the holder can [`LeaseCoordinator::renew`] until stolen.
    pub expires_at_ms: u64,
}

/// A lease-coordination failure.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LeaseError {
    /// [`LeaseCoordinator::acquire`] found a still-valid lease held by
    /// another (or the same) device — CAS failed. The proposal: "acquire
    /// succeeds only if unheld or expired."
    Held {
        agent_id: String,
        holder: String,
        epoch: u64,
        expires_at_ms: u64,
    },
    /// [`LeaseCoordinator::renew`]/[`LeaseCoordinator::release`] found the
    /// caller is no longer the holder at the claimed epoch — someone stole
    /// it (a higher epoch exists), or it was already released. The
    /// fencing-relevant signal: a zombie that can reach the coordinator
    /// learns here that it lost.
    Lost {
        agent_id: String,
        /// The epoch the caller claimed to hold.
        claimed_epoch: u64,
        /// The coordinator's current epoch for the agent (0 if never held).
        current_epoch: u64,
    },
    /// The coordinator backend itself failed — a *distributed* coordinator
    /// (e.g. [`crate::net_relay::NetworkLeaseCoordinator`]) could not reach the
    /// lease register or got an unusable reply. Distinct from `Held`/`Lost`,
    /// which are legitimate CAS verdicts, not failures. The in-process
    /// `InMemoryLeaseCoordinator` never returns this.
    Backend(String),
}

impl fmt::Display for LeaseError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            LeaseError::Held {
                agent_id,
                holder,
                epoch,
                expires_at_ms,
            } => write!(
                f,
                "lease for {agent_id} is held by {holder} at epoch {epoch} (expires at \
                 {expires_at_ms}ms) — acquire refused (unexpired)"
            ),
            LeaseError::Lost {
                agent_id,
                claimed_epoch,
                current_epoch,
            } => write!(
                f,
                "lease for {agent_id} claimed at epoch {claimed_epoch} is lost (coordinator is at \
                 epoch {current_epoch}) — you are no longer the holder"
            ),
            LeaseError::Backend(m) => write!(f, "lease coordinator backend error: {m}"),
        }
    }
}

impl std::error::Error for LeaseError {}

/// The lease-coordination contract — the **linearizable single-key register**
/// per agent the proposal requires (Cosmos `if_match` / a single-writer
/// daemon in B6). Distinct from [`crate::relay::Relay`] on purpose: `Relay` is
/// eventually consistent and cannot host a lease.
///
/// All methods take `&mut self` to mirror `Relay` and leave room for a backend
/// that mutates on read; the reference impl is `Clone` + `Send` + `Sync` so
/// several device handles can share one linearizable register.
///
/// # Contract every implementation MUST hold
///
/// 1. **Linearizable single-key CAS.** [`Self::acquire`] must be a genuine
///    compare-and-swap: given concurrent acquires on an unheld/expired agent,
///    exactly one succeeds. An eventually-consistent store (a synced folder, an
///    `FsRelay`-shaped backend) **cannot** provide this and must not back a
///    coordinator — two sites would each "win" the same acquire, the exact
///    split-brain the fence exists to bound.
/// 2. **Epoch monotonicity that OUTLIVES the process.** The fencing epoch must
///    never regress or repeat across restarts/failovers. This is load-bearing:
///    a coordinator that restarts and re-issues a *lower* epoch would let a
///    stale high-epoch write out-fence a fresh one. A durable backend persists
///    the counter; a stateless one MUST seed each agent's epoch from
///    **`max(max(epoch) over that agent's intent ops still in the oplog,
///    checkpoint.state.intents[agent].fencing_epoch)`** before granting.
///    **Both terms are required:** compaction (B4) truncates below-frontier
///    intent ops out of the journal, so `max(oplog ops)` alone UNDER-seeds
///    after a compaction — the checkpoint's persisted `fencing_epoch` is the
///    high-water mark the truncated ops left behind ([`crate::fold::IntentAgent`]
///    is carried keep-all in the checkpoint precisely so this seed survives).
///    **[`InMemoryLeaseCoordinator`] does NOT persist** — it is the honest
///    single-process reference (like `InMemoryRelay`), correct only while the
///    process lives.
/// 3. **Authority is the epoch, not the expiry.** `Lease::expires_at_ms` is a
///    **liveness hint only** — it merely enables *stealing*. No consumer may
///    read "expired" as license to act without holding the current epoch; the
///    only authority check is "am I still the holder at `epoch`?" via
///    [`Self::current`]/[`Self::renew`]. (The B6 dispatch fence reads the
///    current epoch here, not wall-clock expiry.)
pub trait LeaseCoordinator {
    /// Compare-and-swap acquire. Grants **iff** the agent is unheld or its
    /// lease has expired; on grant the epoch is bumped (`last_epoch + 1`,
    /// monotone, never reused) and `expires_at_ms = now + ttl_ms`. A
    /// still-valid lease (held by anyone, including the caller) returns
    /// [`LeaseError::Held`] — the holder keeps it alive with [`Self::renew`].
    fn acquire(
        &mut self,
        agent_id: &str,
        device_id: &str,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError>;

    /// Heartbeat: extend `expires_at_ms` **without** bumping the epoch. Ok iff
    /// the caller is still the current holder at `epoch` (even slightly past
    /// wall-clock expiry, as long as no one has stolen it — expiry only
    /// enables stealing). Otherwise [`LeaseError::Lost`].
    fn renew(
        &mut self,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError>;

    /// Clean release (hook this to the supervisor's graceful stop): iff the
    /// caller holds `epoch`, drop the holder so the next acquire grants
    /// immediately with the next epoch — no TTL wait. Otherwise
    /// [`LeaseError::Lost`]. The epoch counter is retained across a release so
    /// epochs never repeat.
    fn release(&mut self, agent_id: &str, device_id: &str, epoch: u64) -> Result<(), LeaseError>;

    /// Linearizable read of the current lease (the tier-3 fence's "am I still
    /// the holder at `epoch`?" check reads this). `None` when the agent is
    /// unheld (never acquired, or cleanly released). Note an *expired* but
    /// un-stolen lease still reports its holder — expiry alone doesn't clear
    /// the register, only a steal or release does.
    fn current(&mut self, agent_id: &str) -> Result<Option<Lease>, LeaseError>;
}

/// One agent's lease register cell.
#[derive(Debug, Clone, Default)]
struct Slot {
    /// The last epoch minted for this agent — monotone, retained across
    /// release so a re-acquire never reuses an epoch.
    epoch: u64,
    /// `Some` while held (until a steal or a release), `None` when unheld.
    holder: Option<String>,
    expires_at_ms: u64,
}

/// The honest in-process reference coordinator: a linearizable CAS register
/// behind `Arc<Mutex>` (genuinely linearizable within one process — the
/// mutex serializes every CAS), over an injected [`WallClock`] for
/// server-authoritative expiry. Clone shares the same register, so two device
/// handles contend over one linearizable cell — exactly the concurrency the
/// exclusivity/fencing tests exercise.
///
/// **This cannot be backed by an eventually-consistent store.** A synced-folder
/// / `FsRelay`-style backend gives no consensus, so two sites could each "win"
/// the same acquire — the split-brain the whole design exists to prevent. The
/// distributed coordinator (Cosmos `if_match`, single-writer daemon, etc.) is
/// B6; it must supply *real* single-key linearizability, and this type is the
/// contract it implements.
#[derive(Clone)]
pub struct InMemoryLeaseCoordinator {
    slots: Arc<Mutex<BTreeMap<String, Slot>>>,
    wall: WallClock,
}

impl fmt::Debug for InMemoryLeaseCoordinator {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("InMemoryLeaseCoordinator")
            .finish_non_exhaustive()
    }
}

impl InMemoryLeaseCoordinator {
    pub fn new(wall: WallClock) -> Self {
        Self {
            slots: Arc::new(Mutex::new(BTreeMap::new())),
            wall,
        }
    }

    fn lease_of(agent_id: &str, slot: &Slot) -> Option<Lease> {
        slot.holder.as_ref().map(|holder| Lease {
            agent_id: agent_id.to_string(),
            holder: holder.clone(),
            epoch: slot.epoch,
            expires_at_ms: slot.expires_at_ms,
        })
    }
}

impl LeaseCoordinator for InMemoryLeaseCoordinator {
    fn acquire(
        &mut self,
        agent_id: &str,
        device_id: &str,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError> {
        let now = (self.wall)();
        // The whole CAS runs under the lock — this is what makes acquire
        // linearizable (the honest reference).
        let mut slots = self.slots.lock().expect("lease mutex poisoned");
        let slot = slots.entry(agent_id.to_string()).or_default();
        let held = slot.holder.is_some() && now < slot.expires_at_ms;
        if held {
            return Err(LeaseError::Held {
                agent_id: agent_id.to_string(),
                holder: slot.holder.clone().expect("held ⇒ some holder"),
                epoch: slot.epoch,
                expires_at_ms: slot.expires_at_ms,
            });
        }
        slot.epoch += 1; // monotone fencing token, never reused
        slot.holder = Some(device_id.to_string());
        slot.expires_at_ms = now.saturating_add(ttl_ms);
        Ok(Self::lease_of(agent_id, slot).expect("just set holder"))
    }

    fn renew(
        &mut self,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError> {
        let now = (self.wall)();
        let mut slots = self.slots.lock().expect("lease mutex poisoned");
        let slot = slots.entry(agent_id.to_string()).or_default();
        let ours = slot.holder.as_deref() == Some(device_id) && slot.epoch == epoch;
        if !ours {
            return Err(LeaseError::Lost {
                agent_id: agent_id.to_string(),
                claimed_epoch: epoch,
                current_epoch: slot.epoch,
            });
        }
        // Renew regardless of wall-clock expiry: until someone steals (which
        // bumps the epoch) the holder keeps it. No epoch bump.
        slot.expires_at_ms = now.saturating_add(ttl_ms);
        Ok(Self::lease_of(agent_id, slot).expect("ours ⇒ some holder"))
    }

    fn release(&mut self, agent_id: &str, device_id: &str, epoch: u64) -> Result<(), LeaseError> {
        let mut slots = self.slots.lock().expect("lease mutex poisoned");
        let slot = slots.entry(agent_id.to_string()).or_default();
        let ours = slot.holder.as_deref() == Some(device_id) && slot.epoch == epoch;
        if !ours {
            return Err(LeaseError::Lost {
                agent_id: agent_id.to_string(),
                claimed_epoch: epoch,
                current_epoch: slot.epoch,
            });
        }
        slot.holder = None; // epoch retained → next acquire mints epoch+1
        Ok(())
    }

    fn current(&mut self, agent_id: &str) -> Result<Option<Lease>, LeaseError> {
        let slots = self.slots.lock().expect("lease mutex poisoned");
        Ok(slots
            .get(agent_id)
            .and_then(|slot| Self::lease_of(agent_id, slot)))
    }
}

// ---------------------------------------------------------------------------
// Intent — the leased execution-intent surface's typed payload + the fold-side
// field readers the deterministic fold uses to apply fencing.
// ---------------------------------------------------------------------------

/// Payload field names for [`crate::oplog::Surface::Intent`] ops. These are part of the
/// serialized payload the fold reads; keep them stable.
pub const INTENT_FIELD_ID: &str = "id";
pub const INTENT_FIELD_AGENT: &str = "agent_id";
pub const INTENT_FIELD_EPOCH: &str = "epoch";
pub const INTENT_FIELD_STATUS: &str = "status";

/// The monotone lifecycle of one execution intent: `pending` → `committed`
/// | `failed`. Monotone means a terminal state (either) supersedes `pending`
/// in the fold, and never regresses.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IntentStatus {
    Pending,
    Committed,
    Failed,
}

impl IntentStatus {
    /// Monotone rank: `pending` (0) < terminal (`committed`/`failed`, 1). The
    /// fold prefers the higher rank at a given epoch, so a `committed`/`failed`
    /// always supersedes an earlier `pending` for the same run.
    pub fn rank(self) -> u8 {
        match self {
            IntentStatus::Pending => 0,
            IntentStatus::Committed | IntentStatus::Failed => 1,
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            IntentStatus::Pending => "pending",
            IntentStatus::Committed => "committed",
            IntentStatus::Failed => "failed",
        }
    }
}

/// A leased execution intent — the proposal's write-ahead ledger entry, folded
/// under [`crate::oplog::FoldTier::Leased`]. `run_id` is the B7 deterministic run id (the
/// idempotency key), `epoch` the lease's fencing token.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Intent {
    pub agent_id: String,
    /// The B7 content-addressed run id — the fold's per-run stable key.
    pub run_id: String,
    /// The lease epoch this intent was written under — the fencing token.
    pub epoch: u64,
    pub status: IntentStatus,
}

impl Intent {
    pub fn new(
        agent_id: impl Into<String>,
        run_id: impl Into<String>,
        epoch: u64,
        status: IntentStatus,
    ) -> Self {
        Self {
            agent_id: agent_id.into(),
            run_id: run_id.into(),
            epoch,
            status,
        }
    }

    /// The op payload — `run_id` under `"id"` so the fold's stable key is
    /// `id:<run_id>` (reusing the standard [`crate::oplog::OpRecord::stable_key`]
    /// machinery), plus the fencing fields.
    pub fn payload(&self) -> Value {
        json!({
            INTENT_FIELD_ID: self.run_id,
            INTENT_FIELD_AGENT: self.agent_id,
            INTENT_FIELD_EPOCH: self.epoch,
            INTENT_FIELD_STATUS: self.status.as_str(),
        })
    }

    /// Parse an [`crate::oplog::Surface::Intent`] payload back into an [`Intent`]; `None`
    /// when a required field is missing or malformed.
    pub fn from_payload(payload: &Value) -> Option<Self> {
        let status = match payload.get(INTENT_FIELD_STATUS).and_then(Value::as_str)? {
            "pending" => IntentStatus::Pending,
            "committed" => IntentStatus::Committed,
            "failed" => IntentStatus::Failed,
            _ => return None,
        };
        Some(Self {
            agent_id: payload
                .get(INTENT_FIELD_AGENT)
                .and_then(Value::as_str)?
                .to_string(),
            run_id: payload
                .get(INTENT_FIELD_ID)
                .and_then(Value::as_str)?
                .to_string(),
            epoch: payload.get(INTENT_FIELD_EPOCH).and_then(Value::as_u64)?,
            status,
        })
    }
}

/// The agent an intent op belongs to — the fencing group. Missing/malformed
/// reads as `""` so the fold stays total (a well-formed [`Intent`] never
/// omits it).
pub fn intent_agent(payload: &Value) -> &str {
    payload
        .get(INTENT_FIELD_AGENT)
        .and_then(Value::as_str)
        .unwrap_or("")
}

/// The fencing epoch an intent op carries. Missing/malformed reads as `0`
/// (below any real lease epoch, which start at 1), so a malformed op is fenced
/// rather than promoted.
pub fn intent_epoch(payload: &Value) -> u64 {
    payload
        .get(INTENT_FIELD_EPOCH)
        .and_then(Value::as_u64)
        .unwrap_or(0)
}

/// The monotone status rank of an intent op ([`IntentStatus::rank`]); an
/// unknown/absent status reads as `pending` (0).
pub fn intent_status_rank(payload: &Value) -> u8 {
    match payload.get(INTENT_FIELD_STATUS).and_then(Value::as_str) {
        Some("committed") | Some("failed") => 1,
        _ => 0,
    }
}

/// Is this intent op **committed** specifically (not merely terminal)? Feeds
/// the fence-independent idempotency oracle ([`crate::fold::IntentAgent::committed_runs`])
/// — a `failed` run is terminal but NOT committed, so it may be retried.
pub fn intent_is_committed(payload: &Value) -> bool {
    payload.get(INTENT_FIELD_STATUS).and_then(Value::as_str) == Some("committed")
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
    use std::thread;

    fn manual_clock() -> (Arc<AtomicU64>, WallClock) {
        let t = Arc::new(AtomicU64::new(0));
        let reader = t.clone();
        (t, Arc::new(move || reader.load(Ordering::SeqCst)))
    }

    #[test]
    fn acquire_is_exclusive_and_bumps_epoch_never_reusing() {
        let (_t, wall) = manual_clock(); // clock frozen at 0 → nothing expires
        let mut coord = InMemoryLeaseCoordinator::new(wall);

        // First acquire wins epoch 1.
        let a = coord.acquire("agent-x", "dev-a", 100).unwrap();
        assert_eq!((a.epoch, a.holder.as_str()), (1, "dev-a"));

        // A second, concurrent acquire (clock frozen ⇒ not expired) loses —
        // exactly one holder.
        let err = coord.acquire("agent-x", "dev-b", 100).unwrap_err();
        assert!(
            matches!(err, LeaseError::Held { epoch: 1, .. }),
            "got {err:?}"
        );

        // Clean release, then the next acquire mints epoch 2 — no wait, no
        // reuse of epoch 1.
        coord.release("agent-x", "dev-a", 1).unwrap();
        assert_eq!(coord.current("agent-x").unwrap(), None, "released ⇒ unheld");
        let b = coord.acquire("agent-x", "dev-b", 100).unwrap();
        assert_eq!(
            (b.epoch, b.holder.as_str()),
            (2, "dev-b"),
            "epoch is monotone, never reused"
        );
    }

    #[test]
    fn renew_and_release_require_the_current_holder_and_epoch() {
        let (t, wall) = manual_clock();
        let mut coord = InMemoryLeaseCoordinator::new(wall);
        let lease = coord.acquire("a", "dev-a", 100).unwrap();
        assert_eq!(lease.expires_at_ms, 100);

        // Renew extends expiry, no epoch bump.
        t.store(50, Ordering::SeqCst);
        let renewed = coord.renew("a", "dev-a", 1, 100).unwrap();
        assert_eq!((renewed.epoch, renewed.expires_at_ms), (1, 150));

        // Wrong epoch, wrong holder → Lost, both ways.
        assert!(matches!(
            coord.renew("a", "dev-a", 99, 100),
            Err(LeaseError::Lost {
                current_epoch: 1,
                ..
            })
        ));
        assert!(matches!(
            coord.renew("a", "dev-b", 1, 100),
            Err(LeaseError::Lost { .. })
        ));
        assert!(matches!(
            coord.release("a", "dev-b", 1),
            Err(LeaseError::Lost { .. })
        ));

        // Release by the real holder succeeds; a later renew is Lost.
        coord.release("a", "dev-a", 1).unwrap();
        assert!(matches!(
            coord.renew("a", "dev-a", 1, 100),
            Err(LeaseError::Lost { .. })
        ));
    }

    #[test]
    fn renew_works_past_expiry_until_actually_stolen() {
        let (t, wall) = manual_clock();
        let mut coord = InMemoryLeaseCoordinator::new(wall);
        coord.acquire("a", "dev-a", 100).unwrap();

        // Past wall-clock expiry but nobody stole it → renew still works
        // (expiry only *enables* stealing; the proposal's Azure-blob-lease
        // semantics).
        t.store(500, Ordering::SeqCst);
        assert!(
            coord.renew("a", "dev-a", 1, 100).is_ok(),
            "renew works until stolen"
        );
    }

    #[test]
    fn expired_lease_is_stealable_and_fences_the_old_epoch() {
        let (t, wall) = manual_clock();
        let mut coord = InMemoryLeaseCoordinator::new(wall);
        let old = coord.acquire("a", "dev-a", 100).unwrap();
        assert_eq!(old.epoch, 1);

        // dev-a pauses past its TTL; dev-b steals — epoch bumps to 2.
        t.store(200, Ordering::SeqCst);
        let stolen = coord.acquire("a", "dev-b", 100).unwrap();
        assert_eq!((stolen.epoch, stolen.holder.as_str()), (2, "dev-b"));

        // The zombie dev-a, if it can reach the coordinator, learns it lost.
        assert!(matches!(
            coord.renew("a", "dev-a", 1, 100),
            Err(LeaseError::Lost {
                claimed_epoch: 1,
                current_epoch: 2,
                ..
            })
        ));
        let cur = coord.current("a").unwrap().unwrap();
        assert_eq!((cur.epoch, cur.holder.as_str()), (2, "dev-b"));
    }

    #[test]
    fn reference_impl_is_linearizable_under_concurrent_contention() {
        // Many threads sharing ONE register hammer acquire→work→release. The
        // linearizable CAS must guarantee: at most one holder at any instant,
        // and every granted epoch is unique and monotone (never reused).
        let (_t, wall) = manual_clock(); // frozen ⇒ no expiry ⇒ every hold ends only by release
        let coord = InMemoryLeaseCoordinator::new(wall);
        let concurrent = Arc::new(AtomicUsize::new(0));
        let max_concurrent = Arc::new(AtomicUsize::new(0));
        let granted = Arc::new(Mutex::new(Vec::<u64>::new()));

        let threads: Vec<_> = (0..8)
            .map(|i| {
                let mut coord = coord.clone();
                let concurrent = concurrent.clone();
                let max_concurrent = max_concurrent.clone();
                let granted = granted.clone();
                let device = format!("dev-{i}");
                thread::spawn(move || {
                    for _ in 0..50 {
                        // Spin until we win the lease.
                        let lease = loop {
                            match coord.acquire("agent", &device, 1_000_000) {
                                Ok(l) => break l,
                                Err(LeaseError::Held { .. }) => thread::yield_now(),
                                Err(e) => panic!("unexpected {e:?}"),
                            }
                        };
                        // Critical section: assert we are the sole holder.
                        let now = concurrent.fetch_add(1, Ordering::SeqCst) + 1;
                        max_concurrent.fetch_max(now, Ordering::SeqCst);
                        granted.lock().unwrap().push(lease.epoch);
                        // A little real overlap pressure.
                        for _ in 0..20 {
                            std::hint::spin_loop();
                        }
                        concurrent.fetch_sub(1, Ordering::SeqCst);
                        coord.release("agent", &device, lease.epoch).unwrap();
                    }
                })
            })
            .collect();
        for th in threads {
            th.join().unwrap();
        }

        assert_eq!(
            max_concurrent.load(Ordering::SeqCst),
            1,
            "the CAS is linearizable: never two holders at once"
        );
        // Every successful acquire bumped by exactly 1, so grants are exactly
        // 1..=K with no repeats.
        let mut epochs = granted.lock().unwrap().clone();
        let count = epochs.len();
        epochs.sort_unstable();
        epochs.dedup();
        assert_eq!(epochs.len(), count, "no epoch was ever reused");
        assert_eq!(
            epochs,
            (1..=count as u64).collect::<Vec<_>>(),
            "epochs are dense and monotone"
        );
    }

    #[test]
    fn intent_payload_round_trips_and_ranks() {
        let intent = Intent::new("milo", "run-abc", 3, IntentStatus::Committed);
        let payload = intent.payload();
        assert_eq!(
            payload[INTENT_FIELD_ID],
            json!("run-abc"),
            "run_id lives under id"
        );
        assert_eq!(Intent::from_payload(&payload), Some(intent));

        // Fold-side readers.
        assert_eq!(intent_agent(&payload), "milo");
        assert_eq!(intent_epoch(&payload), 3);
        assert_eq!(intent_status_rank(&payload), 1);
        assert_eq!(
            intent_status_rank(&Intent::new("m", "r", 1, IntentStatus::Pending).payload()),
            0
        );

        // Monotone status ranking.
        assert!(IntentStatus::Pending.rank() < IntentStatus::Committed.rank());
        assert_eq!(IntentStatus::Committed.rank(), IntentStatus::Failed.rank());

        // Missing fields → total, fenceable defaults (never a panic).
        assert_eq!(intent_epoch(&json!({})), 0);
        assert_eq!(intent_agent(&json!({})), "");
        assert_eq!(Intent::from_payload(&json!({"id": "r"})), None);
    }
}