polyc-query 2026.9.0

//! `dashboard` Fleet-only reference table — the dashboard maintained
//! projection ([`crate::dashboard::DashboardRow`], #1584) exposed through
//! the query engine's catalog (#1585, part of the read-path-convergence
//! epic #1574).
//!
//! # Reference data, not a journal decode
//!
//! Like [`crate::decode::persona`], this module decodes an already-typed
//! caller-supplied value ([`crate::dashboard::DashboardRow`], via
//! [`crate::engine::ReferenceData::dashboard`]) into Arrow — never a journal
//! event, and never [`crate::dashboard::DashboardProjection`] itself (this
//! crate's own `authority::ScopedQuery::execute` resolves the row snapshot
//! and hands it over already built, mirroring how [`crate::engine::PartitionEvents`]
//! is always a caller-supplied replay, never a journal handle this crate
//! opens itself).
//!
//! # Column selection
//!
//! Flat, one row per conversation, mirroring [`crate::dashboard::DashboardRow`]
//! field-for-field: `conversation_id`, `total_events`, `committed_turns`,
//! `input_tokens`, `output_tokens`, `summary_active`, `summary_text`,
//! `last_turn_id`, `created_at_ms`, `last_activity_ms`, `persona_id`,
//! `first_message_preview`. The two `repeated`-shaped fields —
//! [`crate::dashboard::DashboardRow::edges`] and
//! [`crate::dashboard::DashboardRow::settlements`] — are not normalized into
//! child tables (unlike `persona_identities`'s treatment of
//! [`polyc_proto::proto::polychrome::persona::v1::PersonaProfile::identities`]):
//! there is no existing per-conversation join key a caller would use to
//! re-join a child row back to this one beyond `conversation_id` itself,
//! which every row already carries, so a JSON-string column (the same
//! encoding [`crate::decode::grant_replays`]'s `covered_capabilities` and
//! [`crate::decode::handoffs`]'s `allowed` already use for a `repeated`
//! field) is queried via this crate's registered `json_get`/`->`/`->>`
//! functions rather than a `JOIN`. Both settlement totals are encoded as
//! decimal STRINGS inside that JSON, not JSON numbers — mirroring
//! [`crate::decode::payments`]'s own `amount` column — since a `u128` value
//! can exceed an IEEE-754 double's exact integer range and JSON numbers have
//! no wider integer type.
//!
//! `settlements_json` carries the two directions as separate keys, never a
//! sum: an outbound receipt is a caller's cost and an inbound one is this
//! deployment's revenue (see [`crate::dashboard::DashboardSettlement`]), so a
//! query that wants one asks for one.
//!
//! # Fleet-only, admin-gated (QRY-2-D, #1537)
//!
//! This table is registered ONLY for [`crate::session::QueryScope::Fleet`] —
//! never even resolvable for any other scope, the same "never registered"
//! posture [`PERSONAS_TABLE`](crate::engine::PERSONAS_TABLE)/
//! [`PARTICIPATIONS_TABLE`](crate::engine::PARTICIPATIONS_TABLE) use (see
//! `crate::engine`'s module docs' "Reference data" section), not the
//! "resolvable but redacted" posture `attribution`/`payments`/... use.
//! `summary_text`/`first_message_preview` need no column-level redaction
//! beyond that: [`crate::authority::QueryAuthority::scope_for`] mints
//! [`crate::session::QueryScope::Fleet`] ONLY for a verified
//! [`crate::authority::Principal::Admin`], never for a persona-scoped or
//! conversation-grant principal, so gating the whole table to Fleet scope
//! IS the admin gate — the identical posture
//! [`SUMMARY_TABLE`](crate::engine::SUMMARY_TABLE) already applies to its
//! own `summary_text` column, which this table also carries.

use std::sync::Arc;

use arrow::array::{ArrayRef, BooleanBuilder, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use serde::Serialize;

use crate::dashboard::DashboardRow;

/// One settlement entry, JSON-encoded into the `dashboard` table's
/// `settlements_json` column — see the module docs' "Column selection"
/// section for why both totals are strings.
#[derive(Serialize)]
struct SettlementJson {
    /// The `subject` the receipts name, or `"(unattributed)"`.
    persona: String,
    /// What settled out of this subject's own wallet, in the settlement
    /// token's base units, as a decimal string (a `u128` can exceed a JSON
    /// number's exact integer range).
    spend_base_units: String,
    /// What the deployment charged this subject, same unit and encoding.
    charged_base_units: String,
}

/// The `dashboard` table's Arrow schema, in column order.
///
/// `conversation_id` (`Utf8`, non-null — this table's join key back to the
/// event tables' `partition` column via `'conv-' ||
/// dashboard.conversation_id`, the same convention
/// `crate::decode::persona`'s module docs describe for `participations`),
/// `total_events`/`committed_turns`/`input_tokens`/`output_tokens`
/// (`UInt64`, non-null), `summary_active` (`Boolean`, non-null),
/// `summary_text` (`Utf8`, nullable), `last_turn_id` (`Utf8`, nullable),
/// `edges_json` (`Utf8`, non-null — a JSON array of edge provider strings,
/// `"[]"` when empty), `settlements_json` (`Utf8`, non-null — a JSON array of
/// [`SettlementJson`] objects, `"[]"` when empty), `created_at_ms`/
/// `last_activity_ms` (`UInt64`, nullable), `persona_id` (`Utf8`, nullable),
/// `first_message_preview` (`Utf8`, nullable).
#[must_use]
pub(crate) fn schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("conversation_id", DataType::Utf8, false),
        Field::new("total_events", DataType::UInt64, false),
        Field::new("committed_turns", DataType::UInt64, false),
        Field::new("input_tokens", DataType::UInt64, false),
        Field::new("output_tokens", DataType::UInt64, false),
        Field::new("summary_active", DataType::Boolean, false),
        Field::new("summary_text", DataType::Utf8, true),
        Field::new("last_turn_id", DataType::Utf8, true),
        Field::new("edges_json", DataType::Utf8, false),
        Field::new("settlements_json", DataType::Utf8, false),
        Field::new("created_at_ms", DataType::UInt64, true),
        Field::new("last_activity_ms", DataType::UInt64, true),
        Field::new("persona_id", DataType::Utf8, true),
        Field::new("first_message_preview", DataType::Utf8, true),
    ]))
}

/// Decode `rows` into the `dashboard` table's Arrow `RecordBatch`, in
/// [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_dashboard_batch(rows: &[DashboardRow]) -> Result<RecordBatch, ArrowError> {
    let mut conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut total_events_b = UInt64Builder::with_capacity(rows.len());
    let mut committed_turns_b = UInt64Builder::with_capacity(rows.len());
    let mut input_tokens_b = UInt64Builder::with_capacity(rows.len());
    let mut output_tokens_b = UInt64Builder::with_capacity(rows.len());
    let mut summary_active_b = BooleanBuilder::with_capacity(rows.len());
    let mut summary_text_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut last_turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut edges_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut settlements_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut created_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut last_activity_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut first_message_preview_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);

    for row in rows {
        conversation_id_b.append_value(&row.conversation_id);
        #[allow(clippy::cast_possible_truncation)] // usize -> u64: never truncates on any
        // platform this crate ships on (64-bit); mirrors `crate::decode::events_batch`'s own
        // `usize`-carrying columns.
        total_events_b.append_value(row.total_events as u64);
        #[allow(clippy::cast_possible_truncation)]
        committed_turns_b.append_value(row.committed_turns as u64);
        input_tokens_b.append_value(row.input_tokens);
        output_tokens_b.append_value(row.output_tokens);
        summary_active_b.append_value(row.summary_active());
        match &row.summary_text {
            Some(v) => summary_text_b.append_value(v),
            None => summary_text_b.append_null(),
        }
        match &row.last_turn_id {
            Some(v) => last_turn_id_b.append_value(v),
            None => last_turn_id_b.append_null(),
        }
        edges_json_b
            .append_value(serde_json::to_string(&row.edges).unwrap_or_else(|_| "[]".to_owned()));
        let settlements: Vec<SettlementJson> = row
            .settlements
            .iter()
            .map(|s| SettlementJson {
                persona: s.persona.clone(),
                spend_base_units: s.spend_base_units.to_string(),
                charged_base_units: s.charged_base_units.to_string(),
            })
            .collect();
        settlements_json_b
            .append_value(serde_json::to_string(&settlements).unwrap_or_else(|_| "[]".to_owned()));
        match row.created_at_ms {
            Some(v) => created_at_ms_b.append_value(v),
            None => created_at_ms_b.append_null(),
        }
        match row.last_activity_ms {
            Some(v) => last_activity_ms_b.append_value(v),
            None => last_activity_ms_b.append_null(),
        }
        match &row.persona_id {
            Some(v) => persona_id_b.append_value(v),
            None => persona_id_b.append_null(),
        }
        match &row.first_message_preview {
            Some(v) => first_message_preview_b.append_value(v),
            None => first_message_preview_b.append_null(),
        }
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(conversation_id_b.finish()),
        Arc::new(total_events_b.finish()),
        Arc::new(committed_turns_b.finish()),
        Arc::new(input_tokens_b.finish()),
        Arc::new(output_tokens_b.finish()),
        Arc::new(summary_active_b.finish()),
        Arc::new(summary_text_b.finish()),
        Arc::new(last_turn_id_b.finish()),
        Arc::new(edges_json_b.finish()),
        Arc::new(settlements_json_b.finish()),
        Arc::new(created_at_ms_b.finish()),
        Arc::new(last_activity_ms_b.finish()),
        Arc::new(persona_id_b.finish()),
        Arc::new(first_message_preview_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

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

    use arrow::array::{Array as _, BooleanArray, StringArray, UInt64Array};

    use super::*;
    use crate::dashboard::DashboardSettlement;

    fn sample_row() -> DashboardRow {
        DashboardRow {
            conversation_id: "conv-a".to_owned(),
            total_events: 9,
            committed_turns: 1,
            input_tokens: 10,
            output_tokens: 5,
            summary_text: Some("condensed".to_owned()),
            last_turn_id: Some("abc123".to_owned()),
            edges: vec!["web".to_owned(), "slack".to_owned()],
            settlements: vec![DashboardSettlement {
                persona: "persona-a".to_owned(),
                spend_base_units: 1_500,
                charged_base_units: 250,
            }],
            created_at_ms: Some(1_700_000_000_000),
            last_activity_ms: Some(1_700_000_001_000),
            persona_id: Some("persona-a".to_owned()),
            first_message_preview: Some("hello there".to_owned()),
        }
    }

    #[test]
    fn decode_dashboard_batch_round_trips_every_column() {
        let batch = decode_dashboard_batch(&[sample_row()]).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

        let col = |i: usize| batch.column(i).clone();
        let conversation_id = col(0);
        let conversation_id = conversation_id
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(conversation_id.value(0), "conv-a");

        let total_events = col(1);
        let total_events = total_events.as_any().downcast_ref::<UInt64Array>().unwrap();
        assert_eq!(total_events.value(0), 9);

        let summary_active = col(5);
        let summary_active = summary_active
            .as_any()
            .downcast_ref::<BooleanArray>()
            .unwrap();
        assert!(summary_active.value(0));

        let edges_json = col(8);
        let edges_json = edges_json.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(edges_json.value(0), r#"["web","slack"]"#);

        let settlements_json = col(9);
        let settlements_json = settlements_json
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        let decoded: serde_json::Value = serde_json::from_str(settlements_json.value(0)).unwrap();
        assert_eq!(decoded[0]["persona"], "persona-a");
        // Two keys, never one sum: `"1750"` would describe nothing.
        assert_eq!(decoded[0]["spend_base_units"], "1500");
        assert_eq!(decoded[0]["charged_base_units"], "250");

        let persona_id = col(12);
        let persona_id = persona_id.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(persona_id.value(0), "persona-a");
    }

    #[test]
    fn decode_dashboard_batch_renders_absent_fields_as_null_and_empty_lists_as_empty_json_arrays() {
        let row = DashboardRow {
            conversation_id: "conv-bare".to_owned(),
            ..Default::default()
        };
        let batch = decode_dashboard_batch(&[row]).expect("batch build");

        let summary_text = batch.column(6);
        let summary_text = summary_text.as_any().downcast_ref::<StringArray>().unwrap();
        assert!(summary_text.is_null(0));

        let created_at_ms = batch.column(10);
        let created_at_ms = created_at_ms
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap();
        assert!(created_at_ms.is_null(0));

        let edges_json = batch.column(8);
        let edges_json = edges_json.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(edges_json.value(0), "[]");

        let settlements_json = batch.column(9);
        let settlements_json = settlements_json
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(settlements_json.value(0), "[]");
    }

    #[test]
    fn decode_dashboard_batch_empty_input_has_zero_rows_and_the_full_schema() {
        let batch = decode_dashboard_batch(&[]).expect("batch build");
        assert_eq!(batch.num_rows(), 0);
        assert_eq!(batch.schema(), schema());
    }
}