polyc-query-credential 2026.10.1

//! The one credential verification and scoping mechanism.
//!
//! Two entry paths need this: `polyc-control-plane`'s forensics
//! conversation-read funnel (which needs an authorization verdict, never a
//! runnable engine session) and the standalone Query service, which verifies
//! a presented credential at its own boundary. The embedded `QueryAuthority`
//! funnel this module once also served is deleted (POLY-196). The remaining
//! paths must not be separate 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 [`crate::principal::Principal`],
//! [`crate::session::QueryScope`], or a [`crate::principal::Scoping`], so a
//! trusted identifier still cannot become authority.

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::principal::{
    AdminFleetPrincipal, AdminPrincipal, ControlFleetPrincipal, ConversationGrantPrincipal,
    PersonaAccess, PersonaPrincipal, Principal, PrincipalError, SEARCH_SCOPE_CAP, Scoping,
    SearchScope, SearchScopeError, canonical_search_scope,
};
use crate::session::{MemorySources, QueryScope};
use polyc_query_model::{GRANT_KIND, GrantClaims, GrantSubject};

/// 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
    }
}

/// The one credential verification and scoping mechanism — see this module's
/// doc.
pub 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(any(test, feature = "test-util"))]
    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 {
    async fn verified_memory_participants(
        &self,
        conversation_id: &str,
        proposed_sources: &[String],
    ) -> Result<Vec<String>, PrincipalError> {
        if proposed_sources.is_empty() {
            return Ok(Vec::new());
        }
        let Some(persona_host) = self.persona.load_full() else {
            return Err(PrincipalError::StoreUnavailable);
        };
        let mut participants = Vec::with_capacity(proposed_sources.len());
        for proposed in proposed_sources {
            let active = persona_host
                .active_persona(proposed.clone())
                .await
                .map_err(|_store_error| PrincipalError::StoreUnavailable)?
                .ok_or(PrincipalError::SourceOutsideScope)?;
            let participations = persona_host
                .participations(active.persona_id.clone())
                .await
                .map_err(|_store_error| PrincipalError::StoreUnavailable)?;
            if !participations
                .iter()
                .any(|participation| participation.conversation_id == conversation_id)
            {
                return Err(PrincipalError::SourceOutsideScope);
            }
            participants.push(active.persona_id);
        }
        participants.sort();
        participants.dedup();
        Ok(participants)
    }

    /// Extends a verified composite-trace scope with the personas read from
    /// that same grant's pinned trace result.
    ///
    /// The list is still only a proposal: every source is resolved through
    /// the persona authority and must participate in the granted
    /// conversation. Any one mismatch refuses the complete scope.
    ///
    /// # Errors
    ///
    /// Returns [`PrincipalError::SourceOutsideScope`] when `scoping` carries
    /// no composite-trace capability, or when a proposed source cannot be
    /// verified. Returns [`PrincipalError::StoreUnavailable`] when the
    /// persona store cannot answer.
    pub async fn composite_trace_memory_scope(
        &self,
        scoping: &Scoping,
        proposed_sources: &[String],
    ) -> Result<QueryScope, PrincipalError> {
        if scoping.composite_trace().is_none() {
            return Err(PrincipalError::SourceOutsideScope);
        }
        let conversation_id = scoping
            .conversation_id()
            .ok_or(PrincipalError::SourceOutsideScope)?;
        let owner = scoping
            .caller_identity()
            .ok_or(PrincipalError::SourceOutsideScope)?;
        let proposed = proposed_sources
            .iter()
            .filter(|source| source.as_str() != owner)
            .cloned()
            .collect::<Vec<_>>();
        let participants = self
            .verified_memory_participants(conversation_id, &proposed)
            .await?;
        let QueryScope::Conversations { conversations, .. } = scoping.scope() else {
            return Err(PrincipalError::SourceOutsideScope);
        };
        Ok(QueryScope::Conversations {
            conversations: conversations.clone(),
            memory: MemorySources {
                owner: Some(owner.to_owned()),
                participants,
            },
        })
    }

    /// Builds the production component: durable bearer authorization, no
    /// process-local revocation fallback.
    #[must_use]
    pub fn current(
        persona: Arc<dyn crate::principal::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.
    #[must_use]
    pub fn verifying(
        persona: Arc<dyn crate::principal::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(any(test, feature = "test-util"))]
            test_search_scope_cap: std::sync::Mutex::new(None),
        }
    }

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

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

    /// Overrides `SEARCH_SCOPE_CAP` for this authority instance — test-only,
    /// so an over-cap refusal can be exercised without constructing
    /// thousands of real participation ties.
    ///
    /// # Panics
    ///
    /// Panics if a prior test thread panicked while holding this cell's
    /// lock.
    #[cfg(any(test, feature = "test-util"))]
    pub 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.
    ///
    /// # Errors
    ///
    /// See [`PrincipalError`]'s variant docs.
    pub 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".
            Err(_store_error) => Err(PrincipalError::StoreUnavailable),
        }
    }

    /// Verify a conversation-grant token minted by
    /// [`polyc_query_model::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 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);
        }
        let (conversation_id, subject, memory_sources) = claims.into_parts();
        // `GrantSubject::ControlFleet` names no conversation — its
        // `conversation_id` is always the empty-string sentinel
        // `mint_control_fleet_grant` fixes, and this arm never reads it,
        // exactly as that mint function's own doc requires of a verifier.
        // It proposes no persona-memory co-participant either, so
        // `memory_sources` is dropped here, unused by this principal.
        if let Some((purpose, statement_digest)) = subject.control_fleet() {
            let digest = polyc_crypto::hex::decode(statement_digest)
                .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok())
                .ok_or_else(|| {
                    tracing::warn!("control-fleet grant statement digest was not 32 hex bytes");
                    PrincipalError::InvalidGrant
                })?;
            return Ok(Principal::ControlFleet(ControlFleetPrincipal::minted(
                purpose.to_owned(),
                digest,
            )));
        }
        // `GrantSubject::AdminFleet` is a caller-held, arbitrary-SQL Fleet
        // capability (PR 10A) — unlike `ControlFleet` it binds no statement
        // digest, so verification here is signature + kind + expiry ONLY
        // (already checked above). The freshness check this subject's own
        // safety actually depends on — is `admin_persona` CURRENTLY admin —
        // cannot run here: it needs a persona-store read, and this function
        // is sync. `Self::scoping_for` (async) does that read before this
        // principal's Fleet scope is ever honored for anything.
        if let Some((admin_persona, session)) = subject.admin_fleet() {
            return Ok(Principal::AdminFleet(AdminFleetPrincipal::minted(
                admin_persona.to_owned(),
                session.to_owned(),
            )));
        }
        // A `GrantSubject::Persona` presented with the empty conversation-id
        // sentinel is the PERSONA-WIDE shape `mint_persona_wide_grant` mints
        // (8C-2's `GetMyPayments` move, not `ControlFleet` and not the
        // routine-fire shape) — it names no single conversation, so it never
        // becomes a `Principal::ConversationGrant` at all. It resolves
        // through the exact same fresh, per-call participation lookup
        // `Principal::Persona` already gets from an explorer session
        // (`Self::scoping_for`'s own `Principal::Persona` arm) — never the
        // fleet, and never a second, divergent implementation of "my own
        // conversations." The routine-fire shape of this same subject always
        // carries a real conversation id, so it is unaffected and still
        // becomes a `Principal::ConversationGrant` below.
        if conversation_id.is_empty()
            && let GrantSubject::Persona(persona_id) = &subject
        {
            return Ok(Principal::Persona(PersonaPrincipal::minted(
                persona_id.clone(),
            )));
        }
        let composite_digests = subject
            .composite_trace()
            .map(|(_persona_id, trace, memory, routine)| vec![trace, memory, routine])
            .or_else(|| {
                subject
                    .admin_composite_trace()
                    .map(|(trace, routine, addresses)| vec![trace, routine, addresses])
            });
        if let Some(digests) = composite_digests {
            for digest in digests {
                let valid = polyc_crypto::hex::decode(digest)
                    .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok())
                    .is_some();
                if !valid {
                    tracing::warn!("composite-trace statement digest was not 32 hex bytes");
                    return Err(PrincipalError::InvalidGrant);
                }
            }
        }
        // A persona composite trace must name a real persona: an empty id
        // would form the owner partition `persona--mem`. The admin subject is
        // the one shape for a caller with no persona (POLY-394).
        if subject
            .composite_trace()
            .is_some_and(|(persona_id, ..)| persona_id.is_empty())
        {
            tracing::warn!("a composite-trace grant named an empty persona");
            return Err(PrincipalError::InvalidGrant);
        }
        // An admin composite trace names no persona and may not read memory
        // (POLY-394). A proposed memory source on it has no legitimate mint
        // site, so it refuses the whole grant rather than widening the scope.
        if subject.admin_composite_trace().is_some() && !memory_sources.is_empty() {
            tracing::warn!("an admin composite-trace grant proposed persona-memory sources");
            return Err(PrincipalError::InvalidGrant);
        }
        Ok(Principal::ConversationGrant(
            ConversationGrantPrincipal::minted(conversation_id, subject, memory_sources),
        ))
    }

    /// 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
    /// [`crate::principal::PersonaSource::participations`] — never cached,
    /// never a replay. A persona with zero participations still scopes
    /// successfully, to an empty catalog.
    ///
    /// # Errors
    ///
    /// Returns [`PrincipalError::StoreUnavailable`] if a
    /// [`Principal::Persona`]'s participation set cannot be resolved.
    /// Infallible for [`Principal::Admin`]. For
    /// [`Principal::ConversationGrant`], returns
    /// [`PrincipalError::MalformedConversationGrantSubject`] only if
    /// `Scoping::for_conversation`'s own upstream invariant broke — not
    /// reachable in production.
    #[allow(
        clippy::too_many_lines,
        reason = "one match arm per Principal variant, each spelling out its own Scoping literal in full"
    )]
    pub async fn scoping_for(&self, principal: &Principal) -> Result<Scoping, PrincipalError> {
        let scoping = match principal {
            Principal::Admin(admin) => Scoping::admin(admin.persona_id()),
            Principal::ConversationGrant(grant) => {
                // The subject's own owner identity always rides along —
                // `GrantSubject::owner_persona_id`'s contract IS "the
                // caller's own verified persona id," so a grant's subject
                // reads its own `persona-memory/v1` partition whether or not
                // it proposed any co-participant. A `Turn` subject has no
                // owner identity (`None`), matching I-4: a turn-subject
                // grant admits no owner-visible table.
                let owner = grant.subject().owner_persona_id().map(str::to_owned);
                // Each PROPOSED co-participant is verified, never trusted:
                // `active_persona` resolves it (following any merge), then
                // its participations must actually include this grant's own
                // conversation. Any failure refuses the WHOLE mint —
                // `SourceOutsideScope` — rather than silently dropping the
                // one persona that did not verify.
                let participants = self
                    .verified_memory_participants(grant.conversation_id(), grant.memory_sources())
                    .await?;
                let mut scoping =
                    Scoping::for_conversation(grant.conversation_id(), grant.subject())?;
                let memory = MemorySources {
                    owner,
                    participants,
                };
                let scope = match scoping.scope() {
                    QueryScope::Conversations { conversations, .. } => QueryScope::Conversations {
                        conversations: conversations.clone(),
                        memory,
                    },
                    QueryScope::Fleet => QueryScope::Fleet,
                    QueryScope::FleetConversation { conversation } => {
                        QueryScope::FleetConversation {
                            conversation: conversation.clone(),
                        }
                    }
                };
                scoping.set_scope(scope);
                scoping
            }
            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::persona(persona.persona_id(), conversation_ids)
            }
            Principal::ControlFleet(control_fleet) => {
                Scoping::from_control_fleet(control_fleet.clone())
            }
            Principal::AdminFleet(admin_fleet) => {
                let Some(persona_host) = self.persona.load_full() else {
                    return Err(PrincipalError::StoreUnavailable);
                };
                match persona_host
                    .active_persona(admin_fleet.admin_persona().to_owned())
                    .await
                {
                    Ok(Some(active)) if active.admin => {
                        Scoping::from_admin_fleet(active.persona_id, admin_fleet.clone())
                    }
                    // Either no profile at all, or one an admin has already
                    // de-admitted since this grant was minted — both refuse
                    // this Fleet scope the same way `verify_admin_session`
                    // refuses a browser session whose persona no longer
                    // carries the attribute.
                    Ok(Some(_) | None) => return Err(PrincipalError::NotAuthorizedForFleet),
                    Err(_store_error) => return Err(PrincipalError::StoreUnavailable),
                }
            }
        };
        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.
    ///
    /// The same derivation the deleted embedded engine's
    /// `QueryAuthority::scope_for` performed (Epic 1565, chunk 10C-1), minus
    /// the engine session: this method takes a bare persona id rather than a
    /// verified [`Principal`], because its caller has already authenticated
    /// the session by another means 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.
    pub 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();
        Ok(Scoping::own_rows(persona_id, conversation_ids))
    }

    /// Authorizes `principal` for a conversation read that needs no runnable
    /// engine session — the identical checks the deleted embedded engine's
    /// `QueryAuthority::scope_for` performed, minus building the session:
    /// refuses [`Principal::ControlFleet`]/
    /// [`Principal::AdminFleet`] the same way, then runs [`Self::scoping_for`]
    /// (the participation check, the admin re-check) and discards the
    /// derived [`Scoping`].
    ///
    /// This is the funnel `polyc-control-plane`'s forensics conversation-read
    /// routes call: they need an authorization verdict, never a `DataFusion`
    /// session (Epic 1565, chunk 10C-1).
    ///
    /// # Errors
    ///
    /// See [`Self::scoping_for`], plus
    /// [`PrincipalError::ControlFleetNotUsableEmbedded`]/
    /// [`PrincipalError::AdminFleetNotUsableEmbedded`] for the two refused
    /// principal kinds.
    pub async fn authorize_conversation_read(
        &self,
        principal: &Principal,
    ) -> Result<(), PrincipalError> {
        if matches!(principal, Principal::ControlFleet(_)) {
            return Err(PrincipalError::ControlFleetNotUsableEmbedded);
        }
        if matches!(principal, Principal::AdminFleet(_)) {
            return Err(PrincipalError::AdminFleetNotUsableEmbedded);
        }
        self.scoping_for(principal).await?;
        Ok(())
    }

    /// Resolve `principal_ref`'s trusted PARTICIPATION search scope for one
    /// turn — the set of PREVIOUS conversations the participation-scoped
    /// search surface may read.
    ///
    /// The trust contract: `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
    ///    [`crate::principal::PersonaSource::active_persona`]. A `None`
    ///    verdict refuses ([`SearchScopeError::PersonaNotActive`]); a store
    ///    error refuses as infrastructure-unavailable
    ///    ([`SearchScopeError::StoreUnavailable`]).
    /// 2. Resolves the BOUNDED participation scope via
    ///    [`crate::principal::PersonaSource::participation_scope`], which
    ///    already applies visibility tombstones and refuses over cap BEFORE
    ///    any per-conversation read runs.
    /// 3. Removes `conversation_id` — the CALLING conversation — from the
    ///    resolved set.
    /// 4. Canonicalizes and hashes what remains.
    ///
    /// Returns a [`SearchScope`], never a [`Principal`].
    ///
    /// # 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.
    pub 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(any(test, feature = "test-util"))]
        let cap = self
            .test_search_scope_cap
            .lock()
            .expect("poison")
            .unwrap_or(SEARCH_SCOPE_CAP);
        #[cfg(not(any(test, feature = "test-util")))]
        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 the trusted turn-scope path's own job.
        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 trait UnixClock: Send + Sync {
    /// Returns the current Unix time in milliseconds.
    fn now_unix_ms(&self) -> u64;
}

/// The production clock.
pub 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 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 struct CredentialWitness {
    credential: PresentedCredential,
    authority: Arc<CredentialAuthority>,
    clock: Arc<dyn UnixClock>,
    composite_trace_memory_sources: Option<Vec<String>>,
    composite_trace_routine: bool,
    composite_trace_addresses: bool,
}

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 {
    /// 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 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
            }
        }
    }

    /// Admits ONLY a signed conversation-grant token — never a raw session
    /// bearer (PR 10A's edge listener).
    ///
    /// Unlike [`Self::admit_bearer`], there is no fallback branch here at
    /// all: `token` is always presented as
    /// [`PresentedCredential::ConversationGrant`], so a caller presenting a
    /// raw session bearer instead reaches
    /// [`CredentialAuthority::verify_conversation_grant`] — which verifies a
    /// signature against the turn-read role's own trust set — and refuses
    /// with [`PrincipalError::InvalidGrant`] there; the session-bearer
    /// verification path ([`CredentialAuthority::verify_admin_session`]) is
    /// structurally unreachable from this function, not merely unreached by
    /// this call.
    ///
    /// # Errors
    ///
    /// Returns the grant refusal for a malformed, forged, expired, or
    /// non-grant-shaped token — see
    /// [`CredentialAuthority::verify_conversation_grant`].
    pub async fn admit_grant_only(
        token: String,
        authority: Arc<CredentialAuthority>,
        clock: Arc<dyn UnixClock>,
    ) -> Result<(Self, Scoping), PrincipalError> {
        Self::admit(
            PresentedCredential::ConversationGrant(token),
            authority,
            clock,
        )
        .await
    }

    /// Verifies `credential` and retains it for revalidation.
    ///
    /// # Errors
    ///
    /// Returns whatever the shared mechanism returns for an invalid, expired,
    /// revoked, unknown, or unresolvable credential.
    pub 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,
                composite_trace_memory_sources: None,
                composite_trace_routine: false,
                composite_trace_addresses: false,
            },
            scoping,
        ))
    }

    /// Retains the trace-attributed memory proposal for stream revalidation.
    ///
    /// The proposal is verified again on every revalidation. It is never an
    /// authorization result by itself.
    pub fn retain_composite_trace_memory_sources(&mut self, sources: Vec<String>) {
        self.composite_trace_memory_sources = Some(sources);
    }

    /// Retains that this witness serves the fixed composite routine statement.
    pub const fn retain_composite_trace_routine(&mut self) {
        self.composite_trace_routine = true;
    }

    /// Retains that this witness serves the fixed participant-address
    /// statement of an admin composite trace (POLY-464).
    ///
    /// Each revalidation then returns the one-conversation Fleet scope, and
    /// only while the grant still carries the address digest.
    pub const fn retain_composite_trace_addresses(&mut self) {
        self.composite_trace_addresses = true;
    }

    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 async fn current_scope(&self) -> Result<QueryScope, PrincipalError> {
        let scoping = Self::verify(&self.credential, &self.authority, self.clock.as_ref()).await?;
        if let Some(sources) = &self.composite_trace_memory_sources {
            self.authority
                .composite_trace_memory_scope(&scoping, sources)
                .await
        } else if self.composite_trace_routine {
            if scoping.composite_trace().is_none() {
                Err(PrincipalError::SourceOutsideScope)
            } else {
                Ok(QueryScope::Fleet)
            }
        } else if self.composite_trace_addresses {
            let addresses = scoping
                .composite_trace()
                .and_then(crate::principal::CompositeTracePrincipal::address_statement_digest);
            match (addresses, scoping.conversation_id()) {
                (Some(_), Some(conversation)) => Ok(QueryScope::FleetConversation {
                    conversation: conversation.to_owned(),
                }),
                _ => Err(PrincipalError::SourceOutsideScope),
            }
        } else {
            Ok(scoping.into_parts().scope)
        }
    }
}