polyc-query 2026.9.0

//! `turn_failed` typed-table decoder — `TurnFailedEvent` → Arrow columns.
//!
//! The fact model's fourth typed table (#1311/#1178's "keyed
//! domain-fact-table" wave; see `crate::decode`'s module docs) —
//! `crate::decode::model_call` is the closest analog, a clean single-kind
//! typed table, and this module mirrors its shape for
//! `polychrome.events.v1.TurnFailedEvent` (`crates/proto/proto/events.proto`,
//! `kind = "turn_failed"`) — the structured sibling of the empty
//! `turn_complete` marker (`#756`): a turn that ended in failure instead of
//! completing, now decoded straight from the payload rather than left in the
//! wide `events` table's opaque `payload` column. "Instead of completing"
//! describes the turn's OUTCOME, not the absence of a commit marker: the
//! control plane's sole `turn_failed` emission
//! (`crates/control-plane/src/grpc/mod.rs`) still pushes the `turn_complete`
//! COMMIT marker unconditionally, in the same atomic batch, immediately
//! after the optional `turn_failed` event — the two co-occur for any
//! normally-failed turn, which is exactly why `crate::engine`'s `turn_failed`
//! view can apply the same committed-turn semijoin every other typed table
//! uses (see `crate::views`'s module docs' "Committed-turn filter
//! invariant" section) rather than needing an exemption. Same design
//! rationale [`crate::decode::usage`] cites: a schema-per-kind typed table
//! beside the wide `events` table.
//!
//! This is the fact model's "why turns failed" table: a query workload that
//! wants a breakdown of turn failures by kind, or every failed turn's
//! diagnostic text within a time window, reads `turn_failed` directly
//! instead of scanning `events_raw`'s opaque `payload` bytes for the
//! `turn_failed` kind-base and hand-decoding each one client-side.
//!
//! # Uniform keys (#1311)
//!
//! Same discipline as [`crate::decode::usage`]/[`crate::decode::model_call`]:
//! 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 = "turn_failed"` 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`/`model_call`/`summary` all
//! call through, rather than each carrying its own copy (`attribution`
//! moved to its own `polyc_facts` fold in #1579 — see that module's own
//! "Shared decode" section).
//!
//! # `event_time` is deferred to #1327
//!
//! Same as every other typed table in this crate: no uniform attested
//! `event_time` column exists yet (tracked as future work in #1327), and this
//! module does not invent one — see [`crate::decode::usage`]'s module docs
//! for the fuller rationale.
//!
//! # No identity, no redaction concern
//!
//! Unlike [`crate::decode::attribution`], a `TurnFailedEvent` carries no
//! external identity of any kind — just a provider-agnostic failure
//! classification and a diagnostic message describing what went wrong inside
//! ONE turn of ONE conversation. It is conversation-scoped-safe the same way
//! `usage`/`model_call` are (a row never spans conversations), with none of
//! `attribution`'s identity-redaction machinery needed: this table's public
//! `turn_failed` view carries IDENTICAL columns for every
//! [`crate::session::QueryScope`] — the raw/view split it does need is the
//! committed-turn-filter one `usage`/`model_call` also use (see
//! `crate::engine::TURN_FAILED_RAW_TABLE`'s doc and `crate::views`'s module
//! docs' "Committed-turn filter invariant" section), not an identity
//! redaction.
//!
//! # The `kind` enum → `failure_kind` string column
//!
//! [`TurnFailedEvent::kind`] is `polychrome.harness.v1.TurnFailureKind`, a
//! proto3 (open) enum — buffa's generated field type is
//! `buffa::EnumValue<TurnFailureKind>`, which holds either `Known(variant)`
//! or `Unknown(i32)` for a wire value no current variant covers (see
//! `buffa::EnumValue`'s own docs). `failure_kind_label` maps every known
//! variant to its stable, lowercase, `snake_case` label — `"rate_limit"`,
//! `"timeout"`, and so on — the same convention
//! `polyc_eventlog::TrustTag::as_str` uses for the wide `events` table's own
//! `trust` column. An `Unknown` wire value (a message from a future harness
//! build carrying a variant this crate does not yet know) maps to
//! `"unspecified"`, the same label `TurnFailureKind::Unspecified` itself gets
//! — an unrecognized classification is exactly as informative to a query as
//! an explicitly-absent one, and both are real, queryable rows rather than a
//! decode failure.
//!
//! No existing helper in this crate already stringifies `TurnFailureKind`
//! (`crates/rpc-client/src/lib.rs`'s own `TurnFailureKind` is a distinct,
//! edge-facing enum with its own variant set, not a mirror of the wire type
//! this module decodes), so `failure_kind_label` maps every variant
//! explicitly rather than reusing one.
//!
//! # Column selection
//!
//! [`TurnFailedEvent`] carries exactly two fields beyond the uniform keys,
//! and this table's first cut keeps both:
//!
//! - `failure_kind` — the stringified [`TurnFailureKind`] (see above).
//! - `message` — the underlying provider/tool diagnostic text, verbatim.
//!   Empty when the turn failed without one (an all-defaults payload; see
//!   below).
//!
//! Decode quirk this module inherits from `decode_typed_kind_events`: an
//! EMPTY payload decodes cleanly to [`TurnFailedEvent`]'s all-defaults value
//! (proto3 elides all-default scalar fields) — `kind` unset (`Unknown(0)`,
//! stringified `"unspecified"`), `message = ""` — 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 buffa::EnumValue;
use polyc_eventlog::Event;
use polyc_proto::events_decode::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::TurnFailedEvent;
use polyc_proto::proto::polychrome::harness::v1::TurnFailureKind;

/// One decoded `turn_failed` row.
///
/// The fact model's uniform key columns (`partition`, `position`, `turn_id`)
/// plus the [`TurnFailedEvent`] fields this table keeps (see the module
/// docs' "Column selection").
#[derive(Debug, Clone)]
pub(crate) struct TurnFailedRow {
    /// 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 `turn_failed` kind with no turn
    /// tagged.
    pub turn_id: Option<String>,
    /// The stringified, provider-agnostic failure classification — see the
    /// module docs' "The `kind` enum → `failure_kind` string column"
    /// section.
    pub failure_kind: String,
    /// Human-readable diagnostic text (the underlying provider/tool error).
    /// Empty when the turn failed without one.
    pub message: String,
}

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

/// Map a decoded [`TurnFailedEvent::kind`] onto its stable, lowercase,
/// `snake_case` label.
///
/// Every [`TurnFailureKind`] variant is matched explicitly — see the module
/// docs' "The `kind` enum → `failure_kind` string column" section for why an
/// `Unknown` wire value maps to `"unspecified"`, the same label
/// [`TurnFailureKind::Unspecified`] itself gets, rather than surfacing as a
/// decode failure.
#[must_use]
fn failure_kind_label(kind: EnumValue<TurnFailureKind>) -> &'static str {
    match kind.as_known() {
        Some(TurnFailureKind::RateLimit) => "rate_limit",
        Some(TurnFailureKind::Timeout) => "timeout",
        Some(TurnFailureKind::Unavailable) => "unavailable",
        Some(TurnFailureKind::Auth) => "auth",
        Some(TurnFailureKind::BadRequest) => "bad_request",
        Some(TurnFailureKind::Other) => "other",
        Some(TurnFailureKind::Unspecified) | None => "unspecified",
    }
}

/// Decode already-framed [`TurnFailedRow`]s into the `turn_failed` table's
/// Arrow `RecordBatch`.
///
/// Columns are `partition`, `position`, `turn_id`, `failure_kind`, `message`,
/// in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_turn_failed_batch(rows: &[TurnFailedRow]) -> 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 failure_kind_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut message_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);

    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(),
        }
        failure_kind_b.append_value(&row.failure_kind);
        message_b.append_value(&row.message);
    }

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

/// Filter `partition`'s framed `events` to kind-base `turn_failed` 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
/// [`TurnFailedEvent`] → [`TurnFailedRow`] field mapping (see the module
/// docs' "Column selection" and "The `kind` enum → `failure_kind` string
/// column" sections) is unique to this module.
#[must_use]
pub(crate) fn decode_turn_failed_events(
    partition: &str,
    events: &[(u64, Event)],
) -> Vec<TurnFailedRow> {
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[kinds::TURN_FAILED],
        "turn_failed",
        try_decode_event_payload::<TurnFailedEvent>,
        |partition, position, turn_id, event: TurnFailedEvent| TurnFailedRow {
            partition,
            position,
            turn_id,
            failure_kind: failure_kind_label(event.kind).to_string(),
            message: event.message,
        },
    )
}

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

    use super::*;

    /// A [`TurnFailedEvent`] naming both real fields explicitly — no
    /// `..Default::default()` spread — 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_turn_failed(kind: TurnFailureKind, message: &str) -> TurnFailedEvent {
        TurnFailedEvent {
            kind: kind.into(),
            message: message.to_string(),
            __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",
                "failure_kind",
                "message"
            ]
        );

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("failure_kind", DataType::Utf8, false),
            ("message", 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 failure_kind_label_covers_every_known_variant() {
        assert_eq!(
            failure_kind_label(TurnFailureKind::Unspecified.into()),
            "unspecified"
        );
        assert_eq!(
            failure_kind_label(TurnFailureKind::RateLimit.into()),
            "rate_limit"
        );
        assert_eq!(
            failure_kind_label(TurnFailureKind::Timeout.into()),
            "timeout"
        );
        assert_eq!(
            failure_kind_label(TurnFailureKind::Unavailable.into()),
            "unavailable"
        );
        assert_eq!(failure_kind_label(TurnFailureKind::Auth.into()), "auth");
        assert_eq!(
            failure_kind_label(TurnFailureKind::BadRequest.into()),
            "bad_request"
        );
        assert_eq!(failure_kind_label(TurnFailureKind::Other.into()), "other");
    }

    #[test]
    fn failure_kind_label_maps_an_unknown_wire_value_to_unspecified() {
        let unknown: EnumValue<TurnFailureKind> = EnumValue::from(999);
        assert_eq!(failure_kind_label(unknown), "unspecified");
    }

    #[test]
    fn decode_turn_failed_batch_round_trips() {
        let rows = vec![
            TurnFailedRow {
                partition: "conv-a".to_string(),
                position: 5,
                turn_id: None,
                failure_kind: "rate_limit".to_string(),
                message: "provider throttled the request".to_string(),
            },
            TurnFailedRow {
                partition: "conv-a".to_string(),
                position: 9,
                turn_id: Some("turn-xyz".to_string()),
                failure_kind: "timeout".to_string(),
                message: "deadline exceeded".to_string(),
            },
        ];
        let batch = decode_turn_failed_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 failure_kind = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(failure_kind.value(0), "rate_limit");
        assert_eq!(failure_kind.value(1), "timeout");

        let message = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(message.value(0), "provider throttled the request");
        assert_eq!(message.value(1), "deadline exceeded");
    }

    #[test]
    fn decode_turn_failed_events_filters_and_decodes_real_buffa_bytes() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let turn_failed =
            sample_turn_failed(TurnFailureKind::Unavailable, "harness pod unreachable");
        let bytes = turn_failed.encode_to_vec();

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

        let decoded = decode_turn_failed_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].failure_kind, "unavailable");
        assert_eq!(decoded[0].message, "harness pod unreachable");
    }

    #[test]
    fn decode_turn_failed_events_bare_kind_has_no_turn_id() {
        let events = vec![(7, Event::new(kinds::TURN_FAILED, Vec::new()))];
        let decoded = decode_turn_failed_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_turn_failed_event_decodes_to_defaults() {
        let events = vec![(1, Event::new(kinds::TURN_FAILED, Vec::new()))];
        let decoded = decode_turn_failed_events("conv-empty", &events);
        assert_eq!(decoded.len(), 1, "an empty payload must decode, not skip");
        assert_eq!(decoded[0].failure_kind, "unspecified");
        assert_eq!(decoded[0].message, "");
    }

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

    #[test]
    fn turn_failed_event_round_trip_through_decode_and_batch() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let turn_failed = sample_turn_failed(TurnFailureKind::BadRequest, "malformed tool call");
        let bytes = turn_failed.encode_to_vec();
        let events = vec![(
            1,
            Event::new(kinds::tagged(kinds::TURN_FAILED, &turn), bytes),
        )];

        let decoded = decode_turn_failed_events("conv-rt", &events);
        let batch = decode_turn_failed_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 failure_kind = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(failure_kind.value(0), "bad_request");

        let message = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(message.value(0), "malformed tool call");
    }
}