polyc-query 2026.9.0

//! Converts between the durable audit's source vector and the protocol's own.
//!
//! This crate is the one inward layer that may see both vocabularies. The
//! query protocol owns a closed, wire-neutral mirror so a client never
//! acquires the state plane's component graph; the durable audit owns the
//! authoritative types. Every conversion below names every field, so a new
//! field on either side breaks the build rather than travelling as a default.
//!
//! The two canonical orders agree by construction: both sort strictly by the
//! same identity rule, so a vector accepted by one is accepted by the other.

#![allow(
    dead_code,
    reason = "the reverse projection rebuilds a source vector from its wire form; only this module's own cases need it, and its helpers are unreachable without it"
)]

use polyc_query_model as model;
use polyc_state::digest::ContentDigest;
use polyc_state::feed::SourceCheckpoint;
use polyc_state::id::{OwnerId, PartitionId};
use polyc_state::immutable::{
    Classification, ContentReference, Generation, ObjectDescriptor, ObjectId, Retention,
};
use polyc_state::journal::{JournalAnchor, JournalAttestation};
use polyc_state::projection::artifact::{ExactObjectRef, ObjectNamespace};
use polyc_state::projection::{
    FamilyId, ProjectionGeneration, ProjectionKey, ProjectionManifest, PublisherFence, PublisherId,
};
use polyc_state::query_audit::{self as audit, ProjectionPin, SourcePin, SourceSnapshot};
use polyc_state::revision::{
    CommitRoot, JournalPosition, JournalSource, PartitionIncarnation, Revision,
};

/// Why a protocol source vector cannot become a durable one.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub(crate) enum EvidenceError {
    /// One field is empty, out of bounds, or unsorted.
    #[error("source evidence field `{0}` is outside its contract")]
    Malformed(&'static str),
}

fn width(bytes: &[u8]) -> Result<[u8; 32], EvidenceError> {
    bytes
        .try_into()
        .map_err(|_| EvidenceError::Malformed("byte width"))
}

/// Projects the durable source vector onto the protocol's own vocabulary.
pub(crate) fn evidence_of(source: &SourceSnapshot) -> Result<model::SourceEvidence, EvidenceError> {
    let pins = source
        .pins()
        .iter()
        .map(pin_of)
        .collect::<Result<Vec<_>, _>>()?;
    model::SourceEvidence::try_new(pins).map_err(|_| EvidenceError::Malformed("source_pins"))
}

fn pin_of(pin: &SourcePin) -> Result<model::SourcePin, EvidenceError> {
    Ok(match pin {
        SourcePin::Projected(projected) => {
            model::SourcePin::Projected(Box::new(manifest_of(projected.manifest())?))
        }
        SourcePin::Journal(anchor) => model::SourcePin::Journal(model::JournalAnchor::new(
            journal_source_of(anchor.source())?,
            anchor.head().get(),
        )),
        SourcePin::Authoritative(revision) => model::SourcePin::Authoritative(revision.get()),
    })
}

fn journal_source_of(source: &JournalSource) -> Result<model::JournalSource, EvidenceError> {
    model::JournalSource::try_new(
        source.partition().as_str().to_owned(),
        width(source.incarnation().as_bytes())?,
    )
    .map_err(|_| EvidenceError::Malformed("partition"))
}

fn manifest_of(manifest: &ProjectionManifest) -> Result<model::ProjectionManifest, EvidenceError> {
    model::ProjectionManifest::try_new(
        key_of(manifest.key())?,
        manifest.generation().get(),
        checkpoint_of(manifest.checkpoint())?,
        manifest.schema_version(),
        manifest.fact_version(),
        object_of(manifest.object_descriptor())?,
        exact_object_of(manifest.artifact_object())?,
        manifest.publisher().as_str().to_owned(),
        fence_of(manifest.fence())?,
    )
    .map_err(|_| EvidenceError::Malformed("manifest"))
}

fn key_of(key: &ProjectionKey) -> Result<model::ProjectionKey, EvidenceError> {
    model::ProjectionKey::try_new(
        key.family().as_str().to_owned(),
        key.source().as_str().to_owned(),
    )
    .map_err(|_| EvidenceError::Malformed("projection_key"))
}

fn fence_of(fence: &PublisherFence) -> Result<model::PublisherFence, EvidenceError> {
    model::PublisherFence::try_new(
        key_of(fence.key())?,
        width(fence.incarnation().as_bytes())?,
        fence.term(),
    )
    .map_err(|_| EvidenceError::Malformed("fence"))
}

fn checkpoint_of(checkpoint: &SourceCheckpoint) -> Result<model::SourceCheckpoint, EvidenceError> {
    Ok(model::SourceCheckpoint::new(
        journal_source_of(checkpoint.source())?,
        checkpoint.feed_position().get(),
        checkpoint.journal_position().get(),
        checkpoint.evidence_leaf(),
        attestation_of(checkpoint.covering_attestation())?,
    ))
}

fn attestation_of(
    attestation: &JournalAttestation,
) -> Result<model::JournalAttestation, EvidenceError> {
    Ok(model::JournalAttestation::new(
        width(attestation.root().as_bytes())?,
        attestation.leaf_count(),
        attestation
            .signature()
            .try_into()
            .map_err(|_| EvidenceError::Malformed("signature"))?,
        attestation
            .signer()
            .try_into()
            .map_err(|_| EvidenceError::Malformed("signer"))?,
    ))
}

fn object_of(object: &ObjectDescriptor) -> Result<model::ObjectDescriptor, EvidenceError> {
    model::ObjectDescriptor::try_new(
        object.object().as_str().to_owned(),
        object.generation().get(),
        width(object.digest().as_bytes())?,
        object.owner().as_str().to_owned(),
        classification_of(object.classification()),
        retention_of(object.retention()),
        object.byte_len(),
        object.content_reference().as_str().to_owned(),
    )
    .map_err(|_| EvidenceError::Malformed("object_descriptor"))
}

fn exact_object_of(object: &ExactObjectRef) -> Result<model::ExactObjectRef, EvidenceError> {
    model::ExactObjectRef::try_new(
        object.namespace().as_str().to_owned(),
        object.key().as_str().to_owned(),
        object.generation(),
    )
    .map_err(|_| EvidenceError::Malformed("artifact_object"))
}

const fn classification_of(classification: Classification) -> model::Classification {
    match classification {
        Classification::Public => model::Classification::Public,
        Classification::Internal => model::Classification::Internal,
        Classification::Confidential => model::Classification::Confidential,
        Classification::Restricted => model::Classification::Restricted,
    }
}

const fn retention_of(retention: Retention) -> model::Retention {
    match retention {
        Retention::For(duration) => model::Retention::For(duration),
        Retention::UntilReleased => model::Retention::UntilReleased,
    }
}

/// Rebuilds the durable source vector from the protocol's own vocabulary.
pub(crate) fn snapshot_of(
    evidence: &model::SourceEvidence,
) -> Result<SourceSnapshot, EvidenceError> {
    let pins = evidence
        .pins()
        .iter()
        .map(durable_pin_of)
        .collect::<Result<Vec<_>, _>>()?;
    SourceSnapshot::from_canonical(pins).map_err(|_| EvidenceError::Malformed("source_pins"))
}

fn durable_pin_of(pin: &model::SourcePin) -> Result<SourcePin, EvidenceError> {
    Ok(match pin {
        model::SourcePin::Projected(manifest) => {
            SourcePin::Projected(ProjectionPin::new(durable_manifest_of(manifest)?))
        }
        model::SourcePin::Journal(anchor) => SourcePin::Journal(JournalAnchor::new(
            durable_journal_source_of(anchor.source()),
            JournalPosition::new(anchor.head()),
        )),
        model::SourcePin::Authoritative(revision) => {
            SourcePin::Authoritative(Revision::new(*revision))
        }
    })
}

fn durable_journal_source_of(source: &model::JournalSource) -> JournalSource {
    JournalSource::new(
        PartitionId::new(source.partition().to_owned()),
        PartitionIncarnation::from_bytes(*source.incarnation()),
    )
}

fn durable_manifest_of(
    manifest: &model::ProjectionManifest,
) -> Result<ProjectionManifest, EvidenceError> {
    let rebuilt = ProjectionManifest::new(
        durable_key_of(manifest.key()),
        ProjectionGeneration::new(manifest.generation()),
        durable_checkpoint_of(manifest.checkpoint())?,
        manifest.schema_version(),
        manifest.fact_version(),
        durable_object_of(manifest.object())?,
        durable_exact_object_of(manifest.artifact_object())?,
        PublisherId::new(manifest.publisher().to_owned()),
        durable_fence_of(manifest.fence()),
    );
    rebuilt
        .validate_structure()
        .map_err(|_| EvidenceError::Malformed("manifest"))?;
    Ok(rebuilt)
}

fn durable_key_of(key: &model::ProjectionKey) -> ProjectionKey {
    ProjectionKey::new(
        FamilyId::new(key.family().to_owned()),
        PartitionId::new(key.source_partition().to_owned()),
    )
}

fn durable_fence_of(fence: &model::PublisherFence) -> PublisherFence {
    PublisherFence::new(
        durable_key_of(fence.key()),
        PartitionIncarnation::from_bytes(*fence.source_incarnation()),
        fence.term(),
    )
}

fn durable_checkpoint_of(
    checkpoint: &model::SourceCheckpoint,
) -> Result<SourceCheckpoint, EvidenceError> {
    SourceCheckpoint::try_new(
        durable_journal_source_of(checkpoint.source()),
        JournalPosition::new(checkpoint.feed_position()),
        JournalPosition::new(checkpoint.journal_position()),
        checkpoint.evidence_leaf(),
        durable_attestation_of(checkpoint.covering_attestation()),
    )
    .map_err(|_| EvidenceError::Malformed("checkpoint"))
}

fn durable_attestation_of(attestation: &model::JournalAttestation) -> JournalAttestation {
    JournalAttestation::new(
        CommitRoot::from_bytes(*attestation.root()),
        attestation.leaf_count(),
        attestation.signature().to_vec(),
        attestation.signer().to_vec(),
    )
}

fn durable_object_of(object: &model::ObjectDescriptor) -> Result<ObjectDescriptor, EvidenceError> {
    Ok(ObjectDescriptor::new(
        ObjectId::new(object.object().to_owned()),
        Generation::new(object.generation()),
        ContentDigest::from_bytes(*object.digest()),
        OwnerId::new(object.owner().to_owned()),
        durable_classification_of(object.classification()),
        durable_retention_of(object.retention()),
        object.byte_len(),
        ContentReference::try_new(object.content_reference().to_owned())
            .map_err(|_| EvidenceError::Malformed("content_reference"))?,
    ))
}

fn durable_exact_object_of(
    object: &model::ExactObjectRef,
) -> Result<ExactObjectRef, EvidenceError> {
    ExactObjectRef::try_new(
        ObjectNamespace::try_new(object.namespace().to_owned())
            .map_err(|_| EvidenceError::Malformed("namespace"))?,
        ContentReference::try_new(object.key().to_owned())
            .map_err(|_| EvidenceError::Malformed("key"))?,
        object.backend_generation(),
    )
    .map_err(|_| EvidenceError::Malformed("artifact_object"))
}

const fn durable_classification_of(classification: model::Classification) -> Classification {
    match classification {
        model::Classification::Public => Classification::Public,
        model::Classification::Internal => Classification::Internal,
        model::Classification::Confidential => Classification::Confidential,
        model::Classification::Restricted => Classification::Restricted,
    }
}

const fn durable_retention_of(retention: model::Retention) -> Retention {
    match retention {
        model::Retention::For(duration) => Retention::For(duration),
        model::Retention::UntilReleased => Retention::UntilReleased,
    }
}

/// Converts the durable audit's error class to the protocol's own.
///
/// Both vocabularies are closed and this match names every variant, so a new
/// class on either side breaks the build rather than arriving as `Internal`.
pub(crate) const fn error_class_of(class: audit::ErrorClass) -> model::ErrorClass {
    match class {
        audit::ErrorClass::Denied => model::ErrorClass::Denied,
        audit::ErrorClass::Deadline => model::ErrorClass::Deadline,
        audit::ErrorClass::Cancelled => model::ErrorClass::Cancelled,
        audit::ErrorClass::Bounds => model::ErrorClass::Bounds,
        audit::ErrorClass::Unavailable => model::ErrorClass::Unavailable,
        audit::ErrorClass::Malformed => model::ErrorClass::Malformed,
        audit::ErrorClass::Internal => model::ErrorClass::Internal,
    }
}

/// Converts the protocol's error class to the durable audit's own.
///
/// The inverse of [`error_class_of`], and closed the same way. A refusal the
/// consumer measures is reported to the terminal latch in this vocabulary, so
/// the durable record carries the class the caller was told.
pub(crate) const fn durable_class_of(class: model::ErrorClass) -> audit::ErrorClass {
    match class {
        model::ErrorClass::Denied => audit::ErrorClass::Denied,
        model::ErrorClass::Deadline => audit::ErrorClass::Deadline,
        model::ErrorClass::Cancelled => audit::ErrorClass::Cancelled,
        model::ErrorClass::Bounds => audit::ErrorClass::Bounds,
        model::ErrorClass::Unavailable => audit::ErrorClass::Unavailable,
        model::ErrorClass::Malformed => audit::ErrorClass::Malformed,
        model::ErrorClass::Internal => audit::ErrorClass::Internal,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use polyc_state::feed::{ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES};
    use std::time::Duration;

    fn durable_source(partition: &str) -> JournalSource {
        JournalSource::new(
            PartitionId::new(partition),
            PartitionIncarnation::from_bytes([3; 32]),
        )
    }

    fn durable_manifest(partition: &str) -> ProjectionManifest {
        let key = ProjectionKey::new(
            FamilyId::new("conversation-core/v1"),
            PartitionId::new(partition),
        );
        ProjectionManifest::new(
            key.clone(),
            ProjectionGeneration::new(4),
            SourceCheckpoint::try_new(
                durable_source(partition),
                JournalPosition::new(11),
                JournalPosition::new(12),
                13,
                JournalAttestation::new(
                    CommitRoot::from_bytes([9; 32]),
                    14,
                    vec![5; ATTESTATION_SIGNATURE_BYTES],
                    vec![6; ATTESTATION_SIGNER_BYTES],
                ),
            )
            .expect("checkpoint evidence is coherent"),
            2,
            3,
            ObjectDescriptor::new(
                key.object(),
                Generation::new(4),
                ContentDigest::from_bytes([8; 32]),
                OwnerId::new("projector"),
                Classification::Confidential,
                Retention::For(Duration::from_mins(1)),
                4096,
                ContentReference::try_new("content/key").expect("content key is valid"),
            ),
            ExactObjectRef::try_new(
                ObjectNamespace::try_new("ns").expect("namespace is valid"),
                ContentReference::try_new("content/key").expect("content key is valid"),
                21,
            )
            .expect("exact object is valid"),
            PublisherId::new("publisher-1"),
            PublisherFence::new(key, PartitionIncarnation::from_bytes([3; 32]), 5),
        )
    }

    fn durable_evidence() -> SourceSnapshot {
        SourceSnapshot::from_canonical(vec![
            SourcePin::Projected(ProjectionPin::new(durable_manifest("conv-a"))),
            SourcePin::Journal(JournalAnchor::new(
                durable_source("conv-b"),
                JournalPosition::new(42),
            )),
            SourcePin::Authoritative(Revision::new(77)),
        ])
        .expect("the fixture vector is canonical")
    }

    /// A durable source vector must survive the protocol round trip exactly.
    ///
    /// The audit digests these bytes. A dropped or defaulted field would keep
    /// the value structurally valid while no longer being what was signed.
    #[test]
    fn a_durable_source_vector_round_trips_without_omission() {
        let durable = durable_evidence();

        let projected = evidence_of(&durable).expect("the durable vector projects");
        let rebuilt = snapshot_of(&projected).expect("the projected vector rebuilds");

        assert_eq!(rebuilt, durable);
        assert_eq!(
            rebuilt.canonical_bytes(),
            durable.canonical_bytes(),
            "the audit digests these bytes, so they must be identical"
        );
    }

    #[test]
    fn both_vocabularies_share_one_canonical_order() {
        let durable = durable_evidence();
        let projected = evidence_of(&durable).expect("the durable vector projects");

        let durable_identities: Vec<Vec<u8>> = durable
            .pins()
            .iter()
            .map(|pin| match pin {
                SourcePin::Projected(projected) => {
                    let mut bytes = vec![0_u8];
                    push(&mut bytes, projected.manifest().key().family().as_str());
                    push(&mut bytes, projected.manifest().key().source().as_str());
                    bytes
                }
                SourcePin::Journal(anchor) => {
                    let mut bytes = vec![1_u8];
                    push(&mut bytes, anchor.partition().as_str());
                    bytes
                }
                SourcePin::Authoritative(_) => vec![2_u8],
            })
            .collect();
        let projected_identities: Vec<Vec<u8>> = projected
            .pins()
            .iter()
            .map(model::SourcePin::identity_bytes)
            .collect();

        assert_eq!(projected_identities, durable_identities);
    }

    fn push(buffer: &mut Vec<u8>, value: &str) {
        buffer.extend_from_slice(&u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
        buffer.extend_from_slice(value.as_bytes());
    }
}