polyc-query 2026.10.2

//! Proves the fixed SQL statement `crate::control_plane::dashboard`
//! (`crates/control-plane/src/forensics.rs`'s `DASHBOARD_CONTEXT_SQL`) sends
//! to the Fleet Query plane for `/api/dashboard`'s per-conversation context
//! (POLY-392) — against real `DataFusion` `MemTable`s built from the exact
//! `conversation-core/v1`, `conversation-execution/v1`, and
//! `conversation-security/v1` schemas the registry declares, not a live
//! projected stack.
//!
//! Mirrors `dashboard_attribution_sql.rs`'s own shape and disclaimer: this is
//! a syntax-and-semantics proof for the SQL text itself, not the Query
//! plane's authority, resolution, or transport.

#![allow(clippy::unwrap_used)]

use std::sync::Arc;

use arrow::array::{BooleanArray, ListArray, RecordBatch, StringArray, UInt64Array};
use datafusion::datasource::MemTable;
use datafusion::prelude::SessionContext;
use polyc_projection::family::{
    CONVERSATION_CORE_ENTRY, CONVERSATION_EXECUTION_ENTRY, CONVERSATION_MESSAGES,
    CONVERSATION_SECURITY_ENTRY, CONVERSATION_TURNS, EXECUTION_SUMMARY, SECURITY_ATTRIBUTION,
    SECURITY_ATTRIBUTION_PROVENANCE,
};

/// The statement under test — the exact constant `crate::forensics`
/// (`crates/control-plane/src/forensics.rs`) runs in production.
const CONTEXT_SQL: &str = polyc_query_model::statements::DASHBOARD_CONTEXT_SQL;

/// The same logical-to-Arrow mapping `polyc_projector::artifact::arrow_schema`
/// applies, restated here rather than imported — see
/// `financial_dashboard_sql.rs`'s own `schema` for why a Component cannot
/// depend on `polyc-projector`, a Container.
fn schema(
    entry: polyc_projection::family::FamilyEntry,
    table: polyc_projection::family::TableId,
) -> arrow::datatypes::SchemaRef {
    use polyc_projection::family::LogicalType;

    let declared = entry.table(table).expect("table declared");
    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()
}

/// `conv-a` carries two committed turns (the second, `turn-2`, has the
/// greater `complete_position` and must win), three messages (a `system`
/// row at the lowest position that must be skipped, an `internal_only`
/// `user` row that must also be skipped, and the real opening `user` row),
/// two summaries (the one with the greater `covers_through_position` must
/// win), and attribution posted through two distinct edges. `conv-b` carries
/// exactly one of each, proving the single-row case. `conv-c` carries
/// nothing in any of the four tables and must be entirely absent from the
/// result — the dashboard still lists it from `dashboard_conversations.sql`;
/// this statement just contributes nothing for it.
#[allow(
    clippy::too_many_lines,
    reason = "one MemTable per one of the five tables this statement reads, each already a \
              single clearly-named fixture; splitting further would only add indirection for \
              one test-only call site"
)]
fn context() -> SessionContext {
    let ctx = SessionContext::new();

    let turns_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_TURNS);
    let turns = RecordBatch::try_new(
        Arc::clone(&turns_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a", "conv-a", "conv-b"])),
            Arc::new(fixed_incarnation(3)),
            Arc::new(StringArray::from(vec!["turn-1", "turn-2", "turn-only"])),
            Arc::new(UInt64Array::from(vec![0_u64, 10, 0])),
            Arc::new(UInt64Array::from(vec![1_u64, 11, 1])),
            Arc::new(UInt64Array::from(vec![2_u64, 12, 2])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "turns",
        Arc::new(MemTable::try_new(turns_schema, vec![vec![turns]]).unwrap()),
    )
    .unwrap();

    let messages_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_MESSAGES);
    let messages = RecordBatch::try_new(
        Arc::clone(&messages_schema),
        vec![
            Arc::new(StringArray::from(vec![
                "conv-a", "conv-a", "conv-a", "conv-b",
            ])),
            Arc::new(fixed_incarnation(4)),
            Arc::new(UInt64Array::from(vec![0_u64, 1, 2, 0])),
            Arc::new(StringArray::from(vec![
                "turn-1",
                "turn-1",
                "turn-1",
                "turn-only",
            ])),
            Arc::new(StringArray::from(vec!["system", "user", "user", "user"])),
            Arc::new(BooleanArray::from(vec![false, true, false, false])),
            Arc::new(BooleanArray::from(vec![false; 4])),
            Arc::new(StringArray::from(vec![
                "context note",
                "internal draft, must be skipped",
                "hello from conv-a, the real opening message",
                "hi from conv-b",
            ])),
            Arc::new(StringArray::from(vec![
                "trusted", "trusted", "trusted", "trusted",
            ])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "messages",
        Arc::new(MemTable::try_new(messages_schema, vec![vec![messages]]).unwrap()),
    )
    .unwrap();

    let summary_schema = schema(CONVERSATION_EXECUTION_ENTRY, EXECUTION_SUMMARY);
    let summary = RecordBatch::try_new(
        Arc::clone(&summary_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a", "conv-a", "conv-b"])),
            Arc::new(fixed_incarnation(3)),
            Arc::new(UInt64Array::from(vec![0_u64, 5, 0])),
            Arc::new(StringArray::from(vec!["sum-1", "sum-2", "sum-only"])),
            Arc::new(StringArray::from(vec![
                "stale summary",
                "latest summary",
                "conv-b's only summary",
            ])),
            Arc::new(UInt64Array::from(vec![3_u64, 12, 2])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "summary",
        Arc::new(MemTable::try_new(summary_schema, vec![vec![summary]]).unwrap()),
    )
    .unwrap();

    let attribution_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION);
    let attribution = RecordBatch::try_new(
        Arc::clone(&attribution_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a", "conv-a", "conv-b"])),
            Arc::new(fixed_incarnation(3)),
            Arc::new(UInt64Array::from(vec![0_u64, 1, 0])),
            Arc::new(StringArray::from(vec!["turn-1", "turn-1", "turn-only"])),
            Arc::new(StringArray::from(vec![
                "persona-a",
                "persona-b",
                "persona-c",
            ])),
            Arc::new(StringArray::from(vec![
                "initiator",
                "participant",
                "initiator",
            ])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "attribution",
        Arc::new(MemTable::try_new(attribution_schema, vec![vec![attribution]]).unwrap()),
    )
    .unwrap();

    let provenance_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION_PROVENANCE);
    let provenance = RecordBatch::try_new(
        Arc::clone(&provenance_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-a", "conv-a", "conv-b"])),
            Arc::new(fixed_incarnation(3)),
            Arc::new(UInt64Array::from(vec![0_u64, 1, 0])),
            Arc::new(StringArray::from(vec!["team", "team", "team"])),
            Arc::new(StringArray::from(vec!["team", "team", "team"])),
            Arc::new(StringArray::from(vec!["u1", "u2", "u3"])),
            Arc::new(StringArray::from(vec!["Alice", "Bob", "Carol"])),
            // conv-a's two attribution rows post through two DISTINCT edges;
            // a fanout or dedup bug in `array_agg` would show up as a
            // three-or-one-element array instead of the correct two.
            Arc::new(StringArray::from(vec!["edge-2", "edge-1", "edge-3"])),
            Arc::new(StringArray::from(vec!["", "", ""])),
            Arc::new(StringArray::from(vec!["", "", ""])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "attribution_provenance",
        Arc::new(MemTable::try_new(provenance_schema, vec![vec![provenance]]).unwrap()),
    )
    .unwrap();

    ctx
}

fn string_list_value(array: &ListArray, row: usize) -> Vec<String> {
    let value = array.value(row);
    let strings = value.as_any().downcast_ref::<StringArray>().unwrap();
    strings.iter().map(|s| s.unwrap().to_owned()).collect()
}

#[tokio::test]
async fn one_row_per_conversation_with_latest_turn_opening_message_summary_and_edges() {
    let ctx = context();
    let df = ctx.sql(CONTEXT_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
    // conv-a (two turns, three messages, two summaries, two attribution
    // rows) and conv-b (one of each) each collapse to exactly one row;
    // conv-c never appears — a fan-out bug would report more than one row
    // for conv-a.
    assert_eq!(rows, 2, "one row per conversation carrying any context");

    let batch = &batches[0];
    let ids = batch
        .column_by_name("id")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let latest_turn_ids = batch
        .column_by_name("latest_turn_id")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let previews = batch
        .column_by_name("opening_message_preview")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let summaries = batch
        .column_by_name("latest_summary_text")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let edge_ids = batch
        .column_by_name("edge_ids")
        .unwrap()
        .as_any()
        .downcast_ref::<ListArray>()
        .unwrap();

    let a = ids
        .iter()
        .position(|id| id == Some("conv-a"))
        .expect("conv-a present");
    assert_eq!(
        latest_turn_ids.value(a),
        "turn-2",
        "turn-2 has the greater complete_position"
    );
    assert_eq!(
        previews.value(a),
        "hello from conv-a, the real opening message",
        "the system row and the internal_only user row are both skipped"
    );
    assert_eq!(
        summaries.value(a),
        "latest summary",
        "sum-2 has the greater covers_through_position"
    );
    assert_eq!(
        string_list_value(edge_ids, a),
        vec!["edge-1".to_owned(), "edge-2".to_owned()],
        "conv-a's two attribution rows post through two distinct edges, sorted"
    );

    let b = ids
        .iter()
        .position(|id| id == Some("conv-b"))
        .expect("conv-b present");
    assert_eq!(latest_turn_ids.value(b), "turn-only");
    assert_eq!(previews.value(b), "hi from conv-b");
    assert_eq!(summaries.value(b), "conv-b's only summary");
    assert_eq!(string_list_value(edge_ids, b), vec!["edge-3".to_owned()]);
}

#[tokio::test]
async fn opening_message_preview_is_cut_to_two_hundred_characters() {
    let ctx = SessionContext::new();

    let turns_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_TURNS);
    let turns = RecordBatch::try_new(
        Arc::clone(&turns_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-long"])),
            Arc::new(fixed_incarnation(1)),
            Arc::new(StringArray::from(vec!["turn-1"])),
            Arc::new(UInt64Array::from(vec![0_u64])),
            Arc::new(UInt64Array::from(vec![1_u64])),
            Arc::new(UInt64Array::from(vec![2_u64])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "turns",
        Arc::new(MemTable::try_new(turns_schema, vec![vec![turns]]).unwrap()),
    )
    .unwrap();

    let long_text: String = "a".repeat(5_000);
    let messages_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_MESSAGES);
    let messages = RecordBatch::try_new(
        Arc::clone(&messages_schema),
        vec![
            Arc::new(StringArray::from(vec!["conv-long"])),
            Arc::new(fixed_incarnation(1)),
            Arc::new(UInt64Array::from(vec![0_u64])),
            Arc::new(StringArray::from(vec!["turn-1"])),
            Arc::new(StringArray::from(vec!["user"])),
            Arc::new(BooleanArray::from(vec![false])),
            Arc::new(BooleanArray::from(vec![false])),
            Arc::new(StringArray::from(vec![long_text.as_str()])),
            Arc::new(StringArray::from(vec!["trusted"])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "messages",
        Arc::new(MemTable::try_new(messages_schema, vec![vec![messages]]).unwrap()),
    )
    .unwrap();

    let summary_schema = schema(CONVERSATION_EXECUTION_ENTRY, EXECUTION_SUMMARY);
    let summary = RecordBatch::new_empty(summary_schema.clone());
    ctx.register_table(
        "summary",
        Arc::new(MemTable::try_new(summary_schema, vec![vec![summary]]).unwrap()),
    )
    .unwrap();

    let attribution_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION);
    let attribution = RecordBatch::new_empty(attribution_schema.clone());
    ctx.register_table(
        "attribution",
        Arc::new(MemTable::try_new(attribution_schema, vec![vec![attribution]]).unwrap()),
    )
    .unwrap();

    let provenance_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION_PROVENANCE);
    let provenance = RecordBatch::new_empty(provenance_schema.clone());
    ctx.register_table(
        "attribution_provenance",
        Arc::new(MemTable::try_new(provenance_schema, vec![vec![provenance]]).unwrap()),
    )
    .unwrap();

    let df = ctx.sql(CONTEXT_SQL).await.unwrap();
    let batches = df.collect().await.unwrap();
    let batch = &batches[0];
    let previews = batch
        .column_by_name("opening_message_preview")
        .unwrap()
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    assert_eq!(
        previews.value(0).chars().count(),
        200,
        "the statement itself cuts the preview to 200 characters — the wire never carries more"
    );
}

/// Directional timing at a synthetic-fleet scale: 1,000 conversations, 100
/// messages each (100,000 `messages` rows — the table none of the three
/// earlier dashboard statements reads, per `ExactParquetTable`'s own
/// doc — `crates/query/src/core_execution/provider.rs` — pushing projection
/// and `LIMIT` but never a filter, so this statement's `WHERE role = 'user'`
/// predicate cannot prune the scan), two summaries and two attribution rows
/// each.
///
/// Ignored by default: `cargo nextest run -p polyc-query --test
/// dashboard_context_sql -- --ignored` (or `cargo test -- --ignored`).
/// SQL-planning-and-execution time over in-memory Arrow batches only — it
/// does NOT include the request path's Parquet reads from object storage,
/// credential/grant admission, or JSON encoding, so it is a lower bound on
/// the real `/api/dashboard` request's added cost, not the number itself.
/// No harness in this repository seeds a real projected fleet at this
/// scale (`crates/query-service/tests/production_composition.rs` composes
/// real planes but not at fleet size), so this is the cheapest real signal
/// available without building one. Observed at introduction (debug profile,
/// developer machine, five runs): median 589.8ms, worst 605.2ms — the SQL
/// itself is not pathologically slow at this scale, but this excludes the
/// Parquet-read and network cost `ExactParquetTable`'s unfiltered scan adds
/// in production, which is unmeasured.
#[tokio::test]
#[ignore = "synthetic-scale timing probe, not a correctness test — run manually"]
#[allow(
    clippy::too_many_lines,
    reason = "one MemTable per one of the five tables this statement reads, at fleet scale, \
              each already a single clearly-named fixture; splitting further would only add \
              indirection for one test-only call site"
)]
async fn dashboard_context_sql_at_synthetic_fleet_scale() {
    const CONVERSATIONS: usize = 1_000;
    const MESSAGES_PER_CONVERSATION: usize = 100;
    const RUNS: usize = 5;

    let ctx = SessionContext::new();
    let ids: Vec<String> = (0..CONVERSATIONS).map(|i| format!("conv-{i:05}")).collect();

    let turns_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_TURNS);
    let turns = RecordBatch::try_new(
        Arc::clone(&turns_schema),
        vec![
            Arc::new(StringArray::from(ids.clone())),
            Arc::new(fixed_incarnation(CONVERSATIONS)),
            Arc::new(StringArray::from(
                ids.iter()
                    .map(|id| format!("{id}-turn"))
                    .collect::<Vec<_>>(),
            )),
            Arc::new(UInt64Array::from(vec![0_u64; CONVERSATIONS])),
            Arc::new(UInt64Array::from(vec![1_u64; CONVERSATIONS])),
            Arc::new(UInt64Array::from(vec![2_u64; CONVERSATIONS])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "turns",
        Arc::new(MemTable::try_new(turns_schema, vec![vec![turns]]).unwrap()),
    )
    .unwrap();

    let total_messages = CONVERSATIONS * MESSAGES_PER_CONVERSATION;
    let message_partitions: Vec<&str> = ids
        .iter()
        .flat_map(|id| std::iter::repeat_n(id.as_str(), MESSAGES_PER_CONVERSATION))
        .collect();
    let message_turn_ids: Vec<String> = ids
        .iter()
        .flat_map(|id| std::iter::repeat_n(format!("{id}-turn"), MESSAGES_PER_CONVERSATION))
        .collect();
    let message_positions: Vec<u64> = (0..CONVERSATIONS)
        .flat_map(|_| 0..u64::try_from(MESSAGES_PER_CONVERSATION).unwrap())
        .collect();
    let message_roles: Vec<&str> = (0..total_messages).map(|_| "user").collect();
    let message_text: String = "m".repeat(400);
    let message_texts: Vec<&str> = (0..total_messages).map(|_| message_text.as_str()).collect();
    let message_trust: Vec<&str> = (0..total_messages).map(|_| "trusted").collect();

    let messages_schema = schema(CONVERSATION_CORE_ENTRY, CONVERSATION_MESSAGES);
    let messages = RecordBatch::try_new(
        Arc::clone(&messages_schema),
        vec![
            Arc::new(StringArray::from(message_partitions)),
            Arc::new(fixed_incarnation(total_messages)),
            Arc::new(UInt64Array::from(message_positions)),
            Arc::new(StringArray::from(message_turn_ids)),
            Arc::new(StringArray::from(message_roles)),
            Arc::new(BooleanArray::from(vec![false; total_messages])),
            Arc::new(BooleanArray::from(vec![false; total_messages])),
            Arc::new(StringArray::from(message_texts)),
            Arc::new(StringArray::from(message_trust)),
        ],
    )
    .unwrap();
    ctx.register_table(
        "messages",
        Arc::new(MemTable::try_new(messages_schema, vec![vec![messages]]).unwrap()),
    )
    .unwrap();

    let summary_partitions: Vec<&str> = ids
        .iter()
        .flat_map(|id| std::iter::repeat_n(id.as_str(), 2))
        .collect();
    let summary_count = summary_partitions.len();
    let summary_schema = schema(CONVERSATION_EXECUTION_ENTRY, EXECUTION_SUMMARY);
    let summary = RecordBatch::try_new(
        Arc::clone(&summary_schema),
        vec![
            Arc::new(StringArray::from(summary_partitions)),
            Arc::new(fixed_incarnation(summary_count)),
            Arc::new(UInt64Array::from(
                (0..CONVERSATIONS)
                    .flat_map(|_| [0_u64, 1])
                    .collect::<Vec<_>>(),
            )),
            Arc::new(StringArray::from(vec!["sum"; summary_count])),
            Arc::new(StringArray::from(vec!["summary text"; summary_count])),
            Arc::new(UInt64Array::from(
                (0..CONVERSATIONS)
                    .flat_map(|_| [0_u64, 1])
                    .collect::<Vec<_>>(),
            )),
        ],
    )
    .unwrap();
    ctx.register_table(
        "summary",
        Arc::new(MemTable::try_new(summary_schema, vec![vec![summary]]).unwrap()),
    )
    .unwrap();

    let attribution_partitions: Vec<&str> = ids
        .iter()
        .flat_map(|id| std::iter::repeat_n(id.as_str(), 2))
        .collect();
    let attribution_count = attribution_partitions.len();
    let attribution_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION);
    let attribution = RecordBatch::try_new(
        Arc::clone(&attribution_schema),
        vec![
            Arc::new(StringArray::from(attribution_partitions.clone())),
            Arc::new(fixed_incarnation(attribution_count)),
            Arc::new(UInt64Array::from(
                (0..CONVERSATIONS)
                    .flat_map(|_| [0_u64, 1])
                    .collect::<Vec<_>>(),
            )),
            Arc::new(StringArray::from(vec!["turn"; attribution_count])),
            Arc::new(StringArray::from(vec!["persona"; attribution_count])),
            Arc::new(StringArray::from(vec!["initiator"; attribution_count])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "attribution",
        Arc::new(MemTable::try_new(attribution_schema, vec![vec![attribution]]).unwrap()),
    )
    .unwrap();

    let provenance_schema = schema(CONVERSATION_SECURITY_ENTRY, SECURITY_ATTRIBUTION_PROVENANCE);
    let provenance = RecordBatch::try_new(
        Arc::clone(&provenance_schema),
        vec![
            Arc::new(StringArray::from(attribution_partitions)),
            Arc::new(fixed_incarnation(attribution_count)),
            Arc::new(UInt64Array::from(
                (0..CONVERSATIONS)
                    .flat_map(|_| [0_u64, 1])
                    .collect::<Vec<_>>(),
            )),
            Arc::new(StringArray::from(vec!["team"; attribution_count])),
            Arc::new(StringArray::from(vec!["team"; attribution_count])),
            Arc::new(StringArray::from(vec!["u"; attribution_count])),
            Arc::new(StringArray::from(vec!["Name"; attribution_count])),
            Arc::new(StringArray::from(vec!["edge-1"; attribution_count])),
            Arc::new(StringArray::from(vec![""; attribution_count])),
            Arc::new(StringArray::from(vec![""; attribution_count])),
        ],
    )
    .unwrap();
    ctx.register_table(
        "attribution_provenance",
        Arc::new(MemTable::try_new(provenance_schema, vec![vec![provenance]]).unwrap()),
    )
    .unwrap();

    let mut durations = Vec::with_capacity(RUNS);
    for _ in 0..RUNS {
        let started = std::time::Instant::now();
        let df = ctx.sql(CONTEXT_SQL).await.unwrap();
        let batches = df.collect().await.unwrap();
        let elapsed = started.elapsed();
        let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
        assert_eq!(rows, CONVERSATIONS, "one row per conversation, at scale");
        durations.push(elapsed);
    }
    durations.sort();
    eprintln!(
        "dashboard_context.sql at {CONVERSATIONS} conversations x \
         {MESSAGES_PER_CONVERSATION} messages: median {:?}, worst {:?} over {RUNS} runs",
        durations[RUNS / 2],
        durations[RUNS - 1]
    );
}