polyc-facts 2026.9.0

//! The attribution fold: attribution was folded independently nine times
//! across five files before this crate existed, each call site running its
//! own small loop over events instead of calling one shared "attribution for
//! this turn" accessor.
//!
//! Unlike the message-content and receipt folds, the eleven call sites this
//! module replaces do NOT share one policy: some scope to committed turns
//! (and disagree on what "committed" means — full `turn_start`+
//! `turn_complete`, `turn_complete`-only, or no scoping at all), some keep
//! the first match, others the last or every match, and whether an empty
//! `persona_id` counts differs site to site. Baking any of that into this
//! crate would silently change behavior the first time two sites' policies
//! were compared side by side. So this module draws a narrower line than the
//! other two folds: [`attribution_events`] is the ONE decode primitive —
//! kind filtering, turn-id recovery, and the raw decoded fields — and every
//! policy choice downstream of it (commit scoping, tie-break, empty-id
//! filtering, alias canonicalization) stays with the caller, exactly as it
//! ran before. [`caller_by_turn_last_wins`] is the one exception: a small
//! helper built on the primitive, added only because three call sites
//! (`persona_committed_token_usage`, `accumulate_conversation_usage`,
//! `compute_stats`) independently implement the identical "last caller wins
//! per turn" fold.
//! Unifying the other eight sites' genuinely different policies is future
//! work, not this change.
//!
//! # `crates/query/src/decode/attribution.rs` (#1579)
//!
//! The query engine's own `attribution` typed-table decoder used to bypass
//! this module entirely and re-derive the identical kind-filter/decode loop
//! by hand — not because its POLICY differs (it applies none: no
//! committed-turn scoping, no tie-break, every fact surfaces as its own
//! row), but because it derives `position` from the journal's own append
//! position on each `(u64, Event)` pair it is handed, while
//! [`attribution_events`] derives `position` from the given slice's own
//! index. [`attribution_events_with_positions`] is the same decode
//! parametrized over an explicit position source, so both call shapes route
//! through one filter/decode body; [`attribution_events`] is now defined in
//! terms of it. `role` (the payload's own free-text field, independent of
//! which kind it was recorded under) is carried on [`AttributionFact`] for
//! this reason — the nine pre-existing call sites never needed it (only
//! [`AttributionEventKind`]), but the query table exposes it verbatim as its
//! own `role` column.
//!
//! `persona_id` here is always the RAW value the event recorded — this crate
//! is sync and persona-store-free by design (see the crate docs), so alias
//! canonicalization (resolving a pre-merge id to its survivor) stays in the
//! async callers that already do it (`polyc_persona::PersonaHost`).

use std::collections::HashMap;

use polyc_eventlog_model::Event;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::AttributionEvent;
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;

/// Which attribution kind a decoded [`AttributionFact`] carries.
///
/// The kind BASE the event was stored under ([`polyc_proto::kinds::CALLER`] /
/// [`polyc_proto::kinds::PARTICIPANT`]), never the payload's own free-text
/// `role` field (which the write side, `AttributionKind` in
/// `control-plane::grpc::attribution`, separately maps to the same two
/// values — this is the read-side twin, decode-only, with no write-side
/// role-string mapping of its own).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AttributionEventKind {
    /// A `caller` event — the persona whose message triggered the turn.
    Caller,
    /// A `participant` event — a persona observed speaking in the turn's
    /// input.
    Participant,
}

/// Which attribution kind(s) [`attribution_events`] decodes.
///
/// A selector, not a policy: it only narrows which kind BASE is decoded.
/// Every other choice (committed-turn scoping, tie-break, empty-id
/// filtering) is the caller's, applied to the returned facts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AttributionScope {
    /// Only `caller` events.
    CallerOnly,
    /// Only `participant` events.
    ParticipantOnly,
    /// Both `caller` and `participant` events.
    Both,
}

impl AttributionScope {
    /// Whether this scope wants a fact of the given kind.
    const fn wants(self, kind: AttributionEventKind) -> bool {
        matches!(
            (self, kind),
            (Self::CallerOnly, AttributionEventKind::Caller)
                | (Self::ParticipantOnly, AttributionEventKind::Participant)
                | (Self::Both, _)
        )
    }
}

/// One decoded `caller`/`participant` fact.
///
/// `persona_id` is RAW — exactly the string the event recorded, including an
/// empty one. Whether an empty `persona_id` is meaningful (some call sites
/// filter it, some don't) is caller policy, not this fold's; see the module
/// docs.
///
/// No `Eq` derive: [`ExternalIdentity`] (a generated protobuf type) does not
/// implement it.
#[derive(Debug, Clone, PartialEq)]
pub struct AttributionFact {
    /// Journal position of the event this fact came from — the `events`
    /// slice index under [`attribution_events`], or the caller-supplied
    /// position under [`attribution_events_with_positions`].
    pub position: u64,
    /// Whether this was a `caller` or `participant` event.
    pub kind: AttributionEventKind,
    /// The turn this attribution belongs to, recovered from the event kind's
    /// `base:{turn_uuid}` suffix — the only place a `caller`/`participant`
    /// event carries a turn id; the payload itself has none. `None` for a
    /// bare (unsuffixed) kind, which does not occur for these two kinds in
    /// practice but is represented rather than assumed away.
    pub turn_id: Option<uuid::Uuid>,
    /// The durable principal (persona) the external identity resolved to, as
    /// recorded on the event — RAW, not filtered or canonicalized.
    pub persona_id: String,
    /// The payload's own free-text `role` field — RAW, independent of
    /// whether this fact's [`kind`](Self::kind) is `Caller` or
    /// `Participant`. Not used by any of the nine call sites this fold
    /// replaces (they only read `kind`); carried so
    /// `crates/query/src/decode/attribution.rs`'s `role` column (which
    /// exposes this exact wire field) can read through this fold too
    /// (#1579).
    pub role: String,
    /// The external identity tuple as observed at the edge, when the event
    /// carried one.
    pub identity: Option<ExternalIdentity>,
    /// Which edge vouched for this attribution via a signed
    /// `AssertedAttribution` envelope, as recorded on the event; empty for
    /// internal/unsigned attribution.
    pub asserting_edge_id: String,
    /// The asserting edge's registered ed25519 public key that verified the
    /// envelope, hex-encoded, as recorded on the event; empty when
    /// `asserting_edge_id` is empty.
    pub signer_pk_hex: String,
    /// The envelope's own `AssertedAttribution.signature_hex`, as recorded
    /// on the event for forensics; empty when `asserting_edge_id` is empty.
    pub signature_hex: String,
}

/// Decode every `caller`/`participant` fact in `events` matching `scope`, in
/// journal order.
///
/// A convenience wrapper over [`attribution_events_with_positions`] that
/// derives each fact's `position` from the fact's own index in `events`
/// (sound for a full, un-pruned replay from position 0, which is every
/// existing caller's shape).
///
/// This is the one canonical decode path for `AttributionEvent`: kind-base
/// filtering (via [`polyc_proto::kinds::parse`], the same `base:{turn_uuid}`
/// grammar every other kind uses), the capped, canonical
/// [`polyc_proto::events_decode::decode_event_payload`] decode, and nothing
/// else — no committed-turn scoping, no tie-break, no empty-`persona_id`
/// filtering, no alias resolution. Every one of those stays with the caller;
/// see the module docs for why.
///
/// An event whose kind base matches but whose payload fails to decode (or
/// exceeds the decode cap) is silently skipped — matching every one of the
/// eleven call sites this primitive replaces, none of which surfaced a decode
/// failure as anything more than a dropped record.
#[must_use = "iterators are lazy — nothing decodes until you consume this"]
pub fn attribution_events(
    events: &[Event],
    scope: AttributionScope,
) -> impl Iterator<Item = AttributionFact> + '_ {
    attribution_events_with_positions(
        events.iter().enumerate().map(|(idx, ev)| (idx as u64, ev)),
        scope,
    )
}

/// [`attribution_events`], parametrized over an explicit `(position, event)`
/// source rather than assuming position = slice index.
///
/// Added for `crates/query/src/decode/attribution.rs` (#1579), whose own
/// `(u64, Event)` pairs carry the journal's own append position — usually,
/// but not by contract, equal to the pair's index in a full replay; the
/// query engine's decode tests deliberately exercise non-zero-based
/// positions, so this fold cannot assume position = index the way
/// [`attribution_events`]'s convenience wrapper does. See the module docs'
/// "`crates/query/src/decode/attribution.rs`" section.
#[must_use = "iterators are lazy — nothing decodes until you consume this"]
pub fn attribution_events_with_positions<'a>(
    events: impl Iterator<Item = (u64, &'a Event)> + 'a,
    scope: AttributionScope,
) -> impl Iterator<Item = AttributionFact> + 'a {
    events.filter_map(move |(position, ev)| {
        let (base, turn_id) = kinds::parse(&ev.kind);
        let kind = if base == kinds::CALLER {
            AttributionEventKind::Caller
        } else if base == kinds::PARTICIPANT {
            AttributionEventKind::Participant
        } else {
            return None;
        };
        if !scope.wants(kind) {
            return None;
        }
        let decoded =
            polyc_proto::events_decode::decode_event_payload::<AttributionEvent>(&ev.payload)?;
        Some(AttributionFact {
            position,
            kind,
            turn_id,
            persona_id: decoded.persona_id,
            role: decoded.role,
            identity: decoded.identity.into_option(),
            asserting_edge_id: decoded.asserting_edge_id,
            signer_pk_hex: decoded.signer_pk_hex,
            signature_hex: decoded.signature_hex,
        })
    })
}

/// The LAST caller persona id recorded per turn, in journal order — later
/// facts overwrite earlier ones for the same turn.
///
/// The one policy this module composes on the primitive, because three call
/// sites (`grpc::attribution::persona_committed_token_usage`,
/// `forensics::accumulate_conversation_usage`, and `forensics::compute_stats`)
/// independently join a turn's caller under the identical "last caller wins"
/// rule. Neither
/// commit-scoping nor tie-break for any OTHER site is folded in here — a
/// caller that wants first-wins, a collected set, or a different notion of
/// "committed" builds it directly off [`attribution_events`] instead.
///
/// Callers filter `facts` to whatever their own committed-turn definition
/// requires BEFORE passing them here (this helper applies no commit scoping
/// of its own) — except `compute_stats`, which deliberately passes unfiltered
/// facts because its downstream lookup only ever reads committed turn ids;
/// the no-op argument is documented at that call site. Pass an
/// already-scoped iterator, e.g.
/// `attribution_events(events, AttributionScope::CallerOnly).filter(|f|
/// committed.contains(&f.turn_id.unwrap_or_default()))`.
#[must_use]
pub fn caller_by_turn_last_wins(
    facts: impl Iterator<Item = AttributionFact>,
) -> HashMap<uuid::Uuid, String> {
    let mut out = HashMap::new();
    for fact in facts {
        if let Some(turn) = fact.turn_id {
            out.insert(turn, fact.persona_id);
        }
    }
    out
}

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

    use buffa::Message as _;
    use polyc_eventlog_model::Event;
    use uuid::Uuid;

    use super::*;

    fn caller_event(turn: &Uuid, persona_id: &str) -> Event {
        Event::new(
            kinds::tagged(kinds::CALLER, turn),
            AttributionEvent {
                persona_id: persona_id.to_owned(),
                role: "initiator".to_owned(),
                ..Default::default()
            }
            .encode_to_vec(),
        )
    }

    fn participant_event(turn: &Uuid, persona_id: &str) -> Event {
        Event::new(
            kinds::tagged(kinds::PARTICIPANT, turn),
            AttributionEvent {
                persona_id: persona_id.to_owned(),
                role: "participant".to_owned(),
                ..Default::default()
            }
            .encode_to_vec(),
        )
    }

    fn caller_event_with_identity(
        turn: &Uuid,
        persona_id: &str,
        identity: ExternalIdentity,
    ) -> Event {
        Event::new(
            kinds::tagged(kinds::CALLER, turn),
            AttributionEvent {
                persona_id: persona_id.to_owned(),
                role: "initiator".to_owned(),
                identity: buffa::MessageField::some(identity),
                ..Default::default()
            }
            .encode_to_vec(),
        )
    }

    #[test]
    fn caller_only_scope_excludes_participant_events() {
        let turn = Uuid::now_v7();
        let events = vec![
            caller_event(&turn, "persona-caller"),
            participant_event(&turn, "persona-participant"),
        ];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts.len(), 1);
        assert_eq!(facts[0].kind, AttributionEventKind::Caller);
        assert_eq!(facts[0].persona_id, "persona-caller");
    }

    #[test]
    fn participant_only_scope_excludes_caller_events() {
        let turn = Uuid::now_v7();
        let events = vec![
            caller_event(&turn, "persona-caller"),
            participant_event(&turn, "persona-participant"),
        ];
        let facts: Vec<_> =
            attribution_events(&events, AttributionScope::ParticipantOnly).collect();
        assert_eq!(facts.len(), 1);
        assert_eq!(facts[0].kind, AttributionEventKind::Participant);
        assert_eq!(facts[0].persona_id, "persona-participant");
    }

    #[test]
    fn both_scope_decodes_caller_and_participant_and_skips_unrelated_kinds() {
        let turn = Uuid::now_v7();
        let events = vec![
            caller_event(&turn, "persona-caller"),
            participant_event(&turn, "persona-participant"),
            Event::new(kinds::tagged(kinds::USER_MSG, &turn), Vec::new()),
        ];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::Both).collect();
        assert_eq!(facts.len(), 2);
        assert_eq!(facts[0].kind, AttributionEventKind::Caller);
        assert_eq!(facts[1].kind, AttributionEventKind::Participant);
    }

    #[test]
    fn turn_id_recovers_from_the_kind_suffix() {
        let turn = Uuid::now_v7();
        let events = vec![caller_event(&turn, "persona-1")];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts[0].turn_id, Some(turn));
    }

    #[test]
    fn bare_kind_with_no_turn_suffix_yields_no_turn_id() {
        let events = vec![Event::new(
            kinds::CALLER.to_owned(),
            AttributionEvent {
                persona_id: "persona-1".to_owned(),
                role: "initiator".to_owned(),
                ..Default::default()
            }
            .encode_to_vec(),
        )];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts.len(), 1);
        assert_eq!(facts[0].turn_id, None);
    }

    #[test]
    fn decode_failure_is_skipped_not_surfaced() {
        let turn = Uuid::now_v7();
        let events = vec![
            Event::new(kinds::tagged(kinds::CALLER, &turn), vec![0xFF, 0xFE, 0xFD]),
            caller_event(&turn, "persona-good"),
        ];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(
            facts.len(),
            1,
            "the undecodable payload is skipped, not errored"
        );
        assert_eq!(facts[0].persona_id, "persona-good");
    }

    #[test]
    fn persona_id_passes_through_raw_including_empty() {
        let turn = Uuid::now_v7();
        let events = vec![caller_event(&turn, "")];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(
            facts.len(),
            1,
            "an empty persona_id is not filtered by the primitive"
        );
        assert_eq!(facts[0].persona_id, "");
    }

    #[test]
    fn identity_extracts_when_present_and_none_when_absent() {
        let turn = Uuid::now_v7();
        let identity = ExternalIdentity {
            provider: "slack".to_owned(),
            scope: "T1".to_owned(),
            external_id: "U1".to_owned(),
            display_name: "Alice".to_owned(),
            ..Default::default()
        };
        let events = vec![
            caller_event_with_identity(&turn, "persona-1", identity.clone()),
            caller_event(&turn, "persona-2"),
        ];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts[0].identity, Some(identity));
        assert_eq!(facts[1].identity, None);
    }

    #[test]
    fn position_reflects_index_in_the_given_slice() {
        let turn = Uuid::now_v7();
        let events = vec![
            Event::new(kinds::tagged(kinds::USER_MSG, &turn), Vec::new()),
            caller_event(&turn, "persona-1"),
        ];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts[0].position, 1);
    }

    /// [`attribution_events_with_positions`] (#1579) uses the CALLER-supplied
    /// position, not the iterator's own index — the query engine's own
    /// `(u64, Event)` pairs are not always index-aligned (its own decode
    /// tests deliberately start position numbering above zero).
    #[test]
    fn attribution_events_with_positions_uses_the_supplied_position_not_the_index() {
        let turn = Uuid::now_v7();
        let events = [
            (
                5_u64,
                Event::new(kinds::tagged(kinds::USER_MSG, &turn), Vec::new()),
            ),
            (9_u64, caller_event(&turn, "persona-1")),
        ];
        let facts: Vec<_> = attribution_events_with_positions(
            events.iter().map(|(pos, ev)| (*pos, ev)),
            AttributionScope::CallerOnly,
        )
        .collect();
        assert_eq!(facts.len(), 1);
        assert_eq!(
            facts[0].position, 9,
            "position must come from the supplied pair, not the 0-based iterator index"
        );
    }

    /// `role` (#1579) is the payload's own free-text field, independent of
    /// `kind` — a `participant`-kind event whose payload happens to say
    /// `"initiator"` still reports `role: "initiator"`, and an empty `role`
    /// on a `caller`-kind event is not filled in from `kind`.
    #[test]
    fn role_is_raw_from_the_payload_not_derived_from_kind() {
        let turn = Uuid::now_v7();
        let events = vec![participant_event(&turn, "persona-1")];
        let facts: Vec<_> =
            attribution_events(&events, AttributionScope::ParticipantOnly).collect();
        assert_eq!(facts[0].kind, AttributionEventKind::Participant);
        assert_eq!(facts[0].role, "participant");

        let bare = Event::new(
            kinds::CALLER.to_owned(),
            AttributionEvent {
                persona_id: "persona-2".to_owned(),
                role: String::new(),
                ..Default::default()
            }
            .encode_to_vec(),
        );
        let facts: Vec<_> = attribution_events(&[bare], AttributionScope::CallerOnly).collect();
        assert_eq!(facts[0].kind, AttributionEventKind::Caller);
        assert_eq!(facts[0].role, "", "role is not backfilled from kind");
    }

    #[test]
    fn caller_by_turn_last_wins_keeps_the_later_fact() {
        let turn = Uuid::now_v7();
        let events = vec![
            caller_event(&turn, "persona-first"),
            caller_event(&turn, "persona-second"),
        ];
        let map =
            caller_by_turn_last_wins(attribution_events(&events, AttributionScope::CallerOnly));
        assert_eq!(map.get(&turn), Some(&"persona-second".to_owned()));
    }

    #[test]
    fn edge_provenance_fields_decode_when_present() {
        let turn = Uuid::now_v7();
        let events = vec![Event::new(
            kinds::tagged(kinds::CALLER, &turn),
            AttributionEvent {
                persona_id: String::new(),
                role: "edge".to_owned(),
                asserting_edge_id: "trigger-edge".to_owned(),
                signer_pk_hex: "deadbeef".to_owned(),
                signature_hex: "cafef00d".to_owned(),
                ..Default::default()
            }
            .encode_to_vec(),
        )];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts.len(), 1);
        assert_eq!(facts[0].asserting_edge_id, "trigger-edge");
        assert_eq!(facts[0].signer_pk_hex, "deadbeef");
        assert_eq!(facts[0].signature_hex, "cafef00d");
    }

    #[test]
    fn edge_provenance_fields_default_empty_when_absent() {
        let turn = Uuid::now_v7();
        let events = vec![caller_event(&turn, "persona-1")];
        let facts: Vec<_> = attribution_events(&events, AttributionScope::CallerOnly).collect();
        assert_eq!(facts[0].asserting_edge_id, "");
        assert_eq!(facts[0].signer_pk_hex, "");
        assert_eq!(facts[0].signature_hex, "");
    }

    #[test]
    fn caller_by_turn_last_wins_tracks_multiple_turns_independently() {
        let turn_a = Uuid::now_v7();
        let turn_b = Uuid::now_v7();
        let events = vec![
            caller_event(&turn_a, "persona-a"),
            caller_event(&turn_b, "persona-b"),
        ];
        let map =
            caller_by_turn_last_wins(attribution_events(&events, AttributionScope::CallerOnly));
        assert_eq!(map.get(&turn_a), Some(&"persona-a".to_owned()));
        assert_eq!(map.get(&turn_b), Some(&"persona-b".to_owned()));
    }
}