polyc-query 2026.9.0

//! Executes exact `conversation-core/v1` files behind a durable permit.
//!
//! This private seam is opt-in. Production constructors do not compose it.
//! Production composition must supply current credential re-verification.
//! This module durably settles a terminal audit before exposing stream EOF.
//! Service framing remains outside this private seam.
//! This seam admits signed manifests through the shared artifact contract.
//! It proves each immutable file before provider construction.
//! It binds the audited logical plan into a fresh catalog.
//! `DataFusion` can then read only the verified generations.
//!
//! Sibling modules isolate artifact admission, provider construction, and bounded release.
//! A real artifact retirement lease also blocks activation.
//! The projection read lease does not protect these objects from artifact garbage collection.
//! Signed physical envelopes now bind the uncompressed profile and a conservative
//! source-decode bound. Query recomputes them from retained exact bytes and
//! charges the complete request sum to the shared `RuntimeEnv` memory pool
//! before reading a segment body. This does not claim that the pool accounts
//! for allocator or operator memory outside that explicit envelope.

#![allow(
    dead_code,
    reason = "the admission handle and the stream's settled-completion view are held for the cases that assert them; production reads the terminal frame instead"
)]

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use datafusion::catalog::{
    CatalogProvider, CatalogProviderList, MemoryCatalogProvider, MemoryCatalogProviderList,
    MemorySchemaProvider, TableProvider,
};
use datafusion::common::tree_node::{Transformed, TreeNode};
use datafusion::datasource::provider_as_source;
use datafusion::error::DataFusionError;
use datafusion::execution::context::{SessionContext, SessionState};
use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
use datafusion::execution::session_state::SessionStateBuilder;
use datafusion::logical_expr::{LogicalPlan, TableScan};
use parquet::errors::ParquetError;
use polyc_projection::family::{CONVERSATION_MESSAGES, CONVERSATION_TURNS, conversation_core};
use polyc_projection_artifact::parquet_profile::source_decode_reservation_bytes;
use polyc_projection_artifact::{
    ArtifactReadError, ArtifactRealm, FileVerificationBudget, FleetArtifactAccess,
    ManifestOpenError, ProjectionFile, ReadContractError, RealmTopology,
    VerifiedProjectionManifest, VisibleArtifactAccess,
};
use polyc_state::error::StateError;
use polyc_state::id::PartitionId;
use polyc_state::projection::artifact::ManifestTrust;
use polyc_state::revision::JournalSource;
use tokio_util::sync::CancellationToken;

use crate::core_resolution::{
    CompiledCoreParts, CoreMetadataAuthority, CoreOperationContext, CoreRealm, CoreResolutionError,
    CoreTable, EffectiveCoreBounds, PreparedCoreParts, PreparedCoreQuery, arrow_schema,
};
use crate::session::QueryScope;
use crate::statement_gate::AllowedStatement;

pub(crate) mod admission;
mod guardian;
mod latch;
mod provider;
mod stream;
#[cfg(test)]
mod tests;

struct ProjectedCompiledParts {
    normalized_plan: String,
    dependencies: Vec<CoreTable>,
    statement: AllowedStatement,
    explain_enabled: bool,
    plan: LogicalPlan,
    base_state: SessionState,
}

fn projected_parts(
    compiled: CompiledCoreParts,
) -> Result<ProjectedCompiledParts, CoreExecutionError> {
    let CompiledCoreParts {
        normalized_plan,
        dependencies,
        legacy_dependencies,
        statement,
        explain_enabled,
        plan,
        base_state,
    } = compiled;
    if !legacy_dependencies.is_empty() {
        return Err(CoreExecutionError::LegacyProviderUnavailable);
    }
    Ok(ProjectedCompiledParts {
        normalized_plan,
        dependencies,
        statement,
        explain_enabled,
        plan,
        base_state,
    })
}

pub(crate) use admission::{CoreExecutionAdmission, CoreExecutionAdmissionInput};
#[cfg(test)]
pub(crate) use guardian::GuardianDispatchPause;
pub(crate) use guardian::PermitGuardian;
pub(crate) use guardian::{classify_error, classify_resolution};
use provider::{ExactParquetTable, VerifiedCoreFile, validate_parquet_file};
pub(crate) use stream::{BoundCoreQuery, CoreResultStream};

/// Defines the current scope authority required before Query releases rows.
///
/// Production composition must implement this from the presented credential.
/// It must also consult current persona, session, and grant authority.
/// A requester identifier or cached principal is not sufficient.
/// This slice cannot compose that credential-bearing adapter.
/// Thus, production activation remains impossible.
/// Source-incarnation checks are complete at this seam.
#[async_trait]
pub(crate) trait CoreScopeRevalidator: Send + Sync {
    async fn current_scope(
        &self,
        operation: &CoreOperationContext,
    ) -> Result<QueryScope, CoreResolutionError>;
}

pub(crate) mod sealed {
    /// Marks an adapter that re-proves a credential, not only a scope.
    pub(crate) trait CredentialProven {}
}

/// Requires a revalidator that rechecks a retained credential.
///
/// [`CoreScopeRevalidator`] alone says nothing about where a scope came from.
/// A caller that only remembers trusted identifiers satisfies it. Such a caller
/// re-proves no bearer session and no signed grant.
///
/// The marker does not verify that claim. It forces each adapter to write one
/// explicit `impl` line stating it. A reviewer can then find every adapter that
/// makes the claim with one grep.
///
/// Only the capability exists today. Two adapters are still missing. One
/// retains a redacted credential witness. One materializes bounded results for
/// embedded callers. No production constructor composes this seam until both
/// land.
pub(crate) trait CurrentCredentialAuthority:
    CoreScopeRevalidator + sealed::CredentialProven
{
}

impl<T> CurrentCredentialAuthority for T where
    T: CoreScopeRevalidator + sealed::CredentialProven + ?Sized
{
}

/// Every prepared field except the permit guardian.
///
/// `bind` splits the guardian away so ownership of the permit is visible in
/// the type of every later step.
struct AdmittedCoreParts {
    compiled: crate::core_resolution::CompiledCoreQuery,
    manifests: Vec<polyc_state::projection::ProjectionManifest>,
    partitions: Vec<PartitionId>,
    scope: QueryScope,
    realm: CoreRealm,
    metadata: Arc<dyn CoreMetadataAuthority>,
    operation: CoreOperationContext,
    bounds: EffectiveCoreBounds,
}

/// A bound physical plan that still needs its permit guardian.
///
/// Planning cannot fail after this value exists, so handing it the guardian is
/// infallible.
struct PlannedCoreQuery {
    dataframe: datafusion::dataframe::DataFrame,
    manifests: Vec<polyc_state::projection::ProjectionManifest>,
    original_scope: QueryScope,
    metadata: Arc<dyn CoreMetadataAuthority>,
    operation: Arc<CoreOperationContext>,
    bounds: EffectiveCoreBounds,
    explain_enabled: bool,
    execution_admission: tokio::sync::OwnedSemaphorePermit,
    source_decode_reservation: MemoryReservation,
    #[cfg(test)]
    providers: BTreeMap<CoreTable, Arc<dyn TableProvider>>,
}

impl PlannedCoreQuery {
    fn into_bound(
        self,
        guardian: PermitGuardian,
        authority: &CoreArtifactAuthority,
        cancellation: CancellationToken,
    ) -> BoundCoreQuery {
        let Self {
            dataframe,
            manifests,
            original_scope,
            metadata,
            operation,
            bounds,
            explain_enabled,
            execution_admission,
            source_decode_reservation,
            #[cfg(test)]
            providers,
        } = self;
        BoundCoreQuery {
            guardian,
            dataframe,
            manifests,
            original_scope,
            metadata,
            scope_revalidator: Arc::clone(&authority.scope),
            operation,
            cancellation,
            bounds,
            revalidation_interval: authority.revalidation_interval,
            _explain_enabled: explain_enabled,
            execution_admission,
            source_decode_reservation,
            #[cfg(test)]
            report_delay: Duration::ZERO,
            #[cfg(test)]
            providers,
        }
    }
}

enum CoreArtifactComposition {
    Visible {
        reader: Arc<dyn VisibleArtifactAccess>,
    },
    Fleet {
        visible: Arc<dyn VisibleArtifactAccess>,
        fleet: Arc<dyn FleetArtifactAccess>,
    },
}

impl CoreArtifactComposition {
    fn visible(&self) -> &Arc<dyn VisibleArtifactAccess> {
        match self {
            Self::Visible { reader } => reader,
            Self::Fleet { visible, .. } => visible,
        }
    }

    const fn realm(&self) -> CoreRealm {
        match self {
            Self::Visible { .. } => CoreRealm::Visible,
            Self::Fleet { .. } => CoreRealm::Fleet,
        }
    }
}

/// Binds an exact artifact composition to one realm.
pub(crate) struct CoreArtifactAuthority {
    composition: CoreArtifactComposition,
    trust: Arc<dyn ManifestTrust>,
    topology: RealmTopology,
    scope: Arc<dyn CurrentCredentialAuthority>,
    revalidation_interval: Duration,
    admission: Arc<CoreExecutionAdmission>,
}

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

impl CoreArtifactAuthority {
    pub(crate) fn visible(
        reader: Arc<dyn VisibleArtifactAccess>,
        trust: Arc<dyn ManifestTrust>,
        topology: RealmTopology,
        scope: Arc<dyn CurrentCredentialAuthority>,
        revalidation_interval: Duration,
        admission: Arc<CoreExecutionAdmission>,
    ) -> Result<Self, CoreExecutionError> {
        Self::new(
            CoreArtifactComposition::Visible { reader },
            trust,
            topology,
            scope,
            revalidation_interval,
            admission,
        )
    }

    pub(crate) fn fleet(
        visible: Arc<dyn VisibleArtifactAccess>,
        fleet: Arc<dyn FleetArtifactAccess>,
        trust: Arc<dyn ManifestTrust>,
        topology: RealmTopology,
        scope: Arc<dyn CurrentCredentialAuthority>,
        revalidation_interval: Duration,
        admission: Arc<CoreExecutionAdmission>,
    ) -> Result<Self, CoreExecutionError> {
        Self::new(
            CoreArtifactComposition::Fleet { visible, fleet },
            trust,
            topology,
            scope,
            revalidation_interval,
            admission,
        )
    }

    fn new(
        composition: CoreArtifactComposition,
        trust: Arc<dyn ManifestTrust>,
        topology: RealmTopology,
        scope: Arc<dyn CurrentCredentialAuthority>,
        revalidation_interval: Duration,
        admission: Arc<CoreExecutionAdmission>,
    ) -> Result<Self, CoreExecutionError> {
        if revalidation_interval.is_zero() {
            return Err(CoreExecutionError::InvalidComposition(
                "the release revalidation interval is zero",
            ));
        }
        Ok(Self {
            composition,
            trust,
            topology,
            scope,
            revalidation_interval,
            admission,
        })
    }

    /// Consumes a permit-owning prepared query and binds its exact files.
    pub(crate) async fn bind(
        &self,
        prepared: PreparedCoreQuery,
    ) -> Result<BoundCoreQuery, CoreExecutionError> {
        let PreparedCoreParts {
            guardian,
            compiled,
            manifests,
            partitions,
            scope,
            realm,
            metadata,
            operation,
            bounds,
        } = prepared.into_parts();
        let cancellation = guardian.cancellation();
        let admitted = AdmittedCoreParts {
            compiled,
            manifests,
            partitions,
            scope,
            realm,
            metadata,
            operation,
            bounds,
        };
        match self.admit_and_plan(guardian, admitted, &cancellation).await {
            Ok(bound) => Ok(bound),
            Err((guardian, error)) => {
                // The intent is already durable. A bind refusal settles its own
                // exact terminal. It does not leave an ordinary unmatched
                // intent for the reconciler.
                //
                // A settlement failure never masks the refusal. The caller
                // needs the reason its query was refused.
                let _ = guardian.fail(classify_error(&error)).await;
                Err(error)
            }
        }
    }

    /// Binds exact files, returning the guardian to the caller on refusal.
    ///
    /// The guardian moves into the returned [`BoundCoreQuery`] on success. It
    /// moves back to the caller on failure. The compiler therefore proves that
    /// exactly one path owns it.
    async fn admit_and_plan(
        &self,
        guardian: PermitGuardian,
        admitted: AdmittedCoreParts,
        cancellation: &CancellationToken,
    ) -> Result<BoundCoreQuery, (PermitGuardian, CoreExecutionError)> {
        // Planning holds every admitted part plus its verified manifests, so
        // its future is large. Box it to keep the caller's future small.
        match Box::pin(self.plan_admitted(admitted, cancellation)).await {
            Ok(planned) => Ok(planned.into_bound(guardian, self, cancellation.clone())),
            Err(error) => Err((guardian, error)),
        }
    }

    async fn plan_admitted(
        &self,
        admitted: AdmittedCoreParts,
        cancellation: &CancellationToken,
    ) -> Result<PlannedCoreQuery, CoreExecutionError> {
        let AdmittedCoreParts {
            compiled,
            manifests,
            partitions,
            scope,
            realm,
            metadata,
            operation,
            bounds,
        } = admitted;
        let ProjectedCompiledParts {
            normalized_plan,
            dependencies,
            statement,
            explain_enabled,
            plan,
            base_state,
        } = projected_parts(compiled.into_parts())?;
        if realm != self.composition.realm() {
            return Err(CoreExecutionError::RealmMismatch);
        }
        let pinned_partitions = manifests
            .iter()
            .map(|manifest| manifest.key().source().clone())
            .collect::<Vec<_>>();
        if pinned_partitions != partitions {
            return Err(CoreExecutionError::PlanIdentityMismatch);
        }
        operation.check().map_err(operation_refusal)?;
        let operation = Arc::new(operation);
        let execution_admission = self.admission.acquire(&operation, cancellation).await?;
        let admitted_manifests = self
            .admit_manifests(&manifests, &operation, cancellation, bounds)
            .await?;

        let source_decode_reservation = self.reserve_source_decode(
            &dependencies,
            &admitted_manifests,
            realm,
            bounds,
            &base_state,
        )?;
        let mut providers = BTreeMap::new();
        for dependency in &dependencies {
            providers.insert(
                *dependency,
                self.exact_provider(
                    *dependency,
                    &admitted_manifests,
                    &operation,
                    cancellation,
                    bounds,
                )
                .await?,
            );
        }

        let context = request_context(&base_state, &providers)?;
        let rebound = rebind_plan(plan, &context, &providers)?;
        if rebound.display_indent().to_string() != normalized_plan
            || dependency_closure(&rebound)? != dependencies
        {
            return Err(CoreExecutionError::PlanIdentityMismatch);
        }
        let dataframe = operation_wait(
            &operation,
            cancellation,
            context.execute_logical_plan(rebound),
        )
        .await??;
        let dataframe = match statement {
            AllowedStatement::Query => dataframe.limit(
                0,
                Some(
                    usize::try_from(bounds.rows())
                        .unwrap_or(usize::MAX)
                        .saturating_add(1),
                ),
            )?,
            AllowedStatement::Explain => dataframe,
        };

        Ok(PlannedCoreQuery {
            dataframe,
            manifests,
            original_scope: scope,
            metadata,
            operation,
            bounds,
            explain_enabled,
            execution_admission,
            source_decode_reservation,
            #[cfg(test)]
            providers,
        })
    }

    fn required_decode_bytes(
        &self,
        dependencies: &[CoreTable],
        manifests: &[(JournalSource, VerifiedProjectionManifest)],
        realm: CoreRealm,
        bounds: EffectiveCoreBounds,
    ) -> Result<u64, CoreExecutionError> {
        let mut total = bounds.artifact_range_bytes().checked_add(1).ok_or(
            CoreExecutionError::ParquetContract("verification request bound overflowed"),
        )?;
        for dependency in dependencies {
            let table = match dependency {
                CoreTable::Turns => CONVERSATION_TURNS,
                CoreTable::Messages => CONVERSATION_MESSAGES,
            };
            for (_, manifest) in manifests {
                add_source_reservations(
                    &mut total,
                    self.composition
                        .visible()
                        .files_for_visible(manifest, table)?,
                    bounds,
                )?;
                if *dependency == CoreTable::Messages
                    && realm == CoreRealm::Fleet
                    && let CoreArtifactComposition::Fleet { fleet, .. } = &self.composition
                {
                    add_source_reservations(
                        &mut total,
                        fleet.files_for_fleet(manifest, table)?,
                        bounds,
                    )?;
                }
            }
        }
        Ok(total)
    }

    fn reserve_source_decode(
        &self,
        dependencies: &[CoreTable],
        manifests: &[(JournalSource, VerifiedProjectionManifest)],
        realm: CoreRealm,
        bounds: EffectiveCoreBounds,
        base_state: &SessionState,
    ) -> Result<MemoryReservation, CoreExecutionError> {
        let decode_bytes = self.required_decode_bytes(dependencies, manifests, realm, bounds)?;
        if decode_bytes > bounds.source_decode_bytes() {
            return Err(CoreExecutionError::SourceDecodeBound {
                observed: decode_bytes,
                limit: bounds.source_decode_bytes(),
            });
        }
        let decode_capacity =
            usize::try_from(decode_bytes).map_err(|_| CoreExecutionError::SourceDecodeBound {
                observed: decode_bytes,
                limit: bounds.source_decode_bytes(),
            })?;
        let reservation = MemoryConsumer::new("projection-source-decode")
            .register(&base_state.runtime_env().memory_pool);
        reservation.try_grow(decode_capacity)?;
        Ok(reservation)
    }

    async fn admit_manifests(
        &self,
        manifests: &[polyc_state::projection::ProjectionManifest],
        operation: &Arc<CoreOperationContext>,
        cancellation: &CancellationToken,
        bounds: EffectiveCoreBounds,
    ) -> Result<Vec<(JournalSource, VerifiedProjectionManifest)>, CoreExecutionError> {
        let mut admitted_manifests = Vec::with_capacity(manifests.len());
        for manifest in manifests {
            operation.check().map_err(operation_refusal)?;
            let admitted = operation_wait(
                operation,
                cancellation,
                self.composition.visible().read_and_verify_manifest(
                    manifest,
                    self.trust.as_ref(),
                    conversation_core(),
                    &self.topology,
                    bounds.manifest_bytes(),
                ),
            )
            .await??;
            admitted_manifests.push((manifest.checkpoint().source().clone(), admitted));
        }
        Ok(admitted_manifests)
    }

    async fn exact_provider(
        &self,
        dependency: CoreTable,
        manifests: &[(JournalSource, VerifiedProjectionManifest)],
        operation: &Arc<CoreOperationContext>,
        cancellation: &CancellationToken,
        bounds: EffectiveCoreBounds,
    ) -> Result<Arc<dyn TableProvider>, CoreExecutionError> {
        let table_id = match dependency {
            CoreTable::Turns => CONVERSATION_TURNS,
            CoreTable::Messages => CONVERSATION_MESSAGES,
        };
        let table_schema =
            conversation_core()
                .table(table_id)
                .ok_or(CoreExecutionError::InvalidComposition(
                    "conversation-core omits a required table",
                ))?;
        let expected_schema = arrow_schema(table_schema);
        let verification = FileVerificationBudget::try_new(
            bounds.artifact_file_bytes(),
            bounds.artifact_range_bytes(),
        )?;
        let mut files = Vec::new();
        for (source, admitted) in manifests {
            let visible_files = self
                .composition
                .visible()
                .files_for_visible(admitted, table_id)?;
            for file in visible_files {
                operation.check().map_err(operation_refusal)?;
                let verified = operation_wait(
                    operation,
                    cancellation,
                    self.composition
                        .visible()
                        .verify_and_retain_file_exact(&file, verification),
                )
                .await??;
                files.push(VerifiedCoreFile::Visible {
                    file: verified,
                    source: source.clone(),
                });
            }
            if dependency == CoreTable::Messages
                && let CoreArtifactComposition::Fleet { fleet, .. } = &self.composition
            {
                let fleet_files = fleet.files_for_fleet(admitted, table_id)?;
                for file in fleet_files {
                    operation.check().map_err(operation_refusal)?;
                    let verified = operation_wait(
                        operation,
                        cancellation,
                        fleet.verify_and_retain_file_exact(&file, verification),
                    )
                    .await??;
                    files.push(VerifiedCoreFile::Fleet {
                        file: verified,
                        source: source.clone(),
                    });
                }
            }
        }
        files.sort_by_key(VerifiedCoreFile::identity);
        for file in &files {
            operation_wait(
                operation,
                cancellation,
                validate_parquet_file(
                    file.clone(),
                    Arc::clone(&expected_schema),
                    Arc::clone(operation),
                    cancellation.clone(),
                ),
            )
            .await??;
        }
        Ok(Arc::new(ExactParquetTable::new(
            expected_schema,
            files,
            operation,
            cancellation,
        )))
    }
}

fn add_source_reservations<R: ArtifactRealm>(
    total: &mut u64,
    files: Vec<ProjectionFile<R>>,
    bounds: EffectiveCoreBounds,
) -> Result<(), CoreExecutionError> {
    for file in files {
        if file.byte_len() > bounds.artifact_file_bytes() {
            return Err(CoreExecutionError::SourceDecodeBound {
                observed: file.byte_len(),
                limit: bounds.artifact_file_bytes(),
            });
        }
        *total = total
            .checked_add(source_decode_reservation_bytes(
                file.byte_len(),
                file.descriptor().physical(),
            )?)
            .ok_or(CoreExecutionError::ParquetContract(
                "aggregate source decode bound overflowed",
            ))?;
    }
    Ok(())
}

pub(super) async fn operation_wait<T>(
    operation: &CoreOperationContext,
    cancellation: &CancellationToken,
    future: impl Future<Output = T>,
) -> Result<T, CoreExecutionError> {
    const POLL_INTERVAL: Duration = Duration::from_millis(10);

    tokio::pin!(future);
    loop {
        if cancellation.is_cancelled() {
            return Err(CoreExecutionError::Cancelled);
        }
        let remaining = operation.remaining().map_err(operation_refusal)?;
        let wait = remaining.min(POLL_INTERVAL);
        match tokio::time::timeout(wait, &mut future).await {
            Ok(result) => {
                if cancellation.is_cancelled() {
                    return Err(CoreExecutionError::Cancelled);
                }
                operation.check().map_err(operation_refusal)?;
                return Ok(result);
            }
            Err(_) if wait == remaining => return Err(CoreExecutionError::Deadline),
            Err(_) => operation.check().map_err(operation_refusal)?,
        }
    }
}

pub(super) fn operation_refusal(error: CoreResolutionError) -> CoreExecutionError {
    match error {
        CoreResolutionError::State(StateError::DeadlineExpired { .. }) => {
            CoreExecutionError::Deadline
        }
        CoreResolutionError::State(StateError::Cancelled { .. }) => CoreExecutionError::Cancelled,
        other => CoreExecutionError::from(other),
    }
}

fn request_context(
    base_state: &SessionState,
    providers: &BTreeMap<CoreTable, Arc<dyn TableProvider>>,
) -> Result<SessionContext, CoreExecutionError> {
    let catalog_name = base_state.config_options().catalog.default_catalog.clone();
    let schema_name = base_state.config_options().catalog.default_schema.clone();
    let catalog_list = Arc::new(MemoryCatalogProviderList::new());
    let catalog = Arc::new(MemoryCatalogProvider::new());
    catalog.register_schema(&schema_name, Arc::new(MemorySchemaProvider::new()))?;
    catalog_list.register_catalog(catalog_name, catalog);
    let state = SessionStateBuilder::new_from_existing(base_state.clone())
        .with_catalog_list(catalog_list)
        .build();
    let context = SessionContext::new_with_state(state);
    for (table, provider) in providers {
        context.register_table(table.name(), Arc::clone(provider))?;
    }
    Ok(context)
}

fn rebind_plan(
    plan: LogicalPlan,
    context: &SessionContext,
    providers: &BTreeMap<CoreTable, Arc<dyn TableProvider>>,
) -> Result<LogicalPlan, CoreExecutionError> {
    let rebound = plan.transform_up(|node| {
        let LogicalPlan::TableScan(scan) = node else {
            return Ok(Transformed::no(node));
        };
        let name = scan.table_name.table();
        let table = CoreTable::from_name(name).map_err(|_| {
            DataFusionError::Plan(format!("audited plan acquired undeclared table {name}"))
        })?;
        let provider = providers.get(&table).ok_or_else(|| {
            DataFusionError::Plan(format!(
                "audited dependency {} is not registered",
                table.name()
            ))
        })?;
        // The fresh catalog supplies a consistency check.
        // The replacement uses the same provider that the catalog holds.
        if !context.table_exist(table.name())? {
            return Err(DataFusionError::Plan(format!(
                "fresh request catalog omits {}",
                table.name()
            )));
        }
        let TableScan {
            table_name,
            source: _,
            projection,
            projected_schema: _,
            filters,
            fetch,
        } = scan;
        let replacement = TableScan::try_new(
            table_name,
            provider_as_source(Arc::clone(provider)),
            projection,
            filters,
            fetch,
        )?;
        Ok(Transformed::yes(LogicalPlan::TableScan(replacement)))
    })?;
    Ok(rebound.data)
}

fn dependency_closure(plan: &LogicalPlan) -> Result<Vec<CoreTable>, CoreExecutionError> {
    let mut dependencies = BTreeSet::new();
    plan.apply(|node| {
        if let LogicalPlan::TableScan(scan) = node {
            let name = scan.table_name.table();
            let table = CoreTable::from_name(name).map_err(|_| {
                DataFusionError::Plan(format!("rebound plan acquired undeclared table {name}"))
            })?;
            dependencies.insert(table);
        }
        Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue)
    })?;
    Ok(dependencies.into_iter().collect())
}

/// Classifies fail-closed exact artifact and projected-stream refusals.
/// A typed refusal from exact projected execution.
///
/// `Display` and `Debug` both report the refusal only. A partition names a
/// real conversation, and a nested engine, Parquet, or Arrow message quotes
/// column names, predicates, and object keys. The variants keep their values
/// so a mechanism can read them; formatting never renders one.
#[derive(thiserror::Error)]
pub(crate) enum CoreExecutionError {
    #[error("the projected core composition is invalid")]
    InvalidComposition(&'static str),
    #[error("the prepared realm and artifact composition disagree")]
    RealmMismatch,
    #[error("the rebound logical plan differs from the audited plan")]
    PlanIdentityMismatch,
    #[error("the audited plan requires a legacy provider that is not bound")]
    LegacyProviderUnavailable,
    #[error("current authority no longer contains the original projected scope")]
    AuthorityNarrowed,
    #[error("the current source lineage changed for an authorized partition")]
    SourceChanged(PartitionId),
    #[error("the projected query was cancelled")]
    Cancelled,
    #[error("the projected query exhausted its original deadline")]
    Deadline,
    #[error("the aggregate projected execution admission was closed")]
    ExecutionAdmissionClosed,
    #[error("the terminal query audit is not durably settled")]
    AuditCompletionUnavailable,
    #[error("the terminal query audit answered for another command identity")]
    AuditReceiptMismatch,
    #[error("the released Arrow result used {observed} bytes past {limit}")]
    ReleaseBound { observed: u64, limit: u64 },
    #[error("the projection source decode needs {observed} bytes past {limit}")]
    SourceDecodeBound { observed: u64, limit: u64 },
    #[error("the exact Parquet contract failed")]
    ParquetContract(&'static str),
    #[error("descriptor planning refused this query")]
    Resolution(Box<CoreResolutionError>),
    #[error("the signed manifest could not be opened")]
    Manifest(#[from] ManifestOpenError),
    #[error("the realm read contract refused this query")]
    Contract(#[from] ReadContractError),
    #[error("an exact artifact read refused this query")]
    Artifact(#[from] ArtifactReadError),
    #[error("projected execution failed")]
    DataFusion(#[from] DataFusionError),
    #[error("the exact Parquet reader refused this file")]
    Parquet(#[from] ParquetError),
    #[error("the signed Parquet profile refused this file")]
    Profile(#[from] polyc_projection_artifact::parquet_profile::ProfileError),
    #[error("an Arrow operation refused this result")]
    Arrow(#[from] arrow::error::ArrowError),
}

impl fmt::Debug for CoreExecutionError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidComposition(_) => "InvalidComposition",
            Self::RealmMismatch => "RealmMismatch",
            Self::PlanIdentityMismatch => "PlanIdentityMismatch",
            Self::LegacyProviderUnavailable => "LegacyProviderUnavailable",
            Self::AuthorityNarrowed => "AuthorityNarrowed",
            Self::SourceChanged(_) => "SourceChanged",
            Self::Cancelled => "Cancelled",
            Self::Deadline => "Deadline",
            Self::ExecutionAdmissionClosed => "ExecutionAdmissionClosed",
            Self::AuditCompletionUnavailable => "AuditCompletionUnavailable",
            Self::AuditReceiptMismatch => "AuditReceiptMismatch",
            Self::ReleaseBound { .. } => "ReleaseBound",
            Self::SourceDecodeBound { .. } => "SourceDecodeBound",
            Self::ParquetContract(_) => "ParquetContract",
            Self::Resolution(_) => "Resolution",
            Self::Manifest(_) => "Manifest",
            Self::Contract(_) => "Contract",
            Self::Artifact(_) => "Artifact",
            Self::DataFusion(_) => "DataFusion",
            Self::Parquet(_) => "Parquet",
            Self::Profile(_) => "Profile",
            Self::Arrow(_) => "Arrow",
        })
    }
}

impl From<CoreResolutionError> for CoreExecutionError {
    fn from(value: CoreResolutionError) -> Self {
        Self::Resolution(Box::new(value))
    }
}