polyc-query-model 2026.9.2

//! `DataFusion`-free vocabulary for the versioned Query protocol.
//!
//! This crate contains only semantic requests and result frames.
//!
//! The REQUEST grants no scope. It names no realm, namespace, partition,
//! table, source, or object. The Query service derives every one of those
//! facts from the credential it verifies at its own boundary.
//!
//! The TERMINAL FRAME is the opposite by design. It carries State's complete
//! source evidence, so `SourceSnapshot` transitively reaches State's
//! manifest and object vocabulary: projection keys, source checkpoints, object
//! descriptors, and namespaces. That evidence flows from the server to the
//! caller and grants the caller nothing. Do not read the small re-export list
//! below as a small type graph.

use std::fmt;
use std::time::Duration;

pub mod evidence;

pub use evidence::{
    ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES, COMMIT_ROOT_BYTES, Classification,
    DIGEST_BYTES, ExactObjectRef, INCARNATION_BYTES, JournalAnchor, JournalAttestation,
    JournalSource, MAX_SOURCE_PINS, ObjectDescriptor, ProjectionKey, ProjectionManifest,
    PublisherFence, Retention, SourceCheckpoint, SourceEvidence, SourcePin,
};

/// How a query ended, as the durable audit recorded it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueryOutcome {
    /// Every requested row was released.
    Succeeded,
    /// The query stopped, and this class says who is responsible.
    Failed(ErrorClass),
}

/// Who is responsible for a query that did not succeed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorClass {
    /// Current authority refused this caller.
    Denied,
    /// The declared execution deadline expired.
    Deadline,
    /// The caller withdrew.
    Cancelled,
    /// A declared resource ceiling was reached.
    Bounds,
    /// A required source could not answer.
    Unavailable,
    /// The statement, its parameters, or the plan could not be used.
    Malformed,
    /// Corruption, or a fault in the serving plane or its deployment.
    Internal,
}

/// Whether a result carries every matching row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Truncation {
    /// Every matching row was released.
    Complete,
    /// Release stopped at this row count with rows still available.
    TruncatedAt(u64),
}

/// The only protocol version this build speaks.
pub const PROTOCOL_VERSION: u32 = 1;
/// Largest UTF-8 SQL statement accepted by the semantic boundary.
pub const MAX_SQL_BYTES: usize = 64 * 1024;
/// Largest number of positional parameters in one request.
pub const MAX_PARAMETERS: usize = 256;
/// Largest total UTF-8 parameter payload in one request.
pub const MAX_PARAMETER_BYTES: usize = 64 * 1024;
/// Largest caller-requested execution duration.
pub const MAX_TIMEOUT: Duration = Duration::from_mins(5);
/// Largest caller-requested result row count.
pub const MAX_ROWS: u64 = 1_000_000;
/// Largest caller-requested released result size.
pub const MAX_RESULT_BYTES: u64 = 64 * 1024 * 1024;
/// Largest caller-requested result frame.
pub const MAX_FRAME_BYTES: u64 = 4 * 1024 * 1024;

/// One request parameter from the protocol's closed vocabulary.
///
/// `Debug` reports the kind only. A parameter value is caller content, so it
/// never reaches a log line through this type.
#[derive(Clone, PartialEq, Eq)]
pub enum Parameter {
    /// A UTF-8 string.
    Utf8(String),
    /// An unsigned 64-bit integer.
    UInt64(u64),
    /// A Boolean.
    Boolean(bool),
    /// An explicit SQL null.
    Null,
}

impl fmt::Debug for Parameter {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let kind = match self {
            Self::Utf8(_) => "utf8",
            Self::UInt64(_) => "uint64",
            Self::Boolean(_) => "boolean",
            Self::Null => "null",
        };
        formatter.write_str(kind)
    }
}

/// Projection consistency requested by a caller.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Consistency {
    /// Read the currently published projection.
    Projected,
    /// Refuse unless the projection covers at least this journal position.
    ///
    /// The position names no partition or source. A result may pin several
    /// independent sources, so this scalar has no single referent across them.
    /// The gate that recognizes this posture refuses it before any artifact
    /// read; a later chunk owns both the target vector and its wait semantics.
    RequireProjectedThrough(u64),
}

/// Caller-requested ceilings. A deployment may only narrow them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RequestedBounds {
    timeout: Duration,
    rows: u64,
    result_bytes: u64,
    frame_bytes: u64,
}

impl RequestedBounds {
    /// Validates all four caller ceilings.
    ///
    /// # Errors
    ///
    /// Returns [`ModelError::Bounds`] when a ceiling is zero or exceeds the
    /// protocol's compile-time maximum.
    pub fn try_new(
        timeout: Duration,
        rows: u64,
        result_bytes: u64,
        frame_bytes: u64,
    ) -> Result<Self, ModelError> {
        if timeout.is_zero() || timeout > MAX_TIMEOUT {
            return Err(ModelError::Bounds("timeout_nanos"));
        }
        if rows == 0 || rows > MAX_ROWS {
            return Err(ModelError::Bounds("rows"));
        }
        if result_bytes == 0 || result_bytes > MAX_RESULT_BYTES {
            return Err(ModelError::Bounds("result_bytes"));
        }
        if frame_bytes == 0 || frame_bytes > MAX_FRAME_BYTES {
            return Err(ModelError::Bounds("frame_bytes"));
        }
        Ok(Self {
            timeout,
            rows,
            result_bytes,
            frame_bytes,
        })
    }

    /// Returns the execution timeout ceiling.
    #[must_use]
    pub const fn timeout(self) -> Duration {
        self.timeout
    }

    /// Returns the row ceiling.
    #[must_use]
    pub const fn rows(self) -> u64 {
        self.rows
    }

    /// Returns the released result byte ceiling.
    #[must_use]
    pub const fn result_bytes(self) -> u64 {
        self.result_bytes
    }

    /// Returns the per-frame byte ceiling.
    #[must_use]
    pub const fn frame_bytes(self) -> u64 {
        self.frame_bytes
    }
}

/// A version-independent semantic Query request.
///
/// `Debug` reports the statement's byte length and the parameter count. The
/// statement text and every parameter value are caller content, so neither
/// reaches a log line through this type.
#[derive(Clone, PartialEq, Eq)]
pub struct QueryRequest {
    sql: String,
    parameters: Vec<Parameter>,
    consistency: Consistency,
    bounds: RequestedBounds,
}

impl QueryRequest {
    /// Validates a complete request.
    ///
    /// # Errors
    ///
    /// Returns a typed refusal for empty or over-bound SQL and parameters.
    pub fn try_new(
        sql: String,
        parameters: Vec<Parameter>,
        consistency: Consistency,
        bounds: RequestedBounds,
    ) -> Result<Self, ModelError> {
        if sql.is_empty() || sql.len() > MAX_SQL_BYTES {
            return Err(ModelError::Bounds("sql"));
        }
        if parameters.len() > MAX_PARAMETERS {
            return Err(ModelError::Bounds("parameters"));
        }
        let parameter_bytes = parameters.iter().try_fold(0_usize, |total, parameter| {
            let bytes = match parameter {
                Parameter::Utf8(value) => value.len(),
                Parameter::UInt64(_) | Parameter::Boolean(_) | Parameter::Null => 0,
            };
            total
                .checked_add(bytes)
                .ok_or(ModelError::Bounds("parameters"))
        })?;
        if parameter_bytes > MAX_PARAMETER_BYTES {
            return Err(ModelError::Bounds("parameters"));
        }
        Ok(Self {
            sql,
            parameters,
            consistency,
            bounds,
        })
    }

    /// Returns the SQL text.
    #[must_use]
    pub fn sql(&self) -> &str {
        &self.sql
    }

    /// Returns positional parameters.
    #[must_use]
    pub fn parameters(&self) -> &[Parameter] {
        &self.parameters
    }

    /// Returns requested consistency.
    #[must_use]
    pub const fn consistency(&self) -> Consistency {
        self.consistency
    }

    /// Returns caller-requested bounds.
    #[must_use]
    pub const fn bounds(&self) -> RequestedBounds {
        self.bounds
    }
}

impl fmt::Debug for QueryRequest {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("QueryRequest")
            .field("sql_bytes", &self.sql.len())
            .field("parameters", &self.parameters.len())
            .field("consistency", &self.consistency)
            .field("bounds", &self.bounds)
            .finish()
    }
}

/// One Arrow IPC schema frame.
///
/// `Debug` reports the encoded length only. The bytes describe result columns.
#[derive(Clone, PartialEq, Eq)]
pub struct SchemaFrame {
    arrow_ipc: Vec<u8>,
}

impl SchemaFrame {
    /// Validates one opaque Arrow IPC schema frame.
    ///
    /// # Errors
    ///
    /// Returns [`ModelError::Bounds`] for an empty or over-bound frame.
    pub fn try_new(arrow_ipc: Vec<u8>) -> Result<Self, ModelError> {
        validate_frame_bytes(&arrow_ipc)?;
        Ok(Self { arrow_ipc })
    }

    /// Returns the opaque Arrow IPC schema bytes.
    #[must_use]
    pub fn arrow_ipc(&self) -> &[u8] {
        &self.arrow_ipc
    }
}

impl fmt::Debug for SchemaFrame {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SchemaFrame")
            .field("arrow_ipc_bytes", &self.arrow_ipc.len())
            .finish()
    }
}

/// One ordered Arrow IPC data frame.
///
/// `Debug` reports the sequence, row count, and encoded length. The bytes are
/// released rows, so they never reach a log line through this type.
#[derive(Clone, PartialEq, Eq)]
pub struct DataFrame {
    sequence: u64,
    rows: u64,
    arrow_ipc: Vec<u8>,
}

impl DataFrame {
    /// Validates one ordered opaque Arrow IPC data frame.
    ///
    /// # Errors
    ///
    /// Returns [`ModelError::Bounds`] for an empty or over-bound frame.
    pub fn try_new(sequence: u64, rows: u64, arrow_ipc: Vec<u8>) -> Result<Self, ModelError> {
        validate_frame_bytes(&arrow_ipc)?;
        Ok(Self {
            sequence,
            rows,
            arrow_ipc,
        })
    }

    /// Returns the zero-based frame sequence.
    #[must_use]
    pub const fn sequence(&self) -> u64 {
        self.sequence
    }

    /// Returns the number of rows encoded in the frame.
    #[must_use]
    pub const fn rows(&self) -> u64 {
        self.rows
    }

    /// Returns opaque Arrow IPC record-batch bytes.
    #[must_use]
    pub fn arrow_ipc(&self) -> &[u8] {
        &self.arrow_ipc
    }
}

impl fmt::Debug for DataFrame {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DataFrame")
            .field("sequence", &self.sequence)
            .field("rows", &self.rows)
            .field("arrow_ipc_bytes", &self.arrow_ipc.len())
            .finish()
    }
}

/// Final audited result facts.
///
/// `Debug` reports counts and outcome only. The source evidence names object
/// keys, namespaces, partitions, signer keys, and signatures, so it never
/// reaches a log line through this type.
#[derive(Clone, PartialEq, Eq)]
pub struct TerminalFrame {
    outcome: QueryOutcome,
    duration: Duration,
    rows: u64,
    result_bytes: u64,
    truncation: Truncation,
    source: SourceEvidence,
}

impl TerminalFrame {
    /// Records the exact durable completion facts returned by Query.
    #[must_use]
    pub const fn new(
        outcome: QueryOutcome,
        duration: Duration,
        rows: u64,
        result_bytes: u64,
        truncation: Truncation,
        source: SourceEvidence,
    ) -> Self {
        Self {
            outcome,
            duration,
            rows,
            result_bytes,
            truncation,
            source,
        }
    }

    /// Returns the durable completion outcome.
    #[must_use]
    pub const fn outcome(&self) -> QueryOutcome {
        self.outcome
    }

    /// Returns measured execution duration.
    #[must_use]
    pub const fn duration(&self) -> Duration {
        self.duration
    }

    /// Returns total rows released across data frames.
    #[must_use]
    pub const fn rows(&self) -> u64 {
        self.rows
    }

    /// Returns total opaque data bytes released across data frames.
    #[must_use]
    pub const fn result_bytes(&self) -> u64 {
        self.result_bytes
    }

    /// Returns whether the result was complete or truncated.
    #[must_use]
    pub const fn truncation(&self) -> Truncation {
        self.truncation
    }

    /// Returns exact canonical source premises recorded by Query audit.
    #[must_use]
    pub const fn source(&self) -> &SourceEvidence {
        &self.source
    }
}

impl fmt::Debug for TerminalFrame {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("TerminalFrame")
            .field("outcome", &self.outcome)
            .field("duration", &self.duration)
            .field("rows", &self.rows)
            .field("result_bytes", &self.result_bytes)
            .field("truncation", &self.truncation)
            .field("source_pins", &self.source.pins().len())
            .finish()
    }
}

fn validate_frame_bytes(bytes: &[u8]) -> Result<(), ModelError> {
    if bytes.is_empty() || u64::try_from(bytes.len()).unwrap_or(u64::MAX) > MAX_FRAME_BYTES {
        return Err(ModelError::Bounds("arrow_ipc"));
    }
    Ok(())
}

/// One version-independent result stream frame.
///
/// Every variant's own `Debug` reports shape only, so this derive carries no
/// statement, parameter, row byte, or source identifier.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResultFrame {
    /// The required first frame.
    Schema(SchemaFrame),
    /// An ordered data frame.
    Data(DataFrame),
    /// The required final frame.
    Terminal(TerminalFrame),
}

/// Semantic request validation failure.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum ModelError {
    /// One field is empty or outside its closed bound.
    #[error("query field `{0}` is outside its protocol bound")]
    Bounds(&'static str),
    /// A canonically ordered vector is unsorted or names one source twice.
    #[error("query field `{0}` is not in strict canonical order")]
    Order(&'static str),
}

#[cfg(test)]
mod tests {
    use super::*;

    fn bounds() -> RequestedBounds {
        RequestedBounds::try_new(Duration::from_secs(1), 10, 1024, 512).unwrap()
    }

    #[test]
    fn request_refuses_empty_and_overbound_content() {
        assert_eq!(
            QueryRequest::try_new(String::new(), vec![], Consistency::Projected, bounds()),
            Err(ModelError::Bounds("sql"))
        );
        assert_eq!(
            QueryRequest::try_new(
                "select ?".into(),
                vec![Parameter::Utf8("x".repeat(MAX_PARAMETER_BYTES + 1))],
                Consistency::Projected,
                bounds(),
            ),
            Err(ModelError::Bounds("parameters"))
        );
    }

    #[test]
    fn bounds_refuse_zero_and_crossed_frame_limits() {
        assert_eq!(
            RequestedBounds::try_new(Duration::ZERO, 1, 1, 1),
            Err(ModelError::Bounds("timeout_nanos"))
        );
        assert_eq!(
            RequestedBounds::try_new(Duration::from_secs(1), 1, 8, MAX_FRAME_BYTES + 1),
            Err(ModelError::Bounds("frame_bytes"))
        );
    }

    #[test]
    fn debug_output_carries_no_caller_content() {
        let request = QueryRequest::try_new(
            "select secret_column from messages".into(),
            vec![Parameter::Utf8("tenant-secret".into())],
            Consistency::Projected,
            bounds(),
        )
        .unwrap();
        let rendered = format!("{request:?}");
        assert!(!rendered.contains("secret_column"), "{rendered}");
        assert!(!rendered.contains("tenant-secret"), "{rendered}");
        assert!(rendered.contains("sql_bytes"), "{rendered}");

        let data = DataFrame::try_new(0, 2, vec![7, 8, 9]).unwrap();
        let rendered = format!("{data:?}");
        assert!(!rendered.contains('7'), "{rendered}");
        assert!(rendered.contains("arrow_ipc_bytes: 3"), "{rendered}");
    }
}