polyc-query 2026.8.3

//! `summary` typed-table decoder — `SummaryEvent` → Arrow columns.
//!
//! The tenth typed table (#1178 epic), decoded straight from
//! `polychrome.events.v1.SummaryEvent`'s two fields, `text` and
//! `covers_through_position` (`crates/proto/proto/events.proto:269-277`), a
//! DIRECT decode — not fold-coupled — through
//! [`crate::decode::decode_typed_kind_events`], the SAME shared loop
//! `usage`/`model_call`/`turn_failed` also call through (`attribution`/
//! `handoffs` moved to their own `polyc_facts` folds in #1579 instead — see
//! each module's own "Shared decode" section — since those two ALSO needed
//! a second, non-`polyc-query` consumer to fold-couple against; `summary`
//! has none).
//!
//! # Uniform keys (#1311)
//!
//! Every typed table carries the fact model's uniform key columns —
//! `partition`/`position`/`turn_id` — exactly as [`crate::decode::usage`]'s
//! module docs describe; see that module for the full rationale (why
//! `kind_base`/`conversation_id`/`event_time` are omitted). The one wrinkle
//! specific to `summary`: unlike `usage`'s `turn_id`, which names the real
//! turn that ran up the usage, a `summary` event's `kind` suffix is tagged
//! with a FRESH `Uuid::now_v7()` minted for the summary itself
//! (`crates/control-plane/src/grpc/summary.rs`'s `summary_id`) — it is not,
//! and is never meant to be, the id of any turn in the conversation. This
//! table's `turn_id` column therefore reflects that synthetic tag verbatim
//! (or `None` for a bare, untagged `summary` kind), not a real turn's
//! identity; a consumer joining `summary.turn_id` against another table's
//! `turn_id` expecting to land on a real turn will find nothing, by design.
//!
//! # `covers_through_position` is NOT a journal position (read this before using it)
//!
//! `SummaryEvent.covers_through_position` is a **count** of the leading
//! COMMITTED events the summary folds in, in the committed-turns' own
//! coordinate space — not a raw journal position
//! (`crates/control-plane/src/grpc/projection.rs`'s `project_full_messages`:
//! "`covers_through` is a COUNT of covered committed events, so the tail
//! begins at index `covers_through`"). It is included on this table for
//! fidelity to the stored payload, but a consumer wanting the transcript's
//! summary CUTOFF — "which events does this summary's own presence let me
//! drop" — must key on the summary ROW'S OWN `position` column (this row's
//! journal position), never on `covers_through_position`.
//! `crates/control-plane/src/forensics.rs`'s `build_transcript` is the
//! reference implementation: it locates the latest summary by its own
//! enumerated log position, discards the payload's
//! `covers_through_position` entirely (bound to `_stored` and never read),
//! and skips every event at or before that position. Using
//! `covers_through_position` for a position-based cutoff query (e.g.
//! `WHERE position > covers_through_position`) would silently compare two
//! different coordinate spaces and is a regression, not a valid query.
//!
//! # Fleet-only, not identity/signer redacted (QRY-1)
//!
//! `summary` text is a condensation of the persona's OWN conversation —
//! `crates/control-plane/src/grpc/summary.rs` folds only that partition's
//! own replayed history (`reconstruct_full`/`project_full_messages`), never
//! another conversation's — and carries no external identity or signer key,
//! so it needs none of `attribution`/`payments`' column redaction. It DOES
//! decode flat, identically for every scope, inside
//! `crate::engine::registration::register_typed_journal_tables` — the same
//! `usage`/`model_call`/`turn_failed` decode-and-batch treatment — but,
//! unlike those three, the resulting `summary_raw` `MemTable` backs a
//! `summary` `CREATE VIEW` built ONLY for
//! [`QueryScope::Fleet`](crate::session::QueryScope::Fleet): a
//! persona/conversation-scoped session must never see summary content, and
//! this table's own `turn_id` is a synthetic tag rather than a real turn's
//! id (see above), which rules out the committed-turn semijoin
//! `usage`/`model_call` use instead of a Fleet-only gate — see
//! `crate::engine`'s module docs' "Committed-turn filter invariant" section
//! and `crate::views`'s own section of the same name for the full
//! mechanism.

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::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::SummaryEvent;

/// One decoded `summary` row: the fact model's uniform key columns
/// (`partition`, `position`, `turn_id`) plus the decoded [`SummaryEvent`]
/// payload.
///
/// Produced by [`decode_summary_events`] and consumed by
/// [`decode_summary_batch`] — kept as an explicit intermediate type (rather
/// than building the `RecordBatch` directly) so a caller can inspect or
/// filter decoded rows before materializing Arrow columns.
#[derive(Debug, Clone)]
pub(crate) struct SummaryRow {
    /// 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 — the
    /// coordinate a transcript summary-cutoff query keys on (see the module
    /// docs' `covers_through_position` caveat).
    pub position: u64,
    /// The `:{turn_uuid}` suffix off the event's `kind`, canonical
    /// hyphenated form, or `None` for a bare `summary` kind with no tag —
    /// this is the summary's OWN synthetic `Uuid::now_v7()` tag, NOT a real
    /// turn id (see the module docs).
    pub turn_id: Option<String>,
    /// The decoded payload.
    pub summary: SummaryEvent,
}

/// The `summary` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable — the summary's own synthetic tag, not a real
/// turn id), `text` (`Utf8`, non-null), `covers_through_position`
/// (`UInt64`, non-null — see the module docs for why this is NOT a journal
/// position and must not be used for a position-based cutoff).
///
/// See the module docs for why `kind_base`/`conversation_id`/`event_time`
/// are deliberately omitted, matching [`crate::decode::usage::schema`].
#[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("text", DataType::Utf8, false),
        Field::new("covers_through_position", DataType::UInt64, false),
    ]))
}

/// Decode already-framed [`SummaryRow`]s into the `summary` table's Arrow
/// `RecordBatch` (`partition`, `position`, `turn_id`, `text`,
/// `covers_through_position` columns, in [`schema`] order).
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_summary_batch(rows: &[SummaryRow]) -> 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 text_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
    let mut covers_through_b = UInt64Builder::with_capacity(rows.len());

    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(),
        }
        text_b.append_value(&row.summary.text);
        covers_through_b.append_value(row.summary.covers_through_position);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(text_b.finish()),
        Arc::new(covers_through_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Filter `partition`'s framed `events` to kind-base `summary` rows, decode
/// each payload, and pair it with that row's key columns.
///
/// `partition` and each event's own `position` are stored verbatim;
/// `turn_id` is derived via [`polyc_proto::kinds::parse`] — the platform's
/// single kind-grammar owner, reused here exactly as
/// [`crate::decode::events_batch`] reuses it — so a `summary:{uuid}` kind
/// yields `turn_id = Some(uuid)` (the summary's own synthetic tag, not a
/// real turn id — see the module docs) and a bare `summary` kind yields
/// `turn_id = None`.
///
/// A non-empty payload that fails to decode is silently skipped — see
/// [`crate::decode::usage`]'s module docs for why that (rather than an
/// empty payload, which decodes cleanly to an all-defaults
/// `SummaryEvent`) is the corruption signal.
///
/// Delegates the filter/derive/decode loop to
/// `crate::decode::decode_typed_kind_events` (see that function's docs) —
/// only the `SummaryEvent` → [`SummaryRow`] mapping is unique to this
/// module.
#[must_use]
pub(crate) fn decode_summary_events(partition: &str, events: &[(u64, Event)]) -> Vec<SummaryRow> {
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[kinds::SUMMARY],
        "summary",
        try_decode_event_payload::<SummaryEvent>,
        |partition, position, turn_id, summary: SummaryEvent| SummaryRow {
            partition,
            position,
            turn_id,
            summary,
        },
    )
}

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

    use super::*;

    #[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",
                "text",
                "covers_through_position",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("text", DataType::Utf8, false),
            ("covers_through_position", DataType::UInt64, 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_summary_events_filters_and_decodes_real_buffa_bytes() {
        let summary_id = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let summary = SummaryEvent {
            text: "the older prefix, condensed".to_string(),
            covers_through_position: 12,
            ..Default::default()
        };
        let bytes = summary.encode_to_vec();

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

        let decoded = decode_summary_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(summary_id.to_string()));
        assert_eq!(decoded[0].summary.text, "the older prefix, condensed");
        assert_eq!(decoded[0].summary.covers_through_position, 12);
    }

    #[test]
    fn decode_summary_events_bare_kind_has_no_turn_id() {
        let summary = SummaryEvent {
            text: "bare".to_string(),
            covers_through_position: 3,
            ..Default::default()
        };
        let events = vec![(7, Event::new(kinds::SUMMARY, summary.encode_to_vec()))];
        let decoded = decode_summary_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 undecodable_non_empty_payload_is_skipped() {
        let good = SummaryEvent {
            text: "ok".to_string(),
            covers_through_position: 1,
            ..Default::default()
        };
        let events = vec![
            (1, Event::new(kinds::SUMMARY, vec![0xFF, 0xFE, 0xFD])),
            (2, Event::new(kinds::SUMMARY, good.encode_to_vec())),
        ];
        let decoded = decode_summary_events("conv-corrupt", &events);
        assert_eq!(decoded.len(), 1, "only the valid payload decodes");
        assert_eq!(decoded[0].position, 2);
        assert_eq!(decoded[0].summary.text, "ok");
    }

    #[test]
    fn decode_summary_batch_round_trips_multiple_rows() {
        let rows = vec![
            SummaryRow {
                partition: "conv-a".to_string(),
                position: 5,
                turn_id: None,
                summary: SummaryEvent {
                    text: "first summary".to_string(),
                    covers_through_position: 4,
                    ..Default::default()
                },
            },
            SummaryRow {
                partition: "conv-a".to_string(),
                position: 40,
                turn_id: Some("summary-xyz".to_string()),
                summary: SummaryEvent {
                    text: "second, later summary".to_string(),
                    covers_through_position: 30,
                    ..Default::default()
                },
            },
        ];
        let batch = decode_summary_batch(&rows).expect("batch build");
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.schema(), schema());

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

        let position = batch
            .column(1)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(position.values(), &[5, 40]);

        let turn_id = batch
            .column(2)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(turn_id.is_null(0));
        assert_eq!(turn_id.value(1), "summary-xyz");

        let text = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(text.value(0), "first summary");
        assert_eq!(text.value(1), "second, later summary");

        let covers_through = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(covers_through.values(), &[4, 30]);
    }

    #[test]
    fn empty_payload_summary_event_decodes_to_defaults() {
        let events = vec![(1, Event::new(kinds::SUMMARY, Vec::new()))];
        let decoded = decode_summary_events("conv-empty", &events);
        assert_eq!(decoded.len(), 1, "an empty payload must decode, not skip");
        assert_eq!(decoded[0].summary.text, String::new());
        assert_eq!(decoded[0].summary.covers_through_position, 0);
    }
}