polyc-query 2026.9.0

//! `routines` reference-table decoder — [`crate::routine_catalog::RoutineStatusRecord`]
//! → Arrow (issue #1592).
//!
//! Mirrors [`crate::decode::persona`]'s own shape: this is reference data,
//! not a journal decode — there is no `partition`/`position`/`turn_id`
//! uniform-key triple here, since a `Routine`'s status is not a journal event
//! at all. See [`crate::routine_catalog`]'s module doc for why the row type
//! this module decodes ([`RoutineStatusRecord`]) carries no spec field.

use std::sync::Arc;

use arrow::array::{ArrayRef, BooleanBuilder, Int64Builder, StringBuilder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;

use crate::routine_catalog::RoutineStatusRecord;

/// The `routines` table's Arrow schema.
///
/// `name` (`Utf8`, non-null), `ready` (`Boolean`, non-null), `phase`
/// (`Utf8`, nullable — `None` before the first reconcile), `message` (`Utf8`,
/// nullable), `last_fire_time_ms`/`next_fire_time_ms` (`Int64`, nullable,
/// epoch ms), `conditions_json`
/// (`Utf8`, non-null — `"[]"` for a routine with no conditions yet, never
/// absent). #1593 adds: `creator_persona`/`provenance_conversation_id`
/// (`Utf8`, non-null — `RoutineProvenance` is required on every spec),
/// `schedule_json`/`next_fires_json` (`Utf8`, non-null — always present,
/// `next_fires_json` is `"[]"` rather than absent when empty; its entries and
/// `schedule_json`'s `once.at_ms` are JSON numbers, epoch ms, never RFC3339
/// text), `suspended` (`Boolean`, non-null),
/// `paused_by`/`pause_reason` (`Utf8`, nullable — `None` while active),
/// `paused_at_ms` (`Int64`, nullable, epoch ms). Signed, not `UInt64`: a
/// pre-1970 instant must still decode, not saturate.
/// `prompt` (`Utf8`, non-null — the routines explorer table's row-hierarchy
/// fix): `RoutinePayload::prompt`, required non-empty by `polyc_controller`'s
/// own admission (`validate_payload`), so this column is never null. #1807
/// adds: `scope` (`Utf8`, non-null — `"public"`/`"private"`, always populated
/// by admission's private default) and `orphaned` (`Boolean`, non-null — the
/// scheduler-written `Orphaned` condition, `false` until the owner departs).
/// INV-OAF18 (part of issue #1882's review) adds `uid` (`Utf8`, non-null —
/// empty string for a CR with no uid yet, never absent): the query layer's
/// owner-scoped `fires` view joins on this column rather than on `name` — see
/// `crate::views::FIRES_OWNED_VIEW_SQL`'s own doc for why a name-keyed join
/// is unsafe.
#[must_use]
pub(crate) fn schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("name", DataType::Utf8, false),
        Field::new("uid", DataType::Utf8, false),
        Field::new("fire_conversation_id", DataType::Utf8, false),
        Field::new("ready", DataType::Boolean, false),
        Field::new("phase", DataType::Utf8, true),
        Field::new("message", DataType::Utf8, true),
        Field::new("last_fire_time_ms", DataType::Int64, true),
        Field::new("next_fire_time_ms", DataType::Int64, true),
        Field::new("conditions_json", DataType::Utf8, false),
        Field::new("creator_persona", DataType::Utf8, false),
        Field::new("provenance_conversation_id", DataType::Utf8, false),
        Field::new("schedule_json", DataType::Utf8, false),
        Field::new("next_fires_json", DataType::Utf8, false),
        Field::new("suspended", DataType::Boolean, false),
        Field::new("paused_by", DataType::Utf8, true),
        Field::new("paused_at_ms", DataType::Int64, true),
        Field::new("pause_reason", DataType::Utf8, true),
        Field::new("prompt", DataType::Utf8, false),
        Field::new("scope", DataType::Utf8, false),
        Field::new("orphaned", DataType::Boolean, false),
        Field::new("display_name", DataType::Utf8, false),
        Field::new("description", DataType::Utf8, false),
        Field::new("schedule_timezone", DataType::Utf8, false),
    ]))
}

/// Decode [`RoutineStatusRecord`]s into the `routines` table's Arrow
/// `RecordBatch` ([`schema`] column order).
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
#[allow(clippy::too_many_lines)] // 15 columns' worth of mechanical builder wiring, see `handoffs`' own precedent
pub(crate) fn decode_routines_batch(
    rows: &[RoutineStatusRecord],
) -> Result<RecordBatch, ArrowError> {
    let mut name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut uid_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut fire_conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut ready_b = BooleanBuilder::with_capacity(rows.len());
    let mut phase_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut message_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut last_fire_time_ms_b = Int64Builder::with_capacity(rows.len());
    let mut next_fire_time_ms_b = Int64Builder::with_capacity(rows.len());
    let mut conditions_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
    let mut creator_persona_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut provenance_conversation_id_b =
        StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut schedule_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 48);
    let mut next_fires_json_b = StringBuilder::with_capacity(rows.len(), rows.len() * 96);
    let mut suspended_b = BooleanBuilder::with_capacity(rows.len());
    let mut paused_by_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut paused_at_ms_b = Int64Builder::with_capacity(rows.len());
    let mut pause_reason_b = StringBuilder::with_capacity(rows.len(), rows.len() * 32);
    let mut prompt_b = StringBuilder::with_capacity(rows.len(), rows.len() * 64);
    let mut scope_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut orphaned_b = BooleanBuilder::with_capacity(rows.len());
    let mut display_name_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    let mut description_b = StringBuilder::with_capacity(rows.len(), rows.len() * 48);
    let mut schedule_timezone_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);

    for row in rows {
        name_b.append_value(&row.name);
        uid_b.append_value(&row.uid);
        fire_conversation_id_b.append_value(&row.fire_conversation_id);
        ready_b.append_value(row.ready);
        match &row.phase {
            Some(v) => phase_b.append_value(v),
            None => phase_b.append_null(),
        }
        match &row.message {
            Some(v) => message_b.append_value(v),
            None => message_b.append_null(),
        }
        match row.last_fire_time_ms {
            Some(v) => last_fire_time_ms_b.append_value(v),
            None => last_fire_time_ms_b.append_null(),
        }
        match row.next_fire_time_ms {
            Some(v) => next_fire_time_ms_b.append_value(v),
            None => next_fire_time_ms_b.append_null(),
        }
        conditions_json_b.append_value(&row.conditions_json);
        creator_persona_b.append_value(&row.creator_persona);
        provenance_conversation_id_b.append_value(&row.provenance_conversation_id);
        schedule_json_b.append_value(&row.schedule_json);
        next_fires_json_b.append_value(&row.next_fires_json);
        suspended_b.append_value(row.suspended);
        match &row.paused_by {
            Some(v) => paused_by_b.append_value(v),
            None => paused_by_b.append_null(),
        }
        match row.paused_at_ms {
            Some(v) => paused_at_ms_b.append_value(v),
            None => paused_at_ms_b.append_null(),
        }
        match &row.pause_reason {
            Some(v) => pause_reason_b.append_value(v),
            None => pause_reason_b.append_null(),
        }
        prompt_b.append_value(&row.prompt);
        scope_b.append_value(&row.scope);
        orphaned_b.append_value(row.orphaned);
        display_name_b.append_value(&row.display_name);
        description_b.append_value(&row.description);
        schedule_timezone_b.append_value(&row.schedule_timezone);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(name_b.finish()),
        Arc::new(uid_b.finish()),
        Arc::new(fire_conversation_id_b.finish()),
        Arc::new(ready_b.finish()),
        Arc::new(phase_b.finish()),
        Arc::new(message_b.finish()),
        Arc::new(last_fire_time_ms_b.finish()),
        Arc::new(next_fire_time_ms_b.finish()),
        Arc::new(conditions_json_b.finish()),
        Arc::new(creator_persona_b.finish()),
        Arc::new(provenance_conversation_id_b.finish()),
        Arc::new(schedule_json_b.finish()),
        Arc::new(next_fires_json_b.finish()),
        Arc::new(suspended_b.finish()),
        Arc::new(paused_by_b.finish()),
        Arc::new(paused_at_ms_b.finish()),
        Arc::new(pause_reason_b.finish()),
        Arc::new(prompt_b.finish()),
        Arc::new(scope_b.finish()),
        Arc::new(orphaned_b.finish()),
        Arc::new(display_name_b.finish()),
        Arc::new(description_b.finish()),
        Arc::new(schedule_timezone_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

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

    use super::*;

    fn sample_record() -> RoutineStatusRecord {
        RoutineStatusRecord {
            name: "daily-standup".to_string(),
            uid: "uid-daily-standup".to_string(),
            fire_conversation_id: "fire-conv-daily-standup".to_string(),
            ready: true,
            phase: Some("Ready".to_string()),
            message: Some("validated".to_string()),
            last_fire_time_ms: Some(1_785_657_600_000),
            next_fire_time_ms: Some(1_785_744_000_000),
            conditions_json: r#"[{"type":"MissedFire","status":"False"}]"#.to_string(),
            creator_persona: "persona-1".to_string(),
            provenance_conversation_id: "conv-1".to_string(),
            schedule_json: r#"{"kind":"cron","expression":"0 9 * * *","timezone":null}"#
                .to_string(),
            next_fires_json: r"[1785744000000,1785830400000]".to_string(),
            suspended: false,
            paused_by: None,
            paused_at_ms: None,
            pause_reason: None,
            prompt: "post the morning standup".to_string(),
            scope: "private".to_string(),
            orphaned: false,
            display_name: String::new(),
            description: String::new(),
            schedule_timezone: "UTC".to_owned(),
        }
    }

    #[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![
                "name",
                "uid",
                "fire_conversation_id",
                "ready",
                "phase",
                "message",
                "last_fire_time_ms",
                "next_fire_time_ms",
                "conditions_json",
                "creator_persona",
                "provenance_conversation_id",
                "schedule_json",
                "next_fires_json",
                "suspended",
                "paused_by",
                "paused_at_ms",
                "pause_reason",
                "prompt",
                "scope",
                "orphaned",
                "display_name",
                "description",
                "schedule_timezone",
            ]
        );
        assert!(!schema.field(0).is_nullable(), "name must be non-null");
        assert!(!schema.field(1).is_nullable(), "uid must be non-null");
        assert!(
            !schema.field(2).is_nullable(),
            "fire_conversation_id must be non-null"
        );
        assert!(!schema.field(3).is_nullable(), "ready must be non-null");
        assert!(schema.field(4).is_nullable());
        assert!(schema.field(5).is_nullable());
        assert!(schema.field(6).is_nullable());
        assert!(schema.field(7).is_nullable());
        assert!(
            !schema.field(8).is_nullable(),
            "conditions_json must be non-null"
        );
        assert!(
            !schema.field(9).is_nullable(),
            "creator_persona must be non-null"
        );
        assert!(
            !schema.field(10).is_nullable(),
            "provenance_conversation_id must be non-null"
        );
        assert!(
            !schema.field(11).is_nullable(),
            "schedule_json must be non-null"
        );
        assert!(
            !schema.field(12).is_nullable(),
            "next_fires_json must be non-null"
        );
        assert!(
            !schema.field(13).is_nullable(),
            "suspended must be non-null"
        );
        assert!(schema.field(14).is_nullable());
        assert!(schema.field(15).is_nullable());
        assert!(schema.field(16).is_nullable());
        assert!(!schema.field(17).is_nullable(), "prompt must be non-null");
        assert!(!schema.field(18).is_nullable(), "scope must be non-null");
        assert!(!schema.field(19).is_nullable(), "orphaned must be non-null");
        assert!(
            !schema.field(20).is_nullable(),
            "display_name must be non-null"
        );
        assert!(
            !schema.field(21).is_nullable(),
            "description must be non-null"
        );
        assert!(
            !schema.field(22).is_nullable(),
            "schedule_timezone must be non-null"
        );
    }

    /// A known [`RoutineStatusRecord`] round-trips through
    /// [`decode_routines_batch`] to exactly its own row — provenance and
    /// schedule included (issue #1593).
    #[test]
    fn routine_status_record_round_trips_to_its_row() {
        let batch = decode_routines_batch(&[sample_record()]).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

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

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

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

        let ready = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(ready.value(0));

        let next_fire_time_ms = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::Int64Array>()
            .unwrap();
        assert_eq!(next_fire_time_ms.value(0), 1_785_744_000_000);

        let conditions_json = batch
            .column(8)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(
            conditions_json.value(0),
            r#"[{"type":"MissedFire","status":"False"}]"#
        );

        let creator_persona = batch
            .column(9)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(creator_persona.value(0), "persona-1");

        let provenance_conversation_id = batch
            .column(10)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(provenance_conversation_id.value(0), "conv-1");

        let schedule_json = batch
            .column(11)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(
            schedule_json.value(0),
            r#"{"kind":"cron","expression":"0 9 * * *","timezone":null}"#
        );

        let next_fires_json = batch
            .column(12)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(next_fires_json.value(0), r"[1785744000000,1785830400000]");

        let suspended = batch
            .column(13)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(!suspended.value(0));

        let paused_by = batch
            .column(14)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(paused_by.is_null(0));

        let prompt = batch
            .column(17)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(prompt.value(0), "post the morning standup");

        let scope = batch
            .column(18)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(scope.value(0), "private");

        let orphaned = batch
            .column(19)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(!orphaned.value(0));
    }

    /// #1807: a public, owner-departed routine's `scope`/`orphaned` round-trip
    /// onto their own columns — snapshot-level prior art mirrored from
    /// #1592/#1593's own table tests.
    #[test]
    fn scope_and_orphaned_round_trip() {
        let mut record = sample_record();
        record.scope = "public".to_string();
        record.orphaned = true;

        let batch = decode_routines_batch(&[record]).expect("batch build");

        let scope = batch
            .column(18)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(scope.value(0), "public");

        let orphaned = batch
            .column(19)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(orphaned.value(0));
    }

    #[test]
    fn unreconciled_routine_round_trips_nulls() {
        let record = RoutineStatusRecord {
            name: "brand-new".to_string(),
            uid: "uid-brand-new".to_string(),
            fire_conversation_id: "fire-conv-brand-new".to_string(),
            ready: false,
            phase: None,
            message: None,
            last_fire_time_ms: None,
            next_fire_time_ms: None,
            conditions_json: "[]".to_string(),
            creator_persona: "persona-2".to_string(),
            provenance_conversation_id: "conv-2".to_string(),
            schedule_json: r#"{"kind":"once","at_ms":1785657600000}"#.to_string(),
            next_fires_json: "[]".to_string(),
            suspended: false,
            paused_by: None,
            paused_at_ms: None,
            pause_reason: None,
            prompt: "check the inbox".to_string(),
            scope: "private".to_string(),
            orphaned: false,
            display_name: String::new(),
            description: String::new(),
            schedule_timezone: "UTC".to_owned(),
        };
        let batch = decode_routines_batch(&[record]).expect("batch build");

        let phase = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert!(phase.is_null(0));

        let next_fire_time_ms = batch
            .column(7)
            .as_any()
            .downcast_ref::<arrow::array::Int64Array>()
            .unwrap();
        assert!(next_fire_time_ms.is_null(0));
    }

    #[test]
    fn paused_routine_round_trips_pause_metadata() {
        let mut record = sample_record();
        record.suspended = true;
        record.paused_by = Some("persona-1".to_string());
        record.paused_at_ms = Some(1_785_542_400_000);
        record.pause_reason = Some("rotating out old announcements".to_string());

        let batch = decode_routines_batch(&[record]).expect("batch build");

        let suspended = batch
            .column(13)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(suspended.value(0));

        let paused_by = batch
            .column(14)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(paused_by.value(0), "persona-1");

        let paused_at_ms = batch
            .column(15)
            .as_any()
            .downcast_ref::<arrow::array::Int64Array>()
            .unwrap();
        assert_eq!(paused_at_ms.value(0), 1_785_542_400_000);

        let pause_reason = batch
            .column(16)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(pause_reason.value(0), "rotating out old announcements");
    }

    #[test]
    fn empty_rows_yield_zero_row_batch_with_the_full_schema() {
        let batch = decode_routines_batch(&[]).expect("batch build");
        assert_eq!(batch.num_rows(), 0);
        assert_eq!(batch.schema(), schema());
    }
}