polyc-query 2026.9.0

//! The one credential verification and scoping mechanism.
//!
//! Two entry paths need this: the embedded [`QueryAuthority`] funnel that
//! control-plane callers already reach, and the standalone Query service that
//! verifies a presented credential at its own boundary. They must not be two
//! implementations. A change that landed on one and missed the other would be
//! an authorization difference reachable from exactly one surface, which is
//! the hardest kind to notice in review.
//!
//! So both hold this component and neither reimplements it. It owns the
//! bearer session check, the `ExplorerRead` scope gate, the fresh active-
//! persona read, the admin and persona role split, the conversation-grant
//! signature, kind, and expiry checks, the fresh participation lookup, and the
//! derivation of a scope from each verified principal. It mints no public
//! constructor for [`Principal`], [`QueryScope`], or a scoping, so a trusted
//! identifier still cannot become authority.
//!
//! [`QueryAuthority`]: crate::authority::QueryAuthority

use std::sync::Arc;

use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use polyc_crypto::session;
use polyc_crypto::signing_role::{RoleTrustSet, SigningRole as _, TurnReadRole};
use polyc_persona::ScopeResolution;

use crate::authority::{
    AdminPrincipal, ConversationGrantPrincipal, GRANT_KIND, GrantClaims, PersonaAccess,
    PersonaPrincipal, Principal, PrincipalError, SEARCH_SCOPE_CAP, Scoping, SearchScope,
    SearchScopeError, canonical_search_scope,
};
use crate::session::QueryScope;

/// Verifies a presented credential and derives exactly what it authorizes.
///
/// `Debug` reports the composition only. The trust sets hold public keys and
/// the persona handle reaches tenant identities, so neither belongs in a log
/// line.
/// The one session capability this mechanism needs.
///
/// Verification, and nothing else. The complete browser-session authority also
/// mints and revokes, and a read-only plane must hold neither. Narrowing the
/// port is what makes that structural: a composition built through
/// `CredentialAuthority::verifying` holds a value with no mint method and no
/// revoke method, so no future edit inside this crate can reach one.
#[async_trait::async_trait]
pub trait SessionVerification: Send + Sync {
    /// Verifies a signed bearer and current-reads its durable authorization.
    ///
    /// # Errors
    ///
    /// Returns the session authority's own refusal for an invalid, expired,
    /// revoked, or unreadable session.
    async fn verify_bearer(
        &self,
        token: &str,
        now_ms: u64,
    ) -> Result<
        polyc_crypto::session::AuthorizedSessionClaims,
        polyc_session_family::authority::SessionAuthorityError,
    >;
}

/// Narrows a complete browser-session authority to verification.
///
/// Control holds the complete authority and composes through
/// [`CredentialAuthority::current`]. This wrapper is how that one capability
/// reaches the mechanism without the mechanism itself holding the rest.
struct FullAuthority(Arc<dyn polyc_session_family::authority::BrowserSessionAuthority>);

#[async_trait::async_trait]
impl SessionVerification for FullAuthority {
    async fn verify_bearer(
        &self,
        token: &str,
        now_ms: u64,
    ) -> Result<
        polyc_crypto::session::AuthorizedSessionClaims,
        polyc_session_family::authority::SessionAuthorityError,
    > {
        self.0.verify_bearer(token, now_ms).await
    }
}

pub(crate) struct CredentialAuthority {
    persona: PersonaAccess,
    bearer_authority: Option<Arc<dyn SessionVerification>>,
    turn_read_trust: RoleTrustSet<TurnReadRole>,
    #[cfg(any(test, feature = "test-util"))]
    legacy_revoked: Option<Arc<polyc_crypto::session::RevokedTokens>>,
    #[cfg(any(test, feature = "test-util"))]
    legacy_session_trust: Option<RoleTrustSet<polyc_crypto::signing_role::SessionRole>>,
    #[cfg(test)]
    test_search_scope_cap: std::sync::Mutex<Option<usize>>,
}

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

impl CredentialAuthority {
    /// Builds the production component: durable bearer authorization, no
    /// process-local revocation fallback.
    pub(crate) fn current(
        persona: Arc<dyn crate::authority::PersonaSource>,
        turn_read_trust: RoleTrustSet<TurnReadRole>,
        authority: Arc<dyn polyc_session_family::authority::BrowserSessionAuthority>,
    ) -> Self {
        Self::verifying(persona, turn_read_trust, Arc::new(FullAuthority(authority)))
    }

    /// Builds the mechanism over verification alone.
    ///
    /// A composition that can only verify cannot mint a session and cannot
    /// revoke one. That is not a claim in a comment: the type it holds has no
    /// such method. A read-only plane composes through here.
    pub(crate) fn verifying(
        persona: Arc<dyn crate::authority::PersonaSource>,
        turn_read_trust: RoleTrustSet<TurnReadRole>,
        sessions: Arc<dyn SessionVerification>,
    ) -> Self {
        Self {
            persona: PersonaAccess::Current(persona),
            bearer_authority: Some(sessions),
            turn_read_trust,
            #[cfg(any(test, feature = "test-util"))]
            legacy_revoked: None,
            #[cfg(any(test, feature = "test-util"))]
            legacy_session_trust: None,
            #[cfg(test)]
            test_search_scope_cap: std::sync::Mutex::new(None),
        }
    }

    /// Builds the retired fixture component.
    #[cfg(any(test, feature = "test-util"))]
    pub(crate) fn legacy(
        persona: crate::authority::PersonaCell,
        revoked: Arc<polyc_crypto::session::RevokedTokens>,
        turn_read_trust: RoleTrustSet<TurnReadRole>,
        legacy_session_trust: RoleTrustSet<polyc_crypto::signing_role::SessionRole>,
    ) -> Self {
        Self {
            persona: PersonaAccess::Legacy(persona),
            bearer_authority: None,
            turn_read_trust,
            legacy_revoked: Some(revoked),
            legacy_session_trust: Some(legacy_session_trust),
            #[cfg(test)]
            test_search_scope_cap: std::sync::Mutex::new(None),
        }
    }

    /// Swaps the turn-read trust set in place for a rotation fixture.
    #[cfg(test)]
    pub(crate) fn replace_turn_read_trust_for_test(&mut self, trust: RoleTrustSet<TurnReadRole>) {
        self.turn_read_trust = trust;
    }

    #[cfg(test)]
    pub(crate) fn set_test_search_scope_cap(&self, cap: usize) {
        *self.test_search_scope_cap.lock().expect("poison") = Some(cap);
    }

    /// Verify an explorer session token (from either the `pc_explorer_session`
    /// cookie or an `Authorization: Bearer` header — the caller extracts the
    /// string, this method only verifies it) and resolve the FRESH admin flag
    /// for the bound persona. Mints [`Principal::Admin`] when that persona
    /// carries the durable admin attribute, [`Principal::Persona`] (A3) for
    /// any other valid session whose persona resolves. See the module doc's
    /// "Fail-closed admin resolution" section for exactly which failure maps
    /// to which [`PrincipalError`] variant.
    ///
    /// # Errors
    ///
    /// See [`PrincipalError`]'s variant docs.
    pub(crate) async fn verify_admin_session(
        &self,
        token: &str,
        now_ms: u64,
    ) -> Result<Principal, PrincipalError> {
        let claims = if let Some(authority) = &self.bearer_authority {
            authority
                .verify_bearer(token, now_ms)
                .await
                .map_err(|error| match error {
                    polyc_session_family::authority::SessionAuthorityError::Invalid => {
                        PrincipalError::InvalidSession
                    }
                    _ => PrincipalError::StoreUnavailable,
                })?
        } else {
            #[cfg(not(any(test, feature = "test-util")))]
            return Err(PrincipalError::StoreUnavailable);

            #[cfg(any(test, feature = "test-util"))]
            {
                let legacy = session::verify_session_with_trust(
                    self.legacy_session_trust
                        .as_ref()
                        .ok_or(PrincipalError::StoreUnavailable)?,
                    token,
                    now_ms,
                    self.legacy_revoked
                        .as_ref()
                        .ok_or(PrincipalError::StoreUnavailable)?,
                )
                .ok_or(PrincipalError::InvalidSession)?;
                polyc_crypto::session::AuthorizedSessionClaims {
                    issuer: legacy.issuer,
                    key_id: legacy.key_id,
                    session_id: "legacy-test-session".to_owned(),
                    authorization_epoch: 0,
                    subject: legacy.subject,
                    scopes: legacy.scopes,
                    issued_ms: legacy.issued_ms,
                    expires_ms: legacy.expires_ms,
                }
            }
        };

        // Mirrors `crate::forensics::resolve_explorer_caller`'s own gate: a
        // session must carry BOTH a resolved persona AND the `ExplorerRead`
        // scope to admit here — a wallet-rooted session with no linked
        // persona, or one that only ever carries `WalletManage`, is treated
        // exactly like no session at all, never as a fleet-wide admin.
        if !claims.has_scope(session::SessionScope::ExplorerRead) {
            return Err(PrincipalError::InvalidSession);
        }
        let persona_id = claims
            .subject
            .persona_id()
            .ok_or(PrincipalError::InvalidSession)?
            .to_owned();

        let Some(persona) = self.persona.load_full() else {
            return Err(PrincipalError::StoreUnavailable);
        };
        match persona.active_persona(persona_id).await {
            Ok(Some(active)) => {
                // `active.persona_id`, not the id the token named: the
                // resolver has already traversed any merge alias, so a
                // session minted before an absorption scopes to the SURVIVING
                // persona rather than to an id nothing is filed under.
                let persona_id = active.persona_id;
                if active.admin {
                    Ok(Principal::Admin(AdminPrincipal::minted(persona_id)))
                } else {
                    // A3: a valid session bound to a non-admin persona mints
                    // a persona-scoped principal instead of a hard refusal —
                    // see `PersonaPrincipal`'s doc for what it can query.
                    Ok(Principal::Persona(PersonaPrincipal::minted(persona_id)))
                }
            }
            // The session's own persona id names nobody who may act — either
            // no profile at all, or one an admin has de-admitted. There is no
            // persona left to mint ANY principal for, and the two cases are
            // deliberately indistinguishable to the caller.
            Ok(None) => Err(PrincipalError::NotAuthorizedForFleet),
            // A store READ error says nothing about whether the caller is an
            // admin — treated as infrastructure-unavailable, never folded
            // into "not admin" (see the module doc for why this differs from
            // `crate::forensics::persona_is_admin`'s stricter fold).
            Err(_store_error) => Err(PrincipalError::StoreUnavailable),
        }
    }

    /// Verify a conversation-grant token minted by [`mint_conversation_grant`]
    /// against this authority's own signer public key and `now_unix_ms`. The
    /// untrusted claims are decoded only to select their role-scoped key;
    /// no authorization field, kind, or expiry is trusted until that key has
    /// verified the exact decoded bytes.
    ///
    /// Deliberately returns one undifferentiated [`PrincipalError::InvalidGrant`]
    /// for every OTHER failure mode (malformed, bad signature, wrong `kind`)
    /// — there is nothing diagnostic here for a probing caller to learn; the
    /// specific reason is available to the caller only via the
    /// `tracing::warn!` this method emits. Expiry alone is carved out as
    /// [`PrincipalError::GrantExpired`] (a DISTINCT, still-opaque-to-forgery
    /// outcome) precisely so a legitimate caller whose grant simply outlived
    /// its TTL can silently re-mint and retry once, rather than being told
    /// nothing beyond "invalid" the way an actual forgery is.
    ///
    /// # Errors
    ///
    /// Returns [`PrincipalError::InvalidGrant`] if the token is malformed, its
    /// signature does not verify, or its `kind` tag does not match; returns
    /// [`PrincipalError::GrantExpired`] if the token verifies but its TTL has
    /// elapsed.
    pub(crate) fn verify_conversation_grant(
        &self,
        token: &str,
        now_unix_ms: u64,
    ) -> Result<Principal, PrincipalError> {
        let (claims_b64, sig_b64) = token.split_once('.').ok_or_else(|| {
            tracing::warn!("conversation grant token malformed: no `.` separator");
            PrincipalError::InvalidGrant
        })?;
        let canonical = URL_SAFE_NO_PAD.decode(claims_b64).map_err(|_| {
            tracing::warn!("conversation grant token malformed: claims segment not base64");
            PrincipalError::InvalidGrant
        })?;
        let signature = URL_SAFE_NO_PAD.decode(sig_b64).map_err(|_| {
            tracing::warn!("conversation grant token malformed: signature segment not base64");
            PrincipalError::InvalidGrant
        })?;
        let claims: GrantClaims = serde_json::from_slice(&canonical).map_err(|_| {
            tracing::warn!("conversation grant token malformed: claims did not decode as JSON");
            PrincipalError::InvalidGrant
        })?;
        if claims.issuer != TurnReadRole::ISSUER
            || !self.turn_read_trust.verify_turn_read_capability(
                &claims.key_id,
                &canonical,
                &signature,
            )
        {
            tracing::warn!("conversation grant token signature invalid");
            return Err(PrincipalError::InvalidGrant);
        }
        if claims.kind != GRANT_KIND {
            tracing::warn!(kind = %claims.kind, "conversation grant token kind tag mismatch");
            return Err(PrincipalError::InvalidGrant);
        }
        if now_unix_ms > claims.expires_at_ms {
            tracing::warn!("conversation grant token expired");
            return Err(PrincipalError::GrantExpired);
        }
        Ok(Principal::ConversationGrant(
            ConversationGrantPrincipal::minted(claims.conversation_id, claims.subject),
        ))
    }

    /// The one query API: scope a session to `principal`'s own authority.
    ///
    /// [`Principal::Admin`] scopes fleet-wide (every conversation partition);
    /// [`Principal::ConversationGrant`] scopes to exactly that grant's own
    /// conversation; [`Principal::Persona`] (A3) scopes to exactly that
    /// persona's own participated conversations, resolved FRESH per call via
    /// [`polyc_persona::PersonaHost::participations`] — never cached, never a
    /// replay. A persona with zero participations still scopes successfully,
    /// to an empty catalog (see [`PersonaPrincipal`]'s doc for the full
    /// boundary conditions). Every non-Fleet scope — conversation grant and
    /// persona alike — reuses the identical redacted registration path
    /// (`allow_explain = false`, no `events_raw`, redacted attribution
    /// identity columns, no `personas`/`participations` reference tables).
    ///
    /// # Errors
    ///
    /// Returns [`PrincipalError::StoreUnavailable`] if a [`Principal::Persona`]'s
    /// participation set cannot be resolved — the persona-store cell is
    /// empty, or the store itself errors reading the index. This mirrors
    /// [`QueryAuthority::verify_admin_session`]'s own posture for its
    /// admin-flag read: infrastructure unavailability, never an
    /// authorization verdict. Infallible for
    /// [`Principal::Admin`]/[`Principal::ConversationGrant`].
    pub(crate) async fn scoping_for(
        &self,
        principal: &Principal,
    ) -> Result<Scoping, PrincipalError> {
        let scoping = match principal {
            Principal::Admin(admin) => Scoping {
                scope: QueryScope::Fleet,
                allow_explain: true,
                caller_identity: Some(admin.persona_id().to_owned()),
                conversation_id: None,
                turn_id: None,
                web_session_id: None,
            },
            Principal::ConversationGrant(grant) => {
                Scoping::for_conversation(grant.conversation_id(), grant.subject())
            }
            Principal::Persona(persona) => {
                let Some(persona_host) = self.persona.load_full() else {
                    return Err(PrincipalError::StoreUnavailable);
                };
                let participations = persona_host
                    .participations(persona.persona_id().to_owned())
                    .await
                    .map_err(|_store_error| PrincipalError::StoreUnavailable)?;
                let conversation_ids = participations
                    .into_iter()
                    .map(|participation| participation.conversation_id)
                    .collect();
                Scoping {
                    scope: QueryScope::Conversations(conversation_ids),
                    allow_explain: false,
                    caller_identity: Some(persona.persona_id().to_owned()),
                    conversation_id: None,
                    turn_id: None,
                    web_session_id: None,
                }
            }
        };
        Ok(scoping)
    }

    /// Scope a session to `persona_id`'s own participated conversations,
    /// ignoring admin status entirely — the "my own rows" scope no
    /// `QueryScope` variant expresses on its own.
    ///
    /// A sibling to [`Self::scope_for`], not a call through it: `scope_for`
    /// maps [`Principal::Admin`] to the fleet scope unconditionally, so
    /// routing an administrator's own-rows read through it would still hand
    /// back the whole deployment. This method takes a bare persona id rather
    /// than a verified [`Principal`], because its caller has already
    /// authenticated the session by another means (a bearer cookie verified
    /// against the explorer session authority, e.g. `ExplorerCaller`) and is
    /// asking for exactly that persona's own rows regardless of what else
    /// that session may be authorized to see. Admin status must never widen
    /// this scope, by construction rather than by convention: this method
    /// never reads it at all.
    ///
    /// # Errors
    ///
    /// Returns [`PrincipalError::StoreUnavailable`] if `persona_id`'s
    /// participation set cannot be resolved — the persona-store cell is
    /// empty, or the store itself errors reading the index. The same
    /// infrastructure-unavailable posture [`Self::scope_for`] gives its own
    /// [`Principal::Persona`] arm.
    pub(crate) async fn own_rows_scoping(
        &self,
        persona_id: &str,
    ) -> Result<Scoping, PrincipalError> {
        let Some(persona_host) = self.persona.load_full() else {
            return Err(PrincipalError::StoreUnavailable);
        };
        let participations = persona_host
            .participations(persona_id.to_owned())
            .await
            .map_err(|_store_error| PrincipalError::StoreUnavailable)?;
        let conversation_ids = participations
            .into_iter()
            .map(|participation| participation.conversation_id)
            .collect();
        let scoping = Scoping {
            scope: QueryScope::Conversations(conversation_ids),
            allow_explain: false,
            caller_identity: Some(persona_id.to_owned()),
            conversation_id: None,
            turn_id: None,
            web_session_id: None,
        };
        Ok(scoping)
    }

    /// Resolve `principal_ref`'s trusted PARTICIPATION search scope for one
    /// turn — the set of PREVIOUS conversations the participation-scoped
    /// search surface may read. This is the read authority registered as
    /// DF-23 in `docs/proposals/separated-planes-authority-registry.md`, never
    /// the calling conversation itself (that stays
    /// `conversation_find`'s job, reached through
    /// [`QueryAuthority::scope_for_turn`] instead).
    ///
    /// The trust contract is IDENTICAL to [`QueryAuthority::scope_for_turn`]'s
    /// own (see that method's doc): `principal_ref`, `conversation_id`, and
    /// `turn_id` are trusted-side values taken from a turn's own dispatch
    /// attribution — never a wire field, a tool argument, or anything a model
    /// can choose. A surface that cannot say that about all three has no
    /// business calling this.
    ///
    /// Runs, in order:
    ///
    /// 1. Verifies `principal_ref` is an ACTIVE persona via
    ///    [`polyc_persona::PersonaHost::active_persona`]. A `None` verdict
    ///    refuses ([`SearchScopeError::PersonaNotActive`]); a store error
    ///    refuses as infrastructure-unavailable
    ///    ([`SearchScopeError::StoreUnavailable`]) — never folded into "not
    ///    active" (see that error variant's own doc for why the distinction
    ///    matters).
    /// 2. Resolves the BOUNDED participation scope via
    ///    [`polyc_persona::PersonaHost::participation_scope`], which already
    ///    applies visibility tombstones and refuses over cap BEFORE any
    ///    per-conversation read runs. This method reimplements none of
    ///    that — an over-cap resolution surfaces directly as
    ///    [`SearchScopeError::OverCap`].
    /// 3. Removes `conversation_id` — the CALLING conversation — from the
    ///    resolved set. Participation-scoped search covers previous
    ///    conversations only.
    /// 4. Canonicalizes and hashes what remains — see [`SearchScope::hash`]'s
    ///    own doc for the exact five-step algorithm.
    ///
    /// Returns a [`SearchScope`], never a [`Principal`] — see this module's
    /// doc for why handing out a [`Principal`] here would crack the
    /// no-public-constructor seal on it.
    ///
    /// # Errors
    ///
    /// See [`SearchScopeError`]'s variant docs.
    ///
    /// # Panics
    ///
    /// Never panics in a production build. Under `#[cfg(test)]` only, this
    /// reads the test-only `SEARCH_SCOPE_CAP` override through a `Mutex` and
    /// panics if that mutex is poisoned (a prior test thread inside this same
    /// process panicked while holding it) — the same poisoning posture this
    /// crate's own `ScopedQuery::race_inject_after_count_read` test hook
    /// already takes.
    pub(crate) async fn resolve_search_scope(
        &self,
        principal_ref: &str,
        conversation_id: &str,
        turn_id: &str,
    ) -> Result<SearchScope, SearchScopeError> {
        let Some(persona_host) = self.persona.load_full() else {
            return Err(SearchScopeError::StoreUnavailable);
        };

        match persona_host.active_persona(principal_ref.to_owned()).await {
            Ok(Some(_active)) => {}
            Ok(None) => {
                tracing::info!(
                    persona_id = %principal_ref,
                    conversation_id = %conversation_id,
                    turn_id = %turn_id,
                    "refusing search-scope resolution: persona is not active"
                );
                return Err(SearchScopeError::PersonaNotActive);
            }
            Err(_store_error) => {
                return Err(SearchScopeError::StoreUnavailable);
            }
        }

        #[cfg(test)]
        let cap = self
            .test_search_scope_cap
            .lock()
            .expect("poison")
            .unwrap_or(SEARCH_SCOPE_CAP);
        #[cfg(not(test))]
        let cap = SEARCH_SCOPE_CAP;

        let resolution = persona_host
            .participation_scope(principal_ref.to_owned(), cap)
            .await
            .map_err(|_store_error| SearchScopeError::StoreUnavailable)?;

        let mut conversation_ids = match resolution {
            ScopeResolution::RefusedOverCap { count } => {
                tracing::warn!(
                    persona_id = %principal_ref,
                    conversation_id = %conversation_id,
                    turn_id = %turn_id,
                    count,
                    cap,
                    "refusing search-scope resolution: participation count exceeds the search cap"
                );
                return Err(SearchScopeError::OverCap { count });
            }
            ScopeResolution::Resolved { conversation_ids } => conversation_ids,
        };

        // Participation-scoped search covers PREVIOUS conversations only —
        // the calling conversation is `conversation_find`'s job (design doc,
        // "Trusted search authority").
        conversation_ids.retain(|id| id != conversation_id);

        let (conversation_ids, hash) = canonical_search_scope(conversation_ids);
        Ok(SearchScope::minted(conversation_ids, hash))
    }
}

// ---------------------------------------------------------------------------
// The retained credential witness
// ---------------------------------------------------------------------------

/// A wall clock, injected so expiry is testable without waiting.
pub(crate) trait UnixClock: Send + Sync {
    /// Returns the current Unix time in milliseconds.
    fn now_unix_ms(&self) -> u64;
}

/// The production clock.
pub(crate) struct SystemUnixClock;

impl UnixClock for SystemUnixClock {
    fn now_unix_ms(&self) -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |elapsed| {
                u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
            })
    }
}

/// The credential a caller presented, retained privately for re-verification.
///
/// Both kinds are re-verified from the raw token, because that is what the
/// verification actually needs: a bearer's signature is over its own bytes and
/// its session id keys a fresh durable read, and a grant's signature is over
/// the canonical claims. A remembered persona id or conversation id would let
/// a revoked or expired credential keep releasing rows.
pub(crate) enum PresentedCredential {
    /// An explorer bearer session.
    Bearer(String),
    /// A signed conversation grant.
    ConversationGrant(String),
}

impl std::fmt::Debug for PresentedCredential {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(match self {
            Self::Bearer(_) => "bearer",
            Self::ConversationGrant(_) => "conversation-grant",
        })
    }
}

/// Re-proves a presented credential for as long as a stream releases rows.
///
/// Admission derives the authorized scope once. This witness re-runs the same
/// checks through the same shared mechanism before the first row and at the
/// bounded interval, and refuses to widen: the current scope must still
/// contain the admitted one. A narrowing, an expiry, a revocation, a destroyed
/// persona, a lost admin role, or a State outage all stop release.
pub(crate) struct CredentialWitness {
    credential: PresentedCredential,
    authority: Arc<CredentialAuthority>,
    clock: Arc<dyn UnixClock>,
}

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

impl CredentialWitness {
    /// Verifies the presented credential and retains it for revalidation.
    ///
    /// # Errors
    ///
    /// Returns whatever the shared mechanism returns for an invalid, expired,
    /// revoked, unknown, or unresolvable credential.
    /// Admits whichever credential one bearer value holds.
    ///
    /// The transport carries one header, so the kind is decided here, once.
    /// The grant path runs first because it is purely cryptographic: a grant
    /// carries the turn-read role's own signature, which a session token
    /// cannot produce, so a session token always falls through. An expired
    /// grant is a real grant and keeps its own refusal rather than being
    /// retried as a session.
    ///
    /// The resolved kind is what the witness retains, so every later
    /// revalidation re-runs the same path. Re-deciding the kind on each
    /// revalidation would let a token that verified one way keep a stream
    /// alive by verifying the other.
    ///
    /// # Errors
    ///
    /// Returns the grant refusal for a grant-shaped token, and the session
    /// refusal otherwise.
    pub(crate) async fn admit_bearer(
        token: String,
        authority: Arc<CredentialAuthority>,
        clock: Arc<dyn UnixClock>,
    ) -> Result<(Self, Scoping), PrincipalError> {
        let now = clock.now_unix_ms();
        match authority.verify_conversation_grant(&token, now) {
            Ok(_grant) => {
                Self::admit(
                    PresentedCredential::ConversationGrant(token),
                    authority,
                    clock,
                )
                .await
            }
            Err(PrincipalError::GrantExpired) => Err(PrincipalError::GrantExpired),
            Err(_not_a_grant) => {
                Self::admit(PresentedCredential::Bearer(token), authority, clock).await
            }
        }
    }

    pub(crate) async fn admit(
        credential: PresentedCredential,
        authority: Arc<CredentialAuthority>,
        clock: Arc<dyn UnixClock>,
    ) -> Result<(Self, Scoping), PrincipalError> {
        let scoping = Self::verify(&credential, &authority, clock.as_ref()).await?;
        Ok((
            Self {
                credential,
                authority,
                clock,
            },
            scoping,
        ))
    }

    async fn verify(
        credential: &PresentedCredential,
        authority: &CredentialAuthority,
        clock: &dyn UnixClock,
    ) -> Result<Scoping, PrincipalError> {
        let now = clock.now_unix_ms();
        let principal = match credential {
            PresentedCredential::Bearer(token) => {
                authority.verify_admin_session(token, now).await?
            }
            PresentedCredential::ConversationGrant(token) => {
                authority.verify_conversation_grant(token, now)?
            }
        };
        authority.scoping_for(&principal).await
    }

    /// Re-proves the credential and returns the scope it authorizes now.
    ///
    /// # Errors
    ///
    /// Returns the shared mechanism's refusal. A caller treats any refusal as
    /// a reason to stop releasing rows.
    pub(crate) async fn current_scope(&self) -> Result<QueryScope, PrincipalError> {
        let scoping = Self::verify(&self.credential, &self.authority, self.clock.as_ref()).await?;
        Ok(scoping.scope)
    }
}

#[async_trait::async_trait]
impl crate::core_execution::CoreScopeRevalidator for CredentialWitness {
    async fn current_scope(
        &self,
        operation: &crate::core_resolution::CoreOperationContext,
    ) -> Result<QueryScope, crate::core_resolution::CoreResolutionError> {
        operation.check()?;
        Self::current_scope(self)
            .await
            .map_err(|_| crate::core_resolution::CoreResolutionError::InvalidAttribution)
    }
}

// A credential witness re-proves the credential itself, so it may carry the
// production seal. Nothing else in this crate may.
impl crate::core_execution::sealed::CredentialProven for CredentialWitness {}