polyc-query 2026.9.0

//! `personas`/`participations`/`persona_identities` (#1312) — see the
//! parent module's own doc for the full column-selection rationale shared
//! across every table in `crate::decode::persona`.

use std::sync::Arc;

use arrow::array::{ArrayRef, BooleanBuilder, StringBuilder, UInt32Builder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_proto::proto::polychrome::persona::v1::{
    ExternalIdentity, Participation, PersonaProfile,
};

/// One decoded `personas` row: the reusable subset of one [`PersonaProfile`]
/// (see the parent module docs' "Column selection" section for what is kept
/// and why), widened (#1578) to carry every lifecycle/audit/profile-document
/// field [`PersonaProfile`] flows end to end but the first cut of this row
/// discarded.
#[derive(Debug, Clone)]
pub(crate) struct PersonaRow {
    /// The persona id (UUID, minted at creation) — this table's join key.
    pub persona_id: String,
    /// Display name as last observed at an edge; presentation only.
    pub display_name: String,
    /// `"provisional"`, `"linked"`, `"merged"`, or `"removed"`.
    pub status: String,
    /// Unix ms the persona record was created.
    pub created_at_ms: u64,
    /// When `status == "merged"`: the surviving persona id this record was
    /// absorbed into. Empty string (never absent) when this persona has not
    /// been merged — `persona.proto`'s `merged_into` is a plain proto3
    /// string, whose zero value is `""`, same as `display_name`.
    pub merged_into: String,
    /// Merge audit: who performed the merge. Empty when never merged.
    pub merged_by: String,
    /// Merge audit: the evidence cited for the merge. Empty when never
    /// merged.
    pub merge_evidence: String,
    /// Merge audit: when the merge happened. `0` when never merged.
    pub merged_at_ms: u64,
    /// Split audit: the persona this record was peeled out of. Empty when
    /// not split-created.
    pub split_from: String,
    /// Split audit: who performed the split. Empty when not split-created.
    pub split_by: String,
    /// Split audit: the reason cited for the split. Empty when not
    /// split-created.
    pub split_reason: String,
    /// Split audit: when the split happened. `0` when not split-created.
    pub split_at_ms: u64,
    /// Removal audit: the admin who de-admitted this persona. Empty unless
    /// `status == "removed"`.
    pub removed_by: String,
    /// Removal audit: when the removal happened. `0` unless `status ==
    /// "removed"`.
    pub removed_at_ms: u64,
    /// Admin-removal audit: the admin who took away this persona's admin
    /// role. Empty unless this persona was demoted from admin.
    pub admin_removed_by: String,
    /// Admin-removal audit: when the demotion happened. `0` unless this
    /// persona was demoted from admin.
    pub admin_removed_at_ms: u64,
    /// `ProfileDocument.language` (BCP-47); empty when unset or `profile`
    /// itself is absent.
    pub profile_language: String,
    /// `ProfileDocument.tone`; empty when unset or `profile` itself is
    /// absent.
    pub profile_tone: String,
    /// `ProfileDocument.updated_at_ms`; `0` when `profile` is absent.
    pub profile_updated_at_ms: u64,
    /// `ProfileDocument.preferences` (`map<string, string>`), JSON-object
    /// text (see the parent module docs' "JSON-shaped string columns"
    /// section); `"{}"` when empty or `profile` itself is absent.
    pub profile_preferences: String,
    /// `Permissions.tool_grants`, JSON-array text; `"[]"` when empty or
    /// `profile`/`permissions` is absent.
    pub permissions_tool_grants: String,
    /// `Permissions.approval_policy`; empty when `profile`/`permissions` is
    /// absent.
    pub permissions_approval_policy: String,
    /// `Permissions.turn_quota`; `0` when `profile`/`permissions` is absent
    /// (the floor default applies).
    pub permissions_turn_quota: u32,
    /// `Permissions.admin`; `false` when `profile`/`permissions` is absent.
    pub permissions_admin: bool,
    /// `Permissions.auto_review`; `false` when `profile`/`permissions` is
    /// absent.
    pub permissions_auto_review: bool,
    /// `Permissions.operator`; `false` when `profile`/`permissions` is
    /// absent.
    pub permissions_operator: bool,
}

impl From<&PersonaProfile> for PersonaRow {
    fn from(profile: &PersonaProfile) -> Self {
        let document = profile.profile.as_option();
        let permissions = document.and_then(|doc| doc.permissions.as_option());
        Self {
            persona_id: profile.persona_id.clone(),
            display_name: profile.display_name.clone(),
            status: profile.status.clone(),
            created_at_ms: profile.created_at_ms,
            merged_into: profile.merged_into.clone(),
            merged_by: profile.merged_by.clone(),
            merge_evidence: profile.merge_evidence.clone(),
            merged_at_ms: profile.merged_at_ms,
            split_from: profile.split_from.clone(),
            split_by: profile.split_by.clone(),
            split_reason: profile.split_reason.clone(),
            split_at_ms: profile.split_at_ms,
            removed_by: profile.removed_by.clone(),
            removed_at_ms: profile.removed_at_ms,
            admin_removed_by: profile.admin_removed_by.clone(),
            admin_removed_at_ms: profile.admin_removed_at_ms,
            profile_language: document.map(|doc| doc.language.clone()).unwrap_or_default(),
            profile_tone: document.map(|doc| doc.tone.clone()).unwrap_or_default(),
            profile_updated_at_ms: document.map_or(0, |doc| doc.updated_at_ms),
            profile_preferences: document.map_or_else(
                || "{}".to_string(),
                |doc| serde_json::to_string(&doc.preferences).unwrap_or_else(|_| "{}".to_string()),
            ),
            permissions_tool_grants: permissions.map_or_else(
                || "[]".to_string(),
                |p| serde_json::to_string(&p.tool_grants).unwrap_or_else(|_| "[]".to_string()),
            ),
            permissions_approval_policy: permissions
                .map(|p| p.approval_policy.clone())
                .unwrap_or_default(),
            permissions_turn_quota: permissions.map_or(0, |p| p.turn_quota),
            permissions_admin: permissions.is_some_and(|p| p.admin),
            permissions_auto_review: permissions.is_some_and(|p| p.auto_review),
            permissions_operator: permissions.is_some_and(|p| p.operator),
        }
    }
}

/// The `personas` table's Arrow schema.
///
/// `persona_id` (`Utf8`, non-null), `display_name` (`Utf8`, non-null —
/// empty string, not null, for an unnamed persona: `persona.proto`'s
/// `display_name` is a plain proto3 string, whose zero value is `""`, never
/// absent), `status` (`Utf8`, non-null), `created_at_ms` (`UInt64`,
/// non-null), `merged_into` (`Utf8`, non-null — empty string, not null, when
/// this persona has not been merged, the same proto3-zero-value convention
/// as `display_name`), then (#1578) the lifecycle/audit columns
/// (`merged_by`, `merge_evidence`, `merged_at_ms`, `split_from`, `split_by`,
/// `split_reason`, `split_at_ms`, `removed_by`, `removed_at_ms`,
/// `admin_removed_by`, `admin_removed_at_ms` — every one `Utf8`/`UInt64`,
/// non-null, empty-string/zero sentinels for "never happened") and the
/// flattened profile-document columns (`profile_language`, `profile_tone`,
/// `profile_updated_at_ms`, `profile_preferences`, `permissions_tool_grants`,
/// `permissions_approval_policy`, `permissions_turn_quota`,
/// `permissions_admin`, `permissions_auto_review`, `permissions_operator` —
/// see [`PersonaRow`]'s own field docs for each column's absent-`profile`
/// sentinel).
#[must_use]
pub(crate) fn personas_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("display_name", DataType::Utf8, false),
        Field::new("status", DataType::Utf8, false),
        Field::new("created_at_ms", DataType::UInt64, false),
        Field::new("merged_into", DataType::Utf8, false),
        Field::new("merged_by", DataType::Utf8, false),
        Field::new("merge_evidence", DataType::Utf8, false),
        Field::new("merged_at_ms", DataType::UInt64, false),
        Field::new("split_from", DataType::Utf8, false),
        Field::new("split_by", DataType::Utf8, false),
        Field::new("split_reason", DataType::Utf8, false),
        Field::new("split_at_ms", DataType::UInt64, false),
        Field::new("removed_by", DataType::Utf8, false),
        Field::new("removed_at_ms", DataType::UInt64, false),
        Field::new("admin_removed_by", DataType::Utf8, false),
        Field::new("admin_removed_at_ms", DataType::UInt64, false),
        Field::new("profile_language", DataType::Utf8, false),
        Field::new("profile_tone", DataType::Utf8, false),
        Field::new("profile_updated_at_ms", DataType::UInt64, false),
        Field::new("profile_preferences", DataType::Utf8, false),
        Field::new("permissions_tool_grants", DataType::Utf8, false),
        Field::new("permissions_approval_policy", DataType::Utf8, false),
        Field::new("permissions_turn_quota", DataType::UInt32, false),
        Field::new("permissions_admin", DataType::Boolean, false),
        Field::new("permissions_auto_review", DataType::Boolean, false),
        Field::new("permissions_operator", DataType::Boolean, false),
    ]))
}

/// Decode [`PersonaRow`]s into the `personas` table's Arrow `RecordBatch`,
/// in [`personas_schema`] column order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
#[allow(clippy::too_many_lines)] // one builder + one append per column; no complexity to extract
pub(crate) fn decode_personas_batch(rows: &[PersonaRow]) -> Result<RecordBatch, ArrowError> {
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut display_name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut status_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut created_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut merged_into_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut merged_by_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut merge_evidence_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut merged_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut split_from_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut split_by_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut split_reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut split_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut removed_by_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut removed_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut admin_removed_by_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut admin_removed_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut profile_language_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut profile_tone_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut profile_updated_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut profile_preferences_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut permissions_tool_grants_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut permissions_approval_policy_b =
        StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut permissions_turn_quota_b = UInt32Builder::with_capacity(rows.len());
    let mut permissions_admin_b = BooleanBuilder::with_capacity(rows.len());
    let mut permissions_auto_review_b = BooleanBuilder::with_capacity(rows.len());
    let mut permissions_operator_b = BooleanBuilder::with_capacity(rows.len());

    for row in rows {
        persona_id_b.append_value(&row.persona_id);
        display_name_b.append_value(&row.display_name);
        status_b.append_value(&row.status);
        created_at_ms_b.append_value(row.created_at_ms);
        merged_into_b.append_value(&row.merged_into);
        merged_by_b.append_value(&row.merged_by);
        merge_evidence_b.append_value(&row.merge_evidence);
        merged_at_ms_b.append_value(row.merged_at_ms);
        split_from_b.append_value(&row.split_from);
        split_by_b.append_value(&row.split_by);
        split_reason_b.append_value(&row.split_reason);
        split_at_ms_b.append_value(row.split_at_ms);
        removed_by_b.append_value(&row.removed_by);
        removed_at_ms_b.append_value(row.removed_at_ms);
        admin_removed_by_b.append_value(&row.admin_removed_by);
        admin_removed_at_ms_b.append_value(row.admin_removed_at_ms);
        profile_language_b.append_value(&row.profile_language);
        profile_tone_b.append_value(&row.profile_tone);
        profile_updated_at_ms_b.append_value(row.profile_updated_at_ms);
        profile_preferences_b.append_value(&row.profile_preferences);
        permissions_tool_grants_b.append_value(&row.permissions_tool_grants);
        permissions_approval_policy_b.append_value(&row.permissions_approval_policy);
        permissions_turn_quota_b.append_value(row.permissions_turn_quota);
        permissions_admin_b.append_value(row.permissions_admin);
        permissions_auto_review_b.append_value(row.permissions_auto_review);
        permissions_operator_b.append_value(row.permissions_operator);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(persona_id_b.finish()),
        Arc::new(display_name_b.finish()),
        Arc::new(status_b.finish()),
        Arc::new(created_at_ms_b.finish()),
        Arc::new(merged_into_b.finish()),
        Arc::new(merged_by_b.finish()),
        Arc::new(merge_evidence_b.finish()),
        Arc::new(merged_at_ms_b.finish()),
        Arc::new(split_from_b.finish()),
        Arc::new(split_by_b.finish()),
        Arc::new(split_reason_b.finish()),
        Arc::new(split_at_ms_b.finish()),
        Arc::new(removed_by_b.finish()),
        Arc::new(removed_at_ms_b.finish()),
        Arc::new(admin_removed_by_b.finish()),
        Arc::new(admin_removed_at_ms_b.finish()),
        Arc::new(profile_language_b.finish()),
        Arc::new(profile_tone_b.finish()),
        Arc::new(profile_updated_at_ms_b.finish()),
        Arc::new(profile_preferences_b.finish()),
        Arc::new(permissions_tool_grants_b.finish()),
        Arc::new(permissions_approval_policy_b.finish()),
        Arc::new(permissions_turn_quota_b.finish()),
        Arc::new(permissions_admin_b.finish()),
        Arc::new(permissions_auto_review_b.finish()),
        Arc::new(permissions_operator_b.finish()),
    ];
    RecordBatch::try_new(personas_schema(), columns)
}

/// One decoded `participations` row: `persona_id` (supplied by the caller —
/// see the parent module docs) plus the reusable subset of one
/// [`Participation`].
#[derive(Debug, Clone)]
pub(crate) struct ParticipationRow {
    /// The persona this tie belongs to — this table's join key back to
    /// `personas`.
    pub persona_id: String,
    /// The conversation this persona is tied to — joins to the event
    /// tables' `partition` column (`conv-{conversation_id}`).
    pub conversation_id: String,
    /// `"initiator"` (triggered a turn) or `"participant"` (spoke in the
    /// thread).
    pub role: String,
    /// Unix ms this tie was first recorded — this table's "joined at".
    pub first_at_ms: u64,
}

impl ParticipationRow {
    /// Pair `persona_id` (not carried on the stored [`Participation`] value
    /// itself — see the parent module docs) with `participation`'s reusable
    /// fields.
    #[must_use]
    pub(crate) fn new(persona_id: &str, participation: &Participation) -> Self {
        Self {
            persona_id: persona_id.to_string(),
            conversation_id: participation.conversation_id.clone(),
            role: participation.role.clone(),
            first_at_ms: participation.first_at_ms,
        }
    }
}

/// The `participations` table's Arrow schema.
///
/// `persona_id` (`Utf8`, non-null), `conversation_id` (`Utf8`, non-null),
/// `role` (`Utf8`, non-null), `first_at_ms` (`UInt64`, non-null).
#[must_use]
pub(crate) fn participations_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("conversation_id", DataType::Utf8, false),
        Field::new("role", DataType::Utf8, false),
        Field::new("first_at_ms", DataType::UInt64, false),
    ]))
}

/// Decode [`ParticipationRow`]s into the `participations` table's Arrow
/// `RecordBatch` (`persona_id`, `conversation_id`, `role`, `first_at_ms`
/// columns, in [`participations_schema`] order).
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_participations_batch(
    rows: &[ParticipationRow],
) -> Result<RecordBatch, ArrowError> {
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut role_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut first_at_ms_b = UInt64Builder::with_capacity(rows.len());

    for row in rows {
        persona_id_b.append_value(&row.persona_id);
        conversation_id_b.append_value(&row.conversation_id);
        role_b.append_value(&row.role);
        first_at_ms_b.append_value(row.first_at_ms);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(persona_id_b.finish()),
        Arc::new(conversation_id_b.finish()),
        Arc::new(role_b.finish()),
        Arc::new(first_at_ms_b.finish()),
    ];
    RecordBatch::try_new(participations_schema(), columns)
}

/// One decoded `persona_identities` row: `persona_id` (supplied by the
/// caller — [`ExternalIdentity`] carries no `persona_id` of its own, the
/// same reason [`ParticipationRow::new`] takes it as a separate argument)
/// plus one [`PersonaProfile::identities`] element verbatim. The child table
/// for `personas`' `repeated ExternalIdentity identities` field — see the
/// parent module docs' "Column selection" section.
#[derive(Debug, Clone)]
pub(crate) struct PersonaIdentityRow {
    /// The persona this identity is linked to — this table's join key back
    /// to `personas`.
    pub persona_id: String,
    /// Edge namespace (`"slack"`, `"telegram"`, `"api-key"`, `"email"`, …).
    pub provider: String,
    /// Provider scope qualifying the id (e.g. the workspace/team id); empty
    /// for providers with globally-unique ids.
    pub scope: String,
    /// The provider-native stable identifier (never a mutable handle).
    pub external_id: String,
    /// Display name as last observed at the edge; presentation only.
    pub display_name: String,
}

impl PersonaIdentityRow {
    /// Pair `persona_id` (not carried on the stored [`ExternalIdentity`]
    /// value itself — see this struct's doc) with `identity`'s fields.
    #[must_use]
    pub(crate) fn new(persona_id: &str, identity: &ExternalIdentity) -> Self {
        Self {
            persona_id: persona_id.to_string(),
            provider: identity.provider.clone(),
            scope: identity.scope.clone(),
            external_id: identity.external_id.clone(),
            display_name: identity.display_name.clone(),
        }
    }
}

/// The `persona_identities` table's Arrow schema.
///
/// `persona_id` (`Utf8`, non-null), `provider` (`Utf8`, non-null), `scope`
/// (`Utf8`, non-null — empty string, not null, for a provider with
/// globally-unique ids, per `persona.proto`'s own field doc), `external_id`
/// (`Utf8`, non-null), `display_name` (`Utf8`, non-null — empty string, not
/// null, for an identity with no observed display name, the plain-proto3
/// zero-value convention `personas.display_name` also follows).
#[must_use]
pub(crate) fn persona_identities_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("provider", DataType::Utf8, false),
        Field::new("scope", DataType::Utf8, false),
        Field::new("external_id", DataType::Utf8, false),
        Field::new("display_name", DataType::Utf8, false),
    ]))
}

/// Decode [`PersonaIdentityRow`]s into the `persona_identities` table's
/// Arrow `RecordBatch` (`persona_id`, `provider`, `scope`, `external_id`,
/// `display_name` columns, in [`persona_identities_schema`] order).
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_persona_identities_batch(
    rows: &[PersonaIdentityRow],
) -> Result<RecordBatch, ArrowError> {
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut provider_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut scope_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut external_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut display_name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);

    for row in rows {
        persona_id_b.append_value(&row.persona_id);
        provider_b.append_value(&row.provider);
        scope_b.append_value(&row.scope);
        external_id_b.append_value(&row.external_id);
        display_name_b.append_value(&row.display_name);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(persona_id_b.finish()),
        Arc::new(provider_b.finish()),
        Arc::new(scope_b.finish()),
        Arc::new(external_id_b.finish()),
        Arc::new(display_name_b.finish()),
    ];
    RecordBatch::try_new(persona_identities_schema(), columns)
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;

    use super::*;

    fn sample_profile() -> PersonaProfile {
        PersonaProfile {
            persona_id: "persona-1".to_string(),
            display_name: "Alice".to_string(),
            created_at_ms: 1_000,
            status: "linked".to_string(),
            ..Default::default()
        }
    }

    /// A persona absorbed into a survivor via merge — `status == "merged"`
    /// and `merged_into` set to the survivor's id.
    fn sample_merged_profile() -> PersonaProfile {
        PersonaProfile {
            persona_id: "persona-absorbed".to_string(),
            display_name: "Alice (old)".to_string(),
            created_at_ms: 1_000,
            status: "merged".to_string(),
            merged_into: "persona-survivor".to_string(),
            ..Default::default()
        }
    }

    fn sample_participation() -> Participation {
        Participation {
            conversation_id: "conv-a".to_string(),
            role: "initiator".to_string(),
            first_at_ms: 2_000,
            ..Default::default()
        }
    }

    fn sample_identity() -> ExternalIdentity {
        ExternalIdentity {
            provider: "slack".to_string(),
            scope: "team-1".to_string(),
            external_id: "U-123".to_string(),
            display_name: "Alice".to_string(),
            ..Default::default()
        }
    }

    #[test]
    fn personas_schema_shape() {
        let schema = personas_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "persona_id",
                "display_name",
                "status",
                "created_at_ms",
                "merged_into",
                "merged_by",
                "merge_evidence",
                "merged_at_ms",
                "split_from",
                "split_by",
                "split_reason",
                "split_at_ms",
                "removed_by",
                "removed_at_ms",
                "admin_removed_by",
                "admin_removed_at_ms",
                "profile_language",
                "profile_tone",
                "profile_updated_at_ms",
                "profile_preferences",
                "permissions_tool_grants",
                "permissions_approval_policy",
                "permissions_turn_quota",
                "permissions_admin",
                "permissions_auto_review",
                "permissions_operator",
            ]
        );
        for field in schema.fields() {
            assert!(!field.is_nullable(), "{} must be non-null", field.name());
        }
    }

    #[test]
    fn participations_schema_shape() {
        let schema = participations_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec!["persona_id", "conversation_id", "role", "first_at_ms"]
        );
        for field in schema.fields() {
            assert!(!field.is_nullable(), "{} must be non-null", field.name());
        }
    }

    #[test]
    fn persona_identities_schema_shape() {
        let schema = persona_identities_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "persona_id",
                "provider",
                "scope",
                "external_id",
                "display_name"
            ]
        );
        for field in schema.fields() {
            assert!(!field.is_nullable(), "{} must be non-null", field.name());
        }
    }

    /// A known [`PersonaProfile`] round-trips through [`PersonaRow`] and
    /// [`decode_personas_batch`] to exactly its own row.
    #[test]
    fn persona_profile_round_trips_to_its_row() {
        let profile = sample_profile();
        let row = PersonaRow::from(&profile);
        let batch = decode_personas_batch(&[row]).expect("batch build");

        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), personas_schema());

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

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

        let status = batch
            .column(2)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(status.value(0), "linked");

        let created_at_ms = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(created_at_ms.value(0), 1_000);

        let merged_into = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(
            merged_into.value(0),
            "",
            "an unmerged persona's merged_into is empty, not null"
        );
    }

    /// A merged [`PersonaProfile`] (`status == "merged"`) round-trips its
    /// `merged_into` as the surviving persona's id — the G1 gap this table
    /// closes so a Fleet consumer can canonicalize an absorbed persona's
    /// data into its survivor without a second `resolve_persona_id`-style
    /// lookup.
    #[test]
    fn merged_persona_round_trips_its_survivor_id() {
        let profile = sample_merged_profile();
        let row = PersonaRow::from(&profile);
        let batch = decode_personas_batch(&[row]).expect("batch build");

        let status = batch
            .column(2)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(status.value(0), "merged");

        let merged_into = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(merged_into.value(0), "persona-survivor");
    }

    /// A [`PersonaProfile`] with no `profile` document set (`sample_profile`'s
    /// shape) decodes every lifecycle/audit/profile-document column to its
    /// type's zero-value sentinel — never null, per [`personas_schema`]'s
    /// non-null columns, and `"{}"`/`"[]"` (not empty string) for the two
    /// JSON-shaped columns.
    #[test]
    fn persona_profile_defaults_when_lifecycle_and_profile_are_absent() {
        let profile = sample_profile();
        let row = PersonaRow::from(&profile);
        let batch = decode_personas_batch(&[row]).expect("batch build");

        let string_col = |i: usize| {
            batch
                .column(i)
                .as_any()
                .downcast_ref::<arrow::array::StringArray>()
                .unwrap()
                .value(0)
                .to_owned()
        };
        let u64_col = |i: usize| {
            batch
                .column(i)
                .as_any()
                .downcast_ref::<arrow::array::UInt64Array>()
                .unwrap()
                .value(0)
        };

        assert_eq!(string_col(5), "", "merged_by");
        assert_eq!(string_col(6), "", "merge_evidence");
        assert_eq!(u64_col(7), 0, "merged_at_ms");
        assert_eq!(string_col(8), "", "split_from");
        assert_eq!(string_col(9), "", "split_by");
        assert_eq!(string_col(10), "", "split_reason");
        assert_eq!(u64_col(11), 0, "split_at_ms");
        assert_eq!(string_col(12), "", "removed_by");
        assert_eq!(u64_col(13), 0, "removed_at_ms");
        assert_eq!(string_col(14), "", "admin_removed_by");
        assert_eq!(u64_col(15), 0, "admin_removed_at_ms");
        assert_eq!(string_col(16), "", "profile_language");
        assert_eq!(string_col(17), "", "profile_tone");
        assert_eq!(u64_col(18), 0, "profile_updated_at_ms");
        assert_eq!(string_col(19), "{}", "profile_preferences");
        assert_eq!(string_col(20), "[]", "permissions_tool_grants");
        assert_eq!(string_col(21), "", "permissions_approval_policy");
        let turn_quota = batch
            .column(22)
            .as_any()
            .downcast_ref::<arrow::array::UInt32Array>()
            .unwrap()
            .value(0);
        assert_eq!(turn_quota, 0, "permissions_turn_quota");
        for (i, name) in [
            (23, "permissions_admin"),
            (24, "permissions_auto_review"),
            (25, "permissions_operator"),
        ] {
            let value = batch
                .column(i)
                .as_any()
                .downcast_ref::<arrow::array::BooleanArray>()
                .unwrap()
                .value(0);
            assert!(!value, "{name} defaults to false");
        }
    }

    /// A [`PersonaProfile`] with every lifecycle/audit field AND a full
    /// `profile` document (language, tone, preferences, and every
    /// `Permissions` field) set round-trips every one of those fields —
    /// the widening #1578 makes: `PersonaRow::from` no longer discards them.
    #[test]
    fn persona_profile_lifecycle_and_profile_document_round_trip() {
        use buffa::MessageField;
        use polyc_proto::proto::polychrome::persona::v1::{Permissions, ProfileDocument};

        let mut preferences = std::collections::HashMap::default();
        preferences.insert("timezone".to_string(), "UTC".to_string());

        let profile = PersonaProfile {
            persona_id: "persona-full".to_string(),
            display_name: "Ada".to_string(),
            created_at_ms: 1_000,
            status: "merged".to_string(),
            merged_into: "persona-survivor".to_string(),
            merged_by: "admin-1".to_string(),
            merge_evidence: "duplicate-signup".to_string(),
            merged_at_ms: 2_000,
            split_from: "persona-origin".to_string(),
            split_by: "admin-2".to_string(),
            split_reason: "org-split".to_string(),
            split_at_ms: 3_000,
            removed_by: "admin-3".to_string(),
            removed_at_ms: 4_000,
            admin_removed_by: "admin-4".to_string(),
            admin_removed_at_ms: 5_000,
            profile: MessageField::some(ProfileDocument {
                language: "es".to_string(),
                tone: "concise".to_string(),
                preferences,
                permissions: MessageField::some(Permissions {
                    tool_grants: vec!["fs.read".to_string(), "web.fetch".to_string()],
                    approval_policy: "always".to_string(),
                    turn_quota: 42,
                    admin: true,
                    auto_review: true,
                    operator: true,
                    ..Default::default()
                }),
                updated_at_ms: 6_000,
                ..Default::default()
            }),
            ..Default::default()
        };

        let row = PersonaRow::from(&profile);
        assert_eq!(row.merged_by, "admin-1");
        assert_eq!(row.merge_evidence, "duplicate-signup");
        assert_eq!(row.merged_at_ms, 2_000);
        assert_eq!(row.split_from, "persona-origin");
        assert_eq!(row.split_by, "admin-2");
        assert_eq!(row.split_reason, "org-split");
        assert_eq!(row.split_at_ms, 3_000);
        assert_eq!(row.removed_by, "admin-3");
        assert_eq!(row.removed_at_ms, 4_000);
        assert_eq!(row.admin_removed_by, "admin-4");
        assert_eq!(row.admin_removed_at_ms, 5_000);
        assert_eq!(row.profile_language, "es");
        assert_eq!(row.profile_tone, "concise");
        assert_eq!(row.profile_updated_at_ms, 6_000);
        assert_eq!(row.profile_preferences, r#"{"timezone":"UTC"}"#);
        assert_eq!(row.permissions_tool_grants, r#"["fs.read","web.fetch"]"#);
        assert_eq!(row.permissions_approval_policy, "always");
        assert_eq!(row.permissions_turn_quota, 42);
        assert!(row.permissions_admin);
        assert!(row.permissions_auto_review);
        assert!(row.permissions_operator);

        let batch = decode_personas_batch(&[row]).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), personas_schema());
    }

    /// A known `(persona_id, Participation)` pair round-trips through
    /// [`ParticipationRow`] and [`decode_participations_batch`] to exactly
    /// its own row.
    #[test]
    fn participation_round_trips_to_its_row() {
        let participation = sample_participation();
        let row = ParticipationRow::new("persona-1", &participation);
        let batch = decode_participations_batch(&[row]).expect("batch build");

        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), participations_schema());

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

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

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

        let first_at_ms = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(first_at_ms.value(0), 2_000);
    }

    /// A known `(persona_id, ExternalIdentity)` pair round-trips through
    /// [`PersonaIdentityRow`] and [`decode_persona_identities_batch`] to
    /// exactly its own row.
    #[test]
    fn persona_identity_round_trips_to_its_row() {
        let identity = sample_identity();
        let row = PersonaIdentityRow::new("persona-1", &identity);
        let batch = decode_persona_identities_batch(&[row]).expect("batch build");

        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), persona_identities_schema());

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

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

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

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

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