polyc-query 2026.8.3

//! `model_call` typed-table decoder — `ModelCallEvent` → Arrow columns.
//!
//! The fact model's second typed table (#1311/#1178's "keyed
//! domain-fact-table" wave; see `crate::decode`'s module docs) —
//! `crate::decode::usage` established the pattern; this module mirrors it
//! for `polychrome.events.v1.ModelCallEvent`
//! (`crates/proto/proto/events.proto:137-168`), 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").
//!
//! # Uniform keys (#1311)
//!
//! Same discipline as [`crate::decode::usage`]: 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 (every row here
//! is `kind_base = "model_call"` by construction) and so is
//! `conversation_id` (`partition` already *is* `conv-{id}`) — see
//! [`crate::decode::usage`]'s module docs for the fuller rationale, which
//! applies unchanged here rather than being repeated.
//!
//! `partition`, `position`, and `turn_id` are derived exactly the way
//! [`crate::decode::events_batch`]/[`crate::decode::usage`] derive them,
//! via `crate::decode::decode_typed_kind_events` — the shared framing/
//! key-derivation loop this module and `usage` both call through, rather
//! than each carrying its own copy (see that function's docs for why it was
//! extracted once a second typed table needed the identical loop).
//!
//! # Shared decode (#1579)
//!
//! The `decode` argument [`decode_model_call_events`] passes to
//! `decode_typed_kind_events` is [`polyc_facts::fold_model_call_event`] —
//! the same primitive the control plane's replay-determinism record
//! decodes through (`crates/control-plane/src/grpc/mod.rs`'s
//! `decode_model_call_payload`), which needs the full
//! [`polyc_facts::ModelCallFact`] (decode params, clearing, persona-block
//! placement) this table deliberately does not surface — see "Column
//! selection" below. `decode_typed_kind_events` takes `decode` as a plain
//! `Fn(&[u8]) -> Result<T, E>` (not a `T: buffa::Message` bound it resolves
//! internally) precisely so this module and [`crate::decode::usage`] can
//! each plug in their own shared `polyc_facts` fold while still sharing the
//! loop's kind-filter/key-derivation/skip-and-warn shape.
//!
//! # `event_time` is deferred to #1327 — and `captured_clock_unix_ms` is NOT it
//!
//! `event_time` — a uniform attested timestamp every typed table will
//! eventually carry — is deliberately **not** included here, for the same
//! reason [`crate::decode::usage`] omits it: no such column exists on
//! `Event` yet, and inventing one (a fake decode-time clock read, or an
//! empty always-null column) would misrepresent the data. Tracked as future
//! work in #1327.
//!
//! This table's own `captured_clock_unix_ms` column (`UInt64`) MUST NOT be
//! confused with that future `event_time`, despite both being
//! millisecond-precision clock values. `captured_clock_unix_ms` is a
//! **determinism/replay input**: the harness's wall-clock read at turn
//! dispatch, frozen so a deterministic replay reconstructs time-dependent
//! prompt content (e.g. an injected current date) identically
//! (`events.proto:152-154`'s own field doc). It is turn-scoped — captured
//! once per dispatch by the harness's local clock, not independently
//! attested per event by the event log itself. `event_time` (#1327) will
//! instead be a uniform, attested append-time column every typed table
//! carries alike, sourced from the event log, not a caller-supplied payload
//! field. The two answer different questions — "what did the harness
//! believe the time was, for replay fidelity" vs. "when did the event log
//! actually record this row" — and must stay two columns, never collapsed
//! into one.
//!
//! # Column selection
//!
//! [`ModelCallEvent`] carries eight fields; this table's first cut keeps
//! only the general, reusable ones — the same "deliberate, reviewed
//! addition" discipline [`crate::decode::REGISTRY`] and
//! [`crate::decode::persona`]'s "Column selection" section apply:
//!
//! - `provider`, `model` — the provider/model identifiers a query workload
//!   actually wants to slice by (e.g. "which turns ran on which model").
//!   Both are plain proto3 strings; an empty value means the turn deferred
//!   to the harness's configured default rather than naming one explicitly
//!   (`events.proto:138-144`) — stored here as `""`, not `NULL`, matching
//!   that "empty means default" contract exactly rather than reinterpreting
//!   it as an absent value.
//! - `captured_clock_unix_ms` — see above.
//!
//! Left out of this first cut: `decode_params` (`temperature`/`top_p`/
//! `max_tokens`/`reasoning_level` — fine-grained generation knobs, each
//! paired with its own presence flag; nested and nullable-per-field, deeper
//! than a flat typed table's first pass needs), `system_config_ref` (a digest
//! reference into safety/system configuration this crate has no store to
//! resolve against, so the raw digest string alone is not independently
//! queryable), and `clear_trigger_bytes`/`clear_keep_recent`/`clear_marker`
//! (stale-tool-result clearing parameters — replay-determinism inputs, not
//! general query facts). Any of these can become its own column once a
//! query workload justifies it, same as [`crate::decode::persona`]'s
//! deferred fields.
//!
//! No token-count-shaped field exists on [`ModelCallEvent`] itself —
//! `input_tokens`/`output_tokens` are `usage`'s own columns
//! ([`crate::decode::usage`]), a separate event kind entirely.
//!
//! Decode quirk this module inherits from the shared bytes→struct decode
//! (`try_decode_event_payload`, via [`polyc_facts::fold_model_call_event`]):
//! an EMPTY payload decodes cleanly to [`ModelCallEvent`]'s all-defaults value
//! (proto3 elides all-default scalar fields) — `provider = ""`,
//! `model = ""`, `captured_clock_unix_ms = 0` — and is a legitimate row,
//! not a hole in the batch. A NON-empty payload that fails to decode is
//! corruption or a schema mismatch and is skipped, never surfaced as an
//! error; see [`crate::decode::usage`]'s module docs for the fuller
//! rationale.

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::kinds;

/// One decoded `model_call` row.
///
/// The fact model's uniform key columns (`partition`, `position`,
/// `turn_id`) plus the fields of
/// [`polyc_facts::ModelCallFact`](polyc_facts::ModelCallFact) this table's
/// first cut keeps (see the module docs' "Column selection").
#[derive(Debug, Clone)]
pub(crate) struct ModelCallRow {
    /// 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 `model_call` kind with no turn
    /// tagged.
    pub turn_id: Option<String>,
    /// The provider the turn's model belongs to. Empty when the turn
    /// deferred to the harness's configured default provider.
    pub provider: String,
    /// The model id the turn ran. Empty when the turn deferred to the
    /// harness's default model.
    pub model: String,
    /// Wall-clock value captured at turn dispatch, in Unix milliseconds — a
    /// determinism/replay input, NOT the uniform attested `event_time`
    /// #1327 will add (see the module docs).
    pub captured_clock_unix_ms: u64,
}

/// The `model_call` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `provider` (`Utf8`, non-null), `model`
/// (`Utf8`, non-null), `captured_clock_unix_ms` (`UInt64`, non-null).
///
/// See the module docs for why `kind_base`/`conversation_id` are omitted,
/// why `event_time` is deferred to #1327 rather than faked, and why
/// `captured_clock_unix_ms` is not that column.
#[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("provider", DataType::Utf8, false),
        Field::new("model", DataType::Utf8, false),
        Field::new("captured_clock_unix_ms", DataType::UInt64, false),
    ]))
}

/// Decode already-framed [`ModelCallRow`]s into the `model_call` table's
/// Arrow `RecordBatch`.
///
/// Columns are `partition`, `position`, `turn_id`, `provider`, `model`,
/// `captured_clock_unix_ms`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_model_call_batch(rows: &[ModelCallRow]) -> 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 provider_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut model_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut captured_clock_unix_ms_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(),
        }
        provider_b.append_value(&row.provider);
        model_b.append_value(&row.model);
        captured_clock_unix_ms_b.append_value(row.captured_clock_unix_ms);
    }

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

/// Filter `partition`'s framed `events` to kind-base `model_call` rows,
/// decode each payload, and pair it with that row's key columns.
///
/// Delegates the filter/derive/decode loop to
/// `crate::decode::decode_typed_kind_events` — see that function's docs and
/// [`crate::decode::usage::decode_usage_events`] (the pattern's first
/// instance) for the full behavior this reuses rather than re-implements:
/// `turn_id` derivation via [`polyc_proto::kinds::parse`], and a non-empty
/// undecodable payload skipped (logged) rather than erroring. Only the
/// [`polyc_facts::fold_model_call_event`] `decode` argument and the
/// [`polyc_facts::ModelCallFact`] → [`ModelCallRow`] field mapping (see the
/// module docs' "Column selection") are unique to this module.
#[must_use]
pub(crate) fn decode_model_call_events(
    partition: &str,
    events: &[(u64, Event)],
) -> Vec<ModelCallRow> {
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[kinds::MODEL_CALL],
        "model_call",
        polyc_facts::fold_model_call_event,
        |partition, position, turn_id, fact: polyc_facts::ModelCallFact| ModelCallRow {
            partition,
            position,
            turn_id,
            provider: fact.provider,
            model: fact.model,
            captured_clock_unix_ms: fact.captured_clock_unix_ms,
        },
    )
}

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

    use super::*;

    /// A [`ModelCallEvent`] naming only the three columns this table keeps —
    /// every other field explicit rather than `..Default::default()`, so a
    /// reader sees exactly what a real payload carries alongside them.
    /// `__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_model_call(
        provider: &str,
        model: &str,
        captured_clock_unix_ms: u64,
    ) -> ModelCallEvent {
        ModelCallEvent {
            provider: provider.to_string(),
            model: model.to_string(),
            decode_params: buffa::MessageField::default(),
            system_config_ref: String::new(),
            captured_clock_unix_ms,
            clear_trigger_bytes: 0,
            clear_keep_recent: 0,
            clear_marker: String::new(),
            persona_block_placement: 0,
            __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",
                "provider",
                "model",
                "captured_clock_unix_ms",
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("provider", DataType::Utf8, false),
            ("model", DataType::Utf8, false),
            ("captured_clock_unix_ms", 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_model_call_batch_round_trips() {
        let rows = vec![
            ModelCallRow {
                partition: "conv-a".to_string(),
                position: 5,
                turn_id: None,
                provider: "stub-provider".to_string(),
                model: "stub-model-a".to_string(),
                captured_clock_unix_ms: 1_700_000_000_000,
            },
            ModelCallRow {
                partition: "conv-a".to_string(),
                position: 9,
                turn_id: Some("turn-xyz".to_string()),
                provider: "alt-provider".to_string(),
                model: "stub-model-b".to_string(),
                captured_clock_unix_ms: 1_700_000_050_000,
            },
        ];
        let batch = decode_model_call_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, 9]);

        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), "turn-xyz");

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

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

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

    #[test]
    fn decode_model_call_events_filters_and_decodes_real_buffa_bytes() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let model_call = sample_model_call("stub-provider", "stub-model", 1_700_000_000_123);
        let bytes = model_call.encode_to_vec();

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

        let decoded = decode_model_call_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].provider, "stub-provider");
        assert_eq!(decoded[0].model, "stub-model");
        assert_eq!(decoded[0].captured_clock_unix_ms, 1_700_000_000_123);
    }

    #[test]
    fn decode_model_call_events_bare_kind_has_no_turn_id() {
        let events = vec![(7, Event::new(kinds::MODEL_CALL, Vec::new()))];
        let decoded = decode_model_call_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_model_call_event_decodes_to_defaults() {
        let events = vec![(1, Event::new(kinds::MODEL_CALL, Vec::new()))];
        let decoded = decode_model_call_events("conv-empty", &events);
        assert_eq!(decoded.len(), 1, "an empty payload must decode, not skip");
        assert_eq!(decoded[0].provider, "");
        assert_eq!(decoded[0].model, "");
        assert_eq!(decoded[0].captured_clock_unix_ms, 0);
    }

    #[test]
    fn undecodable_non_empty_payload_is_skipped() {
        let events = vec![
            (1, Event::new(kinds::MODEL_CALL, vec![0xFF, 0xFE, 0xFD])),
            (2, Event::new(kinds::MODEL_CALL, Vec::new())),
        ];
        let decoded = decode_model_call_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].provider, "");
        assert_eq!(decoded[0].model, "");
    }

    #[test]
    fn model_call_event_round_trip_through_decode_and_batch() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let model_call = sample_model_call("stub-provider", "stub-model", 1_700_000_099_999);
        let bytes = model_call.encode_to_vec();
        let events = vec![(
            1,
            Event::new(kinds::tagged(kinds::MODEL_CALL, &turn), bytes),
        )];

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

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

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

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

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

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

        let captured_clock = batch
            .column(5)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap();
        assert_eq!(captured_clock.value(0), 1_700_000_099_999);
    }
}