polyc-query 2026.8.3

//! `turn_dispatch` typed-table decoder — `TurnDispatchedEvent` → Arrow
//! columns (issue #1593).
//!
//! Mirrors [`crate::decode::turn_failed`]'s shape exactly: a clean,
//! single-kind, DIRECT decode of `kinds::TURN_DISPATCHED` through
//! [`crate::decode::decode_typed_kind_events`], committed-turn filtered,
//! identical columns for every [`crate::session::QueryScope`] (no external
//! identity on this table either). `kinds::TURN_DISPATCHED` was
//! [`crate::decode::Decode::Opaque`] until this table — this is the fact
//! model's twelfth typed table.
//!
//! # Why this table exists: joining a routine's fires to the turns they opened
//!
//! [`polyc_proto::proto::polychrome::events::v1::TurnDispatchedEvent::occurrence`]
//! carries the SAME scheduled-occurrence identity
//! ([`crate::decode::fires::FireRow::occurrence`]) a routine's fire marker
//! carries — mirroring `polychrome.agent.v1.AgentStart.occurrence` — so a
//! turn opened by a routine's schedule can be joined back to the fire that
//! opened it: `SELECT * FROM fires f JOIN turn_dispatch t ON t.occurrence =
//! f.occurrence`. Before this table, `occurrence` was reachable only as raw
//! protobuf bytes inside `events_raw.payload` (not even `payload_json`,
//! since a protobuf-encoded string field is not valid JSON text) — this is
//! the routines explorer page's (#1592/#1593) fire-history-joins-turns
//! source: the routines page (`crate::decode::fires`'s own module docs)
//! joins `fires` to this table by occurrence, keeping only fires a turn
//! actually opened for. In practice `crate::routine_scheduler::fire_and_mark_tick`
//! (control plane, #1594) dispatches a turn unconditionally for every due
//! tick once its in-process turn-dispatch handle and the routine's `uid`
//! both resolve, so this join surfaces a match for nearly every fire — but
//! it is not a hard guarantee: a tick fired before that handle existed, or
//! whose dispatch handle/uid were unset at fire time
//! (`TickOutcome::AllFailed`, no marker recorded at all in that case, so no
//! orphan fire to worry about here), is the one case this join legitimately
//! drops.
//!
//! # `occurrence` is empty (never `NULL`) for a turn with no scheduled origin
//!
//! Every OTHER turn kind (a chat turn, or a firing from the trigger edge's
//! queue/webhook/fs kinds) still carries a `turn_dispatch` row — the
//! dispatch-intent marker is unconditional (`crates/control-plane/src/grpc/mod.rs`)
//! — just with an empty `occurrence`, matching
//! [`polyc_proto::proto::polychrome::events::v1::TurnDispatchedEvent`]'s own
//! doc ("empty for every turn with no scheduled origin"). This table keeps
//! that literal value (`""`, not `NULL`) rather than converting it — the same
//! "empty payload's literal field values stay literal" convention
//! [`crate::decode::turn_failed`]'s `message` column already establishes —
//! so `WHERE occurrence != ''` (or the join above, which an empty string
//! never accidentally matches since no real occurrence is ever empty) is how
//! a caller selects only scheduled-origin turns.
//!
//! # The audience columns (`#2702`, §5.1)
//!
//! Four columns carry what the dispatch record says about who may read the
//! turn: `visibility` (what the control plane ESTABLISHED),
//! `visibility_source` (whether it decided that here or inherited it from the
//! paused turn it resumes), `source_turn_id` (which turn, when it inherited),
//! and `edge_asserted_visibility` (the claim it was decided against, as the
//! ingest gate admitted it).
//!
//! `visibility_source` is the reason the other three are worth querying. A
//! resume asserts nothing about the room, so it takes the paused turn's
//! recorded conclusion; without the source column a borrowed conclusion and a
//! fresh one are the same row. `SELECT * FROM turn_dispatch WHERE
//! visibility_source = 'inherited' AND visibility = 'direct'` is the audit
//! this table did not previously support.
//!
//! ## An unnameable enum value reads as `unrecognized`, not folded
//!
//! [`crate::decode::turn_failed`] maps an `Unknown` wire value onto the same
//! label its `UNSPECIFIED` variant gets, arguing the two are equally
//! informative. That argument does not hold here, and these columns
//! deliberately do not follow it. `RECORDED_TURN_VISIBILITY_UNKNOWN` is not
//! an absent value — it is a recorded DECISION that nothing was established,
//! and it withholds. A value written by a newer build says something this
//! build cannot read. Folding the second into the first would erase exactly
//! the distinction an audit table exists to show, so an unnameable value
//! reads as `unrecognized` in all three enum columns.
//!
//! This is a rendering choice, not a policy one: every READ path in the
//! workspace collapses an unnameable value to "withhold" (`§5.2.1`,
//! unparseable fails closed). The column is more precise than the decision
//! because a person reading it is asking a different question.
//!
//! # Uniform keys (#1311)
//!
//! `partition`, `position`, `turn_id` — derived exactly as
//! [`crate::decode::turn_failed`] derives them, via
//! [`crate::decode::decode_typed_kind_events`]. In practice `turn_id` is
//! always `Some` (`crates/control-plane/src/grpc/mod.rs` always tags this
//! kind with the dispatching turn), but the column stays nullable, matching
//! every other typed table's uniform-key discipline rather than assuming
//! that invariant here.

use std::sync::Arc;

use arrow::array::{ArrayRef, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_eventlog::Event;
use polyc_proto::events_decode::try_decode_event_payload;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::TurnDispatchedEvent;

/// One decoded `turn_dispatch` row.
#[derive(Debug, Clone)]
pub(crate) struct TurnDispatchRow {
    /// 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` — see the module
    /// docs for why this is always `Some` in practice, though the column
    /// stays nullable.
    pub turn_id: Option<String>,
    /// [`TurnDispatchedEvent::occurrence`] — the scheduled-occurrence
    /// identity this turn was fired from, or empty for a turn with no
    /// scheduled origin (see the module docs).
    pub occurrence: String,
    /// The stringified [`TurnDispatchedEvent::visibility`] — what the control
    /// plane ESTABLISHED about who reads this turn.
    pub visibility: &'static str,
    /// The stringified [`TurnDispatchedEvent::visibility_source`] — whether
    /// [`Self::visibility`] was decided from this turn's own inputs or
    /// inherited from the turn it resumes.
    pub visibility_source: &'static str,
    /// [`TurnDispatchedEvent::source_turn_id`] — the turn whose conclusion an
    /// inherited row took, or empty (never `NULL`, matching `occurrence`'s own
    /// convention) on a row that inherited nothing.
    pub source_turn_id: String,
    /// The stringified [`TurnDispatchedEvent::edge_asserted_visibility`] — the
    /// claim this dispatch acted on, as the ingest gate admitted it.
    pub edge_asserted_visibility: &'static str,
}

/// The `turn_dispatch` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null),
/// `turn_id` (`Utf8`, nullable), `occurrence` (`Utf8`, non-null — empty, not
/// `NULL`, for a turn with no scheduled origin; see the module docs),
/// `visibility` / `visibility_source` / `edge_asserted_visibility` (`Utf8`,
/// non-null — stringified enums, see the module docs' audience-columns
/// section), and `source_turn_id` (`Utf8`, non-null — empty, not `NULL`, on a
/// row that inherited nothing).
#[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("occurrence", DataType::Utf8, false),
        Field::new("visibility", DataType::Utf8, false),
        Field::new("visibility_source", DataType::Utf8, false),
        Field::new("source_turn_id", DataType::Utf8, false),
        Field::new("edge_asserted_visibility", DataType::Utf8, false),
    ]))
}

/// Decode already-framed [`TurnDispatchRow`]s into the `turn_dispatch_raw`
/// table's Arrow `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_turn_dispatch_batch(
    rows: &[TurnDispatchRow],
) -> 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 occurrence_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut visibility_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut visibility_source_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut source_turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut edge_asserted_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);

    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(),
        }
        occurrence_b.append_value(&row.occurrence);
        visibility_b.append_value(row.visibility);
        visibility_source_b.append_value(row.visibility_source);
        source_turn_id_b.append_value(&row.source_turn_id);
        edge_asserted_b.append_value(row.edge_asserted_visibility);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(occurrence_b.finish()),
        Arc::new(visibility_b.finish()),
        Arc::new(visibility_source_b.finish()),
        Arc::new(source_turn_id_b.finish()),
        Arc::new(edge_asserted_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Filter `partition`'s framed `events` to `turn_dispatched` rows, decode
/// each payload as [`TurnDispatchedEvent`], and pair a successful decode with
/// that row's uniform key columns — via
/// [`crate::decode::decode_typed_kind_events`], the same shared loop
/// `usage`/`model_call`/`turn_failed`/`fires` call through.
#[must_use]
pub(crate) fn decode_turn_dispatch_events(
    partition: &str,
    events: &[(u64, Event)],
) -> Vec<TurnDispatchRow> {
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[kinds::TURN_DISPATCHED],
        "turn_dispatch",
        try_decode_event_payload::<TurnDispatchedEvent>,
        |partition, position, turn_id, event: TurnDispatchedEvent| TurnDispatchRow {
            partition,
            position,
            turn_id,
            occurrence: event.occurrence,
            visibility: polyc_proto::audience_display::recorded_visibility_label(event.visibility),
            visibility_source: polyc_proto::audience_display::recorded_source_label(
                event.visibility_source,
            ),
            source_turn_id: event.source_turn_id,
            edge_asserted_visibility: polyc_proto::audience_display::asserted_visibility_label(
                event.edge_asserted_visibility,
            ),
        },
    )
}

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

    use super::*;

    #[test]
    fn schema_shape() {
        let schema = schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "partition",
                "position",
                "turn_id",
                "occurrence",
                "visibility",
                "visibility_source",
                "source_turn_id",
                "edge_asserted_visibility",
            ]
        );
        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("occurrence", DataType::Utf8, false),
            ("visibility", DataType::Utf8, false),
            ("visibility_source", DataType::Utf8, false),
            ("source_turn_id", DataType::Utf8, false),
            ("edge_asserted_visibility", 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);
        }
    }

    /// A real `turn_dispatched` marker carrying a routine's occurrence
    /// round-trips through decode and batch.
    #[test]
    fn decode_round_trips_a_scheduled_origin_dispatch() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_2345);
        let event = TurnDispatchedEvent {
            occurrence: "daily-standup-28461600".to_owned(),
            ..Default::default()
        };
        let events = vec![(
            3,
            Event::new(
                kinds::tagged(kinds::TURN_DISPATCHED, &turn),
                event.encode_to_vec(),
            ),
        )];

        let decoded = decode_turn_dispatch_events("conv-rt", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].partition, "conv-rt");
        assert_eq!(decoded[0].position, 3);
        assert_eq!(decoded[0].turn_id, Some(turn.to_string()));
        assert_eq!(decoded[0].occurrence, "daily-standup-28461600");

        let batch = decode_turn_dispatch_batch(&decoded).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());
        let occurrence = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(occurrence.value(0), "daily-standup-28461600");
    }

    /// An inherited stamp decodes with the turn it was inherited FROM.
    ///
    /// The row a resume writes. Without `visibility_source` this row and a
    /// turn that proved one reader on its own inputs are the same row, which
    /// is what `#2702`'s §5.1 is about.
    #[test]
    fn an_inherited_stamp_names_the_turn_it_came_from() {
        use polyc_proto::proto::polychrome::agent::v1::ConversationVisibility;

        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_7777);
        let source = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_8888);
        let event = TurnDispatchedEvent {
            visibility: buffa::EnumValue::Known(
                RecordedTurnVisibility::RECORDED_TURN_VISIBILITY_DIRECT,
            ),
            visibility_source: buffa::EnumValue::Known(
                RecordedVisibilitySource::RECORDED_VISIBILITY_SOURCE_INHERITED,
            ),
            source_turn_id: source.to_string(),
            // A redrive asserts nothing of its own, and the record says so.
            edge_asserted_visibility: buffa::EnumValue::Known(
                ConversationVisibility::CONVERSATION_VISIBILITY_UNKNOWN,
            ),
            ..Default::default()
        };
        let events = vec![(
            9,
            Event::new(
                kinds::tagged(kinds::TURN_DISPATCHED, &turn),
                event.encode_to_vec(),
            ),
        )];

        let decoded = decode_turn_dispatch_events("conv-resume", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].visibility, "direct");
        assert_eq!(decoded[0].visibility_source, "inherited");
        assert_eq!(decoded[0].source_turn_id, source.to_string());
        assert_eq!(decoded[0].edge_asserted_visibility, "unknown");

        let batch = decode_turn_dispatch_batch(&decoded).expect("batch build");
        let column = |index: usize| {
            batch
                .column(index)
                .as_any()
                .downcast_ref::<arrow::array::StringArray>()
                .expect("string column")
                .value(0)
                .to_owned()
        };
        assert_eq!(column(4), "direct");
        assert_eq!(column(5), "inherited");
        assert_eq!(column(6), source.to_string());
        assert_eq!(column(7), "unknown");
    }

    /// A dispatch written before the source field existed reads as
    /// `unrecorded`, never as `asserted`.
    ///
    /// The wire zero says nothing about how that turn's audience was decided.
    /// Reading it as a fresh decision would invent the exact assurance this
    /// column exists to supply.
    #[test]
    fn a_record_without_a_source_reads_as_unrecorded() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_6666);
        let event = TurnDispatchedEvent {
            visibility: buffa::EnumValue::Known(
                RecordedTurnVisibility::RECORDED_TURN_VISIBILITY_DIRECT,
            ),
            ..Default::default()
        };
        let events = vec![(
            2,
            Event::new(
                kinds::tagged(kinds::TURN_DISPATCHED, &turn),
                event.encode_to_vec(),
            ),
        )];
        let decoded = decode_turn_dispatch_events("conv-legacy", &events);
        assert_eq!(decoded[0].visibility_source, "unrecorded");
        assert_eq!(decoded[0].source_turn_id, "");
    }

    /// An enum value this build cannot name reads as `unrecognized` in every
    /// audience column — never folded into `unknown`.
    ///
    /// `unknown` is a recorded decision that withholds; an unnameable value is
    /// a record this build cannot read. An audit table that could not tell
    /// them apart would answer the question it exists for wrongly.
    #[test]
    fn an_unnameable_audience_value_reads_as_unrecognized() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_5555);
        let event = TurnDispatchedEvent {
            visibility: buffa::EnumValue::Unknown(97),
            visibility_source: buffa::EnumValue::Unknown(98),
            edge_asserted_visibility: buffa::EnumValue::Unknown(99),
            ..Default::default()
        };
        let events = vec![(
            4,
            Event::new(
                kinds::tagged(kinds::TURN_DISPATCHED, &turn),
                event.encode_to_vec(),
            ),
        )];
        let decoded = decode_turn_dispatch_events("conv-future", &events);
        assert_eq!(decoded[0].visibility, "unrecognized");
        assert_eq!(decoded[0].visibility_source, "unrecognized");
        assert_eq!(decoded[0].edge_asserted_visibility, "unrecognized");
    }

    /// A chat turn (no scheduled origin) decodes with an empty `occurrence`,
    /// not `NULL` — see the module docs.
    #[test]
    fn chat_turn_has_empty_not_null_occurrence() {
        let turn = Uuid::from_u128(0x0195_abcd_ef01_2345_6789_abcd_ef01_9999);
        let events = vec![(
            1,
            Event::new(kinds::tagged(kinds::TURN_DISPATCHED, &turn), Vec::new()),
        )];
        let decoded = decode_turn_dispatch_events("conv-chat", &events);
        assert_eq!(decoded.len(), 1, "an empty payload must decode, not skip");
        assert_eq!(decoded[0].occurrence, "");
    }

    #[test]
    fn undecodable_non_empty_payload_is_skipped() {
        let events = vec![(
            1,
            Event::new(kinds::TURN_DISPATCHED, vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_turn_dispatch_events("conv-corrupt", &events);
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn unrelated_kind_is_not_decoded_as_a_dispatch() {
        let events = vec![(1, Event::new(kinds::USAGE, Vec::new()))];
        let decoded = decode_turn_dispatch_events("conv-unrelated", &events);
        assert_eq!(decoded.len(), 0);
    }
}