polyc-query 2026.9.6

//! Production-only capabilities for closed projected conversation 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>,
    versioned: Arc<dyn polyc_state::versioned::VersionedRead>,
    persona_memory: Arc<dyn polyc_state::persona_memory::journal::PersonaMemoryHistory>,
    observation: Arc<dyn polyc_state::observation::ObservationRead>,
}

#[cfg(test)]
impl DirectCoreMetadata {
    #[allow(clippy::too_many_arguments)]
    pub(crate) const fn new(
        journal: Arc<dyn JournalRead>,
        projections: Arc<dyn ProjectionCatalog>,
        audit: Arc<dyn DirectQueryAudit>,
        versioned: Arc<dyn polyc_state::versioned::VersionedRead>,
        persona_memory: Arc<dyn polyc_state::persona_memory::journal::PersonaMemoryHistory>,
        observation: Arc<dyn polyc_state::observation::ObservationRead>,
    ) -> Self {
        Self {
            journal,
            projections,
            audit,
            versioned,
            persona_memory,
            observation,
        }
    }
}

#[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 versioned_source_head(
        &self,
        operation: &CoreOperationContext,
        scope: &polyc_state::command::CommandScope,
    ) -> Result<polyc_state::versioned::VersionedSourceHead, CoreResolutionError> {
        Ok(self.versioned.source_head(
            polyc_state::versioned::GetVersionedSourceHead::new(scope.clone()),
            operation.local_context()?,
        )?)
    }

    async fn persona_memory_source_head(
        &self,
        operation: &CoreOperationContext,
        partition: &polyc_state::persona_memory::journal::MemoryJournalPartition,
    ) -> Result<polyc_state::persona_memory::journal::PersonaMemorySourceHead, CoreResolutionError>
    {
        let page = self.persona_memory.read_history(
            partition,
            None,
            polyc_state::persona_memory::journal::MemoryReplayRange::new(
                0,
                1,
                polyc_state::persona_memory::journal::MAX_PAYLOAD_BYTES,
            ),
            operation.local_context()?,
        )?;
        Ok(
            polyc_state::persona_memory::journal::PersonaMemorySourceHead::new(
                page.incarnation(),
                page.head(),
            ),
        )
    }

    async fn persona_memory_directory_page(
        &self,
        operation: &CoreOperationContext,
        after: Option<&str>,
        limit: u32,
    ) -> Result<polyc_state::persona_memory::journal::MemoryLineagePage, CoreResolutionError> {
        Ok(self
            .persona_memory
            .list_lineage(after, limit, operation.local_context()?)?)
    }

    async fn create_versioned_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        family: &str,
    ) -> Result<polyc_state::versioned::VersionedDirectorySnapshot, CoreResolutionError> {
        Ok(self.versioned.create_directory_snapshot(
            polyc_state::versioned::CreateVersionedDirectorySnapshot::new(family),
            operation.local_context()?,
        )?)
    }

    async fn versioned_directory_page(
        &self,
        operation: &CoreOperationContext,
        request: &polyc_state::versioned::ListVersionedDirectorySnapshot,
    ) -> Result<polyc_state::versioned::VersionedDirectoryPage, CoreResolutionError> {
        Ok(self
            .versioned
            .directory_page(request.clone(), operation.local_context()?)?)
    }

    async fn release_versioned_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        snapshot: &polyc_state::versioned::VersionedDirectorySnapshotId,
    ) -> Result<(), CoreResolutionError> {
        Ok(self.versioned.release_directory_snapshot(
            polyc_state::versioned::ReleaseVersionedDirectorySnapshot::new(snapshot.clone()),
            operation.local_context()?,
        )?)
    }

    async fn observed_head(
        &self,
        operation: &CoreOperationContext,
        collection: &polyc_state::observation::CollectionId,
    ) -> Result<Option<polyc_state::observation::ObservationHead>, CoreResolutionError> {
        operation.check()?;
        Ok(self.observation.latest(collection)?)
    }

    async fn observed_collections(
        &self,
        operation: &CoreOperationContext,
        kind: polyc_state::observation::CollectionKind,
    ) -> Result<polyc_state::observation::ObservedCollectionListing, CoreResolutionError> {
        operation.check()?;
        Ok(self.observation.collections(kind)?)
    }

    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>,
    versioned: polyc_state_connect::versioned_source::VersionedSourceClient<T>,
    /// State's authoritative persona-memory journal client, and the
    /// deployment namespace it is asked under.
    ///
    /// Unlike a Versioned scope, a persona-memory partition carries no
    /// namespace of its own — the client's `read_history` call takes it
    /// separately — so this deployment's own namespace travels alongside the
    /// client rather than inside each request's source.
    persona_memory: polyc_state_connect::persona_memory_journal::PersonaMemoryJournalClient<T>,
    /// State's observation authority client (7-O).
    observation: polyc_state_connect::observation::ObservationClient<T>,
    namespace: String,
}

impl<T> ConnectCoreMetadata<T> {
    #[allow(clippy::too_many_arguments)]
    pub(crate) const fn new(
        journal: JournalClient<T>,
        projections: ProjectionCatalogClient<T>,
        audit: QueryAuditClient<T>,
        versioned: polyc_state_connect::versioned_source::VersionedSourceClient<T>,
        persona_memory: polyc_state_connect::persona_memory_journal::PersonaMemoryJournalClient<T>,
        observation: polyc_state_connect::observation::ObservationClient<T>,
        namespace: String,
    ) -> Self {
        Self {
            journal,
            projections,
            audit,
            versioned,
            persona_memory,
            observation,
            namespace,
        }
    }
}

#[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 versioned_source_head(
        &self,
        operation: &CoreOperationContext,
        scope: &polyc_state::command::CommandScope,
    ) -> Result<polyc_state::versioned::VersionedSourceHead, CoreResolutionError> {
        Ok(self
            .versioned
            .source_head(&operation.declared()?, scope)
            .await?)
    }

    async fn persona_memory_source_head(
        &self,
        operation: &CoreOperationContext,
        partition: &polyc_state::persona_memory::journal::MemoryJournalPartition,
    ) -> Result<polyc_state::persona_memory::journal::PersonaMemorySourceHead, CoreResolutionError>
    {
        let page = self
            .persona_memory
            .read_history(
                &operation.declared()?,
                &self.namespace,
                partition,
                None,
                polyc_state::persona_memory::journal::MemoryReplayRange::new(
                    0,
                    1,
                    polyc_state::persona_memory::journal::MAX_PAYLOAD_BYTES,
                ),
            )
            .await?;
        Ok(
            polyc_state::persona_memory::journal::PersonaMemorySourceHead::new(
                page.incarnation(),
                page.head(),
            ),
        )
    }

    async fn persona_memory_directory_page(
        &self,
        operation: &CoreOperationContext,
        after: Option<&str>,
        limit: u32,
    ) -> Result<polyc_state::persona_memory::journal::MemoryLineagePage, CoreResolutionError> {
        Ok(self
            .persona_memory
            .list_lineage(&operation.declared()?, &self.namespace, after, limit)
            .await?)
    }

    async fn create_versioned_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        family: &str,
    ) -> Result<polyc_state::versioned::VersionedDirectorySnapshot, CoreResolutionError> {
        Ok(self
            .versioned
            .create_directory_snapshot(&operation.declared()?, family)
            .await?)
    }

    async fn versioned_directory_page(
        &self,
        operation: &CoreOperationContext,
        request: &polyc_state::versioned::ListVersionedDirectorySnapshot,
    ) -> Result<polyc_state::versioned::VersionedDirectoryPage, CoreResolutionError> {
        Ok(self
            .versioned
            .directory_page(&operation.declared()?, request)
            .await?)
    }

    async fn release_versioned_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        snapshot: &polyc_state::versioned::VersionedDirectorySnapshotId,
    ) -> Result<(), CoreResolutionError> {
        Ok(self
            .versioned
            .release_directory_snapshot(&operation.declared()?, snapshot)
            .await?)
    }

    async fn observed_head(
        &self,
        operation: &CoreOperationContext,
        collection: &polyc_state::observation::CollectionId,
    ) -> Result<Option<polyc_state::observation::ObservationHead>, CoreResolutionError> {
        Ok(self
            .observation
            .latest(&operation.declared()?, collection)
            .await?)
    }

    async fn observed_collections(
        &self,
        operation: &CoreOperationContext,
        kind: polyc_state::observation::CollectionKind,
    ) -> Result<polyc_state::observation::ObservedCollectionListing, CoreResolutionError> {
        Ok(self
            .observation
            .collections(&operation.declared()?, kind)
            .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 {
    readers: BTreeMap<ObjectNamespace, (GcsReadClient, AtRestProtection)>,
}

impl GcsExactBackend {
    fn try_new<I>(sources: I) -> Result<Self, CoreExecutionError>
    where
        I: IntoIterator<Item = (GcsReadClient, Vec<GcsReadNamespace>)>,
    {
        let mut source_count = 0_usize;
        let mut readers = BTreeMap::new();
        for (client, profiles) in sources {
            for profile in profiles {
                source_count = source_count.saturating_add(1);
                readers.insert(
                    profile.namespace.clone(),
                    (client.clone(), profile.protection),
                );
            }
        }
        if source_count == 0 || readers.len() != source_count {
            return Err(CoreExecutionError::InvalidComposition(
                "an artifact realm has empty or duplicate namespace protection",
            ));
        }
        Ok(Self { readers })
    }

    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.readers
            .get(namespace)
            .map_or(AtRestProtection::None, |(_, protection)| *protection)
    }

    fn head_exact(
        &self,
        reference: &ExactObjectRef,
    ) -> ArtifactReadFuture<'_, ExactObjectMetadata> {
        let name = Self::object_name(reference);
        let generation = reference.generation();
        let client = self
            .readers
            .get(reference.namespace())
            .map(|(client, _)| client.clone());
        Box::pin(async move {
            let client = client.ok_or_else(|| ArtifactReadError::Refused {
                reason: format!("{name} names an unconfigured namespace"),
            })?;
            let address = i64::try_from(generation).map_err(|_| ArtifactReadError::Refused {
                reason: format!("{name} names a generation GCS cannot address"),
            })?;
            match 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();
        let client = self
            .readers
            .get(reference.namespace())
            .map(|(client, _)| client.clone());
        Box::pin(async move {
            let client = client.ok_or_else(|| ArtifactReadError::Refused {
                reason: format!("{name} names an unconfigured namespace"),
            })?;
            let address = i64::try_from(generation).map_err(|_| ArtifactReadError::Refused {
                reason: format!("{name} names a generation GCS cannot address"),
            })?;
            let served = 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}"))
            ),
        },
    }
}

#[derive(Clone)]
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 }
    }
}

#[derive(Clone)]
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 visible_namespace_set(sources: &[VisibleGcsSource]) -> BTreeSet<ObjectNamespace> {
    sources
        .iter()
        .flat_map(|source| &source.profiles)
        .map(|profile| profile.namespace.clone())
        .collect()
}

fn fleet_namespace_set(sources: &[FleetGcsSource]) -> BTreeSet<ObjectNamespace> {
    sources
        .iter()
        .flat_map(|source| &source.profiles)
        .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: Vec<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 = visible_namespace_set(&visible);
    let topology = RealmTopology::try_new(visible_namespaces.clone(), fleet_namespaces)?;
    let backend = GcsExactBackend::try_new(
        visible
            .into_iter()
            .map(|source| (source.client, source.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: Vec<VisibleGcsSource>,
    fleet: Vec<FleetGcsSource>,
    trust: Arc<dyn ManifestTrust>,
    scope: Arc<dyn CurrentCredentialAuthority>,
    revalidation_interval: Duration,
    admission: Arc<CoreExecutionAdmission>,
) -> Result<CoreArtifactAuthority, CoreExecutionError> {
    let visible_namespaces = visible_namespace_set(&visible);
    let fleet_namespaces = fleet_namespace_set(&fleet);
    let topology = RealmTopology::try_new(visible_namespaces.clone(), fleet_namespaces.clone())?;
    let visible_backend = GcsExactBackend::try_new(
        visible
            .into_iter()
            .map(|source| (source.client, source.profiles)),
    )?;
    let fleet_backend = GcsExactBackend::try_new(
        fleet
            .into_iter()
            .map(|source| (source.client, source.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;