polyc-query 2026.8.3

//! `attribution` typed-table decoder — `AttributionEvent` → Arrow columns.
//!
//! The fact model's third typed table (#1311/#1178's "keyed
//! domain-fact-table" wave; see `crate::decode`'s module docs) —
//! `crate::decode::usage` and `crate::decode::model_call` established the
//! pattern; this module mirrors it for
//! `polychrome.events.v1.AttributionEvent`
//! (`crates/proto/proto/events.proto`'s `AttributionEvent` message),
//! decoded straight from the payload rather than left in the wide `events`
//! table's opaque `payload` column — the same design rationale
//! [`crate::decode::usage`] cites (docs/reference/datafusion-data-layer.md,
//! "Typed tables beside the wide events table").
//!
//! # Two kinds, one message (the pattern's first multi-kind table)
//!
//! Every earlier typed table matches exactly one kind-base. `attribution`
//! is the first that does not: `AttributionEvent` is written under BOTH
//! `caller` and `participant` (`crates/proto/proto/events.proto`'s
//! kind-to-message table lists both arrows pointing at
//! `polychrome.events.v1.AttributionEvent`), because a persona attributed
//! to a turn is recorded once as the turn's caller and, for every persona
//! who spoke in the thread (caller included), again as a participant —
//! `crates/control-plane/src/grpc/attribution.rs`'s `Caller`/`Participant`
//! kind pairing is the write side of exactly this split. `role` (this
//! table's own column) is what actually distinguishes the two at read
//! time — `"initiator"` for a `caller`-kind row, `"participant"` for a
//! `participant`-kind row (`polyc_persona::ROLE_INITIATOR`/
//! `ROLE_PARTICIPANT`, the same two strings `crate::decode::persona`'s
//! `participations.role` column already carries for the *directory's* own
//! initiator/participant distinction) — so a caller reading this table
//! never needs to know which of the two kinds a row came from, only what
//! `role` it decoded to.
//!
//! [`decode_attribution_events`] handles the two kinds by passing both
//! kind-bases to `crate::decode::decode_typed_kind_events` in one call,
//! rather than calling it twice and concatenating the results: that
//! function's `kind_bases` parameter (generalized from a single `&str` to
//! a `&[&str]` for exactly this table — see its own docs) already exists
//! to answer "does this row's kind match the set this table decodes",
//! which is precisely attribution's shape. A second call-and-concat would
//! duplicate the same filter/derive/decode loop with a different
//! `kind_base` argument and then re-merge two `Vec`s that were never
//! independent to begin with (one output row per input event, order
//! doesn't matter since the caller wraps every row's own `position`
//! already) — an extra step the slice-based filter already avoids for
//! free, and the one future multi-kind table can reuse identically rather
//! than each inventing its own concat.
//!
//! # Uniform keys (#1311)
//!
//! Same discipline as [`crate::decode::usage`]/[`crate::decode::model_call`]:
//! every row carries the fact model's uniform key columns — `partition`,
//! `position`, `turn_id` — so a row joins back to the conversation/turn it
//! belongs to (`crate::decode::events_batch`). `kind_base` is omitted here
//! too, but for a different reason than the single-kind tables: this
//! table's rows are NOT all one kind-base (see above), so a `kind_base`
//! column would actually carry information for `attribution` — the choice
//! is still to omit it, because `role` already carries the same
//! distinction in a form more useful to a query (`WHERE role =
//! 'initiator'` reads the same whether the writer ever renames the
//! underlying kind strings) and duplicating it as a second column would
//! just be two spellings of one fact. `conversation_id` is omitted as
//! usual — `partition` already *is* `conv-{id}`.
//!
//! `event_time` is deferred to #1327, same as every other typed table
//! today — no such column exists on `Event` yet, and this table does not
//! invent one; see [`crate::decode::usage`]'s module docs for the fuller
//! rationale.
//!
//! # Flattening `identity` (`ExternalIdentity`)
//!
//! [`AttributionEvent::identity`] is a nested message
//! (`polychrome.persona.v1.ExternalIdentity`,
//! `crates/proto/proto/persona.proto`), not a flat scalar. It is flattened
//! into four `identity_*` columns rather than exposed as a nested/struct
//! Arrow column: `ExternalIdentity` has exactly four fields —
//! `provider`, `scope`, `external_id`, `display_name` — and every one of
//! them is a plain proto3 `string`, so all four flatten cleanly with no
//! deferred remainder (unlike, say, [`crate::decode::persona`]'s
//! `identities` field, a `repeated` message left out of the `personas`
//! table entirely because a repeated field cannot flatten to a single
//! row's scalar column). Keeping this as a nested column would force a SQL
//! caller through `DataFusion`'s struct-field syntax for what is, on the
//! wire, four independent strings — flattening keeps `attribution` a plain
//! table any of `DataFusion`'s ordinary column predicates already work
//! against, matching the fact model's "clean queryable table" goal (#1311)
//! more directly than a struct column would.
//!
//! An unset `identity` field (`MessageField::is_unset`) decodes to
//! `ExternalIdentity::default()` — all four `identity_*` columns empty
//! strings, not `NULL` — the same "proto3 default, not absent" contract
//! every other typed table's plain string field already follows (see
//! [`crate::decode::model_call`]'s `provider`/`model` column docs for the
//! same rule applied to a top-level field).
//!
//! # Column selection
//!
//! [`AttributionRow`] keeps every field [`AttributionEvent`] carries —
//! unlike [`crate::decode::model_call`], which defers several of
//! [`ModelCallEvent`](polyc_proto::proto::polychrome::events::v1::ModelCallEvent)'s
//! fields, every one of `AttributionEvent`'s fields (`persona_id`,
//! `identity`, `role`, and the edge-provenance trio `asserting_edge_id`/
//! `signer_pk_hex`/`signature_hex` the edge-auth redesign added) is a
//! general, reusable, queryable fact about who — and which edge vouched
//! for — a turn's attribution, so there is nothing left to defer.
//!
//! # Shared decode (#1579)
//!
//! [`decode_attribution_events`] now decodes through
//! [`polyc_facts::attribution_events_with_positions`] — the SAME kind-filter/
//! decode primitive nine other call sites across control-plane already read
//! through (`polyc_facts::attribution_events`) — rather than its own
//! independent loop over `crate::decode::decode_typed_kind_events`. This
//! table's `role` column and this fold's policy-free posture (no
//! committed-turn scoping, no tie-break) were never in conflict with the
//! other nine sites' policy disagreements the fold's own module docs
//! describe; what blocked convergence was purely mechanical — this table's
//! `position` comes from the journal's own append position on each
//! `(u64, Event)` pair, while [`polyc_facts::attribution_events`]'s
//! convenience wrapper derives it from a plain slice's index — closed by
//! `polyc_facts::attribution_events_with_positions`, and the payload's raw
//! `role` field, which the fold now also carries (see
//! `polyc_facts::AttributionFact::role`'s doc). See
//! `crates/facts/src/attribution.rs`'s own module docs for the fuller
//! account. The edge-provenance trio rides that same shared fact, so
//! those columns come through the fold too rather than a second decode.
//!
//! One observability behavior does NOT come from the fold:
//! [`decode_attribution_events`] logged a WARN on a corrupt non-empty
//! payload before this convergence, and still does — via
//! [`warn_undecodable_attribution_payloads`], a separate read-only probe
//! run alongside the fold rather than a change to it, since the fold is
//! shared by nine OTHER call sites that never wanted that log side effect.

use std::sync::Arc;

use arrow::array::{ArrayRef, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_eventlog::Event;
use polyc_proto::events_decode::decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::AttributionEvent;

/// One decoded `attribution` row.
///
/// The fact model's uniform key columns (`partition`, `position`,
/// `turn_id`) plus [`AttributionEvent`]'s own fields, with `identity`
/// flattened to four `identity_*` columns (see the module docs' "Flattening
/// `identity`" section).
#[derive(Debug, Clone)]
pub(crate) struct AttributionRow {
    /// The journal partition this row's event was read from (`conv-{id}`).
    pub partition: String,
    /// The journal's own monotonic append position for this event.
    pub position: u64,
    /// The `:{turn_uuid}` suffix off the event's `kind`, canonical
    /// hyphenated form, or `None` for a bare `caller`/`participant` kind
    /// with no turn tagged.
    pub turn_id: Option<String>,
    /// The durable persona this row attributes the turn/message to.
    pub persona_id: String,
    /// `"initiator"` for a row decoded from a `caller`-kind event,
    /// `"participant"` for one decoded from a `participant`-kind event
    /// (`polyc_persona::ROLE_INITIATOR`/`ROLE_PARTICIPANT` — this crate
    /// does not depend on `polyc-persona`, so the row stores the plain
    /// string the payload itself carries rather than that crate's
    /// constants).
    pub role: String,
    /// [`ExternalIdentity::provider`] — the edge namespace ("slack",
    /// "telegram", "api-key", "email", …) the identity was observed
    /// through.
    pub identity_provider: String,
    /// [`ExternalIdentity::scope`] — provider scope qualifying the id
    /// (e.g. the workspace/team id); empty for providers with
    /// globally-unique ids.
    pub identity_scope: String,
    /// [`ExternalIdentity::external_id`] — the provider-native stable
    /// identifier.
    pub identity_external_id: String,
    /// [`ExternalIdentity::display_name`] — display name as last observed
    /// at the edge; presentation only.
    pub identity_display_name: String,
    /// [`AttributionEvent::asserting_edge_id`] — which edge vouched for
    /// this attribution via a signed `AssertedAttribution` envelope; empty
    /// for internal/unsigned attribution.
    pub asserting_edge_id: String,
    /// [`AttributionEvent::signer_pk_hex`] — the asserting edge's
    /// registered ed25519 public key, hex-encoded; empty when
    /// `asserting_edge_id` is empty.
    pub signer_pk_hex: String,
    /// [`AttributionEvent::signature_hex`] — the envelope's own signature,
    /// hex-encoded, recorded for forensics; empty when `asserting_edge_id`
    /// is empty.
    pub signature_hex: String,
}

/// The `attribution` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `persona_id` (`Utf8`, non-null), `role`
/// (`Utf8`, non-null), `identity_provider` (`Utf8`, non-null),
/// `identity_scope` (`Utf8`, non-null), `identity_external_id` (`Utf8`,
/// non-null), `identity_display_name` (`Utf8`, non-null),
/// `asserting_edge_id` (`Utf8`, non-null), `signer_pk_hex` (`Utf8`,
/// non-null), `signature_hex` (`Utf8`, non-null).
///
/// See the module docs for why `kind_base`/`conversation_id` are omitted,
/// why `identity` is flattened rather than nested, and why `event_time` is
/// deferred to #1327 rather than faked.
#[must_use]
pub(crate) fn schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("partition", DataType::Utf8, false),
        Field::new("position", DataType::UInt64, false),
        Field::new("turn_id", DataType::Utf8, true),
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("role", DataType::Utf8, false),
        Field::new("identity_provider", DataType::Utf8, false),
        Field::new("identity_scope", DataType::Utf8, false),
        Field::new("identity_external_id", DataType::Utf8, false),
        Field::new("identity_display_name", DataType::Utf8, false),
        Field::new("asserting_edge_id", DataType::Utf8, false),
        Field::new("signer_pk_hex", DataType::Utf8, false),
        Field::new("signature_hex", DataType::Utf8, false),
    ]))
}

/// Decode already-framed [`AttributionRow`]s into the `attribution` table's
/// Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_attribution_batch(rows: &[AttributionRow]) -> Result<RecordBatch, ArrowError> {
    let mut partition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut position_b = UInt64Builder::with_capacity(rows.len());
    let mut turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut role_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut identity_provider_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut identity_scope_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut identity_external_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut identity_display_name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut asserting_edge_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut signer_pk_hex_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
    let mut signature_hex_b = StringBuilder::with_capacity(rows.len(), rows.len() * 128);

    for row in rows {
        partition_b.append_value(&row.partition);
        position_b.append_value(row.position);
        match &row.turn_id {
            Some(id) => turn_id_b.append_value(id),
            None => turn_id_b.append_null(),
        }
        persona_id_b.append_value(&row.persona_id);
        role_b.append_value(&row.role);
        identity_provider_b.append_value(&row.identity_provider);
        identity_scope_b.append_value(&row.identity_scope);
        identity_external_id_b.append_value(&row.identity_external_id);
        identity_display_name_b.append_value(&row.identity_display_name);
        asserting_edge_id_b.append_value(&row.asserting_edge_id);
        signer_pk_hex_b.append_value(&row.signer_pk_hex);
        signature_hex_b.append_value(&row.signature_hex);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(persona_id_b.finish()),
        Arc::new(role_b.finish()),
        Arc::new(identity_provider_b.finish()),
        Arc::new(identity_scope_b.finish()),
        Arc::new(identity_external_id_b.finish()),
        Arc::new(identity_display_name_b.finish()),
        Arc::new(asserting_edge_id_b.finish()),
        Arc::new(signer_pk_hex_b.finish()),
        Arc::new(signature_hex_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Filter `partition`'s framed `events` to `caller`/`participant` rows,
/// decode each payload via the shared
/// [`polyc_facts::attribution_events_with_positions`] fold, and pair it
/// with that row's key columns.
///
/// Delegates the filter/derive/decode loop to the shared fold (see the
/// module docs' "Shared decode" section) — [`polyc_facts::AttributionScope::Both`]
/// covers both kind-bases in one call, mirroring `ATTRIBUTION_KIND_BASES`'s
/// old two-kinds-one-call shape. Only the
/// [`polyc_facts::AttributionFact`] → [`AttributionRow`] field mapping,
/// including flattening `identity`, is unique to this module.
///
/// The shared fold itself silently skips a corrupt payload (matching every
/// one of the nine other call sites it replaces, none of which surfaced a
/// decode failure as more than a dropped record — see the fold's own
/// module docs). This table logged a WARN on that case before #1579's
/// convergence, the same posture `crate::decode::decode_typed_kind_events`
/// still applies for `usage`/`model_call`/`turn_failed`/`summary`, so
/// [`warn_undecodable_attribution_payloads`] restores it here as an
/// independent, policy-free probe — it never influences which rows this
/// function returns, only what gets logged.
#[must_use]
pub(crate) fn decode_attribution_events(
    partition: &str,
    events: &[(u64, Event)],
) -> Vec<AttributionRow> {
    warn_undecodable_attribution_payloads(events);
    polyc_facts::attribution_events_with_positions(
        events.iter().map(|(position, event)| (*position, event)),
        polyc_facts::AttributionScope::Both,
    )
    .map(|fact| {
        let identity = fact.identity.unwrap_or_default();
        AttributionRow {
            partition: partition.to_string(),
            position: fact.position,
            turn_id: fact.turn_id.map(|u| u.to_string()),
            persona_id: fact.persona_id,
            role: fact.role,
            identity_provider: identity.provider,
            identity_scope: identity.scope,
            identity_external_id: identity.external_id,
            identity_display_name: identity.display_name,
            // The edge-provenance trio rides the SAME shared fact (see
            // `polyc_facts::AttributionFact`'s own field docs), so these
            // columns come through the fold rather than a second decode.
            asserting_edge_id: fact.asserting_edge_id,
            signer_pk_hex: fact.signer_pk_hex,
            signature_hex: fact.signature_hex,
        }
    })
    .collect()
}

/// Log a WARN for each `caller`/`participant` event whose non-empty payload
/// fails to decode as [`AttributionEvent`] — restores the observability
/// [`decode_attribution_events`] otherwise lost once it moved onto
/// [`polyc_facts::attribution_events_with_positions`] (a fold shared by nine
/// OTHER call sites too, none of which want a log side effect baked into
/// the primitive itself — see that fold's own module docs on why it stays
/// policy-free).
///
/// A separate, read-only pass over the same `events` rather than a change to
/// the fold: it re-runs the identical decode
/// ([`polyc_proto::events_decode::decode_event_payload`] is a pure,
/// side-effect-free function) purely to notice a failure worth logging, and
/// never affects which rows [`decode_attribution_events`] returns — the
/// exact "pre-decode probe, keep the fold policy-free" shape matching
/// `crate::decode::decode_typed_kind_events`'s own skip-and-warn posture for
/// every OTHER typed table in this crate.
fn warn_undecodable_attribution_payloads(events: &[(u64, Event)]) {
    for (position, event) in events {
        let position: u64 = *position;
        let (base, _) = kinds::parse(&event.kind);
        if base != kinds::CALLER && base != kinds::PARTICIPANT {
            continue;
        }
        if event.payload.is_empty() {
            continue;
        }
        if decode_event_payload::<AttributionEvent>(&event.payload).is_none() {
            tracing::warn!(
                len = event.payload.len(),
                table = "attribution",
                position,
                "corrupt event payload; skipping row"
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use buffa::Message as _;
    use polyc_proto::kinds;
    use polyc_proto::proto::polychrome::events::v1::AttributionEvent;
    use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
    use uuid::Uuid;

    use super::*;

    /// An [`AttributionEvent`] naming every field explicit rather than
    /// `..Default::default()`, so a reader sees exactly what a real
    /// payload carries. `__buffa_unknown_fields` is buffa's own
    /// hidden-but-public round-trip-fidelity field
    /// (`buffa::UnknownFields`, always empty for a hand-built fixture) —
    /// every generated message carries it, so it is named here too rather
    /// than reached for a struct-update spread.
    fn sample_attribution(
        persona_id: &str,
        role: &str,
        identity: ExternalIdentity,
    ) -> AttributionEvent {
        AttributionEvent {
            persona_id: persona_id.to_string(),
            identity: buffa::MessageField::some(identity),
            role: role.to_string(),
            asserting_edge_id: String::new(),
            signer_pk_hex: String::new(),
            signature_hex: String::new(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn sample_identity(
        provider: &str,
        scope: &str,
        external_id: &str,
        display_name: &str,
    ) -> ExternalIdentity {
        ExternalIdentity {
            provider: provider.to_string(),
            scope: scope.to_string(),
            external_id: external_id.to_string(),
            display_name: display_name.to_string(),
            time_zone: String::new(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    #[test]
    fn schema_shape() {
        let schema = schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "partition",
                "position",
                "turn_id",
                "persona_id",
                "role",
                "identity_provider",
                "identity_scope",
                "identity_external_id",
                "identity_display_name",
                "asserting_edge_id",
                "signer_pk_hex",
                "signature_hex",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("persona_id", DataType::Utf8, false),
            ("role", DataType::Utf8, false),
            ("identity_provider", DataType::Utf8, false),
            ("identity_scope", DataType::Utf8, false),
            ("identity_external_id", DataType::Utf8, false),
            ("identity_display_name", DataType::Utf8, false),
            ("asserting_edge_id", DataType::Utf8, false),
            ("signer_pk_hex", DataType::Utf8, false),
            ("signature_hex", DataType::Utf8, false),
        ];
        for (field, (name, ty, nullable)) in schema.fields().iter().zip(expect) {
            assert_eq!(field.name(), name);
            assert_eq!(field.data_type(), &ty);
            assert_eq!(field.is_nullable(), nullable);
        }
    }

    #[test]
    fn decode_attribution_batch_round_trips() {
        let rows = vec![AttributionRow {
            partition: "conv-a".to_string(),
            position: 5,
            turn_id: Some("turn-xyz".to_string()),
            persona_id: "persona-1".to_string(),
            role: "initiator".to_string(),
            identity_provider: "slack".to_string(),
            identity_scope: "team-1".to_string(),
            identity_external_id: "U123".to_string(),
            identity_display_name: "Ada".to_string(),
            asserting_edge_id: "slack-edge".to_string(),
            signer_pk_hex: "deadbeef".to_string(),
            signature_hex: "cafef00d".to_string(),
        }];
        let batch = decode_attribution_batch(&rows).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

        let persona_id = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(persona_id.value(0), "persona-1");

        let role = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(role.value(0), "initiator");

        let identity_provider = batch
            .column(5)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(identity_provider.value(0), "slack");

        let identity_external_id = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(identity_external_id.value(0), "U123");

        let asserting_edge_id = batch
            .column(9)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(asserting_edge_id.value(0), "slack-edge");

        let signer_pk_hex = batch
            .column(10)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(signer_pk_hex.value(0), "deadbeef");

        let signature_hex = batch
            .column(11)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(signature_hex.value(0), "cafef00d");
    }

    /// A real buffa-encoded `AttributionEvent` under a `caller` kind with a
    /// `:turn_uuid` suffix decodes to one row with its keys, `persona_id`,
    /// `role`, and every flattened `identity_*` field intact.
    #[test]
    fn decode_attribution_events_filters_and_decodes_real_buffa_bytes_for_caller() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let identity = sample_identity("slack", "team-9", "U999", "Grace");
        let attribution = sample_attribution("persona-caller", "initiator", identity);
        let bytes = attribution.encode_to_vec();

        let events = vec![
            (1, Event::new(kinds::TURN_START, Vec::new())),
            (2, Event::new(kinds::tagged(kinds::CALLER, &turn), bytes)),
            (3, Event::new(kinds::USER_MSG, b"not attribution".to_vec())),
        ];

        let decoded = decode_attribution_events("conv-real", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].partition, "conv-real");
        assert_eq!(decoded[0].position, 2);
        assert_eq!(decoded[0].turn_id, Some(turn.to_string()));
        assert_eq!(decoded[0].persona_id, "persona-caller");
        assert_eq!(decoded[0].role, "initiator");
        assert_eq!(decoded[0].identity_provider, "slack");
        assert_eq!(decoded[0].identity_scope, "team-9");
        assert_eq!(decoded[0].identity_external_id, "U999");
        assert_eq!(decoded[0].identity_display_name, "Grace");
    }

    /// Both source kinds decode through the same call: a `caller`-kind row
    /// and a `participant`-kind row in the same partition both appear,
    /// each with the role its own kind implies.
    #[test]
    fn both_source_kinds_decode_with_their_own_role() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let caller_identity = sample_identity("slack", "team-1", "U1", "Alice");
        let caller = sample_attribution("persona-alice", "initiator", caller_identity);
        let participant_identity = sample_identity("telegram", "", "T2", "Bob");
        let participant = sample_attribution("persona-bob", "participant", participant_identity);

        let events = vec![
            (
                1,
                Event::new(kinds::tagged(kinds::CALLER, &turn), caller.encode_to_vec()),
            ),
            (
                2,
                Event::new(
                    kinds::tagged(kinds::PARTICIPANT, &turn),
                    participant.encode_to_vec(),
                ),
            ),
        ];

        let mut decoded = decode_attribution_events("conv-both", &events);
        decoded.sort_by_key(|row| row.position);
        assert_eq!(decoded.len(), 2);
        assert_eq!(decoded[0].persona_id, "persona-alice");
        assert_eq!(decoded[0].role, "initiator");
        assert_eq!(decoded[1].persona_id, "persona-bob");
        assert_eq!(decoded[1].role, "participant");
    }

    #[test]
    fn decode_attribution_events_bare_kind_has_no_turn_id() {
        let events = vec![(7, Event::new(kinds::CALLER, Vec::new()))];
        let decoded = decode_attribution_events("conv-bare", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].partition, "conv-bare");
        assert_eq!(decoded[0].position, 7);
        assert_eq!(decoded[0].turn_id, None);
    }

    #[test]
    fn empty_payload_attribution_event_decodes_to_defaults() {
        let events = vec![(1, Event::new(kinds::PARTICIPANT, Vec::new()))];
        let decoded = decode_attribution_events("conv-empty", &events);
        assert_eq!(decoded.len(), 1, "an empty payload must decode, not skip");
        assert_eq!(decoded[0].persona_id, "");
        assert_eq!(decoded[0].role, "");
        assert_eq!(decoded[0].identity_provider, "");
        assert_eq!(decoded[0].identity_scope, "");
        assert_eq!(decoded[0].identity_external_id, "");
        assert_eq!(decoded[0].identity_display_name, "");
        assert_eq!(decoded[0].asserting_edge_id, "");
        assert_eq!(decoded[0].signer_pk_hex, "");
        assert_eq!(decoded[0].signature_hex, "");
    }

    /// The edge-provenance trio (`asserting_edge_id`/`signer_pk_hex`/
    /// `signature_hex`) decodes off a real buffa-encoded payload exactly
    /// like `persona_id`/`role` do — the read-side twin of
    /// `crate::grpc::attribution::EdgeProvenance` being stamped onto the
    /// event on the write side.
    #[test]
    fn decode_attribution_events_surfaces_edge_provenance_fields() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_7777);
        let attribution = AttributionEvent {
            persona_id: String::new(),
            identity: buffa::MessageField::none(),
            role: "edge".to_string(),
            asserting_edge_id: "trigger-edge".to_string(),
            signer_pk_hex: "deadbeef".to_string(),
            signature_hex: "cafef00d".to_string(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let events = vec![(
            1,
            Event::new(
                kinds::tagged(kinds::CALLER, &turn),
                attribution.encode_to_vec(),
            ),
        )];

        let decoded = decode_attribution_events("conv-edge-provenance", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].role, "edge");
        assert_eq!(decoded[0].persona_id, "");
        assert_eq!(decoded[0].asserting_edge_id, "trigger-edge");
        assert_eq!(decoded[0].signer_pk_hex, "deadbeef");
        assert_eq!(decoded[0].signature_hex, "cafef00d");
    }

    #[test]
    fn undecodable_non_empty_payload_is_skipped() {
        let events = vec![
            (1, Event::new(kinds::CALLER, vec![0xFF, 0xFE, 0xFD])),
            (2, Event::new(kinds::PARTICIPANT, Vec::new())),
        ];
        let decoded = decode_attribution_events("conv-corrupt", &events);
        assert_eq!(decoded.len(), 1, "only the empty (valid) payload decodes");
        assert_eq!(decoded[0].position, 2);
        assert_eq!(decoded[0].persona_id, "");
    }

    /// An event kind outside `{caller, participant}` is not attribution
    /// data at all and must never surface as a row here, even if its
    /// payload happens to decode as `AttributionEvent` bytes.
    #[test]
    fn unrelated_kind_is_not_decoded_as_attribution() {
        let identity = sample_identity("slack", "team-1", "U1", "Alice");
        let attribution = sample_attribution("persona-alice", "initiator", identity);
        let events = vec![(1, Event::new(kinds::USAGE, attribution.encode_to_vec()))];
        let decoded = decode_attribution_events("conv-unrelated", &events);
        assert_eq!(
            decoded.len(),
            0,
            "a usage-kind event must never decode as attribution"
        );
    }

    #[test]
    fn attribution_event_round_trip_through_decode_and_batch() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_1111);
        let identity = sample_identity("email", "", "user@example.test", "Sam");
        let attribution = sample_attribution("persona-rt", "participant", identity);
        let events = vec![(
            1,
            Event::new(
                kinds::tagged(kinds::PARTICIPANT, &turn),
                attribution.encode_to_vec(),
            ),
        )];

        let decoded = decode_attribution_events("conv-rt", &events);
        let batch = decode_attribution_batch(&decoded).expect("batch build");

        let turn_id = batch
            .column(2)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(turn_id.value(0), turn.to_string());

        let persona_id = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(persona_id.value(0), "persona-rt");

        let identity_display_name = batch
            .column(8)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(identity_display_name.value(0), "Sam");
    }
}