polyc-query 2026.10.2

//! 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.

pub(crate) mod ipc;

use std::collections::{BTreeSet, VecDeque};
use std::fmt;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use connectrpc::client::ClientTransport;
use datafusion::common::config::Dialect;
use datafusion::error::DataFusionError;
use datafusion::execution::SessionStateDefaults;
use datafusion::execution::config::SessionConfig;
use datafusion::execution::context::SessionState;
use datafusion::execution::memory_pool::FairSpillPool;
use datafusion::execution::runtime_env::{RuntimeEnv, RuntimeEnvBuilder};
use datafusion::execution::session_state::SessionStateBuilder;
use futures::{Stream, StreamExt as _};
use polyc_crypto::signing_role::{RoleTrustSet, TurnReadRole};
use polyc_query_model::{
    CatalogColumn, CatalogColumnType, CatalogReply, CatalogTable, 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 sha2::Digest as _;
use tokio_util::sync::CancellationToken;

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,
    CoreOperationContext, CoreParameter, CorePlanOutcome, CorePlanningAuthority, CoreQueryRequest,
    CoreRealm, CoreRequestedBounds, CoreResolutionError, ProjectedCorePolicy,
    ProjectedCorePolicyInput, catalog_tables,
};
use crate::limits::{QueryLimits, SpillDirGuard};
use polyc_query_credential::credential::{
    CredentialAuthority, CredentialWitness, SessionVerification, SystemUnixClock, UnixClock,
};
use polyc_query_credential::principal::{PersonaSource, PrincipalError, PrincipalKind, Scoping};
use polyc_query_credential::session::QueryScope;

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,
}

/// One object-store location and the namespaces it serves.
///
/// A composite fixed statement may pin families published under different
/// prefixes. The namespace selects the one closed location that may open its
/// exact references.
pub struct ArtifactSource {
    /// The read-only client for this location.
    pub artifacts: GcsReadClient,
    /// The namespaces routed to this client.
    pub namespaces: Vec<ArtifactNamespace>,
}

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

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>,
        /// Additional visible family locations.
        additional_artifacts: Vec<ArtifactSource>,
        /// 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>,
        /// Additional visible family locations.
        additional_visible_artifacts: Vec<ArtifactSource>,
        /// The read-only client for Fleet objects.
        fleet_artifacts: GcsReadClient,
        /// The Fleet namespaces this workload may open.
        fleet_namespaces: Vec<ArtifactNamespace>,
        /// Additional Fleet family locations.
        additional_fleet_artifacts: Vec<ArtifactSource>,
    },
}

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 once supplied these through
/// `QueryAuthority::new_state_backed`, which is deleted with the embedded
/// engine (POLY-196). This composition root supplies them
/// directly, because it needs no journal and no dashboard: it replays nothing.
pub struct CredentialSource {
    /// The persona authority a participation scope is resolved through.
    pub persona: Arc<dyn 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,
    /// State namespace holding persona-memory journal partitions.
    pub persona_memory_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>,
    /// State's Versioned source-coordinate client.
    ///
    /// A family whose declared kind is `SourceKind::Versioned` is proven live
    /// against this authority, not against a journal partition head.
    pub versioned: polyc_state_connect::versioned_source::VersionedSourceClient<T>,
    /// State's authoritative persona-memory journal client.
    ///
    /// A family whose declared kind is `SourceKind::PersonaMemory` is proven
    /// live against this authority, not against a journal partition head or
    /// a Versioned aggregate.
    pub persona_memory: polyc_state_connect::persona_memory_journal::PersonaMemoryJournalClient<T>,
    /// State's observation authority client (7-O).
    ///
    /// A family whose declared kind is `SourceKind::Observed` is proven live
    /// against this authority, not against a journal partition head, a
    /// Versioned aggregate, or a persona-memory partition.
    pub observation: polyc_state_connect::observation::ObservationClient<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,
    composite_traces: CompositeTraceCache,
    /// Removes `limits.spill_dir` when this service drops — see
    /// [`SpillDirGuard`]'s own doc for why cleanup lives here rather than in
    /// `DataFusion`'s `DiskManager` (POLY-314).
    #[allow(dead_code, reason = "held only for its Drop side effect")]
    spill_guard: SpillDirGuard,
}

const MAX_COMPOSITE_TRACE_GRANTS: usize = 256;

#[derive(Clone, Default)]
struct CompositeTraceCache {
    entries: Arc<Mutex<VecDeque<CompositeTraceGrant>>>,
}

#[derive(Clone)]
struct CompositeTraceGrant {
    key: [u8; 32],
    conversation_id: String,
    personas: Vec<String>,
}

impl CompositeTraceCache {
    fn remember(&self, grant: CompositeTraceGrant) {
        let Ok(mut entries) = self.entries.lock() else {
            return;
        };
        entries.retain(|entry| entry.key != grant.key);
        entries.push_back(grant);
        while entries.len() > MAX_COMPOSITE_TRACE_GRANTS {
            entries.pop_front();
        }
    }

    fn personas(&self, key: &[u8; 32], conversation_id: &str) -> Option<Vec<String>> {
        let entries = self.entries.lock().ok()?;
        entries
            .iter()
            .find(|entry| &entry.key == key && entry.conversation_id == conversation_id)
            .map(|entry| entry.personas.clone())
    }
}

struct CompositeTraceCollector {
    cache: CompositeTraceCache,
    key: [u8; 32],
    conversation_id: String,
    caller_column: usize,
    record_column: usize,
    personas: BTreeSet<String>,
}

impl CompositeTraceCollector {
    fn try_new(
        cache: CompositeTraceCache,
        key: [u8; 32],
        conversation_id: String,
        schema: &arrow::datatypes::Schema,
    ) -> Result<Self, QueryServiceError> {
        let caller_column = schema
            .index_of("caller_persona_id")
            .map_err(|_missing| QueryServiceError::Unauthorized)?;
        let record_column = schema
            .index_of("record_persona_id")
            .map_err(|_missing| QueryServiceError::Unauthorized)?;
        Ok(Self {
            cache,
            key,
            conversation_id,
            caller_column,
            record_column,
            personas: BTreeSet::new(),
        })
    }

    fn observe(&mut self, batch: &arrow::record_batch::RecordBatch) -> Result<(), ErrorClass> {
        for index in [self.caller_column, self.record_column] {
            let values = batch
                .column(index)
                .as_any()
                .downcast_ref::<arrow::array::StringArray>()
                .ok_or(ErrorClass::Internal)?;
            for value in values.iter().flatten().filter(|value| !value.is_empty()) {
                self.personas.insert(value.to_owned());
                if self.personas.len() >= polyc_query_model::MAX_SOURCE_PINS {
                    return Err(ErrorClass::Bounds);
                }
            }
        }
        Ok(())
    }

    fn finish(self) {
        self.cache.remember(CompositeTraceGrant {
            key: self.key,
            conversation_id: self.conversation_id,
            personas: self.personas.into_iter().collect(),
        });
    }
}

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()
    }
}

/// Which Query realm a request is being admitted into — the typed
/// replacement for comparing a string label, so every admission check below
/// matches it exhaustively instead of by equality against a literal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RealmKind {
    /// The visible-only realm.
    Visible,
    /// The Fleet realm.
    Fleet,
}

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

/// The realm-typed artifact sources, held for the life of the service.
enum ComposedRealm {
    Visible {
        sources: Vec<VisibleGcsSource>,
        fleet_namespaces: Vec<ObjectNamespace>,
    },
    Fleet {
        visible_sources: Vec<VisibleGcsSource>,
        fleet_sources: Vec<FleetGcsSource>,
    },
}

impl ComposedRealm {
    const fn kind(&self) -> RealmKind {
        match self {
            Self::Visible { .. } => RealmKind::Visible,
            Self::Fleet { .. } => RealmKind::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 {
                sources,
                fleet_namespaces,
            } => visible_gcs_artifacts(
                sources.clone(),
                fleet_namespaces.clone(),
                trust,
                witness,
                revalidation_interval,
                admission,
            ),
            Self::Fleet {
                visible_sources,
                fleet_sources,
            } => fleet_gcs_artifacts(
                visible_sources.clone(),
                fleet_sources.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()
}

/// Builds the one `SessionState` every projected query plans and executes
/// under (POLY-374).
///
/// One function owns every `DataFusion` option this service's safety or
/// behaviour relies on, so a future `DataFusion` default that changes
/// fails a named test rather than silently changing the surface. Every
/// pin below names the value it replaces:
///
/// - `datafusion.execution.enable_recursive_ctes = false` (today `true`)
///   — the second line behind `crate::statement_gate`'s AST refusal of
///   `WITH RECURSIVE`.
/// - `datafusion.execution.target_partitions = 1` and
///   `datafusion.execution.planning_concurrency = 1` (today both derive
///   from the NODE's core count, not the pod's 500m CPU request — a
///   derived default is replaced, not carried). One partition keeps the
///   memory pool's consumer count and the plan's parallelism
///   deterministic under a throttle.
/// - `datafusion.execution.batch_size = 8192` (today's value).
/// - `datafusion.execution.collect_statistics = true` (today's value).
/// - `datafusion.catalog.information_schema = false` (today's value);
///   `crate::statement_gate`'s ordering guard stays as defence in depth.
/// - `datafusion.catalog.default_catalog = "datafusion"`,
///   `datafusion.catalog.default_schema = "public"`, and
///   `datafusion.catalog.create_default_catalog_and_schema = true`
///   (today's values): `core_resolution` and `core_execution` read the
///   two names from the composed state to register their per-request
///   catalog.
/// - `datafusion.sql_parser.dialect = Generic`,
///   `datafusion.sql_parser.enable_ident_normalization = true`, and
///   `datafusion.sql_parser.parse_float_as_decimal = false` (today's
///   values). Under `Generic`, `->` and `->>` reach the JSON expression
///   planner; `?` tokenizes as a parameter placeholder, so the contains
///   operator stays function-form only — see
///   `docs/architecture/query-json-port.md`.
/// - `datafusion.execution.parquet.pushdown_filters = false`,
///   `reorder_filters = false`, `enable_page_index = true`, and
///   `bloom_filter_on_read = true` (all today's values). The exact
///   provider declines filter pushdown either way; the pins make a
///   flipped upstream default a test failure, not a silent change.
/// - URL tables are never enabled: `SessionContext::enable_url_table` is
///   a builder call, not an option, and nothing on this surface calls
///   it; a test proves a `'file.csv'` table reference still fails to
///   plan.
///
/// `polyc_query_json::register_all` then registers the thirteen JSON
/// scalar functions, the JSON function rewrite, and the JSON expression
/// planner on the built state. `SessionStateBuilder::new_from_existing`
/// propagates all three — plus every pin above — into each per-request
/// session.
///
/// POLY-371: the function registries are closed, not default. The builder
/// starts empty (`with_default_features` is not called) and installs only
/// [`crate::function_surface`]'s reviewed lists plus the core expression
/// planners — no table functions, no higher-order planners, no file-format
/// factories the custom table providers never consult. A statement that
/// names anything else fails planning with the collapsed invalid-query
/// error, and the statement gate refuses it earlier with a countable
/// reason.
///
/// # Errors
///
/// Returns the `DataFusion` error if JSON function, rewrite, or planner
/// registration fails.
pub(crate) fn projected_session_state(
    runtime: Arc<RuntimeEnv>,
) -> Result<SessionState, DataFusionError> {
    let mut config = SessionConfig::new()
        .with_information_schema(false)
        .with_target_partitions(1)
        .with_batch_size(8192);
    {
        let options = config.options_mut();
        options.execution.enable_recursive_ctes = false;
        options.execution.planning_concurrency = 1;
        options.execution.collect_statistics = true;
        options.execution.parquet.pushdown_filters = false;
        options.execution.parquet.reorder_filters = false;
        options.execution.parquet.enable_page_index = true;
        options.execution.parquet.bloom_filter_on_read = true;
        options.catalog.information_schema = false;
        options.catalog.default_catalog = "datafusion".to_string();
        options.catalog.default_schema = "public".to_string();
        options.catalog.create_default_catalog_and_schema = true;
        options.sql_parser.dialect = Dialect::Generic;
        options.sql_parser.enable_ident_normalization = true;
        options.sql_parser.parse_float_as_decimal = false;
    }
    let mut state = SessionStateBuilder::new()
        .with_runtime_env(runtime)
        .with_config(config)
        .with_expr_planners(SessionStateDefaults::default_expr_planners())
        .with_scalar_functions(crate::function_surface::closed_scalar_functions())
        .with_aggregate_functions(crate::function_surface::closed_aggregate_functions())
        .with_window_functions(crate::function_surface::closed_window_functions())
        .build();
    polyc_query_json::register_all(&mut state)?;
    Ok(state)
}

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 —
    /// `polyc_query_credential`'s `CredentialAuthority`, built here from the
    /// same three inputs the service passes. `tests/credential_admission.rs`
    /// covers the admission behaviors this surface relies on.
    ///
    /// # Errors
    ///
    /// Returns [`QueryServiceError::InvalidComposition`] for any unusable
    /// namespace, bound, identifier, or realm wiring.
    #[allow(
        clippy::too_many_lines,
        reason = "the constructor validates and transfers every realm-typed composition field in one audit boundary"
    )]
    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,
            persona_memory_namespace,
            projection_owner,
            policy,
            limits,
            realm,
            trust,
            journal,
            projections,
            audit,
            versioned,
            persona_memory,
            observation,
        } = 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,
            versioned,
            persona_memory,
            observation,
            persona_memory_namespace,
        ));
        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,
                additional_artifacts,
                fleet_namespaces,
            } => {
                let mut sources = vec![VisibleGcsSource::new(
                    artifacts,
                    read_namespaces(namespaces)?,
                )];
                for source in additional_artifacts {
                    sources.push(VisibleGcsSource::new(
                        source.artifacts,
                        read_namespaces(source.namespaces)?,
                    ));
                }
                ComposedRealm::Visible {
                    sources,
                    fleet_namespaces: topology_namespaces(fleet_namespaces)?,
                }
            }
            QueryServiceRealm::Fleet {
                visible_artifacts,
                visible_namespaces,
                additional_visible_artifacts,
                fleet_artifacts,
                fleet_namespaces,
                additional_fleet_artifacts,
            } => {
                let mut visible_sources = vec![VisibleGcsSource::new(
                    visible_artifacts,
                    read_namespaces(visible_namespaces)?,
                )];
                for source in additional_visible_artifacts {
                    visible_sources.push(VisibleGcsSource::new(
                        source.artifacts,
                        read_namespaces(source.namespaces)?,
                    ));
                }
                let mut fleet_sources = vec![FleetGcsSource::new(
                    fleet_artifacts,
                    read_namespaces(fleet_namespaces)?,
                )];
                for source in additional_fleet_artifacts {
                    fleet_sources.push(FleetGcsSource::new(
                        source.artifacts,
                        read_namespaces(source.namespaces)?,
                    ));
                }
                ComposedRealm::Fleet {
                    visible_sources,
                    fleet_sources,
                }
            }
        };

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

        // A zero spill quota is refused here rather than at the first query
        // that spills. `DataFusion` accepts it and then fails every spilling
        // query at execution time, which reads as a healthy pod that answers
        // some queries and not others.
        if policy.revalidation_interval.is_zero()
            || policy.execution_memory_bytes == 0
            || limits.spill_quota_bytes == 0
        {
            return Err(QueryServiceError::InvalidComposition);
        }
        // The spill root and quota come from `limits`, so the deployment decides
        // where a spilling query writes and how much it may write. `build_arc`
        // CREATES the directory here, at composition, which makes an unwritable
        // root a startup refusal rather than a failure on the first query that
        // spills.
        //
        // The quota must sit below the volume's own limit, which must sit below
        // the pod's. `DataFusion` refuses a query that would exceed this one;
        // the kubelet evicts a pod that exceeds either of the others, and an
        // eviction kills every concurrent stream rather than one query.
        let spill_guard = SpillDirGuard::new(limits.spill_dir.clone());
        let runtime = RuntimeEnvBuilder::new()
            .with_memory_pool(Arc::new(FairSpillPool::new(policy.execution_memory_bytes)))
            .with_temp_file_path(&limits.spill_dir)
            .with_max_temp_directory_size(limits.spill_quota_bytes)
            .build_arc()
            .map_err(|_datafusion| QueryServiceError::InvalidComposition)?;
        let session = projected_session_state(runtime)
            .map_err(|_datafusion| QueryServiceError::InvalidComposition)?;

        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,
            composite_traces: CompositeTraceCache::default(),
            spill_guard,
        })
    }

    /// Runs one credential-bearing projected query, admitted through the
    /// INTERNAL path — the plane-to-plane mutual-TLS listener, which also
    /// accepts a raw session bearer
    /// (`CredentialWitness::admit_bearer`'s own fallback).
    ///
    /// 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> {
        self.start_projected_admitted(credential, query_id, request, Admission::Internal)
            .await
    }

    /// Runs one credential-bearing projected query, admitted through the
    /// EDGE path — the public-reachable, grant-only listener (PR 10A) — see
    /// `docs/decisions/0022-explicit-query-credentials.md`'s "The edge
    /// listener" section.
    ///
    /// Two differences from [`Self::start_projected`], both enforced before
    /// anything else runs: the credential MUST be a signed grant — a raw
    /// session bearer is refused outright, never falling through to
    /// `CredentialWitness::admit_bearer`'s own fallback — and the resolved
    /// principal must be one this realm's edge admits
    /// (`admit_edge_principal_kind`): `Persona`/`ConversationGrant` on the
    /// Visible realm, `AdminFleet` on the Fleet realm, `ControlFleet` on
    /// neither (it is a capability Control holds for itself, never a caller).
    ///
    /// # Errors
    ///
    /// See [`Self::start_projected`]'s own doc — the same error shapes,
    /// [`QueryServiceError::Unauthorized`] additionally covering both edge-
    /// only refusals above.
    pub async fn start_projected_edge(
        &self,
        credential: QueryCredential,
        query_id: &str,
        request: &QueryRequest,
    ) -> Result<ProjectedResultStream, QueryServiceError> {
        self.start_projected_admitted(credential, query_id, request, Admission::Edge)
            .await
    }

    /// Answers `DescribeCatalog` on the internal listener — the tables and
    /// columns the presented credential may name, nothing more (POLY-367).
    ///
    /// The reply is a pure function of the build's registry and the verified
    /// scope: no artifact opens, no source resolves, and no audit record is
    /// written — a deliberate first, since the call changes nothing a caller
    /// could not already learn by planning each listed table. Every
    /// credential refusal [`Self::start_projected`] can return applies here
    /// identically, through the same shared admission prefix.
    ///
    /// # Errors
    ///
    /// See [`Self::start_projected`]. Additionally, a principal kind outside
    /// the catalog's self-service allowlist is refused on either listener —
    /// see `Self::describe_catalog_admitted`.
    pub async fn describe_catalog(
        &self,
        credential: QueryCredential,
    ) -> Result<CatalogReply, QueryServiceError> {
        self.describe_catalog_admitted(credential, Admission::Internal)
            .await
    }

    /// Answers `DescribeCatalog` on the EDGE listener — grant-only admission
    /// before anything else, exactly as [`Self::start_projected_edge`].
    ///
    /// # Errors
    ///
    /// See [`Self::describe_catalog`].
    pub async fn describe_catalog_edge(
        &self,
        credential: QueryCredential,
    ) -> Result<CatalogReply, QueryServiceError> {
        self.describe_catalog_admitted(credential, Admission::Edge)
            .await
    }

    async fn describe_catalog_admitted(
        &self,
        credential: QueryCredential,
        admission: Admission,
    ) -> Result<CatalogReply, QueryServiceError> {
        let (_witness, scoping) = self.admit_credential(credential, admission).await?;
        // "What may I query" is a self-service surface: it admits exactly
        // the edge listener's principal set on EITHER listener. The
        // statement-bound kinds — ControlFleet's fixed digest, the
        // composite-trace fixed statements, a turn's own bearer, a raw
        // admin session — have no catalog question a fixed answer could
        // serve, so the same allowlist refuses them here.
        admit_edge_principal_kind(&scoping, self.realm.kind())?;
        // The catalog holds the same execution permit `Execute` does: it is
        // the deployment's one bound on admitted work, and a lookup that
        // bypassed it would be an unmetered third path through the
        // semaphore. Acquired here, not in the shared prefix, because
        // `Execute` takes it inside `plan_admitted` after planning and the
        // prefix must not change that.
        let operation = CoreOperationContext::from_declared(
            &DeclaredCall::live(polyc_state_connect::state_audience(), self.limits.timeout),
            self.limits.timeout,
        );
        let _permit = self
            .admission
            .acquire(&operation, &CancellationToken::new())
            .await
            .map_err(|error| execution_refusal(&error))?;
        let realm = CoreRealm::from_scope(scoping.scope());
        let reply = catalog_reply(realm, scoping.scope())?;
        crate::metrics::record_catalog_described(
            catalog_realm_label(realm),
            catalog_principal_label(scoping.principal_kind()),
        );
        tracing::debug!(
            realm = catalog_realm_label(realm),
            principal_kind = catalog_principal_label(scoping.principal_kind()),
            tables = reply.tables().len(),
            "described the query catalog"
        );
        Ok(reply)
    }

    #[allow(
        clippy::too_many_lines,
        reason = "one method keeps credential admission, fixed-statement scoping, planning, binding, and observed release in order"
    )]
    async fn start_projected_admitted(
        &self,
        credential: QueryCredential,
        query_id: &str,
        request: &QueryRequest,
        admission: Admission,
    ) -> Result<ProjectedResultStream, QueryServiceError> {
        let credential_key: [u8; 32] = sha2::Sha256::digest(credential.0.as_bytes()).into();
        let (mut witness, scoping) = self.admit_credential(credential, admission).await?;
        admit_control_fleet(&scoping, self.realm.kind(), request.sql())?;
        let composite_statement = admit_composite_trace(&scoping, self.realm.kind(), request)?;
        let composite_scope = match composite_statement {
            Some(CompositeTraceStatement::Trace) | None => None,
            Some(CompositeTraceStatement::Memory) => {
                let conversation = scoping
                    .conversation_id()
                    .ok_or(QueryServiceError::Unauthorized)?;
                let personas = self
                    .composite_traces
                    .personas(&credential_key, conversation)
                    .ok_or(QueryServiceError::Unauthorized)?;
                witness.retain_composite_trace_memory_sources(personas.clone());
                Some(
                    self.credential
                        .composite_trace_memory_scope(&scoping, &personas)
                        .await
                        .map_err(refusal)?,
                )
            }
            Some(CompositeTraceStatement::Routine) => {
                witness.retain_composite_trace_routine();
                Some(QueryScope::Fleet)
            }
            Some(CompositeTraceStatement::Addresses) => {
                // One conversation on the Fleet realm, never Fleet-wide
                // (POLY-464). The witness re-derives this same scope on
                // every revalidation.
                let conversation = scoping
                    .conversation_id()
                    .ok_or(QueryServiceError::Unauthorized)?
                    .to_owned();
                witness.retain_composite_trace_addresses();
                Some(QueryScope::FleetConversation { conversation })
            }
        };
        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, &scoping))?;
        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 planned = if composite_statement == Some(CompositeTraceStatement::Memory) {
            self.planning
                .plan_composite_trace_memory(
                    &self.compiler,
                    &self.limits,
                    composite_scope.as_ref().unwrap_or_else(|| scoping.scope()),
                    scoping.allow_explain(),
                    audit,
                    core_request,
                )
                .await
        } else {
            self.planning
                .plan(
                    &self.compiler,
                    &self.limits,
                    composite_scope.as_ref().unwrap_or_else(|| scoping.scope()),
                    scoping.allow_explain(),
                    audit,
                    core_request,
                )
                .await
        };
        let prepared = match planned.map_err(|error| resolution_refusal(&error, &scoping))? {
            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();
        // Only a grant that may read memory retains the trace's attributed
        // personas; an admin composite trace has no memory statement to follow.
        let collects_memory_sources = scoping
            .composite_trace()
            .is_some_and(|capability| capability.memory_statement_digest().is_some());
        let collector = if composite_statement == Some(CompositeTraceStatement::Trace)
            && collects_memory_sources
        {
            Some(CompositeTraceCollector::try_new(
                self.composite_traces.clone(),
                credential_key,
                scoping
                    .conversation_id()
                    .map(str::to_owned)
                    .ok_or(QueryServiceError::Unauthorized)?,
                rows.schema(),
            )?)
        } else {
            None
        };
        ProjectedResultStream::start_observed(
            rows,
            bounds.response_frame_bytes(),
            bounds.result_release_bytes(),
            collector,
        )
    }

    /// Admits `credential` through `admission` and returns the witness and
    /// scoping it verified to — the prefix every RPC on this service shares
    /// (POLY-367).
    ///
    /// Covers only what every RPC enforces identically: witness verification
    /// (`admit_grant_only` at the edge, so a raw session bearer stays
    /// refused there), the edge principal-kind allowlist, and the
    /// admin-fleet realm check. Statement-bound admission — the control-fleet
    /// digest and the composite-trace statement match — stays in
    /// [`Self::start_projected_admitted`]: it needs the SQL a catalog call
    /// does not carry, and the catalog path refuses those principal kinds by
    /// allowlist instead.
    ///
    /// `admit_admin_fleet` runs ahead of `admit_control_fleet` here rather
    /// than after it as the inlined sequence once did. The two checks gate
    /// disjoint `Option` fields a single [`Scoping`] can never populate
    /// together, so the reorder is unobservable — and every refusal is the
    /// same collapsed [`QueryServiceError::Unauthorized`] regardless.
    ///
    /// The shared execution permit deliberately is NOT acquired here:
    /// `Execute` takes it inside `plan_admitted` only after planning, and
    /// pulling the acquisition forward would hold a slot across planning.
    /// `DescribeCatalog` acquires the same semaphore itself.
    ///
    /// # Errors
    ///
    /// Returns [`QueryServiceError::Unauthorized`] for any credential or
    /// principal-kind refusal; the reason stays off the wire.
    async fn admit_credential(
        &self,
        credential: QueryCredential,
        admission: Admission,
    ) -> Result<(CredentialWitness, Scoping), QueryServiceError> {
        let (witness, scoping) = match admission {
            Admission::Internal => {
                CredentialWitness::admit_bearer(
                    credential.0,
                    Arc::clone(&self.credential),
                    Arc::clone(&self.clock),
                )
                .await
            }
            Admission::Edge => {
                CredentialWitness::admit_grant_only(
                    credential.0,
                    Arc::clone(&self.credential),
                    Arc::clone(&self.clock),
                )
                .await
            }
        }
        .map_err(refusal)?;
        if admission == Admission::Edge {
            admit_edge_principal_kind(&scoping, self.realm.kind())?;
        }
        admit_admin_fleet(&scoping, self.realm.kind())?;
        Ok((witness, scoping))
    }
}

fn resolution_refusal(error: &CoreResolutionError, scoping: &Scoping) -> 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.
    let realm = catalog_realm_label(CoreRealm::from_scope(scoping.scope()));
    let principal_kind = catalog_principal_label(scoping.principal_kind());
    if let CoreResolutionError::Statement(rejected) = error {
        // POLY-371: the gate's refusal is the one place the reason is safe
        // to name — `reason_key` is a bounded label, never SQL text.
        crate::metrics::record_statement_refusal(realm, principal_kind, rejected.reason_key());
        tracing::warn!(
            realm,
            principal_kind,
            reason = rejected.reason_key(),
            "the projected read path refused a query statement"
        );
    } else {
        tracing::warn!(
            ?error,
            realm,
            principal_kind,
            "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
}

/// The `realm` label value [`crate::metrics::record_catalog_described`]
/// reports — the scope-derived realm the reply was computed for, not the
/// listener's.
const fn catalog_realm_label(realm: CoreRealm) -> &'static str {
    match realm {
        CoreRealm::Visible => "visible",
        CoreRealm::Fleet => "fleet",
    }
}

/// The `principal_kind` label value
/// [`crate::metrics::record_catalog_described`] reports. Exhaustive over
/// [`PrincipalKind`] even though only the catalog's admitted kinds can reach
/// the call — a future kind gets a label, not a panic or a fallback bucket.
const fn catalog_principal_label(kind: PrincipalKind) -> &'static str {
    match kind {
        PrincipalKind::Admin => "admin",
        PrincipalKind::ConversationGrantTurn => "conversation_grant_turn",
        PrincipalKind::ConversationGrantWebSession => "conversation_grant_web_session",
        PrincipalKind::ConversationGrantPersona => "conversation_grant_persona",
        PrincipalKind::ConversationGrantCompositeTrace => "conversation_grant_composite_trace",
        PrincipalKind::ConversationGrantAdminCompositeTrace => {
            "conversation_grant_admin_composite_trace"
        }
        PrincipalKind::Persona => "persona",
        PrincipalKind::ControlFleet => "control_fleet",
        PrincipalKind::AdminFleet => "admin_fleet",
    }
}

/// Builds one `DescribeCatalog` reply for (`realm`, `scope`) from the closed
/// registry — names and SQL-visible logical types only (POLY-367).
///
/// The schema source is each table's `public_schema`, not the declared
/// `TableSchema`: a view-renamed column — `Summary`'s `summary_id AS
/// turn_id`, for example — is what `SELECT` actually resolves, so it is what
/// the caller must see.
///
/// # Errors
///
/// Returns [`QueryServiceError::Execution`] with [`ErrorClass::Internal`]
/// when the registry describes a shape the protocol cannot carry — a column
/// type outside the closed [`CatalogColumnType`] set, or a name outside the
/// model's bounds.
/// That is a build defect, not caller input, so it fails the whole reply
/// rather than misdescribing one column.
fn catalog_reply(realm: CoreRealm, scope: &QueryScope) -> Result<CatalogReply, QueryServiceError> {
    let internal = || QueryServiceError::Execution(ErrorClass::Internal);
    let mut tables = Vec::new();
    for table in catalog_tables(realm, scope) {
        let columns = table
            .public_schema()
            .fields()
            .iter()
            .map(|field| {
                CatalogColumn::try_new(
                    field.name().clone(),
                    catalog_column_type(field.data_type())?,
                    field.is_nullable(),
                )
                .map_err(|_| internal())
            })
            .collect::<Result<Vec<_>, QueryServiceError>>()?;
        tables
            .push(CatalogTable::try_new(table.name().to_owned(), columns).map_err(|_| internal())?);
    }
    tables.sort_by(|a, b| a.name().cmp(b.name()));
    CatalogReply::try_new(tables).map_err(|_| internal())
}

/// Maps one public-schema Arrow type onto the catalog's closed logical-type
/// enum — the exact inverse of the registry's own `LogicalType` → `DataType`
/// mapping (`core_resolution::arrow_schema`, POLY-367).
///
/// A type outside the closed set — a registry change that outran this arm —
/// fails the reply closed rather than describe a column with the wrong type.
fn catalog_column_type(
    data_type: &arrow::datatypes::DataType,
) -> Result<CatalogColumnType, QueryServiceError> {
    match data_type {
        arrow::datatypes::DataType::Utf8 => Ok(CatalogColumnType::Utf8),
        arrow::datatypes::DataType::FixedSizeBinary(len) => Ok(CatalogColumnType::FixedBytes {
            len: u32::try_from(*len)
                .map_err(|_| QueryServiceError::Execution(ErrorClass::Internal))?,
        }),
        arrow::datatypes::DataType::UInt64 => Ok(CatalogColumnType::UInt64),
        arrow::datatypes::DataType::Boolean => Ok(CatalogColumnType::Boolean),
        _ => Err(QueryServiceError::Execution(ErrorClass::Internal)),
    }
}

/// The purpose an admin composite-trace read names in its audit requester.
const ADMIN_COMPOSITE_TRACE_PURPOSE: &str = "conversation.trace";

/// Derives the audit's requester from the verified scope.
///
/// Fails closed. A scope that names neither a caller, a conversation, nor a
/// Control-Fleet capability could not be attributed, and an unattributed
/// query is not one this plane runs.
fn requester_of(scoping: &Scoping) -> Result<RequesterId, QueryServiceError> {
    if let Some(control_fleet) = scoping.control_fleet() {
        // Names both the acting principal (Control, exercising this
        // purpose-bound capability rather than an unrestricted admin
        // session) and the purpose, in the one string the query-audit
        // schema has for a requester — see proof (3) in
        // `docs/decisions/0016-control-fleet-query-capability.md`.
        return Ok(RequesterId::new(format!(
            "control-fleet:{}",
            control_fleet.purpose()
        )));
    }
    if scoping.principal_kind() == PrincipalKind::ConversationGrantAdminCompositeTrace {
        // An admin composite trace names no persona (POLY-394). Control
        // mints it for itself after its own admission gate admitted a
        // service bearer, so the audit names it in the `ControlFleet` form,
        // never as an empty persona. The conversation rides in its own
        // audit field.
        return Ok(RequesterId::new(format!(
            "control-fleet:{ADMIN_COMPOSITE_TRACE_PURPOSE}"
        )));
    }
    if let Some(admin_fleet) = scoping.admin_fleet() {
        // Names both the acting admin persona and the session the grant
        // rides — the query-audit trail must name both, per
        // `docs/decisions/0022-explicit-query-credentials.md`'s proof (4),
        // never just a bare persona id indistinguishable from an ordinary
        // browser admin session's own audit row.
        return Ok(RequesterId::new(format!(
            "admin-fleet:{}:{}",
            admin_fleet.admin_persona(),
            admin_fleet.session()
        )));
    }
    scoping.caller_identity().map_or_else(
        || {
            scoping
                .conversation_id()
                .map_or(Err(QueryServiceError::Unauthorized), |conversation| {
                    Ok(RequesterId::new(format!("conversation:{conversation}")))
                })
        },
        |persona| Ok(RequesterId::new(format!("persona:{persona}"))),
    )
}

/// Admits (or refuses) a verified [`Principal::ControlFleet`](polyc_query_credential::principal::Principal::ControlFleet)
/// scope for the statement it is about to run.
///
/// Every other principal passes through untouched: this check applies
/// exclusively to the one capability that carries a `control_fleet` value at
/// all — an admin session, a persona session, and a conversation grant are
/// unaffected regardless of `realm` or `sql`.
///
/// Three independent conditions must all hold, and each is checked
/// explicitly rather than trusted from how the scope was derived — a
/// defense-in-depth posture against a future change to
/// `CredentialAuthority::scoping_for` that got the mapping wrong:
///
/// 1. `realm` must be [`RealmKind::Fleet`] — this capability exists to let
///    Control reach the standalone Fleet Query plane; the Visible realm has
///    no business honoring it at all.
/// 2. `scoping.scope` must be [`QueryScope::Fleet`] — a `ControlFleet`
///    principal resolved to any other scope is a contradiction this check
///    refuses rather than silently accepts.
/// 3. The SHA-256 of `sql`, lowercase-hex, must equal the grant's own
///    `statement_digest` exactly — a grant minted for one fixed statement
///    authorizes that statement alone.
///
/// # Errors
///
/// Returns [`QueryServiceError::Unauthorized`] when any of the three
/// conditions fails, or when the digest comparison itself cannot be formed
/// (never — the digest is always computable — but a failure here fails
/// closed on the same terms rather than unwrapping).
fn admit_control_fleet(
    scoping: &Scoping,
    realm: RealmKind,
    sql: &str,
) -> Result<(), QueryServiceError> {
    let Some(control_fleet) = scoping.control_fleet() else {
        return Ok(());
    };
    // Exhaustive `match`, not `!=`: a third `RealmKind` variant must force
    // every caller of this function to decide what it means here, rather
    // than silently inheriting whatever the "not fleet" branch already did.
    match realm {
        RealmKind::Fleet => {}
        RealmKind::Visible => {
            tracing::warn!(
                %realm,
                purpose = control_fleet.purpose(),
                "a control-fleet grant was presented outside the Fleet realm"
            );
            return Err(QueryServiceError::Unauthorized);
        }
    }
    if scoping.scope() != &QueryScope::Fleet {
        tracing::warn!(
            purpose = control_fleet.purpose(),
            "a control-fleet grant resolved to a non-Fleet scope"
        );
        return Err(QueryServiceError::Unauthorized);
    }
    let actual: [u8; 32] = sha2::Sha256::digest(sql.as_bytes()).into();
    if &actual != control_fleet.statement_digest() {
        tracing::warn!(
            purpose = control_fleet.purpose(),
            "a control-fleet grant's statement digest did not match the statement presented"
        );
        return Err(QueryServiceError::Unauthorized);
    }
    Ok(())
}

/// Admits (or refuses) a verified
/// [`Principal::AdminFleet`](polyc_query_credential::principal::Principal::AdminFleet) scope.
///
/// A sibling of [`admit_control_fleet`] with one fewer check: this
/// capability binds no statement digest, so there is nothing to recompute
/// against the SQL actually presented — the whole point of this subject is
/// that the caller may run ANY ad-hoc `SELECT`/`WITH` statement the shared
/// statement gate admits. Every other principal passes through untouched,
/// exactly like [`admit_control_fleet`].
///
/// Two independent conditions must both hold:
///
/// 1. `realm` must be [`RealmKind::Fleet`] — the Visible realm has no
///    business honoring an ad-hoc, fleet-wide credential at all.
/// 2. `scoping.scope` must be [`QueryScope::Fleet`] — an `AdminFleet`
///    principal resolved to any other scope is a contradiction this check
///    refuses rather than silently accepts.
///
/// The freshness check this credential's own safety depends on — that
/// `admin_persona` is CURRENTLY admin — already ran, in
/// `CredentialAuthority::scoping_for`, before this scope ever reached this
/// call: a `Scoping` carrying `admin_fleet: Some(_)` only exists because that
/// check passed.
///
/// # Errors
///
/// Returns [`QueryServiceError::Unauthorized`] when either condition fails.
/// Which listener admitted this call — the internal, mutual-TLS plane-to-
/// plane surface, or the public-reachable, grant-only edge (PR 10A). Decides
/// both which credential shapes [`ProjectedCoreService::start_projected_admitted`]
/// accepts and, for [`Self::Edge`], which resolved principal KINDS
/// [`admit_edge_principal_kind`] admits.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Admission {
    /// The mutual-TLS plane-to-plane listener — every credential shape this
    /// realm already admits, unchanged.
    Internal,
    /// The public-reachable edge listener — grant-only, and additionally
    /// restricted per realm; see [`admit_edge_principal_kind`].
    Edge,
}

/// Restricts which resolved principal KIND reaches a self-service surface:
/// the edge listener's `Execute`, and `DescribeCatalog` on EITHER listener
/// (POLY-367) — a caller allowed to ask what it may query over the public
/// listener is allowed the same answer internally, and nothing more is.
/// This is on top of whatever [`admit_control_fleet`]/[`admit_admin_fleet`]
/// already check.
///
/// An explicit ALLOWLIST over `(principal_kind, realm)`, not a blocklist: the
/// Visible edge admits exactly [`PrincipalKind::Persona`],
/// [`PrincipalKind::ConversationGrantWebSession`], and
/// [`PrincipalKind::ConversationGrantPersona`] (the routine-fire shape of a
/// `Persona` subject — see that variant's own doc); the Fleet edge admits
/// exactly [`PrincipalKind::AdminFleet`]. Everything else is refused,
/// including [`PrincipalKind::ConversationGrantTurn`] (a harness turn's own
/// budget-scoped bearer, never meant to leave the plane-to-plane listener —
/// the edge admitting it would let an exfiltrated turn credential reach a
/// public-facing listener it was never designed to be presented to) and
/// [`PrincipalKind::ControlFleet`] (Control's own capability, minted for
/// Control's own dashboard reads, never presented by an outside caller). An
/// allowlist rather than a blocklist means a future [`PrincipalKind`] variant
/// is refused by default until someone deliberately adds it here, not
/// admitted by default until someone remembers to exclude it.
///
/// # Errors
///
/// Returns [`QueryServiceError::Unauthorized`] when `scoping` carries a
/// principal kind this realm's edge does not admit.
fn admit_edge_principal_kind(scoping: &Scoping, realm: RealmKind) -> Result<(), QueryServiceError> {
    let admitted = matches!(
        (scoping.principal_kind(), realm),
        (
            PrincipalKind::Persona
                | PrincipalKind::ConversationGrantWebSession
                | PrincipalKind::ConversationGrantPersona,
            RealmKind::Visible
        ) | (PrincipalKind::AdminFleet, RealmKind::Fleet)
    );
    if admitted {
        Ok(())
    } else {
        tracing::warn!(
            %realm,
            principal_kind = ?scoping.principal_kind(),
            "a grant resolved to a principal kind this realm's edge listener does not admit"
        );
        Err(QueryServiceError::Unauthorized)
    }
}

fn admit_admin_fleet(scoping: &Scoping, realm: RealmKind) -> Result<(), QueryServiceError> {
    let Some(admin_fleet) = scoping.admin_fleet() else {
        return Ok(());
    };
    match realm {
        RealmKind::Fleet => {}
        RealmKind::Visible => {
            tracing::warn!(
                %realm,
                admin_persona = admin_fleet.admin_persona(),
                "an admin-fleet grant was presented outside the Fleet realm"
            );
            return Err(QueryServiceError::Unauthorized);
        }
    }
    if scoping.scope() != &QueryScope::Fleet {
        tracing::warn!(
            admin_persona = admin_fleet.admin_persona(),
            "an admin-fleet grant resolved to a non-Fleet scope"
        );
        return Err(QueryServiceError::Unauthorized);
    }
    Ok(())
}

/// Admits one of the fixed statements carried by a composite-trace grant and
/// returns the scope the selected Query realm must plan.
///
/// Trace and memory run on the visible service with the grant's verified
/// conversation scope. The routine statement runs on the Fleet service and
/// receives Fleet catalog scope, but only after its fixed SQL and its
/// conversation parameter match the signed grant. The same bearer therefore
/// crosses both services without becoming a general Fleet credential.
///
/// The participant-address statement (POLY-464) runs on the Fleet service
/// only, under an admin composite-trace grant only, and only for the grant's
/// own conversation partition. It receives the one-conversation Fleet scope,
/// never the Fleet-wide one.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CompositeTraceStatement {
    Trace,
    Memory,
    Routine,
    Addresses,
}

fn admit_composite_trace(
    scoping: &Scoping,
    realm: RealmKind,
    request: &QueryRequest,
) -> Result<Option<CompositeTraceStatement>, QueryServiceError> {
    let Some(capability) = scoping.composite_trace() else {
        return Ok(None);
    };
    let conversation = scoping
        .conversation_id()
        .ok_or(QueryServiceError::Unauthorized)?;
    let digest = polyc_crypto::hex::lower(&sha2::Sha256::digest(request.sql().as_bytes()));
    let parameters = request.parameters();
    let conversation_parameter = |index: usize| matches!(parameters.get(index), Some(polyc_query_model::Parameter::Utf8(value)) if value == conversation);
    let conversation_partition = format!("conv-{conversation}");
    let partition_parameter = |index: usize| matches!(parameters.get(index), Some(polyc_query_model::Parameter::Utf8(value)) if value == &conversation_partition);
    if realm == RealmKind::Visible
        && digest == capability.trace_statement_digest()
        && parameters.len() == 1
        && partition_parameter(0)
    {
        return Ok(Some(CompositeTraceStatement::Trace));
    }
    // The memory statement needs both a memory digest and an owner persona.
    // An admin composite trace carries neither (POLY-394), so no parameter
    // can make it match: there is no owner partition to form.
    if let (Some(memory_digest), Some(owner)) = (
        capability.memory_statement_digest(),
        scoping.caller_identity(),
    ) && realm == RealmKind::Visible
        && digest == memory_digest
        && parameters.len() == 3
        && matches!(parameters.first(), Some(polyc_query_model::Parameter::Utf8(value)) if value == &format!("persona-{owner}-mem"))
        && conversation_parameter(1)
        && partition_parameter(2)
    {
        return Ok(Some(CompositeTraceStatement::Memory));
    }
    if realm == RealmKind::Fleet
        && digest == capability.routine_statement_digest()
        && parameters.len() == 1
        && conversation_parameter(0)
    {
        return Ok(Some(CompositeTraceStatement::Routine));
    }
    // The participant-address statement needs the address digest, which
    // only an admin composite-trace grant carries (POLY-464). A persona
    // grant has none, so no parameter can make it match. The digest must
    // also name this build's own fixed statement, not only the signed one.
    let own_address_digest = polyc_crypto::hex::lower(&sha2::Sha256::digest(
        polyc_query_model::statements::COMPOSITE_TRACE_ADDRESSES_SQL.as_bytes(),
    ));
    if let Some(address_digest) = capability.address_statement_digest()
        && scoping.principal_kind() == PrincipalKind::ConversationGrantAdminCompositeTrace
        && realm == RealmKind::Fleet
        && digest == address_digest
        && digest == own_address_digest
        && parameters.len() == 1
        && partition_parameter(0)
    {
        return Ok(Some(CompositeTraceStatement::Addresses));
    }
    tracing::warn!(%realm, "a composite-trace grant did not match its fixed statement");
    Err(QueryServiceError::Unauthorized)
}

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,
    composite_trace: Option<CompositeTraceCollector>,
}

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 {
    #[cfg(test)]
    pub(crate) fn start(
        rows: CoreResultStream,
        frame_ceiling: u64,
        release_ceiling: u64,
    ) -> Result<Self, QueryServiceError> {
        Self::start_observed(rows, frame_ceiling, release_ceiling, None)
    }

    fn start_observed(
        rows: CoreResultStream,
        frame_ceiling: u64,
        release_ceiling: u64,
        composite_trace: Option<CompositeTraceCollector>,
    ) -> 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,
            composite_trace,
        })
    }

    /// 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;
        if matches!(&outcome, QueryOutcome::Succeeded)
            && !self.byte_truncated
            && let Some(observer) = self.composite_trace.take()
        {
            observer.finish();
        }
        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)) => {
                    if let Some(observer) = &mut self.composite_trace
                        && let Err(class) = observer.observe(&batch)
                    {
                        let _ = self
                            .rows
                            .report_failure(crate::core_evidence::durable_class_of(class));
                        return Some(self.release_terminal(QueryOutcome::Failed(class)));
                    }
                    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)))
    })
}

#[cfg(test)]
mod control_fleet_tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use polyc_query_credential::principal::ControlFleetPrincipal;

    use super::*;

    const STATEMENT_A: &str = "SELECT partition FROM turns";
    const STATEMENT_B: &str = "SELECT partition FROM usage";

    fn digest_of(sql: &str) -> [u8; 32] {
        sha2::Sha256::digest(sql.as_bytes()).into()
    }

    fn control_fleet_scoping(purpose: &str, digest: [u8; 32]) -> Scoping {
        Scoping::for_test(
            QueryScope::Fleet,
            false,
            None,
            None,
            None,
            None,
            Some(ControlFleetPrincipal::minted(purpose.to_owned(), digest)),
            None,
            None,
            PrincipalKind::ControlFleet,
        )
    }

    fn admin_fleet_scoping(admin_persona: &str, session: &str) -> Scoping {
        Scoping::for_test(
            QueryScope::Fleet,
            true,
            Some(admin_persona.to_owned()),
            None,
            None,
            None,
            None,
            Some(
                polyc_query_credential::principal::AdminFleetPrincipal::minted(
                    admin_persona.to_owned(),
                    session.to_owned(),
                ),
            ),
            None,
            PrincipalKind::AdminFleet,
        )
    }

    /// A grant is admitted when its realm, scope, and digest all agree.
    #[test]
    fn a_matching_grant_is_admitted() {
        let scoping = control_fleet_scoping("dashboard.conversations", digest_of(STATEMENT_A));
        assert!(admit_control_fleet(&scoping, RealmKind::Fleet, STATEMENT_A).is_ok());
    }

    /// Proof: a grant for statement A is refused for statement B.
    #[test]
    fn a_grant_for_one_statement_is_refused_for_another() {
        let scoping = control_fleet_scoping("dashboard.conversations", digest_of(STATEMENT_A));
        assert!(matches!(
            admit_control_fleet(&scoping, RealmKind::Fleet, STATEMENT_B),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof: refused in the Visible realm, even with a matching digest.
    #[test]
    fn a_matching_grant_is_refused_outside_the_fleet_realm() {
        let scoping = control_fleet_scoping("dashboard.conversations", digest_of(STATEMENT_A));
        assert!(matches!(
            admit_control_fleet(&scoping, RealmKind::Visible, STATEMENT_A),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof: refused with a conversation scope. This scope shape can never
    /// come out of `CredentialAuthority::scoping_for`'s own `ControlFleet`
    /// arm — the point of this test is that `admit_control_fleet` refuses it
    /// on its own, independent of that mapping ever being correct.
    #[test]
    fn a_matching_grant_is_refused_under_a_conversation_scope() {
        let mut scoping = control_fleet_scoping("dashboard.conversations", digest_of(STATEMENT_A));
        scoping.set_scope_for_test(QueryScope::Conversations {
            conversations: vec!["conv-1".to_owned()],
            memory: polyc_query_credential::session::MemorySources::default(),
        });
        assert!(matches!(
            admit_control_fleet(&scoping, RealmKind::Fleet, STATEMENT_A),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// A scope with no `control_fleet` value passes through untouched,
    /// regardless of realm or statement — this check applies to exactly one
    /// principal kind.
    #[test]
    fn a_non_control_fleet_scope_is_never_refused_here() {
        let scoping = Scoping::for_test(
            QueryScope::Fleet,
            true,
            Some("persona-1".to_owned()),
            None,
            None,
            None,
            None,
            None,
            None,
            PrincipalKind::Admin,
        );
        assert!(admit_control_fleet(&scoping, RealmKind::Visible, "SELECT 1").is_ok());
    }

    /// Proof: the audit row names both the caller and the purpose — the one
    /// requester string a query-audit intent carries.
    #[test]
    fn the_requester_names_the_control_fleet_purpose() {
        let scoping = control_fleet_scoping("dashboard.spend", digest_of(STATEMENT_A));
        let requester = requester_of(&scoping).expect("a control-fleet scope is attributable");
        assert_eq!(requester.as_str(), "control-fleet:dashboard.spend");
    }

    /// A scope with no caller, conversation, or control-fleet value is
    /// unattributable and fails closed, unchanged from before this
    /// capability existed.
    #[test]
    fn an_unattributed_scope_still_refuses() {
        let scoping = Scoping::for_test(
            QueryScope::Fleet,
            false,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            PrincipalKind::Admin,
        );
        assert!(matches!(
            requester_of(&scoping),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (extra, PR 10A): an admin-fleet grant is admitted in the Fleet
    /// realm and refused outside it.
    #[test]
    fn an_admin_fleet_grant_is_refused_outside_the_fleet_realm() {
        let scoping = admin_fleet_scoping("persona-admin", "session-a");
        assert!(admit_admin_fleet(&scoping, RealmKind::Fleet).is_ok());
        assert!(matches!(
            admit_admin_fleet(&scoping, RealmKind::Visible),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (extra, PR 10A): an admin-fleet grant resolved to a non-Fleet
    /// scope is refused, mirroring `a_matching_grant_is_refused_under_a_
    /// conversation_scope` for `ControlFleet`.
    #[test]
    fn an_admin_fleet_grant_is_refused_under_a_conversation_scope() {
        let mut scoping = admin_fleet_scoping("persona-admin", "session-a");
        scoping.set_scope_for_test(QueryScope::Conversations {
            conversations: vec!["conv-1".to_owned()],
            memory: polyc_query_credential::session::MemorySources::default(),
        });
        assert!(matches!(
            admit_admin_fleet(&scoping, RealmKind::Fleet),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (extra, PR 10A): the audit row names both the admin persona and
    /// the session — never just a bare persona id.
    #[test]
    fn the_requester_names_the_admin_fleet_persona_and_session() {
        let scoping = admin_fleet_scoping("persona-admin", "session-a");
        let requester = requester_of(&scoping).expect("an admin-fleet scope is attributable");
        assert_eq!(requester.as_str(), "admin-fleet:persona-admin:session-a");
    }

    /// The `Principal::Persona` shape — a persona-wide grant or an ordinary
    /// non-admin explorer session — the standalone-visible-edge-eligible
    /// case that carries no `conversation_id` at all.
    fn persona_principal_scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conv-1".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            Some("persona-1".to_owned()),
            None,
            None,
            None,
            None,
            None,
            None,
            PrincipalKind::Persona,
        )
    }

    /// A `Principal::ConversationGrant` whose subject is
    /// `GrantSubject::WebSession`.
    fn websession_conversation_grant_scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conv-1".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            None,
            Some("conv-1".to_owned()),
            None,
            Some("session-1".to_owned()),
            None,
            None,
            None,
            PrincipalKind::ConversationGrantWebSession,
        )
    }

    /// A `Principal::ConversationGrant` whose subject is `GrantSubject::Persona`
    /// with a real conversation id — the routine-fire shape.
    fn persona_conversation_grant_scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conv-1".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            Some("persona-1".to_owned()),
            Some("conv-1".to_owned()),
            None,
            None,
            None,
            None,
            None,
            PrincipalKind::ConversationGrantPersona,
        )
    }

    /// A `Principal::ConversationGrant` whose subject is `GrantSubject::Turn`
    /// — a harness turn's own budget-scoped bearer, never meant to reach the
    /// public edge listener at all.
    fn turn_conversation_grant_scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conv-1".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            None,
            Some("conv-1".to_owned()),
            Some("turn-1".to_owned()),
            None,
            None,
            None,
            None,
            PrincipalKind::ConversationGrantTurn,
        )
    }

    /// Proof (extra, PR 10A): `ControlFleet` is refused on EITHER edge — it
    /// is Control's own capability, never a caller's.
    #[test]
    fn edge_refuses_a_control_fleet_grant_on_either_realm() {
        let scoping = control_fleet_scoping("dashboard.conversations", digest_of(STATEMENT_A));
        assert!(matches!(
            admit_edge_principal_kind(&scoping, RealmKind::Visible),
            Err(QueryServiceError::Unauthorized)
        ));
        assert!(matches!(
            admit_edge_principal_kind(&scoping, RealmKind::Fleet),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (extra, PR 10A): the Visible edge admits a `Persona` principal,
    /// a `WebSession`-subject conversation grant, and the `Persona`-subject
    /// (routine-fire) conversation grant — the three shapes item 2 of the
    /// review names — and refuses every one of them at the Fleet edge.
    #[test]
    fn visible_edge_admits_persona_and_websession_shapes_fleet_edge_refuses_them() {
        for scoping in [
            persona_principal_scoping(),
            websession_conversation_grant_scoping(),
            persona_conversation_grant_scoping(),
        ] {
            assert!(admit_edge_principal_kind(&scoping, RealmKind::Visible).is_ok());
            assert!(matches!(
                admit_edge_principal_kind(&scoping, RealmKind::Fleet),
                Err(QueryServiceError::Unauthorized)
            ));
        }
    }

    /// Proof (extra, PR 10A, review round 2): a `Turn`-subject conversation
    /// grant — a harness turn's own budget-scoped bearer — is refused at
    /// BOTH edges, not just the Fleet one. Before this review round,
    /// `admit_edge_principal_kind` admitted anything on the Visible realm
    /// that wasn't `admin_fleet`, which let a Turn subject through: mutate
    /// the allowlist back to that shape (drop the `ConversationGrantTurn`
    /// exclusion, or drop this whole match arm) and this test catches it.
    #[test]
    fn edge_refuses_a_turn_grant_on_either_realm() {
        let scoping = turn_conversation_grant_scoping();
        assert!(matches!(
            admit_edge_principal_kind(&scoping, RealmKind::Visible),
            Err(QueryServiceError::Unauthorized)
        ));
        assert!(matches!(
            admit_edge_principal_kind(&scoping, RealmKind::Fleet),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (extra, PR 10A): the Visible edge admits
    /// `Persona`/`ConversationGrant` — never `AdminFleet`, a Fleet-only
    /// capability.
    #[test]
    fn visible_edge_refuses_admin_fleet() {
        let scoping = admin_fleet_scoping("persona-admin", "session-a");
        assert!(matches!(
            admit_edge_principal_kind(&scoping, RealmKind::Visible),
            Err(QueryServiceError::Unauthorized)
        ));
        assert!(admit_edge_principal_kind(&scoping, RealmKind::Fleet).is_ok());
    }

    /// A grant minted for one real ControlFleet statement is refused for
    /// every other. Uses production SQL text, not synthetic strings, so a
    /// purpose-bound digest isolates the dashboard statements and the two
    /// fires statements from one another.
    #[test]
    fn each_real_control_fleet_statement_is_refused_for_the_others() {
        let statements = [
            (
                "dashboard.conversations",
                polyc_query_model::statements::DASHBOARD_CONVERSATIONS_SQL,
            ),
            (
                "dashboard.spend",
                polyc_query_model::statements::DASHBOARD_SPEND_SQL,
            ),
            (
                "dashboard.attribution",
                polyc_query_model::statements::DASHBOARD_ATTRIBUTION_SQL,
            ),
            (
                "dashboard.context",
                polyc_query_model::statements::DASHBOARD_CONTEXT_SQL,
            ),
            (
                "routines.fires",
                polyc_query_model::statements::ROUTINE_FIRES_SQL,
            ),
            (
                "routines.fire",
                polyc_query_model::statements::ROUTINE_FIRE_SQL,
            ),
        ];
        for (purpose, sql) in statements {
            let scoping = control_fleet_scoping(purpose, digest_of(sql));
            assert!(
                admit_control_fleet(&scoping, RealmKind::Fleet, sql).is_ok(),
                "{purpose}'s own grant must admit its own statement"
            );
            for (_, other_sql) in statements {
                if other_sql == sql {
                    continue;
                }
                assert!(
                    matches!(
                        admit_control_fleet(&scoping, RealmKind::Fleet, other_sql),
                        Err(QueryServiceError::Unauthorized)
                    ),
                    "{purpose}'s grant must be refused for a different fixed statement"
                );
            }
        }
    }
}

#[cfg(test)]
mod composite_trace_tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use arrow::array::{ArrayRef, StringArray};
    use arrow::datatypes::{DataType, Field, Schema};
    use arrow::record_batch::RecordBatch;
    use polyc_query_model::{Consistency, Parameter, RequestedBounds};

    use polyc_query_credential::principal::CompositeTracePrincipal;

    use super::*;

    const TRACE: &str = polyc_query_model::statements::COMPOSITE_TRACE_SQL;
    const MEMORY: &str = polyc_query_model::statements::COMPOSITE_TRACE_MEMORY_SQL;
    const ROUTINE: &str = polyc_query_model::statements::COMPOSITE_TRACE_ROUTINES_SQL;
    const ADDRESSES: &str = polyc_query_model::statements::COMPOSITE_TRACE_ADDRESSES_SQL;

    fn digest(sql: &str) -> [u8; 32] {
        sha2::Sha256::digest(sql.as_bytes()).into()
    }

    fn scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conversation-a".to_owned()],
                memory: polyc_query_credential::session::MemorySources {
                    owner: Some("persona-a".to_owned()),
                    participants: Vec::new(),
                },
            },
            false,
            Some("persona-a".to_owned()),
            Some("conversation-a".to_owned()),
            None,
            None,
            None,
            None,
            Some(CompositeTracePrincipal::minted(
                &polyc_crypto::hex::lower(&digest(TRACE)),
                &polyc_crypto::hex::lower(&digest(MEMORY)),
                &polyc_crypto::hex::lower(&digest(ROUTINE)),
            )),
            PrincipalKind::ConversationGrantCompositeTrace,
        )
    }

    fn request(sql: &str, parameters: Vec<Parameter>) -> QueryRequest {
        QueryRequest::try_new(
            sql.to_owned(),
            parameters,
            Consistency::Projected,
            RequestedBounds::try_new(Duration::from_secs(5), 100, 1024 * 1024, 64 * 1024).unwrap(),
        )
        .unwrap()
    }

    #[test]
    fn only_the_three_exact_statements_and_parameters_are_admitted() {
        let scoped = scoping();
        let trace = request(
            TRACE,
            vec![Parameter::Utf8("conv-conversation-a".to_owned())],
        );
        assert_eq!(
            admit_composite_trace(&scoped, RealmKind::Visible, &trace).unwrap(),
            Some(CompositeTraceStatement::Trace)
        );
        let memory = request(
            MEMORY,
            vec![
                Parameter::Utf8("persona-persona-a-mem".to_owned()),
                Parameter::Utf8("conversation-a".to_owned()),
                Parameter::Utf8("conv-conversation-a".to_owned()),
            ],
        );
        assert_eq!(
            admit_composite_trace(&scoped, RealmKind::Visible, &memory).unwrap(),
            Some(CompositeTraceStatement::Memory)
        );
        let routine = request(ROUTINE, vec![Parameter::Utf8("conversation-a".to_owned())]);
        assert_eq!(
            admit_composite_trace(&scoped, RealmKind::Fleet, &routine).unwrap(),
            Some(CompositeTraceStatement::Routine)
        );

        let caller_statement = request(
            &format!("{TRACE}\n-- caller text"),
            vec![Parameter::Utf8("conversation-a".to_owned())],
        );
        assert!(matches!(
            admit_composite_trace(&scoped, RealmKind::Visible, &caller_statement),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// The scope an admin composite-trace grant resolves to: no caller
    /// identity, no owner memory partition, and no memory digest (POLY-394).
    fn admin_scoping() -> Scoping {
        Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conversation-a".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            None,
            Some("conversation-a".to_owned()),
            None,
            None,
            None,
            None,
            Some(CompositeTracePrincipal::minted_without_memory(
                &polyc_crypto::hex::lower(&digest(TRACE)),
                &polyc_crypto::hex::lower(&digest(ROUTINE)),
                &polyc_crypto::hex::lower(&digest(ADDRESSES)),
            )),
            PrincipalKind::ConversationGrantAdminCompositeTrace,
        )
    }

    fn memory_request(owner_partition: &str) -> QueryRequest {
        request(
            MEMORY,
            vec![
                Parameter::Utf8(owner_partition.to_owned()),
                Parameter::Utf8("conversation-a".to_owned()),
                Parameter::Utf8("conv-conversation-a".to_owned()),
            ],
        )
    }

    /// Proof (4): the memory statement is refused under the admin subject,
    /// with the exact memory statement and a well-formed owner partition.
    /// Even a scope that somehow carried the memory digest and a caller
    /// identity would refuse: the admin grant has neither to offer.
    #[test]
    fn the_admin_subject_never_admits_the_memory_statement() {
        let scoped = admin_scoping();
        for owner_partition in ["persona-x-mem", "persona--mem"] {
            assert!(matches!(
                admit_composite_trace(
                    &scoped,
                    RealmKind::Visible,
                    &memory_request(owner_partition)
                ),
                Err(QueryServiceError::Unauthorized)
            ));
        }
    }

    /// Proof (5): the trace and routine statements are admitted under the
    /// admin subject with the same parameter rules as the persona subject.
    #[test]
    fn the_admin_subject_admits_trace_and_routine_under_the_same_rules() {
        let scoped = admin_scoping();
        let trace = request(
            TRACE,
            vec![Parameter::Utf8("conv-conversation-a".to_owned())],
        );
        assert_eq!(
            admit_composite_trace(&scoped, RealmKind::Visible, &trace).unwrap(),
            Some(CompositeTraceStatement::Trace)
        );
        let routine = request(ROUTINE, vec![Parameter::Utf8("conversation-a".to_owned())]);
        assert_eq!(
            admit_composite_trace(&scoped, RealmKind::Fleet, &routine).unwrap(),
            Some(CompositeTraceStatement::Routine)
        );

        for (realm, refused) in [
            (
                RealmKind::Visible,
                request(
                    TRACE,
                    vec![Parameter::Utf8("conv-conversation-b".to_owned())],
                ),
            ),
            (RealmKind::Fleet, trace.clone()),
            (
                RealmKind::Fleet,
                request(ROUTINE, vec![Parameter::Utf8("conversation-b".to_owned())]),
            ),
            (RealmKind::Visible, routine.clone()),
        ] {
            assert!(matches!(
                admit_composite_trace(&scoped, realm, &refused),
                Err(QueryServiceError::Unauthorized)
            ));
        }
    }

    fn address_request(parameters: Vec<Parameter>) -> QueryRequest {
        request(ADDRESSES, parameters)
    }

    /// POLY-464: the admin subject admits the participant-address statement
    /// only on the Fleet service, and only with its own conversation
    /// partition as the one parameter.
    #[test]
    fn the_admin_subject_admits_the_address_statement_only_on_fleet() {
        let scoped = admin_scoping();
        let own = || vec![Parameter::Utf8("conv-conversation-a".to_owned())];
        let admitted = admit_composite_trace(&scoped, RealmKind::Fleet, &address_request(own()));
        assert!(
            matches!(admitted, Ok(Some(CompositeTraceStatement::Addresses))),
            "the admin subject reads its own addresses on Fleet: {admitted:?}"
        );

        for (realm, refused) in [
            (RealmKind::Visible, address_request(own())),
            (
                RealmKind::Fleet,
                address_request(vec![Parameter::Utf8("conv-conversation-b".to_owned())]),
            ),
            (
                RealmKind::Fleet,
                address_request(vec![Parameter::Utf8("conversation-a".to_owned())]),
            ),
            (
                RealmKind::Fleet,
                address_request(vec![
                    Parameter::Utf8("conv-conversation-a".to_owned()),
                    Parameter::Utf8("conv-conversation-a".to_owned()),
                ]),
            ),
            (RealmKind::Fleet, address_request(Vec::new())),
        ] {
            assert!(
                matches!(
                    admit_composite_trace(&scoped, realm, &refused),
                    Err(QueryServiceError::Unauthorized)
                ),
                "refused on {realm} with {:?}",
                refused.parameters()
            );
        }
    }

    /// POLY-464: a scoping of the persona composite-trace kind never admits
    /// the participant-address statement, even when its capability carries
    /// the address digest. Only the admin kind reads the addresses.
    #[test]
    fn a_persona_kind_with_an_address_digest_is_refused() {
        let scoped = Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conversation-a".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            Some("persona-a".to_owned()),
            Some("conversation-a".to_owned()),
            None,
            None,
            None,
            None,
            Some(CompositeTracePrincipal::minted_without_memory(
                &polyc_crypto::hex::lower(&digest(TRACE)),
                &polyc_crypto::hex::lower(&digest(ROUTINE)),
                &polyc_crypto::hex::lower(&digest(ADDRESSES)),
            )),
            PrincipalKind::ConversationGrantCompositeTrace,
        );
        assert!(matches!(
            admit_composite_trace(
                &scoped,
                RealmKind::Fleet,
                &address_request(vec![Parameter::Utf8("conv-conversation-a".to_owned())]),
            ),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// POLY-464: an admin grant whose signed address digest names other SQL
    /// is refused. The statement must be this build's own fixed text.
    #[test]
    fn an_address_digest_for_other_sql_is_refused() {
        const OTHER: &str = "SELECT address FROM trace_participant_addresses";
        let scoped = Scoping::for_test(
            QueryScope::Conversations {
                conversations: vec!["conversation-a".to_owned()],
                memory: polyc_query_credential::session::MemorySources::default(),
            },
            false,
            None,
            Some("conversation-a".to_owned()),
            None,
            None,
            None,
            None,
            Some(CompositeTracePrincipal::minted_without_memory(
                &polyc_crypto::hex::lower(&digest(TRACE)),
                &polyc_crypto::hex::lower(&digest(ROUTINE)),
                &polyc_crypto::hex::lower(&digest(OTHER)),
            )),
            PrincipalKind::ConversationGrantAdminCompositeTrace,
        );
        assert!(matches!(
            admit_composite_trace(
                &scoped,
                RealmKind::Fleet,
                &request(
                    OTHER,
                    vec![Parameter::Utf8("conv-conversation-a".to_owned())]
                ),
            ),
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// POLY-464: a persona composite-trace grant never admits the
    /// participant-address statement. It carries no address digest, so the
    /// statement matches nothing on either service.
    #[test]
    fn no_persona_subject_admits_the_address_statement() {
        let scoped = scoping();
        for realm in [RealmKind::Visible, RealmKind::Fleet] {
            assert!(matches!(
                admit_composite_trace(
                    &scoped,
                    realm,
                    &address_request(vec![Parameter::Utf8("conv-conversation-a".to_owned())]),
                ),
                Err(QueryServiceError::Unauthorized)
            ));
        }
    }

    /// Proof (6): the audit requester for an admin composite trace names the
    /// Control-Fleet form, never an empty persona. A persona composite trace
    /// still names its persona.
    #[test]
    fn the_admin_composite_trace_requester_is_never_an_empty_persona() {
        let requester = requester_of(&admin_scoping()).expect("the admin read is attributable");
        assert_eq!(requester.as_str(), "control-fleet:conversation.trace");
        let persona = requester_of(&scoping()).expect("the persona read is attributable");
        assert_eq!(persona.as_str(), "persona:persona-a");
    }

    #[test]
    fn trace_rows_supply_the_complete_memory_source_proposal() {
        let cache = CompositeTraceCache::default();
        let schema = Arc::new(Schema::new(vec![
            Field::new("caller_persona_id", DataType::Utf8, false),
            Field::new("record_persona_id", DataType::Utf8, false),
        ]));
        let mut collector = CompositeTraceCollector::try_new(
            cache.clone(),
            [7; 32],
            "conversation-a".to_owned(),
            &schema,
        )
        .unwrap();
        let columns: Vec<ArrayRef> = vec![
            Arc::new(StringArray::from(vec!["persona-a", "persona-a"])),
            Arc::new(StringArray::from(vec!["", "persona-b"])),
        ];
        collector
            .observe(&RecordBatch::try_new(schema, columns).unwrap())
            .unwrap();
        collector.finish();
        assert_eq!(
            cache.personas(&[7; 32], "conversation-a").unwrap(),
            vec!["persona-a".to_owned(), "persona-b".to_owned()]
        );
        assert!(cache.personas(&[7; 32], "conversation-b").is_none());
    }
}

#[cfg(test)]
mod catalog_tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    //! `DescribeCatalog` end to end through `ProjectedCoreService`: the real
    //! admission prefix, the real principal allowlist, the real permit — over
    //! fakes for the two authority ports (persona reads, session
    //! verification) and a transport that fails if any State call is made.
    //! The catalog path makes none.

    use std::collections::BTreeMap;
    use std::task::{Context, Poll};

    use bytes::Bytes;
    use connectrpc::client::{ClientBody, ClientConfig, ClientTransport};
    use futures::future::BoxFuture;
    use polyc_crypto::session::{AuthorizedSessionClaims, SessionScope, SessionSubject};
    use polyc_crypto::signing_role::RoleSigner;
    use polyc_persona::{ActivePersona, PersonaError, PersonaReferenceSnapshot, ScopeResolution};
    use polyc_proto::proto::polychrome::persona::v1::Participation;
    use polyc_query_model::GrantSubject;
    use polyc_session_family::authority::SessionAuthorityError;
    use polyc_state::projection::artifact::ManifestVerdict;
    use polyc_state_connect::observation::ObservationClient;
    use polyc_state_connect::persona_memory_journal::PersonaMemoryJournalClient;
    use polyc_state_connect::versioned_source::VersionedSourceClient;
    use polyc_storage_gcs::{GcsConfig, GcsCredentials};

    use super::*;
    use crate::core_resolution::all_tables;

    const CONVERSATION: &str = "conv-a";
    const PERSONA: &str = "persona-a";
    const ADMIN: &str = "persona-admin";
    const PERSONA_BEARER: &str = "bearer-persona-a";
    const ADMIN_BEARER: &str = "bearer-persona-admin";
    /// Far-future expiry for every minted grant: `expiry > now` is all the
    /// verifier checks, and `u64::MAX` halves cleanly under any clock.
    const EXPIRES: u64 = u64::MAX / 2;

    fn signer() -> RoleSigner<TurnReadRole> {
        RoleSigner::from_key_bytes(&[23_u8; 32]).expect("a well-formed signing key")
    }

    /// The persona reads the credential mechanism needs, faked.
    struct FakePersonas {
        personas: BTreeMap<String, ActivePersona>,
        participations: BTreeMap<String, Vec<Participation>>,
    }

    #[async_trait::async_trait]
    impl PersonaSource for FakePersonas {
        async fn active_persona(
            &self,
            persona_id: String,
        ) -> Result<Option<ActivePersona>, PersonaError> {
            Ok(self.personas.get(&persona_id).cloned())
        }

        async fn participations(
            &self,
            persona_id: String,
        ) -> Result<Vec<Participation>, PersonaError> {
            Ok(self
                .participations
                .get(&persona_id)
                .cloned()
                .unwrap_or_default())
        }

        async fn participation_scope(
            &self,
            _persona_id: String,
            _cap: usize,
        ) -> Result<ScopeResolution, PersonaError> {
            Err(PersonaError::Authority(
                "the catalog path never resolves a search scope".to_owned(),
            ))
        }

        async fn usage_rollup_index(&self) -> Result<Vec<String>, PersonaError> {
            Ok(Vec::new())
        }

        async fn reference_snapshot(
            &self,
            _persona_id: String,
        ) -> Result<Option<PersonaReferenceSnapshot>, PersonaError> {
            Ok(None)
        }
    }

    /// Bearer verification, faked: only the two canned tokens resolve.
    struct FakeSessions;

    #[async_trait::async_trait]
    impl SessionVerification for FakeSessions {
        async fn verify_bearer(
            &self,
            token: &str,
            _now_ms: u64,
        ) -> Result<AuthorizedSessionClaims, SessionAuthorityError> {
            let (session_id, persona_id) = match token {
                PERSONA_BEARER => ("session-persona-a", PERSONA),
                ADMIN_BEARER => ("session-persona-admin", ADMIN),
                _ => return Err(SessionAuthorityError::Invalid),
            };
            Ok(AuthorizedSessionClaims {
                issuer: "session-test".to_owned(),
                key_id: "key-test".to_owned(),
                session_id: session_id.to_owned(),
                authorization_epoch: 1,
                subject: SessionSubject::Persona {
                    persona_id: persona_id.to_owned(),
                },
                scopes: vec![SessionScope::ExplorerRead],
                issued_ms: 1,
                expires_ms: EXPIRES,
            })
        }
    }

    /// A transport that must never be called: the catalog path touches no
    /// State client.
    #[derive(Clone)]
    struct NeverTransport;

    struct NeverBody;

    impl connectrpc::http_body::Body for NeverBody {
        type Data = Bytes;
        type Error = std::io::Error;

        fn poll_frame(
            self: std::pin::Pin<&mut Self>,
            _context: &mut Context<'_>,
        ) -> Poll<Option<Result<connectrpc::http_body::Frame<Bytes>, Self::Error>>> {
            Poll::Ready(None)
        }
    }

    impl ClientTransport for NeverTransport {
        type ResponseBody = NeverBody;
        type Error = std::io::Error;

        fn send(
            &self,
            _request: http::Request<ClientBody>,
        ) -> BoxFuture<'static, Result<http::Response<Self::ResponseBody>, Self::Error>> {
            Box::pin(async {
                Err(std::io::Error::other(
                    "the catalog path makes no state call",
                ))
            })
        }
    }

    /// No manifest is ever classified on the catalog path.
    #[derive(Debug)]
    struct NoTrust;

    impl ManifestTrust for NoTrust {
        fn classify(
            &self,
            _signer: &[u8],
            _canonical: &[u8],
            _signature: &[u8],
        ) -> ManifestVerdict {
            ManifestVerdict::Untrusted
        }
    }

    fn personas() -> FakePersonas {
        FakePersonas {
            personas: BTreeMap::from([
                (
                    PERSONA.to_owned(),
                    ActivePersona {
                        persona_id: PERSONA.to_owned(),
                        admin: false,
                        provisional: false,
                    },
                ),
                (
                    ADMIN.to_owned(),
                    ActivePersona {
                        persona_id: ADMIN.to_owned(),
                        admin: true,
                        provisional: false,
                    },
                ),
            ]),
            participations: BTreeMap::from([(
                PERSONA.to_owned(),
                vec![Participation {
                    conversation_id: CONVERSATION.to_owned(),
                    role: "participant".to_owned(),
                    first_at_ms: 1,
                    via_persona_id: String::new(),
                    __buffa_unknown_fields: buffa::UnknownFields::new(),
                }],
            )]),
        }
    }

    fn dead_reader() -> GcsReadClient {
        GcsReadClient::new(GcsConfig {
            bucket: "catalog-tests".to_owned(),
            prefix: "unused".to_owned(),
            credentials: GcsCredentials::Anonymous,
            api_base: "http://127.0.0.1:1".to_owned(),
            http_timeout: Duration::from_secs(1),
        })
        .expect("an anonymous read client composes")
    }

    /// Composes the real service over the fakes — `concurrency` and
    /// `timeout` are the permit's capacity and the operation deadline.
    fn service(realm_fleet: bool, concurrency: usize, timeout: Duration) -> ProjectedCoreService {
        service_with_personas(personas(), realm_fleet, concurrency, timeout)
    }

    /// [`service`], with the caller's persona fixture — a test that varies
    /// the participation count varies nothing else.
    fn service_with_personas(
        source: FakePersonas,
        realm_fleet: bool,
        concurrency: usize,
        timeout: Duration,
    ) -> ProjectedCoreService {
        let config = || ClientConfig::new("http://127.0.0.1:1".parse().unwrap());
        let composition = ProjectedCoreComposition {
            namespace: "conversation-core".to_owned(),
            persona_memory_namespace: "polychrome".to_owned(),
            projection_owner: "projector".to_owned(),
            policy: QueryServicePolicy {
                result_release_bytes: 1024 * 1024,
                response_frame_bytes: 256 * 1024,
                manifest_bytes: 256 * 1024,
                artifact_file_bytes: 64 * 1024 * 1024,
                artifact_range_bytes: 16 * 1024 * 1024,
                source_decode_bytes: 512 * 1024 * 1024,
                max_concurrent_executions: concurrency,
                revalidation_interval: Duration::from_secs(30),
                execution_memory_bytes: 8 * 1024 * 1024,
            },
            limits: {
                let defaults = QueryLimits::default();
                QueryLimits {
                    timeout,
                    row_cap: defaults.row_cap,
                    spill_dir: std::env::temp_dir().join(format!(
                        "polyc-query-catalog-test-{}-{realm_fleet}-{concurrency}",
                        std::process::id()
                    )),
                    spill_quota_bytes: 1024 * 1024,
                }
            },
            realm: if realm_fleet {
                QueryServiceRealm::Fleet {
                    visible_artifacts: dead_reader(),
                    visible_namespaces: vec![ArtifactNamespace {
                        namespace: "conversation-core-visible".to_owned(),
                        protection: AtRestProtection::TenantKey,
                    }],
                    additional_visible_artifacts: Vec::new(),
                    fleet_artifacts: dead_reader(),
                    fleet_namespaces: vec![ArtifactNamespace {
                        namespace: "conversation-core-fleet".to_owned(),
                        protection: AtRestProtection::TenantKey,
                    }],
                    additional_fleet_artifacts: Vec::new(),
                }
            } else {
                QueryServiceRealm::Visible {
                    artifacts: dead_reader(),
                    namespaces: vec![ArtifactNamespace {
                        namespace: "conversation-core-visible".to_owned(),
                        protection: AtRestProtection::TenantKey,
                    }],
                    additional_artifacts: Vec::new(),
                    fleet_namespaces: vec!["conversation-core-fleet".to_owned()],
                }
            },
            trust: Arc::new(NoTrust),
            journal: JournalClient::new(NeverTransport, config()),
            projections: ProjectionCatalogClient::new(NeverTransport, config()),
            audit: QueryAuditClient::new(NeverTransport, config()),
            versioned: VersionedSourceClient::new(NeverTransport, config()),
            persona_memory: PersonaMemoryJournalClient::new(NeverTransport, config()),
            observation: ObservationClient::new(NeverTransport, config()),
        };
        ProjectedCoreService::try_new(
            CredentialSource {
                persona: Arc::new(source),
                turn_read_trust: RoleTrustSet::current(&signer()),
                sessions: Arc::new(FakeSessions),
            },
            composition,
        )
        .expect("the test service composes")
    }

    fn bearer(token: &str) -> QueryCredential {
        QueryCredential::from_bearer(token.to_owned())
    }

    fn conversation_grant(subject: GrantSubject) -> String {
        polyc_query_model::mint_conversation_grant(&signer(), CONVERSATION, subject, EXPIRES)
    }

    fn names(reply: &CatalogReply) -> Vec<&str> {
        reply.tables().iter().map(|table| table.name()).collect()
    }

    /// A Visible-realm service answers a persona session and the two
    /// conversation-grant kinds — and names only tables a Visible scope may
    /// query.
    #[tokio::test]
    async fn the_internal_listener_describes_the_visible_catalog() {
        let service = service(false, 2, Duration::from_secs(30));
        let reply = service
            .describe_catalog(bearer(PERSONA_BEARER))
            .await
            .expect("a persona session describes its catalog");
        let names = names(&reply);
        assert!(names.contains(&"turns"), "a persona sees turns: {names:?}");
        assert!(names.windows(2).all(|pair| pair[0] < pair[1]));

        for credential in [
            conversation_grant(GrantSubject::WebSession("session-1".to_owned())),
            conversation_grant(GrantSubject::Persona(PERSONA.to_owned())),
        ] {
            service
                .describe_catalog(bearer(&credential))
                .await
                .expect("a conversation grant describes its catalog");
        }
    }

    /// Proof (2): no Fleet-only table name ever reaches a Visible caller.
    /// The mutation this pins is "list every registry table".
    #[tokio::test]
    async fn a_visible_caller_never_sees_a_fleet_only_table() {
        let service = service(false, 2, Duration::from_secs(30));
        let reply = service
            .describe_catalog(bearer(PERSONA_BEARER))
            .await
            .unwrap();
        // The scope a persona session resolves to: its participated
        // conversations, and itself as memory owner.
        let scope = QueryScope::Conversations {
            conversations: vec![CONVERSATION.to_owned()],
            memory: polyc_query_credential::session::MemorySources {
                owner: Some(PERSONA.to_owned()),
                participants: Vec::new(),
            },
        };
        let listed = names(&reply);
        for table in all_tables() {
            // The independent half: a table whose declared realm shape is
            // Fleet-only must never be named to a Visible caller, whatever
            // the listing rule says.
            if !table.visible_in(CoreRealm::Visible) {
                assert!(
                    !listed.contains(&table.name()),
                    "a Fleet-only table name reached a Visible caller: `{}`",
                    table.name()
                );
            }
            // The agreement half: the reply is exactly the listing rule's
            // answer under this scope.
            assert_eq!(
                listed.contains(&table.name()),
                catalog_tables(CoreRealm::Visible, &scope).contains(&table),
                "the reply must be exactly the listing rule's answer for `{}`",
                table.name()
            );
        }
    }

    /// Proof (3): every statement-bound or wrong-realm principal kind is
    /// refused on EITHER listener; a raw session bearer is refused on the
    /// edge listener even though the internal one admits it.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn refused_principal_kinds_and_the_edge_session_bearer() {
        let service = service(false, 2, Duration::from_secs(30));

        // A session bearer verifies on the internal listener and is refused
        // on the edge one — `admit_grant_only`, not the kind allowlist.
        service
            .describe_catalog(bearer(PERSONA_BEARER))
            .await
            .expect("the internal listener admits a session bearer");
        assert!(matches!(
            service.describe_catalog_edge(bearer(PERSONA_BEARER)).await,
            Err(QueryServiceError::Unauthorized)
        ));

        let refused = [
            conversation_grant(GrantSubject::Turn("turn-1".to_owned())),
            conversation_grant(GrantSubject::CompositeTrace {
                persona_id: PERSONA.to_owned(),
                trace_statement_digest: polyc_crypto::hex::lower(&[1; 32]),
                memory_statement_digest: polyc_crypto::hex::lower(&[2; 32]),
                routine_statement_digest: polyc_crypto::hex::lower(&[3; 32]),
            }),
            polyc_query_model::mint_control_fleet_grant(
                &signer(),
                "dashboard.conversations",
                &polyc_crypto::hex::lower(&[4; 32]),
                EXPIRES,
            ),
            // An admin-fleet grant on the Visible realm — wrong realm.
            polyc_query_model::mint_admin_fleet_grant(&signer(), ADMIN, "session-a", EXPIRES),
            // A raw admin session is not a self-service catalog caller.
            ADMIN_BEARER.to_owned(),
        ];
        for credential in &refused {
            assert!(
                matches!(
                    service.describe_catalog(bearer(credential)).await,
                    Err(QueryServiceError::Unauthorized)
                ),
                "the internal listener refuses it"
            );
            assert!(
                matches!(
                    service.describe_catalog_edge(bearer(credential)).await,
                    Err(QueryServiceError::Unauthorized)
                ),
                "the edge listener refuses it"
            );
        }
    }

    /// The Fleet realm serves an admin-fleet grant its full catalog and
    /// refuses every Visible-realm kind — including a persona session the
    /// internal listener would otherwise admit.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_fleet_realm_describes_its_catalog_to_admin_fleet_only() {
        let service = service(true, 2, Duration::from_secs(30));
        let admin_fleet =
            polyc_query_model::mint_admin_fleet_grant(&signer(), ADMIN, "session-a", EXPIRES);
        let reply = service
            .describe_catalog(bearer(&admin_fleet))
            .await
            .expect("an admin-fleet grant describes the fleet catalog");
        // Fleet sees every SQL-served table, including Fleet-only ones.
        assert_eq!(
            names(&reply).len(),
            catalog_tables(CoreRealm::Fleet, &QueryScope::Fleet).len()
        );
        assert!(
            all_tables().any(|table| !table.visible_in(CoreRealm::Visible)
                && names(&reply).contains(&table.name())),
            "a fleet-only table name is present"
        );
        for credential in [
            PERSONA_BEARER.to_owned(),
            ADMIN_BEARER.to_owned(),
            conversation_grant(GrantSubject::WebSession("session-1".to_owned())),
            conversation_grant(GrantSubject::Turn("turn-1".to_owned())),
            polyc_query_model::mint_control_fleet_grant(
                &signer(),
                "dashboard.conversations",
                &polyc_crypto::hex::lower(&[4; 32]),
                EXPIRES,
            ),
        ] {
            assert!(
                matches!(
                    service.describe_catalog(bearer(&credential)).await,
                    Err(QueryServiceError::Unauthorized)
                ),
                "the Fleet realm refuses a non-admin-fleet caller"
            );
        }
        // And the edge listener refuses the session bearer outright.
        assert!(matches!(
            service.describe_catalog_edge(bearer(PERSONA_BEARER)).await,
            Err(QueryServiceError::Unauthorized)
        ));
    }

    /// Proof (4): `DescribeCatalog` holds the same execution permit
    /// `Execute` does — the semaphore this test saturates is the one field
    /// `start_projected_admitted` hands to planning.
    #[tokio::test]
    async fn a_saturated_permit_refuses_the_catalog_call() {
        let service = service(false, 1, Duration::from_millis(80));
        let operation = CoreOperationContext::for_test(Duration::from_secs(30));
        let held = service
            .admission
            .acquire(&operation, &CancellationToken::new())
            .await
            .expect("the one slot is free");
        let outcome = service.describe_catalog(bearer(PERSONA_BEARER)).await;
        assert!(
            matches!(outcome, Err(QueryServiceError::Execution(_))),
            "a saturated admission refuses the catalog: {outcome:?}"
        );
        drop(held);
        service
            .describe_catalog(bearer(PERSONA_BEARER))
            .await
            .expect("the released slot admits the catalog again");
    }

    /// The reply's schema is the PUBLIC schema — the Fleet-only summary
    /// view exposes `turn_id`, the name `SELECT` resolves, never the
    /// declared `summary_id`.
    ///
    /// The table is found by the rename itself, never by its name: the
    /// execution-cutover guard owns the table-name literal outside the
    /// projected modules.
    #[tokio::test]
    async fn the_reply_reports_public_schema_names() {
        let service = service(true, 2, Duration::from_secs(30));
        let admin_fleet =
            polyc_query_model::mint_admin_fleet_grant(&signer(), ADMIN, "session-a", EXPIRES);
        let reply = service
            .describe_catalog(bearer(&admin_fleet))
            .await
            .unwrap();

        // The registry's renaming table: declared `summary_id`, public
        // `turn_id`. Found by the rename, not by the table's name.
        let renamed = all_tables()
            .find(|table| {
                let Some(declared) = table
                    .family()
                    .tables()
                    .iter()
                    .find(|schema| schema.table().as_str() == table.name())
                else {
                    return false;
                };
                let declared_names: Vec<&str> =
                    declared.fields().iter().map(|field| field.name()).collect();
                let public_schema = table.public_schema();
                let public_names: Vec<&str> = public_schema
                    .fields()
                    .iter()
                    .map(|field| field.name().as_str())
                    .collect();
                declared_names.contains(&"summary_id") && public_names.contains(&"turn_id")
            })
            .expect("the registry renames at least one table's public schema");

        let listed = reply
            .tables()
            .iter()
            .find(|table| table.name() == renamed.name())
            .expect("the reply lists the renaming table");
        let columns: Vec<&str> = listed
            .columns()
            .iter()
            .map(|column| column.name())
            .collect();
        let public_schema = renamed.public_schema();
        let public_names: Vec<&str> = public_schema
            .fields()
            .iter()
            .map(|field| field.name().as_str())
            .collect();
        assert_eq!(columns, public_names, "the reply speaks the view's names");
        assert!(
            columns.contains(&"turn_id"),
            "the view's renamed column is present: {columns:?}"
        );
        assert!(
            !columns.contains(&"summary_id"),
            "the declared column name never leaks: {columns:?}"
        );
    }

    /// The two fixed statements `crates/control-plane/src/payments_read.rs`
    /// runs for `GetMyPayments`, hand-copied exactly as
    /// `tests/payments_read_sql.rs` does.
    const GET_MY_PAYMENTS_SQL: &str = "SELECT partition, position, turn_id, direction, reference, \
amount_base_units, asset, recipient, method, tool_call_id, payer_kind, timestamp_unix \
FROM payments WHERE subject = $1 \
UNION ALL \
SELECT partition, position, turn_id, direction, reference, amount_base_units, asset, \
recipient, method, tool_call_id, payer_kind, timestamp_unix FROM outbound_payments \
WHERE subject = $1 \
ORDER BY timestamp_unix DESC, partition DESC, position DESC LIMIT 500";

    const GET_MY_REFUSALS_SQL: &str = "SELECT partition, position, turn_id, reason, reason_detail, \
merchant_host, requested_base_units, permitted_base_units, tool_call_id, timestamp_unix \
FROM refusals WHERE subject = $1 \
ORDER BY timestamp_unix DESC, partition DESC, position DESC LIMIT 500";

    /// `personas()`, with `PERSONA` taking part in `conversations`
    /// conversations — the participation set a persona-wide grant resolves to.
    fn personas_in(conversations: usize) -> FakePersonas {
        let mut source = personas();
        source.participations.insert(
            PERSONA.to_owned(),
            (0..conversations)
                .map(|index| Participation {
                    conversation_id: format!("c-{index}"),
                    role: "participant".to_owned(),
                    first_at_ms: 1,
                    via_persona_id: String::new(),
                    __buffa_unknown_fields: buffa::UnknownFields::new(),
                })
                .collect(),
        );
        source
    }

    /// One `GetMyPayments`-shaped request, with the caller persona bound to
    /// `$1` exactly as `payments_read.rs` binds it.
    fn my_statement_request(sql: &str) -> QueryRequest {
        QueryRequest::try_new(
            sql.to_owned(),
            vec![polyc_query_model::Parameter::Utf8(PERSONA.to_owned())],
            polyc_query_model::Consistency::Projected,
            polyc_query_model::RequestedBounds::try_new(
                Duration::from_secs(30),
                500,
                1024 * 1024,
                256 * 1024,
            )
            .unwrap(),
        )
        .unwrap()
    }

    /// POLY-479, service-level: a persona-wide grant resolves to one source
    /// partition per participation, and past `MAX_SOURCE_PINS` the plan
    /// refuses before it ever resolves a source — the transport here fails
    /// on ANY State call, so reaching the bound is the only way this call
    /// returns at all. The bound is the load-bearing contract Control's
    /// one-conversation fan-out routes around; it never moves.
    #[tokio::test]
    async fn a_persona_wide_grant_past_the_source_pin_bound_is_refused_bounds() {
        let over = usize::try_from(polyc_state::query_audit::MAX_SOURCE_PINS).unwrap() + 1;
        for (label, sql) in [
            ("payments", GET_MY_PAYMENTS_SQL),
            ("refusals", GET_MY_REFUSALS_SQL),
        ] {
            let service =
                service_with_personas(personas_in(over), false, 2, Duration::from_secs(30));
            let grant = polyc_query_model::mint_persona_wide_grant(&signer(), PERSONA, EXPIRES);
            let error = service
                .start_projected(
                    bearer(&grant),
                    &format!("q-479-{label}"),
                    &my_statement_request(sql),
                )
                .await
                .expect_err("a persona-wide statement past the source-pin bound must refuse");
            assert!(
                matches!(error, QueryServiceError::Resolution(ErrorClass::Bounds)),
                "{label} refused for the wrong reason: {error:?}"
            );
        }
    }

    /// At exactly the bound the same statements pass the pin check and reach
    /// source resolution — where the deliberately dead transport refuses.
    /// Any class but `Bounds` is the proof; nothing here has sources to serve.
    #[tokio::test]
    async fn a_persona_wide_grant_at_the_source_pin_bound_passes_the_bound() {
        let at = usize::try_from(polyc_state::query_audit::MAX_SOURCE_PINS).unwrap();
        for (label, sql) in [
            ("payments", GET_MY_PAYMENTS_SQL),
            ("refusals", GET_MY_REFUSALS_SQL),
        ] {
            let service = service_with_personas(personas_in(at), false, 2, Duration::from_secs(30));
            let grant = polyc_query_model::mint_persona_wide_grant(&signer(), PERSONA, EXPIRES);
            let error = service
                .start_projected(
                    bearer(&grant),
                    &format!("q-479-at-{label}"),
                    &my_statement_request(sql),
                )
                .await
                .expect_err("the dead transport must refuse once the bound passes");
            assert!(
                matches!(error, QueryServiceError::Resolution(class) if class != ErrorClass::Bounds),
                "{label} must pass the bound and fail in source resolution: {error:?}"
            );
        }
    }
}

/// The pinned-option, refusal, and JSON-registration proofs for
/// [`projected_session_state`] (POLY-374).
///
/// Each assertion runs twice: once on the composed state and once on the
/// per-request state `SessionStateBuilder::new_from_existing` produces —
/// the second proves no pin or registration is lost where planning and
/// execution actually inherit their session.
#[cfg(test)]
mod session_tests {
    use super::*;
    use datafusion::execution::FunctionRegistry;
    use datafusion::execution::context::SessionContext;

    /// Builds the composed state the way `compose` does, minus the spill
    /// directory the production `RuntimeEnv` carries.
    fn composed() -> SessionState {
        let runtime = RuntimeEnvBuilder::new()
            .build_arc()
            .expect("a bare runtime env builds");
        projected_session_state(runtime).expect("the projected session composes")
    }

    /// The per-request session shape `core_execution::request_context`
    /// and `core_resolution::compile` both derive from the composed state.
    fn per_request(state: &SessionState) -> SessionState {
        SessionStateBuilder::new_from_existing(state.clone()).build()
    }

    /// Every option the service relies on is pinned, on the composed
    /// session and on the per-request one alike. A `DataFusion` release
    /// that flips one of these defaults fails this test rather than
    /// silently changing the surface (POLY-374).
    #[test]
    fn the_composed_session_carries_every_pinned_option() {
        let state = composed();
        let request_state = per_request(&state);
        for options in [state.config_options(), request_state.config_options()] {
            assert!(
                !options.execution.enable_recursive_ctes,
                "recursive CTE planning is off"
            );
            assert!(!options.catalog.information_schema);
            assert_eq!(options.execution.target_partitions, 1);
            assert_eq!(options.execution.planning_concurrency, 1);
            assert_eq!(options.execution.batch_size.get(), 8192);
            assert!(options.execution.collect_statistics);
            assert!(!options.execution.parquet.pushdown_filters);
            assert!(!options.execution.parquet.reorder_filters);
            assert!(options.execution.parquet.enable_page_index);
            assert!(options.execution.parquet.bloom_filter_on_read);
            assert_eq!(options.catalog.default_catalog, "datafusion");
            assert_eq!(options.catalog.default_schema, "public");
            assert_eq!(options.sql_parser.dialect, Dialect::Generic);
            assert!(options.sql_parser.enable_ident_normalization);
            assert!(!options.sql_parser.parse_float_as_decimal);
        }
        // `SessionStateBuilder::new_from_existing` recomputes this flag to
        // false when the default catalog already exists, which the composed
        // state's does. Both values are the pinned contract, not a drift.
        assert!(
            state
                .config_options()
                .catalog
                .create_default_catalog_and_schema
        );
        assert!(
            !request_state
                .config_options()
                .catalog
                .create_default_catalog_and_schema
        );
    }

    /// The planner refuses `WITH RECURSIVE` on its own — the gate's AST
    /// check is bypassed here deliberately, so this proves the pinned
    /// option is a working second line, not a dead one.
    #[tokio::test]
    async fn the_planner_alone_refuses_a_recursive_statement() {
        let context = SessionContext::new_with_state(composed());
        let refusal = context
            .sql(
                "WITH RECURSIVE n AS (SELECT 1 AS v UNION ALL SELECT v + 1 FROM n) \
                 SELECT v FROM n",
            )
            .await;
        assert!(refusal.is_err(), "planning refuses recursion: {refusal:?}");
    }

    /// URL tables are never enabled on this surface — a string-literal
    /// table reference has no catalog to resolve against.
    #[tokio::test]
    async fn a_url_table_reference_fails_to_plan() {
        let context = SessionContext::new_with_state(composed());
        let refusal = context.sql("SELECT * FROM 'file.csv'").await;
        assert!(refusal.is_err(), "URL tables stay off: {refusal:?}");
    }

    /// The per-request session carries all three JSON registration kinds:
    /// the scalar functions, the function rewrite, and the expression
    /// planner. Each line asserts on the `new_from_existing` session —
    /// the one `core_execution` and `core_resolution` actually plan
    /// against — not only the composed one. Skipping `register_all` in
    /// the composer turns every assertion red.
    #[tokio::test]
    async fn the_per_request_session_carries_the_json_surface() {
        let state = per_request(&composed());
        let context = SessionContext::new_with_state(state);

        // The scalar function resolves.
        context
            .sql("SELECT json_get_str('{\"q\":\"a\"}', 'q')")
            .await
            .expect("json_get_str is registered on the per-request session");

        // The expression planner maps `->` to `json_get`.
        let operator_plan = context
            .sql("SELECT '{\"q\":\"a\"}' -> 'q'")
            .await
            .expect("`->` reaches the JSON expression planner")
            .logical_plan()
            .display_indent()
            .to_string();
        assert!(
            operator_plan.contains("json_get"),
            "`->` plans as json_get: {operator_plan}"
        );

        // The function rewrite fires inside `create_physical_expr`, not
        // during logical planning, so the proof is execution: it collapses
        // `CAST(json_get(..) AS int)` to `json_get_int`. Without it the
        // physical planner must cast the sparse JSON union to `Int32`,
        // which Arrow refuses — the query fails instead of returning 1.
        let batches = context
            .sql("SELECT CAST(json_get('{\"q\":1}', 'q') AS int)")
            .await
            .expect("the cast-collapse query plans")
            .collect()
            .await
            .expect("the JSON rewrite made the union cast executable");
        let value = batches[0]
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Int64Array>()
            .expect("the collapsed getter returns Int64")
            .value(0);
        assert_eq!(value, 1);

        // Under the pinned Generic dialect `?` tokenizes as a parameter
        // placeholder, never as `BinaryOperator::Question` — the contains
        // operator is function-form only on this surface.
        let question = context.sql("SELECT '{\"q\":\"a\"}' ? 'q'").await;
        assert!(
            question.is_err(),
            "`?` does not plan under the pinned dialect: {question:?}"
        );
    }

    /// The composed session's function registries hold exactly the closed
    /// surface (POLY-371): every registered name — canonical or alias —
    /// is in [`crate::function_surface::allowed_names`], every allowed
    /// name is registered, and the table-function map is empty. The same
    /// assertions run on the per-request session, proving
    /// `new_from_existing` widens nothing.
    #[test]
    fn the_composed_session_registers_only_the_allowed_functions() {
        let allowed = crate::function_surface::allowed_names();
        for state in [&composed(), &per_request(&composed())] {
            for name in state.scalar_functions().keys() {
                assert!(
                    allowed.contains(name),
                    "scalar registry holds unlisted name {name:?}"
                );
            }
            for name in state
                .aggregate_functions()
                .keys()
                .chain(state.window_functions().keys())
            {
                assert!(
                    allowed.contains(name),
                    "aggregate/window registry holds unlisted name {name:?}"
                );
            }
            assert!(
                state.table_functions().is_empty(),
                "no table function is registered"
            );
            // Every name the gate admits is one the session can plan.
            for name in allowed {
                assert!(
                    state.scalar_functions().contains_key(name)
                        || state.aggregate_functions().contains_key(name)
                        || state.window_functions().contains_key(name),
                    "allowed name {name:?} is not registered"
                );
            }
        }
    }

    /// A refused function fails planning on the per-request session —
    /// canonical name, alias, and case variation alike. These assertions
    /// bypass the gate deliberately (`context.sql` plans directly), so a
    /// registry that still carried `repeat` fails here, not just in the
    /// gate tests.
    #[tokio::test]
    async fn a_refused_function_fails_planning_on_the_per_request_session() {
        let context = SessionContext::new_with_state(per_request(&composed()));
        for sql in [
            "SELECT repeat('x', 2)",
            "SELECT REPEAT('x', 2)",
            "SELECT array_repeat(1, 2)",
            "SELECT list_repeat(1, 2)",
            "SELECT approx_percentile_cont(x, 0.5, 1024) FROM t",
            "SELECT generate_series(1, 10)",
        ] {
            let refusal = context.sql(sql).await;
            assert!(refusal.is_err(), "{sql} must fail planning: {refusal:?}");
        }
    }

    /// Mutation run: registering one extra function on the composed state
    /// breaks the equality the previous test asserts — the per-request
    /// session inherits the drift, which is exactly what the proof
    /// watches for.
    #[test]
    fn an_extra_registration_widens_the_per_request_session() {
        let mut state = composed();
        let before = state.scalar_functions().len();
        let extra = SessionStateDefaults::default_scalar_functions()
            .into_iter()
            .find(|function| function.name() == "repeat")
            .expect("repeat is a default function upstream");
        let _ = state.register_udf(extra);
        assert_eq!(state.scalar_functions().len(), before + 1);
        let request = per_request(&state);
        assert!(
            request.scalar_functions().contains_key("repeat"),
            "the drift propagates: the per-request session sees it"
        );
    }

    /// The seven core expression planners stay installed — the SQL
    /// desugars (`EXTRACT`, `POSITION`, `SUBSTRING`, `TRIM`, array
    /// literals, `IN`, `LIKE`) still plan — and the JSON planner joins
    /// them via `register_all`.
    #[tokio::test]
    async fn the_desugared_sql_forms_still_plan() {
        let context = SessionContext::new_with_state(per_request(&composed()));
        for sql in [
            "SELECT EXTRACT(YEAR FROM CAST('2026-01-01' AS DATE))",
            "SELECT POSITION('b' IN 'abc')",
            "SELECT SUBSTRING('abc' FROM 2)",
            "SELECT TRIM(BOTH ' ' FROM ' a ')",
            "SELECT make_array(1, 2)[1]",
            "SELECT * FROM (VALUES (1)) v(x) WHERE x IN (1, 2)",
            "SELECT 'abc' LIKE 'a%'",
            "SELECT COALESCE(NULL, 1)",
            "SELECT NULLIF(1, 2)",
        ] {
            let planned = context.sql(sql).await;
            assert!(planned.is_ok(), "{sql} must plan: {planned:?}");
        }
    }
}