polyc-query 2026.8.3

//! `fires` typed-table decoder — the scheduler's durable `routine_fired`
//! markers → Arrow (issue #1592, the routines explorer page's fire-history
//! source).
//!
//! `kinds::ROUTINE_FIRED` was `Decode::Opaque` in [`crate::decode::REGISTRY`]
//! until this table: `crates/control-plane/src/routine_scheduler.rs`'s
//! `routine_fired_event` appends one of these to its own dedicated
//! `"routine-scheduler"` event-log partition per actual fire (both the
//! `fixedContent` class in production today and the `enrolled` class kept
//! undeleted for a future payload variant — see that module's own doc),
//! recording exactly [`polyc_proto::proto::polychrome::events::v1::RoutineFiredEvent`]'s
//! four fields: `routine`, `occurrence`, `scheduled_at_ms`, `fired_at_ms`. A
//! DIRECT decode (like [`crate::decode::handoffs`]/[`crate::decode::summary`]),
//! not fold-coupled — there is no signature/trust check on a
//! `routine_fired` marker (`Event::trusted`, appended by the control plane's
//! own scheduler, never caller-supplied), so [`decode_typed_kind_events`]
//! covers this table exactly the way it covers `usage`/`model_call`.
//!
//! # Fleet sees every fire; a persona-scoped session sees its own (issue #1882)
//!
//! `fires` is registered for [`crate::session::QueryScope::Fleet`] (every
//! row, unfiltered — [`crate::views::FIRES_VIEW_SQL`]) and for a
//! persona-scoped session (rows belonging to that persona's OWN routines
//! only, via an INNER JOIN against the already owner-filtered `routines`
//! table — [`crate::views::FIRES_OWNED_VIEW_SQL`]). It is never registered
//! for a conversation-grant scope, which resolves no caller persona to own
//! anything. This table's one source partition
//! (`crate::authority`'s admitted `"routine-scheduler"` partition) used to
//! be admitted into replay only for a Fleet scope; a persona-scoped replay
//! now admits it too, narrowly and leniently (an unreadable/missing
//! scheduler partition is skipped, never a hard failure for that scope) —
//! see `crate::authority`'s module doc, "Admitting the scheduler partition"
//! section.
//!
//! # Uniform keys, with one caveat
//!
//! `partition`/`position`/`turn_id` are derived exactly the way every other
//! typed table derives them, via [`polyc_proto::kinds::parse`] — `partition`
//! is always `"routine-scheduler"` (this table's one source partition) and
//! `turn_id` is always `None` (`kinds::ROUTINE_FIRED` is never tagged with a
//! turn: a fire opens no turn of its own, see
//! `crate::authority`'s doc for `ROUTINE_SCHEDULER_PARTITION`). Both columns
//! are kept anyway, matching every other typed table's uniform-key
//! discipline (#1311) rather than a special-cased narrower schema for this
//! one table.
//!
//! # `outcome` (#1656)
//!
//! A fire's `routine_fired` marker records only that a dispatch was
//! attempted, never how it resolved (see that event's own doc: the marker
//! lands unconditionally on any attempted dispatch). #1656 appends a sibling
//! `kinds::ROUTINE_FIRE_OUTCOME` event in the SAME atomic batch as the
//! marker it accompanies, keyed to it by `(routine, occurrence)` — this
//! decoder folds that sibling event into the `outcome` column by that key,
//! the same "second kind, same table, folded by key" shape
//! [`crate::decode::routine_lifecycle`]'s phases use, except here the join
//! key is `(routine, occurrence)` rather than a single signed record's own
//! `phase` field. `outcome` is `NULL` for any `routine_fired` marker with no
//! matching outcome event — every fire recorded before #1656 shipped, and
//! the rare case where the outcome append's own batch failed.

use std::collections::HashMap;
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::proto::polychrome::events::v1::{
    RoutineFireOutcome, RoutineFireOutcomeEvent, RoutineFiredEvent,
};

/// One decoded `fires` row: the fact model's uniform key columns
/// (`partition`, `position`, `turn_id`) plus every
/// [`RoutineFiredEvent`] field.
#[derive(Debug, Clone)]
pub(crate) struct FireRow {
    /// The journal partition this row's event was read from — always
    /// `"routine-scheduler"` (see the module doc).
    pub partition: String,
    /// The journal's own monotonic append position for this event.
    pub position: u64,
    /// Always `None` — see the module doc's "Uniform keys" section.
    pub turn_id: Option<String>,
    /// [`RoutineFiredEvent::routine`] — the `Routine` resource's name, joins
    /// to `routines.name`.
    pub routine: String,
    /// [`RoutineFiredEvent::occurrence`] — `{routine}-{scheduled_epoch_minutes}`.
    pub occurrence: String,
    /// [`RoutineFiredEvent::scheduled_at_ms`] — the SCHEDULED tick, not
    /// necessarily when the fire itself ran (a caught-up fire's tick
    /// predates its fire time).
    pub scheduled_at_ms: u64,
    /// [`RoutineFiredEvent::fired_at_ms`] — when the fire itself ran.
    pub fired_at_ms: u64,
    /// [`RoutineFiredEvent::routine_uid`] — the firing `Routine` CR's own
    /// stable `uid`, empty for a marker recorded before this field existed
    /// (INV-OAF18, part of issue #1882's review). The owner-scoped `fires`
    /// view joins on this rather than on `routine` — see
    /// `crate::views::FIRES_OWNED_VIEW_SQL`'s doc for why a name-keyed join
    /// is unsafe, and for what an empty `routine_uid` means for a
    /// pre-this-change row.
    pub routine_uid: String,
    /// The classified result of this fire's dispatch, folded in from a
    /// sibling `routine_fire_outcome` event keyed by `(routine, occurrence)`
    /// — `NULL` when no such event was ever appended: pre-#1656 history
    /// only, since #1656 the outcome event lands in the SAME atomic batch
    /// as the marker itself, so a marker from that point on can never exist
    /// without one. One of `"ok"`, `"paused"`, `"retryable_error"`,
    /// `"terminal_error"`, `"timed_out"`, `"stopped_ungranted"` when present.
    pub outcome: Option<String>,
    /// [`RoutineFireOutcomeEvent::grant_drift_tools`] — tool names
    /// a per-tool routine grant admitted although the tool's descriptor hash
    /// no longer matched the hash the owner approved, folded in from the SAME
    /// sibling outcome event as [`Self::outcome`] and rendered as JSON-array
    /// text (`crate::decode::handoffs`' `allowed` column's convention for a
    /// list this crate has no reason to model as a native Arrow list type).
    /// `NULL` exactly when [`Self::outcome`] is `NULL`; `"[]"` for a fire
    /// where no admitted call drifted.
    pub grant_drift_tools: Option<String>,
}

/// The `fires` typed table's Arrow schema.
///
/// `partition` (`Utf8`, non-null), `position` (`UInt64`, non-null), `turn_id`
/// (`Utf8`, nullable — always `NULL`, see the module doc), `routine` (`Utf8`,
/// non-null), `occurrence` (`Utf8`, non-null), `scheduled_at_ms` (`UInt64`,
/// non-null), `fired_at_ms` (`UInt64`, non-null), `routine_uid` (`Utf8`,
/// non-null — empty string for a marker recorded before this column existed,
/// INV-OAF18, part of issue #1882's review), `outcome` (`Utf8`, nullable —
/// see the module doc's "outcome" section), `grant_drift_tools` (`Utf8`,
/// nullable JSON-array text — see [`FireRow::grant_drift_tools`]).
#[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("routine", DataType::Utf8, false),
        Field::new("occurrence", DataType::Utf8, false),
        Field::new("scheduled_at_ms", DataType::UInt64, false),
        Field::new("fired_at_ms", DataType::UInt64, false),
        Field::new("routine_uid", DataType::Utf8, false),
        Field::new("outcome", DataType::Utf8, true),
        Field::new("grant_drift_tools", DataType::Utf8, true),
    ]))
}

/// Decode already-framed [`FireRow`]s into the `fires_raw` table's Arrow
/// `RecordBatch`, in [`schema`] order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_fires_batch(rows: &[FireRow]) -> Result<RecordBatch, ArrowError> {
    let mut partition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut position_b = UInt64Builder::with_capacity(rows.len());
    let mut turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut routine_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut occurrence_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut scheduled_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut fired_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut routine_uid_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut outcome_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut grant_drift_tools_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(),
        }
        routine_b.append_value(&row.routine);
        occurrence_b.append_value(&row.occurrence);
        scheduled_at_ms_b.append_value(row.scheduled_at_ms);
        fired_at_ms_b.append_value(row.fired_at_ms);
        routine_uid_b.append_value(&row.routine_uid);
        match &row.outcome {
            Some(outcome) => outcome_b.append_value(outcome),
            None => outcome_b.append_null(),
        }
        match &row.grant_drift_tools {
            Some(drift) => grant_drift_tools_b.append_value(drift),
            None => grant_drift_tools_b.append_null(),
        }
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(routine_b.finish()),
        Arc::new(occurrence_b.finish()),
        Arc::new(scheduled_at_ms_b.finish()),
        Arc::new(fired_at_ms_b.finish()),
        Arc::new(routine_uid_b.finish()),
        Arc::new(outcome_b.finish()),
        Arc::new(grant_drift_tools_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Classify a [`RoutineFireOutcome`] wire value into the `outcome` column's
/// string vocabulary — mirrors
/// `polyc_control_plane::routine_scheduler::prompt_fire::PromptFireOutcome`'s
/// own variant set. `Unspecified` (a decode default, never deliberately
/// appended) maps to `None`, so a malformed or absent classification reads
/// as `NULL` rather than a fabricated status string.
#[must_use]
const fn outcome_str(outcome: RoutineFireOutcome) -> Option<&'static str> {
    match outcome {
        RoutineFireOutcome::Ok => Some("ok"),
        RoutineFireOutcome::Paused => Some("paused"),
        RoutineFireOutcome::RetryableError => Some("retryable_error"),
        RoutineFireOutcome::TerminalError => Some("terminal_error"),
        RoutineFireOutcome::TimedOut => Some("timed_out"),
        RoutineFireOutcome::StoppedUngranted => Some("stopped_ungranted"),
        RoutineFireOutcome::Unspecified => None,
    }
}

/// Fold `partition`'s `routine_fire_outcome` events into a
/// `(routine, occurrence) -> (outcome, grant_drift_tools_json)` map —
/// [`decode_fires_events`]'s join key for filling each `routine_fired` row's
/// `outcome` and `grant_drift_tools` columns.
///
/// A structurally malformed payload is skipped, same as every other
/// [`crate::decode::decode_typed_kind_events`] caller.
///
/// `.collect()` into the `HashMap` last-wins on a duplicate `(routine,
/// occurrence)` key, but the scheduler's own occurrence dedup
/// (`fold_fired_markers`'s `already_fired` set, `seed_from_replay`'s replay
/// check) makes a genuine duplicate outcome event for the same occurrence
/// unreachable in practice — this is a defensive fallback order, not a
/// meaningful choice between two real outcomes.
#[must_use]
fn fold_outcome_events(
    partition: &str,
    events: &[(u64, Event)],
) -> HashMap<(String, String), (&'static str, String)> {
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[polyc_proto::kinds::ROUTINE_FIRE_OUTCOME],
        "fires",
        try_decode_event_payload::<RoutineFireOutcomeEvent>,
        |_partition, _position, _turn_id, decoded: RoutineFireOutcomeEvent| decoded,
    )
    .into_iter()
    .filter_map(|decoded| {
        let outcome = decoded.outcome.as_known().unwrap_or_default();
        let drift =
            serde_json::to_string(&decoded.grant_drift_tools).unwrap_or_else(|_| "[]".to_owned());
        outcome_str(outcome).map(|s| ((decoded.routine, decoded.occurrence), (s, drift)))
    })
    .collect()
}

/// Filter `partition`'s framed `events` to `routine_fired` rows and decode
/// each payload as [`RoutineFiredEvent`] via
/// [`crate::decode::decode_typed_kind_events`], pairing a successful decode
/// with the fact model's uniform key columns, then fills each row's
/// `outcome` column from [`fold_outcome_events`]'s `(routine, occurrence)`
/// map (#1656) — `NULL` when no matching outcome event exists.
///
/// A non-empty payload that fails to decode as [`RoutineFiredEvent`] is
/// skipped (logged via `tracing::warn!`) — never surfaced as an error, same
/// as every other [`crate::decode::decode_typed_kind_events`] caller.
#[must_use]
pub(crate) fn decode_fires_events(partition: &str, events: &[(u64, Event)]) -> Vec<FireRow> {
    let outcomes = fold_outcome_events(partition, events);
    crate::decode::decode_typed_kind_events(
        partition,
        events,
        &[polyc_proto::kinds::ROUTINE_FIRED],
        "fires",
        try_decode_event_payload::<RoutineFiredEvent>,
        |partition, position, turn_id, decoded: RoutineFiredEvent| {
            let matched = outcomes.get(&(decoded.routine.clone(), decoded.occurrence.clone()));
            FireRow {
                partition,
                position,
                turn_id,
                routine: decoded.routine,
                occurrence: decoded.occurrence,
                scheduled_at_ms: decoded.scheduled_at_ms,
                fired_at_ms: decoded.fired_at_ms,
                routine_uid: decoded.routine_uid,
                outcome: matched.map(|(s, _)| (*s).to_owned()),
                grant_drift_tools: matched.map(|(_, drift)| drift.clone()),
            }
        },
    )
}

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

    use super::*;

    fn sample_event() -> RoutineFiredEvent {
        RoutineFiredEvent {
            routine: "daily-standup".to_string(),
            occurrence: "daily-standup-28461600".to_string(),
            scheduled_at_ms: 1_753_300_800_000,
            fired_at_ms: 1_753_300_805_000,
            routine_uid: "uid-daily-standup".to_string(),
            ..Default::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",
                "routine",
                "occurrence",
                "scheduled_at_ms",
                "fired_at_ms",
                "routine_uid",
                "outcome",
                "grant_drift_tools",
            ]
        );
        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("turn_id", DataType::Utf8, true),
            ("routine", DataType::Utf8, false),
            ("occurrence", DataType::Utf8, false),
            ("scheduled_at_ms", DataType::UInt64, false),
            ("fired_at_ms", DataType::UInt64, false),
            ("routine_uid", DataType::Utf8, false),
            ("outcome", DataType::Utf8, true),
            ("grant_drift_tools", DataType::Utf8, true),
        ];
        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 `routine_fired` marker (the exact shape
    /// `routine_scheduler::routine_fired_event` appends) round-trips through
    /// decode and batch, with `turn_id` always `NULL` — a fire is a bare
    /// kind, never tagged with a turn.
    #[test]
    fn decode_round_trips_a_routine_fired_marker() {
        let ev = sample_event();
        let events = vec![(9, Event::trusted(kinds::ROUTINE_FIRED, ev.encode_to_vec()))];

        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].partition, "routine-scheduler");
        assert_eq!(decoded[0].position, 9);
        assert_eq!(decoded[0].turn_id, None);
        assert_eq!(decoded[0].routine, "daily-standup");
        assert_eq!(decoded[0].occurrence, "daily-standup-28461600");
        assert_eq!(decoded[0].scheduled_at_ms, 1_753_300_800_000);
        assert_eq!(decoded[0].fired_at_ms, 1_753_300_805_000);
        assert_eq!(decoded[0].routine_uid, "uid-daily-standup");
        assert_eq!(
            decoded[0].outcome, None,
            "no sibling routine_fire_outcome event was appended"
        );
        assert_eq!(
            decoded[0].grant_drift_tools, None,
            "grant_drift_tools is NULL exactly when outcome is NULL"
        );

        let batch = decode_fires_batch(&decoded).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

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

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

        let routine_uid = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(routine_uid.value(0), "uid-daily-standup");
    }

    /// A sibling `routine_fire_outcome` event (#1656), appended in the same
    /// partition and keyed to the marker by `(routine, occurrence)`, folds
    /// into that marker's row as its `outcome` column.
    #[test]
    fn decode_folds_a_sibling_outcome_event_into_its_markers_row() {
        let marker = sample_event();
        let outcome_ev = RoutineFireOutcomeEvent {
            routine: marker.routine.clone(),
            occurrence: marker.occurrence.clone(),
            outcome: RoutineFireOutcome::Ok.into(),
            fired_at_ms: marker.fired_at_ms,
            ..Default::default()
        };
        let events = vec![
            (
                9,
                Event::trusted(kinds::ROUTINE_FIRED, marker.encode_to_vec()),
            ),
            (
                10,
                Event::trusted(kinds::ROUTINE_FIRE_OUTCOME, outcome_ev.encode_to_vec()),
            ),
        ];

        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(
            decoded.len(),
            1,
            "the outcome event mints no row of its own"
        );
        assert_eq!(decoded[0].outcome.as_deref(), Some("ok"));
        assert_eq!(
            decoded[0].grant_drift_tools.as_deref(),
            Some("[]"),
            "an outcome event with no drifted tool renders an empty JSON array, not NULL"
        );
    }

    /// A `stopped_ungranted` outcome event carrying `grant_drift_tools`
    /// folds both the abort classification and the drifted tool
    /// names into its marker's row.
    #[test]
    fn decode_folds_a_stopped_ungranted_abort_and_its_drifted_tools() {
        let marker = sample_event();
        let outcome_ev = RoutineFireOutcomeEvent {
            routine: marker.routine.clone(),
            occurrence: marker.occurrence.clone(),
            outcome: RoutineFireOutcome::StoppedUngranted.into(),
            fired_at_ms: marker.fired_at_ms,
            grant_drift_tools: vec!["fs_write".to_string(), "http_post".to_string()],
            ..Default::default()
        };
        let events = vec![
            (
                9,
                Event::trusted(kinds::ROUTINE_FIRED, marker.encode_to_vec()),
            ),
            (
                10,
                Event::trusted(kinds::ROUTINE_FIRE_OUTCOME, outcome_ev.encode_to_vec()),
            ),
        ];

        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].outcome.as_deref(), Some("stopped_ungranted"));
        assert_eq!(
            decoded[0].grant_drift_tools.as_deref(),
            Some(r#"["fs_write","http_post"]"#)
        );

        let batch = decode_fires_batch(&decoded).expect("batch build");
        let drift = batch
            .column(9)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(drift.value(0), r#"["fs_write","http_post"]"#);
    }

    /// A `routine_fired` marker with no matching outcome event — only
    /// possible for a fire recorded before #1656 shipped, since the outcome
    /// event now lands in the same atomic batch as the marker — decodes
    /// with `outcome` `NULL`, never a fabricated status.
    #[test]
    fn decode_leaves_outcome_null_when_no_sibling_event_exists() {
        let marker = sample_event();
        let events = vec![(
            9,
            Event::trusted(kinds::ROUTINE_FIRED, marker.encode_to_vec()),
        )];

        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].outcome, None);
    }

    /// An event kind outside `routine_fired` is not fire data at all and must
    /// never surface as a row here.
    #[test]
    fn unrelated_kind_is_not_decoded_as_a_fire() {
        let events = vec![(1, Event::new(kinds::USAGE, Vec::new()))];
        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(decoded.len(), 0);
    }

    /// A structurally malformed payload (not even valid protobuf for
    /// `RoutineFiredEvent`) is skipped — no row, no panic.
    #[test]
    fn structurally_malformed_payload_drops_the_row() {
        let events = vec![(
            1,
            Event::trusted(kinds::ROUTINE_FIRED, vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_fires_events("routine-scheduler", &events);
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn empty_events_yield_zero_rows() {
        let decoded = decode_fires_events("routine-scheduler", &[]);
        assert_eq!(decoded.len(), 0);
    }
}