agentplane/core/

id.rs

//! Identifiers, digests, and the hash-chain primitive.
//!
//! ULIDs are used for run and case ids because they are lexicographically
//! sortable: a journal scan for one run is a range scan, and cases sort by
//! creation time for free.

use std::fmt;

use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
use sha2::{Digest as _, Sha256};

/// Wall-clock instant. Only ever obtained through a journaled effect
/// (`StepCtx::now`), never by reading the ambient clock.
///
/// # Format it before it reaches a wire
///
/// This is a re-export of [`time::OffsetDateTime`], and that type's **derived**
/// `Serialize` is not a date. Unless the `time` crate's `serde-human-readable`
/// feature is on — it is not here, and enabling it would silently change every
/// stored shape — a bare `Timestamp` field serialises to its **component
/// array**:
///
/// ```text
/// [2027, 15, 8, 0, 0, 0, 0, 0, 0]
///  year  ordinal-day  h  m  s  ns  offset-h  offset-m  offset-s
/// ```
///
/// It parses, it round-trips, and every consumer that expected a date gets nine
/// numbers. A model asked to do arithmetic on one produces confident nonsense; a
/// dashboard renders `2027` as the hour. The failure is quiet in exactly the way
/// a format error is not.
///
/// So **every timestamp on a wire in this crate is RFC 3339**, and the two ways
/// to spell that are:
///
/// ```
/// # use agentplane::core::Timestamp;
/// #[derive(serde::Serialize, serde::Deserialize)]
/// struct Deadline {
///     #[serde(with = "time::serde::rfc3339")]
///     resolved_at: Timestamp,
///     #[serde(default, with = "time::serde::rfc3339::option")]
///     warn_at: Option<Timestamp>,
/// }
/// ```
///
/// and [`format_timestamp`] where the value is being placed into a
/// `serde_json::json!` literal rather than a struct field — an effect
/// descriptor, a `CloudEvents` envelope, a log line.
///
/// `tests/guards` walks the crate's serialized types and fails on a component
/// array, so this is a rule the build enforces rather than one a reviewer has to
/// remember. The hazard is worth stating here anyway, because `Timestamp` is
/// public API: it lands in **your** tool payloads too, and the crate cannot
/// check those.
pub type Timestamp = time::OffsetDateTime;

/// One instant, RFC 3339, for a place that is not a struct field.
///
/// `#[serde(with = "time::serde::rfc3339")]` is the answer wherever there is a
/// field to attach it to. This is the answer where there is not — inside a
/// `json!` literal, where a bare `Timestamp` would take the derived component
/// array with nothing in the source to hint at it. See [`Timestamp`].
///
/// Falls back to the numeric Unix second if formatting fails, which needs an
/// instant outside RFC 3339's representable range. That is not a case worth
/// failing an effect key over, and a number is at least unambiguous.
#[must_use]
pub fn format_timestamp(at: Timestamp) -> String {
    at.format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_else(|_| at.unix_timestamp().to_string())
}

/// Per-run monotonic journal position, starting at 1.
pub type Seq = u64;

/// Ownership fencing epoch.
///
/// Every journal append carries the writer's epoch and the store rejects stale
/// epochs *in the same transaction that writes*, so a paused or partitioned
/// instance that wakes up and keeps writing is fenced by the store rather than
/// by timing. Split-brain cannot corrupt the chain by construction.
pub type Epoch = u64;

macro_rules! ulid_newtype {
    ($(#[$m:meta])* $name:ident, $prefix:literal) => {
        $(#[$m])*
        ///
        /// # One spelling, everywhere
        ///
        #[doc = concat!(" Prefixed (`", $prefix, "_01J8Z…`) in every direction:")]
        /// [`Display`](std::fmt::Display), `Serialize`, the store key, the
        /// column. An id is self-describing wherever it lands — a log line, a
        /// URL, a JSON field, a SIEM — and grepping one against another finds
        /// it whichever way round you do it.
        ///
        /// What this rules out is a store spelling one id two ways — prefixed in
        /// its keys and bare in its payloads — which costs a consumer an
        /// afternoon and looks like two ids until it does.
        ///
        /// [`Deserialize`] and [`parse`](Self::parse) accept a bare ULID too,
        /// because one arrives from somewhere else often enough. The prefix is
        /// checked rather than stripped blindly, so a `case_…` string is refused
        /// where a [`RunId`] is wanted instead of silently becoming one.
        #[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
        pub struct $name(pub ulid::Ulid);

        impl $name {
            /// Mint a fresh id from the ambient clock and a random tail.
            ///
            /// Reserved for the runtime's admission path, which journals the
            /// result. Skills must never call this — see `clippy.toml`.
            ///
            /// **Sorted to the millisecond, not beyond it.** Ordering comes
            /// from the timestamp prefix, so two ids minted in the same
            /// millisecond order by their random tails — arbitrarily, and
            /// differently on each instance. A range scan over one run's
            /// records is exact because the run id is a *prefix*; "cases sort
            /// by creation time" is true between milliseconds and a coin flip
            /// inside one.
            #[allow(clippy::disallowed_methods)]
            #[must_use]
            pub fn generate() -> Self {
                Self(ulid::Ulid::generate())
            }

            /// Reconstruct from a stored string.
            ///
            /// Accepts both the prefixed form produced by [`std::fmt::Display`]
            /// and a bare ULID, so ids round-trip through logs, URLs, and
            /// database columns without the caller having to know which form it
            /// is holding.
            ///
            /// # Errors
            ///
            /// If what remains after the prefix is not a ULID.
            pub fn parse(s: &str) -> Result<Self, ulid::DecodeError> {
                let bare = s.strip_prefix(concat!($prefix, "_")).unwrap_or(s);
                ulid::Ulid::from_string(bare).map(Self)
            }
        }

        /// The trait an id arrives through: an `axum` path segment, a `clap`
        /// argument, a TOML key, a `serde` field with `#[serde(with = …)]`.
        ///
        /// `.parse()` is where every consumer reaches first, and its absence
        /// made [`parse`](Self::parse) something each of them had to find.
        impl std::str::FromStr for $name {
            type Err = ulid::DecodeError;
            fn from_str(s: &str) -> Result<Self, Self::Err> {
                Self::parse(s)
            }
        }

        /// The written form, so a record says which kind of id it holds.
        impl Serialize for $name {
            fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
                s.collect_str(self)
            }
        }

        /// Through [`parse`](Self::parse), so a bare ULID from somewhere else
        /// still reads.
        ///
        /// A refusal names the type and the input: `ulid`'s own error is
        /// `invalid length`, which says nothing about which field was wrong or
        /// what it wanted.
        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
                let raw = <std::borrow::Cow<'de, str>>::deserialize(d)?;
                Self::parse(&raw).map_err(|e| {
                    D::Error::custom(format!(
                        concat!("not a ", stringify!($name), " ('{}'): {}"),
                        raw, e
                    ))
                })
            }
        }

        impl fmt::Display for $name {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                write!(f, "{}_{}", $prefix, self.0)
            }
        }

        impl fmt::Debug for $name {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                write!(f, "{self}")
            }
        }
    };
}

ulid_newtype!(
    /// A single execution: one goal, one plan, one lifetime.
    ///
    /// Runs stay short *by design* — longevity lives in the [`CaseId`], so a
    /// six-week business process never pins a code version.
    RunId, "run"
);

ulid_newtype!(
    /// A long-lived, correlated business fact spanning many runs and weeks.
    CaseId, "case"
);

ulid_newtype!(
    /// One business act made of many independent ones.
    ///
    /// Owns N runs sharing one frozen plan — see [`crate::batch`]. Distinct from
    /// a [`CaseId`] because the relationship is different in kind: a case is a
    /// matter that several runs *touch* over weeks, while a batch is a single
    /// act that several runs *constitute* in one pass.
    BatchId, "batch"
);

/// Position of a step within a plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(transparent)]
pub struct StepId(pub u32);

impl fmt::Display for StepId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "s{}", self.0)
    }
}

/// SHA-256 over an artifact's canonical serialization.
///
/// Also the hash-chain link type. [`Digest::chain`] is the only way to extend a
/// chain, and it hashes `prev ‖ bytes` — where `bytes` are the record's *wire*
/// bytes, never a re-serialized or upcast form. Rehashing after an upcast would
/// destroy tamper evidence for all history the moment a schema changed.
#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default)]
pub struct Digest([u8; 32]);

impl Digest {
    /// The chain's genesis link.
    pub const ZERO: Self = Self([0u8; 32]);

    /// Hash a byte string.
    #[must_use]
    pub fn of(bytes: &[u8]) -> Self {
        let mut h = Sha256::new();
        h.update(bytes);
        Self(h.finalize().into())
    }

    /// Extend a hash chain: `H(prev ‖ bytes)`.
    #[must_use]
    pub fn chain(prev: Self, bytes: &[u8]) -> Self {
        let mut h = Sha256::new();
        h.update(prev.0);
        h.update(bytes);
        Self(h.finalize().into())
    }

    #[must_use]
    pub const fn from_bytes(b: [u8; 32]) -> Self {
        Self(b)
    }

    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }

    #[must_use]
    pub fn to_hex(self) -> String {
        hex::encode(self.0)
    }

    pub fn from_hex(s: &str) -> Result<Self, hex::FromHexError> {
        let mut out = [0u8; 32];
        hex::decode_to_slice(s, &mut out)?;
        Ok(Self(out))
    }

    /// First 8 hex chars — enough to identify a record in a log line.
    #[must_use]
    pub fn short(self) -> String {
        hex::encode(&self.0[..4])
    }
}

impl fmt::Display for Digest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.to_hex())
    }
}

impl fmt::Debug for Digest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}…", self.short())
    }
}

impl Serialize for Digest {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(&self.to_hex())
    }
}

impl<'de> Deserialize<'de> for Digest {
    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let s = String::deserialize(d)?;
        Self::from_hex(&s).map_err(D::Error::custom)
    }
}

/// Stable identity of one effect within one run.
///
/// `H(step ‖ ordinal ‖ kind ‖ canonical(args))`. Two properties follow:
///
/// * **Exactly-once** — the store's unique index on `(run_id, effect_key)` makes
///   "an effect is started at most once per run" a database invariant.
/// * **Divergence detection** — on replay the key is recomputed from the
///   deterministic zone. A mismatch means the code took a different path than
///   the recorded one, and the run is quarantined rather than allowed to
///   silently diverge.
///
/// Skills never construct these: the runtime derives the key from the effect's
/// descriptor plus its position, so a skill cannot forge or collide one.
#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(transparent)]
pub struct EffectKey(Digest);

/// Which pass of a step an effect belongs to.
///
/// A step can run twice for entirely legitimate reasons: once going forward,
/// and once again in reverse when a later step fails and the saga unwinds. The
/// two are different work with different effects, and the journal has to be
/// able to tell them apart.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Phase {
    /// Doing the work.
    #[default]
    Forward,
    /// Undoing it.
    Compensating,
}

impl Phase {
    #[must_use]
    pub const fn is_forward(self) -> bool {
        matches!(self, Self::Forward)
    }

    /// The spelling every store writes.
    ///
    /// One, because three stores keep a phase column and a phase that reads
    /// back as the other half of the saga hands the unwind logic a compensating
    /// record wearing the forward pass's name.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Forward => "forward",
            Self::Compensating => "compensating",
        }
    }

    /// The inverse of [`as_str`](Self::as_str), written over
    /// [`ALL`](Self::ALL) for the reason [`CaseStatus::parse`] is.
    ///
    /// [`CaseStatus::parse`]: crate::core::CaseStatus::parse
    #[must_use]
    pub fn parse(s: &str) -> Option<Self> {
        Self::ALL.iter().copied().find(|c| c.as_str() == s)
    }

    /// Both passes.
    pub const ALL: [Self; 2] = [Self::Forward, Self::Compensating];

    /// By-reference form, for `skip_serializing_if`.
    #[allow(clippy::trivially_copy_pass_by_ref)]
    #[must_use]
    pub const fn is_forward_ref(v: &Self) -> bool {
        v.is_forward()
    }
}

impl EffectKey {
    /// `phase` separates the forward pass from the compensating one.
    ///
    /// Without it a step's compensation would restart its ordinal at zero and
    /// collide with the step's own forward effects: replay would read the
    /// forward result back as the compensation's, and the store's uniqueness
    /// constraint would reject the second announcement.
    ///
    /// `attempt` is 1-based and part of the identity, so a retry is a *new*
    /// effect in the journal rather than a second record under an existing key.
    ///
    /// Without it, attempt 2 would collide with attempt 1's recorded failure:
    /// replay would read back the failure instead of the retry that followed,
    /// and the store's uniqueness constraint on `EffectStarted` would reject
    /// the second attempt outright.
    pub(crate) fn derive(
        step: StepId,
        phase: Phase,
        ordinal: u32,
        attempt: u32,
        kind: &str,
        canonical_args: &[u8],
    ) -> Self {
        let mut h = Sha256::new();
        h.update(step.0.to_be_bytes());
        h.update([phase as u8]);
        h.update(ordinal.to_be_bytes());
        h.update(attempt.to_be_bytes());
        h.update((kind.len() as u64).to_be_bytes());
        h.update(kind.as_bytes());
        h.update(canonical_args);
        Self(Digest(h.finalize().into()))
    }

    #[must_use]
    pub fn to_hex(self) -> String {
        self.0.to_hex()
    }

    pub fn from_hex(s: &str) -> Result<Self, hex::FromHexError> {
        Digest::from_hex(s).map(Self)
    }
}

impl fmt::Display for EffectKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "ek:{}", self.0.to_hex())
    }
}

impl fmt::Debug for EffectKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "ek:{}…", self.0.short())
    }
}

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

    #[test]
    fn chain_is_order_sensitive() {
        let a = Digest::chain(Digest::ZERO, b"a");
        let b = Digest::chain(a, b"b");
        let swapped = Digest::chain(Digest::chain(Digest::ZERO, b"b"), b"a");
        assert_ne!(b, swapped, "chain must not be commutative");
    }

    #[test]
    fn chain_detects_any_mutation() {
        let genuine = Digest::chain(Digest::ZERO, b"record-1");
        let tampered = Digest::chain(Digest::ZERO, b"record-2");
        assert_ne!(genuine, tampered);
    }

    #[test]
    fn digest_hex_roundtrips() {
        let d = Digest::of(b"hello");
        assert_eq!(Digest::from_hex(&d.to_hex()).unwrap(), d);
    }

    #[test]
    fn effect_key_separates_step_phase_ordinal_and_attempt() {
        let fwd = Phase::Forward;
        let base = EffectKey::derive(StepId(0), fwd, 0, 1, "tool", b"{}");
        let other_ordinal = EffectKey::derive(StepId(0), fwd, 1, 1, "tool", b"{}");
        let other_step = EffectKey::derive(StepId(1), fwd, 0, 1, "tool", b"{}");
        let other_attempt = EffectKey::derive(StepId(0), fwd, 0, 2, "tool", b"{}");
        let compensating = EffectKey::derive(StepId(0), Phase::Compensating, 0, 1, "tool", b"{}");
        assert_ne!(base, other_ordinal, "ordinal must be part of the key");
        assert_ne!(base, other_step, "step must be part of the key");
        assert_ne!(
            base, other_attempt,
            "attempt must be part of the key, or a retry collides with the \
             failure it is retrying"
        );
        assert_ne!(
            base, compensating,
            "phase must be part of the key, or a step's compensation collides \
             with its own forward pass"
        );
    }

    /// Length-prefixing `kind` prevents a boundary-shifting collision: without
    /// it, `("ab", "c")` and `("a", "bc")` would hash identically.
    #[test]
    fn effect_key_kind_is_length_prefixed() {
        let a = EffectKey::derive(StepId(0), Phase::Forward, 0, 1, "ab", b"c");
        let b = EffectKey::derive(StepId(0), Phase::Forward, 0, 1, "a", b"bc");
        assert_ne!(a, b);
    }

    #[test]
    fn ids_display_with_prefix() {
        let r = RunId::generate();
        assert!(r.to_string().starts_with("run_"));
    }

    /// The property that broke first time round: `Display` and `parse` must be
    /// inverses, or every id that goes through a database column comes back
    /// unreadable.
    #[test]
    fn ids_round_trip_through_their_displayed_form() {
        let r = RunId::generate();
        assert_eq!(RunId::parse(&r.to_string()).unwrap(), r);
        let c = CaseId::generate();
        assert_eq!(CaseId::parse(&c.to_string()).unwrap(), c);
    }

    /// A bare ULID is still accepted, so ids written by other tools parse.
    #[test]
    fn bare_ulids_still_parse() {
        let r = RunId::generate();
        assert_eq!(RunId::parse(&r.0.to_string()).unwrap(), r);
    }

    /// Prefixes are not interchangeable in *meaning*, but parsing is lenient by
    /// design: a `CaseId` column holds case ids, and the prefix is a display
    /// affordance rather than a type check.
    #[test]
    fn parsing_rejects_garbage() {
        assert!(RunId::parse("run_not-a-ulid").is_err());
    }

    /// **One spelling, in every direction.**
    ///
    /// `Display`, `Serialize`, the store key and the column all write the
    /// prefixed form, so the same id greps against itself wherever it is read.
    /// They did not always: keys carried the prefix and record payloads did
    /// not, so a consumer who stored the string an operator sees in a log could
    /// not deserialize it — `invalid length`, from a check inside `ulid`, naming
    /// neither the field nor the fact that there were two forms.
    #[test]
    fn one_spelling_is_written_and_both_are_read() {
        let r = RunId::generate();

        assert_eq!(
            serde_json::to_string(&r).expect("serialises"),
            format!("\"{r}\""),
            "the written form is the one Display shows"
        );

        for text in [r.to_string(), r.0.to_string()] {
            let json = serde_json::to_string(&text).expect("a string");
            assert_eq!(
                serde_json::from_str::<RunId>(&json).expect("both forms read"),
                r,
                "the {text} spelling did not deserialize"
            );
        }
    }

    /// Leniency stops at the prefix: a case id is not a run id.
    ///
    /// The half that makes accepting two spellings safe. `parse` strips only
    /// *this* type's prefix, so `case_…` fails the ULID decode rather than
    /// quietly becoming a `RunId` — which a blind `strip_prefix` up to the
    /// underscore would have allowed.
    #[test]
    fn another_types_prefix_is_refused_rather_than_stripped() {
        let case = CaseId::generate();
        assert!(
            serde_json::from_str::<RunId>(&format!("\"{case}\"")).is_err(),
            "a case id deserialized as a run id"
        );
        assert!(RunId::parse(&case.to_string()).is_err());
        // And the positive half, so the negative one is not vacuous.
        assert!(CaseId::parse(&case.to_string()).is_ok());
    }

    /// `.parse()` is where an id arrives — a path segment, a CLI argument.
    #[test]
    fn ids_arrive_through_from_str() {
        let r = RunId::generate();
        assert_eq!(r.to_string().parse::<RunId>().unwrap(), r);
        assert_eq!(r.0.to_string().parse::<RunId>().unwrap(), r);
        assert!("nope".parse::<RunId>().is_err());
    }

    /// **The digest is SHA-256, pinned to values computed outside this crate.**
    ///
    /// Every effect key, chain link, Merkle leaf and blob address in the system
    /// comes out of these two functions, so what they emit is a durable format
    /// even though nothing declares it one: change a byte and every historical
    /// run becomes unverifiable, silently, because the whole suite would agree
    /// with itself under the new value.
    ///
    /// That is what makes a *self-consistent* test worthless here and why these
    /// literals were produced by `shasum -a 256` and Python's `hashlib` rather
    /// than by running this code and writing down the answer. They survived a
    /// `sha2` 0.10 → 0.11 upgrade, which is the class of change they exist for:
    /// the hasher was swapped underneath and the bytes had to be shown, not
    /// assumed, to be the same ones.
    ///
    /// What this does not pin: the canonicalization that decides *which* bytes
    /// reach the hasher. That is `canon`'s own contract.
    #[test]
    fn the_digest_matches_sha256_computed_elsewhere() {
        assert_eq!(
            Digest::of(b"").to_hex(),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "the empty digest moved, so every digest in every journal moved with it"
        );
        assert_eq!(
            Digest::of(b"agentplane").to_hex(),
            "c0f8f77669f4860960387db0dc9984894587bcfcc75d1846ffd3574563833443"
        );
        // `chain` is `H(prev ‖ bytes)`, and the concatenation order is the half
        // a reimplementation gets wrong — reversed, it still hashes, still
        // verifies against itself, and agrees with no other reader.
        assert_eq!(
            Digest::chain(Digest::of(b""), b"agentplane").to_hex(),
            "171c5eddf30189b0efde9c71eb14f77685fab77d7750152d032f9e7cf4181159"
        );
    }
}