agentplane 0.45.0

//! Record schema evolution.
//!
//! The journal is forever, so record shapes must evolve without rewriting
//! history. Five rules, and the last two are the ones that are easy to get
//! wrong:
//!
//! 1. Records carry `(kind, v)`; every writer stamps the version.
//! 2. **Backward compatibility is permanent.** New code must read every shape
//!    ever written. There is no "we migrated past that".
//! 3. **Upcast on read; never rewrite history.**
//! 4. **Upcasters are pure and total.** One that reads a clock, a config file,
//!    or the network breaks replay — the same bug class as a non-deterministic
//!    effect, but harder to find because it only manifests on old records.
//! 5. **Hash the wire bytes, not the upcast form.** Rehashing after an upcast
//!    would destroy tamper evidence for all history the first time a schema
//!    changed. See [`Record::from_stored_with`](super::Record::from_stored_with), which
//!    verifies against the stored bytes and never re-serializes.

use serde_json::Value;

use crate::core::StoreError;

/// A pure, total transform from an older record version to the current one.
///
/// Lives inside the deterministic zone: given the same `(kind, version,
/// payload)` it must always produce the same output, in this process and in one
/// started a year from now.
pub trait Upcaster: Send + Sync + std::fmt::Debug {
    /// Highest version this upcaster knows how to produce.
    fn current_version(&self, kind: &str) -> u16;

    /// Lift one payload one or more versions forward.
    fn upcast(&self, kind: &str, version: u16, payload: Value) -> Result<Value, StoreError>;
}

/// The identity upcaster: every kind is at v1 and nothing needs lifting.
///
/// Kept as a real type rather than an `Option<Box<dyn Upcaster>>` so the call
/// site has no branch and the first genuine migration is a one-line swap.
#[derive(Debug, Clone, Copy, Default)]
pub struct Identity;

impl Upcaster for Identity {
    fn current_version(&self, _kind: &str) -> u16 {
        1
    }

    fn upcast(&self, kind: &str, version: u16, payload: Value) -> Result<Value, StoreError> {
        match version {
            1 => Ok(payload),
            // Refused rather than guessed at, in both directions.
            //
            // Below: pre-freeze record shapes change by hard cut, and the danger
            // is that an older record still *parses* — a field that moved or was
            // added comes back as its serde default, so the journal answers an
            // audit question falsely instead of failing to answer it. There is
            // also nothing to lift *to*: the missing field's value was never
            // written, and inventing one would fabricate provenance. The whole
            // journal is refused rather than the affected kinds, because a
            // journal holding one such record holds the rest.
            //
            // Above: a version this build has never heard of means somebody
            // deployed a writer ahead of its readers.
            _ => Err(StoreError::UnknownRecordVersion {
                kind: kind.to_owned(),
                version,
                reads: 1,
            }),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    /// A journal from an older cut is refused, not read with fields missing.
    ///
    /// The danger is that it *parses*. A record whose shape has since changed
    /// still deserialises, with every moved or added field taking its serde
    /// default — so a resumed run reports, say, that no policy governed it. That
    /// is a false answer to an audit question, which is worse than a refusal to
    /// answer, and it is the reason nothing is lifted rather than refused.
    ///
    /// And it is refused as a **version** rather than as damage: the same
    /// distinction the sealed envelope draws, for the same reason. A rolling
    /// deploy reported as tampering spends the one alarm that has to stay
    /// believable.
    #[test]
    fn a_journal_from_an_older_cut_is_refused_rather_than_misread() {
        let err = Identity
            .upcast("RunAdmitted", 0, json!({ "agent": "a", "input": null }))
            .unwrap_err();
        assert!(
            matches!(
                err,
                StoreError::UnknownRecordVersion {
                    version: 0,
                    reads: 1,
                    ..
                }
            ),
            "a version this build does not read is not damage: {err:?}"
        );
        let text = err.to_string();
        assert!(
            text.contains("deploy readers before writers"),
            "the refusal must say what to do about it: {text}"
        );
    }

    /// A future-dated record means someone deployed a writer before the readers.
    /// Failing loudly beats silently dropping fields.
    #[test]
    fn future_versions_are_refused_not_guessed() {
        let err = Identity.upcast("EffectDone", 7, json!({})).unwrap_err();
        assert!(matches!(
            err,
            StoreError::UnknownRecordVersion { version: 7, .. }
        ));
    }

    #[test]
    fn current_version_passes_through() {
        // Asked for rather than restated: a hardcoded number here silently
        // becomes a test of an *old* version the moment the format is bumped,
        // and then passes by being refused for the wrong reason.
        let v = Identity.current_version("EffectDone");
        assert_eq!(
            Identity.upcast("EffectDone", v, json!({"a": 1})).unwrap(),
            json!({"a": 1})
        );
    }

    /// Upcasting must be deterministic — the property that keeps replay sound
    /// once real migrations exist.
    #[test]
    fn upcasting_is_pure() {
        let v = Identity.current_version("StepStarted");
        let a = Identity
            .upcast("StepStarted", v, json!({"skill": "x"}))
            .unwrap();
        let b = Identity
            .upcast("StepStarted", v, json!({"skill": "x"}))
            .unwrap();
        assert_eq!(a, b);
    }
}