polyc-query 2026.9.0

//! `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`
//!
//! The fact model 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 — the same fact, decoded once, never folded twice.
//! `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
//! the fact model'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 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 the fact model'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
//! the fact model's variant→block class 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
//! the access-control 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 polyc_eventlog::Event;
use polyc_proto::events_decode::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::agent::v1::Message;

pub(crate) mod messages;
pub(crate) mod tool_calls;

pub(crate) use messages::messages_schema;
use messages::{MessageProjection, MessageRows};
pub(crate) use tool_calls::tool_calls_schema;
use tool_calls::{ToolCallProjection, ToolCallRows, ToolResultProjection};

/// 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];

/// Filters `partition`'s events to [`MESSAGE_CONTENT_KIND_BASES`].
///
/// The function decodes each admitted payload as a wire [`Message`]. It folds
/// each message once through [`polyc_facts::fold_message_content`]. The fold
/// dispatches each fact to an independent row sink.
///
/// This function cannot use [`crate::decode::decode_typed_kind_events`].
/// That helper returns exactly one row for each decoded payload.
/// One decoded [`Message`] can produce a message row, a tool-call row, or no
/// row. This function uses [`polyc_proto::kinds::parse`] for the same kind
/// parsing. It skips an invalid payload and logs a non-empty invalid payload.
pub(crate) fn decode_message_content_events(
    partition: &str,
    events: &[(u64, Event)],
    messages: &mut MessageRows,
    tool_calls: &mut ToolCallRows,
) {
    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) => {
                messages.push(MessageProjection {
                    partition: partition.to_string(),
                    role: message.role,
                    internal_only,
                    fact: t,
                });
            }
            polyc_facts::MessageContent::ToolCall(call) => {
                tool_calls.push_call(ToolCallProjection {
                    partition: partition.to_string(),
                    internal_only,
                    fact: call,
                });
            }
            polyc_facts::MessageContent::ToolResult(res) => {
                tool_calls.push_result(ToolResultProjection {
                    partition: partition.to_string(),
                    internal_only,
                    fact: res,
                });
            }
            polyc_facts::MessageContent::None => {}
        }
    }
}

#[cfg(test)]
mod tests {
    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(),
        }
    }

    fn decode_rows(partition: &str, events: &[(u64, Event)]) -> (MessageRows, ToolCallRows) {
        let mut messages = MessageRows::default();
        let mut tool_calls = ToolCallRows::default();
        decode_message_content_events(partition, events, &mut messages, &mut tool_calls);
        (messages, tool_calls)
    }

    #[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 (messages, tool_calls) = decode_rows("conv-real", &events);
        assert_eq!(messages.as_slice().len(), 1);
        assert!(tool_calls.as_slice().is_empty());
        let row = &messages.as_slice()[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 (messages, tool_calls) = decode_rows("conv-internal", &events);
        assert_eq!(messages.as_slice().len(), 1);
        assert!(messages.as_slice()[0].internal_only);
        assert!(tool_calls.as_slice().is_empty());
    }

    #[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::trusted(
                kinds::tagged(kinds::OUTPUT_MSG, &turn),
                message.encode_to_vec(),
            ),
        )];

        let (messages, tool_calls) = decode_rows("conv-real", &events);
        assert!(messages.as_slice().is_empty());
        assert_eq!(tool_calls.as_slice().len(), 1);
        let row = &tool_calls.as_slice()[0];
        assert_eq!(row.partition, "conv-real");
        assert_eq!(row.position, 2);
        assert_eq!(row.turn_id, Some(turn.to_string()));
        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);
        assert!(!row.internal_only);
        assert_eq!(row.trust, polyc_eventlog::TrustTag::TrustedUser.as_str());
    }

    #[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 (messages, tool_calls) = decode_rows("conv-paired", &events);
        assert!(messages.as_slice().is_empty());
        let tool_calls = tool_calls.as_slice();
        assert_eq!(tool_calls.len(), 2);
        assert_eq!(tool_calls[0].tool_call_id, "call-1");
        assert_eq!(tool_calls[0].block_type, "call");
        assert_eq!(tool_calls[1].tool_call_id, "call-1");
        assert_eq!(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!(tool_calls[1].result.as_deref(), Some(r#"{"hits":1.0}"#));
        assert_eq!(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 (messages, tool_calls) = decode_rows("conv-empty", &events);
        assert!(messages.as_slice().is_empty());
        assert!(tool_calls.as_slice().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 (messages, tool_calls) = decode_rows("conv-corrupt", &events);
        assert!(messages.as_slice().is_empty());
        assert!(tool_calls.as_slice().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 (messages, tool_calls) = decode_rows("conv-unrelated", &events);
        assert!(messages.as_slice().is_empty());
        assert!(tool_calls.as_slice().is_empty());
    }
}