polyc-query 2026.9.0

//! Guards one durable intent from permit issuance through terminal settlement.
//!
//! The guardian is an actor. It owns the sole [`CoreExecutionPermit`] from the
//! moment State grants it. Every later stage holds only a handle. A stage that
//! fails, or a handle that drops, cannot lose the permit by accident.
//!
//! One permit becomes at most one stable completion command. The command's
//! content is fixed at the latch's dispatch point, and every ambiguous retry
//! then settles that same identity and digest.
//!
//! Two paths deliberately record no completion. A crossed permit addresses
//! another operation's audit trail, so the guardian abandons it. A settlement
//! that exhausts its bounded budget also stops. Both leave an unmatched intent
//! for State's own reconciliation; neither is claimed as a durable terminal.

use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use datafusion::error::DataFusionError;
use polyc_projection_artifact::{ArtifactReadError, ManifestOpenError, ReadContractError};
use polyc_state::error::{RetryClass, StateError};
use polyc_state::immutable::ObjectError;
use polyc_state::projection::ProjectionCatalogError;
use polyc_state::projection::artifact::ArtifactAdmissionError;
use polyc_state::query_audit::{
    ErrorClass, QueryAuditError, QueryCompletion, QueryOutcome, SourceSnapshot,
};
#[cfg(test)]
use tokio::sync::Notify;
use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;

use super::CoreExecutionError;
use super::latch::TerminalLatch;
use crate::core_resolution::{
    CoreCompletionContext, CoreExecutionPermit, CoreMetadataAuthority, CoreResolutionError,
};

/// Bounds the complete terminal settlement, including every retry.
const COMPLETION_TIMEOUT: Duration = Duration::from_secs(5);
/// Bounds one settlement or receipt call inside that budget.
const ATTEMPT_TIMEOUT: Duration = Duration::from_millis(500);
/// Bounds how many times the actor may re-present the same command.
///
/// Three attempts plus their pacing stay well inside [`COMPLETION_TIMEOUT`].
/// The timeout is the outer guard for a call that hangs, not the usual limit.
const MAX_ATTEMPTS: usize = 3;
/// Paces the bounded completion retry.
///
/// A transient State refusal must not be re-presented in a tight loop.
const RETRY_BACKOFF: Duration = Duration::from_millis(10);
/// Reserves settlement time when waiting for a withdrawn query's exact report.
///
/// The wait shares one budget with settlement itself. This reserve keeps
/// enough of that budget for the attempts and their pacing, so a slow producer
/// cannot starve the durable write.
const SETTLEMENT_RESERVE: Duration = Duration::from_millis(1500);

struct FinishRequest {
    /// The outcome only. The completion is built inside the latch's own
    /// critical section, from the counts that same section holds, so nothing
    /// can be released between reading them and recording the terminal.
    outcome: QueryOutcome,
    duration: Duration,
    response: oneshot::Sender<Result<(), CoreExecutionError>>,
}

/// Owns the sole permit and settles one stable terminal for its intent.
///
/// Every stage after `BeginOutcome::Granted` carries this value, never the
/// permit. A stage withdraws through [`Self::cancellation`]. It reports a
/// measured terminal through [`Self::finish`] or [`Self::fail`]. Dropping the
/// guardian withdraws and nothing more; the actor performs the settlement, so
/// no destructor does I/O.
pub(crate) struct PermitGuardian {
    request: Option<oneshot::Sender<FinishRequest>>,
    cancellation: CancellationToken,
    latch: Arc<TerminalLatch>,
    source: polyc_state::query_audit::SourceSnapshot,
    started: tokio::time::Instant,
    #[cfg(test)]
    dispatch_gate: Arc<GuardianTestGate>,
    armed: bool,
}

impl PermitGuardian {
    /// Takes ownership of a freshly granted permit and starts its actor.
    ///
    /// # Panics
    ///
    /// Panics when no Tokio runtime is active. The actor is spawned here so
    /// settlement never depends on a caller that may be dropped.
    pub(crate) fn new(
        permit: CoreExecutionPermit,
        metadata: Arc<dyn CoreMetadataAuthority>,
    ) -> Self {
        let (request, receiver) = oneshot::channel();
        let cancellation = CancellationToken::new();
        let latch = Arc::new(TerminalLatch::default());
        let source = permit.source().clone();
        let started = tokio::time::Instant::now();
        #[cfg(test)]
        let dispatch_gate = Arc::new(GuardianTestGate::default());
        tokio::spawn(guard_permit(
            permit,
            metadata,
            cancellation.clone(),
            Arc::clone(&latch),
            started,
            #[cfg(test)]
            Arc::clone(&dispatch_gate),
            receiver,
        ));
        Self {
            request: Some(request),
            cancellation,
            latch,
            source,
            started,
            #[cfg(test)]
            dispatch_gate,
            armed: true,
        }
    }

    /// Returns the token that withdraws the query.
    pub(crate) fn cancellation(&self) -> CancellationToken {
        self.cancellation.clone()
    }

    /// Returns the latch the consumer records delivered rows into.
    pub(crate) fn latch(&self) -> Arc<TerminalLatch> {
        Arc::clone(&self.latch)
    }

    /// Returns the exact source premises the intent recorded.
    pub(crate) const fn source(&self) -> &polyc_state::query_audit::SourceSnapshot {
        &self.source
    }

    /// Offers the exact terminal the caller measured.
    ///
    /// The latch decides whether this terminal, a withdrawal, or a bound the
    /// consumer settled becomes the command. A measured failure always
    /// survives: it outranks both of the others, in either arrival order. A
    /// measured success does not survive a withdrawal that reached the latch
    /// first, nor a consumer bound, because both describe what the caller
    /// actually received and this does not.
    ///
    /// # Errors
    ///
    /// Returns [`CoreExecutionError::AuditCompletionUnavailable`] when the
    /// actor cannot record a durable completion inside its budget. Returns
    /// [`CoreExecutionError::AuditReceiptMismatch`] when State answers with a
    /// receipt for another command. Both leave the intent unmatched for
    /// State's reconciliation, and neither reports success.
    pub(crate) async fn finish(mut self, outcome: QueryOutcome) -> Result<(), CoreExecutionError> {
        let (response, completed) = oneshot::channel();
        let request = FinishRequest {
            outcome,
            duration: self.started.elapsed(),
            response,
        };
        self.armed = false;
        self.request
            .take()
            .ok_or(CoreExecutionError::AuditCompletionUnavailable)?
            .send(request)
            .map_err(|_| CoreExecutionError::AuditCompletionUnavailable)?;
        completed
            .await
            .unwrap_or(Err(CoreExecutionError::AuditCompletionUnavailable))
    }

    /// Reports a refusal that released no row.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`Self::finish`].
    pub(crate) async fn fail(self, class: ErrorClass) -> Result<(), CoreExecutionError> {
        self.finish(QueryOutcome::Failed(class)).await
    }

    /// Releases the permit without settling, leaving the intent unmatched.
    ///
    /// A crossed permit addresses another operation's audit trail. Completing
    /// it would write this query's terminal onto that trail. The caller fails
    /// closed instead, and State's reconciliation owns the unmatched intent.
    ///
    /// This is the only caller-chosen path that records no completion.
    pub(crate) fn abandon(mut self) {
        self.armed = false;
        drop(self.request.take());
    }

    #[cfg(test)]
    pub(crate) fn pause_dispatch(&self) -> GuardianDispatchPause {
        self.dispatch_gate
            .paused
            .store(true, std::sync::atomic::Ordering::SeqCst);
        GuardianDispatchPause {
            gate: Arc::clone(&self.dispatch_gate),
        }
    }
}

#[cfg(test)]
#[derive(Default)]
struct GuardianTestGate {
    paused: std::sync::atomic::AtomicBool,
    waiting: Notify,
    resume: Notify,
}

#[cfg(test)]
pub(crate) struct GuardianDispatchPause {
    gate: Arc<GuardianTestGate>,
}

#[cfg(test)]
impl GuardianDispatchPause {
    pub(crate) async fn wait(&self) {
        self.gate.waiting.notified().await;
    }

    pub(crate) fn resume(&self) {
        self.gate
            .paused
            .store(false, std::sync::atomic::Ordering::SeqCst);
        self.gate.resume.notify_waiters();
    }
}

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

impl Drop for PermitGuardian {
    fn drop(&mut self) {
        if self.armed {
            self.latch.cancel();
            self.cancellation.cancel();
        }
    }
}

async fn guard_permit(
    permit: CoreExecutionPermit,
    metadata: Arc<dyn CoreMetadataAuthority>,
    cancellation: CancellationToken,
    latch: Arc<TerminalLatch>,
    started: tokio::time::Instant,
    #[cfg(test)] dispatch_gate: Arc<GuardianTestGate>,
    mut receiver: oneshot::Receiver<FinishRequest>,
) {
    let source = permit.source().clone();
    let operation = CoreCompletionContext::server_owned(COMPLETION_TIMEOUT);
    let mut response = None;

    // The query ends when the producer reports a measured terminal or the
    // consumer withdraws. Both transitions go through the latch, so which one
    // becomes the command is decided under one lock rather than by whichever
    // task the scheduler ran first.
    // A oneshot receiver may be polled only until it resolves, so remember
    // whether a later report can still arrive at all.
    let mut report_possible = true;
    let abandoned = tokio::select! {
        biased;
        request = &mut receiver => {
            report_possible = false;
            match request {
                Ok(request) => {
                    let _ = latch.report(request.outcome, request.duration, &source);
                    response = Some(request.response);
                    false
                }
                // An armed handle that drops has already withdrawn through the
                // latch. An unarmed one was abandoned: its permit crossed
                // another audit trail, so this actor settles nothing at all.
                Err(_) => !latch.is_settled(),
            }
        }
        () = cancellation.cancelled() => {
            latch.cancel();
            false
        }
    };
    if abandoned {
        return;
    }

    // A withdrawn query may still have a measured terminal in flight. Wait for
    // it inside the settlement budget rather than a scheduling grace, so a
    // descheduled producer cannot cost the audit its exact class. The latch
    // already holds the delivered counts, so this wait only sharpens the
    // outcome. It can never be the difference between real rows and zero.
    if response.is_none() && report_possible {
        response = await_late_report(&latch, &operation, &mut receiver, &source).await;
    }

    // The linearization point. Exactly one dispatch wins, and after it a
    // withdrawal cannot change the command's content.
    dispatch_boundary(
        #[cfg(test)]
        dispatch_gate.as_ref(),
    )
    .await;
    if cancellation.is_cancelled() {
        latch.cancel();
    }
    let completion = latch.dispatch(&source, started.elapsed());

    let result = settle_completion(&operation, metadata, permit, completion).await;
    if let Some(response) = response {
        let _ = response.send(result);
    }
}

/// Waits inside the settlement budget for a withdrawn query's exact report.
///
/// Returns the response channel when one arrives, so the reporting task still
/// learns whether its terminal was durably settled.
async fn await_late_report(
    latch: &TerminalLatch,
    operation: &CoreCompletionContext,
    receiver: &mut oneshot::Receiver<FinishRequest>,
    source: &SourceSnapshot,
) -> Option<oneshot::Sender<Result<(), CoreExecutionError>>> {
    let headroom = report_headroom(operation)?;
    let Ok(Ok(request)) = tokio::time::timeout(headroom, receiver).await else {
        return None;
    };
    let _ = latch.report(request.outcome, request.duration, source);
    Some(request.response)
}

/// Returns how long the actor may wait for a withdrawn query's exact report.
///
/// The wait and the settlement share one budget. This keeps a reserve for the
/// bounded attempts and their pacing.
fn report_headroom(operation: &CoreCompletionContext) -> Option<Duration> {
    let remaining = operation.remaining().ok()?;
    let headroom = remaining.checked_sub(SETTLEMENT_RESERVE)?;
    (!headroom.is_zero()).then_some(headroom)
}

/// Returns the future awaited at the exact permit-to-command boundary.
///
/// Production has no work at that boundary. The test build inserts a pause
/// there. A test then holds the guardian inside the pre-dispatch window and
/// proves which transition the latch selects.
#[cfg(not(test))]
fn dispatch_boundary() -> std::future::Ready<()> {
    std::future::ready(())
}

#[cfg(test)]
async fn dispatch_boundary(gate: &GuardianTestGate) {
    if gate.paused.load(std::sync::atomic::Ordering::SeqCst) {
        gate.waiting.notify_one();
        gate.resume.notified().await;
    }
}

/// Reports whether a durable completion already answers this exact command.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReceiptVerdict {
    /// State holds this exact command's completion.
    Settled,
    /// State holds no completion for this command yet.
    Absent,
    /// State answered with a receipt for another command.
    Crossed,
    /// State could not answer inside the remaining budget.
    Unknown,
}

async fn settle_completion(
    operation: &CoreCompletionContext,
    metadata: Arc<dyn CoreMetadataAuthority>,
    permit: CoreExecutionPermit,
    completion: QueryCompletion,
) -> Result<(), CoreExecutionError> {
    let command = permit.into_completion(completion)?;
    for attempt in 0..MAX_ATTEMPTS {
        operation
            .check()
            .map_err(|_| CoreExecutionError::AuditCompletionUnavailable)?;
        let result =
            bounded_terminal_call(operation, metadata.complete_audit(operation, &command)).await;
        let receipt_may_exist = match result {
            Some(Ok(receipt)) if receipt.answers(command.metadata()) => return Ok(()),
            Some(Ok(_)) => return Err(CoreExecutionError::AuditReceiptMismatch),
            Some(Err(error)) => match completion_retry_class(&error) {
                RetryClass::Terminal => return Err(CoreExecutionError::AuditCompletionUnavailable),
                RetryClass::Ambiguous => true,
                RetryClass::Transient => false,
            },
            None => true,
        };
        if receipt_may_exist {
            match confirm_recorded_receipt(operation, metadata.as_ref(), &command).await {
                ReceiptVerdict::Settled => return Ok(()),
                ReceiptVerdict::Crossed => return Err(CoreExecutionError::AuditReceiptMismatch),
                ReceiptVerdict::Absent | ReceiptVerdict::Unknown => {}
            }
        }
        if attempt + 1 < MAX_ATTEMPTS {
            tokio::time::sleep(RETRY_BACKOFF).await;
        }
    }
    Err(CoreExecutionError::AuditCompletionUnavailable)
}

async fn confirm_recorded_receipt(
    operation: &CoreCompletionContext,
    metadata: &dyn CoreMetadataAuthority,
    command: &crate::core_resolution::CoreCompletionCommand,
) -> ReceiptVerdict {
    match bounded_terminal_call(operation, metadata.completion_receipt(operation, command)).await {
        Some(Ok(Some(receipt))) if receipt.answers(command.metadata()) => ReceiptVerdict::Settled,
        Some(Ok(Some(_))) => ReceiptVerdict::Crossed,
        Some(Ok(None)) => ReceiptVerdict::Absent,
        Some(Err(error)) if completion_retry_class(&error) == RetryClass::Terminal => {
            ReceiptVerdict::Crossed
        }
        Some(Err(_)) | None => ReceiptVerdict::Unknown,
    }
}

async fn bounded_terminal_call<T>(
    operation: &CoreCompletionContext,
    future: impl Future<Output = T>,
) -> Option<T> {
    let remaining = operation.remaining().ok()?;
    tokio::time::timeout(remaining.min(ATTEMPT_TIMEOUT), future)
        .await
        .ok()
}

const fn completion_retry_class(error: &CoreResolutionError) -> RetryClass {
    match error {
        CoreResolutionError::Audit(error) => error.retry_class(),
        CoreResolutionError::State(error) => error.retry_class(),
        CoreResolutionError::Projection(error) => error.retry_class(),
        CoreResolutionError::Unavailable
        | CoreResolutionError::Statement(_)
        | CoreResolutionError::DataFusion(_)
        | CoreResolutionError::NoSourceDependency
        | CoreResolutionError::UnknownDependency(_)
        | CoreResolutionError::ParameterMismatch
        | CoreResolutionError::InvalidComposition
        | CoreResolutionError::InvalidAttribution
        | CoreResolutionError::EmptyConversationIdentity
        | CoreResolutionError::FreshnessUnsupported { .. }
        | CoreResolutionError::InvalidBounds
        | CoreResolutionError::CrossedPermit
        | CoreResolutionError::CompletionReceiptMismatch
        | CoreResolutionError::MissingSource(_)
        | CoreResolutionError::SourceVectorMismatch
        | CoreResolutionError::SourceMismatch(_)
        | CoreResolutionError::MissingProjection(_)
        | CoreResolutionError::Superseded(_)
        | CoreResolutionError::IncompatibleDescriptor(_)
        | CoreResolutionError::DuplicatePartition
        | CoreResolutionError::DuplicateDescriptor => RetryClass::Terminal,
    }
}

/// Maps one refusal onto the class the durable audit records.
///
/// The taxonomy divides by responsibility. Every classifier below applies the
/// same rule, and none carries a wildcard arm, so a new error variant breaks
/// the build rather than defaulting to a class nobody chose.
///
/// | Class | Who is responsible |
/// |---|---|
/// | `Malformed` | The statement, its parameters, or the plan the caller asked for. |
/// | `Denied` | Current authority refused this caller. |
/// | `Bounds` | A declared resource ceiling was reached. |
/// | `Deadline` | The declared execution deadline expired. |
/// | `Cancelled` | The caller withdrew. |
/// | `Unavailable` | A required source is absent, moved, retired, compacted, or cannot answer now. |
/// | `Internal` | Corruption or a broken deployment: a digest, signed descriptor, generation, length, topology, or protection disagreement, or a fault in this plane. |
pub(crate) fn classify_error(error: &CoreExecutionError) -> ErrorClass {
    match error {
        CoreExecutionError::AuthorityNarrowed => ErrorClass::Denied,
        CoreExecutionError::Deadline => ErrorClass::Deadline,
        CoreExecutionError::Cancelled => ErrorClass::Cancelled,
        CoreExecutionError::ReleaseBound { .. } | CoreExecutionError::SourceDecodeBound { .. } => {
            ErrorClass::Bounds
        }
        // A pinned source moved, the plane is draining, or the durable write
        // could not land. A later query may succeed; this one cannot.
        CoreExecutionError::SourceChanged(_)
        | CoreExecutionError::ExecutionAdmissionClosed
        | CoreExecutionError::AuditCompletionUnavailable => ErrorClass::Unavailable,
        CoreExecutionError::Artifact(error) => classify_artifact(error),
        CoreExecutionError::Manifest(error) => classify_manifest(error),
        CoreExecutionError::Contract(error) => classify_contract(error),
        CoreExecutionError::Resolution(error) => classify_resolution(error),
        CoreExecutionError::DataFusion(error) => classify_datafusion(error),
        // The caller asked for a table this gate cannot execute.
        CoreExecutionError::LegacyProviderUnavailable => ErrorClass::Malformed,
        // Artifact bytes that do not match their signed envelope, a plan that
        // differs from the audited plan, and a receipt for another command are
        // all faults of this plane or its stored data.
        CoreExecutionError::ParquetContract(_)
        | CoreExecutionError::Profile(_)
        | CoreExecutionError::PlanIdentityMismatch
        | CoreExecutionError::AuditReceiptMismatch
        | CoreExecutionError::InvalidComposition(_)
        | CoreExecutionError::RealmMismatch
        | CoreExecutionError::Parquet(_)
        | CoreExecutionError::Arrow(_) => ErrorClass::Internal,
    }
}

/// Classifies an execution-time engine error.
///
/// The statement was compiled before the intent, so a failure here is not the
/// caller's syntax. Only the memory pool's own refusal is a declared ceiling.
const fn classify_datafusion(error: &DataFusionError) -> ErrorClass {
    match error {
        DataFusionError::ResourcesExhausted(_) => ErrorClass::Bounds,
        _ => ErrorClass::Internal,
    }
}

pub(crate) fn classify_resolution(error: &CoreResolutionError) -> ErrorClass {
    match error {
        CoreResolutionError::State(error) => classify_state(error),
        CoreResolutionError::Audit(error) => classify_audit(error),
        CoreResolutionError::Projection(error) => classify_projection(error),
        // A required source is absent or has moved past what this plan pinned.
        CoreResolutionError::MissingSource(_)
        | CoreResolutionError::MissingProjection(_)
        | CoreResolutionError::Superseded(_) => ErrorClass::Unavailable,
        CoreResolutionError::InvalidBounds => ErrorClass::Bounds,
        // The statement, its parameters, or the plan it asked for.
        CoreResolutionError::Statement(_)
        | CoreResolutionError::DataFusion(_)
        | CoreResolutionError::NoSourceDependency
        | CoreResolutionError::UnknownDependency(_)
        | CoreResolutionError::ParameterMismatch
        | CoreResolutionError::EmptyConversationIdentity
        | CoreResolutionError::FreshnessUnsupported { .. } => ErrorClass::Malformed,
        // A missing composition, an absent audit attribution, a crossed
        // permit, a crossed or duplicated source response, an incompatible
        // descriptor, and a durable completion that disagrees are all faults
        // of this plane or its deployment.
        CoreResolutionError::Unavailable
        | CoreResolutionError::InvalidComposition
        | CoreResolutionError::InvalidAttribution
        | CoreResolutionError::CrossedPermit
        | CoreResolutionError::CompletionReceiptMismatch
        | CoreResolutionError::SourceVectorMismatch
        | CoreResolutionError::SourceMismatch(_)
        | CoreResolutionError::IncompatibleDescriptor(_)
        | CoreResolutionError::DuplicatePartition
        | CoreResolutionError::DuplicateDescriptor => ErrorClass::Internal,
    }
}

const fn classify_state(error: &StateError) -> ErrorClass {
    match error {
        StateError::Denied { .. } => ErrorClass::Denied,
        StateError::DeadlineExpired { .. } => ErrorClass::Deadline,
        StateError::Cancelled { .. } => ErrorClass::Cancelled,
        StateError::BoundsExceeded { .. } => ErrorClass::Bounds,
        // A premise the query pinned moved, or the source could not answer.
        // Each variant's own documentation separates it from a caller error,
        // and the class says nothing about whether a retry would help: some
        // members clear on their own and some need an operator.
        StateError::Unavailable { .. }
        | StateError::AmbiguousOutcome { .. }
        | StateError::DuplicateCommand { .. }
        | StateError::PartitionHeld { .. }
        | StateError::RevisionConflict { .. }
        | StateError::IncarnationConflict { .. }
        | StateError::SourceIncarnationChanged { .. }
        | StateError::RetiredRevision { .. }
        | StateError::CompactedRange { .. }
        | StateError::SnapshotUnavailable { .. }
        // State's own storage is damaged, so a source this query depends on
        // genuinely cannot answer. That is what this class says.
        //
        // It is not `Malformed`: no request a caller can send repairs storage.
        // It is not `Internal`: the fault is the state plane's, not this one's.
        // The class carries no retry promise — it is the audit taxonomy — and
        // this outcome is terminal until an operator repairs the partition.
        | StateError::JournalDamaged => ErrorClass::Unavailable,
        StateError::Malformed { .. } => ErrorClass::Malformed,
        // Both mean this plane presented content State did not expect under an
        // identity it already holds.
        StateError::StaleFence { .. } | StateError::DigestConflict { .. } => ErrorClass::Internal,
    }
}

const fn classify_audit(error: &QueryAuditError) -> ErrorClass {
    match error {
        QueryAuditError::State(error) => classify_state(error),
        QueryAuditError::NoRecordedIntent { .. } => ErrorClass::Internal,
    }
}

fn classify_projection(error: &ProjectionCatalogError) -> ErrorClass {
    match error {
        ProjectionCatalogError::State(error) => classify_state(error),
        ProjectionCatalogError::Object(error) => classify_object(error),
        ProjectionCatalogError::Artifact(error) => classify_admission(error),
        ProjectionCatalogError::UnknownManifest { .. } => ErrorClass::Unavailable,
        // The catalog holds a publisher, generation, or manifest state that
        // disagrees with what this read expected.
        ProjectionCatalogError::StalePublisher { .. }
        | ProjectionCatalogError::GenerationConflict { .. }
        | ProjectionCatalogError::NonMonotonic { .. }
        | ProjectionCatalogError::CursorRegression { .. }
        | ProjectionCatalogError::ManifestConflict { .. } => ErrorClass::Internal,
    }
}

const fn classify_object(error: &ObjectError) -> ErrorClass {
    match error {
        ObjectError::State(error) => classify_state(error),
        ObjectError::UnknownGeneration { .. } => ErrorClass::Unavailable,
        // A generation, descriptor, retention, or protection disagreement.
        ObjectError::ProtectionUnavailable { .. }
        | ObjectError::StaleGeneration { .. }
        | ObjectError::NotMonotonic { .. }
        | ObjectError::Retained { .. }
        | ObjectError::CurrentGeneration { .. }
        | ObjectError::DescriptorConflict { .. } => ErrorClass::Internal,
    }
}

const fn classify_admission(error: &ArtifactAdmissionError) -> ErrorClass {
    match error {
        ArtifactAdmissionError::Malformed(error) => classify_state(error),
        // Stored bytes that do not match their signed descriptor, an untrusted
        // signature, and a store that cannot protect the family are all
        // corruption or a broken deployment.
        ArtifactAdmissionError::ProtectionUnavailable { .. }
        | ArtifactAdmissionError::ContentMismatch { .. }
        | ArtifactAdmissionError::Signature { .. }
        | ArtifactAdmissionError::Disagreement { .. } => ErrorClass::Internal,
    }
}

const fn classify_artifact(error: &ArtifactReadError) -> ErrorClass {
    match error {
        // The object store could not answer now, or the named generation is
        // simply not there. Neither is the caller's statement.
        ArtifactReadError::Unavailable { .. } | ArtifactReadError::NotFound { .. } => {
            ErrorClass::Unavailable
        }
        // A declared verification ceiling.
        ArtifactReadError::VerificationBudgetExceeded { .. } => ErrorClass::Bounds,
        // Every remaining variant is a disagreement between the signed
        // descriptor and the stored object: a refused realm, the wrong
        // generation, a bad digest, or a length that does not match.
        ArtifactReadError::Refused { .. }
        | ArtifactReadError::GenerationMismatch { .. }
        | ArtifactReadError::DigestMismatch { .. }
        | ArtifactReadError::MetadataLengthMismatch { .. }
        | ArtifactReadError::RangeLengthMismatch { .. }
        | ArtifactReadError::RangeOutOfBounds { .. } => ErrorClass::Internal,
    }
}

const fn classify_manifest(error: &ManifestOpenError) -> ErrorClass {
    match error {
        ManifestOpenError::Read(error) => classify_artifact(error),
        ManifestOpenError::TooLarge { .. } => ErrorClass::Bounds,
        // A namespace the deployment never configured, a realm topology that
        // disagrees, a store that cannot protect the family, a descriptor that
        // does not match, and a manifest this deployment does not trust.
        ManifestOpenError::NamespaceNotConfigured { .. }
        | ManifestOpenError::TopologyMismatch { .. }
        | ManifestOpenError::ProtectionUnavailable { .. }
        | ManifestOpenError::DescriptorMismatch { .. }
        | ManifestOpenError::Verification(_) => ErrorClass::Internal,
    }
}

const fn classify_contract(error: &ReadContractError) -> ErrorClass {
    match error {
        // The admitted manifest carries no such table, so the source cannot
        // answer this plan.
        ReadContractError::MissingTable { .. } => ErrorClass::Unavailable,
        // Every remaining variant reports how this process was configured.
        ReadContractError::InvalidVerificationBudget { .. }
        | ReadContractError::EmptyRealm { .. }
        | ReadContractError::SharedNamespace { .. }
        | ReadContractError::NamespaceNotConfigured { .. }
        | ReadContractError::ProtectionUnavailable { .. } => ErrorClass::Internal,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use polyc_state::id::PartitionId;
    use polyc_state::projection::artifact::AccessRealm;
    use polyc_state::revision::{JournalPosition, Revision};

    /// Damaged journal storage is an unavailable source, not a caller error
    /// and not this plane's fault.
    ///
    /// The class is the audit taxonomy. It records why a query ended and does
    /// not decide whether anything retries, which is why a terminal outcome
    /// may share it with outcomes that clear on their own. Nothing in this
    /// crate reads the class to choose a retry.
    ///
    /// `Malformed` would blame the caller for storage no request can repair.
    /// `Internal` would place the fault in the query plane, when it is the
    /// state plane whose storage is damaged.
    #[test]
    fn damaged_journal_storage_is_unavailable_not_malformed_or_internal() {
        let class = classify_state(&StateError::JournalDamaged);

        assert_eq!(class, ErrorClass::Unavailable);
        assert_ne!(
            class,
            ErrorClass::Malformed,
            "no request a caller can send repairs damaged storage"
        );
        assert_ne!(
            class,
            ErrorClass::Internal,
            "the fault is the state plane's, not the query plane's"
        );

        // The kernel still calls the outcome terminal. The class above must
        // not be read as disagreeing with that: they answer different
        // questions, and only this one reaches the audit record.
        assert_eq!(
            StateError::JournalDamaged.retry_class(),
            polyc_state::error::RetryClass::Terminal,
            "an operator repairs this; waiting never clears it"
        );
        assert!(!StateError::JournalDamaged.is_retry_safe());
    }

    #[test]
    fn a_corrupt_or_disagreeing_artifact_is_internal_not_malformed() {
        for error in [
            ArtifactReadError::DigestMismatch {
                key: "content-free".to_owned(),
                generation: 1,
            },
            ArtifactReadError::GenerationMismatch {
                expected: 2,
                observed: 3,
                key: "content-free".to_owned(),
            },
            ArtifactReadError::MetadataLengthMismatch {
                key: "content-free".to_owned(),
                expected: 4,
                observed: 5,
            },
            ArtifactReadError::RangeLengthMismatch {
                key: "content-free".to_owned(),
                offset: 0,
                expected: 4,
                observed: 5,
            },
            ArtifactReadError::RangeOutOfBounds {
                key: "content-free".to_owned(),
                offset: 9,
                requested: 4,
                byte_len: 2,
            },
            ArtifactReadError::Refused {
                reason: "content-free".to_owned(),
            },
        ] {
            assert_eq!(
                classify_artifact(&error),
                ErrorClass::Internal,
                "a stored-content or deployment disagreement is never Malformed"
            );
        }
    }

    #[test]
    fn an_absent_source_is_unavailable_and_a_declared_ceiling_is_bounds() {
        assert_eq!(
            classify_artifact(&ArtifactReadError::NotFound {
                key: "content-free".to_owned(),
                generation: 1,
            }),
            ErrorClass::Unavailable
        );
        assert_eq!(
            classify_artifact(&ArtifactReadError::Unavailable {
                reason: "content-free".to_owned(),
            }),
            ErrorClass::Unavailable
        );
        assert_eq!(
            classify_artifact(&ArtifactReadError::VerificationBudgetExceeded {
                key: "content-free".to_owned(),
                byte_len: 2,
                max_byte_len: 1,
            }),
            ErrorClass::Bounds
        );
    }

    #[test]
    fn nested_state_classes_remain_distinct() {
        let family = polyc_state::id::OperationFamily::new("query-test");
        assert_eq!(
            classify_state(&StateError::Denied {
                family: family.clone(),
            }),
            ErrorClass::Denied
        );
        assert_eq!(
            classify_state(&StateError::Cancelled { family }),
            ErrorClass::Cancelled
        );
    }

    #[test]
    fn a_retired_or_compacted_source_is_unavailable_not_malformed() {
        assert_eq!(
            classify_state(&StateError::RetiredRevision {
                requested: Revision::new(4),
                earliest: Revision::new(9),
            }),
            ErrorClass::Unavailable
        );
        assert_eq!(
            classify_state(&StateError::CompactedRange {
                partition: PartitionId::new("conv-a"),
                requested: JournalPosition::new(4),
                earliest: JournalPosition::new(9),
            }),
            ErrorClass::Unavailable
        );
    }

    #[test]
    fn a_deployment_fault_is_internal_and_a_missing_table_is_unavailable() {
        assert_eq!(
            classify_contract(&ReadContractError::NamespaceNotConfigured {
                realm: AccessRealm::Visible,
                namespace: "absent".to_owned(),
            }),
            ErrorClass::Internal
        );
        assert_eq!(
            classify_contract(&ReadContractError::MissingTable { table: "messages" }),
            ErrorClass::Unavailable
        );
    }

    #[test]
    fn only_the_memory_pool_makes_an_engine_error_a_ceiling() {
        assert_eq!(
            classify_datafusion(&DataFusionError::ResourcesExhausted(
                "content-free".to_owned()
            )),
            ErrorClass::Bounds
        );
        assert_eq!(
            classify_datafusion(&DataFusionError::Internal("content-free".to_owned())),
            ErrorClass::Internal
        );
    }
}