polyc-query 2026.9.0

//! Production-only capabilities for projected `conversation-core/v1` reads.
//!
//! The types here remain crate-private until the service and atomic cutover
//! slices compose them. They close two authority gaps before that point:
//!
//! - metadata planning has complete direct-State and State-Connect adapters;
//! - artifacts arrive through a GCS client whose type and OAuth scope are both
//!   read-only, then immediately disappear behind realm-typed sealed readers.
//!
//! No constructor accepts [`polyc_storage_gcs::GcsClient`]. A Query deployment
//! therefore cannot gain create, mutable-name read, list, or delete authority
//! through this module.

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

use async_trait::async_trait;
use connectrpc::client::ClientTransport;
use polyc_projection_artifact::{
    ArtifactReadError, ArtifactReadFuture, ExactArtifactBackend, ExactObjectMetadata,
    ExactObjectRange, FleetArtifactAccess, FleetArtifactReader, RealmTopology,
    VisibleArtifactAccess, VisibleArtifactReader,
};
use polyc_state::immutable::AtRestProtection;
use polyc_state::journal::{
    CreateJournalDirectorySnapshot, GetJournalSource, JournalDirectoryPage,
    JournalDirectorySnapshot, JournalSourceHead, ListJournalDirectorySnapshot,
    ReleaseJournalDirectorySnapshot,
};
// Reached only by the in-process adapter below, which the tests compose.
#[cfg(test)]
use polyc_state::journal::JournalRead;
#[cfg(test)]
use polyc_state::projection::ProjectionCatalog;
use polyc_state::projection::artifact::{ExactObjectRef, ManifestTrust, ObjectNamespace};
use polyc_state::projection::{ProjectionResolution, ResolveManifest};
#[cfg(test)]
use polyc_state::query_audit::{AuditPhase, QueryAuditRead, QueryAuditWrite, ReadQueryAudit};
use polyc_state::query_audit::{BeginOutcome, BeginQueryAudit};
use polyc_state::receipt::Receipt;
use polyc_state_connect::journal::client::JournalClient;
use polyc_state_connect::projection::client::ProjectionCatalogClient;
use polyc_state_connect::query_audit::client::QueryAuditClient;
use polyc_storage_gcs::{GcsError, GcsReadClient};

use crate::core_execution::{
    CoreArtifactAuthority, CoreExecutionAdmission, CoreExecutionError, CurrentCredentialAuthority,
};
use crate::core_resolution::{
    CoreCompletionCommand, CoreCompletionContext, CoreExecutionPermit, CoreMetadataAuthority,
    CoreOperationContext, CoreResolutionError,
};

/// Names the in-process State audit capability Query composes directly.
///
/// The transported composition reaches the same two capabilities through
/// `polyc_state_connect`. This alias exists so the direct composition states
/// which pair it needs, in one place, without naming a transport type.
#[cfg(test)]
pub(crate) trait DirectQueryAudit: QueryAuditRead + QueryAuditWrite {}

#[cfg(test)]
impl<T> DirectQueryAudit for T where T: QueryAuditRead + QueryAuditWrite + ?Sized {}

/// Direct in-process access to the three State capabilities planning and
/// terminal settlement need. The journal, catalog, and audit implementations
/// may be backed by different modules; no umbrella State authority is accepted.
#[cfg(test)]
pub(crate) struct DirectCoreMetadata {
    journal: Arc<dyn JournalRead>,
    projections: Arc<dyn ProjectionCatalog>,
    audit: Arc<dyn DirectQueryAudit>,
}

#[cfg(test)]
impl DirectCoreMetadata {
    pub(crate) const fn new(
        journal: Arc<dyn JournalRead>,
        projections: Arc<dyn ProjectionCatalog>,
        audit: Arc<dyn DirectQueryAudit>,
    ) -> Self {
        Self {
            journal,
            projections,
            audit,
        }
    }
}

#[async_trait]
#[cfg(test)]
impl CoreMetadataAuthority for DirectCoreMetadata {
    async fn create_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
    ) -> Result<JournalDirectorySnapshot, CoreResolutionError> {
        Ok(self.journal.create_directory_snapshot(
            CreateJournalDirectorySnapshot,
            operation.local_context()?,
        )?)
    }

    async fn directory_page(
        &self,
        operation: &CoreOperationContext,
        request: ListJournalDirectorySnapshot,
    ) -> Result<JournalDirectoryPage, CoreResolutionError> {
        Ok(self
            .journal
            .directory_page(request, operation.local_context()?)?)
    }

    async fn release_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        request: ReleaseJournalDirectorySnapshot,
    ) -> Result<(), CoreResolutionError> {
        Ok(self
            .journal
            .release_directory_snapshot(request, operation.local_context()?)?)
    }

    async fn source_head(
        &self,
        operation: &CoreOperationContext,
        request: GetJournalSource,
    ) -> Result<Option<JournalSourceHead>, CoreResolutionError> {
        Ok(self
            .journal
            .source_head(request, operation.local_context()?)?)
    }

    async fn resolve_manifest(
        &self,
        operation: &CoreOperationContext,
        request: ResolveManifest,
    ) -> Result<ProjectionResolution, CoreResolutionError> {
        Ok(self
            .projections
            .resolve(request, operation.local_context()?)?)
    }

    async fn begin_audit(
        &self,
        operation: &CoreOperationContext,
        command: BeginQueryAudit,
    ) -> Result<BeginOutcome<CoreExecutionPermit>, CoreResolutionError> {
        Ok(map_begin_outcome(
            self.audit.begin(command, operation.local_context()?)?,
        ))
    }

    async fn complete_audit(
        &self,
        operation: &CoreCompletionContext,
        command: &CoreCompletionCommand,
    ) -> Result<Receipt, CoreResolutionError> {
        let CoreCompletionCommand::Local(command) = command else {
            return Err(CoreResolutionError::InvalidComposition);
        };
        Ok(self
            .audit
            .complete(command.clone(), operation.local_context()?)?)
    }

    async fn completion_receipt(
        &self,
        operation: &CoreCompletionContext,
        command: &CoreCompletionCommand,
    ) -> Result<Option<Receipt>, CoreResolutionError> {
        let CoreCompletionCommand::Local(command) = command else {
            return Err(CoreResolutionError::InvalidComposition);
        };
        let context = operation.local_context()?;
        let Some(receipt) = self.audit.recorded_receipt(
            command.query(),
            command.namespace(),
            AuditPhase::Completion,
        )?
        else {
            return Ok(None);
        };
        if !receipt.is_deduplicated() || !receipt.answers(command.metadata()) {
            return Err(CoreResolutionError::CompletionReceiptMismatch);
        }
        let audit = self
            .audit
            .audit(
                ReadQueryAudit::new(command.query().clone(), command.namespace().clone()),
                context,
            )?
            .ok_or(CoreResolutionError::CompletionReceiptMismatch)?;
        if audit.intent().source() != command.intent_source()
            || audit.completion() != Some(command.completion())
        {
            return Err(CoreResolutionError::CompletionReceiptMismatch);
        }
        Ok(Some(receipt))
    }
}

/// State-Connect access to the same closed metadata capability set.
pub(crate) struct ConnectCoreMetadata<T> {
    journal: JournalClient<T>,
    projections: ProjectionCatalogClient<T>,
    audit: QueryAuditClient<T>,
}

impl<T> ConnectCoreMetadata<T> {
    pub(crate) const fn new(
        journal: JournalClient<T>,
        projections: ProjectionCatalogClient<T>,
        audit: QueryAuditClient<T>,
    ) -> Self {
        Self {
            journal,
            projections,
            audit,
        }
    }
}

#[async_trait]
impl<T> CoreMetadataAuthority for ConnectCoreMetadata<T>
where
    T: ClientTransport + Send + Sync,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: fmt::Display,
{
    async fn create_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
    ) -> Result<JournalDirectorySnapshot, CoreResolutionError> {
        Ok(self
            .journal
            .create_directory_snapshot(&operation.declared()?, &CreateJournalDirectorySnapshot)
            .await?)
    }

    async fn directory_page(
        &self,
        operation: &CoreOperationContext,
        request: ListJournalDirectorySnapshot,
    ) -> Result<JournalDirectoryPage, CoreResolutionError> {
        Ok(self
            .journal
            .directory_page(&operation.declared()?, &request)
            .await?)
    }

    async fn release_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        request: ReleaseJournalDirectorySnapshot,
    ) -> Result<(), CoreResolutionError> {
        Ok(self
            .journal
            .release_directory_snapshot(&operation.declared()?, &request)
            .await?)
    }

    async fn source_head(
        &self,
        operation: &CoreOperationContext,
        request: GetJournalSource,
    ) -> Result<Option<JournalSourceHead>, CoreResolutionError> {
        Ok(self
            .journal
            .source_head(&operation.declared()?, &request)
            .await?)
    }

    async fn resolve_manifest(
        &self,
        operation: &CoreOperationContext,
        request: ResolveManifest,
    ) -> Result<ProjectionResolution, CoreResolutionError> {
        Ok(self
            .projections
            .resolve(&operation.declared()?, &request)
            .await?)
    }

    async fn begin_audit(
        &self,
        operation: &CoreOperationContext,
        command: BeginQueryAudit,
    ) -> Result<BeginOutcome<CoreExecutionPermit>, CoreResolutionError> {
        Ok(map_begin_outcome(
            self.audit.begin(&operation.declared()?, &command).await?,
        ))
    }

    async fn complete_audit(
        &self,
        operation: &CoreCompletionContext,
        command: &CoreCompletionCommand,
    ) -> Result<Receipt, CoreResolutionError> {
        let CoreCompletionCommand::Remote(command) = command else {
            return Err(CoreResolutionError::InvalidComposition);
        };
        let receipt = self
            .audit
            .complete(&operation.declared()?, command)
            .await
            .map_err(settlement_answer)?;
        // Defence in depth. The client already refuses a receipt that does
        // not answer this command; checking again here means one broken or
        // hostile State is classified the same way whichever code path
        // noticed it first.
        if !receipt.answers(command.metadata()) {
            return Err(CoreResolutionError::CompletionReceiptMismatch);
        }
        Ok(receipt)
    }

    async fn completion_receipt(
        &self,
        operation: &CoreCompletionContext,
        command: &CoreCompletionCommand,
    ) -> Result<Option<Receipt>, CoreResolutionError> {
        let CoreCompletionCommand::Remote(command) = command else {
            return Err(CoreResolutionError::InvalidComposition);
        };
        let Some(receipt) = self
            .audit
            .completion_receipt(&operation.declared()?, command)
            .await
            .map_err(settlement_answer)?
        else {
            return Ok(None);
        };
        if !receipt.is_deduplicated() || !receipt.answers(command.metadata()) {
            return Err(CoreResolutionError::CompletionReceiptMismatch);
        }
        Ok(Some(receipt))
    }
}

/// Classifies a settlement answer that does not match the presented command.
///
/// The Connect client refuses a response whose receipt or audit trail
/// disagrees with the command it sent, and names the exact field. Those are
/// not State declining the operation: they are a remote answering about
/// something else, which is what a broken or compromised listener looks like
/// from here.
///
/// The direct adapter calls that `CompletionReceiptMismatch`. This adapter
/// agrees, so one dishonest answer is classified identically whichever
/// transport carried it, and never as a retryable State outage.
fn settlement_answer(error: polyc_state::query_audit::QueryAuditError) -> CoreResolutionError {
    if let polyc_state::query_audit::QueryAuditError::State(
        polyc_state::error::StateError::Malformed { field, .. },
    ) = &error
        && (field.starts_with("receipt") || field.starts_with("audit"))
    {
        return CoreResolutionError::CompletionReceiptMismatch;
    }
    error.into()
}

fn map_begin_outcome<P: Into<CoreExecutionPermit>>(
    outcome: BeginOutcome<P>,
) -> BeginOutcome<CoreExecutionPermit> {
    match outcome {
        BeginOutcome::Granted(permit) => BeginOutcome::Granted(permit.into()),
        BeginOutcome::AlreadyRecorded(receipt) => BeginOutcome::AlreadyRecorded(receipt),
    }
}

/// `Debug` reports the protection only. A namespace names where tenant bytes
/// live, so it never reaches a log line through this type.
#[derive(Clone, PartialEq, Eq)]
pub(crate) struct GcsReadNamespace {
    namespace: ObjectNamespace,
    protection: AtRestProtection,
}

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

impl GcsReadNamespace {
    pub(crate) const fn new(namespace: ObjectNamespace, protection: AtRestProtection) -> Self {
        Self {
            namespace,
            protection,
        }
    }
}

#[derive(Clone)]
struct GcsExactBackend {
    client: GcsReadClient,
    protection: BTreeMap<ObjectNamespace, AtRestProtection>,
}

impl GcsExactBackend {
    fn try_new(
        client: GcsReadClient,
        profiles: &[GcsReadNamespace],
    ) -> Result<Self, CoreExecutionError> {
        let protection = profiles
            .iter()
            .map(|profile| (profile.namespace.clone(), profile.protection))
            .collect::<BTreeMap<_, _>>();
        if profiles.is_empty() || protection.len() != profiles.len() {
            return Err(CoreExecutionError::InvalidComposition(
                "an artifact realm has empty or duplicate namespace protection",
            ));
        }
        Ok(Self { client, protection })
    }

    fn object_name(reference: &ExactObjectRef) -> String {
        format!(
            "{}/{}",
            reference.namespace().as_str(),
            reference.key().as_str()
        )
    }
}

impl ExactArtifactBackend for GcsExactBackend {
    fn declared_protection(&self, namespace: &ObjectNamespace) -> AtRestProtection {
        self.protection
            .get(namespace)
            .copied()
            .unwrap_or(AtRestProtection::None)
    }

    fn head_exact(
        &self,
        reference: &ExactObjectRef,
    ) -> ArtifactReadFuture<'_, ExactObjectMetadata> {
        let name = Self::object_name(reference);
        let generation = reference.generation();
        Box::pin(async move {
            let address = i64::try_from(generation).map_err(|_| ArtifactReadError::Refused {
                reason: format!("{name} names a generation GCS cannot address"),
            })?;
            match self.client.head_exact(&name, address).await {
                Ok(metadata) => {
                    let observed = u64::try_from(metadata.generation).map_err(|_| {
                        ArtifactReadError::Refused {
                            reason: format!("{name} returned an invalid generation"),
                        }
                    })?;
                    Ok(ExactObjectMetadata::new(observed, metadata.size))
                }
                Err(error) => Err(classify_gcs_read(&name, generation, None, error)),
            }
        })
    }

    fn read_exact_range(
        &self,
        reference: &ExactObjectRef,
        offset: u64,
        len: u64,
    ) -> ArtifactReadFuture<'_, ExactObjectRange> {
        let name = Self::object_name(reference);
        let generation = reference.generation();
        Box::pin(async move {
            let address = i64::try_from(generation).map_err(|_| ArtifactReadError::Refused {
                reason: format!("{name} names a generation GCS cannot address"),
            })?;
            let served = self
                .client
                .read_exact_range(&name, address, offset, len)
                .await
                .map_err(|error| classify_gcs_read(&name, generation, Some(offset), error))?;
            // This adapter translates the transport's served range into the
            // shared one. It does not decide. `accept_served_range` in the
            // shared realm reader owns the comparison, so one reader makes it
            // for every backend rather than each adapter making its own.
            let observed =
                u64::try_from(served.generation()).map_err(|_| ArtifactReadError::Refused {
                    reason: format!("{name} served a generation this reader cannot represent"),
                })?;
            Ok(ExactObjectRange::new(observed, served.into_bytes()))
        })
    }
}

fn classify_gcs_read(
    name: &str,
    generation: u64,
    offset: Option<u64>,
    error: GcsError,
) -> ArtifactReadError {
    match error {
        GcsError::NotFound { .. } => ArtifactReadError::NotFound {
            key: name.to_owned(),
            generation,
        },
        GcsError::GenerationMismatch {
            expected, observed, ..
        } => match (u64::try_from(expected), u64::try_from(observed)) {
            (Ok(expected), Ok(observed)) => ArtifactReadError::GenerationMismatch {
                expected,
                observed,
                key: name.to_owned(),
            },
            _ => ArtifactReadError::Refused {
                reason: format!("{name} returned an invalid generation mismatch"),
            },
        },
        GcsError::ShortRead {
            offset,
            expected,
            observed,
            ..
        } => ArtifactReadError::RangeLengthMismatch {
            key: name.to_owned(),
            offset,
            expected,
            observed,
        },
        GcsError::OverlongRead {
            offset,
            expected,
            observed_at_least,
            ..
        } => ArtifactReadError::RangeLengthMismatch {
            key: name.to_owned(),
            offset,
            expected,
            observed: observed_at_least,
        },
        GcsError::Http(inner) => ArtifactReadError::Unavailable {
            reason: format!("{name}: {inner}"),
        },
        GcsError::Status { status, message }
            if status == 408 || status == 429 || (500..600).contains(&status) =>
        {
            ArtifactReadError::Unavailable {
                reason: format!("{name}: {status} {message}"),
            }
        }
        GcsError::RangeTooLarge { len, .. } => ArtifactReadError::Refused {
            reason: format!("{name}: exact range of {len} bytes cannot be represented"),
        },
        other => ArtifactReadError::Refused {
            reason: format!(
                "{name}{}: {other}",
                offset.map_or_else(String::new, |value| format!(" at offset {value}"))
            ),
        },
    }
}

pub(crate) struct VisibleGcsSource {
    client: GcsReadClient,
    profiles: Vec<GcsReadNamespace>,
}

impl VisibleGcsSource {
    pub(crate) const fn new(client: GcsReadClient, profiles: Vec<GcsReadNamespace>) -> Self {
        Self { client, profiles }
    }
}

pub(crate) struct FleetGcsSource {
    client: GcsReadClient,
    profiles: Vec<GcsReadNamespace>,
}

impl FleetGcsSource {
    pub(crate) const fn new(client: GcsReadClient, profiles: Vec<GcsReadNamespace>) -> Self {
        Self { client, profiles }
    }
}

fn namespace_set(profiles: &[GcsReadNamespace]) -> BTreeSet<ObjectNamespace> {
    profiles
        .iter()
        .map(|profile| profile.namespace.clone())
        .collect()
}

/// Composes a visible-only artifact authority. Fleet namespace names are
/// topology data only; this path receives no Fleet credential or reader.
#[allow(clippy::too_many_arguments)]
pub(crate) fn visible_gcs_artifacts(
    visible: VisibleGcsSource,
    fleet_namespaces: impl IntoIterator<Item = ObjectNamespace>,
    trust: Arc<dyn ManifestTrust>,
    scope: Arc<dyn CurrentCredentialAuthority>,
    revalidation_interval: Duration,
    admission: Arc<CoreExecutionAdmission>,
) -> Result<CoreArtifactAuthority, CoreExecutionError> {
    let visible_namespaces = namespace_set(&visible.profiles);
    let topology = RealmTopology::try_new(visible_namespaces.clone(), fleet_namespaces)?;
    let backend = GcsExactBackend::try_new(visible.client, &visible.profiles)?;
    let reader: Arc<dyn VisibleArtifactAccess> =
        Arc::new(VisibleArtifactReader::try_new(backend, visible_namespaces)?);
    CoreArtifactAuthority::visible(
        reader,
        trust,
        topology,
        scope,
        revalidation_interval,
        admission,
    )
}

/// Composes Fleet execution with separately typed visible and Fleet-only
/// clients. The visible reader still serves visible artifacts; the Fleet
/// reader cannot mint visible tokens.
#[allow(clippy::too_many_arguments)]
pub(crate) fn fleet_gcs_artifacts(
    visible: VisibleGcsSource,
    fleet: FleetGcsSource,
    trust: Arc<dyn ManifestTrust>,
    scope: Arc<dyn CurrentCredentialAuthority>,
    revalidation_interval: Duration,
    admission: Arc<CoreExecutionAdmission>,
) -> Result<CoreArtifactAuthority, CoreExecutionError> {
    let visible_namespaces = namespace_set(&visible.profiles);
    let fleet_namespaces = namespace_set(&fleet.profiles);
    let topology = RealmTopology::try_new(visible_namespaces.clone(), fleet_namespaces.clone())?;
    let visible_backend = GcsExactBackend::try_new(visible.client, &visible.profiles)?;
    let fleet_backend = GcsExactBackend::try_new(fleet.client, &fleet.profiles)?;
    let visible_reader: Arc<dyn VisibleArtifactAccess> = Arc::new(VisibleArtifactReader::try_new(
        visible_backend,
        visible_namespaces,
    )?);
    let fleet_reader: Arc<dyn FleetArtifactAccess> = Arc::new(FleetArtifactReader::try_new(
        fleet_backend,
        fleet_namespaces,
    )?);
    CoreArtifactAuthority::fleet(
        visible_reader,
        fleet_reader,
        trust,
        topology,
        scope,
        revalidation_interval,
        admission,
    )
}

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

    #[test]
    fn read_status_classification_is_retry_safe_and_content_free() {
        let throttled = classify_gcs_read(
            "visible/object",
            7,
            None,
            GcsError::Status {
                status: 429,
                message: "response body withheld".to_owned(),
            },
        );
        assert!(throttled.is_retryable());

        let denied = classify_gcs_read(
            "visible/object",
            7,
            None,
            GcsError::Status {
                status: 403,
                message: "response body withheld".to_owned(),
            },
        );
        assert!(!denied.is_retryable());
        assert!(!denied.to_string().contains("provider-secret"));
    }

    #[test]
    fn crossed_generation_remains_typed() {
        let error = classify_gcs_read(
            "visible/object",
            7,
            None,
            GcsError::GenerationMismatch {
                name: "visible/object".to_owned(),
                expected: 7,
                observed: 8,
            },
        );
        assert!(matches!(
            error,
            ArtifactReadError::GenerationMismatch {
                expected: 7,
                observed: 8,
                ..
            }
        ));
    }
}

/// The two production metadata adapters, proved against real State.
#[cfg(test)]
mod adapter_conformance;