polyc-state-connect 2026.8.3

//! Verified replay: a service-level operation, deliberately not a port one.
//!
//! Every other journal operation is a question any adapter could answer, which
//! is why they live on the partition-journal contract in `polyc_state::journal`
//! and why the conformance kit judges three implementations against the same
//! semantics. This one is not. Half of what a verified replay checks is the
//! storage host's own internal state — the durable event-count floor that
//! catches a self-healed truncation of already-committed data — and no port
//! operation exposes it. An implementation-independent contract would therefore
//! be a contract whose honest implementations answer different questions: one
//! that holds a floor refuses a truncated partition, and one that does not
//! reports it verified.
//!
//! So the semantics live here, at the State service, exactly as the B6 charter
//! records. [`JournalRead`] stays unwidened, the [`PartitionJournal`] alias
//! stays three traits, and the in-memory reference is not asked to fake a floor
//! it has no storage to keep. What mounts the journal over the wire requires
//! both this and the port contract, which [`JournalAuthority`] names.
//!
//! [`JournalRead`]: polyc_state::journal::JournalRead
//! [`PartitionJournal`]: polyc_state::journal::PartitionJournal

use polyc_state::{
    context::CallContext, error::StateError, id::PartitionId, journal::PartitionJournal,
};

/// A request to replay one partition and check its tamper-evidence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyPartitionReplay {
    partition: PartitionId,
}

impl VerifyPartitionReplay {
    /// Asks whether `partition` replays and verifies.
    #[must_use]
    pub const fn new(partition: PartitionId) -> Self {
        Self { partition }
    }

    /// Returns the partition asked about.
    #[must_use]
    pub const fn partition(&self) -> &PartitionId {
        &self.partition
    }
}

/// The stable class one verification violation falls into.
///
/// A violation's prose says what was found and is written for a log; this says
/// what KIND of failure it was, and is written for a caller that has to decide
/// something. The two travel together because neither substitutes for the
/// other: prose a client switches on is a contract nobody wrote down, and a
/// category alone loses the leaf counts and roots an operator needs to locate
/// the damage.
///
/// The categories are the distinctions the tamper-evidence itself already
/// draws, so nothing here is invented: a forged marker, a root signed under an
/// unexpected key, a recomputed root that disagrees, a tree that could not be
/// rebuilt, a replay shorter than its durable floor. [`Self::Unclassified`] is
/// the honest answer for a failure a build has no narrower name for, never a
/// default the others fall into.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ViolationCategory {
    /// The replay returned fewer events than the durable event-count floor —
    /// already-committed data is missing.
    TruncatedReplay,
    /// A signed-root marker's payload is not the attestation this journal
    /// writes: a corrupted or foreign record landed under the reserved kind.
    MalformedRoot,
    /// A root's signature does not verify under the expected signer — the
    /// marker was forged, or signed under a key the reader does not trust.
    SignatureInvalid,
    /// The root recomputed from the replay disagrees with the one signed over
    /// it: the content changed after it was attested.
    RootMismatch,
    /// The tamper-evident tree itself could not be rebuilt or proved.
    MerkleLog,
    /// The replay could not be read far enough to check anything — reported as
    /// a verdict, not an error, only where the module reached the storage and
    /// the storage refused the content.
    Unreadable,
    /// A repair of this partition destroyed it and did not finish restoring
    /// it. The content on disk is not what the repair promised, and the
    /// durable stage that holds the survivors is still there. This is the one
    /// category that names a state the journal can leave on purpose, so it is
    /// also the one with a remedy: retry the repair.
    RepairInterrupted,
    /// A violation this build has no narrower category for. Reported as-is
    /// rather than folded into a neighbour it does not mean.
    Unclassified,
}

impl ViolationCategory {
    /// Returns the stable identifier a caller matches on.
    ///
    /// These strings are contract, not prose: they are what a client — the
    /// forensics trace among them — persists, compares, and renders against.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::TruncatedReplay => "truncated_replay",
            Self::MalformedRoot => "malformed_root_marker",
            Self::SignatureInvalid => "signature_invalid",
            Self::RootMismatch => "root_mismatch",
            Self::MerkleLog => "mmr_error",
            Self::Unreadable => "unreadable_replay",
            Self::RepairInterrupted => "repair_interrupted",
            Self::Unclassified => "integrity_check_failed",
        }
    }

    /// Returns the category `identifier` names, or [`Self::Unclassified`] for
    /// one this build does not know.
    ///
    /// The decoding half of [`Self::as_str`]: a peer that grew a category this
    /// build predates is read as unclassified rather than refused, because a
    /// verdict that arrived is worth more than the name of its class.
    #[must_use]
    pub fn from_identifier(identifier: &str) -> Self {
        match identifier {
            "truncated_replay" => Self::TruncatedReplay,
            "malformed_root_marker" => Self::MalformedRoot,
            "signature_invalid" => Self::SignatureInvalid,
            "root_mismatch" => Self::RootMismatch,
            "mmr_error" => Self::MerkleLog,
            "unreadable_replay" => Self::Unreadable,
            "repair_interrupted" => Self::RepairInterrupted,
            _ => Self::Unclassified,
        }
    }
}

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

/// What one verified replay found.
///
/// Three outcomes, and the distinction between them is the whole value of the
/// operation: the partition verified, the partition read and does not verify,
/// or the module could not answer at all. The first two are this enum; the
/// third is a [`StateError`], because "cannot tell" is not a verdict and a
/// caller that read it as one would report a healthy log over a lost one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PartitionVerification {
    /// The partition replayed in full and every signed root over it holds.
    ///
    /// The count is what the verification was performed over, not a separate
    /// tally: a caller comparing it against its own expectation is comparing
    /// against the very events that were checked.
    Verified {
        /// How many events the verified replay returned.
        event_count: u64,
        /// How many signed roots over those events were recomputed and held.
        ///
        /// The verified tally, which is why it rides the verified arm alone: a
        /// violation stops at the first root that does not hold, so no honest
        /// total exists for one. A partition that has committed nothing yet
        /// reports zero, which is the transitional state and not a finding.
        signed_root_count: u64,
    },
    /// The partition read, and its tamper-evidence does not hold.
    ///
    /// The category is what a caller decides on; the reason is the module's own
    /// account of what it found — a tampered event, a forged marker, a root
    /// signed under an unexpected key, or a replay shorter than the durable
    /// floor. A partition that a structural trim shortened is NOT this: the
    /// trim drops the roots that attested the removed content and re-baselines
    /// the floor, so what remains verifies.
    Violation {
        /// The stable class this failure falls into.
        category: ViolationCategory,
        /// What the verification found, in the module's own words.
        reason: String,
    },
}

impl PartitionVerification {
    /// Reports that the partition verified over `event_count` events, of which
    /// `signed_root_count` signed roots were recomputed and held.
    #[must_use]
    pub const fn verified(event_count: u64, signed_root_count: u64) -> Self {
        Self::Verified {
            event_count,
            signed_root_count,
        }
    }

    /// Reports that the partition's tamper-evidence does not hold, in which
    /// class, and why.
    #[must_use]
    pub fn violation(category: ViolationCategory, reason: impl Into<String>) -> Self {
        Self::Violation {
            category,
            reason: reason.into(),
        }
    }

    /// Reports whether the partition verified.
    #[must_use]
    pub const fn is_verified(&self) -> bool {
        matches!(self, Self::Verified { .. })
    }

    /// Returns how many events verified, where the partition verified.
    #[must_use]
    pub const fn event_count(&self) -> Option<u64> {
        match self {
            Self::Verified { event_count, .. } => Some(*event_count),
            Self::Violation { .. } => None,
        }
    }

    /// Returns how many signed roots held, where the partition verified.
    #[must_use]
    pub const fn signed_root_count(&self) -> Option<u64> {
        match self {
            Self::Verified {
                signed_root_count, ..
            } => Some(*signed_root_count),
            Self::Violation { .. } => None,
        }
    }

    /// Returns the failure's class, where it found a violation.
    #[must_use]
    pub const fn category(&self) -> Option<ViolationCategory> {
        match self {
            Self::Verified { .. } => None,
            Self::Violation { category, .. } => Some(*category),
        }
    }

    /// Returns what the verification found, where it found a violation.
    #[must_use]
    pub fn reason(&self) -> Option<&str> {
        match self {
            Self::Verified { .. } => None,
            Self::Violation { reason, .. } => Some(reason),
        }
    }
}

/// Replaying one partition and checking its tamper-evidence.
///
/// # Cancellation safety
///
/// Read-only: dropping the call loses an answer and nothing more, and
/// reissuing it is free.
pub trait VerifiedReplay: Send + Sync {
    /// Replays `request`'s partition and reports whether it verifies.
    ///
    /// A partition whose tamper-evidence does not hold is
    /// [`PartitionVerification::Violation`] and never an error: the module
    /// answered the question it was asked, and the answer is bad news. An error
    /// means the module could not answer.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] for a partition name the module could
    /// not address or a storage failure that stopped the replay from being read
    /// at all, [`StateError::AmbiguousOutcome`] when the module underneath has
    /// stopped answering, and the budget outcomes every read shares.
    fn verify_replay(
        &self,
        request: VerifyPartitionReplay,
        context: &CallContext,
    ) -> Result<PartitionVerification, StateError>;
}

/// Everything State's journal authority serves over the wire.
///
/// The port contract plus the one service-level operation above. A convenience
/// alias so a composition names one bound; it adds no operations of its own.
pub trait JournalAuthority:
    PartitionJournal + polyc_state::journal::JournalDestructionWorkflow + VerifiedReplay
{
}

impl<T> JournalAuthority for T where
    T: PartitionJournal + polyc_state::journal::JournalDestructionWorkflow + VerifiedReplay
{
}

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

    use super::*;

    #[test]
    fn a_verdict_reports_exactly_one_of_a_count_and_a_reason() {
        let verified = PartitionVerification::verified(7, 2);
        assert!(verified.is_verified());
        assert_eq!(verified.event_count(), Some(7));
        assert_eq!(verified.signed_root_count(), Some(2));
        assert_eq!(verified.category(), None);
        assert_eq!(verified.reason(), None);

        let violation = PartitionVerification::violation(
            ViolationCategory::RootMismatch,
            "a root does not hold",
        );
        assert!(!violation.is_verified());
        assert_eq!(violation.event_count(), None);
        assert_eq!(violation.signed_root_count(), None);
        assert_eq!(violation.category(), Some(ViolationCategory::RootMismatch));
        assert_eq!(violation.reason(), Some("a root does not hold"));
    }

    /// The categories are a wire contract, so every one of them survives a
    /// round trip through its own identifier — and an identifier this build
    /// does not know reads as unclassified rather than as a neighbour it does
    /// not mean.
    #[test]
    fn every_category_round_trips_through_its_stable_identifier() {
        for category in [
            ViolationCategory::TruncatedReplay,
            ViolationCategory::MalformedRoot,
            ViolationCategory::SignatureInvalid,
            ViolationCategory::RootMismatch,
            ViolationCategory::MerkleLog,
            ViolationCategory::Unreadable,
            ViolationCategory::RepairInterrupted,
            ViolationCategory::Unclassified,
        ] {
            assert_eq!(
                ViolationCategory::from_identifier(category.as_str()),
                category
            );
        }
        assert_eq!(
            ViolationCategory::from_identifier("a_category_from_the_future"),
            ViolationCategory::Unclassified
        );
        assert_eq!(
            ViolationCategory::from_identifier(""),
            ViolationCategory::Unclassified
        );
    }

    /// A partition that verified over nothing is a verdict, not an absence: an
    /// empty log verifies, and reporting it as anything else would make "no
    /// events" indistinguishable from "could not tell".
    #[test]
    fn an_empty_partition_verifies_over_zero_events() {
        let verdict = PartitionVerification::verified(0, 0);
        assert!(verdict.is_verified());
        assert_eq!(verdict.event_count(), Some(0));
        assert_eq!(verdict.signed_root_count(), Some(0));
    }

    #[test]
    fn a_request_carries_the_partition_it_asks_about() {
        let request = VerifyPartitionReplay::new(PartitionId::new("conv-1"));
        assert_eq!(request.partition(), &PartitionId::new("conv-1"));
    }
}