polyc-query 2026.9.0

//! The one public seam a Query deployment composes.
//!
//! A service process outside this crate needs exactly two things: a way to
//! build the projected read path once at startup, and a way to run one
//! credential-bearing request against it. Everything between those two points
//! stays private.
//!
//! That is deliberate. The scope a query may read, the principal it is
//! attributed to, the permit that makes its audit exactly-once, the artifacts
//! it opens, and the trust that admits a manifest are all decided in here. A
//! shallower seam — one that handed out a verified scope, or a prepared query,
//! or an artifact reader — would let a second composition assemble those parts
//! in an order this crate never checked. There is no public way to supply a
//! scope, a principal, or a permit from outside.
//!
//! The service therefore takes a raw credential and returns encoded protocol
//! frames. It does not take a `QueryScope`, and it cannot:
//! that type is crate-private and has no public constructor.

mod ipc;

use std::fmt;
use std::sync::Arc;
use std::time::Duration;

use connectrpc::client::ClientTransport;
use datafusion::execution::memory_pool::FairSpillPool;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
use datafusion::execution::session_state::SessionStateBuilder;
use futures::{Stream, StreamExt as _};
use polyc_crypto::signing_role::{RoleTrustSet, TurnReadRole};
use polyc_query_model::{
    DataFrame, ErrorClass, QueryOutcome, QueryRequest, ResultFrame, SchemaFrame, TerminalFrame,
    Truncation,
};
use polyc_state::id::{Audience, NamespaceId, OwnerId};
use polyc_state::immutable::AtRestProtection;
use polyc_state::projection::artifact::{ManifestTrust, ObjectNamespace};
use polyc_state::query_audit::{QueryId, RequesterId};
use polyc_state_connect::journal::client::JournalClient;
use polyc_state_connect::projection::client::ProjectionCatalogClient;
use polyc_state_connect::query_audit::client::QueryAuditClient;
use polyc_state_connect::wire::DeclaredCall;
use polyc_storage_gcs::GcsReadClient;

use crate::authority::{PrincipalError, Scoping};
use crate::core_execution::{
    CoreArtifactAuthority, CoreExecutionAdmission, CoreExecutionAdmissionInput, CoreExecutionError,
    CoreResultStream, classify_error, classify_resolution,
};
use crate::core_production::{
    ConnectCoreMetadata, FleetGcsSource, GcsReadNamespace, VisibleGcsSource, fleet_gcs_artifacts,
    visible_gcs_artifacts,
};
use crate::core_resolution::{
    CatalogCompiler, CoreAuditContext, CoreConsistency, CoreMetadataAuthority, CoreParameter,
    CorePlanOutcome, CorePlanningAuthority, CoreQueryRequest, CoreRequestedBounds,
    CoreResolutionError, ProjectedCorePolicy, ProjectedCorePolicyInput,
};
use crate::credential::{CredentialAuthority, CredentialWitness, SystemUnixClock, UnixClock};
use crate::engine::QueryLimits;

pub use crate::credential::SessionVerification;
pub use ipc::FrameEncodeError;

use ipc::FrameEncoder;

/// The credential a caller presented with one request.
///
/// The transport carries one bearer header, so this carries one value. Which
/// credential it holds — an explorer session or a signed conversation grant —
/// is decided inside the shared mechanism, once, and retained for the life of
/// the stream.
///
/// `Debug` reveals nothing. A bearer token is a session another person can
/// replay, and a grant token is a conversation another person can read.
pub struct QueryCredential(String);

impl QueryCredential {
    /// Takes the validated bearer value from the transport.
    #[must_use]
    pub const fn from_bearer(token: String) -> Self {
        Self(token)
    }
}

impl fmt::Debug for QueryCredential {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("QueryCredential([REDACTED])")
    }
}

/// One object namespace this workload may read.
///
/// `Debug` reports the protection alone. A namespace name identifies a
/// deployment's storage layout.
#[derive(Clone)]
pub struct ArtifactNamespace {
    /// The namespace name.
    pub namespace: String,
    /// The at-rest protection every object in it carries.
    pub protection: AtRestProtection,
}

impl fmt::Debug for ArtifactNamespace {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ArtifactNamespace")
            .field("protection", &self.protection)
            .finish_non_exhaustive()
    }
}

/// The realm one Query workload serves.
///
/// A workload serves exactly one. The visible variant receives no Fleet
/// client, so a visible deployment cannot read a Fleet artifact even if its
/// topology names one: it holds the Fleet namespace names for the topology
/// check and no credential to open them.
pub enum QueryServiceRealm {
    /// Visible artifacts only.
    Visible {
        /// The read-only client for visible objects.
        artifacts: GcsReadClient,
        /// The visible namespaces this workload may open.
        namespaces: Vec<ArtifactNamespace>,
        /// Fleet namespace names, for the topology check alone.
        fleet_namespaces: Vec<String>,
    },
    /// Visible and Fleet artifacts, through separately typed clients.
    Fleet {
        /// The read-only client for visible objects.
        visible_artifacts: GcsReadClient,
        /// The visible namespaces this workload may open.
        visible_namespaces: Vec<ArtifactNamespace>,
        /// The read-only client for Fleet objects.
        fleet_artifacts: GcsReadClient,
        /// The Fleet namespaces this workload may open.
        fleet_namespaces: Vec<ArtifactNamespace>,
    },
}

impl fmt::Debug for QueryServiceRealm {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Visible { .. } => "visible",
            Self::Fleet { .. } => "fleet",
        })
    }
}

/// Deployment-owned ceilings and capacities.
///
/// Every value here narrows what a caller may request. None of them widens a
/// caller's own request.
#[derive(Debug, Clone, Copy)]
pub struct QueryServicePolicy {
    /// Largest total result a single query may release.
    pub result_release_bytes: u64,
    /// Largest single protocol frame.
    pub response_frame_bytes: u64,
    /// Largest signed manifest this deployment reads.
    pub manifest_bytes: u64,
    /// Largest artifact file this deployment opens.
    pub artifact_file_bytes: u64,
    /// Largest single ranged artifact read.
    pub artifact_range_bytes: u64,
    /// Largest decoded source footprint for one query.
    pub source_decode_bytes: u64,
    /// Queries that may execute at once.
    pub max_concurrent_executions: usize,
    /// How often a running stream re-proves its credential.
    pub revalidation_interval: Duration,
    /// `DataFusion`'s memory pool size for this workload.
    pub execution_memory_bytes: usize,
}

/// Failure composing or running the projected read path.
///
/// `Debug` and `Display` report the failure and never the query, the caller,
/// or the partitions involved.
#[derive(Debug, thiserror::Error)]
pub enum QueryServiceError {
    /// The deployment's own configuration is not usable.
    #[error("the query service composition is invalid")]
    InvalidComposition,
    /// The presented credential does not authorize this query.
    #[error("the presented credential does not authorize this query")]
    Unauthorized,
    /// The query could not be planned, and this class says who is
    /// responsible.
    ///
    /// The class is the protocol's own vocabulary. The internal refusal that
    /// produced it stays inside this crate and inside the durable record: a
    /// caller learns responsibility, never which partition, manifest, or
    /// object was involved.
    #[error("the query could not be planned")]
    Resolution(ErrorClass),
    /// The query id was already recorded, so this call minted no new
    /// execution capability.
    #[error("this query identity was already recorded")]
    AlreadyRecorded,
    /// The query could not be executed, and this class says who is
    /// responsible. See [`QueryServiceError::Resolution`] on why the class
    /// travels and the refusal does not.
    #[error("the query could not be executed")]
    Execution(ErrorClass),
    /// A released batch could not be encoded into protocol frames.
    #[error("the result could not be encoded")]
    Encode(#[source] FrameEncodeError),
    /// The durable source vector could not be reported to the caller.
    #[error("the result source evidence could not be reported")]
    Evidence,
}

/// What the shared credential mechanism needs.
///
/// The in-process entry point supplies exactly these three through
/// `QueryAuthority::new_state_backed`. This composition root supplies them
/// directly, because it needs no journal and no dashboard: it replays nothing.
/// Both reach the same constructor, so there is still one mechanism.
pub struct CredentialSource {
    /// The persona authority a participation scope is resolved through.
    pub persona: Arc<dyn crate::authority::PersonaSource>,
    /// The turn-read role's public trust, for verifying a signed grant.
    pub turn_read_trust: RoleTrustSet<TurnReadRole>,
    /// The durable authority a bearer session is verified against.
    ///
    /// Verification only. A composition root that supplies this cannot mint a
    /// session through it and cannot revoke one: the port has neither method.
    pub sessions: Arc<dyn SessionVerification>,
}

impl fmt::Debug for CredentialSource {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("CredentialSource")
            .finish_non_exhaustive()
    }
}

/// Everything a [`ProjectedCoreService`] needs at startup.
///
/// One struct, not ten arguments: every field states what it is at the call
/// site, so a deployment cannot silently swap its visible client for its Fleet
/// one, or its namespace for its owner.
pub struct ProjectedCoreComposition<T> {
    /// The projection namespace this deployment reads.
    pub namespace: String,
    /// The projection owner whose generations this deployment reads.
    pub projection_owner: String,
    /// The deployment's own ceilings.
    pub policy: QueryServicePolicy,
    /// The deployment's query limits.
    pub limits: QueryLimits,
    /// The realm this workload serves.
    pub realm: QueryServiceRealm,
    /// The public trust that admits a signed manifest.
    pub trust: Arc<dyn ManifestTrust>,
    /// State's journal client.
    pub journal: JournalClient<T>,
    /// State's projection catalog client.
    pub projections: ProjectionCatalogClient<T>,
    /// State's query audit client.
    pub audit: QueryAuditClient<T>,
}

impl<T> fmt::Debug for ProjectedCoreComposition<T> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProjectedCoreComposition")
            .field("policy", &self.policy)
            .field("realm", &self.realm)
            .finish_non_exhaustive()
    }
}

/// The composed projected read path.
///
/// Built once at startup and shared by every request. It holds no credential
/// and no scope: both arrive with each call to [`Self::start_projected`].
pub struct ProjectedCoreService {
    credential: Arc<CredentialAuthority>,
    clock: Arc<dyn UnixClock>,
    planning: CorePlanningAuthority,
    compiler: CatalogCompiler,
    realm: ComposedRealm,
    trust: Arc<dyn ManifestTrust>,
    admission: Arc<CoreExecutionAdmission>,
    limits: QueryLimits,
    revalidation_interval: Duration,
}

impl fmt::Debug for ProjectedCoreService {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProjectedCoreService")
            .field("realm", &self.realm.kind())
            .field("revalidation_interval", &self.revalidation_interval)
            .finish_non_exhaustive()
    }
}

/// The realm-typed artifact sources, held for the life of the service.
enum ComposedRealm {
    Visible {
        artifacts: GcsReadClient,
        namespaces: Vec<GcsReadNamespace>,
        fleet_namespaces: Vec<ObjectNamespace>,
    },
    Fleet {
        visible_artifacts: GcsReadClient,
        visible_namespaces: Vec<GcsReadNamespace>,
        fleet_artifacts: GcsReadClient,
        fleet_namespaces: Vec<GcsReadNamespace>,
    },
}

impl ComposedRealm {
    const fn kind(&self) -> &'static str {
        match self {
            Self::Visible { .. } => "visible",
            Self::Fleet { .. } => "fleet",
        }
    }

    /// Builds this request's artifact authority around its own witness.
    ///
    /// The witness is per request because the credential is. Composing the
    /// authority here, rather than once at startup, is what makes it
    /// impossible to run a query against a witness that proved some other
    /// caller's credential.
    fn authority(
        &self,
        trust: Arc<dyn ManifestTrust>,
        witness: Arc<CredentialWitness>,
        revalidation_interval: Duration,
        admission: Arc<CoreExecutionAdmission>,
    ) -> Result<CoreArtifactAuthority, CoreExecutionError> {
        match self {
            Self::Visible {
                artifacts,
                namespaces,
                fleet_namespaces,
            } => visible_gcs_artifacts(
                VisibleGcsSource::new(artifacts.clone(), namespaces.clone()),
                fleet_namespaces.clone(),
                trust,
                witness,
                revalidation_interval,
                admission,
            ),
            Self::Fleet {
                visible_artifacts,
                visible_namespaces,
                fleet_artifacts,
                fleet_namespaces,
            } => fleet_gcs_artifacts(
                VisibleGcsSource::new(visible_artifacts.clone(), visible_namespaces.clone()),
                FleetGcsSource::new(fleet_artifacts.clone(), fleet_namespaces.clone()),
                trust,
                witness,
                revalidation_interval,
                admission,
            ),
        }
    }
}

fn read_namespaces(
    namespaces: Vec<ArtifactNamespace>,
) -> Result<Vec<GcsReadNamespace>, QueryServiceError> {
    if namespaces.is_empty() {
        return Err(QueryServiceError::InvalidComposition);
    }
    namespaces
        .into_iter()
        .map(|entry| {
            ObjectNamespace::try_new(entry.namespace)
                .map(|namespace| GcsReadNamespace::new(namespace, entry.protection))
                .map_err(|_invalid| QueryServiceError::InvalidComposition)
        })
        .collect()
}

fn topology_namespaces(namespaces: Vec<String>) -> Result<Vec<ObjectNamespace>, QueryServiceError> {
    namespaces
        .into_iter()
        .map(|namespace| {
            ObjectNamespace::try_new(namespace)
                .map_err(|_invalid| QueryServiceError::InvalidComposition)
        })
        .collect()
}

impl ProjectedCoreService {
    /// Composes the projected read path, or refuses.
    ///
    /// Every input is checked here. Nothing in this constructor panics on a
    /// bad deployment value: a workload with an unusable namespace, an
    /// out-of-range bound, or an empty owner fails to start and says so.
    ///
    /// `credential` supplies the one shared credential mechanism. The
    /// in-process entry point builds the same mechanism from the same three
    /// inputs through the same constructor, so the two surfaces cannot decide
    /// a credential differently. The parity suite in `authority::tests` proves
    /// it case by case.
    ///
    /// # Errors
    ///
    /// Returns [`QueryServiceError::InvalidComposition`] for any unusable
    /// namespace, bound, identifier, or realm wiring.
    pub fn try_new<T>(
        credential: CredentialSource,
        composition: ProjectedCoreComposition<T>,
    ) -> Result<Self, QueryServiceError>
    where
        T: ClientTransport + Send + Sync + 'static,
        <T::ResponseBody as connectrpc::http_body::Body>::Error: fmt::Display,
    {
        let ProjectedCoreComposition {
            namespace,
            projection_owner,
            policy,
            limits,
            realm,
            trust,
            journal,
            projections,
            audit,
        } = composition;
        let namespace = namespace.as_str();
        let projection_owner = projection_owner.as_str();
        let core_policy = ProjectedCorePolicy::try_from(ProjectedCorePolicyInput {
            result_release_bytes: policy.result_release_bytes,
            response_frame_bytes: policy.response_frame_bytes,
            manifest_bytes: policy.manifest_bytes,
            artifact_file_bytes: policy.artifact_file_bytes,
            artifact_range_bytes: policy.artifact_range_bytes,
            source_decode_bytes: policy.source_decode_bytes,
        })
        .map_err(|_invalid| QueryServiceError::InvalidComposition)?;

        let metadata: Arc<dyn CoreMetadataAuthority> =
            Arc::new(ConnectCoreMetadata::new(journal, projections, audit));
        let planning = CorePlanningAuthority::new(
            NamespaceId::new(namespace),
            OwnerId::new(projection_owner),
            core_policy,
            metadata,
        )
        .map_err(|_invalid| QueryServiceError::InvalidComposition)?;

        let realm = match realm {
            QueryServiceRealm::Visible {
                artifacts,
                namespaces,
                fleet_namespaces,
            } => ComposedRealm::Visible {
                artifacts,
                namespaces: read_namespaces(namespaces)?,
                fleet_namespaces: topology_namespaces(fleet_namespaces)?,
            },
            QueryServiceRealm::Fleet {
                visible_artifacts,
                visible_namespaces,
                fleet_artifacts,
                fleet_namespaces,
            } => ComposedRealm::Fleet {
                visible_artifacts,
                visible_namespaces: read_namespaces(visible_namespaces)?,
                fleet_artifacts,
                fleet_namespaces: read_namespaces(fleet_namespaces)?,
            },
        };

        let admission = Arc::new(
            CoreExecutionAdmission::try_from(CoreExecutionAdmissionInput {
                max_concurrent_executions: policy.max_concurrent_executions,
            })
            .map_err(|_invalid| QueryServiceError::InvalidComposition)?,
        );

        if policy.revalidation_interval.is_zero() || policy.execution_memory_bytes == 0 {
            return Err(QueryServiceError::InvalidComposition);
        }
        let runtime = RuntimeEnvBuilder::new()
            .with_memory_pool(Arc::new(FairSpillPool::new(policy.execution_memory_bytes)))
            .build_arc()
            .map_err(|_datafusion| QueryServiceError::InvalidComposition)?;
        let session = SessionStateBuilder::new()
            .with_runtime_env(runtime)
            .with_default_features()
            .build();

        Ok(Self {
            credential: Arc::new(CredentialAuthority::verifying(
                credential.persona,
                credential.turn_read_trust,
                credential.sessions,
            )),
            clock: Arc::new(SystemUnixClock),
            planning,
            compiler: CatalogCompiler::new(session),
            realm,
            trust,
            admission,
            limits,
            revalidation_interval: policy.revalidation_interval,
        })
    }

    /// Runs one credential-bearing projected query.
    ///
    /// The credential is verified before anything is planned, retained in a
    /// private witness, and re-proved before the first row and at the
    /// revalidation interval for as long as the stream releases rows.
    ///
    /// # Errors
    ///
    /// Returns [`QueryServiceError::Unauthorized`] when the credential does
    /// not authorize the query, [`QueryServiceError::Resolution`] when it
    /// cannot be planned, [`QueryServiceError::AlreadyRecorded`] when the
    /// query identity was already durably recorded, and
    /// [`QueryServiceError::Execution`] when execution itself refuses.
    pub async fn start_projected(
        &self,
        credential: QueryCredential,
        query_id: &str,
        request: &QueryRequest,
    ) -> Result<ProjectedResultStream, QueryServiceError> {
        let (witness, scoping) = CredentialWitness::admit_bearer(
            credential.0,
            Arc::clone(&self.credential),
            Arc::clone(&self.clock),
        )
        .await
        .map_err(refusal)?;
        let requester = requester_of(&scoping)?;

        let caller_bounds = requested_bounds(request);
        let bounds = self
            .planning
            .effective_bounds(&self.limits, caller_bounds)
            .map_err(|error| resolution_refusal(&error))?;
        let declared = DeclaredCall::live(Audience::new("state"), bounds.timeout());
        let audit =
            CoreAuditContext::from_scoped(QueryId::new(query_id), requester, &declared, bounds);
        let core_request = CoreQueryRequest::new(
            request.sql().to_owned(),
            request.parameters().iter().map(parameter_of).collect(),
            consistency_of(request),
            caller_bounds,
        );

        let prepared = match self
            .planning
            .plan(
                &self.compiler,
                &self.limits,
                &scoping.scope,
                scoping.allow_explain,
                audit,
                core_request,
            )
            .await
            .map_err(|error| resolution_refusal(&error))?
        {
            CorePlanOutcome::Granted(prepared) => *prepared,
            CorePlanOutcome::AlreadyRecorded(_) => {
                return Err(QueryServiceError::AlreadyRecorded);
            }
        };

        let artifacts = self
            .realm
            .authority(
                Arc::clone(&self.trust),
                Arc::new(witness),
                self.revalidation_interval,
                Arc::clone(&self.admission),
            )
            .map_err(|error| execution_refusal(&error))?;
        // The artifact authority exists only to bind this one query's exact
        // files. Dropped here so its realm-typed readers do not outlive the
        // bind they were composed for.
        let bound = artifacts.bind(prepared).await;
        drop(artifacts);
        let rows = bound.map_err(|error| execution_refusal(&error))?.execute();
        ProjectedResultStream::start(
            rows,
            bounds.response_frame_bytes(),
            bounds.result_release_bytes(),
        )
    }
}

fn resolution_refusal(error: &CoreResolutionError) -> QueryServiceError {
    // The caller learns only the class. Which resolution step refused is this
    // deployment's to know, and the class alone cannot distinguish an absent
    // projection from a superseded source.
    tracing::warn!(?error, "the projected read path could not resolve a query");
    QueryServiceError::Resolution(crate::core_evidence::error_class_of(classify_resolution(
        error,
    )))
}

fn execution_refusal(error: &CoreExecutionError) -> QueryServiceError {
    // As with resolution: the caller learns the class, and which step refused
    // stays here rather than being discarded.
    tracing::warn!(%error, source = ?std::error::Error::source(error), "the projected read path could not execute a query");
    QueryServiceError::Execution(crate::core_evidence::error_class_of(classify_error(error)))
}

const fn refusal(_error: PrincipalError) -> QueryServiceError {
    // Every refusal reaching the wire is the same fact: this credential does
    // not authorize this query. The reason stays in the durable record.
    QueryServiceError::Unauthorized
}

/// Derives the audit's requester from the verified scope.
///
/// Fails closed. A scope that names neither a caller nor a conversation could
/// not be attributed, and an unattributed query is not one this plane runs.
fn requester_of(scoping: &Scoping) -> Result<RequesterId, QueryServiceError> {
    scoping.caller_identity.as_ref().map_or_else(
        || {
            scoping
                .conversation_id
                .as_ref()
                .map_or(Err(QueryServiceError::Unauthorized), |conversation| {
                    Ok(RequesterId::new(format!("conversation:{conversation}")))
                })
        },
        |persona| Ok(RequesterId::new(format!("persona:{persona}"))),
    )
}

const fn consistency_of(request: &QueryRequest) -> CoreConsistency {
    match request.consistency() {
        polyc_query_model::Consistency::Projected => CoreConsistency::Projected,
        polyc_query_model::Consistency::RequireProjectedThrough(position) => {
            CoreConsistency::RequireProjectedThrough(polyc_state::revision::JournalPosition::new(
                position,
            ))
        }
    }
}

fn parameter_of(parameter: &polyc_query_model::Parameter) -> CoreParameter {
    match parameter {
        polyc_query_model::Parameter::Utf8(value) => CoreParameter::Utf8(value.clone()),
        polyc_query_model::Parameter::UInt64(value) => CoreParameter::UInt64(*value),
        polyc_query_model::Parameter::Boolean(value) => CoreParameter::Boolean(*value),
        polyc_query_model::Parameter::Null => CoreParameter::Null,
    }
}

const fn requested_bounds(request: &QueryRequest) -> CoreRequestedBounds {
    let bounds = request.bounds();
    CoreRequestedBounds::from_requested(
        bounds.timeout(),
        bounds.rows(),
        bounds.result_bytes(),
        bounds.frame_bytes(),
    )
}

/// The protocol frames one query releases.
///
/// The schema frame is produced before the first batch exists, so a caller
/// learns the result shape without waiting for a row.
///
/// A terminal frame that comes from the underlying stream reaching its own end
/// is produced after the durable completion is settled. Two do not: a frame
/// this type could not build, and a stop at the caller's release ceiling. Both
/// are measured here and reported to the latch here, and the guardian settles
/// them when this stream is dropped — so for those two the caller reads its
/// terminal before the record is durable, and a crash in between leaves an
/// unmatched intent for State's reconciler.
///
/// # Release accounting
///
/// This type is the one place that knows which rows reached the caller. It
/// re-frames each batch, so a batch leaving the execution stream is not the
/// same event as its rows being received, and it may stop with frames of that
/// batch unbuilt. It therefore owns the count: the rows and encoded bytes it
/// released, whether it stopped at a ceiling, the terminal frame those numbers
/// build, and the durable completion recorded when the stop is its own. The
/// execution stream defers to it — see `CoreResultStream::account_at_consumer`
/// — so the two never keep separate tallies that can disagree.
///
/// Owning the count is not enough on its own, because the terminal is chosen
/// by whichever party ends the query first. So a release is not recorded, it
/// is admitted: `CoreResultStream::admit_release` takes the latch, admits the
/// release only while no terminal has been chosen, and counts it in the same
/// critical section. One boundary therefore decides whether a frame belongs to
/// the result. If it does, every terminal built afterwards reports it; if it
/// does not, the frame is discarded here and never reaches the caller. No
/// interleaving leaves the caller holding a row the record omits, or the
/// record carrying a row the caller never got, and the answer does not depend
/// on which task the scheduler ran first.
///
/// # Buffering
///
/// What this type holds does not grow with the result. The schema frame is
/// held until it is released; after that, one data frame is built and released
/// in the same step — except a frame the boundary refuses, which is built,
/// measured, and dropped unreleased: one that crosses the release ceiling, or
/// one the latch declines because the query has already ended.
///
/// `core_execution` charges decode against the shared memory pool and says
/// plainly that it does not account for memory outside that envelope. This
/// buffer is outside it, so its bound is stated here.
///
/// Alive at once: one encoded frame, which the encoder holds under
/// `frame_bytes`, and one search candidate, which the encoder bounds by ROWS
/// rather than bytes — under twice the rows of the frame that fits. Rows are
/// not uniform in width, so that is not a byte bound. The byte bound comes
/// from the other side: a candidate is a prefix of one released batch, and
/// execution holds each of those under `result_release_bytes`
/// (`core_execution/stream.rs`). That ceiling is measured in decoded memory
/// and the candidate in encoded bytes, so it bounds this buffer by one batch
/// rather than exactly — which is the property the cursor exists for. What is
/// exact is that the buffer does not grow with the number of frames the
/// caller's `frame_bytes` divides that batch into.
pub struct ProjectedResultStream {
    rows: CoreResultStream,
    encoder: FrameEncoder,
    /// The schema frame, until it is released. Nothing else is ever held: a
    /// data frame is built on demand and released in the same step, so what
    /// this type buffers does not grow with the result.
    queued: Option<ResultFrame>,
    stage: Stage,
    /// Rows released across data frames, counted as each one leaves.
    released_rows: u64,
    /// Opaque bytes released across data frames, counted as each one leaves.
    released_bytes: u64,
    /// The caller's ceiling on those opaque bytes.
    ///
    /// The engine bounds release by the decoded size of a batch in memory.
    /// The protocol bounds the same caller number by the encoded bytes summed
    /// across data frames, and every data frame repeats the schema message, so
    /// the encoded total grows with the frame count — which the caller chooses
    /// through `frame_bytes`. The two numbers are not interchangeable, and
    /// only this type sees both.
    release_ceiling: u64,
    /// Whether release stopped at `release_ceiling` with rows still to come.
    byte_truncated: bool,
}

impl fmt::Debug for ProjectedResultStream {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProjectedResultStream")
            .field("stage", &self.stage)
            .field("queued", &self.queued.is_some())
            .finish_non_exhaustive()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Stage {
    /// Frames are still being produced.
    Streaming,
    /// The terminal frame was released. Nothing follows it.
    Terminated,
}

impl ProjectedResultStream {
    pub(crate) fn start(
        rows: CoreResultStream,
        frame_ceiling: u64,
        release_ceiling: u64,
    ) -> Result<Self, QueryServiceError> {
        let Ok(ceiling) = usize::try_from(frame_ceiling) else {
            // Reported before `rows` drops. Dropping it withdraws through the
            // latch, and a withdrawal is a claim about the caller that no
            // refusal on this path can honestly make.
            let _ = rows.report_failure(polyc_state::query_audit::ErrorClass::Internal);
            return Err(QueryServiceError::InvalidComposition);
        };
        let (encoder, schema) = match FrameEncoder::start(rows.schema().clone(), ceiling) {
            Ok(started) => started,
            Err(error) => {
                // The one refusal a caller can provoke here: a frame ceiling
                // too small for this result's schema. Logged with its own
                // numbers, which the wrapping error's message does not carry.
                let class = error.class();
                tracing::warn!(%error, ?class, "the result schema does not fit the frame ceiling");
                let _ = rows.report_failure(crate::core_evidence::durable_class_of(class));
                return Err(QueryServiceError::Encode(error));
            }
        };
        // This type re-frames every batch, so it — not the stream that hands
        // it one — is what knows which rows reached the caller.
        rows.account_at_consumer();
        Ok(Self {
            rows,
            encoder,
            queued: Some(ResultFrame::Schema(schema)),
            stage: Stage::Streaming,
            released_rows: 0,
            released_bytes: 0,
            release_ceiling,
            byte_truncated: false,
        })
    }

    /// Releases the terminal frame and closes the stream.
    ///
    /// The stage moves to [`Stage::Terminated`] whether or not the frame could
    /// be built, because both outcomes end the stream: nothing may follow a
    /// terminal, and a terminal that cannot be built is still terminal.
    ///
    /// Closing here rather than at each call site is what makes a second
    /// terminal unreachable. The success path returned one without closing, so
    /// every later poll found an exhausted row stream and reported the same
    /// terminal again.
    fn release_terminal(
        &mut self,
        outcome: QueryOutcome,
    ) -> Result<ResultFrame, QueryServiceError> {
        self.stage = Stage::Terminated;
        self.terminal(outcome)
    }

    /// Builds the terminal frame from what the consumer actually received.
    ///
    /// The counts come from the data frames this stream released, not from
    /// execution's release budget. The two measure different things: the
    /// budget counts decoded row bytes against a ceiling, while the protocol
    /// defines the terminal's total as the opaque bytes released across data
    /// frames. Both peers check that equality, so reporting the budget's
    /// number broke every stream that carried a row.
    fn terminal(&self, outcome: QueryOutcome) -> Result<ResultFrame, QueryServiceError> {
        let evidence = crate::core_evidence::evidence_of(self.rows.source())
            .map_err(|_unrepresentable| QueryServiceError::Evidence)?;
        let truncation = if self.byte_truncated || self.rows.delivered().truncated() {
            Truncation::TruncatedAt(self.released_rows)
        } else {
            Truncation::Complete
        };
        Ok(ResultFrame::Terminal(TerminalFrame::new(
            outcome,
            self.rows.elapsed(),
            self.released_rows,
            self.released_bytes,
            truncation,
            evidence,
        )))
    }

    /// Settles this stream at the caller's own release ceiling.
    ///
    /// The caller asked for at most this many opaque bytes and received
    /// exactly `released_rows` inside them. That is a bounded success, and it
    /// is recorded as one on both sides: the terminal frame the caller reads
    /// and the durable completion carry the same outcome, the same row count,
    /// and the same truncation.
    ///
    /// It must not be recorded as a withdrawal. Dropping this stream is how
    /// the producer is stopped once the bound is reached, and that drop
    /// withdraws through the latch — so the bound is settled here, first,
    /// where the released numbers are known.
    fn settle_at_release_ceiling(&mut self) -> Option<Result<ResultFrame, QueryServiceError>> {
        let settled = self.rows.settle_consumer_bound()?;
        self.byte_truncated = true;
        // The record's counts and this stream's are the same numbers: every
        // one of them was admitted through the latch that just built the
        // record, and no release can be admitted after it.
        debug_assert_eq!(settled.rows(), self.released_rows);
        Some(self.release_terminal(QueryOutcome::Succeeded))
    }

    /// Builds the next data frame and releases it, or reports why it cannot.
    ///
    /// One frame is built per call. A frame that would carry the release
    /// total past the caller's ceiling is not built for nothing and then
    /// dropped: it is the last one built, and the stream settles instead of
    /// framing the rest of the batch.
    fn release_next_data_frame(&mut self) -> Option<Result<ResultFrame, QueryServiceError>> {
        let data = match self.encoder.next() {
            Ok(Some(data)) => data,
            Ok(None) => return None,
            Err(error) => {
                // The class says who is responsible: a ceiling too small for
                // one row is the caller's bound, an Arrow failure is this
                // deployment's. Both end the query, and both are measured, so
                // both are reported rather than left to the withdrawal this
                // stream's drop records.
                let class = error.class();
                let selected = self
                    .rows
                    .report_failure(crate::core_evidence::durable_class_of(class));
                tracing::warn!(%error, ?class, "a released batch could not be framed");
                if selected {
                    return Some(self.release_terminal(QueryOutcome::Failed(class)));
                }
                // Another terminal already owns the record. This locally
                // measured failure cannot describe the caller's result, so
                // discard the batch and take the selected terminal from the
                // execution stream.
                self.encoder.discard();
                return None;
            }
        };
        let bytes = data.arrow_ipc().len() as u64;
        let rows = data.rows();
        let total = self.released_bytes.saturating_add(bytes);
        if total > self.release_ceiling {
            // The caller's byte ceiling, measured in the unit the protocol
            // defines. Releasing this frame would produce a stream this
            // server's own contract refuses, which reads as an internal fault
            // for a bound the caller chose. Stopping one frame short is the
            // honest answer: the result is complete up to here and there is
            // more.
            //
            // The first data frame cannot reach this. Bounds resolution
            // refuses a request whose `response_frame_bytes` exceeds its
            // `result_release_bytes` — `EffectiveCoreBounds::mint`, which is
            // their only constructor — and the encoder holds every frame
            // under the former.
            // Another party may have ended the query first, in which case its
            // terminal is the record and this frame is not part of it.
            let settled = self.settle_at_release_ceiling();
            if settled.is_none() {
                self.encoder.discard();
            }
            return settled;
        }
        // The one boundary. The latch either takes this release into the
        // record it will settle, or the query has already ended and the frame
        // must not reach the caller: the record is fixed without it.
        if !self.rows.admit_release(rows, bytes) {
            self.encoder.discard();
            return None;
        }
        self.released_rows = self.released_rows.saturating_add(rows);
        self.released_bytes = total;
        Some(Ok(ResultFrame::Data(data)))
    }

    /// Releases the next frame, or `None` after the terminal frame.
    ///
    /// # Errors
    ///
    /// Returns the refusal that ended the query. A refusal is terminal: the
    /// stream releases nothing after it.
    pub async fn next_frame(&mut self) -> Option<Result<ResultFrame, QueryServiceError>> {
        loop {
            // The schema frame, and only ever that: it is built before any
            // row exists, so it cannot be produced on demand beside the data.
            if let Some(frame) = self.queued.take() {
                return Some(Ok(frame));
            }
            if self.stage == Stage::Terminated {
                return None;
            }
            if self.encoder.has_rows() {
                // A refusal means the query ended under this stream — a
                // deadline fired, or the producer measured a failure, while
                // this type still held a batch. The frame is discarded and
                // the already-chosen terminal is what the stream yields next.
                if let Some(frame) = self.release_next_data_frame() {
                    return Some(frame);
                }
                continue;
            }
            match self.rows.next().await {
                Some(Ok(batch)) => self.encoder.begin(batch),
                Some(Err(error)) => {
                    let outcome = QueryOutcome::Failed(crate::core_evidence::error_class_of(
                        classify_error(&error),
                    ));
                    return Some(self.release_terminal(outcome));
                }
                None => {
                    return Some(self.release_terminal(QueryOutcome::Succeeded));
                }
            }
        }
    }

    /// Returns how many frames the encoder has built.
    ///
    /// A case reads this to prove that framing stops where release stops.
    #[cfg(test)]
    pub(crate) const fn frames_built(&self) -> u64 {
        self.encoder.built()
    }

    /// Requests one batch and leaves it buffered before this framer.
    #[cfg(test)]
    pub(crate) fn request_buffered_batch(&mut self) {
        self.rows.request_buffered_batch();
    }

    /// Returns how many producer frames wait before this framer.
    #[cfg(test)]
    pub(crate) fn buffered_frames(&self) -> usize {
        self.rows.buffered_frames()
    }

    /// Returns whether a producer or consumer terminal has won the latch.
    #[cfg(test)]
    pub(crate) fn terminal_selected(&self) -> bool {
        self.rows.terminal_selected()
    }
}

/// A bounded, fully materialized projected result.
///
/// The embedded caller does not stream. It asks for a whole result and gets
/// one, under the same ceilings a streaming caller runs under: the query's own
/// row cap and release-byte bound are enforced by the same execution path, so
/// this cannot return more than a streaming caller would have received.
#[derive(Debug)]
pub struct MaterializedResult {
    frames: Vec<DataFrame>,
    schema: SchemaFrame,
    terminal: TerminalFrame,
}

impl MaterializedResult {
    /// Returns the schema frame.
    #[must_use]
    pub const fn schema(&self) -> &SchemaFrame {
        &self.schema
    }

    /// Returns the ordered data frames.
    #[must_use]
    pub fn frames(&self) -> &[DataFrame] {
        &self.frames
    }

    /// Returns the terminal frame.
    #[must_use]
    pub const fn terminal(&self) -> &TerminalFrame {
        &self.terminal
    }
}

impl ProjectedResultStream {
    /// Drains this stream into one bounded result.
    ///
    /// # Errors
    ///
    /// Returns the refusal that ended the query, or
    /// [`QueryServiceError::Evidence`] if a stream ended without a terminal
    /// frame, which no correct producer does.
    pub async fn materialize(mut self) -> Result<MaterializedResult, QueryServiceError> {
        let mut schema = None;
        let mut frames = Vec::new();
        let mut terminal = None;
        while let Some(frame) = self.next_frame().await {
            match frame? {
                ResultFrame::Schema(value) => schema = Some(value),
                ResultFrame::Data(value) => frames.push(value),
                ResultFrame::Terminal(value) => terminal = Some(value),
            }
        }
        match (schema, terminal) {
            (Some(schema), Some(terminal)) => Ok(MaterializedResult {
                frames,
                schema,
                terminal,
            }),
            _ => Err(QueryServiceError::Evidence),
        }
    }
}

/// Adapts the frame stream to [`futures::Stream`] for a transport that wants
/// one.
pub fn frames(
    stream: ProjectedResultStream,
) -> impl Stream<Item = Result<ResultFrame, QueryServiceError>> {
    futures::stream::unfold(Some(stream), |state| async move {
        let mut stream = state?;
        let frame = stream.next_frame().await?;
        Some((frame, Some(stream)))
    })
}