polyc-query 2026.9.6

//! Proves the fixed SQL statements `ListRoutineGrants`, `ListRoutineRefusals`,
//! and `GetRoutineFire`'s dispatch read send to the Fleet Query plane
//! (POLY-361) against real `DataFusion` `MemTable`s built from the exact
//! `conversation-security/v1`/`conversation-execution/v1` schemas the
//! registry declares.
//!
//! Same posture as `routine_overview_sql.rs`/`routine_fires_sql.rs`: a
//! syntax-and-semantics proof for the SQL text itself, run against a fixed
//! table rather than a live projected stack. In production every statement
//! here except `routine_fire_count.sql` runs under a `GrantSubject::Persona`
//! conversation grant already scoped to one routine's fire conversation, so
//! none of them names a `partition` — this fixture registers exactly one
//! conversation's rows for the same reason.
//!
//! Grants (`routine_grant_mutations`) and refusals/pending-setup (`approvals`) are
//! DIFFERENT tables, proven separately: a grant/revocation record is a
//! bare, requestless response `append_routine_grant` commits outside any
//! turn, which `approvals`' committed-turn and request-matching rules would
//! drop — see `polyc_facts::RoutineGrantMutationFact`'s own doc.

#![allow(clippy::unwrap_used)]

use std::sync::Arc;

use arrow::array::{Array, BooleanArray, Int64Array, RecordBatch, StringArray, UInt64Array};
use datafusion::datasource::MemTable;
use datafusion::prelude::SessionContext;
use polyc_projection::family::{
    CONVERSATION_EXECUTION_ENTRY, CONVERSATION_SECURITY_ENTRY, EXECUTION_TURN_DISPATCH,
    ROUTINE_FIRES, ROUTINE_LIFECYCLE_ENTRY, SECURITY_APPROVALS, SECURITY_ROUTINE_GRANTS,
    TableSchema,
};

const ACTIVE_GRANTS_SQL: &str = polyc_query_model::statements::ROUTINE_OWNER_ACTIVE_GRANTS_SQL;
const AGGREGATES_SQL: &str = polyc_query_model::statements::ROUTINE_OWNER_APPROVAL_AGGREGATES_SQL;
const REFUSALS_SQL: &str = polyc_query_model::statements::ROUTINE_OWNER_REFUSALS_SQL;
const FIRE_DISPATCH_SQL: &str = polyc_query_model::statements::ROUTINE_OWNER_FIRE_DISPATCH_SQL;
const FIRE_COUNT_SQL: &str = polyc_query_model::statements::ROUTINE_FIRE_COUNT_SQL;
const FIRE_LAST_SQL: &str = polyc_query_model::statements::ROUTINE_FIRE_LAST_SQL;
const STOPPED_TOOL_SQL: &str = polyc_query_model::statements::ROUTINE_OWNER_STOPPED_TOOL_SQL;

fn arrow_schema_for(declared: &TableSchema) -> arrow::datatypes::SchemaRef {
    use polyc_projection::family::LogicalType;

    let fields: Vec<arrow::datatypes::Field> = declared
        .fields()
        .iter()
        .map(|field| {
            let data_type = match field.logical_type() {
                LogicalType::Utf8 => arrow::datatypes::DataType::Utf8,
                LogicalType::FixedBytes { len } => {
                    arrow::datatypes::DataType::FixedSizeBinary(i32::try_from(len).unwrap())
                }
                LogicalType::UInt64 => arrow::datatypes::DataType::UInt64,
                LogicalType::Boolean => arrow::datatypes::DataType::Boolean,
            };
            arrow::datatypes::Field::new(field.name(), data_type, field.nullable())
        })
        .collect();
    Arc::new(arrow::datatypes::Schema::new(fields))
}

fn fixed_incarnation(len: usize) -> arrow::array::FixedSizeBinaryArray {
    let mut builder = arrow::array::FixedSizeBinaryBuilder::with_capacity(len, 32);
    for _ in 0..len {
        builder.append_value([0_u8; 32]).unwrap();
    }
    builder.finish()
}

/// One fire conversation's routine-grant ledger, six records:
///
/// 1. `fs_write` granted.
/// 2. `fs_write` revoked — a LATER record for the same key, `approved =
///    false` — proving the windowing keys per tool and a revocation wins.
/// 3. `shell_exec` granted, the sole surviving per-tool key.
/// 4. `network_call`, a surviving BLANKET grant (`blanket_all`).
fn routine_grant_mutations_context() -> SessionContext {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        CONVERSATION_SECURITY_ENTRY
            .table(SECURITY_ROUTINE_GRANTS)
            .expect("routine_grant_mutations table declared"),
    );
    let n = 4;
    let batch = RecordBatch::try_new(
        Arc::clone(&schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a"; n])), // partition
            Arc::new(fixed_incarnation(n)),                 // source_incarnation
            Arc::new(UInt64Array::from(vec![1_u64, 2, 3, 4])), // position
            Arc::new(StringArray::from(vec![
                "turn-1", "turn-2", "turn-3", "turn-4",
            ])), // turn_id
            Arc::new(StringArray::from(vec![
                "fs_write",
                "fs_write",
                "shell_exec",
                "network_call",
            ])), // tool_name
            Arc::new(StringArray::from(vec!["hash-1"; n])), // tool_descriptor_hash
            Arc::new(StringArray::from(vec![
                "tool",
                "tool",
                "tool",
                "blanket_all",
            ])), // grant_scope
            Arc::new(BooleanArray::from(vec![true, false, true, true])), // approved
            Arc::new(StringArray::from(vec![
                "granted", "revoked", "granted", "granted",
            ])), // response_reason
            Arc::new(StringArray::from(vec!["verified"; n])), // signature_status
        ],
    )
    .unwrap();
    ctx.register_table(
        "routine_grant_mutations",
        Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()),
    )
    .unwrap();
    ctx
}

#[tokio::test]
async fn active_grants_windows_per_tool_and_keeps_a_surviving_blanket() {
    let ctx = routine_grant_mutations_context();
    let df = ctx.sql(ACTIVE_GRANTS_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
    assert_eq!(
        rows, 2,
        "fs_write's grant was revoked by a later record; shell_exec's per-tool \
         grant and the blanket grant survive"
    );
    let scopes: Vec<String> = batches
        .iter()
        .flat_map(|batch| {
            let column = batch
                .column_by_name("grant_scope")
                .unwrap()
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap()
                .clone();
            (0..batch.num_rows())
                .map(move |row| column.value(row).to_owned())
                .collect::<Vec<_>>()
        })
        .collect();
    assert!(scopes.contains(&"tool".to_owned()));
    assert!(scopes.contains(&"blanket_all".to_owned()));
}

/// One fire conversation's ordinary tool-call approval history, three rows:
/// an unanswered request (the pending-setup case) and two true refusals
/// (`outcome = denied`, `routine_grant = false`) — every row here is a
/// real request+response pair inside a committed turn, unlike a grant
/// mutation.
fn approvals_context() -> SessionContext {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        CONVERSATION_SECURITY_ENTRY
            .table(SECURITY_APPROVALS)
            .expect("approvals table declared"),
    );
    let n = 3;
    let batch = RecordBatch::try_new(
        Arc::clone(&schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a"; n])), // partition
            Arc::new(fixed_incarnation(n)),                 // source_incarnation
            Arc::new(UInt64Array::from(vec![1_u64, 2, 3])), // position
            Arc::new(StringArray::from(vec!["turn-1", "turn-2", "turn-3"])), // turn_id
            Arc::new(StringArray::from(vec!["req-1", "req-2", "req-3"])), // request_id
            Arc::new(StringArray::from(vec![
                "network_call",
                "shell_exec",
                "fs_write",
            ])), // tool_name
            Arc::new(StringArray::from(vec!["{}"; n])),     // args_json
            Arc::new(StringArray::from(vec!["unanswered", "denied", "denied"])), // outcome
            Arc::new(StringArray::from(vec![
                "",
                "not routine-safe",
                "not routine-safe",
            ])), // response_reason
            Arc::new(StringArray::from(vec!["", "verified", "verified"])), // signature_status
            Arc::new(BooleanArray::from(vec![false, false, false])), // routine_grant
            Arc::new(StringArray::from(vec!["", "", ""])),  // tool_descriptor_hash
            Arc::new(StringArray::from(vec!["", "", ""])),  // grant_scope
        ],
    )
    .unwrap();
    ctx.register_table(
        "approvals",
        Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()),
    )
    .unwrap();
    ctx
}

#[tokio::test]
async fn approval_aggregates_count_refusals_and_unanswered_requests() {
    let ctx = approvals_context();
    let df = ctx.sql(AGGREGATES_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    assert_eq!(
        batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
        1,
        "an unconditional aggregate always returns exactly one row"
    );
    let denial_count = batches[0]
        .column_by_name("denial_count")
        .unwrap()
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap()
        .value(0);
    let pending = batches[0]
        .column_by_name("pending_setup_approvals")
        .unwrap()
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap()
        .value(0);
    assert_eq!(denial_count, 2, "positions 2 and 3 are true refusals");
    assert_eq!(pending, 1, "only position 1 is unanswered");
}

#[tokio::test]
async fn aggregates_over_zero_rows_still_returns_one_row_at_zero() {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        CONVERSATION_SECURITY_ENTRY
            .table(SECURITY_APPROVALS)
            .expect("approvals table declared"),
    );
    let empty = RecordBatch::new_empty(Arc::clone(&schema));
    ctx.register_table(
        "approvals",
        Arc::new(MemTable::try_new(schema, vec![vec![empty]]).unwrap()),
    )
    .unwrap();
    let df = ctx.sql(AGGREGATES_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
    assert_eq!(
        batches[0]
            .column_by_name("denial_count")
            .unwrap()
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0),
        0
    );
}

/// `approvals_context` plus one `turn_dispatch` row: `turn-2` (the
/// unattended-fire refusal) was dispatched from a real occurrence. `turn-3`
/// (the attended-dispatch refusal) has no `turn_dispatch` row at all,
/// matching an attended dispatch's own turn-open path.
fn refusals_context() -> SessionContext {
    let ctx = approvals_context();
    let td_declared = CONVERSATION_EXECUTION_ENTRY
        .table(EXECUTION_TURN_DISPATCH)
        .expect("turn_dispatch table declared");
    let td_schema = arrow_schema_for(td_declared);
    let n = 1;
    let batch = RecordBatch::try_new(
        Arc::clone(&td_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a"; n])), // partition
            Arc::new(fixed_incarnation(n)),                 // source_incarnation
            Arc::new(UInt64Array::from(vec![10_u64])),      // position
            Arc::new(StringArray::from(vec!["turn-2"])),    // turn_id
            Arc::new(StringArray::from(vec!["daily-standup-1"])), // occurrence
            Arc::new(StringArray::from(vec!["direct"])),    // visibility
            Arc::new(StringArray::from(vec!["decided"])),   // visibility_source
            Arc::new(StringArray::from(vec![""])),          // source_turn_id
            Arc::new(StringArray::from(vec!["unknown"])),   // edge_asserted_visibility
        ],
    )
    .unwrap();
    ctx.register_table(
        "turn_dispatch",
        Arc::new(MemTable::try_new(td_schema, vec![vec![batch]]).unwrap()),
    )
    .unwrap();
    ctx
}

#[tokio::test]
async fn refusals_names_unattended_and_attended_dispatch_by_occurrence() {
    let ctx = refusals_context();
    let df = ctx.sql(REFUSALS_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    let mut by_position: std::collections::HashMap<u64, (String, String)> =
        std::collections::HashMap::new();
    for batch in &batches {
        let position = batch
            .column_by_name("position")
            .unwrap()
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap()
            .clone();
        let dispatch = batch
            .column_by_name("dispatch")
            .unwrap()
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap()
            .clone();
        let fire_id = batch
            .column_by_name("fire_id")
            .unwrap()
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap()
            .clone();
        for row in 0..batch.num_rows() {
            by_position.insert(
                position.value(row),
                (
                    dispatch.value(row).to_owned(),
                    fire_id.value(row).to_owned(),
                ),
            );
        }
    }
    assert_eq!(by_position.len(), 2, "both true refusals surface");
    assert_eq!(
        by_position[&2],
        ("unattended_fire".to_owned(), "daily-standup-1".to_owned())
    );
    assert_eq!(by_position[&3], ("attended".to_owned(), String::new()));
}

#[tokio::test]
async fn fire_dispatch_filters_by_occurrence() {
    let ctx = refusals_context();
    let sql = FIRE_DISPATCH_SQL.replace("$1", "'daily-standup-1'");
    let df = ctx.sql(&sql).await.unwrap();
    let batches = df.collect().await.unwrap();
    let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
    assert_eq!(rows, 1, "only turn-2 was dispatched from this occurrence");
    let turn_id = batches[0]
        .column_by_name("turn_id")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap()
        .value(0);
    assert_eq!(turn_id, "turn-2");
}

#[tokio::test]
async fn fire_count_is_uid_bound_and_excludes_pre_uid_rows() {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        ROUTINE_LIFECYCLE_ENTRY
            .table(ROUTINE_FIRES)
            .expect("fires table declared"),
    );
    let n = 3;
    let batch = RecordBatch::try_new(
        Arc::clone(&schema),
        vec![
            Arc::new(StringArray::from(vec!["routine-scheduler"; n])),
            Arc::new(fixed_incarnation(n)),
            Arc::new(UInt64Array::from(vec![0_u64, 1, 2])),
            Arc::new(StringArray::from(vec!["r-1", "r-1", "r-other"])),
            Arc::new(StringArray::from(vec!["u1", "u1", ""])), // routine_uid — last row pre-uid
            Arc::new(StringArray::from(vec!["o-1", "o-2", "o-3"])),
            Arc::new(UInt64Array::from(vec![0_u64, 0, 0])),
            Arc::new(UInt64Array::from(vec![1_000_u64, 2_000, 3_000])),
            Arc::new(BooleanArray::from(vec![true, true, true])),
            Arc::new(StringArray::from(vec!["ok", "ok", "ok"])),
            Arc::new(BooleanArray::from(vec![false, false, false])),
            Arc::new(StringArray::from(vec!["", "", ""])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "fires",
        Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()),
    )
    .unwrap();
    let sql = FIRE_COUNT_SQL.replace("$1", "'u1'");
    let df = ctx.sql(&sql).await.unwrap();
    let batches = df.collect().await.unwrap();
    assert_eq!(
        batches[0]
            .column_by_name("fire_count")
            .unwrap()
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0),
        2,
        "only u1's two fires count; the empty-uid row is a different, unrelated pre-uid fire"
    );
}

/// `ListRoutines`' latest-fire read (Epic 1565, 10C precursor): uid-bound,
/// latest by `fired_at_ms DESC, position DESC`, and a `NULL` outcome passes
/// through rather than being fabricated.
fn fires_context_for_last_fire() -> SessionContext {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        ROUTINE_LIFECYCLE_ENTRY
            .table(ROUTINE_FIRES)
            .expect("fires table declared"),
    );
    let n = 3;
    let batch = RecordBatch::try_new(
        Arc::clone(&schema),
        vec![
            Arc::new(StringArray::from(vec!["routine-scheduler"; n])),
            Arc::new(fixed_incarnation(n)),
            Arc::new(UInt64Array::from(vec![0_u64, 1, 0])),
            Arc::new(StringArray::from(vec!["r-1", "r-1", "r-other"])),
            Arc::new(StringArray::from(vec!["u1", "u1", "u2"])), // routine_uid
            Arc::new(StringArray::from(vec!["o-1", "o-2", "o-3"])),
            Arc::new(UInt64Array::from(vec![0_u64, 0, 0])),
            Arc::new(UInt64Array::from(vec![1_000_u64, 2_000, 9_000])), // fired_at_ms
            Arc::new(BooleanArray::from(vec![true, false, true])),      // has_outcome
            Arc::new(StringArray::from(vec!["ok", "", "ok"])),
            Arc::new(BooleanArray::from(vec![false, false, false])),
            Arc::new(StringArray::from(vec!["", "", ""])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "fires",
        Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()),
    )
    .unwrap();
    ctx
}

#[tokio::test]
async fn fire_last_is_uid_bound_and_picks_the_most_recent_by_fired_at_ms() {
    let ctx = fires_context_for_last_fire();
    let sql = FIRE_LAST_SQL.replace("$1", "'u1'");
    let df = ctx.sql(&sql).await.unwrap();
    let batches = df.collect().await.unwrap();
    let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
    assert_eq!(
        rows, 1,
        "exactly one row: the latest fire, never the history"
    );
    let fired_at_ms = batches
        .iter()
        .find(|batch| batch.num_rows() > 0)
        .unwrap()
        .column_by_name("fired_at_ms")
        .unwrap()
        .as_any()
        .downcast_ref::<UInt64Array>()
        .unwrap()
        .value(0);
    assert_eq!(
        fired_at_ms, 2_000,
        "u1's later fire, not u2's more recent one"
    );
    let outcome_column = batches
        .iter()
        .find(|batch| batch.num_rows() > 0)
        .unwrap()
        .column_by_name("outcome")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap()
        .clone();
    assert!(
        outcome_column.is_null(0),
        "has_outcome = false must read as SQL null, never a fabricated value"
    );
}

#[tokio::test]
async fn fire_last_over_a_never_fired_uid_returns_zero_rows() {
    let ctx = fires_context_for_last_fire();
    let sql = FIRE_LAST_SQL.replace("$1", "'no-such-uid'");
    let df = ctx.sql(&sql).await.unwrap();
    let batches = df.collect().await.unwrap();
    assert_eq!(
        batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
        0,
        "a never-fired routine returns zero rows, never a row of nulls"
    );
}

/// `ListRoutines`' stopped-tool read (Epic 1565, 10C precursor): the LATEST
/// denied, non-grant tool call by position -- reuses `approvals_context`,
/// the same three-row fixture `approval_aggregates_count_refusals_and_
/// unanswered_requests` proves against (positions 2 and 3 are true
/// refusals; position 3, `fs_write`, is later).
#[tokio::test]
async fn stopped_tool_picks_the_latest_denial_by_position() {
    let ctx = approvals_context();
    let df = ctx.sql(STOPPED_TOOL_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
    assert_eq!(rows, 1);
    let tool_name = batches
        .iter()
        .find(|batch| batch.num_rows() > 0)
        .unwrap()
        .column_by_name("tool_name")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap()
        .value(0);
    assert_eq!(
        tool_name, "fs_write",
        "position 3 is the latest true refusal; position 1 is unanswered, not denied"
    );
}

#[tokio::test]
async fn stopped_tool_over_zero_denials_returns_zero_rows() {
    let ctx = SessionContext::new();
    let schema = arrow_schema_for(
        CONVERSATION_SECURITY_ENTRY
            .table(SECURITY_APPROVALS)
            .expect("approvals table declared"),
    );
    let empty = RecordBatch::new_empty(Arc::clone(&schema));
    ctx.register_table(
        "approvals",
        Arc::new(MemTable::try_new(schema, vec![vec![empty]]).unwrap()),
    )
    .unwrap();
    let df = ctx.sql(STOPPED_TOOL_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    assert_eq!(
        batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
        0,
        "no denial ever happened; ListRoutines reads this as its own empty-string zero value"
    );
}