polyc-facts 2026.10.2

//! Committed conversation-delegation history facts.
//!
//! This fold is the semantic boundary shared by Projector and the end-to-end
//! parity proof that holds the projected delegation shapes.
//!
//! # Every decodable record, no turn-commit gate
//!
//! Production commits a handoff and a handoff denial as their own signed
//! commands, so a record can be durable while the parent turn never completes.
//! The embedded table this replaces states that and applies no committed-turn
//! filter. This fold keeps that rule. Gating on committed turns here would
//! delete exactly the denial evidence the surface exists to show.
//!
//! # A malformed payload refuses the generation
//!
//! The embedded decoder skipped a malformed payload and logged a warning, so a
//! damaged record read as an absent one. A projected generation is an
//! authority a reader trusts, so this fold refuses instead. That is the posture
//! `conversation-execution/v1` already takes.
//!
//! # The fact keeps domain values
//!
//! `polyc-projection` owns the family's logical schema and the projector owns
//! Arrow and Parquet, so a list stays a list and a verdict stays a verdict
//! here. The artifact encoder spells both. `phase` already worked that way.
//!
//! # Tampered and untrusted records stay
//!
//! Only a payload that does not decode is refused. A record whose signature
//! fails, or whose key the deployment does not trust, keeps every field and
//! carries its verdict.

use std::collections::HashSet;

use polyc_crypto::signing_role::{HandoffRole, RoleTrustSet, SignatureVerdict, SigningKeyIdentity};
use polyc_eventlog_model::Event;
use polyc_proto::kinds;

use crate::{HandoffFact, PreparedSource, fold_handoff_event, prepare_conversation_core};

/// All delegation-history rows folded from one prepared source prefix.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ConversationDelegationFacts {
    /// One row per handoff or handoff-denial record.
    pub handoffs: Vec<HandoffHistoryFact>,
    /// Fleet-only signer evidence. One row per handoff record whose embedded
    /// key is a usable ed25519 public key, so this is shorter than
    /// [`Self::handoffs`] exactly where a record reads `invalid` for that
    /// reason.
    pub signers: Vec<HandoffSignerFact>,
}

/// Which command one delegation row records.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HandoffPhase {
    /// The parent asked to transfer to a child agent.
    Handoff,
    /// The transfer was refused.
    Denied,
}

impl HandoffPhase {
    /// Every phase, in a fixed order.
    ///
    /// A reader that restates this vocabulary iterates this instead of
    /// copying the spellings. `every_phase_appears_in_all` makes a new
    /// variant a compile error, so the list cannot fall behind the type.
    pub const ALL: [Self; 2] = [Self::Handoff, Self::Denied];

    /// Stable lowercase name, matching the column the caller reads today.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Handoff => "handoff",
            Self::Denied => "handoff_denied",
        }
    }
}

/// One delegation row, discriminated by [`Self::phase`].
///
/// A field the other phase does not carry holds its type's empty value. No
/// column of this family is nullable.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HandoffHistoryFact {
    /// Journal position.
    pub position: u64,
    /// The parent turn the record was tagged with, empty when it carried none.
    pub turn_id: String,
    /// Which command this row records.
    pub phase: HandoffPhase,
    /// `handoff` only: the child conversation the transfer named.
    pub child_conversation_id: String,
    /// The child agent the transfer named.
    pub child_agent_id: String,
    /// `handoff` only: how many parent messages were carried.
    pub carried_count: u64,
    /// The freeform reason given for the requested transfer.
    pub reason: String,
    /// `handoff_denied` only: the parent's bound agent at request time.
    pub parent_agent_id: String,
    /// `handoff_denied` only: why the transfer was refused.
    pub denial_reason: String,
    /// `handoff_denied` only: the parent's declared allowlist, in order.
    pub allowed: Vec<String>,
    /// How the signature reads against this deployment's handoff trust set.
    pub signature_status: SignatureVerdict,
}

/// The signer of one delegation record. Fleet-only.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HandoffSignerFact {
    /// Journal position of the record this key signed.
    pub position: u64,
    /// The embedded ed25519 public key, exactly 32 bytes.
    pub signed_by: [u8; 32],
    /// This deployment's trust-role identity for the key.
    pub signer_key_id: String,
}

/// Why a durable delegation fact could not be folded.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConversationDelegationError {
    /// Two source events claim one journal position.
    #[error("position {position} appears more than once in one source prefix")]
    RepeatedPosition {
        /// The repeated journal position.
        position: u64,
    },
    /// A delegation payload did not decode under its declared kind.
    #[error("the {kind} record at position {position} is malformed")]
    Malformed {
        /// Stable event kind, never payload content.
        kind: &'static str,
        /// Journal position of the malformed record.
        position: u64,
    },
}

/// Folds one prepared source prefix into its delegation history.
///
/// Applies no committed-turn filter: see the module docs.
///
/// # Errors
///
/// Returns [`ConversationDelegationError`] for a repeated source position or a
/// malformed delegation payload. An unusable embedded key is NOT an error: the
/// record publishes with the `invalid` verdict and emits no signer row.
pub fn fold_conversation_delegation(
    events: &[(u64, Event)],
    handoff_trust: &RoleTrustSet<HandoffRole>,
) -> Result<ConversationDelegationFacts, ConversationDelegationError> {
    refuse_repeated_positions(events)?;
    let mut facts = ConversationDelegationFacts::default();
    for (position, event) in events {
        let (base, turn) = kinds::parse(&event.kind);
        if base != kinds::HANDOFF && base != kinds::HANDOFF_DENIED {
            continue;
        }
        let kind = if base == kinds::HANDOFF {
            kinds::HANDOFF
        } else {
            kinds::HANDOFF_DENIED
        };
        let fact = fold_handoff_event(base, &event.payload, handoff_trust).ok_or(
            ConversationDelegationError::Malformed {
                kind,
                position: *position,
            },
        )?;
        let turn_id = turn.map_or_else(String::new, |id| id.to_string());
        let (row, signed_by) = match fact {
            HandoffFact::Handoff(h) => (
                HandoffHistoryFact {
                    position: *position,
                    turn_id,
                    phase: HandoffPhase::Handoff,
                    child_conversation_id: h.child_conversation_id,
                    child_agent_id: h.child_agent_id,
                    carried_count: u64::from(h.carried_count),
                    reason: h.reason,
                    parent_agent_id: String::new(),
                    denial_reason: String::new(),
                    allowed: Vec::new(),
                    signature_status: h.signature_status,
                },
                h.signed_by,
            ),
            HandoffFact::Denied(d) => (
                HandoffHistoryFact {
                    position: *position,
                    turn_id,
                    phase: HandoffPhase::Denied,
                    child_conversation_id: String::new(),
                    child_agent_id: d.child_agent_id,
                    carried_count: 0,
                    reason: d.reason,
                    parent_agent_id: d.parent_agent_id,
                    denial_reason: d.denial_reason,
                    allowed: d.allowed,
                    signature_status: d.signature_status,
                },
                d.signed_by,
            ),
        };
        facts.signers.extend(signer(*position, signed_by));
        facts.handoffs.push(row);
    }
    Ok(facts)
}

/// Prepares one source prefix, then folds its delegation history.
///
/// Preparation is the shared one `conversation-core/v1` defines, so a verified
/// excision strips its record before this fold reads the prefix.
///
/// # Errors
///
/// Returns [`ConversationDelegationError`] for every reason
/// [`fold_conversation_delegation`] does.
pub fn prepare_and_fold_conversation_delegation(
    events: &mut [(u64, Event)],
    partition: &str,
    handoff_trust: &RoleTrustSet<HandoffRole>,
) -> Result<(PreparedSource, ConversationDelegationFacts), ConversationDelegationError> {
    let prepared = prepare_conversation_core(events, partition);
    let facts = fold_conversation_delegation(events, handoff_trust)?;
    Ok((prepared, facts))
}

/// The signer row for one record, or `None` where the embedded key is not a
/// usable ed25519 public key.
///
/// An unusable key is a verdict, not a refusal. `classify_handoff` cannot
/// verify a signature against such a key, so the record already reads
/// `invalid`, and the Fleet-only signer table simply has no key to publish
/// for it. Refusing here instead would refuse the whole generation, which
/// takes the conversation's entire delegation history dark over one record
/// the authority already accepted and this fold already classified.
fn signer(position: u64, signed_by: Vec<u8>) -> Option<HandoffSignerFact> {
    let identity = SigningKeyIdentity::for_public_key::<HandoffRole>(signed_by).ok()?;
    let signed_by: [u8; 32] = identity.public_key().try_into().ok()?;
    Some(HandoffSignerFact {
        position,
        signed_by,
        signer_key_id: identity.key_id().to_owned(),
    })
}

fn refuse_repeated_positions(events: &[(u64, Event)]) -> Result<(), ConversationDelegationError> {
    let mut positions = HashSet::with_capacity(events.len());
    for (position, _) in events {
        if !positions.insert(*position) {
            return Err(ConversationDelegationError::RepeatedPosition {
                position: *position,
            });
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {

    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use buffa::Message as _;
    use polyc_crypto::signing_role::HandoffSigner;
    use polyc_proto::proto::polychrome::handoff::v1::{Handoff, HandoffDenied};
    use uuid::Uuid;

    use super::*;

    /// `ALL` must list every phase.
    ///
    /// The match has no wildcard, so a new variant fails to compile here.
    /// The length assertion then forces the new variant into `ALL`.
    #[test]
    fn every_phase_appears_in_all() {
        for phase in HandoffPhase::ALL {
            match phase {
                HandoffPhase::Handoff | HandoffPhase::Denied => {}
            }
        }
        assert_eq!(HandoffPhase::ALL.len(), 2);
    }

    fn trust(signer: &HandoffSigner) -> RoleTrustSet<HandoffRole> {
        RoleTrustSet::current(signer)
    }

    fn handoff_event(signer: &HandoffSigner, turn: Option<&Uuid>) -> Event {
        let mut h = Handoff {
            child_conversation_id: "child-a".to_owned(),
            child_agent_id: "researcher".to_owned(),
            carried_count: 2,
            reason: "delegate research".to_owned(),
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_into(signer, &mut h);
        let kind = turn.map_or_else(
            || kinds::HANDOFF.to_owned(),
            |turn| kinds::tagged(kinds::HANDOFF, turn),
        );
        Event::new(kind, h.encode_to_vec())
    }

    fn denied_event(signer: &HandoffSigner, turn: &Uuid) -> Event {
        let mut d = HandoffDenied {
            parent_conversation_id: "parent-7".to_owned(),
            parent_agent_id: "assistant".to_owned(),
            child_agent_id: "banned-agent".to_owned(),
            reason: "delegate weird task".to_owned(),
            denial_reason: "this agent cannot hand off to that agent".to_owned(),
            allowed: vec!["coding".to_owned(), "research".to_owned()],
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_denied_into(signer, &mut d);
        Event::new(
            kinds::tagged(kinds::HANDOFF_DENIED, turn),
            d.encode_to_vec(),
        )
    }

    /// The trap this family exists to avoid. Production commits a handoff as
    /// its own signed command, so gating on a completed parent turn would
    /// delete exactly the denial evidence the surface shows.
    #[test]
    fn a_record_whose_turn_never_completed_still_folds() {
        let signer = HandoffSigner::from_seed(31);
        let turn = Uuid::from_u128(7);
        // No turn_dispatched and no turn_complete anywhere in the prefix.
        let events = vec![
            (9, handoff_event(&signer, Some(&turn))),
            (10, denied_event(&signer, &turn)),
        ];

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

        assert_eq!(
            facts.handoffs.len(),
            2,
            "a committed-turn gate would empty this"
        );
        assert_eq!(facts.handoffs[0].phase, HandoffPhase::Handoff);
        assert_eq!(facts.handoffs[1].phase, HandoffPhase::Denied);
    }

    /// The embedded decoder skipped and warned, so damage read as absence.
    #[test]
    fn a_malformed_payload_refuses_instead_of_vanishing() {
        let signer = HandoffSigner::from_seed(32);
        let events = vec![(
            4,
            Event::new(kinds::HANDOFF.to_owned(), vec![0xff, 0xff, 0xff]),
        )];

        assert_eq!(
            fold_conversation_delegation(&events, &trust(&signer)),
            Err(ConversationDelegationError::Malformed {
                kind: kinds::HANDOFF,
                position: 4,
            })
        );
    }

    #[test]
    fn a_tampered_record_keeps_every_field_and_reads_invalid() {
        let signer = HandoffSigner::from_seed(33);
        let mut h = Handoff {
            child_conversation_id: "child-a".to_owned(),
            child_agent_id: "researcher".to_owned(),
            carried_count: 2,
            reason: "delegate research".to_owned(),
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_into(&signer, &mut h);
        h.child_conversation_id = "child-evil".to_owned();
        let events = vec![(1, Event::new(kinds::HANDOFF.to_owned(), h.encode_to_vec()))];

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

        assert_eq!(facts.handoffs.len(), 1);
        assert_eq!(facts.handoffs[0].child_conversation_id, "child-evil");
        assert_eq!(
            facts.handoffs[0].signature_status,
            SignatureVerdict::Invalid
        );
    }

    #[test]
    fn a_key_this_deployment_does_not_trust_reads_untrusted() {
        let signer = HandoffSigner::from_seed(34);
        let stranger = HandoffSigner::from_seed(99);
        let events = vec![(1, handoff_event(&signer, None))];

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

        assert_eq!(
            facts.handoffs[0].signature_status,
            SignatureVerdict::Untrusted
        );
    }

    #[test]
    fn a_denial_carries_its_allowlist_and_a_handoff_carries_none() {
        let signer = HandoffSigner::from_seed(35);
        let turn = Uuid::from_u128(3);
        let events = vec![
            (1, handoff_event(&signer, Some(&turn))),
            (2, denied_event(&signer, &turn)),
        ];

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

        assert!(facts.handoffs[0].allowed.is_empty());
        assert_eq!(
            facts.handoffs[1].allowed,
            vec!["coding".to_owned(), "research".to_owned()]
        );
        assert_eq!(facts.handoffs[0].carried_count, 2);
        assert_eq!(facts.handoffs[1].carried_count, 0);
        assert_eq!(facts.handoffs[0].turn_id, turn.to_string());
    }

    #[test]
    fn an_untagged_record_carries_an_empty_turn_id() {
        let signer = HandoffSigner::from_seed(36);
        let events = vec![(1, handoff_event(&signer, None))];

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

        assert_eq!(facts.handoffs[0].turn_id, "");
    }

    #[test]
    fn every_record_contributes_one_signer_row_at_its_own_position() {
        let signer = HandoffSigner::from_seed(37);
        let turn = Uuid::from_u128(5);
        let events = vec![
            (11, handoff_event(&signer, Some(&turn))),
            (12, denied_event(&signer, &turn)),
        ];

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

        assert_eq!(facts.signers.len(), 2);
        assert_eq!(facts.signers[0].position, 11);
        assert_eq!(facts.signers[1].position, 12);
        assert_eq!(
            facts.signers[0].signed_by.to_vec(),
            signer.public_key_bytes()
        );
        assert!(!facts.signers[0].signer_key_id.is_empty());
        assert_eq!(
            facts.signers[0].signer_key_id,
            facts.signers[1].signer_key_id
        );
    }

    /// An unusable embedded key publishes the record as `invalid` and emits
    /// no signer row. It does not refuse.
    ///
    /// The record is authority-accepted and already in the journal, and a
    /// fold refusal is a generation refusal, so refusing here would take the
    /// conversation's whole delegation history dark and keep it dark. The
    /// verdict already carries the fact: a signature cannot verify against a
    /// key this malformed.
    #[test]
    fn a_record_carrying_no_usable_key_reads_invalid_and_publishes_no_signer() {
        let signer = HandoffSigner::from_seed(38);
        let mut h = Handoff {
            child_conversation_id: "child-a".to_owned(),
            ..Default::default()
        };
        polyc_crypto::handoff::sign_handoff_into(&signer, &mut h);
        h.signed_by = vec![0u8; 31];
        let events = vec![(6, Event::new(kinds::HANDOFF.to_owned(), h.encode_to_vec()))];

        let facts = fold_conversation_delegation(&events, &trust(&signer))
            .expect("an unusable key must not refuse the generation");
        assert_eq!(facts.handoffs.len(), 1, "the record still publishes");
        assert_eq!(facts.handoffs[0].position, 6);
        assert_eq!(facts.handoffs[0].child_conversation_id, "child-a");
        assert_eq!(
            facts.handoffs[0].signature_status,
            SignatureVerdict::Invalid,
            "a signature cannot verify against an unusable key"
        );
        assert!(
            facts.signers.is_empty(),
            "there is no usable key to publish in the Fleet-only signer table"
        );
    }

    #[test]
    fn a_repeated_source_position_is_refused() {
        let signer = HandoffSigner::from_seed(39);
        let events = vec![
            (2, handoff_event(&signer, None)),
            (2, handoff_event(&signer, None)),
        ];

        assert_eq!(
            fold_conversation_delegation(&events, &trust(&signer)),
            Err(ConversationDelegationError::RepeatedPosition { position: 2 })
        );
    }

    #[test]
    fn an_unrelated_kind_contributes_no_row() {
        let signer = HandoffSigner::from_seed(40);
        let events = vec![(1, Event::new(kinds::USAGE.to_owned(), vec![]))];

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

        assert!(facts.handoffs.is_empty());
        assert!(facts.signers.is_empty());
    }
}