polyc-state-connect 2026.9.0

//! The mapping between the journal contract's vocabulary and its encoding.
//!
//! Same rules as [`crate::wire`], and the same helpers: every conversion is an
//! explicit `From` or `TryFrom` naming each field, [`Kernel`] carries the local
//! type the orphan rule needs, kernel to wire is infallible, and wire to kernel
//! returns [`StateError::Malformed`] naming the field it could not interpret.
//! The decoding helpers — `malformed`, `fixed_bytes`, `required`, `known`, and
//! `completeness` — are imported from that module rather than restated here, so
//! a peer sending an enum value from a newer protocol is refused by one rule
//! and not by two that could drift apart.
//!
//! One thing is deliberately not on the wire: a batch's resource bounds. They
//! belong to the operation family, which both sides already know, and a bound a
//! caller could send is a bound a caller could widen.

use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{
        CommandEnvelope, CommandMetadata, CommandScope, FencingToken, Precondition, ResourceBounds,
    },
    digest::ContentDigest,
    error::StateError,
    id::{AggregateId, Audience, CommandId, NamespaceId, PartitionId, Purpose},
    journal::{
        self, CommitJournalBatch, DestroyPartition, ExcisePartitionRecords, GetJournalHead,
        GetJournalProof, GetJournalRoot, GetJournalSource, JournalAnchor, JournalAttestation,
        JournalDirectoryPage, JournalDirectorySnapshot, JournalDirectorySnapshotId,
        JournalInclusionProof, JournalProof, JournalRange, JournalRecord, JournalRecordDraft,
        JournalSourceHead, ListJournalDirectorySnapshot, LogicalJournalHead, QuarantinedRecord,
        ReadJournalRange, RecordDecision, RecordDisposition, RecordKind, RecordTrust,
        RepairPartition,
    },
    revision::{CommitRoot, JournalPosition, JournalSource, PartitionIncarnation},
};

use crate::{
    journal::verify::{PartitionVerification, ViolationCategory},
    wire::{Kernel, completeness, fixed_bytes, known, malformed, required},
};

// ---------------------------------------------------------------------------
// Records
// ---------------------------------------------------------------------------

impl From<Kernel<RecordTrust>> for pb::RecordTrust {
    fn from(value: Kernel<RecordTrust>) -> Self {
        match value.0 {
            RecordTrust::Unspecified => Self::RECORD_TRUST_UNSPECIFIED,
            RecordTrust::TrustedUser => Self::RECORD_TRUST_TRUSTED_USER,
            RecordTrust::QuarantinedContent => Self::RECORD_TRUST_QUARANTINED_CONTENT,
        }
    }
}

/// Reads a record's provenance off the wire.
///
/// An unset tag is [`RecordTrust::Unspecified`] rather than an error — that is
/// exactly what "the writer declared no provenance" means — but a value from a
/// newer protocol is refused, because guessing at a provenance class is the one
/// mistake that would silently widen trust.
fn record_trust(
    field: &str,
    value: buffa::EnumValue<pb::RecordTrust>,
) -> Result<RecordTrust, StateError> {
    match known(field, value)? {
        pb::RecordTrust::RECORD_TRUST_UNSPECIFIED => Ok(RecordTrust::Unspecified),
        pb::RecordTrust::RECORD_TRUST_TRUSTED_USER => Ok(RecordTrust::TrustedUser),
        pb::RecordTrust::RECORD_TRUST_QUARANTINED_CONTENT => Ok(RecordTrust::QuarantinedContent),
    }
}

impl From<Kernel<&JournalRecordDraft>> for pb::JournalRecordDraft {
    fn from(value: Kernel<&JournalRecordDraft>) -> Self {
        Self {
            kind: value.0.kind().as_str().to_owned(),
            trust: pb::RecordTrust::from(Kernel(value.0.trust())).into(),
            payload: value.0.payload().to_vec(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalRecordDraft> for Kernel<JournalRecordDraft> {
    type Error = StateError;

    /// Reads one record a batch asks the journal to commit.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `kind` when the record declares
    /// no schema discriminator, and naming `trust` for a provenance class this
    /// build does not know.
    fn try_from(value: pb::JournalRecordDraft) -> Result<Self, Self::Error> {
        let kind = RecordKind::new(value.kind);
        if kind.is_empty() {
            return Err(malformed(
                "kind",
                "a record declares the schema its payload follows",
            ));
        }
        let trust = record_trust("trust", value.trust)?;
        Ok(Self(
            JournalRecordDraft::new(kind, value.payload).with_trust(trust),
        ))
    }
}

impl From<Kernel<&JournalRecord>> for pb::JournalRecord {
    fn from(value: Kernel<&JournalRecord>) -> Self {
        use polyc_state::page::Positioned as _;
        Self {
            position: value.0.position().get(),
            kind: value.0.kind().as_str().to_owned(),
            trust: pb::RecordTrust::from(Kernel(value.0.trust())).into(),
            payload: value.0.payload().to_vec(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalRecord> for Kernel<JournalRecord> {
    type Error = StateError;

    /// Reads one committed record.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `trust` for a provenance class
    /// this build does not know.
    fn try_from(value: pb::JournalRecord) -> Result<Self, Self::Error> {
        let trust = record_trust("trust", value.trust)?;
        Ok(Self(JournalRecord::new(
            JournalPosition::new(value.position),
            RecordKind::new(value.kind),
            trust,
            value.payload,
        )))
    }
}

// ---------------------------------------------------------------------------
// Commit
// ---------------------------------------------------------------------------

/// Returns the bounds every journal command declares.
///
/// The family's own, not the caller's: a command may be refused against these
/// but can never widen them, which is why they never travel.
const fn journal_bounds() -> ResourceBounds {
    ResourceBounds::new(
        journal::MAX_BATCH_PAYLOAD_BYTES,
        journal::MAX_RECORDS_PER_BATCH,
    )
}

/// The identity, addressing, and preconditions every journal command carries,
/// as they arrive off the wire.
///
/// A batch and a mutation declare the same nine things in the same way, so they
/// are read once here rather than twice with a chance of drifting.
struct WireMetadata {
    command_id: String,
    partition: String,
    aggregate: String,
    namespace: String,
    purpose: String,
    audience: String,
    digest: Vec<u8>,
    precondition: Option<pb::Precondition>,
    fence: Option<u64>,
}

impl TryFrom<WireMetadata> for Kernel<CommandMetadata> {
    type Error = StateError;

    /// Rebuilds the metadata a journal command carries from its wire fields.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digest` when the digest is not
    /// [`ContentDigest::LEN`] bytes, and `precondition` when the command names
    /// none.
    fn try_from(wire: WireMetadata) -> Result<Self, Self::Error> {
        let digest = ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
            "digest",
            &wire.digest,
        )?);
        let precondition = wire.precondition.ok_or_else(|| {
            malformed(
                "precondition",
                "a precondition names what durable state the command requires",
            )
        })?;
        let precondition = Kernel::<Precondition>::try_from(precondition)?.into_inner();

        // The protocol version is deliberately left at
        // `ProtocolVersion::CURRENT` rather than read off the command: it rides
        // on the call's `CallContext` and is checked at admission, before a
        // handler runs. A second copy inside the command would be a second
        // place a peer could disagree with itself.
        let mut metadata = CommandMetadata::new(
            CommandId::new(wire.command_id),
            journal::family(),
            digest,
            CommandScope::new(
                AggregateId::new(wire.aggregate),
                PartitionId::new(wire.partition),
                NamespaceId::new(wire.namespace),
            ),
            CommandEnvelope::new(
                Purpose::new(wire.purpose),
                Audience::new(wire.audience),
                journal_bounds(),
            ),
        )
        .with_precondition(precondition);
        if let Some(fence) = wire.fence {
            metadata = metadata.with_fence(FencingToken::new(fence));
        }
        Ok(Self(metadata))
    }
}

impl From<Kernel<&CommitJournalBatch>> for pb::CommitJournalBatch {
    fn from(value: Kernel<&CommitJournalBatch>) -> Self {
        let metadata = value.0.metadata();
        Self {
            command_id: metadata.command_id().as_str().to_owned(),
            partition: metadata.scope().partition().as_str().to_owned(),
            aggregate: metadata.scope().aggregate().as_str().to_owned(),
            namespace: metadata.scope().namespace().as_str().to_owned(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            audience: metadata.envelope().audience().as_str().to_owned(),
            digest: metadata.digest().as_bytes().to_vec(),
            precondition: buffa::MessageField::some(pb::Precondition::from(Kernel(
                metadata.precondition(),
            ))),
            fence: metadata.fence().map(FencingToken::get),
            records: value
                .0
                .records()
                .iter()
                .map(|record| pb::JournalRecordDraft::from(Kernel(record)))
                .collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::CommitJournalBatch> for Kernel<CommitJournalBatch> {
    type Error = StateError;

    /// Reads one atomic append.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digest` when the digest is not
    /// [`ContentDigest::LEN`] bytes, `precondition` when the batch names none,
    /// and `kind` or `trust` for a record this build cannot read.
    fn try_from(value: pb::CommitJournalBatch) -> Result<Self, Self::Error> {
        let metadata = Kernel::<CommandMetadata>::try_from(WireMetadata {
            command_id: value.command_id,
            partition: value.partition,
            aggregate: value.aggregate,
            namespace: value.namespace,
            purpose: value.purpose,
            audience: value.audience,
            digest: value.digest,
            precondition: value.precondition.into_option(),
            fence: value.fence,
        })?
        .into_inner();
        let records = value
            .records
            .into_iter()
            .map(|record| Kernel::<JournalRecordDraft>::try_from(record).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Self(CommitJournalBatch::new(metadata, records)))
    }
}

// ---------------------------------------------------------------------------
// Reads
// ---------------------------------------------------------------------------

impl From<Kernel<&JournalSource>> for pb::JournalSource {
    fn from(value: Kernel<&JournalSource>) -> Self {
        Self {
            partition: value.0.partition().as_str().to_owned(),
            incarnation: value.0.incarnation().as_bytes().to_vec(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalSource> for Kernel<JournalSource> {
    type Error = StateError;

    fn try_from(value: pb::JournalSource) -> Result<Self, Self::Error> {
        let pb::JournalSource {
            partition,
            incarnation,
            __buffa_unknown_fields: _,
        } = value;
        if partition.is_empty() {
            return Err(malformed(
                "source",
                "a journal source names a non-empty partition",
            ));
        }
        let incarnation =
            PartitionIncarnation::from_bytes(fixed_bytes::<32>("incarnation", &incarnation)?);
        Ok(Self(JournalSource::new(
            PartitionId::new(partition),
            incarnation,
        )))
    }
}

impl From<Kernel<&JournalAnchor>> for pb::JournalAnchor {
    fn from(value: Kernel<&JournalAnchor>) -> Self {
        Self {
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(value.0.source()))),
            head: value.0.head().get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalAnchor> for Kernel<JournalAnchor> {
    type Error = StateError;

    fn try_from(value: pb::JournalAnchor) -> Result<Self, Self::Error> {
        let pb::JournalAnchor {
            source,
            head,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a journal anchor names an exact physical source",
            source,
        )?)?
        .into_inner();
        Ok(Self(JournalAnchor::new(source, JournalPosition::new(head))))
    }
}

impl From<Kernel<&ReadJournalRange>> for pb::ReadJournalRange {
    fn from(value: Kernel<&ReadJournalRange>) -> Self {
        Self {
            partition: value.0.partition().as_str().to_owned(),
            start: value.0.start().get(),
            limit: value.0.limit(),
            anchor: value
                .0
                .anchor()
                .map_or_else(buffa::MessageField::default, |anchor| {
                    buffa::MessageField::some(pb::JournalAnchor::from(Kernel(anchor)))
                }),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::ReadJournalRange> for Kernel<ReadJournalRange> {
    type Error = StateError;

    fn try_from(value: pb::ReadJournalRange) -> Result<Self, Self::Error> {
        let pb::ReadJournalRange {
            partition,
            start,
            limit,
            anchor,
            __buffa_unknown_fields: _,
        } = value;
        let request = ReadJournalRange::new(
            PartitionId::new(partition),
            JournalPosition::new(start),
            limit,
        );
        Ok(Self(match anchor.into_option() {
            Some(anchor) => request.within(Kernel::<JournalAnchor>::try_from(anchor)?.into_inner()),
            None => request,
        }))
    }
}

impl From<Kernel<&JournalRange>> for pb::JournalRange {
    /// Writes exactly one of `anchor` and `absent_partition`.
    ///
    /// An anchored range takes its partition from the anchor's own source, so
    /// `absent_partition` stays unset rather than repeating it. An absent one
    /// names the partition and carries no anchor, because no durable source
    /// exists to name.
    fn from(value: Kernel<&JournalRange>) -> Self {
        let (anchor, absent_partition) = value.0.anchor().map_or_else(
            || {
                (
                    buffa::MessageField::none(),
                    Some(value.0.partition().as_str().to_owned()),
                )
            },
            |anchor| {
                (
                    buffa::MessageField::some(pb::JournalAnchor::from(Kernel(anchor))),
                    None,
                )
            },
        );
        Self {
            anchor,
            records: value
                .0
                .records()
                .iter()
                .map(|record| pb::JournalRecord::from(Kernel(record)))
                .collect(),
            next_position: value.0.next_position().map(JournalPosition::get),
            completeness: pb::PageCompleteness::from(Kernel(value.0.completeness())).into(),
            absent_partition,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalRange> for Kernel<JournalRange> {
    type Error = StateError;

    /// Reads one bounded range and what it observed.
    ///
    /// Exactly one of `anchor` and `absent_partition` must be set. Neither is
    /// normalized into the other: a reply claiming both has described two
    /// different observations, and a reply claiming neither has described none.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `anchor` when the range sets
    /// both or neither observation, `absent_partition` when an absent range
    /// carries records, a resume cursor, or anything but complete, and
    /// `completeness` when it does not say whether records remain.
    fn try_from(value: pb::JournalRange) -> Result<Self, Self::Error> {
        let completeness = completeness("completeness", value.completeness)?;
        let records = value
            .records
            .into_iter()
            .map(|record| Kernel::<JournalRecord>::try_from(record).map(Kernel::into_inner))
            .collect::<Result<Vec<_>, _>>()?;
        let next = value.next_position.map(JournalPosition::new);

        match (value.anchor.into_option(), value.absent_partition) {
            (Some(_), Some(_)) => Err(StateError::Malformed {
                field: "anchor".to_owned(),
                reason: "a bounded read observes either an anchored prefix or an absent \
                         partition, never both"
                    .to_owned(),
            }),
            (None, None) => Err(StateError::Malformed {
                field: "anchor".to_owned(),
                reason: "a bounded read names the immutable prefix it observed, or the \
                         partition it found nothing physical for"
                    .to_owned(),
            }),
            (Some(anchor), None) => Ok(Self(JournalRange::anchored(
                Kernel::<JournalAnchor>::try_from(anchor)?.into_inner(),
                records,
                next,
                completeness,
            ))),
            (None, Some(partition)) => {
                if partition.is_empty() {
                    return Err(StateError::Malformed {
                        field: "absent_partition".to_owned(),
                        reason: "an absent observation names the exact partition it read"
                            .to_owned(),
                    });
                }
                if !records.is_empty()
                    || next.is_some()
                    || completeness != polyc_state::page::PageCompleteness::Complete
                {
                    return Err(StateError::Malformed {
                        field: "absent_partition".to_owned(),
                        reason: "a partition that does not exist holds no records, resumes \
                                 nowhere, and is complete"
                            .to_owned(),
                    });
                }
                Ok(Self(JournalRange::absent(PartitionId::new(partition))))
            }
        }
    }
}

impl From<Kernel<LogicalJournalHead>> for pb::JournalHead {
    fn from(value: Kernel<LogicalJournalHead>) -> Self {
        Self {
            position: value.0.position().get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalHead> for Kernel<LogicalJournalHead> {
    type Error = StateError;

    /// Reads one partition's head.
    ///
    /// # Errors
    ///
    fn try_from(value: pb::JournalHead) -> Result<Self, Self::Error> {
        let pb::JournalHead {
            position,
            __buffa_unknown_fields: _,
        } = value;
        Ok(Self(LogicalJournalHead::new(JournalPosition::new(
            position,
        ))))
    }
}

impl From<Kernel<&JournalDirectorySnapshot>> for pb::JournalDirectorySnapshot {
    fn from(value: Kernel<&JournalDirectorySnapshot>) -> Self {
        Self {
            snapshot: value.0.id().as_str().to_owned(),
            partition_count: value.0.partition_count(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalDirectorySnapshot> for Kernel<JournalDirectorySnapshot> {
    type Error = StateError;

    /// Reads one bounded directory snapshot identity and count.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] when the identity is empty or the
    /// captured count exceeds the directory snapshot bound.
    fn try_from(value: pb::JournalDirectorySnapshot) -> Result<Self, Self::Error> {
        if value.snapshot.is_empty() {
            return Err(malformed(
                "snapshot",
                "a successful creation carries an identity",
            ));
        }
        if value.partition_count > journal::MAX_DIRECTORY_SNAPSHOT_PARTITIONS as u64 {
            return Err(malformed(
                "partition_count",
                "the count exceeds the snapshot bound",
            ));
        }
        Ok(Self(JournalDirectorySnapshot::new(
            JournalDirectorySnapshotId::new(value.snapshot),
            value.partition_count,
        )))
    }
}

impl From<Kernel<&JournalDirectoryPage>> for pb::JournalDirectoryPage {
    fn from(value: Kernel<&JournalDirectoryPage>) -> Self {
        Self {
            snapshot: value.0.snapshot().as_str().to_owned(),
            partitions: value
                .0
                .partitions()
                .iter()
                .map(|partition| partition.as_str().to_owned())
                .collect(),
            next_after: value
                .0
                .next_after()
                .map(|partition| partition.as_str().to_owned()),
            completeness: pb::PageCompleteness::from(Kernel(value.0.completeness())).into(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalDirectoryPage> for Kernel<JournalDirectoryPage> {
    type Error = StateError;

    /// Reads one bounded page of partitions.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] when the snapshot identity is empty,
    /// the page exceeds its transport bound, a partition name is empty, or
    /// `completeness` does not say whether partitions remain.
    fn try_from(value: pb::JournalDirectoryPage) -> Result<Self, Self::Error> {
        if value.snapshot.is_empty() {
            return Err(malformed(
                "snapshot",
                "a directory page carries an identity",
            ));
        }
        if value.partitions.len() > journal::MAX_DIRECTORY_PAGE_PARTITIONS as usize {
            return Err(malformed(
                "partitions",
                "the page exceeds the transport bound",
            ));
        }
        let completeness = completeness("completeness", value.completeness)?;
        let partitions = value
            .partitions
            .into_iter()
            .map(|partition| {
                if partition.is_empty() {
                    Err(malformed("partition", "a directory name is non-empty"))
                } else {
                    Ok(PartitionId::new(partition))
                }
            })
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Self(JournalDirectoryPage::new(
            JournalDirectorySnapshotId::new(value.snapshot),
            partitions,
            value.next_after.map(PartitionId::new),
            completeness,
        )))
    }
}

// ---------------------------------------------------------------------------
// Proof
// ---------------------------------------------------------------------------

impl From<Kernel<&JournalAttestation>> for pb::JournalAttestation {
    fn from(value: Kernel<&JournalAttestation>) -> Self {
        Self {
            root: value.0.root().as_bytes().to_vec(),
            leaf_count: value.0.leaf_count(),
            signature: value.0.signature().to_vec(),
            signer: value.0.signer().to_vec(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalAttestation> for Kernel<JournalAttestation> {
    type Error = StateError;

    /// Reads one partition's signed root.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `root` when the attested root
    /// is not [`CommitRoot::LEN`] bytes.
    fn try_from(value: pb::JournalAttestation) -> Result<Self, Self::Error> {
        let pb::JournalAttestation {
            root,
            leaf_count,
            signature,
            signer,
            __buffa_unknown_fields: _,
        } = value;
        let root = CommitRoot::from_bytes(fixed_bytes::<{ CommitRoot::LEN }>("root", &root)?);
        let signature = fixed_bytes::<{ polyc_state::feed::ATTESTATION_SIGNATURE_BYTES }>(
            "signature",
            &signature,
        )?;
        let signer =
            fixed_bytes::<{ polyc_state::feed::ATTESTATION_SIGNER_BYTES }>("signer", &signer)?;
        Ok(Self(JournalAttestation::new(
            root,
            leaf_count,
            signature.to_vec(),
            signer.to_vec(),
        )))
    }
}

impl From<Kernel<&JournalInclusionProof>> for pb::JournalInclusionProof {
    fn from(value: Kernel<&JournalInclusionProof>) -> Self {
        Self {
            position: value.0.position().get(),
            leaf_count: value.0.leaf_count(),
            inactive_peaks: value.0.inactive_peaks(),
            digests: value
                .0
                .digests()
                .iter()
                .map(|digest| digest.to_vec())
                .collect(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalInclusionProof> for Kernel<JournalInclusionProof> {
    type Error = StateError;

    /// Reads one record's path to the root that covers it.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digests` when any sibling
    /// digest is not [`CommitRoot::LEN`] bytes. A proof with a wrong-width
    /// digest is one no verifier could evaluate, so it is refused rather than
    /// padded.
    fn try_from(value: pb::JournalInclusionProof) -> Result<Self, Self::Error> {
        let digests = value
            .digests
            .into_iter()
            .map(|digest| fixed_bytes::<{ CommitRoot::LEN }>("digests", &digest))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Self(JournalInclusionProof::new(
            JournalPosition::new(value.position),
            value.leaf_count,
            value.inactive_peaks,
            digests,
        )))
    }
}

impl From<Kernel<&JournalProof>> for pb::JournalProof {
    fn from(value: Kernel<&JournalProof>) -> Self {
        Self {
            attestation: buffa::MessageField::some(pb::JournalAttestation::from(Kernel(
                value.0.attestation(),
            ))),
            inclusion: buffa::MessageField::some(pb::JournalInclusionProof::from(Kernel(
                value.0.inclusion(),
            ))),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalProof> for Kernel<JournalProof> {
    type Error = StateError;

    /// Reads one record's proof and the attestation it resolves to.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `attestation` or `inclusion`
    /// when either half is absent — half a proof proves nothing.
    fn try_from(value: pb::JournalProof) -> Result<Self, Self::Error> {
        let attestation = Kernel::<JournalAttestation>::try_from(required(
            "attestation",
            "a proof names the signed root it resolves to",
            value.attestation,
        )?)?
        .into_inner();
        let inclusion = Kernel::<JournalInclusionProof>::try_from(required(
            "inclusion",
            "a proof names the record's path to its root",
            value.inclusion,
        )?)?
        .into_inner();
        Ok(Self(JournalProof::new(attestation, inclusion)))
    }
}

// ---------------------------------------------------------------------------
// Repair and mutation
// ---------------------------------------------------------------------------

impl From<Kernel<&RecordDecision>> for pb::RecordDecision {
    fn from(value: Kernel<&RecordDecision>) -> Self {
        use pb::__buffa::oneof::record_decision::Disposition;
        let disposition = match value.0.disposition() {
            RecordDisposition::Drop => Disposition::from(pb::DropRecord::default()),
            RecordDisposition::Replace(payload) => Disposition::Replace(payload.clone()),
        };
        Self {
            position: value.0.position().get(),
            disposition: Some(disposition),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::RecordDecision> for Kernel<RecordDecision> {
    type Error = StateError;

    /// Reads what excision does to one record.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `disposition` when the sender
    /// set no variant. A decision that says nothing is not "keep": keeping is
    /// what happens to every record the command does not name at all.
    fn try_from(value: pb::RecordDecision) -> Result<Self, Self::Error> {
        use pb::__buffa::oneof::record_decision::Disposition;
        let disposition = match value.disposition {
            Some(Disposition::Drop(_)) => RecordDisposition::Drop,
            Some(Disposition::Replace(payload)) => RecordDisposition::Replace(payload),
            None => {
                return Err(malformed(
                    "disposition",
                    "an excision decision names what happens to the record",
                ));
            }
        };
        Ok(Self(RecordDecision::new(
            JournalPosition::new(value.position),
            disposition,
        )))
    }
}

impl From<Kernel<&CommandMetadata>> for pb::JournalMutationCommand {
    /// Encodes the identity, addressing, and preconditions a mutation carries.
    fn from(value: Kernel<&CommandMetadata>) -> Self {
        let metadata = value.0;
        Self {
            command_id: metadata.command_id().as_str().to_owned(),
            partition: metadata.scope().partition().as_str().to_owned(),
            aggregate: metadata.scope().aggregate().as_str().to_owned(),
            namespace: metadata.scope().namespace().as_str().to_owned(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            audience: metadata.envelope().audience().as_str().to_owned(),
            digest: metadata.digest().as_bytes().to_vec(),
            precondition: buffa::MessageField::some(pb::Precondition::from(Kernel(
                metadata.precondition(),
            ))),
            fence: metadata.fence().map(FencingToken::get),
            expected_root: None,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalMutationCommand> for Kernel<CommandMetadata> {
    type Error = StateError;

    /// Reads the metadata a journal mutation carries.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `digest` or `precondition`,
    /// whichever the sender sized wrong or left unset.
    fn try_from(value: pb::JournalMutationCommand) -> Result<Self, Self::Error> {
        Self::try_from(WireMetadata {
            command_id: value.command_id,
            partition: value.partition,
            aggregate: value.aggregate,
            namespace: value.namespace,
            purpose: value.purpose,
            audience: value.audience,
            digest: value.digest,
            precondition: value.precondition.into_option(),
            fence: value.fence,
        })
    }
}

impl From<Kernel<&DestroyPartition>> for pb::JournalMutationCommand {
    fn from(value: Kernel<&DestroyPartition>) -> Self {
        let mut command = Self::from(Kernel(value.0.metadata()));
        command.expected_root = value.0.expected_root().map(|root| root.as_bytes().to_vec());
        command
    }
}

impl TryFrom<pb::JournalMutationCommand> for Kernel<DestroyPartition> {
    type Error = StateError;

    fn try_from(value: pb::JournalMutationCommand) -> Result<Self, Self::Error> {
        let expected_root = value
            .expected_root
            .as_deref()
            .map(|bytes| fixed_bytes::<{ CommitRoot::LEN }>("expected_root", bytes))
            .transpose()?
            .map(CommitRoot::from_bytes);
        let metadata = Kernel::<CommandMetadata>::try_from(value)?.into_inner();
        let command = DestroyPartition::new(metadata);
        Ok(Self(match expected_root {
            Some(root) => command.at_root(root),
            None => command,
        }))
    }
}

impl From<Kernel<&RepairPartition>> for pb::JournalMutationCommand {
    fn from(value: Kernel<&RepairPartition>) -> Self {
        Self::from(Kernel(value.0.metadata()))
    }
}

impl From<Kernel<&ExcisePartitionRecords>> for pb::JournalMutationCommand {
    fn from(value: Kernel<&ExcisePartitionRecords>) -> Self {
        Self::from(Kernel(value.0.metadata()))
    }
}

impl From<Kernel<&QuarantinedRecord>> for pb::QuarantinedRecord {
    fn from(value: Kernel<&QuarantinedRecord>) -> Self {
        Self {
            position: value.0.position().get(),
            reason: value.0.reason().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<pb::QuarantinedRecord> for Kernel<QuarantinedRecord> {
    fn from(value: pb::QuarantinedRecord) -> Self {
        Self(QuarantinedRecord::new(
            JournalPosition::new(value.position),
            value.reason,
        ))
    }
}

// ---------------------------------------------------------------------------
// Request shapes the client and the listener both build
// ---------------------------------------------------------------------------

impl From<Kernel<&GetJournalHead>> for String {
    fn from(value: Kernel<&GetJournalHead>) -> Self {
        value.0.partition().as_str().to_owned()
    }
}

impl From<Kernel<&GetJournalSource>> for String {
    fn from(value: Kernel<&GetJournalSource>) -> Self {
        value.0.partition().as_str().to_owned()
    }
}

impl From<Kernel<&JournalSourceHead>> for pb::JournalSourceHead {
    fn from(value: Kernel<&JournalSourceHead>) -> Self {
        Self {
            source: buffa::MessageField::some(pb::JournalSource::from(Kernel(value.0.source()))),
            position: value.0.position().get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::JournalSourceHead> for Kernel<JournalSourceHead> {
    type Error = StateError;

    fn try_from(value: pb::JournalSourceHead) -> Result<Self, Self::Error> {
        let pb::JournalSourceHead {
            source,
            position,
            __buffa_unknown_fields: _,
        } = value;
        let source = Kernel::<JournalSource>::try_from(required(
            "source",
            "a source head names an exact physical source",
            source,
        )?)?
        .into_inner();
        Ok(Self(JournalSourceHead::new(
            source,
            JournalPosition::new(position),
        )))
    }
}

impl From<Kernel<&GetJournalRoot>> for String {
    fn from(value: Kernel<&GetJournalRoot>) -> Self {
        value.0.partition().as_str().to_owned()
    }
}

/// Builds the request a bounded partition listing asks for.
///
/// `start_after` is the name to resume after; [`None`] starts from the first
/// partition, which is what a caller paging from the beginning sends.
#[must_use]
pub fn directory_listing(
    snapshot: String,
    start_after: Option<String>,
    limit: u32,
) -> ListJournalDirectorySnapshot {
    let request =
        ListJournalDirectorySnapshot::new(JournalDirectorySnapshotId::new(snapshot), limit);
    match start_after {
        Some(after) => request.after(PartitionId::new(after)),
        None => request,
    }
}

/// Builds the request a proof read asks for.
#[must_use]
pub fn proof_request(partition: String, position: u64) -> GetJournalProof {
    GetJournalProof::new(PartitionId::new(partition), JournalPosition::new(position))
}

// ---------------------------------------------------------------------------
// The verified-replay verdict
// ---------------------------------------------------------------------------

impl From<Kernel<&PartitionVerification>> for pb::VerifyPartitionReplayReply {
    fn from(value: Kernel<&PartitionVerification>) -> Self {
        use pb::__buffa::oneof::verify_partition_replay_reply::Verdict;
        let verdict = match value.0 {
            PartitionVerification::Verified {
                event_count,
                signed_root_count,
            } => Verdict::from(pb::ReplayVerified {
                event_count: *event_count,
                signed_root_count: *signed_root_count,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }),
            PartitionVerification::Violation { category, reason } => {
                Verdict::from(pb::ReplayViolation {
                    reason: reason.clone(),
                    category: category.as_str().to_owned(),
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                })
            }
        };
        Self {
            verdict: Some(verdict),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl TryFrom<pb::VerifyPartitionReplayReply> for Kernel<PartitionVerification> {
    type Error = StateError;

    /// Reads the verdict one verified replay reached.
    ///
    /// # Errors
    ///
    /// Returns [`StateError::Malformed`] naming `verdict` when the listener set
    /// no variant. A missing verdict is not "verified": the caller asked
    /// whether a log holds, and an empty answer must never read as yes.
    fn try_from(value: pb::VerifyPartitionReplayReply) -> Result<Self, Self::Error> {
        use pb::__buffa::oneof::verify_partition_replay_reply::Verdict;
        let verification = match value.verdict {
            Some(Verdict::Verified(verified)) => {
                PartitionVerification::verified(verified.event_count, verified.signed_root_count)
            }
            Some(Verdict::Violation(violation)) => PartitionVerification::violation(
                ViolationCategory::from_identifier(&violation.category),
                violation.reason.clone(),
            ),
            None => {
                return Err(malformed(
                    "verdict",
                    "a verified replay answers with a verdict, and an absent one is not a pass",
                ));
            }
        };
        Ok(Self(verification))
    }
}

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

    use super::*;
    use polyc_state::{
        consistency::Consistency,
        id::SnapshotId,
        page::PageCompleteness,
        receipt::{CommitEvidence, Receipt},
    };

    fn drafts() -> Vec<JournalRecordDraft> {
        vec![
            JournalRecordDraft::new(RecordKind::new("turn_input"), b"one".to_vec()),
            JournalRecordDraft::new(RecordKind::new("turn_output"), b"two".to_vec())
                .with_trust(RecordTrust::QuarantinedContent),
        ]
    }

    fn batch() -> CommitJournalBatch {
        let records = drafts();
        let metadata = CommandMetadata::new(
            CommandId::new("cmd-1"),
            journal::family(),
            ContentDigest::from_bytes([5; ContentDigest::LEN]),
            CommandScope::new(
                AggregateId::new("conv-1"),
                PartitionId::new("conv-1"),
                NamespaceId::new("default"),
            ),
            CommandEnvelope::new(
                Purpose::new("turn"),
                crate::state_audience(),
                journal_bounds(),
            ),
        )
        .with_precondition(Precondition::JournalHead(JournalPosition::new(4)))
        .with_fence(FencingToken::new(3));
        CommitJournalBatch::new(metadata, records)
    }

    fn source() -> JournalSource {
        JournalSource::new(
            PartitionId::new("conv-1"),
            PartitionIncarnation::from_bytes([7; PartitionIncarnation::LEN]),
        )
    }

    #[test]
    fn a_batch_round_trips_field_for_field() {
        let original = batch();
        let back =
            Kernel::<CommitJournalBatch>::try_from(pb::CommitJournalBatch::from(Kernel(&original)))
                .unwrap()
                .into_inner();
        assert_eq!(back, original);
        assert_eq!(back.metadata().fence(), Some(FencingToken::new(3)));
        assert_eq!(
            back.metadata().precondition(),
            Precondition::JournalHead(JournalPosition::new(4))
        );
        assert_eq!(back.canonical_bytes(), original.canonical_bytes());
    }

    #[test]
    fn a_batch_with_a_wrong_width_digest_is_malformed() {
        let mut wire = pb::CommitJournalBatch::from(Kernel(&batch()));
        wire.digest = vec![1, 2, 3];
        let error = Kernel::<CommitJournalBatch>::try_from(wire).unwrap_err();
        assert!(matches!(error, StateError::Malformed { ref field, .. } if field == "digest"));
    }

    #[test]
    fn a_record_with_no_kind_is_malformed() {
        let mut wire = pb::CommitJournalBatch::from(Kernel(&batch()));
        wire.records[0].kind = String::new();
        let error = Kernel::<CommitJournalBatch>::try_from(wire).unwrap_err();
        assert!(matches!(error, StateError::Malformed { ref field, .. } if field == "kind"));
    }

    /// A provenance class from a newer protocol is refused rather than
    /// downgraded: guessing here would widen trust silently.
    #[test]
    fn an_unknown_trust_class_is_refused() {
        let mut wire = pb::CommitJournalBatch::from(Kernel(&batch()));
        wire.records[0].trust = buffa::EnumValue::from(99);
        let error = Kernel::<CommitJournalBatch>::try_from(wire).unwrap_err();
        assert!(matches!(error, StateError::Malformed { ref field, .. } if field == "trust"));
    }

    #[test]
    fn a_range_round_trips_with_its_anchor() {
        let anchor = JournalAnchor::new(source(), JournalPosition::new(7));
        let original = JournalRange::anchored(
            anchor.clone(),
            vec![JournalRecord::new(
                JournalPosition::new(1),
                RecordKind::new("k"),
                RecordTrust::TrustedUser,
                b"body".to_vec(),
            )],
            Some(JournalPosition::new(2)),
            PageCompleteness::Truncated,
        );
        let back = Kernel::<JournalRange>::try_from(pb::JournalRange::from(Kernel(&original)))
            .unwrap()
            .into_inner();
        assert_eq!(back, original);
        assert_eq!(
            back.anchor().expect("an anchored range").snapshot(),
            anchor.snapshot()
        );
        assert!(!back.is_absent());
        assert_eq!(back.partition(), &PartitionId::new("conv-1"));
        // An anchored reply carries its partition in the anchor's own source,
        // and must not repeat it in the absent field.
        assert_eq!(
            pb::JournalRange::from(Kernel(&original)).absent_partition,
            None
        );
    }

    /// An absent observation survives the wire naming the exact partition.
    #[test]
    fn an_absent_range_round_trips_with_the_partition_it_read() {
        let original = JournalRange::absent(PartitionId::new("conv-never-written"));
        let wire = pb::JournalRange::from(Kernel(&original));
        assert_eq!(
            wire.absent_partition.as_deref(),
            Some("conv-never-written"),
            "an absent observation names the exact partition it read"
        );
        assert!(
            wire.anchor.clone().into_option().is_none(),
            "and carries no anchor to name"
        );

        let back = Kernel::<JournalRange>::try_from(wire).unwrap().into_inner();
        assert_eq!(back, original);
        assert!(back.is_absent());
        assert_eq!(back.anchor(), None);
        assert_eq!(back.partition(), &PartitionId::new("conv-never-written"));
        assert!(back.is_empty());
        assert_eq!(back.next_position(), None);
        assert_eq!(back.completeness(), PageCompleteness::Complete);
    }

    fn anchored_wire() -> pb::JournalRange {
        pb::JournalRange::from(Kernel(&JournalRange::anchored(
            JournalAnchor::new(
                JournalSource::new(
                    PartitionId::new("p"),
                    PartitionIncarnation::from_bytes([8; PartitionIncarnation::LEN]),
                ),
                JournalPosition::new(1),
            ),
            Vec::new(),
            None,
            PageCompleteness::Complete,
        )))
    }

    #[test]
    fn a_range_that_names_no_observation_is_malformed() {
        let mut wire = anchored_wire();
        wire.anchor = buffa::MessageField::default();
        let error = Kernel::<JournalRange>::try_from(wire).unwrap_err();
        assert!(matches!(error, StateError::Malformed { ref field, .. } if field == "anchor"));
    }

    /// Both observations at once describes two different reads, so it is
    /// refused rather than normalized into either one.
    #[test]
    fn a_range_that_names_both_observations_is_malformed() {
        let mut wire = anchored_wire();
        wire.absent_partition = Some("p".to_owned());
        let error = Kernel::<JournalRange>::try_from(wire).unwrap_err();
        assert!(matches!(error, StateError::Malformed { ref field, .. } if field == "anchor"));
    }

    /// A partition that does not exist holds nothing, resumes nowhere, and is
    /// complete. Every other absent shape is a contradiction on the wire.
    #[test]
    fn a_malformed_absent_range_is_refused() {
        let base = pb::JournalRange::from(Kernel(&JournalRange::absent(PartitionId::new("p"))));

        let mut with_records = base.clone();
        with_records.records = vec![pb::JournalRecord::from(Kernel(&JournalRecord::new(
            JournalPosition::new(1),
            RecordKind::new("k"),
            RecordTrust::Unspecified,
            Vec::new(),
        )))];
        let error = Kernel::<JournalRange>::try_from(with_records).unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "absent_partition")
        );

        let mut with_cursor = base.clone();
        with_cursor.next_position = Some(2);
        let error = Kernel::<JournalRange>::try_from(with_cursor).unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "absent_partition")
        );

        let mut truncated = base.clone();
        truncated.completeness =
            pb::PageCompleteness::from(Kernel(PageCompleteness::Truncated)).into();
        let error = Kernel::<JournalRange>::try_from(truncated).unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "absent_partition")
        );

        let mut unnamed = base;
        unnamed.absent_partition = Some(String::new());
        let error = Kernel::<JournalRange>::try_from(unnamed).unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "absent_partition")
        );
    }

    #[test]
    fn a_read_request_round_trips_with_and_without_an_anchor() {
        let bare = ReadJournalRange::new(PartitionId::new("conv-1"), JournalPosition::new(3), 8);
        assert_eq!(
            Kernel::<ReadJournalRange>::try_from(pb::ReadJournalRange::from(Kernel(&bare)))
                .unwrap()
                .into_inner(),
            bare
        );

        let anchored = bare
            .clone()
            .within(JournalAnchor::new(source(), JournalPosition::new(9)));
        assert_eq!(
            Kernel::<ReadJournalRange>::try_from(pb::ReadJournalRange::from(Kernel(&anchored)))
                .unwrap()
                .into_inner(),
            anchored
        );
    }

    #[test]
    fn a_head_wire_carries_only_the_logical_position() {
        for head in [
            LogicalJournalHead::new(JournalPosition::new(0)),
            LogicalJournalHead::new(JournalPosition::new(5)),
        ] {
            let back = Kernel::<LogicalJournalHead>::try_from(pb::JournalHead::from(Kernel(head)))
                .unwrap()
                .into_inner();
            assert_eq!(back.position(), head.position());
        }
    }

    #[test]
    fn a_directory_page_round_trips() {
        let original = JournalDirectoryPage::new(
            JournalDirectorySnapshotId::new("directory-1"),
            vec![PartitionId::new("conv-1")],
            Some(PartitionId::new("conv-1")),
            PageCompleteness::Truncated,
        );
        let back = Kernel::<JournalDirectoryPage>::try_from(pb::JournalDirectoryPage::from(
            Kernel(&original),
        ))
        .unwrap()
        .into_inner();
        assert_eq!(back, original);
        assert!(
            back.validate(&ListJournalDirectorySnapshot::new(
                JournalDirectorySnapshotId::new("directory-1"),
                1,
            ))
            .is_ok()
        );
    }

    #[test]
    fn a_directory_snapshot_round_trips() {
        let original =
            JournalDirectorySnapshot::new(JournalDirectorySnapshotId::new("directory-1"), 7);
        let back = Kernel::<JournalDirectorySnapshot>::try_from(
            pb::JournalDirectorySnapshot::from(Kernel(&original)),
        )
        .unwrap()
        .into_inner();
        assert_eq!(back, original);

        let mut malformed = pb::JournalDirectorySnapshot::from(Kernel(&original));
        malformed.snapshot.clear();
        assert!(Kernel::<JournalDirectorySnapshot>::try_from(malformed).is_err());
    }

    #[test]
    fn malformed_directory_snapshot_messages_fail_closed() {
        let page = JournalDirectoryPage::new(
            JournalDirectorySnapshotId::new("directory-1"),
            vec![PartitionId::new("conv-1")],
            None,
            PageCompleteness::Complete,
        );
        let mut missing_identity = pb::JournalDirectoryPage::from(Kernel(&page));
        missing_identity.snapshot.clear();
        assert!(Kernel::<JournalDirectoryPage>::try_from(missing_identity).is_err());

        let mut empty_name = pb::JournalDirectoryPage::from(Kernel(&page));
        empty_name.partitions[0].clear();
        assert!(Kernel::<JournalDirectoryPage>::try_from(empty_name).is_err());

        let mut oversized = pb::JournalDirectoryPage::from(Kernel(&page));
        oversized.partitions =
            vec!["conv-1".to_owned(); journal::MAX_DIRECTORY_PAGE_PARTITIONS as usize + 1];
        assert!(Kernel::<JournalDirectoryPage>::try_from(oversized).is_err());

        let mut unspecified = pb::JournalDirectoryPage::from(Kernel(&page));
        unspecified.completeness = pb::PageCompleteness::PAGE_COMPLETENESS_UNSPECIFIED.into();
        assert!(Kernel::<JournalDirectoryPage>::try_from(unspecified).is_err());
    }

    #[test]
    fn a_proof_round_trips_and_a_wrong_width_digest_is_refused() {
        let original = JournalProof::new(
            JournalAttestation::new(
                CommitRoot::from_bytes([8; CommitRoot::LEN]),
                6,
                vec![1; polyc_state::feed::ATTESTATION_SIGNATURE_BYTES],
                vec![4; polyc_state::feed::ATTESTATION_SIGNER_BYTES],
            ),
            JournalInclusionProof::new(
                JournalPosition::new(2),
                6,
                1,
                vec![[9; CommitRoot::LEN], [10; CommitRoot::LEN]],
            ),
        );
        let wire = pb::JournalProof::from(Kernel(&original));
        assert_eq!(
            Kernel::<JournalProof>::try_from(wire.clone())
                .unwrap()
                .into_inner(),
            original
        );
        let wire = pb::JournalProof::from(Kernel(&original));

        let mut broken_inclusion = wire.inclusion.into_option().unwrap();
        broken_inclusion.digests[0] = vec![1];
        let broken = pb::JournalProof {
            attestation: wire.attestation,
            inclusion: buffa::MessageField::some(broken_inclusion),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        assert!(matches!(
            Kernel::<JournalProof>::try_from(broken).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "digests"
        ));
    }

    #[test]
    fn an_excision_decision_round_trips_both_dispositions() {
        for decision in [
            RecordDecision::new(JournalPosition::new(1), RecordDisposition::Drop),
            RecordDecision::new(
                JournalPosition::new(2),
                RecordDisposition::Replace(b"redacted".to_vec()),
            ),
        ] {
            let back =
                Kernel::<RecordDecision>::try_from(pb::RecordDecision::from(Kernel(&decision)))
                    .unwrap()
                    .into_inner();
            assert_eq!(back, decision);
        }
    }

    /// A decision that names nothing is not "keep": keeping is what happens to
    /// every record a command does not name at all.
    #[test]
    fn a_decision_with_no_disposition_is_malformed() {
        let mut wire = pb::RecordDecision::from(Kernel(&RecordDecision::new(
            JournalPosition::new(1),
            RecordDisposition::Drop,
        )));
        wire.disposition = None;
        assert!(matches!(
            Kernel::<RecordDecision>::try_from(wire).unwrap_err(),
            StateError::Malformed { ref field, .. } if field == "disposition"
        ));
    }

    #[test]
    fn a_mutation_commands_identity_and_preconditions_round_trip() {
        let original = ExcisePartitionRecords::new(
            batch().metadata().clone(),
            vec![RecordDecision::new(
                JournalPosition::new(1),
                RecordDisposition::Drop,
            )],
        );
        let back = Kernel::<CommandMetadata>::try_from(pb::JournalMutationCommand::from(Kernel(
            &original,
        )))
        .unwrap()
        .into_inner();
        assert_eq!(&back, original.metadata());

        let destroy = DestroyPartition::new(batch().metadata().clone())
            .at_root(CommitRoot::from_bytes([7; CommitRoot::LEN]));
        assert_eq!(
            Kernel::<DestroyPartition>::try_from(pb::JournalMutationCommand::from(Kernel(
                &destroy
            )))
            .unwrap()
            .into_inner(),
            destroy
        );

        let repair = RepairPartition::new(batch().metadata().clone());
        assert_eq!(
            Kernel::<CommandMetadata>::try_from(pb::JournalMutationCommand::from(Kernel(&repair)))
                .unwrap()
                .into_inner(),
            *repair.metadata()
        );
    }

    #[test]
    fn a_quarantined_record_round_trips() {
        let original = QuarantinedRecord::new(JournalPosition::new(3), "decode failed");
        let back =
            Kernel::<QuarantinedRecord>::from(pb::QuarantinedRecord::from(Kernel(&original)))
                .into_inner();
        assert_eq!(back, original);
    }

    #[test]
    fn the_small_request_shapes_carry_what_they_name() {
        assert_eq!(
            String::from(Kernel(&GetJournalHead::new(PartitionId::new("conv-1")))),
            "conv-1"
        );
        assert_eq!(
            String::from(Kernel(&GetJournalRoot::new(PartitionId::new("conv-1")))),
            "conv-1"
        );
        assert_eq!(
            directory_listing("directory-1".to_owned(), Some("conv-1".to_owned()), 4,)
                .start_after(),
            Some(&PartitionId::new("conv-1"))
        );
        assert_eq!(
            directory_listing("directory-1".to_owned(), None, 4).start_after(),
            None
        );
        assert_eq!(
            proof_request("conv-1".to_owned(), 2).position(),
            JournalPosition::new(2)
        );
    }

    /// Both verdicts round-trip, and a reply that reached none is refused
    /// rather than read as a pass.
    #[test]
    fn a_verdict_round_trips_and_an_absent_one_is_never_a_pass() {
        for verdict in [
            PartitionVerification::verified(0, 0),
            PartitionVerification::verified(9, 2),
            PartitionVerification::violation(
                ViolationCategory::RootMismatch,
                "a root does not hold",
            ),
            PartitionVerification::violation(
                ViolationCategory::TruncatedReplay,
                "the replay is shorter than its durable floor",
            ),
            PartitionVerification::violation(
                ViolationCategory::Unclassified,
                "something this build has no name for",
            ),
        ] {
            let back = Kernel::<PartitionVerification>::try_from(
                pb::VerifyPartitionReplayReply::from(Kernel(&verdict)),
            )
            .unwrap()
            .into_inner();
            assert_eq!(back, verdict);
        }

        let error =
            Kernel::<PartitionVerification>::try_from(pb::VerifyPartitionReplayReply::default())
                .unwrap_err();
        assert!(
            matches!(error, StateError::Malformed { ref field, .. } if field == "verdict"),
            "got {error}"
        );
    }

    /// The journal family's receipts travel over the receipt mapping B0 already
    /// ships, so a commit's evidence survives the wire unchanged.
    #[test]
    fn a_journal_receipt_keeps_its_position_root_and_anchor() {
        let anchor = JournalAnchor::new(source(), JournalPosition::new(2));
        let evidence = CommitEvidence::new()
            .with_position(JournalPosition::new(2))
            .with_root(CommitRoot::from_bytes([6; CommitRoot::LEN]))
            .with_snapshot(anchor.snapshot());
        let receipt = Receipt::committed(
            batch().metadata(),
            evidence,
            Consistency::OrderedPerAggregate,
        );
        let back = Kernel::<Receipt>::try_from(pb::Receipt::from(Kernel(&receipt)))
            .unwrap()
            .into_inner();
        assert_eq!(back.evidence(), receipt.evidence());
        assert_eq!(
            back.evidence().snapshot().map(SnapshotId::as_str),
            Some(anchor.snapshot().as_str())
        );
    }
}