polyc-facts 2026.10.2

//! The `administrator-audit/v1` fold: one row per administrator model change.
//!
//! # One partition, one kind, one row shape
//!
//! The source is the `admin-audit` journal partition, which one writer owns
//! (`polyc_control_plane::forensics`) and which carries one kind,
//! [`kinds::ADMIN_MODEL_CHANGE`]. Every other family in this crate folds a
//! conversation partition, where an unrecognized kind belongs to a sibling
//! family and is skipped. Here it does not: a kind nobody declared means the
//! writer and this fold disagree about what the partition holds, and
//! publishing under that disagreement would state a history neither of them
//! holds. So this is the crate's one closed-vocabulary fold, and an unknown
//! kind refuses the generation.
//!
//! Journal framing kinds cannot reach here to trip that rule. Commit markers,
//! incarnation markers, excision intents, and signed MMR roots are dropped by
//! State's record layer before any read returns, so the records this fold sees
//! are the ones a caller committed.
//!
//! # There is no prepare step
//!
//! [`crate::prepare_conversation_core`] applies excision and text withholding,
//! and both are conversation semantics: its excision match is keyed on
//! `conv-{conversation_id}` and its withholding reads paused turns. This
//! partition has neither conversations nor turns, so running it would be a
//! no-op that implied a conversation shape the partition does not have.
//!
//! # A record is never dropped
//!
//! [`AdminSignatureStatus`] states the verdict on the record's signature, and
//! every decodable record becomes a row whatever that verdict says. A payload
//! that does not decode becomes a row too, with status
//! [`AdminSignatureStatus::Malformed`] and empty payload fields — its position
//! and its status are the evidence that something was written there. Dropping
//! it would leave a hole in an administrative trail with nothing to say a hole
//! is there.
//!
//! # A row is a change that was RECORDED
//!
//! Not one that was necessarily applied. The writer appends first and installs
//! only after a receipt, so a record can be durable while the change it
//! describes was refused — an acknowledgement lost, then a receipt read that
//! failed too. That is the recoverable direction, and it is the one this
//! ordering chooses: an unmatched record says a change was attempted, while an
//! applied change with no record says nothing at all.
//!
//! Nothing in the row distinguishes the two. A reader auditing what
//! administrators did reads attempts as well as changes, and the deployment's
//! live selection is what says which took effect.
//!
//! # What this fold refuses to carry
//!
//! Signature bytes, signer public keys, attestation material, and object keys
//! never leave this module. A verdict says whether a signature verified; the
//! bytes that prove it stay on the journal, where a reader holding the
//! partition can check them for itself. `principal` is the one bearer identity
//! a row carries, because who made a change is the fact the trail exists for.

use std::collections::HashSet;

use polyc_crypto::approval::{decode_admin_model_change, verify_admin_model_change};
use polyc_crypto::signing_role::{ApprovalRole, KeyStatus, RoleTrustSet};
use polyc_eventlog_model::Event;
use polyc_proto::kinds;

/// The verdict published for one administrator-audit record.
///
/// Four values rather than [`polyc_crypto::signing_role::SignatureVerdict`]'s
/// three. Two of them split `verified` by the lifecycle of the key that signed
/// — the fact a reader auditing a key rotation is looking for — and the fourth
/// names a record that could not be read at all, which that type has no way to
/// express because a malformed payload never reaches a signature check.
///
/// # The verdict states today's trust, not the moment of signing
///
/// The two verified values read the deployment's CURRENT trust set. A key
/// retired after a record was written moves that record from
/// `verified_current` to `verified_retired`, and a rebuild republishes the
/// same journal positions under the new reading. Nothing was rewritten: the
/// column answers "what does this deployment vouch for now", which is the
/// question an auditor is asking. A reader comparing two generations sees the
/// change with no journal write behind it, and that is correct.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AdminSignatureStatus {
    /// The signature checks out under the key the role signs with NOW.
    VerifiedCurrent,
    /// The signature checks out under a key this deployment has retired and
    /// still trusts.
    VerifiedRetired,
    /// The record decoded, and its signature did not check out against a
    /// trusted key.
    ///
    /// One value for two causes on purpose. A signature that fails against the
    /// key the record carries, and a good signature under a key this
    /// deployment does not trust, are both "this deployment cannot vouch for
    /// this record" — and a reader acts on them identically.
    Unverified,
    /// The payload is not an `admin_model_change` record at all.
    Malformed,
}

impl AdminSignatureStatus {
    /// Every status, in a fixed order.
    ///
    /// A reader that restates this vocabulary iterates this instead of copying
    /// the spellings. `every_status_appears_in_all` makes a new variant a
    /// compile error, so the list cannot fall behind the type.
    pub const ALL: [Self; 4] = [
        Self::VerifiedCurrent,
        Self::VerifiedRetired,
        Self::Unverified,
        Self::Malformed,
    ];

    /// The stable spelling published in the `signature_status` column.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::VerifiedCurrent => "verified_current",
            Self::VerifiedRetired => "verified_retired",
            Self::Unverified => "unverified",
            Self::Malformed => "malformed",
        }
    }
}

impl std::fmt::Display for AdminSignatureStatus {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// One administrator model change, as the partition recorded it.
///
/// The row identity is `(partition, source_incarnation, position)`. The
/// encoder supplies the first two from the generation's source; `position` is
/// carried here because it is the only one of the three the fold can know.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdminModelChangeFact {
    /// The journal position the record was committed at.
    pub position: u64,
    /// The verdict on this record's signature.
    pub signature_status: AdminSignatureStatus,
    /// The principal the record names, empty when it did not decode.
    pub principal: String,
    /// Provider before the change, empty when it did not decode.
    pub previous_provider: String,
    /// Model before the change, empty when it did not decode.
    pub previous_model: String,
    /// Provider after the change, empty when it did not decode.
    pub new_provider: String,
    /// Model after the change, empty when it did not decode.
    pub new_model: String,
    /// Unix ms the change was applied, zero when it did not decode.
    pub changed_at_ms: u64,
}

/// Everything `administrator-audit/v1` publishes for one generation.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AdministratorAuditFacts {
    /// One row per record on the partition, in position order.
    pub model_changes: Vec<AdminModelChangeFact>,
}

/// Why an administrator-audit generation could not be folded.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum AdministratorAuditError {
    /// Two records claim one position, so the prefix is not one history.
    #[error("position {position} appears more than once in one source prefix")]
    RepeatedPosition {
        /// The position claimed twice.
        position: u64,
    },
    /// The partition carries a kind this family does not declare.
    ///
    /// A refusal rather than a skipped record: see the module doc.
    #[error("the administrator-audit partition carries an undeclared kind at position {position}")]
    UndeclaredKind {
        /// The position carrying it.
        position: u64,
    },
}

/// Folds the `admin-audit` partition prefix into its published rows.
///
/// One row per record, in the order the positions give, with no record
/// dropped. `trust` decides only the two verified statuses apart from
/// [`AdminSignatureStatus::Unverified`]; it never decides whether a row exists.
///
/// # Errors
///
/// Returns [`AdministratorAuditError::RepeatedPosition`] when one position
/// appears twice, and [`AdministratorAuditError::UndeclaredKind`] when the
/// partition carries any kind but [`kinds::ADMIN_MODEL_CHANGE`].
pub fn fold_administrator_audit(
    events: &[(u64, Event)],
    trust: &RoleTrustSet<ApprovalRole>,
) -> Result<AdministratorAuditFacts, AdministratorAuditError> {
    refuse_repeated_positions(events)?;
    let mut model_changes = Vec::with_capacity(events.len());
    for (position, event) in events {
        // The exact kind, not its base: `kinds::parse` splits a turn tag off,
        // and this partition has no turns. A tagged spelling of the one
        // declared kind is as undeclared as any other.
        if event.kind != kinds::ADMIN_MODEL_CHANGE {
            return Err(AdministratorAuditError::UndeclaredKind {
                position: *position,
            });
        }
        model_changes.push(fold_admin_model_change(*position, &event.payload, trust));
    }
    Ok(AdministratorAuditFacts { model_changes })
}

/// Folds one record, whatever its payload turns out to be.
///
/// Never returns `None` and never fails. Every outcome a record can have is a
/// row: that is the whole posture of this family, and expressing it as a total
/// function is what stops a later edit from adding a `?` that drops one.
fn fold_admin_model_change(
    position: u64,
    payload: &[u8],
    trust: &RoleTrustSet<ApprovalRole>,
) -> AdminModelChangeFact {
    let Some(decoded) = decode_admin_model_change(payload) else {
        return AdminModelChangeFact {
            position,
            signature_status: AdminSignatureStatus::Malformed,
            principal: String::new(),
            previous_provider: String::new(),
            previous_model: String::new(),
            new_provider: String::new(),
            new_model: String::new(),
            changed_at_ms: 0,
        };
    };
    // Verified against the key the payload carries first, then against what
    // this deployment trusts. A record that passes the first check and fails
    // the second is a genuine signature by a signer nobody here vouches for,
    // which is the same answer to a reader as a forged one.
    let signature_status = match verify_admin_model_change(payload) {
        None => AdminSignatureStatus::Unverified,
        Some(verified) => match trust.approval_key_status(&verified.signer_public_key) {
            None => AdminSignatureStatus::Unverified,
            Some(KeyStatus::Current) => AdminSignatureStatus::VerifiedCurrent,
            Some(KeyStatus::Retired) => AdminSignatureStatus::VerifiedRetired,
        },
    };
    AdminModelChangeFact {
        position,
        signature_status,
        principal: decoded.principal,
        previous_provider: decoded.previous_provider,
        previous_model: decoded.previous_model,
        new_provider: decoded.new_provider,
        new_model: decoded.new_model,
        changed_at_ms: decoded.changed_at_ms,
    }
}

/// Refuses a prefix that claims one position twice.
///
/// Two records at one position are not one history, and folding them would
/// publish two rows under one row identity. The same check every journal fold
/// in this crate makes, for the same reason.
fn refuse_repeated_positions(events: &[(u64, Event)]) -> Result<(), AdministratorAuditError> {
    let mut seen: HashSet<u64> = HashSet::with_capacity(events.len());
    for (position, _) in events {
        if !seen.insert(*position) {
            return Err(AdministratorAuditError::RepeatedPosition {
                position: *position,
            });
        }
    }
    Ok(())
}

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

    use polyc_crypto::approval::{ApprovalSigner, admin_model_change_payload};
    use polyc_crypto::signing_role::{SigningKeyIdentity, TrustedKey};

    use super::*;

    fn record(position: u64, payload: Vec<u8>) -> (u64, Event) {
        (position, Event::new(kinds::ADMIN_MODEL_CHANGE, payload))
    }

    fn change(signer: &ApprovalSigner, principal: &str, new_model: &str) -> Vec<u8> {
        admin_model_change_payload(
            principal,
            "prov-a",
            "model-a",
            "prov-b",
            new_model,
            1_750_000_000_000,
            signer,
        )
        .0
    }

    fn identity(signer: &ApprovalSigner) -> SigningKeyIdentity {
        SigningKeyIdentity::for_public_key::<ApprovalRole>(signer.public_key_bytes()).unwrap()
    }

    /// The current key first, the retired key second.
    fn rotated_trust(
        current: &ApprovalSigner,
        retired: &ApprovalSigner,
    ) -> RoleTrustSet<ApprovalRole> {
        RoleTrustSet::checked(vec![
            TrustedKey::current(identity(current)),
            TrustedKey::retired(identity(retired)),
        ])
        .unwrap()
    }

    #[test]
    fn every_status_appears_in_all() {
        for status in AdminSignatureStatus::ALL {
            // Exhaustive: a variant added without joining `ALL` fails to
            // compile here rather than publishing an unspelled status.
            let spelled = match status {
                AdminSignatureStatus::VerifiedCurrent => "verified_current",
                AdminSignatureStatus::VerifiedRetired => "verified_retired",
                AdminSignatureStatus::Unverified => "unverified",
                AdminSignatureStatus::Malformed => "malformed",
            };
            assert_eq!(status.as_str(), spelled);
        }
        assert_eq!(AdminSignatureStatus::ALL.len(), 4);
    }

    /// A record signed by the current key reads `verified_current`.
    #[test]
    fn a_current_signature_verifies_as_current() {
        let current = ApprovalSigner::from_seed(1);
        let retired = ApprovalSigner::from_seed(2);
        let trust = rotated_trust(&current, &retired);
        let events = vec![record(1, change(&current, "admin:root", "model-b"))];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        assert_eq!(facts.model_changes.len(), 1);
        assert_eq!(
            facts.model_changes[0].signature_status,
            AdminSignatureStatus::VerifiedCurrent
        );
        assert_eq!(facts.model_changes[0].principal, "admin:root");
        assert_eq!(facts.model_changes[0].new_model, "model-b");
        assert_eq!(facts.model_changes[0].position, 1);
    }

    /// A record signed by a retired key reads `verified_retired`.
    ///
    /// The record is genuine and the deployment still trusts what that key
    /// signed. Collapsing it into `verified_current` would hide the one fact a
    /// reader auditing a rotation is looking for.
    #[test]
    fn a_retired_signature_verifies_as_retired() {
        let current = ApprovalSigner::from_seed(1);
        let retired = ApprovalSigner::from_seed(2);
        let trust = rotated_trust(&current, &retired);
        let events = vec![record(1, change(&retired, "admin:root", "model-b"))];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        assert_eq!(
            facts.model_changes[0].signature_status,
            AdminSignatureStatus::VerifiedRetired
        );
    }

    /// The status is the key's, not the key's position in the set.
    ///
    /// The same two keys, listed in the opposite order. A fold that read
    /// "first is current" would swap both verdicts here and pass every other
    /// test in this file.
    #[test]
    fn key_order_does_not_decide_the_status() {
        let current = ApprovalSigner::from_seed(1);
        let retired = ApprovalSigner::from_seed(2);
        let reversed = RoleTrustSet::<ApprovalRole>::checked(vec![
            TrustedKey::retired(identity(&retired)),
            TrustedKey::current(identity(&current)),
        ])
        .unwrap();
        let events = vec![
            record(1, change(&current, "admin:root", "model-b")),
            record(2, change(&retired, "admin:root", "model-c")),
        ];

        let facts = fold_administrator_audit(&events, &reversed).unwrap();

        assert_eq!(
            facts.model_changes[0].signature_status,
            AdminSignatureStatus::VerifiedCurrent
        );
        assert_eq!(
            facts.model_changes[1].signature_status,
            AdminSignatureStatus::VerifiedRetired
        );
    }

    /// A genuine signature by a key nobody here trusts reads `unverified`.
    #[test]
    fn a_foreign_signature_reads_unverified_and_keeps_its_fields() {
        let current = ApprovalSigner::from_seed(1);
        let foreign = ApprovalSigner::from_seed(9);
        let trust = RoleTrustSet::<ApprovalRole>::current(&current);
        let events = vec![record(1, change(&foreign, "attacker", "evil-model"))];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        assert_eq!(
            facts.model_changes[0].signature_status,
            AdminSignatureStatus::Unverified
        );
        assert_eq!(
            facts.model_changes[0].principal, "attacker",
            "the evidence of what was written is the point of keeping the row"
        );
    }

    /// A tampered record keeps every field it still decodes.
    #[test]
    fn a_tampered_record_reads_unverified_and_is_not_dropped() {
        let current = ApprovalSigner::from_seed(1);
        let trust = RoleTrustSet::<ApprovalRole>::current(&current);
        let genuine = change(&current, "admin:root", "model-b");
        let mut value: serde_json::Value = serde_json::from_slice(&genuine).unwrap();
        value["new_model"] = serde_json::json!("evil-model");
        let events = vec![record(1, value.to_string().into_bytes())];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        assert_eq!(facts.model_changes.len(), 1);
        assert_eq!(
            facts.model_changes[0].signature_status,
            AdminSignatureStatus::Unverified
        );
        assert_eq!(facts.model_changes[0].new_model, "evil-model");
    }

    /// A malformed record is a row with empty fields, never a gap.
    #[test]
    fn a_malformed_record_is_a_row_with_empty_fields() {
        let trust = RoleTrustSet::<ApprovalRole>::current(&ApprovalSigner::from_seed(1));
        let events = vec![record(1, b"{\"not\":\"a record\"}".to_vec())];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        assert_eq!(facts.model_changes.len(), 1, "the record is not dropped");
        let row = &facts.model_changes[0];
        assert_eq!(row.signature_status, AdminSignatureStatus::Malformed);
        assert_eq!(row.position, 1);
        assert!(row.principal.is_empty());
        assert!(row.previous_provider.is_empty());
        assert!(row.previous_model.is_empty());
        assert!(row.new_provider.is_empty());
        assert!(row.new_model.is_empty());
        assert_eq!(row.changed_at_ms, 0);
    }

    /// An undeclared kind refuses the generation rather than becoming a row.
    #[test]
    fn an_undeclared_kind_refuses_the_generation() {
        let signer = ApprovalSigner::from_seed(1);
        let trust = RoleTrustSet::<ApprovalRole>::current(&signer);
        let events = vec![
            record(1, change(&signer, "admin:root", "model-b")),
            (2, Event::new("turn_start", Vec::new())),
        ];

        assert_eq!(
            fold_administrator_audit(&events, &trust),
            Err(AdministratorAuditError::UndeclaredKind { position: 2 })
        );
    }

    /// A turn-tagged spelling of the one kind is undeclared too.
    #[test]
    fn a_turn_tagged_kind_is_undeclared() {
        let signer = ApprovalSigner::from_seed(1);
        let trust = RoleTrustSet::<ApprovalRole>::current(&signer);
        let tagged = format!("{}:{}", kinds::ADMIN_MODEL_CHANGE, uuid::Uuid::nil());
        let events = vec![(1, Event::new(tagged, change(&signer, "a", "m")))];

        assert_eq!(
            fold_administrator_audit(&events, &trust),
            Err(AdministratorAuditError::UndeclaredKind { position: 1 })
        );
    }

    /// One position claimed twice refuses the prefix.
    #[test]
    fn a_repeated_position_is_refused() {
        let signer = ApprovalSigner::from_seed(1);
        let trust = RoleTrustSet::<ApprovalRole>::current(&signer);
        let events = vec![
            record(1, change(&signer, "admin:root", "model-b")),
            record(1, change(&signer, "admin:root", "model-c")),
        ];

        assert_eq!(
            fold_administrator_audit(&events, &trust),
            Err(AdministratorAuditError::RepeatedPosition { position: 1 })
        );
    }

    /// Rows come back in the order the positions give.
    #[test]
    fn rows_follow_position_order() {
        let signer = ApprovalSigner::from_seed(1);
        let trust = RoleTrustSet::<ApprovalRole>::current(&signer);
        let events = vec![
            record(4, change(&signer, "admin:root", "model-b")),
            record(9, change(&signer, "admin:root", "model-c")),
            record(11, change(&signer, "admin:root", "model-d")),
        ];

        let facts = fold_administrator_audit(&events, &trust).unwrap();

        let positions: Vec<u64> = facts.model_changes.iter().map(|row| row.position).collect();
        assert_eq!(positions, vec![4, 9, 11]);
    }

    /// An empty prefix folds to no rows rather than refusing.
    #[test]
    fn an_empty_prefix_folds_to_no_rows() {
        let trust = RoleTrustSet::<ApprovalRole>::current(&ApprovalSigner::from_seed(1));
        assert_eq!(
            fold_administrator_audit(&[], &trust).unwrap(),
            AdministratorAuditFacts::default()
        );
    }
}