polyc-query 2026.8.3

//! `messages`/`tool_calls` typed-table decoder — the fact model's sixth and
//! seventh typed tables, and the first pair decoded through a SHARED
//! CONTENT-BLOCK FOLD rather than a plain per-kind protobuf decode.
//!
//! # Why this module decodes through `polyc_facts::fold_message_content`
//!
//! `docs/reference/datafusion-fact-model.md` names the "variant→block" class —
//! `user_msg`/`output_msg`'s single `Content` oneof per wire `Message` — as
//! "the single highest-volume slice of the log", and names
//! `crates/control-plane/src/trace.rs`'s `decode_message_steps` as the
//! fold's proof that it is extractable: that function already turns a
//! `Message`'s content block into `ToolCallRec`/`ToolResultRec` facts
//! independent of any UI. That trace-projector logic itself now calls the
//! ONE shared fold this module also calls —
//! [`polyc_facts::fold_message_content`] (`crates/facts/src/message_content.rs`)
//! — so this module is not a second, independently-written reimplementation
//! of "what does this message's content block mean"; it is the SAME fold's
//! second caller, exactly the guarantee the fact model exists to provide
//! (see that doc's "Problem: the same fact, folded twice, one wrong").
//! `crates/control-plane/src/grpc/approvals.rs`'s
//! `collect_pending_approval_responses` is the fold's third production
//! caller.
//!
//! [`decode_message_content_events`] therefore never matches on
//! `content::Type` itself — it calls [`polyc_facts::fold_message_content`]
//! once per decoded `Message` and only maps the returned
//! [`polyc_facts::MessageContent`] variant onto this crate's own row shapes.
//!
//! # One content-block fold, two SQL tables (not one)
//!
//! The fact model's doc describes the variant→block class as normalizing
//! into "one content-block table — a row per block, a discriminator column
//! for which variant it is". This module ships that one fold as TWO
//! sibling SQL tables rather than one, for query ergonomics: `messages`
//! (the [`polyc_facts::MessageContent::Text`] slice — a message's own
//! `role`/`internal_only` plus its full text) and `tool_calls` (the
//! [`polyc_facts::MessageContent::ToolCall`]/[`polyc_facts::MessageContent::ToolResult`]
//! slice, unioned into ONE table via a `block_type` discriminator column —
//! see "`tool_calls` unions calls and results" below). This split is a
//! decode/mod.rs `REGISTRY` matter, not a fold matter: see that module's
//! `REGISTRY` doc for how one kind's single `Decode::Typed` disposition
//! covers a fold whose output fans out into two table registrations in
//! `crate::engine`.
//!
//! [`polyc_facts::MessageContent::None`] — no content block, or one this
//! fold does not expand into a fact in v1 (a thought block, for example) —
//! produces NO row in either table, the same "nothing to project" contract
//! [`polyc_facts::MessageContent::None`]'s own doc states. This includes an
//! EMPTY event payload: unlike every earlier typed table in this crate
//! (`usage`/`model_call`/`attribution`/`turn_failed`), where an empty
//! payload decodes to a legitimate all-defaults row, an empty `Message`
//! payload decodes to a `Message` with no content block set at all, which
//! the fold folds to `MessageContent::None` — correctly zero rows, not a
//! defaults row, because there is no text/call/result fact an empty message
//! carries.
//!
//! # Which kinds drive the fold
//!
//! Two kinds carry a `Content` block: `user_msg` and `output_msg`. Both are
//! folded here. [`MESSAGE_CONTENT_KIND_BASES`] lists exactly those two, and
//! `docs/reference/datafusion-fact-model.md`'s variant→block class names the
//! same pair. Every production caller
//! (`crates/control-plane/src/trace.rs`'s `decode_message_steps` and
//! `crates/control-plane/src/grpc/approvals.rs`'s two call sites) folds only
//! those two kinds.
//!
//! # Uniform keys (#1311)
//!
//! Same discipline as every earlier typed table: `partition`, `position`,
//! and `turn_id` are derived via [`polyc_proto::kinds::parse`], exactly as
//! [`crate::decode::events_batch`] derives them. Unlike every earlier typed
//! table, `position` and `turn_id` are not re-derived independently here —
//! they are the SAME values [`polyc_facts::fold_message_content`] stamps
//! onto its own [`polyc_facts::TextFact`]/[`polyc_facts::ToolCallFact`]/
//! [`polyc_facts::ToolResultFact`] output, because the fact model's uniform
//! key discipline (docs/reference/datafusion-fact-model.md, "Grain and
//! synthesized keys") requires every fact struct to carry its own identity
//! key rather than making the caller re-derive it. `event_time` is deferred
//! to #1327, unmodified from every other table's rationale.
//!
//! # `role`/`internal_only`: read off `Message` directly, not through the fold
//!
//! [`polyc_facts::fold_message_content`] returns a content-block FACT — it
//! deliberately does not carry the wire [`Message`]'s own `role` or
//! `internal_only` scalar fields, since those describe the MESSAGE, not the
//! block folded out of it. This module reads both directly off the already
//! -decoded `Message` (via the crate's one canonical
//! [`polyc_proto::events_decode::try_decode_event_payload`] path) alongside
//! calling the fold — this is not a second decode of CONTENT (the fold
//! remains the only place that interprets `content::Type`), only two plain
//! scalar field reads off a struct already decoded once.
//!
//! `role` is `Message.role` verbatim (`"user"`/`"assistant"`/`"model"`/…,
//! `crates/proto/proto/agent_service.proto:335-343`'s own doc comment) — NOT
//! `crates/control-plane/src/trace.rs`'s `step_type` normalization
//! (`"user_input"`/`"model_output"`, derived from which KIND the row came
//! from). That normalization is presentation logic for one projector's step
//! list, the kind of thing docs/reference/datafusion-fact-model.md's
//! decode-vs-project boundary keeps out of the fact layer; `role` is the
//! smaller, already-on-the-wire fact this table exposes instead, matching
//! this module's "prefer the smaller set trace.rs already projects" brief.
//!
//! `internal_only` is kept as its own column — not dropped — because it
//! drives this table's redaction posture; see "Redaction" below.
//!
//! # `tool_calls` unions calls and results
//!
//! A tool call and its result are ALWAYS two separate wire `Message`s (the
//! model emits the call in one `output_msg`; the executed result lands in a
//! later message), so [`decode_message_content_events`] never merges them
//! into one row — doing so would require the FIFO pairing
//! `crates/control-plane/src/trace.rs`'s `build_turns` performs (nearest-preceding
//! match, because a provider can reuse a `tool_call_id` across turns), which
//! is presentation-level projection, not mechanical fact extraction (see
//! the fact model doc's decode-vs-project test). Instead, `tool_calls`
//! carries ONE ROW PER BLOCK — a `"call"`-typed row from
//! [`polyc_facts::MessageContent::ToolCall`], a `"result"`-typed row from
//! [`polyc_facts::MessageContent::ToolResult`] — sharing the SAME
//! `tool_call_id` join key, exactly the "a discriminator column for which
//! variant it is, and `tool_call_id` as the join key" shape
//! docs/reference/datafusion-fact-model.md's variant→block section names. A
//! caller pairs a call to its result with an ordinary SQL self-join on
//! `tool_call_id`, the same join key `crate::decode::attribution`'s `role`
//! column precedent already establishes for "two shapes, one table,
//! disambiguated by a column" (there: `caller`/`participant`; here:
//! `"call"`/`"result"`).
//!
//! `arguments` (populated only for `"call"` rows) and `result` (populated
//! only for `"result"` rows) are both `Utf8` columns holding the FULL
//! `serde_json::to_string` of the fact's `serde_json::Value`, following
//! [`crate::decode::events_batch`]'s own `payload_json` precedent for
//! typing a schema-less JSON payload as `Utf8` rather than a nested Arrow
//! struct column — this crate's pinned `datafusion = "=54.0.0"` ships no
//! `json_get`/`->`-style extraction function (see that module's docs), so a
//! caller that wants a specific field parses the returned text client-side
//! either way. `first_party` is populated only for `"result"` rows
//! ([`polyc_facts::ToolResultFact::first_party`] has no equivalent on a
//! call); `NULL` on a `"call"` row, never `false`, so a caller cannot
//! mistake "not applicable to this block type" for "observed false".
//!
//! # Redaction: `internal_only` rows are Fleet-only
//!
//! `Message.internal_only`'s own wire doc states the message "is recorded in
//! the execution log (for resumption) but is not part of the conversation
//! history and is not emitted to clients"
//! (`crates/proto/proto/agent_service.proto:340-342`). That is precisely
//! docs/reference/datafusion-data-layer.md's boundary condition for what a
//! persona-scoped session may see: "the persona-scoped `payload` column is
//! bounded to the fields the existing transcript and approval surfaces
//! already show a participant... This redaction applies to every
//! non-maintainer registration — the conversation-scoped agent tool
//! included." An `internal_only` message is BY DEFINITION never shown on
//! the transcript/approval surface a participant's own client renders, so
//! exposing it to a `QueryScope::Conversations` session would show that
//! session MORE than its own existing transcript surface already shows it —
//! exactly the leak the boundary condition forbids. Every OTHER field on
//! both tables (full tool-call arguments, full tool-result payloads, plain
//! message text) is already within that boundary:
//! `crates/control-plane/src/forensics.rs`'s `wire_text`/`format_tool_call`/
//! `format_tool_result` already render a committed turn's full (if
//! display-truncated) tool arguments/results on the classic transcript, so
//! this table's full, untruncated columns are the same fact the transcript
//! already shows a participant, just untruncated — narrower text-length
//! truncation is presentation, not a new disclosure.
//!
//! Mechanism: [`crate::engine`] registers this module's full decode as
//! `messages_raw`/`tool_calls_raw` for every scope (needed to build the
//! views below), builds `messages`/`tool_calls` as `CREATE VIEW`s whose row
//! set is scope-dependent — every row for
//! [`Fleet`](crate::session::QueryScope::Fleet), every row EXCEPT
//! `internal_only = true` for every other scope — then deregisters
//! `messages_raw`/`tool_calls_raw` for every non-Fleet scope. This is a ROW
//! filter, not [`crate::decode::attribution`]'s COLUMN filter (there is no
//! column here that is itself raw identity data the way `identity_*` is);
//! `internal_only` stays present and visible in every scope's view — it is
//! always `false` for a non-Fleet row by construction of the filter, not a
//! second raw-identity-shaped column that needs hiding on its own. The
//! precedent for a ROW-level (not column-level) redaction view is
//! [`crate::views::COMMITTED_TURNS_VIEW_SQL`], which already filters
//! `events_raw` down to committed-turn rows the same way.

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 polyc_eventlog::Event;
use polyc_proto::events_decode::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::agent::v1::Message;

/// `block_type`'s value for a row folded from
/// [`polyc_facts::MessageContent::ToolCall`].
const BLOCK_TYPE_CALL: &str = "call";

/// `block_type`'s value for a row folded from
/// [`polyc_facts::MessageContent::ToolResult`].
const BLOCK_TYPE_RESULT: &str = "result";

/// The kind-bases [`decode_message_content_events`] folds — see the module
/// docs' "Which kinds drive the fold" section.
const MESSAGE_CONTENT_KIND_BASES: &[&str] = &[kinds::USER_MSG, kinds::OUTPUT_MSG];

/// One decoded `messages` row — a [`polyc_facts::MessageContent::Text`]
/// block plus the fact model's uniform key columns and the owning
/// `Message`'s own `role`/`internal_only` scalars (see the module docs'
/// "`role`/`internal_only`" section for why those two are read directly off
/// `Message` rather than through the fold).
#[derive(Debug, Clone)]
pub(crate) struct MessageRow {
    /// 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 `user_msg`/`output_msg` kind
    /// with no turn tagged.
    pub turn_id: Option<String>,
    /// [`Message::role`] verbatim — NOT `crates/control-plane/src/trace.rs`'s
    /// `step_type` normalization; see the module docs.
    pub role: String,
    /// [`Message::internal_only`] — drives this table's redaction posture;
    /// see the module docs' "Redaction" section.
    pub internal_only: bool,
    /// [`polyc_facts::TextFact::text`] — full, untruncated text.
    pub text: String,
    /// [`polyc_facts::TextFact::trust`].
    pub trust: String,
}

/// One decoded `tool_calls` row — either a
/// [`polyc_facts::MessageContent::ToolCall`] or a
/// [`polyc_facts::MessageContent::ToolResult`] block, disambiguated by
/// `block_type` (see the module docs' "`tool_calls` unions calls and
/// results" section).
#[derive(Debug, Clone)]
pub(crate) struct ToolCallRow {
    /// 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 `user_msg`/`output_msg` kind
    /// with no turn tagged.
    pub turn_id: Option<String>,
    /// Cross-message join key correlating a call to its result (and, one
    /// join further, its approval/settlement) — see the module docs.
    pub tool_call_id: String,
    /// `"call"` or `"result"` — which fold variant this row came from.
    pub block_type: &'static str,
    /// Tool name. Empty when the block carried no `function_call`/
    /// `function_result`.
    pub name: String,
    /// Full call arguments, `serde_json::to_string`'d. `Some` only for a
    /// `"call"` row, `None` for a `"result"` row.
    pub arguments: Option<String>,
    /// Full result value, `serde_json::to_string`'d. `Some` only for a
    /// `"result"` row, `None` for a `"call"` row.
    pub result: Option<String>,
    /// [`polyc_facts::ToolResultFact::first_party`]. `Some` only for a
    /// `"result"` row, `None` (never `false`) for a `"call"` row — see the
    /// module docs for why `NULL`, not `false`, marks "not applicable".
    pub first_party: Option<bool>,
    /// [`Message::internal_only`] — drives this table's redaction posture;
    /// see the module docs' "Redaction" section.
    pub internal_only: bool,
    /// Trust tag carried by the fact ([`polyc_facts::ToolCallFact::trust`]/
    /// [`polyc_facts::ToolResultFact::trust`]).
    pub trust: String,
}

/// Both tables' rows decoded from one partition's events in a single pass —
/// see [`decode_message_content_events`].
#[derive(Debug, Clone, Default)]
pub(crate) struct MessageContentRows {
    /// Every `messages` row this partition's events fold to.
    pub messages: Vec<MessageRow>,
    /// Every `tool_calls` row this partition's events fold to.
    pub tool_calls: Vec<ToolCallRow>,
}

/// The `messages` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `role` (`Utf8`, non-null), `internal_only`
/// (`Boolean`, non-null), `text` (`Utf8`, non-null), `trust` (`Utf8`,
/// non-null).
#[must_use]
pub(crate) fn messages_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("role", DataType::Utf8, false),
        Field::new("internal_only", DataType::Boolean, false),
        Field::new("text", DataType::Utf8, false),
        Field::new("trust", DataType::Utf8, false),
    ]))
}

/// The `tool_calls` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `tool_call_id` (`Utf8`, non-null),
/// `block_type` (`Utf8`, non-null), `name` (`Utf8`, non-null), `arguments`
/// (`Utf8`, nullable), `result` (`Utf8`, nullable), `first_party`
/// (`Boolean`, nullable), `internal_only` (`Boolean`, non-null), `trust`
/// (`Utf8`, non-null).
#[must_use]
pub(crate) fn tool_calls_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("tool_call_id", DataType::Utf8, false),
        Field::new("block_type", DataType::Utf8, false),
        Field::new("name", DataType::Utf8, false),
        Field::new("arguments", DataType::Utf8, true),
        Field::new("result", DataType::Utf8, true),
        Field::new("first_party", DataType::Boolean, true),
        Field::new("internal_only", DataType::Boolean, false),
        Field::new("trust", DataType::Utf8, false),
    ]))
}

/// Decode already-framed [`MessageRow`]s into the `messages` table's Arrow
/// `RecordBatch`, in [`messages_schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_messages_batch(rows: &[MessageRow]) -> 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 role_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut internal_only_b = BooleanBuilder::with_capacity(rows.len());
    let mut text_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
    let mut trust_b = StringBuilder::with_capacity(rows.len(), rows.len() * 20);

    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(),
        }
        role_b.append_value(&row.role);
        internal_only_b.append_value(row.internal_only);
        text_b.append_value(&row.text);
        trust_b.append_value(&row.trust);
    }

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

/// Decode already-framed [`ToolCallRow`]s into the `tool_calls` table's
/// Arrow `RecordBatch`, in [`tool_calls_schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_tool_calls_batch(rows: &[ToolCallRow]) -> 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 tool_call_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut block_type_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut arguments_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut result_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut first_party_b = BooleanBuilder::with_capacity(rows.len());
    let mut internal_only_b = BooleanBuilder::with_capacity(rows.len());
    let mut trust_b = StringBuilder::with_capacity(rows.len(), rows.len() * 20);

    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(),
        }
        tool_call_id_b.append_value(&row.tool_call_id);
        block_type_b.append_value(row.block_type);
        name_b.append_value(&row.name);
        match &row.arguments {
            Some(args) => arguments_b.append_value(args),
            None => arguments_b.append_null(),
        }
        match &row.result {
            Some(result) => result_b.append_value(result),
            None => result_b.append_null(),
        }
        match row.first_party {
            Some(first_party) => first_party_b.append_value(first_party),
            None => first_party_b.append_null(),
        }
        internal_only_b.append_value(row.internal_only);
        trust_b.append_value(&row.trust);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(tool_call_id_b.finish()),
        Arc::new(block_type_b.finish()),
        Arc::new(name_b.finish()),
        Arc::new(arguments_b.finish()),
        Arc::new(result_b.finish()),
        Arc::new(first_party_b.finish()),
        Arc::new(internal_only_b.finish()),
        Arc::new(trust_b.finish()),
    ];
    RecordBatch::try_new(tool_calls_schema(), columns)
}

/// Best-effort JSON serialization of a fact's `serde_json::Value` field —
/// `"null"` on the (practically unreachable) serialize failure, matching
/// `crates/control-plane/src/trace.rs`'s own `args_json_full` fallback pattern for
/// the identical fold output.
fn json_string(value: &serde_json::Value) -> String {
    serde_json::to_string(value).unwrap_or_else(|_| "null".to_string())
}

/// Filter `partition`'s events to [`MESSAGE_CONTENT_KIND_BASES`], decode
/// each payload as a wire [`Message`], fold its content block through the
/// SHARED [`polyc_facts::fold_message_content`] fold, and sort the result
/// into `messages`/`tool_calls` rows in one pass.
///
/// Unlike [`crate::decode::decode_typed_kind_events`], this cannot delegate
/// to that shared loop: `decode_typed_kind_events`'s `build` closure returns
/// exactly one `R` per decoded payload, but one decoded `Message` here
/// fans out into a `MessageRow`, a `ToolCallRow`, or NO row at all
/// (`MessageContent::None`) — see the module docs. This function still
/// reuses [`polyc_proto::kinds::parse`] for the same key derivation
/// `decode_typed_kind_events` performs, and applies the identical
/// "non-empty undecodable payload is skipped and logged, never surfaced as
/// an error" contract — a payload that fails to decode as [`Message`] is
/// skipped (logged via `tracing::warn!` when non-empty), never surfaced to
/// the caller as an error.
#[must_use]
pub(crate) fn decode_message_content_events(
    partition: &str,
    events: &[(u64, Event)],
) -> MessageContentRows {
    let mut out = MessageContentRows::default();

    for (position, event) in events {
        let (base, turn_uuid) = kinds::parse(&event.kind);
        if !MESSAGE_CONTENT_KIND_BASES.contains(&base) {
            continue;
        }
        let turn_id = turn_uuid.map(|id| id.to_string());

        let message = match try_decode_event_payload::<Message>(&event.payload) {
            Ok(message) => message,
            Err(e) => {
                if !event.payload.is_empty() {
                    tracing::warn!(
                        error = %e,
                        len = event.payload.len(),
                        table = "messages/tool_calls",
                        "corrupt message payload; skipping row"
                    );
                }
                continue;
            }
        };
        let internal_only = message.internal_only;
        let trust = event.trust.as_str();

        let folded =
            polyc_facts::fold_message_content(&message, *position, turn_id.as_deref(), trust);
        for warning in folded.warnings {
            tracing::warn!(table = "tool_calls", %warning, "message-content fold warning");
        }

        match folded.content {
            polyc_facts::MessageContent::Text(t) => {
                out.messages.push(MessageRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    role: message.role,
                    internal_only,
                    text: t.text,
                    trust: t.trust,
                });
            }
            polyc_facts::MessageContent::ToolCall(call) => {
                out.tool_calls.push(ToolCallRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    tool_call_id: call.tool_call_id,
                    block_type: BLOCK_TYPE_CALL,
                    name: call.name,
                    arguments: Some(json_string(&call.arguments)),
                    result: None,
                    first_party: None,
                    internal_only,
                    trust: call.trust,
                });
            }
            polyc_facts::MessageContent::ToolResult(res) => {
                out.tool_calls.push(ToolCallRow {
                    partition: partition.to_string(),
                    position: *position,
                    turn_id,
                    tool_call_id: res.tool_call_id,
                    block_type: BLOCK_TYPE_RESULT,
                    name: res.name,
                    arguments: None,
                    result: Some(json_string(&res.result)),
                    first_party: Some(res.first_party),
                    internal_only,
                    trust: res.trust,
                });
            }
            polyc_facts::MessageContent::None => {}
        }
    }

    out
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use buffa::Message as _;
    use polyc_proto::proto::polychrome::agent::v1::{
        Content, FunctionCallContent, FunctionResultContent, TextContent, ToolCallContent,
        ToolResultContent, content, function_result_content, tool_call_content,
        tool_result_content,
    };
    use serde_json::json;
    use uuid::Uuid;

    use super::*;

    fn text_message(role: &str, text: &str, internal_only: bool) -> Message {
        Message {
            role: role.to_string(),
            content: buffa::MessageField::some(Content {
                r#type: Some(content::Type::Text(Box::new(TextContent {
                    text: text.to_string(),
                    ..Default::default()
                }))),
                ..Default::default()
            }),
            internal_only,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn tool_call_message(id: &str, name: &str, args: serde_json::Value) -> Message {
        let arguments = serde_json::from_value::<buffa_types::google::protobuf::Struct>(args)
            .map(buffa::MessageField::some)
            .unwrap_or_default();
        Message {
            role: "model".to_string(),
            content: buffa::MessageField::some(Content {
                r#type: Some(content::Type::ToolCall(Box::new(ToolCallContent {
                    id: id.to_string(),
                    r#type: Some(tool_call_content::Type::FunctionCall(Box::new(
                        FunctionCallContent {
                            name: name.to_string(),
                            arguments,
                            ..Default::default()
                        },
                    ))),
                    ..Default::default()
                }))),
                ..Default::default()
            }),
            internal_only: false,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    fn tool_result_message(
        id: &str,
        name: &str,
        result: serde_json::Value,
        first_party: bool,
    ) -> Message {
        let response = serde_json::from_value::<buffa_types::google::protobuf::Struct>(result)
            .ok()
            .map(|s| function_result_content::Result::Response(Box::new(s)));
        Message {
            role: "tool".to_string(),
            content: buffa::MessageField::some(Content {
                r#type: Some(content::Type::ToolResult(Box::new(ToolResultContent {
                    call_id: id.to_string(),
                    first_party,
                    r#type: Some(tool_result_content::Type::FunctionResult(Box::new(
                        FunctionResultContent {
                            name: name.to_string(),
                            result: response,
                            ..Default::default()
                        },
                    ))),
                    ..Default::default()
                }))),
                ..Default::default()
            }),
            internal_only: false,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    #[test]
    fn messages_schema_shape() {
        let schema = messages_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "partition",
                "position",
                "turn_id",
                "role",
                "internal_only",
                "text",
                "trust",
            ]
        );
        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("role", DataType::Utf8, false),
            ("internal_only", DataType::Boolean, false),
            ("text", DataType::Utf8, false),
            ("trust", 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 tool_calls_schema_shape() {
        let schema = tool_calls_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "partition",
                "position",
                "turn_id",
                "tool_call_id",
                "block_type",
                "name",
                "arguments",
                "result",
                "first_party",
                "internal_only",
                "trust",
            ]
        );
        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("tool_call_id", DataType::Utf8, false),
            ("block_type", DataType::Utf8, false),
            ("name", DataType::Utf8, false),
            ("arguments", DataType::Utf8, true),
            ("result", DataType::Utf8, true),
            ("first_party", DataType::Boolean, true),
            ("internal_only", DataType::Boolean, false),
            ("trust", 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_messages_batch_round_trips() {
        let rows = vec![MessageRow {
            partition: "conv-a".to_string(),
            position: 5,
            turn_id: Some("turn-xyz".to_string()),
            role: "user".to_string(),
            internal_only: false,
            text: "hello".to_string(),
            trust: "trusted_user".to_string(),
        }];
        let batch = decode_messages_batch(&rows).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), messages_schema());

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

    #[test]
    fn decode_tool_calls_batch_round_trips_a_call_and_a_result() {
        let rows = vec![
            ToolCallRow {
                partition: "conv-a".to_string(),
                position: 5,
                turn_id: Some("turn-xyz".to_string()),
                tool_call_id: "call-1".to_string(),
                block_type: BLOCK_TYPE_CALL,
                name: "search".to_string(),
                arguments: Some(r#"{"q":"a"}"#.to_string()),
                result: None,
                first_party: None,
                internal_only: false,
                trust: "trusted_user".to_string(),
            },
            ToolCallRow {
                partition: "conv-a".to_string(),
                position: 6,
                turn_id: Some("turn-xyz".to_string()),
                tool_call_id: "call-1".to_string(),
                block_type: BLOCK_TYPE_RESULT,
                name: "search".to_string(),
                arguments: None,
                result: Some(r#"{"hits":1}"#.to_string()),
                first_party: Some(true),
                internal_only: false,
                trust: "trusted_user".to_string(),
            },
        ];
        let batch = decode_tool_calls_batch(&rows).expect("batch build");
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.schema(), tool_calls_schema());

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

        let arguments = batch
            .column(6)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(arguments.value(0), r#"{"q":"a"}"#);
        assert!(arguments.is_null(1));

        let result = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(result.is_null(0));
        assert_eq!(result.value(1), r#"{"hits":1}"#);

        let first_party = batch
            .column(8)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(first_party.is_null(0));
        assert!(first_party.value(1));
    }

    #[test]
    fn text_message_decodes_to_a_messages_row_with_role_and_internal_only() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let message = text_message("user", "hi there", false);
        let events = vec![(
            1,
            Event::trusted(
                kinds::tagged(kinds::USER_MSG, &turn),
                message.encode_to_vec(),
            ),
        )];

        let rows = decode_message_content_events("conv-real", &events);
        assert_eq!(rows.messages.len(), 1);
        assert!(rows.tool_calls.is_empty());
        let row = &rows.messages[0];
        assert_eq!(row.partition, "conv-real");
        assert_eq!(row.position, 1);
        assert_eq!(row.turn_id, Some(turn.to_string()));
        assert_eq!(row.role, "user");
        assert!(!row.internal_only);
        assert_eq!(row.text, "hi there");
        assert_eq!(row.trust, polyc_eventlog::TrustTag::TrustedUser.as_str());
    }

    #[test]
    fn internal_only_flag_is_carried_onto_the_row() {
        let message = text_message("model", "ground truth note", true);
        let events = vec![(1, Event::new(kinds::OUTPUT_MSG, message.encode_to_vec()))];
        let rows = decode_message_content_events("conv-internal", &events);
        assert_eq!(rows.messages.len(), 1);
        assert!(rows.messages[0].internal_only);
    }

    #[test]
    fn tool_call_message_decodes_to_a_tool_calls_row() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let message = tool_call_message("call-1", "search", json!({"q": "a"}));
        let events = vec![(
            2,
            Event::new(
                kinds::tagged(kinds::OUTPUT_MSG, &turn),
                message.encode_to_vec(),
            ),
        )];

        let rows = decode_message_content_events("conv-real", &events);
        assert!(rows.messages.is_empty());
        assert_eq!(rows.tool_calls.len(), 1);
        let row = &rows.tool_calls[0];
        assert_eq!(row.tool_call_id, "call-1");
        assert_eq!(row.block_type, "call");
        assert_eq!(row.name, "search");
        assert_eq!(row.arguments.as_deref(), Some(r#"{"q":"a"}"#));
        assert_eq!(row.result, None);
        assert_eq!(row.first_party, None);
    }

    #[test]
    fn tool_result_message_decodes_to_a_tool_calls_row_paired_by_id() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let call = tool_call_message("call-1", "search", json!({"q": "a"}));
        let result = tool_result_message("call-1", "search", json!({"hits": 1}), true);

        let events = vec![
            (
                2,
                Event::new(
                    kinds::tagged(kinds::OUTPUT_MSG, &turn),
                    call.encode_to_vec(),
                ),
            ),
            (
                3,
                Event::new(
                    kinds::tagged(kinds::USER_MSG, &turn),
                    result.encode_to_vec(),
                ),
            ),
        ];

        let rows = decode_message_content_events("conv-paired", &events);
        assert_eq!(rows.tool_calls.len(), 2);
        assert_eq!(rows.tool_calls[0].tool_call_id, "call-1");
        assert_eq!(rows.tool_calls[0].block_type, "call");
        assert_eq!(rows.tool_calls[1].tool_call_id, "call-1");
        assert_eq!(rows.tool_calls[1].block_type, "result");
        // `google.protobuf.Struct`'s `number_value` is a double, so the
        // integer JSON literal above round-trips as a float — the same
        // quirk `polyc_facts::message_content`'s own tests already document.
        assert_eq!(
            rows.tool_calls[1].result.as_deref(),
            Some(r#"{"hits":1.0}"#)
        );
        assert_eq!(rows.tool_calls[1].first_party, Some(true));
    }

    #[test]
    fn empty_payload_folds_to_no_rows() {
        let events = vec![(1, Event::new(kinds::USER_MSG, Vec::new()))];
        let rows = decode_message_content_events("conv-empty", &events);
        assert!(rows.messages.is_empty());
        assert!(rows.tool_calls.is_empty());
    }

    #[test]
    fn undecodable_non_empty_payload_is_skipped() {
        let events = vec![
            (1, Event::new(kinds::USER_MSG, vec![0xFF, 0xFE, 0xFD])),
            (2, Event::new(kinds::USER_MSG, Vec::new())),
        ];
        let rows = decode_message_content_events("conv-corrupt", &events);
        assert!(rows.messages.is_empty());
        assert!(rows.tool_calls.is_empty());
    }

    #[test]
    fn unrelated_kind_is_not_decoded() {
        let message = text_message("user", "hi", false);
        let events = vec![(1, Event::new(kinds::USAGE, message.encode_to_vec()))];
        let rows = decode_message_content_events("conv-unrelated", &events);
        assert!(rows.messages.is_empty());
        assert!(rows.tool_calls.is_empty());
    }
}