polyc-state-connect 2026.8.3

//! Explicit delegation-witness burn wire mapping.
//!
//! One conversion per type, every field named. Nothing here spreads a default
//! over a struct: a field added on either side must be written down again in
//! both directions or this file stops compiling, which is the whole point.
//!
//! Every reference crosses as exactly [`REF_BYTES`] bytes and is decoded
//! through [`fixed_bytes`], so a short or overlong reference is refused rather
//! than padded into a different identity. That matters most for the issuer
//! reference: a witness admitted under a reference the wire silently reshaped
//! would be a witness admitted under a deployment secret nobody named.
//!
//! Absence is a wire variant rather than an empty message. A witness State
//! holds no record for arrives as [`pb::StateBurnUnrecorded`], so the answer
//! that has to fail closed cannot be produced by a decoder filling in a
//! default.

use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    burn::{
        BurnCommand, BurnOperation, IssuerRef, MAX_PENDING_PER_SUBJECT, PendingIndex,
        PlannedWitness, REF_BYTES, SubjectRef, WitnessRecord, WitnessRef, WitnessState,
        WitnessStatus, burn_scope,
    },
    command::{CommandEnvelope, CommandMetadata, ResourceBounds},
    digest::ContentDigest,
    error::StateError,
    id::{Audience, CommandId, NamespaceId, Purpose},
    revision::Revision,
    versioned::{EntryExpectation, MAX_MUTATIONS_PER_TRANSACTION, MAX_TRANSACTION_PAYLOAD_BYTES},
};

use crate::wire::{fixed_bytes, known, malformed, required};

/// Reads exactly one reference, refusing any other width.
fn reference_from_wire(field: &str, value: &[u8]) -> Result<[u8; REF_BYTES], StateError> {
    fixed_bytes::<REF_BYTES>(field, value)
}

pub(crate) fn witness_to_wire(value: WitnessRef) -> Vec<u8> {
    value.as_bytes().to_vec()
}

pub(crate) fn witness_from_wire(field: &str, value: &[u8]) -> Result<WitnessRef, StateError> {
    Ok(WitnessRef::from_bytes(reference_from_wire(field, value)?))
}

pub(crate) fn subject_to_wire(value: SubjectRef) -> Vec<u8> {
    value.as_bytes().to_vec()
}

pub(crate) fn subject_from_wire(field: &str, value: &[u8]) -> Result<SubjectRef, StateError> {
    Ok(SubjectRef::from_bytes(reference_from_wire(field, value)?))
}

fn issuer_to_wire(value: IssuerRef) -> Vec<u8> {
    value.as_bytes().to_vec()
}

fn issuer_from_wire(field: &str, value: &[u8]) -> Result<IssuerRef, StateError> {
    Ok(IssuerRef::from_bytes(reference_from_wire(field, value)?))
}

const fn status_to_wire(value: WitnessStatus) -> pb::StateBurnWitnessStatus {
    match value {
        WitnessStatus::Live => pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_LIVE,
        WitnessStatus::Burned => pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_BURNED,
        WitnessStatus::Redeemed => pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_REDEEMED,
    }
}

fn status_from_wire(
    field: &str,
    value: buffa::EnumValue<pb::StateBurnWitnessStatus>,
) -> Result<WitnessStatus, StateError> {
    match known(field, value)? {
        pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_LIVE => Ok(WitnessStatus::Live),
        pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_BURNED => Ok(WitnessStatus::Burned),
        pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_REDEEMED => {
            Ok(WitnessStatus::Redeemed)
        }
        pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_UNSPECIFIED => Err(malformed(
            field,
            "a witness record names one known lifecycle status",
        )),
    }
}

fn record_to_wire(value: &WitnessRecord) -> pb::StateBurnWitnessRecord {
    pb::StateBurnWitnessRecord {
        witness: witness_to_wire(value.witness()),
        subject: subject_to_wire(value.subject()),
        issuer: issuer_to_wire(value.issuer()),
        status: status_to_wire(value.status()).into(),
        recorded_at_ms: value.recorded_at_ms(),
        expires_at_ms: value.expires_at_ms(),
        settled_at_ms: value.settled_at_ms(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn record_from_wire(value: &pb::StateBurnWitnessRecord) -> Result<WitnessRecord, StateError> {
    Ok(WitnessRecord::from_parts(
        witness_from_wire("witness", &value.witness)?,
        subject_from_wire("subject", &value.subject)?,
        issuer_from_wire("issuer", &value.issuer)?,
        status_from_wire("status", value.status)?,
        value.recorded_at_ms,
        value.expires_at_ms,
        value.settled_at_ms,
    ))
}

pub(crate) fn witness_state_to_wire(value: &WitnessState) -> pb::StateBurnWitnessState {
    use pb::__buffa::oneof::state_burn_witness_state::State;
    let state = match value {
        WitnessState::Unrecorded => State::from(pb::StateBurnUnrecorded {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        WitnessState::Recorded(record) => State::from(record_to_wire(record)),
    };
    pb::StateBurnWitnessState {
        state: Some(state),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Reads what State holds for one witness.
///
/// A reply carrying no variant is refused rather than read as
/// [`WitnessState::Unrecorded`]. Both are a refusal at the caller, but only one
/// of them is a fact State actually stated, and a redemption path that cannot
/// tell them apart is the failure this family exists to remove.
pub(crate) fn witness_state_from_wire(
    value: pb::StateBurnWitnessState,
) -> Result<WitnessState, StateError> {
    use pb::__buffa::oneof::state_burn_witness_state::State;
    match value.state {
        Some(State::Unrecorded(_)) => Ok(WitnessState::Unrecorded),
        Some(State::Recorded(record)) => Ok(WitnessState::Recorded(record_from_wire(&record)?)),
        None => Err(malformed(
            "state",
            "a witness reply states either that State holds no record or which record it holds",
        )),
    }
}

pub(crate) fn pending_to_wire(value: &PendingIndex) -> pb::StateBurnPendingIndex {
    pb::StateBurnPendingIndex {
        witnesses: value
            .entries()
            .iter()
            .copied()
            .map(witness_to_wire)
            .collect(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn pending_from_wire(
    value: &pb::StateBurnPendingIndex,
) -> Result<PendingIndex, StateError> {
    if value.witnesses.len() > MAX_PENDING_PER_SUBJECT {
        return Err(malformed(
            "pending",
            "a pending index carries no more than the outstanding-witness ceiling",
        ));
    }
    Ok(PendingIndex::new(
        value
            .witnesses
            .iter()
            .map(|witness| witness_from_wire("pending", witness))
            .collect::<Result<Vec<_>, _>>()?,
    ))
}

fn expected_to_wire(value: EntryExpectation) -> pb::StateBurnExpectedEntry {
    use pb::__buffa::oneof::state_burn_expected_entry::Expected;
    let expected = match value {
        EntryExpectation::Absent => Expected::from(pb::StateBurnExpectedAbsent {
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        EntryExpectation::Revision(revision) => Expected::from(pb::StateBurnExpectedRevision {
            revision: revision.get(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::StateBurnExpectedEntry {
        expected: Some(expected),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn expected_from_wire(value: pb::StateBurnExpectedEntry) -> Result<EntryExpectation, StateError> {
    use pb::__buffa::oneof::state_burn_expected_entry::Expected;
    match value.expected {
        Some(Expected::Absent(_)) => Ok(EntryExpectation::Absent),
        Some(Expected::Revision(value)) => {
            Ok(EntryExpectation::Revision(Revision::new(value.revision)))
        }
        None => Err(malformed(
            "expected",
            "a burn operation declares its exact row premise",
        )),
    }
}

pub(crate) fn planned_to_wire(value: &PlannedWitness) -> pb::StateBurnPlannedWitness {
    pb::StateBurnPlannedWitness {
        record: buffa::MessageField::some(record_to_wire(value.record())),
        expected: buffa::MessageField::some(expected_to_wire(value.expected())),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn planned_from_wire(
    value: pb::StateBurnPlannedWitness,
) -> Result<PlannedWitness, StateError> {
    Ok(PlannedWitness::new(
        record_from_wire(&required(
            "record",
            "a planned burn carries the record it settles",
            value.record,
        )?)?,
        expected_from_wire(required(
            "expected",
            "a planned burn carries the row premise it was planned against",
            value.expected,
        )?)?,
    ))
}

/// Reads a bounded set of planned burns.
///
/// Bounded on the way in as well as on the way out: a subject's whole
/// outstanding set settles in one transaction, so a peer that names more than
/// the ceiling is naming a transaction that could not commit anyway.
pub(crate) fn planned_set_from_wire(
    field: &str,
    value: Vec<pb::StateBurnPlannedWitness>,
) -> Result<Vec<PlannedWitness>, StateError> {
    if value.len() > MAX_PENDING_PER_SUBJECT {
        return Err(malformed(
            field,
            "a subject burn covers no more than the outstanding-witness ceiling",
        ));
    }
    value.into_iter().map(planned_from_wire).collect()
}

pub(crate) fn operation_to_wire(value: &BurnOperation) -> pb::StateBurnOperation {
    use pb::__buffa::oneof::state_burn_operation::Operation;
    let operation = match value {
        BurnOperation::Record {
            record,
            record_expected,
            pending,
            pending_expected,
        } => Operation::from(pb::StateBurnRecordWitness {
            record: buffa::MessageField::some(record_to_wire(record)),
            record_expected: buffa::MessageField::some(expected_to_wire(*record_expected)),
            pending: buffa::MessageField::some(pending_to_wire(pending)),
            pending_expected: buffa::MessageField::some(expected_to_wire(*pending_expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        BurnOperation::Burn {
            now_ms,
            record,
            record_expected,
            pending,
            pending_expected,
        } => Operation::from(pb::StateBurnBurnWitness {
            now_ms: *now_ms,
            record: buffa::MessageField::some(record_to_wire(record)),
            record_expected: buffa::MessageField::some(expected_to_wire(*record_expected)),
            pending: buffa::MessageField::some(pending_to_wire(pending)),
            pending_expected: buffa::MessageField::some(expected_to_wire(*pending_expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        BurnOperation::Redeem {
            now_ms,
            issuer,
            record,
            record_expected,
            pending,
            pending_expected,
        } => Operation::from(pb::StateBurnRedeemWitness {
            now_ms: *now_ms,
            issuer: issuer_to_wire(*issuer),
            record: buffa::MessageField::some(record_to_wire(record)),
            record_expected: buffa::MessageField::some(expected_to_wire(*record_expected)),
            pending: buffa::MessageField::some(pending_to_wire(pending)),
            pending_expected: buffa::MessageField::some(expected_to_wire(*pending_expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
        BurnOperation::BurnSubject {
            now_ms,
            subject,
            burns,
            pending_expected,
        } => Operation::from(pb::StateBurnBurnSubject {
            now_ms: *now_ms,
            subject: subject_to_wire(*subject),
            burns: burns.iter().map(planned_to_wire).collect(),
            pending_expected: buffa::MessageField::some(expected_to_wire(*pending_expected)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }),
    };
    pb::StateBurnOperation {
        operation: Some(operation),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

fn operation_from_wire(value: pb::StateBurnOperation) -> Result<BurnOperation, StateError> {
    use pb::__buffa::oneof::state_burn_operation::Operation;
    let record = |value| {
        record_from_wire(&required(
            "record",
            "a burn operation carries its result record",
            value,
        )?)
    };
    let pending = |value| {
        pending_from_wire(&required(
            "pending",
            "a burn operation carries the exact resulting index",
            value,
        )?)
    };
    let expected = |field: &'static str, value| {
        expected_from_wire(required(
            field,
            "a burn operation carries its premise",
            value,
        )?)
    };
    match value.operation {
        Some(Operation::Record(value)) => Ok(BurnOperation::Record {
            record: record(value.record)?,
            record_expected: expected("record_expected", value.record_expected)?,
            pending: pending(value.pending)?,
            pending_expected: expected("pending_expected", value.pending_expected)?,
        }),
        Some(Operation::Burn(value)) => Ok(BurnOperation::Burn {
            now_ms: value.now_ms,
            record: record(value.record)?,
            record_expected: expected("record_expected", value.record_expected)?,
            pending: pending(value.pending)?,
            pending_expected: expected("pending_expected", value.pending_expected)?,
        }),
        Some(Operation::Redeem(value)) => Ok(BurnOperation::Redeem {
            now_ms: value.now_ms,
            issuer: issuer_from_wire("issuer", &value.issuer)?,
            record: record(value.record)?,
            record_expected: expected("record_expected", value.record_expected)?,
            pending: pending(value.pending)?,
            pending_expected: expected("pending_expected", value.pending_expected)?,
        }),
        Some(Operation::BurnSubject(value)) => Ok(BurnOperation::BurnSubject {
            now_ms: value.now_ms,
            subject: subject_from_wire("subject", &value.subject)?,
            burns: planned_set_from_wire("burns", value.burns)?,
            pending_expected: expected("pending_expected", value.pending_expected)?,
        }),
        None => Err(malformed("operation", "a burn command names one operation")),
    }
}

pub(crate) fn metadata_to_wire(command: &BurnCommand) -> pb::StateBurnCommandMetadata {
    let value = command.metadata();
    pb::StateBurnCommandMetadata {
        command_id: value.command_id().as_str().to_owned(),
        namespace: value.scope().namespace().as_str().to_owned(),
        purpose: value.envelope().purpose().as_str().to_owned(),
        command_audience: value.envelope().audience().as_str().to_owned(),
        digest: value.digest().as_bytes().to_vec(),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

pub(crate) fn command_from_wire(
    metadata: pb::StateBurnCommandMetadata,
    operation: pb::StateBurnOperation,
) -> Result<BurnCommand, StateError> {
    let namespace = NamespaceId::new(metadata.namespace);
    let digest = ContentDigest::from_bytes(fixed_bytes::<{ ContentDigest::LEN }>(
        "digest",
        &metadata.digest,
    )?);
    Ok(BurnCommand::new(
        CommandMetadata::new(
            CommandId::new(metadata.command_id),
            polyc_state::burn::family(),
            digest,
            burn_scope(&namespace),
            CommandEnvelope::new(
                Purpose::new(metadata.purpose),
                Audience::new(metadata.command_audience),
                ResourceBounds::new(MAX_TRANSACTION_PAYLOAD_BYTES, MAX_MUTATIONS_PER_TRANSACTION),
            ),
        ),
        operation_from_wire(operation)?,
    ))
}

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

    const NOW: u64 = 1_700_000_000_000;
    const TTL: u64 = 5 * 60 * 1_000;

    fn witness(seed: u8) -> WitnessRef {
        WitnessRef::from_bytes([seed; REF_BYTES])
    }

    fn record() -> WitnessRecord {
        WitnessRecord::live(
            witness(1),
            SubjectRef::for_persona("persona-1"),
            IssuerRef::for_secret_reference("deployment/witness-secret"),
            NOW,
            NOW + TTL,
        )
    }

    fn pending() -> PendingIndex {
        PendingIndex::default().with(witness(1)).with(witness(2))
    }

    #[test]
    fn every_operation_round_trips_through_the_wire() {
        let live = record();
        let operations = [
            BurnOperation::Record {
                record: live,
                record_expected: EntryExpectation::Absent,
                pending: pending(),
                pending_expected: EntryExpectation::Absent,
            },
            BurnOperation::Burn {
                now_ms: NOW + 1,
                record: live.burned(NOW + 1),
                record_expected: EntryExpectation::Revision(Revision::new(9)),
                pending: PendingIndex::default(),
                pending_expected: EntryExpectation::Revision(Revision::new(11)),
            },
            BurnOperation::Redeem {
                now_ms: NOW + 2,
                issuer: live.issuer(),
                record: live.redeemed(NOW + 2),
                record_expected: EntryExpectation::Revision(Revision::new(13)),
                pending: PendingIndex::default(),
                pending_expected: EntryExpectation::Revision(Revision::new(15)),
            },
            BurnOperation::BurnSubject {
                now_ms: NOW + 3,
                subject: live.subject(),
                burns: vec![PlannedWitness::new(
                    live.burned(NOW + 3),
                    EntryExpectation::Revision(Revision::new(17)),
                )],
                pending_expected: EntryExpectation::Revision(Revision::new(19)),
            },
        ];
        for operation in operations {
            let restored = operation_from_wire(operation_to_wire(&operation))
                .expect("an operation survives its own encoding");
            assert_eq!(restored, operation);
        }
    }

    /// Both halves of the answer the fail-closed path reads survive the hop,
    /// and they stay two answers rather than collapsing into one.
    #[test]
    fn a_witness_state_round_trips_and_keeps_unrecorded_distinct() {
        for state in [
            WitnessState::Unrecorded,
            WitnessState::Recorded(record()),
            WitnessState::Recorded(record().burned(NOW + 1)),
            WitnessState::Recorded(record().redeemed(NOW + 1)),
        ] {
            assert_eq!(
                witness_state_from_wire(witness_state_to_wire(&state))
                    .expect("a witness state survives its own encoding"),
                state
            );
        }
        assert_ne!(
            witness_state_to_wire(&WitnessState::Unrecorded),
            witness_state_to_wire(&WitnessState::Recorded(record()))
        );
    }

    #[test]
    fn the_pending_index_round_trips_through_the_wire() {
        assert_eq!(
            pending_from_wire(&pending_to_wire(&pending())).expect("pending index round trip"),
            pending()
        );
    }

    /// A reference either decodes to the exact identity it named or is refused.
    /// A truncated issuer reference read as a shorter one is a witness admitted
    /// under a deployment secret nobody named.
    #[test]
    fn a_misshapen_reference_is_refused_rather_than_reshaped() {
        let mut wire = record_to_wire(&record());
        wire.issuer.truncate(REF_BYTES - 1);
        assert!(
            record_from_wire(&wire).is_err(),
            "a truncated issuer reference was accepted"
        );
        let mut wire = record_to_wire(&record());
        wire.witness.push(0);
        assert!(
            record_from_wire(&wire).is_err(),
            "an overlong witness reference was accepted"
        );
        let mut wire = record_to_wire(&record());
        wire.subject.clear();
        assert!(
            record_from_wire(&wire).is_err(),
            "an absent subject reference was accepted"
        );
    }

    #[test]
    fn an_unset_state_a_missing_oneof_and_a_missing_premise_fail_closed() {
        assert!(
            witness_state_from_wire(pb::StateBurnWitnessState::default()).is_err(),
            "a witness reply with no variant was read as an answer"
        );
        assert!(
            operation_from_wire(pb::StateBurnOperation::default()).is_err(),
            "an operation with no variant was accepted"
        );
        assert!(
            expected_from_wire(pb::StateBurnExpectedEntry::default()).is_err(),
            "a premise with no variant was accepted"
        );
        let mut wire = record_to_wire(&record());
        wire.status = pb::StateBurnWitnessStatus::STATE_BURN_WITNESS_STATUS_UNSPECIFIED.into();
        assert!(
            record_from_wire(&wire).is_err(),
            "a record naming no lifecycle status was accepted"
        );
        let burn = pb::StateBurnBurnWitness {
            now_ms: NOW,
            record: buffa::MessageField::some(record_to_wire(&record().burned(NOW))),
            record_expected: buffa::MessageField::none(),
            pending: buffa::MessageField::some(pending_to_wire(&PendingIndex::default())),
            pending_expected: buffa::MessageField::some(expected_to_wire(EntryExpectation::Absent)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        assert!(
            operation_from_wire(pb::StateBurnOperation {
                operation: Some(pb::__buffa::oneof::state_burn_operation::Operation::from(
                    burn
                )),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
            .is_err(),
            "a burn with no record premise was accepted"
        );
    }

    /// The transaction ceiling is the reason the outstanding set is bounded, so
    /// a peer naming more than it is refused before a command is ever built.
    #[test]
    fn an_oversized_pending_set_is_refused_at_the_boundary() {
        let mut wire = pending_to_wire(&PendingIndex::default());
        wire.witnesses = (0..=u8::try_from(MAX_PENDING_PER_SUBJECT)
            .expect("ceiling fits one byte"))
            .map(|seed| witness_to_wire(witness(seed)))
            .collect();
        assert!(
            pending_from_wire(&wire).is_err(),
            "a pending index past the ceiling was accepted"
        );
        let burns: Vec<pb::StateBurnPlannedWitness> = (0..=MAX_PENDING_PER_SUBJECT)
            .map(|_| planned_to_wire(&PlannedWitness::new(record(), EntryExpectation::Absent)))
            .collect();
        assert!(
            planned_set_from_wire("burns", burns).is_err(),
            "a subject burn past the ceiling was accepted"
        );
    }
}