polyc_runtime/

warm_swap.rs

//! Warm drain-swap-resume — the click-to-restart path for a wire- and
//! schema-compatible binary swap (see the update PRD, [`crate::compat`], and
//! [`crate::stager`]).
//!
//! This is the "click to restart" layer. A [`Compatibility::Warm`] release is a
//! new binary whose fingerprint is otherwise identical to the running one — only
//! binary internals moved — so it cannot be picked up at a turn boundary the way
//! a hot config bundle is ([`crate::hot_reload`]); the process itself has to be
//! replaced. But a live conversation must not be yanked mid-turn. This module
//! orchestrates the smallest safe restart: stop admitting new turns, let
//! in-flight turns and paused approvals reach a committed checkpoint, release the
//! single-writer lease, swap the process or image, re-acquire the lease, and
//! resume from the last committed turn.
//!
//! # It orchestrates; it does not reinvent
//!
//! Every invariant this path relies on is already built and tested elsewhere —
//! the job here is to sequence those primitives, not to duplicate them:
//!
//! - **The commit marker is the drain point.** A turn is persisted to the event
//!   log — bracketed by its `turn_start`/`turn_complete` pair — before the
//!   harness is dialed, and a turn missing its `turn_complete` is discarded on
//!   replay (the control plane's `filter_committed_turns`). So "drain" is not a
//!   new barrier: it is waiting for the in-flight turn to reach its existing
//!   commit marker, which doubles as the rollback net.
//! - **The lease is the existing single-writer guard.** The per-conversation
//!   `coordination.k8s.io/v1` Lease already enforces exactly one writer; a warm
//!   swap releases it before the swap and re-acquires it after, so the swapped-in
//!   process is the sole writer with no split-brain window.
//! - **Resume is the existing replay path.** Restarting from the last committed
//!   turn — re-driving any suspended HITL approval — is the same reconstruct the
//!   control plane runs on every reconnect.
//! - **The swap core reuses the stager wholesale.** Verifying the new binary's
//!   signature, applying the pointer flip, health-checking it, and auto-rolling
//!   back on failure are [`crate::stager`]'s job; the only bespoke piece is the
//!   [`Activator`], which for a warm swap is a process or image swap.
//!
//! Because those primitives live in the control plane (a container) while this
//! crate is a foundation component, the orchestrator is expressed over injectable
//! seams — the same shape [`crate::stager`] and [`crate::hot_reload`] use. The
//! control plane wires each seam to its real drain / lease / resume code; the
//! state machine and its ordering invariant are exercised here without a cluster.
//!
//! # The classifier gate
//!
//! Only a release that classifies [`Compatibility::Warm`] takes this path.
//! [`ensure_warm`] refuses every other verdict: a [`Compatibility::Hot`] change
//! reloads at the turn boundary with no restart, a [`Compatibility::Cold`] change
//! needs a coordinated fleet redeploy, and an
//! [`Incompatible`](crate::compat::Compatibility::Incompatible) bundle was
//! authored against a runtime this build cannot satisfy. The gate runs before any
//! seam is touched, so a non-warm release never closes admission, drains, or
//! releases the lease.
//!
//! # The ordering invariant
//!
//! [`WarmSwap::run`] walks a fixed sequence, and the ordering is the whole point:
//!
//! ```text
//!   ensure_warm ─▶ close admission ─▶ drain to committed checkpoint
//!                                              │
//!                                              ▼
//!            release lease ─▶ swap (stager: verify→apply→health→rollback)
//!                                              │
//!                                              ▼
//!            re-acquire lease ─▶ resume from the last committed turn
//! ```
//!
//! - No new turn is admitted once `close` runs, which is *before* the drain
//!   begins.
//! - In-flight turns and paused approvals reach a committed checkpoint *before*
//!   the swap — the checkpoint is captured while the lease is still held.
//! - The lease is released *before* it is re-acquired, so there is never a moment
//!   with two writers.
//! - Resume restarts from exactly the drained checkpoint, so no committed turn is
//!   dropped and none is replayed twice; a paused approval captured in the
//!   checkpoint is re-driven.
//!
//! The lease is re-acquired and the conversation is resumed on *whichever* binary
//! is live after the swap — the new one on a committed swap, the previous one if
//! the health check rolled back — so a conversation is never left drained-but-dead
//! behind a failed swap.

use thiserror::Error;

use crate::compat::Compatibility;
use crate::stager::{
    Activator, HealthCheck, Outcome, ReleaseId, SignatureVerifier, StageError, Stager, UpdateSource,
};

/// The drained resume point: the last fully-committed turn plus the paused
/// approvals that must survive the swap.
///
/// Captured by [`Drain::drain`] while the lease is still held, and threaded into
/// [`Resume::resume`] so the swapped-in process restarts from exactly here. Turns
/// that never reached their `turn_complete` marker are not part of it — they are
/// the interrupted turns replay discards — so resuming from this point neither
/// drops a committed turn nor replays an uncommitted one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Checkpoint {
    /// The id of the last fully-committed turn — the point resume restarts from.
    last_committed_turn: String,
    /// How many HITL approvals were paused at the checkpoint. They are persisted
    /// with the turn, so they survive the swap and are re-driven on resume.
    paused_approvals: usize,
}

impl Checkpoint {
    /// Build a checkpoint from the last committed turn id and the count of paused
    /// approvals carried across the swap.
    #[must_use]
    pub fn new(last_committed_turn: impl Into<String>, paused_approvals: usize) -> Self {
        Self {
            last_committed_turn: last_committed_turn.into(),
            paused_approvals,
        }
    }

    /// The id of the last fully-committed turn — the resume restart point.
    #[must_use]
    pub fn last_committed_turn(&self) -> &str {
        &self.last_committed_turn
    }

    /// How many paused HITL approvals were captured at the checkpoint and must be
    /// re-driven after the swap.
    #[must_use]
    pub const fn paused_approvals(&self) -> usize {
        self.paused_approvals
    }
}

/// What resume did after the swap: where it restarted and how many paused
/// approvals it re-drove.
///
/// [`WarmSwap::run`] checks this against the drained [`Checkpoint`] so a seam that
/// restarts from the wrong turn — dropping or duplicating a committed turn — is a
/// hard error rather than a silent divergence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResumeReport {
    /// The turn id resume restarted from. Must equal the checkpoint's
    /// [`Checkpoint::last_committed_turn`].
    resumed_from: String,
    /// How many paused approvals were re-driven on resume.
    redriven_approvals: usize,
}

impl ResumeReport {
    /// Build a resume report from the turn resumed and the approvals re-driven.
    #[must_use]
    pub fn new(resumed_from: impl Into<String>, redriven_approvals: usize) -> Self {
        Self {
            resumed_from: resumed_from.into(),
            redriven_approvals,
        }
    }

    /// The turn id resume restarted from.
    #[must_use]
    pub fn resumed_from(&self) -> &str {
        &self.resumed_from
    }

    /// How many paused approvals resume re-drove.
    #[must_use]
    pub const fn redriven_approvals(&self) -> usize {
        self.redriven_approvals
    }
}

/// The result of a warm swap that got as far as resuming.
///
/// Carries the swap [`Outcome`] (committed onto the new binary, or rolled back to
/// the previous one), the drained [`Checkpoint`], and the [`ResumeReport`]. The
/// swap outcome is honest about a rollback, so the surface presenting the restart
/// can say which binary is actually serving.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WarmSwapReport {
    /// What the swap did — [`Outcome::Committed`] onto the new binary or
    /// [`Outcome::RolledBack`] to the previous one after a failed health check.
    pub swap: Outcome,
    /// The point the conversation drained to and resumed from.
    pub checkpoint: Checkpoint,
    /// Where resume restarted and how many approvals it re-drove.
    pub resume: ResumeReport,
}

/// Why a warm swap was refused or could not complete.
#[derive(Debug, Error)]
pub enum WarmSwapError {
    /// The release did not classify [`Compatibility::Warm`], so it does not take
    /// the click-to-restart path. A hot change reloads with no restart, a cold
    /// change needs a coordinated redeploy, and an incompatible bundle was built
    /// for a runtime this binary cannot satisfy.
    #[error("refused: only a warm binary swap takes the click-to-restart path ({})", verdict_label(.0))]
    NotWarm(Compatibility),
    /// Admission could not be closed, so new turns cannot be reliably stopped and
    /// the swap is abandoned before the lease is touched.
    #[error("could not stop admitting new turns: {0}")]
    Admission(String),
    /// The drain could not reach a committed checkpoint, so the swap is abandoned
    /// with the lease still held — nothing was released or swapped.
    #[error("draining to a committed checkpoint failed: {0}")]
    Drain(String),
    /// A lease operation failed — either releasing the lease before the swap or
    /// re-acquiring it after.
    #[error("lease {op} failed: {reason}")]
    Lease {
        /// Which lease operation failed: `"release"` or `"re-acquire"`.
        op: &'static str,
        /// The underlying failure.
        reason: String,
    },
    /// The stager could not verify or apply the binary swap. Wraps the underlying
    /// [`StageError`] — most importantly [`StageError::Unverified`], which means
    /// the new binary's signature did not verify and nothing was swapped.
    #[error(transparent)]
    Stage(#[from] StageError),
    /// Resume could not restart the conversation from the checkpoint after the
    /// swap. The lease has already been re-acquired, so this is a resume failure
    /// on a live binary, not a lost lease.
    #[error("resuming from the last committed turn failed: {0}")]
    Resume(String),
    /// Resume restarted from a different turn than the one drained — a dropped or
    /// duplicated turn. Named separately from [`Self::Resume`] because it is a
    /// correctness violation, not a transport failure.
    #[error(
        "resume restarted from turn {resumed_from:?} but drained to {expected:?} — \
         a committed turn would be dropped or replayed"
    )]
    ResumeMismatch {
        /// The turn resume actually restarted from.
        resumed_from: String,
        /// The checkpoint turn it should have restarted from.
        expected: String,
    },
}

/// A short, plain-language name for a refused verdict, for [`WarmSwapError`].
const fn verdict_label(verdict: &Compatibility) -> &'static str {
    match verdict {
        Compatibility::Warm => "warm",
        Compatibility::Hot => "a config change that reloads with no restart",
        Compatibility::Cold => "a format change that needs a coordinated redeploy",
        Compatibility::Incompatible(_) => "built for a different runtime",
    }
}

/// Gate a classification verdict onto the warm path: [`Ok`] only for
/// [`Compatibility::Warm`].
///
/// This is the classifier interlock in one place — the swap proceeds only when
/// the change is a wire- and schema-compatible binary swap. Any other verdict
/// (`Hot`, `Cold`, or `Incompatible`) is refused before any seam is touched.
///
/// # Errors
///
/// Returns [`WarmSwapError::NotWarm`] carrying the refused verdict for anything
/// other than [`Compatibility::Warm`].
pub fn ensure_warm(verdict: &Compatibility) -> Result<(), WarmSwapError> {
    if *verdict == Compatibility::Warm {
        Ok(())
    } else {
        Err(WarmSwapError::NotWarm(verdict.clone()))
    }
}

/// Stops admitting new turns — the readiness flip / turn-admission gate.
///
/// Called once, before the drain, so no turn is admitted while the conversation
/// drains. In production this flips the readiness signal the edge and the
/// admission check read, so the load balancer stops routing new turns here.
pub trait AdmissionGate {
    /// Stop admitting new turns.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::Admission`] when admission cannot be closed.
    fn close(&self) -> Result<(), WarmSwapError>;
}

impl<F> AdmissionGate for F
where
    F: Fn() -> Result<(), WarmSwapError>,
{
    fn close(&self) -> Result<(), WarmSwapError> {
        self()
    }
}

/// Drains in-flight turns and paused approvals to a committed checkpoint.
///
/// Runs while the lease is still held. It waits for the in-flight turn to reach
/// its `turn_complete` marker — the commit point every turn is already bracketed
/// by — and captures the last committed turn plus any paused approvals as a
/// [`Checkpoint`]. It never forces a partial turn to commit: an interrupted turn
/// is simply excluded, exactly as replay would discard it.
pub trait Drain {
    /// Drain to a committed checkpoint and return it.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::Drain`] when the conversation cannot reach a
    /// committed checkpoint (for example a journal sync failed).
    fn drain(&self) -> Result<Checkpoint, WarmSwapError>;
}

impl<F> Drain for F
where
    F: Fn() -> Result<Checkpoint, WarmSwapError>,
{
    fn drain(&self) -> Result<Checkpoint, WarmSwapError> {
        self()
    }
}

/// Releases and re-acquires the single-writer per-conversation lease.
///
/// A warm swap releases the lease after draining and re-acquires it after the
/// swap, so the swapped-in process is the sole writer with no split-brain window.
/// In production this wraps the `coordination.k8s.io/v1` Lease: `release` is the
/// lease handle's drop (clears `holderIdentity`) and `reacquire` is a fresh
/// `try_acquire`.
pub trait LeaseControl {
    /// Release the lease held for this conversation.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::Lease`] (with `op = "release"`) when the lease
    /// cannot be released.
    fn release(&self) -> Result<(), WarmSwapError>;

    /// Re-acquire the lease for this conversation.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::Lease`] (with `op = "re-acquire"`) when the lease
    /// cannot be re-acquired.
    fn reacquire(&self) -> Result<(), WarmSwapError>;
}

/// Resumes the conversation from the last committed turn after the swap.
///
/// Given the drained [`Checkpoint`], it replays from the last committed turn and
/// re-drives any paused HITL approval — the same reconstruct the control plane
/// runs on a reconnect. It restarts from exactly the checkpoint turn, so no
/// committed turn is dropped and none is replayed twice.
pub trait Resume {
    /// Resume from `checkpoint` and report where it restarted.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::Resume`] when the conversation cannot be resumed.
    fn resume(&self, checkpoint: &Checkpoint) -> Result<ResumeReport, WarmSwapError>;
}

impl<F> Resume for F
where
    F: Fn(&Checkpoint) -> Result<ResumeReport, WarmSwapError>,
{
    fn resume(&self, checkpoint: &Checkpoint) -> Result<ResumeReport, WarmSwapError> {
        self(checkpoint)
    }
}

/// The warm drain-swap-resume orchestrator over its four control seams.
///
/// Holds the admission gate, the drain, the lease control, and the resume seam.
/// [`WarmSwap::run`] drives the fixed sequence, delegating the swap core (verify →
/// apply → health check → auto-rollback) to a [`Stager`] whose [`Activator`] is a
/// process or image swap. The orchestrator adds only the ordering: close before
/// drain, drain before release, release before swap, swap before re-acquire,
/// re-acquire before resume.
pub struct WarmSwap<G, D, L, R> {
    gate: G,
    drain: D,
    lease: L,
    resume: R,
}

impl<G, D, L, R> WarmSwap<G, D, L, R>
where
    G: AdmissionGate,
    D: Drain,
    L: LeaseControl,
    R: Resume,
{
    /// Build a warm-swap orchestrator over its four control seams.
    pub const fn new(gate: G, drain: D, lease: L, resume: R) -> Self {
        Self {
            gate,
            drain,
            lease,
            resume,
        }
    }

    /// Run the warm drain-swap-resume for a `Warm`-classified `release`.
    ///
    /// The sequence is fixed and its ordering is the invariant:
    ///
    /// 1. gate on the classifier — refuse anything but [`Compatibility::Warm`]
    ///    before touching any seam;
    /// 2. close admission, so no new turn is admitted while draining;
    /// 3. drain to a committed checkpoint — in-flight turns and paused approvals
    ///    reach their `turn_complete` marker, captured as a [`Checkpoint`], while
    ///    the lease is still held;
    /// 4. release the lease;
    /// 5. swap via the [`Stager`] — verify the new binary's signature, apply the
    ///    pointer flip, health-check, and auto-roll-back on failure;
    /// 6. re-acquire the lease — always, on whichever binary is now live, so the
    ///    conversation is never left drained-but-dead;
    /// 7. resume from the checkpoint, re-driving paused approvals, and verify it
    ///    restarted from exactly the drained turn.
    ///
    /// The lease re-acquire in step 6 runs even when the swap rolled back or its
    /// signature was refused, so a failed swap surfaces its error only after the
    /// conversation is back under a held lease.
    ///
    /// # Errors
    ///
    /// Returns [`WarmSwapError::NotWarm`] when `verdict` is not warm (nothing is
    /// drained or swapped); [`WarmSwapError::Admission`] or
    /// [`WarmSwapError::Drain`] when the pre-swap steps fail (the lease is still
    /// held); [`WarmSwapError::Lease`] when a release or re-acquire fails;
    /// [`WarmSwapError::Stage`] when the stager cannot verify or apply the swap
    /// (surfaced after the lease is re-acquired); and [`WarmSwapError::Resume`] or
    /// [`WarmSwapError::ResumeMismatch`] when resume fails or restarts from the
    /// wrong turn.
    pub fn run<S, V, A, H>(
        &self,
        verdict: &Compatibility,
        stager: &mut Stager<S, V, A, H>,
        release: &ReleaseId,
    ) -> Result<WarmSwapReport, WarmSwapError>
    where
        S: UpdateSource,
        V: SignatureVerifier,
        A: Activator,
        H: HealthCheck,
    {
        // 1. Classifier gate — no seam is touched for a non-warm release.
        ensure_warm(verdict)?;

        // 2. Stop admitting new turns BEFORE the drain begins.
        self.gate.close()?;

        // 3. Drain to a committed checkpoint while the lease is still held. This
        //    is the existing commit marker doubling as the drain point and the
        //    rollback net — an interrupted turn is excluded, never forced.
        let checkpoint = self.drain.drain()?;

        // 4. Release the lease before the swap, so the re-acquire in step 6 never
        //    races a still-held writer.
        self.lease
            .release()
            .map_err(|e| relabel_lease(e, "release"))?;

        // 5. Swap: reuse the stager wholesale for verify → apply → health →
        //    auto-rollback. Capture the result rather than `?`-returning it, so
        //    the lease is re-acquired even when the swap is refused or rolls back.
        let swap = stager.stage_and_apply(release);

        // 6. Re-acquire the lease on whichever binary is now live, ALWAYS — a
        //    failed swap must not leave the conversation drained-but-dead.
        self.lease
            .reacquire()
            .map_err(|e| relabel_lease(e, "re-acquire"))?;

        // A swap that could not even apply (unverified binary, failed flip) is a
        // hard error — surfaced now that the lease is back and the previous binary
        // is live for the caller's retry. We never paper the swap error over.
        let swap = swap?;

        // 7. Resume from exactly the drained checkpoint, re-driving paused
        //    approvals, and verify no committed turn was dropped or duplicated.
        let resume = self.resume.resume(&checkpoint)?;
        if resume.resumed_from() != checkpoint.last_committed_turn() {
            return Err(WarmSwapError::ResumeMismatch {
                resumed_from: resume.resumed_from().to_owned(),
                expected: checkpoint.last_committed_turn().to_owned(),
            });
        }

        Ok(WarmSwapReport {
            swap,
            checkpoint,
            resume,
        })
    }
}

/// Stamp the failing lease operation onto a [`WarmSwapError::Lease`] so the error
/// names whether the release or the re-acquire failed, regardless of how the seam
/// built it.
fn relabel_lease(err: WarmSwapError, op: &'static str) -> WarmSwapError {
    match err {
        WarmSwapError::Lease { reason, .. } => WarmSwapError::Lease { op, reason },
        other => other,
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use std::cell::RefCell;
    use std::path::PathBuf;
    use std::rc::Rc;

    use super::*;
    use crate::compat::{Incompatibility, RuntimeTarget, StagedBundle};
    use crate::stager::{Health, StagedArtifact};

    /// A shared ordered log of every step across all seams, so a test can assert
    /// the exact sequence a warm swap walked — the deterministic-runtime style of
    /// the interrupted-turn / resume tests.
    type Trace = Rc<RefCell<Vec<String>>>;

    fn trace() -> Trace {
        Rc::new(RefCell::new(Vec::new()))
    }

    fn log(trace: &Trace, step: impl Into<String>) {
        trace.borrow_mut().push(step.into());
    }

    /// An admission gate that records the close and exposes whether a turn would
    /// still be admitted — so a test can prove no turn is admitted after close.
    #[derive(Clone)]
    struct RecordingGate {
        trace: Trace,
        open: Rc<RefCell<bool>>,
    }

    impl RecordingGate {
        fn new(trace: &Trace) -> Self {
            Self {
                trace: Rc::clone(trace),
                open: Rc::new(RefCell::new(true)),
            }
        }

        /// Whether a new turn would be admitted right now.
        fn admits(&self) -> bool {
            *self.open.borrow()
        }
    }

    impl AdmissionGate for RecordingGate {
        fn close(&self) -> Result<(), WarmSwapError> {
            *self.open.borrow_mut() = false;
            log(&self.trace, "close");
            Ok(())
        }
    }

    /// A drain over a tiny model event log: a list of `(turn_id, committed)`. It
    /// records the drain, asserts admission is already closed, and returns the
    /// last COMMITTED turn — an interrupted turn is excluded exactly as
    /// `filter_committed_turns` would discard it.
    #[derive(Clone)]
    struct RecordingDrain {
        trace: Trace,
        gate: RecordingGate,
        log: Vec<(&'static str, bool)>,
        paused_approvals: usize,
    }

    impl RecordingDrain {
        fn new(
            trace: &Trace,
            gate: &RecordingGate,
            model_log: Vec<(&'static str, bool)>,
            paused_approvals: usize,
        ) -> Self {
            Self {
                trace: Rc::clone(trace),
                gate: gate.clone(),
                log: model_log,
                paused_approvals,
            }
        }
    }

    impl Drain for RecordingDrain {
        fn drain(&self) -> Result<Checkpoint, WarmSwapError> {
            // Admission MUST already be closed before the drain begins.
            assert!(
                !self.gate.admits(),
                "drain began while new turns were still admitted"
            );
            log(&self.trace, "drain");
            // The resume point is the last fully-committed turn; an uncommitted
            // (interrupted) turn is never the checkpoint.
            let last_committed = self
                .log
                .iter()
                .rev()
                .find_map(|(id, committed)| committed.then_some(*id))
                .expect("model log has at least one committed turn");
            Ok(Checkpoint::new(last_committed, self.paused_approvals))
        }
    }

    /// Lease control that records release / re-acquire and tracks held state, so a
    /// test can prove the lease was released before it was re-acquired and never
    /// double-held.
    #[derive(Clone)]
    struct RecordingLease {
        trace: Trace,
        held: Rc<RefCell<bool>>,
        fail_release: bool,
        fail_reacquire: bool,
    }

    impl RecordingLease {
        fn new(trace: &Trace) -> Self {
            Self {
                trace: Rc::clone(trace),
                held: Rc::new(RefCell::new(true)),
                fail_release: false,
                fail_reacquire: false,
            }
        }

        fn failing_reacquire(trace: &Trace) -> Self {
            Self {
                fail_reacquire: true,
                ..Self::new(trace)
            }
        }

        fn is_held(&self) -> bool {
            *self.held.borrow()
        }
    }

    impl LeaseControl for RecordingLease {
        fn release(&self) -> Result<(), WarmSwapError> {
            if self.fail_release {
                return Err(WarmSwapError::Lease {
                    op: "release",
                    reason: "api timeout".to_owned(),
                });
            }
            assert!(self.is_held(), "released a lease that was not held");
            *self.held.borrow_mut() = false;
            log(&self.trace, "lease.release");
            Ok(())
        }

        fn reacquire(&self) -> Result<(), WarmSwapError> {
            if self.fail_reacquire {
                return Err(WarmSwapError::Lease {
                    op: "re-acquire",
                    reason: "held by another".to_owned(),
                });
            }
            assert!(
                !self.is_held(),
                "re-acquired a lease that was never released — split-brain writer"
            );
            *self.held.borrow_mut() = true;
            log(&self.trace, "lease.reacquire");
            Ok(())
        }
    }

    /// A resume seam that records the restart, re-drives the paused approvals from
    /// the checkpoint, and (unless told to drift) restarts from exactly the
    /// checkpoint turn.
    #[derive(Clone)]
    struct RecordingResume {
        trace: Trace,
        gate: RecordingGate,
        lease: RecordingLease,
        drift_to: Option<&'static str>,
    }

    impl RecordingResume {
        fn new(trace: &Trace, gate: &RecordingGate, lease: &RecordingLease) -> Self {
            Self {
                trace: Rc::clone(trace),
                gate: gate.clone(),
                lease: lease.clone(),
                drift_to: None,
            }
        }

        fn drifting(
            trace: &Trace,
            gate: &RecordingGate,
            lease: &RecordingLease,
            to: &'static str,
        ) -> Self {
            Self {
                drift_to: Some(to),
                ..Self::new(trace, gate, lease)
            }
        }
    }

    impl Resume for RecordingResume {
        fn resume(&self, checkpoint: &Checkpoint) -> Result<ResumeReport, WarmSwapError> {
            // Resume runs only under a re-held lease, on a live binary.
            assert!(self.lease.is_held(), "resumed without a re-acquired lease");
            log(
                &self.trace,
                format!("resume@{}", checkpoint.last_committed_turn()),
            );
            // The swapped-in process comes back ready to admit turns again.
            *self.gate.open.borrow_mut() = true;
            let from = self.drift_to.unwrap_or(checkpoint.last_committed_turn());
            Ok(ResumeReport::new(from, checkpoint.paused_approvals()))
        }
    }

    /// A recording swap activator (the process / image swap), sharing its flip log
    /// with the trace.
    #[derive(Clone)]
    struct SwapActivator {
        trace: Trace,
    }

    impl SwapActivator {
        fn new(trace: &Trace) -> Self {
            Self {
                trace: Rc::clone(trace),
            }
        }
    }

    impl Activator for SwapActivator {
        fn activate(&self, release: &ReleaseId) -> Result<(), StageError> {
            log(&self.trace, format!("swap:{release}"));
            Ok(())
        }
    }

    fn artifact_for(release: &ReleaseId) -> StagedArtifact {
        StagedArtifact {
            release: release.clone(),
            staged_path: PathBuf::from(format!("/var/lib/polychrome/staged/{release}")),
            bundle: StagedBundle::new(
                RuntimeTarget::new(3, 7, "polychrome.uno/v1"),
                format!("catalog-{release}"),
            ),
            signed_bytes: format!("bytes-of-{release}").into_bytes(),
            signature: vec![0xAB; 4],
            signer_public_key: vec![0xCD; 4],
        }
    }

    fn ok_source(release: &ReleaseId) -> Result<StagedArtifact, StageError> {
        Ok(artifact_for(release))
    }

    // --- The classifier gate ---------------------------------------------------

    #[test]
    fn ensure_warm_admits_only_warm() {
        assert!(ensure_warm(&Compatibility::Warm).is_ok());
        for verdict in [
            Compatibility::Hot,
            Compatibility::Cold,
            Compatibility::Incompatible(Incompatibility::Wire),
        ] {
            let err = ensure_warm(&verdict).unwrap_err();
            assert!(
                matches!(err, WarmSwapError::NotWarm(v) if v == verdict),
                "non-warm verdict must be refused: {verdict:?}",
            );
        }
    }

    #[test]
    fn a_non_warm_release_touches_no_seam() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let drain = RecordingDrain::new(&trace, &gate, vec![("t1", true)], 0);
        let lease = RecordingLease::new(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate.clone(), drain, lease.clone(), resume);

        let mut stager = Stager::new(
            |_: &ReleaseId| -> Result<StagedArtifact, StageError> {
                panic!("download must not run for a non-warm release")
            },
            |_: &StagedArtifact| panic!("verify must not run for a non-warm release"),
            SwapActivator::new(&trace),
            || panic!("health check must not run for a non-warm release"),
            ReleaseId::new("v1"),
        );

        // A HOT release must not take the warm path.
        let err = swap
            .run(&Compatibility::Hot, &mut stager, &ReleaseId::new("v2"))
            .unwrap_err();

        assert!(matches!(err, WarmSwapError::NotWarm(Compatibility::Hot)));
        // No seam ran: admission is still open, the lease is still held, and the
        // trace is empty.
        assert!(
            gate.admits(),
            "admission must stay open for a refused release"
        );
        assert!(
            lease.is_held(),
            "the lease must stay held for a refused release"
        );
        assert!(
            trace.borrow().is_empty(),
            "no seam runs for a refused release"
        );
    }

    // --- The ordering invariant, happy path ------------------------------------

    #[test]
    fn warm_swap_walks_close_drain_release_swap_reacquire_resume_in_order() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        // Model log: one committed turn, then an INTERRUPTED turn (no complete).
        // The drain checkpoint must be the committed turn, never the orphan.
        let drain = RecordingDrain::new(
            &trace,
            &gate,
            vec![("t1", true), ("t2-interrupted", false)],
            2,
        );
        let lease = RecordingLease::new(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate.clone(), drain, lease.clone(), resume);

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            SwapActivator::new(&trace),
            || Health::Healthy,
            ReleaseId::new("v1"),
        );

        let report = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap();

        // The exact ordering invariant, observed across every seam.
        assert_eq!(
            *trace.borrow(),
            vec![
                "close".to_owned(),
                "drain".to_owned(),
                "lease.release".to_owned(),
                "swap:v2".to_owned(),
                "lease.reacquire".to_owned(),
                "resume@t1".to_owned(),
            ],
        );
        // The swap committed onto the new binary.
        assert_eq!(
            report.swap,
            Outcome::Committed {
                version: ReleaseId::new("v2"),
            }
        );
        // The checkpoint is the last COMMITTED turn, not the interrupted one.
        assert_eq!(report.checkpoint.last_committed_turn(), "t1");
        // Resume restarted from exactly the drained turn — no drop, no duplicate.
        assert_eq!(report.resume.resumed_from(), "t1");
        // The paused approvals survived the swap and were re-driven on resume.
        assert_eq!(report.checkpoint.paused_approvals(), 2);
        assert_eq!(report.resume.redriven_approvals(), 2);
        // The lease is held again and admission is open on the swapped-in binary.
        assert!(lease.is_held());
        assert!(gate.admits());
    }

    #[test]
    fn no_new_turn_is_admitted_once_the_drain_begins() {
        // The gate's `close` runs before `drain`, and the drain asserts admission
        // is already closed. This test makes the guarantee explicit: after the
        // swap kicks off, a turn arriving mid-drain is refused.
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        assert!(gate.admits(), "admission starts open");

        // Closing admission (step 2) flips it before any drain work.
        gate.close().unwrap();
        assert!(
            !gate.admits(),
            "a turn arriving after admission closes is refused"
        );
    }

    #[test]
    fn a_paused_approval_survives_the_warm_restart() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        // Three approvals paused at the checkpoint.
        let drain = RecordingDrain::new(&trace, &gate, vec![("t7", true)], 3);
        let lease = RecordingLease::new(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate, drain, lease, resume);

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            SwapActivator::new(&trace),
            || Health::Healthy,
            ReleaseId::new("v1"),
        );

        let report = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap();

        // The approvals paused before the swap are carried across it and re-driven
        // after resume — none is lost to the restart.
        assert_eq!(report.checkpoint.paused_approvals(), 3);
        assert_eq!(report.resume.redriven_approvals(), 3);
        assert_eq!(report.resume.resumed_from(), "t7");
    }

    // --- The swap rolled back: lease still re-acquired, conversation resumed ----

    #[test]
    fn a_rolled_back_swap_still_reacquires_the_lease_and_resumes() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let drain = RecordingDrain::new(&trace, &gate, vec![("t1", true)], 1);
        let lease = RecordingLease::new(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate.clone(), drain, lease.clone(), resume);

        // The new binary fails its health check → the stager auto-rolls back to v1.
        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            SwapActivator::new(&trace),
            || Health::Unhealthy("readiness probe timed out".to_owned()),
            ReleaseId::new("v1"),
        );

        let report = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap();

        // The swap rolled back to the previous binary...
        assert_eq!(
            report.swap,
            Outcome::RolledBack {
                stayed_on: ReleaseId::new("v1"),
                reason: "readiness probe timed out".to_owned(),
            }
        );
        // ...but the conversation was NOT left drained-but-dead: the lease is held
        // again and it resumed from the committed turn on the previous binary.
        assert!(
            lease.is_held(),
            "lease re-acquired even on a rolled-back swap"
        );
        assert!(gate.admits(), "admission re-opened on the previous binary");
        assert_eq!(report.resume.resumed_from(), "t1");
        // The pointer flipped forward to v2 then back to v1 — a rollback flip.
        assert_eq!(
            *trace.borrow(),
            vec![
                "close".to_owned(),
                "drain".to_owned(),
                "lease.release".to_owned(),
                "swap:v2".to_owned(),
                "swap:v1".to_owned(),
                "lease.reacquire".to_owned(),
                "resume@t1".to_owned(),
            ],
        );
    }

    // --- An unverified new binary is refused; the lease is put back -------------

    #[test]
    fn an_unverified_binary_is_refused_and_the_lease_is_reacquired() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let drain = RecordingDrain::new(&trace, &gate, vec![("t1", true)], 0);
        let lease = RecordingLease::new(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate, drain, lease.clone(), resume);

        // The new binary's signature does not verify — the stager applies nothing.
        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| false,
            SwapActivator::new(&trace),
            || panic!("health check must not run for an unverified binary"),
            ReleaseId::new("v1"),
        );

        let err = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap_err();

        assert!(matches!(err, WarmSwapError::Stage(StageError::Unverified)));
        // Nothing swapped, but the lease is back so the conversation is not stuck
        // behind a released lease: release then re-acquire framed the refused swap.
        assert!(
            lease.is_held(),
            "the lease is re-acquired after a refused swap"
        );
        assert_eq!(
            *trace.borrow(),
            vec![
                "close".to_owned(),
                "drain".to_owned(),
                "lease.release".to_owned(),
                "lease.reacquire".to_owned(),
            ],
        );
        // Resume never runs for a refused swap — the swap error is surfaced first.
        assert!(!trace.borrow().iter().any(|s| s.starts_with("resume")));
    }

    // --- Resume that drifts off the checkpoint is a hard correctness error ------

    #[test]
    fn resume_from_the_wrong_turn_is_a_mismatch_error() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let drain = RecordingDrain::new(&trace, &gate, vec![("t1", true)], 0);
        let lease = RecordingLease::new(&trace);
        // Resume drifts to a different turn than the checkpoint — a dropped or
        // duplicated committed turn.
        let resume = RecordingResume::drifting(&trace, &gate, &lease, "t0-stale");
        let swap = WarmSwap::new(gate, drain, lease, resume);

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            SwapActivator::new(&trace),
            || Health::Healthy,
            ReleaseId::new("v1"),
        );

        let err = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap_err();

        assert!(matches!(
            err,
            WarmSwapError::ResumeMismatch { resumed_from, expected }
                if resumed_from == "t0-stale" && expected == "t1"
        ));
    }

    // --- A failed drain leaves the lease held; nothing is released or swapped ---

    #[test]
    fn a_failed_drain_never_releases_the_lease_or_swaps() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let lease = RecordingLease::new(&trace);
        let failing_drain = {
            let trace = Rc::clone(&trace);
            move || -> Result<Checkpoint, WarmSwapError> {
                log(&trace, "drain");
                Err(WarmSwapError::Drain("journal sync failed".to_owned()))
            }
        };
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate.clone(), failing_drain, lease.clone(), resume);

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| panic!("verify must not run when the drain fails"),
            SwapActivator::new(&trace),
            || panic!("health check must not run when the drain fails"),
            ReleaseId::new("v1"),
        );

        let err = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap_err();

        assert!(matches!(err, WarmSwapError::Drain(_)));
        // The lease is still held — a failed drain must not release it — and
        // nothing swapped.
        assert!(lease.is_held(), "a failed drain keeps the lease held");
        assert_eq!(
            *trace.borrow(),
            vec!["close".to_owned(), "drain".to_owned()]
        );
    }

    // --- A failed re-acquire is reported as a lease re-acquire error ------------

    #[test]
    fn a_failed_reacquire_surfaces_as_a_lease_error() {
        let trace = trace();
        let gate = RecordingGate::new(&trace);
        let drain = RecordingDrain::new(&trace, &gate, vec![("t1", true)], 0);
        let lease = RecordingLease::failing_reacquire(&trace);
        let resume = RecordingResume::new(&trace, &gate, &lease);
        let swap = WarmSwap::new(gate, drain, lease, resume);

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            SwapActivator::new(&trace),
            || Health::Healthy,
            ReleaseId::new("v1"),
        );

        let err = swap
            .run(&Compatibility::Warm, &mut stager, &ReleaseId::new("v2"))
            .unwrap_err();

        assert!(matches!(
            err,
            WarmSwapError::Lease {
                op: "re-acquire",
                ..
            }
        ));
    }

    #[test]
    fn not_warm_error_reads_plainly() {
        let msg = WarmSwapError::NotWarm(Compatibility::Hot).to_string();
        assert_eq!(
            msg,
            "refused: only a warm binary swap takes the click-to-restart path \
             (a config change that reloads with no restart)",
        );
        for banned in ["sorry", "please", "unfortunately"] {
            assert!(!msg.to_lowercase().contains(banned));
        }
    }
}