polyc_runtime/

hot_reload.rs

//! Hot config-as-data reload — apply a verified bundle at the turn boundary,
//! with no process restart (see the update PRD and [`crate::compat`] /
//! [`crate::stager`]).
//!
//! This is the OTA-instant layer. A config-as-data bundle — system prompts,
//! persona definitions, the tool catalog, model routing — is staged, verified,
//! and then swapped into a single in-memory handle the turn loop reads *at
//! turn-start*. The next turn of any conversation picks up the new bundle; an
//! in-flight turn keeps the bundle it captured when it began. No binary changes,
//! no restart.
//!
//! # The turn-boundary handle
//!
//! [`HotConfig<C>`] is an atomically-swappable pointer to the live config-as-data
//! value. A turn reads it exactly once, at turn-start, via
//! [`HotConfig::current`] — the returned [`Arc`] is that turn's *pinned
//! snapshot*. Because the snapshot is an owned `Arc`, a swap that lands while the
//! turn is still running does not disturb it: the in-flight turn finishes on the
//! value it started with, and only the *next* [`HotConfig::current`] observes the
//! new value. That is the whole boundary guarantee, and it holds because the read
//! happens once, at the start.
//!
//! The control plane already runs exactly this pattern for one config dimension —
//! model routing is held behind an `Arc<ArcSwap<ModelRef>>` and read once per
//! turn at dispatch, so a live `model set` lands on the next turn of every
//! conversation. [`HotConfig`] generalizes that mechanism to the whole
//! config-as-data surface and ties it to the verify + classify + rollback
//! machinery below.
//!
//! # The classifier gate
//!
//! Only a bundle that classifies [`Compatibility::Hot`] takes this no-restart
//! path. [`ensure_hot`] refuses any other verdict, and [`apply_hot_reload`]
//! routes a [`StagedBundle`] through the compat interlock
//! ([`StagedBundle::evaluate`]) so a bundle authored against a different runtime
//! is refused before anything is applied. A warm or cold change is a binary
//! release, delivered on the other channel and picked up on restart — it never
//! arrives here as a config bundle.
//!
//! # Reusing the stager
//!
//! The apply itself reuses [`crate::stager`] wholesale: the same signature
//! verify (an unverified bundle is never applied), the same health check, and
//! the same pointer-flip rollback. The only thing that differs for a hot reload
//! is the [`Activator`] — instead of a symlink or image swap, it is a
//! config-pointer swap on a [`HotConfig`]. [`HotConfigActivator`] is that
//! activator, so a caller drives an ordinary [`Stager`] whose activator swaps the
//! live config value, and a failed health check flips the value back to the
//! previous bundle just as a binary rollback flips a symlink back.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, PoisonError};

use arc_swap::ArcSwap;
use thiserror::Error;

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

/// An atomically-swappable handle to the live config-as-data value the turn loop
/// reads at turn-start.
///
/// Clone to share the same underlying cell: a reader (the turn loop, per turn)
/// and a writer (a verified hot reload) hold clones and operate on one value.
/// [`HotConfig::current`] is the turn-start read; [`HotConfig::install`] is the
/// swap. A swap is visible only to reads that begin after it, so an in-flight
/// turn is never disturbed.
pub struct HotConfig<C> {
    cell: Arc<ArcSwap<C>>,
}

impl<C> Clone for HotConfig<C> {
    fn clone(&self) -> Self {
        // Share the same cell — never a deep copy of the config value.
        Self {
            cell: Arc::clone(&self.cell),
        }
    }
}

impl<C> HotConfig<C> {
    /// Seed the handle with an initial config value.
    #[must_use]
    pub fn new(initial: C) -> Self {
        Self {
            cell: Arc::new(ArcSwap::from_pointee(initial)),
        }
    }

    /// Seed the handle from an already-shared config value.
    #[must_use]
    pub fn from_arc(initial: Arc<C>) -> Self {
        Self {
            cell: Arc::new(ArcSwap::new(initial)),
        }
    }

    /// The live config-as-data value, read at turn-start.
    ///
    /// The returned [`Arc`] is the calling turn's pinned snapshot: a later
    /// [`HotConfig::install`] swaps the cell, but this owned handle keeps
    /// pointing at the value that was live when the turn began. Read it once, at
    /// the start of a turn, and thread that snapshot through the rest of the
    /// turn.
    #[must_use]
    pub fn current(&self) -> Arc<C> {
        self.cell.load_full()
    }

    /// Swap the live config-as-data value.
    ///
    /// The swap is atomic and lock-free. Turns that already read
    /// [`HotConfig::current`] keep their snapshot; the next turn to read sees
    /// `next`.
    pub fn install(&self, next: Arc<C>) {
        self.cell.store(next);
    }
}

/// A config-pointer-swap [`Activator`]: the hot-reload analog of the stager's
/// symlink or image swap.
///
/// It resolves a [`ReleaseId`] to a staged config value and swaps the live
/// [`HotConfig`] to it, so driving a [`Stager`] with this activator makes
/// `stage_and_apply` verify, apply, health-check, and roll back a *config*
/// reload with no changes to the stager itself. The previous bundle's value
/// stays registered, so an auto-rollback (a flip back to the previous release)
/// is a pointer swap, never a re-download.
///
/// Clone to hold a staging handle alongside the copy the stager owns: both
/// clones share the same live handle and the same registry.
pub struct HotConfigActivator<C> {
    live: HotConfig<C>,
    // Release identity → the config value that release delivers. The stager
    // flips forward to a newly-staged release and, on rollback, back to the
    // previously-committed one, so both must resolve here.
    staged: Arc<Mutex<HashMap<ReleaseId, Arc<C>>>>,
}

impl<C> Clone for HotConfigActivator<C> {
    fn clone(&self) -> Self {
        Self {
            live: self.live.clone(),
            staged: Arc::clone(&self.staged),
        }
    }
}

impl<C> HotConfigActivator<C> {
    /// Build an activator over `live`, registering its current value under
    /// `initial` so a rollback to the starting release resolves.
    ///
    /// `initial` is the [`ReleaseId`] the [`Stager`] is constructed with as its
    /// live version; registering the current value under it makes the first
    /// auto-rollback a pointer swap back to the value that is live right now.
    #[must_use]
    pub fn new(live: HotConfig<C>, initial: ReleaseId) -> Self {
        let mut staged = HashMap::new();
        staged.insert(initial, live.current());
        Self {
            live,
            staged: Arc::new(Mutex::new(staged)),
        }
    }

    /// Register the config value a release delivers, before it is applied.
    ///
    /// Call this for a release as it is staged (so `activate` can resolve the
    /// forward flip). A committed release stays registered, so a later rollback
    /// to it is a pointer swap.
    pub fn stage(&self, release: ReleaseId, value: Arc<C>) {
        self.lock().insert(release, value);
    }

    /// The live handle this activator swaps — clone it to read the config the
    /// same way a turn does.
    #[must_use]
    pub fn live(&self) -> HotConfig<C> {
        self.live.clone()
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<ReleaseId, Arc<C>>> {
        // A poisoned registry still holds valid entries — a panic elsewhere does
        // not corrupt the release → value map — so recover the guard rather than
        // propagate the poison and fail an otherwise-sound reload.
        self.staged.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

impl<C> Activator for HotConfigActivator<C> {
    fn activate(&self, release: &ReleaseId) -> Result<(), StageError> {
        let value = self.lock().get(release).map(Arc::clone);
        value.map_or_else(
            // A flip to an unregistered release cannot swap a value it never
            // received — refuse it as an activation failure rather than silently
            // leaving the live config unchanged.
            || {
                Err(StageError::Activate(format!(
                    "no staged config registered for release {release}"
                )))
            },
            |value| {
                self.live.install(value);
                Ok(())
            },
        )
    }
}

/// Why a hot config reload was refused or could not complete.
#[derive(Debug, Error)]
pub enum HotReloadError {
    /// The bundle did not classify [`Compatibility::Hot`] against the running
    /// runtime, so it does not take the no-restart config-reload path. A warm or
    /// cold change is a binary release picked up on restart; an incompatible
    /// bundle was authored against a runtime this build cannot satisfy.
    #[error("refused: only a hot config bundle reloads without a restart ({})", verdict_label(.0))]
    NotHot(Compatibility),
    /// The stager could not stage, verify, or apply the bundle. Wraps the
    /// underlying [`StageError`] — most importantly [`StageError::Unverified`],
    /// which means the signature did not verify and nothing was applied.
    #[error(transparent)]
    Stage(#[from] StageError),
}

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

/// Gate a classification verdict onto the hot path: [`Ok`] only for
/// [`Compatibility::Hot`].
///
/// This is the classifier interlock in one place — the reload proceeds only when
/// the change is config-as-data on a matching runtime. Any other verdict
/// (`Warm`, `Cold`, or `Incompatible`) is refused.
///
/// # Errors
///
/// Returns [`HotReloadError::NotHot`] carrying the refused verdict for anything
/// other than [`Compatibility::Hot`].
pub fn ensure_hot(verdict: &Compatibility) -> Result<(), HotReloadError> {
    if *verdict == Compatibility::Hot {
        Ok(())
    } else {
        Err(HotReloadError::NotHot(verdict.clone()))
    }
}

/// Apply a staged config-as-data `bundle` to the live config via `stager`, only
/// if it classifies hot against `running`.
///
/// The sequence is: classify (the [`StagedBundle::evaluate`] interlock from
/// [`crate::compat`]) → refuse anything but [`Compatibility::Hot`] → hand the
/// release to the [`Stager`], which verifies the signature, applies it (a
/// config-pointer swap when the stager's activator is a [`HotConfigActivator`]),
/// health-checks, and rolls back on failure. The classifier gate runs *before*
/// the stager touches anything, so an incompatible bundle never reaches verify
/// or apply.
///
/// On [`Outcome::Committed`] the live [`HotConfig`] now serves the new bundle to
/// the next turn; on [`Outcome::RolledBack`] the health check failed and the
/// pointer flipped back, so the live config is honestly the previous bundle.
///
/// # Errors
///
/// Returns [`HotReloadError::NotHot`] when the bundle does not classify hot
/// (nothing is staged, verified, or applied), or [`HotReloadError::Stage`] when
/// the stager cannot stage, verify (an unverified bundle is refused), or apply
/// the release.
pub fn apply_hot_reload<S, V, A, H>(
    running: &Fingerprint,
    bundle: &StagedBundle,
    stager: &mut Stager<S, V, A, H>,
    release: &ReleaseId,
) -> Result<Outcome, HotReloadError>
where
    S: UpdateSource,
    V: SignatureVerifier,
    A: Activator,
    H: HealthCheck,
{
    ensure_hot(&bundle.evaluate(running))?;
    Ok(stager.stage_and_apply(release)?)
}

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

    use std::path::PathBuf;

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

    /// A stand-in config-as-data value: a tag identifies which bundle a turn read.
    #[derive(Debug, Clone, PartialEq, Eq)]
    struct Cfg {
        tag: &'static str,
    }

    impl Cfg {
        fn new(tag: &'static str) -> Arc<Self> {
            Arc::new(Self { tag })
        }
    }

    /// The running build's fingerprint used across these tests.
    fn running() -> Fingerprint {
        Fingerprint::new(3, 7, "polychrome.uno/v1", "catalog-v1")
    }

    /// A bundle authored against the running runtime — classifies hot.
    fn hot_bundle() -> StagedBundle {
        StagedBundle::new(running().runtime_target(), "catalog-v2")
    }

    fn artifact_for(release: &ReleaseId) -> StagedArtifact {
        StagedArtifact {
            release: release.clone(),
            staged_path: PathBuf::from(format!("/var/lib/polychrome/staged/{release}")),
            bundle: hot_bundle(),
            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 turn-boundary invariant, on the handle in isolation ----------------

    #[test]
    fn a_swap_lands_on_the_next_turn_never_an_in_flight_one() {
        let live = HotConfig::new(Cfg { tag: "v1" });

        // Turn A begins: it snapshots the config-as-data at turn-start.
        let turn_a = live.current();

        // A verified hot bundle is applied between turns.
        live.install(Cfg::new("v2"));

        // Turn B begins AFTER the swap and snapshots at its own turn-start.
        let turn_b = live.current();

        // Turn A, still in flight, finishes on the config it started with...
        assert_eq!(
            turn_a.tag, "v1",
            "in-flight turn keeps its turn-start snapshot"
        );
        // ...while the next turn reads the newly installed config.
        assert_eq!(turn_b.tag, "v2", "the next turn reads the new bundle");
        // The boundary is exactly turn-start: A never observes v2 mid-turn.
        assert_eq!(live.current().tag, "v2", "the live handle now serves v2");
    }

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

    #[test]
    fn ensure_hot_admits_only_hot() {
        assert!(ensure_hot(&Compatibility::Hot).is_ok());
        for verdict in [
            Compatibility::Warm,
            Compatibility::Cold,
            Compatibility::Incompatible(crate::compat::Incompatibility::Wire),
        ] {
            let err = ensure_hot(&verdict).unwrap_err();
            assert!(
                matches!(err, HotReloadError::NotHot(v) if v == verdict),
                "non-hot verdict must be refused: {verdict:?}",
            );
        }
    }

    #[test]
    fn apply_refuses_a_bundle_built_for_another_runtime_before_touching_anything() {
        let live = HotConfig::new(Cfg { tag: "v1" });
        let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
        // A bundle authored against a newer wire version → Incompatible, not hot.
        let bundle = StagedBundle::new(RuntimeTarget::new(4, 7, "polychrome.uno/v1"), "catalog-v2");

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

        let err =
            apply_hot_reload(&running(), &bundle, &mut stager, &ReleaseId::new("v2")).unwrap_err();

        assert!(matches!(err, HotReloadError::NotHot(_)));
        // Nothing was applied: the live config is untouched.
        assert_eq!(live.current().tag, "v1");
    }

    // --- Reuse of the stager's verify -------------------------------------------

    #[test]
    fn apply_never_swaps_an_unverified_bundle() {
        let live = HotConfig::new(Cfg { tag: "v1" });
        let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
        activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));

        let mut stager = Stager::new(
            ok_source,
            // Signature does not verify.
            |_: &StagedArtifact| false,
            activator,
            || panic!("health check must not run for an unverified bundle"),
            ReleaseId::new("v1"),
        );

        let err = apply_hot_reload(
            &running(),
            &hot_bundle(),
            &mut stager,
            &ReleaseId::new("v2"),
        )
        .unwrap_err();

        assert!(matches!(err, HotReloadError::Stage(StageError::Unverified)));
        // The verify refusal means the pointer never flipped.
        assert_eq!(live.current().tag, "v1");
    }

    // --- The end-to-end hot reload: gate + verify + swap at the boundary --------

    #[test]
    fn a_verified_hot_reload_swaps_the_config_for_the_next_turn() {
        let live = HotConfig::new(Cfg { tag: "v1" });
        let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
        // Register the value the new release delivers, as it is staged.
        activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));

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

        // A turn already in flight captured its snapshot before the reload.
        let in_flight = live.current();

        let outcome = apply_hot_reload(
            &running(),
            &hot_bundle(),
            &mut stager,
            &ReleaseId::new("v2"),
        )
        .unwrap();

        assert_eq!(
            outcome,
            Outcome::Committed {
                version: ReleaseId::new("v2"),
            }
        );
        // The in-flight turn finishes on the prior config...
        assert_eq!(
            in_flight.tag, "v1",
            "in-flight turn completes on the prior bundle"
        );
        // ...and the next turn reads the newly installed one.
        assert_eq!(
            live.current().tag,
            "v2",
            "the next turn reads the reloaded bundle"
        );
    }

    #[test]
    fn a_failed_health_check_flips_the_config_back_to_the_previous_bundle() {
        let live = HotConfig::new(Cfg { tag: "v1" });
        let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
        activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));

        let mut stager = Stager::new(
            ok_source,
            |_: &StagedArtifact| true,
            activator,
            || Health::Unhealthy("readiness probe timed out".to_owned()),
            ReleaseId::new("v1"),
        );

        let outcome = apply_hot_reload(
            &running(),
            &hot_bundle(),
            &mut stager,
            &ReleaseId::new("v2"),
        )
        .unwrap();

        assert_eq!(
            outcome,
            Outcome::RolledBack {
                stayed_on: ReleaseId::new("v1"),
                reason: "readiness probe timed out".to_owned(),
            }
        );
        // The config was swapped to v2, then flipped back — a pointer swap, not a
        // re-download — so the next turn honestly reads the previous bundle.
        assert_eq!(
            live.current().tag,
            "v1",
            "rollback restores the previous bundle"
        );
    }

    #[test]
    fn not_hot_error_reads_plainly() {
        let msg = HotReloadError::NotHot(Compatibility::Warm).to_string();
        assert_eq!(
            msg,
            "refused: only a hot config bundle reloads without a restart \
             (a binary change that needs a restart)",
        );
        for banned in ["sorry", "please", "unfortunately"] {
            assert!(!msg.to_lowercase().contains(banned));
        }
    }
}