polyc-query 2026.8.3

//! The sealed per-request authorization funnel (A2, the 2026-07-21
//! access-control retrofit — docs/reference/datafusion-data-layer.md, "Access
//! control": "One scope path, sealed").
//!
//! # The one-scope-path invariant
//!
//! [`QueryAuthority`] is this crate's ONLY public way to obtain a runnable
//! query session. Its primary query API is [`QueryAuthority::scope_for`],
//! which accepts nothing but an already-verified [`Principal`] and returns a
//! [`ScopedQuery`] whose catalog is bounded to exactly what that principal's
//! own verification established — Fleet-wide for a verified admin, one
//! conversation's own partition for a verified conversation grant.
//! [`QueryAuthority::scope_for_turn`] is the one other entry point, for a
//! trusted-side caller that already holds a conversation and turn from the
//! turn's own dispatch attribution and so has nothing to verify; it returns a
//! [`ScopedQuery`] identical to the conversation-grant arm's, never a
//! [`Principal`], so the no-public-constructor seal on [`Principal`] is
//! untouched. Both funnel through one private session constructor
//! (`QueryAuthority::session`) over one private `Scoping` input — code spans,
//! not intra-doc links, because a public doc may not link a private item — so
//! a change to what a scoped session may read cannot land on one entry point
//! and miss the other. There is
//! no third, looser constructor anywhere in this crate's public surface: a
//! caller with a [`Principal`] can reach a [`ScopedQuery`], and a caller
//! WITHOUT one cannot name `crate::engine::QueryEngine`,
//! `crate::session::QueryScope`, `crate::engine::PartitionEvents`,
//! `crate::engine::ReferenceData`, or any decode/provider/view internal at
//! all — every one of those is `pub(crate)` (see [`crate`]'s own module doc
//! for the full sealed/public split, and the "Pinning the seal" section
//! below for how that split is verified, not merely asserted).
//!
//! This module's OWN public surface, beyond [`QueryAuthority`]/[`Principal`]/
//! [`ScopedQuery`]/[`PrincipalError`]/[`ScopedQueryError`]/
//! [`mint_conversation_grant`] (each covered by its own section below or its
//! own doc comment), is exactly five more items: [`PersonaCell`],
//! [`GrantSubject`], (issue #1592) `crate::routine_catalog::RoutineCatalog`,
//! and (the participation-scoped search authority,
//! docs/proposals/participation-scoped-agent-search.md) [`SearchScope`] and
//! [`SearchScopeError`].
//! `PersonaCell` is retained for the compatibility constructor used by
//! external test support; production composition uses
//! [`QueryAuthority::new_state_backed`] with a narrow [`PersonaSource`].
//! Neither handle carries scoped-data access on its own: it is the sealed funnel's
//! [`QueryAuthority::verify_admin_session`] that does the actual scoped read
//! through it, never a caller holding the handle directly. Likewise, a
//! `RoutineCatalog` implementation carries no scoped-data access of its own —
//! this module's `ScopedQuery::resolve_routines` is what actually calls
//! into it, for a verified `QueryScope::Fleet` session (every routine,
//! unfiltered) and, as of issue #1882, a verified persona-scoped session
//! too (that persona's own routines only — see `resolve_routines`'s own
//! doc for the filter) — never for a conversation-grant session, which
//! resolves no caller persona to filter by.
//! `GrantSubject` is `pub` for the same reason as `mint_conversation_grant`
//! itself: a caller minting a conversation grant (a harness turn dispatch
//! or, as of #1576, the explorer's own web-session mint endpoint) must be
//! able to name and construct the subject it is minting for. [`SearchScope`]
//! and [`SearchScopeError`] are `pub` because
//! [`QueryAuthority::resolve_search_scope`] is itself a narrow, TRUSTED-side
//! entry point exactly like [`QueryAuthority::scope_for_turn`] — see that
//! method's own doc for the full trust contract and the "Trusted search
//! authority" design section it implements. It returns a [`SearchScope`],
//! never a [`Principal`], for the identical reason `scope_for_turn` returns a
//! [`ScopedQuery`] rather than one: a [`Principal`] has no public
//! constructor, and handing one out here would be the first crack in that
//! seal.
//!
//! # `Principal`: minted only by verification, never constructed
//!
//! [`Principal`] is a `pub` enum — a caller outside this crate can name it,
//! `match` it, and read its resolved fields through the accessor methods on
//! [`AdminPrincipal`]/[`ConversationGrantPrincipal`]/[`PersonaPrincipal`] —
//! but every one of those inner types keeps its fields PRIVATE and exposes
//! NO public constructor. The only way to obtain a [`Principal`] value is to
//! call [`QueryAuthority::verify_admin_session`] (which mints either an
//! admin or, as of A3, a persona-scoped [`Principal`] for a valid non-admin
//! session — see the "Fail-closed admin resolution" section below and
//! [`PersonaPrincipal`]'s own doc) or
//! [`QueryAuthority::verify_conversation_grant`]. Neither
//! verification method takes a caller-supplied identity as a trusted input:
//! both re-derive the principal from a signed, verified artifact (a session
//! token or a grant token) plus, for the admin path, a FRESH per-request read
//! of the durable persona store — never a cached flag, never a bare
//! `persona_id: String` argument a caller could hand in unchecked.
//!
//! # Fail-closed admin resolution
//!
//! [`QueryAuthority::verify_admin_session`] distinguishes three ways a
//! request can fail to become an admin [`Principal`], and the HTTP layer
//! (`crates/control-plane/src/query_http.rs`) maps each to a DIFFERENT status
//! code — this distinction is why [`PrincipalError`] has three separate
//! variants instead of one:
//!
//! - [`PrincipalError::InvalidSession`] — no token, or a token that fails
//!   [`polyc_crypto::session::verify_session`] (bad signature, expired,
//!   revoked, malformed) → 401. A caller with no credential at all lands
//!   here.
//! - [`PrincipalError::StoreUnavailable`] — the token verifies, but the
//!   admin flag cannot be resolved: the persona-store cell is empty (the
//!   State-backed authority is unavailable OR the read itself returned an
//!   error → 503. This is transient
//!   INFRASTRUCTURE state, never an authorization verdict — mirrors
//!   `crate::forensics::resolve_explorer_caller`'s own documented posture
//!   for its identical ArcSwap-empty case, extended here to also cover a
//!   store READ error (that forensics helper's own `persona_is_admin` folds
//!   a read error into "not admin"/403; this funnel treats it as 503
//!   instead, because a store-read failure says nothing about whether the
//!   caller IS an admin — collapsing it into a 403 would misrepresent an
//!   infrastructure fault as a deliberate access denial).
//! - [`PrincipalError::NotAuthorizedForFleet`] — the token verifies AND the
//!   store answered, but the persona id the token names may no longer act →
//!   403. There is no persona left to mint ANY principal for, admin or
//!   persona-scoped, so this stays a hard refusal. A verified session whose
//!   persona IS active, but does not carry the durable admin attribute, is no
//!   longer refused here (A3): it mints a [`Principal::Persona`] instead —
//!   see [`PersonaPrincipal`]'s doc for exactly what that principal can and
//!   cannot query.
//!
//!   "May no longer act" covers two cases a caller cannot tell apart, and
//!   deliberately so: a persona id naming no profile, and one naming a
//!   profile an admin has de-admitted. The second is the reason this reads
//!   through [`polyc_persona::PersonaStore::active_persona`] rather than
//!   `profile`. Removal keeps the record — a forensics-readable tombstone,
//!   not a deletion — so `profile` answers `Some` for a removed persona and
//!   this funnel used to mint a persona-scoped principal for one. Removal
//!   de-admission and bearer revocation are separate State mutations.
//!   Checking active status here, fresh per request, prevents a bearer minted
//!   before de-admission from retaining persona access even before a separate
//!   revocation command lands.
//!
//! The admin flag is resolved FRESH on every call — never cached across
//! requests or even across two calls in the same process — so a stripped
//! admin bit takes effect on the very next query, matching the design's
//! "the session ... never caches its privileges" invariant
//! (docs/reference/datafusion-data-layer.md, "Access control").
//!
//! # Conversation grants
//!
//! [`QueryAuthority::verify_conversation_grant`] is the
//! signature-checked-before-claims-trusted, `kind`-tagged, TTL-bound grant
//! verification the explorer's read path runs on, moved into this crate from
//! the control plane byte-for-byte so the funnel has exactly one verification path
//! per principal kind, not one inside this crate and a second, independently
//! written one in the control plane. Minting stays a control-plane
//! responsibility ([`mint_conversation_grant`] is `pub` for that reason
//! alone — minting needs the private half of the signing key, which only the
//! control plane holds); verification needs only the public half, which is
//! why it belongs in the crate that also builds the scoped session the
//! grant authorizes.
//!
//! # QRY-3: a decode-amplification backstop, and the spill mount's ordering dependency
//!
//! [`crate::engine::QueryLimits::max_source_events`] is a backstop against a
//! runaway/pathological query's decode fan-out — it is NOT a memory or
//! out-of-memory bound, does not cover the replay allocation that precedes
//! it, and is an event COUNT, not bytes. See that field's own doc for the
//! full reasoning and issue #1541 for the tracked real fixes (a byte-based
//! meter, selective/lazy decode, retention, infra-backed spill).
//!
//! It is enforced TWICE, at two different points in [`ScopedQuery::execute`]:
//!
//! - An O(1) PRE-CHECK (`ScopedQuery::estimate_source_event_total`) sums
//!   `PartitionJournal::partition_event_count` across the scope's own
//!   partitions and calls `ScopedQuery::enforce_source_budget` over that sum
//!   BEFORE `ScopedQuery::resolve_partitions` — and therefore before any
//!   replay, any `crate::cache` lookup, or any
//!   `crate::engine::decode_partition_tables` call — ever runs. This is the
//!   check that actually keeps the reject-before-decode invariant true:
//!   `crate::cache`'s decode cache means a `Lookup::Tail`/`Lookup::Miss`
//!   resolution (and therefore a decode) now happens INSIDE
//!   `resolve_partitions` itself, so a check placed only AFTER that call
//!   returns — as this crate had before this pre-check existed — decodes
//!   first and rejects second, exactly backwards from what this backstop is
//!   for.
//! - `ScopedQuery::enforce_source_budget` runs a SECOND time, over
//!   `ResolvedPartitions::replayed_events` right after `resolve_partitions`
//!   returns — defense in depth against the same TOCTOU window the
//!   pre-check itself cannot close (an append landing between the pre-check's
//!   own count reads and the actual replay), not the primary gate.
//!
//! Both calls share one message set (`FLEET_BUDGET_EXCEEDED_MESSAGE`/
//! `CONVERSATION_BUDGET_EXCEEDED_MESSAGE`), so a caller cannot distinguish
//! which one fired.
//!
//! Separately, `build_base_session_state` points `DataFusion`'s spill at
//! [`crate::engine::QueryLimits::spill_dir`] — this crate's OWN default is a
//! portable `std::env::temp_dir`-rooted path, and the control plane's
//! `Config::query_spill_dir` keeps that same portable default. A Kubernetes
//! deployment overrides it to a DEDICATED mount (e.g. `/var/query-spill`) via
//! `POLYCHROME_QUERY_SPILL_DIR`. **That override REQUIRES a PAIRED, SEPARATE
//! manifest PR (the dedicated `emptyDir` at that path) to land and deploy
//! FIRST** — this crate creates neither the directory's PARENT mount nor the
//! path itself ahead of time, and `build_base_session_state` creates
//! `spill_dir` EAGERLY (not lazily), so a missing mount PANICS at
//! [`QueryAuthority::new_state_backed`] — control-plane startup, not merely the first
//! spilling query. See [`crate::engine::QueryLimits::spill_dir`]'s doc, and
//! `build_base_session_state`'s, for the full ordering note and the
//! three-tier quota reasoning (`DataFusion` quota < `emptyDir` `sizeLimit` <
//! `ephemeral-storage` limit).
//!
//! # Admitting the scheduler partition (issue #1592)
//!
//! `ScopedQuery::replay_scoped_partitions`'s `QueryScope::Fleet` arm used
//! to hard-filter every discovered partition to the `"conv-"` prefix —
//! correct until the `fires` typed table (`crate::decode::fires`) needed the
//! routine scheduler's own dedicated, NON-conversation partition,
//! `"routine-scheduler"` (`ROUTINE_SCHEDULER_PARTITION` in this module,
//! kept in exact lockstep with the literal
//! `crates/control-plane/src/routine_scheduler.rs`'s own
//! `ROUTINE_SCHEDULER_PARTITION` constant names). That partition is admitted
//! NARROWLY: the Fleet replay loop now accepts a partition name iff it
//! starts with `"conv-"` OR equals `"routine-scheduler"` exactly — no other
//! non-conversation partition is, or ever will be by this same code path,
//! admitted. This module's `#[cfg(test)]`
//! `fleet_replay_admits_the_routine_scheduler_partition_but_no_other_non_conversation_partition`
//! test proves the narrowness for Fleet.
//!
//! A `QueryScope::Conversations` replay never LISTS partitions at all (it
//! replays exactly the caller's own `conv-{id}` names) — but, as of issue
//! #1882, its own replay ADDITIONALLY admits this same
//! `"routine-scheduler"` partition by name whenever `caller_identity` is
//! `Some` (a verified persona-scoped session, never a conversation grant —
//! see [`GrantSubject`]'s own doc and `Scoping::for_conversation`, which
//! always sets `caller_identity` to `None`). This is what makes a
//! persona-scoped `fires` possible at all: `fires`' one source partition
//! being Fleet-only by construction is no longer true once a persona scope
//! also reads it. The admission stays leniently OPTIONAL for that scope with
//! respect to READABILITY (an unreadable or not-yet-existing scheduler
//! partition is skipped-and-logged, never a hard failure for the rest of an
//! otherwise healthy persona query) — unlike this scope's own conversation
//! partitions, where any one read failure IS a hard error. This leniency
//! does NOT extend to `ReplayError::BytesBudgetExceeded` (private to this
//! module): that check is a
//! global, resource-protection backstop (QRY-3/#1541), not a per-partition
//! readability concern, and it hard-fails the whole query for the scheduler
//! partition exactly as it does for every other partition in every
//! scope — mirroring the Fleet arm's own identical admission, not an
//! inconsistency with it. See `ScopedQuery::replay_scoped_partitions`'s own
//! doc for both arms, and its scheduler-partition admission block
//! specifically for why "never fail" there is scoped to readability alone.
//!
//! # Pinning the seal
//!
//! Two independent checks pin this seal, catching the two ways it could
//! drift in opposite directions:
//!
//! - `tests::sealed_set_is_crate_only` (`#[cfg(test)]`, in this crate) is a
//!   documented, exhaustive enumeration of every item this module doc claims
//!   is `pub(crate)` — it re-asserts each one's visibility by constructing a
//!   value of the type from WITHIN this crate (which only compiles if the
//!   item is at least crate-visible) alongside a doc comment on the same
//!   test naming the exact item. It can only catch the seal being narrowed
//!   TOO FAR — an item this crate still needs demoted below crate-visibility
//!   — since it compiles identically whether an item stays `pub(crate)` or
//!   is widened back to `pub`; it is NOT a compile-time backstop against a
//!   reviewer re-widening one.
//! - `crates/query/tests/compile_fail.rs` (QRY-6) is exactly that backstop,
//!   using [`trybuild`](https://docs.rs/trybuild): it compiles
//!   `tests/compile-fail/sealed_modules.rs` as a genuinely EXTERNAL crate —
//!   the same vantage point `polyc-control-plane` or any other downstream
//!   crate has — and asserts every `use polyc_query::{engine, decode,
//!   provider, session, statement_gate}::...` in it fails to resolve, diffing
//!   the real rustc output against a checked-in `.stderr`. A PR that widens
//!   any of those modules back to `pub` makes this fixture start compiling,
//!   which trybuild reports as a hard failure ("expected test case to fail
//!   to compile, but it compiled successfully") — the gap the module doc
//!   used to describe as unfilled ("adding [a trybuild harness] for a single
//!   invariant was judged more machinery than the invariant warrants") is
//!   now closed directly, not left to `just arch` and PR review alone.
//!
//! The remaining backstops still apply on top of both: the layer-boundary
//! arch check (`just arch`, which fails if any Container reaches around this
//! funnel by depending on `polyc-query`'s internals directly rather than
//! through `polyc-control-plane`'s own composition of [`QueryAuthority`]) and
//! a cargo-doc/public-surface review at PR time: `missing_docs` deny means
//! every genuinely `pub` item in this crate carries its own doc comment
//! already, so the crate's *entire* public surface is enumerable by running
//! `cargo doc -p polyc-query --no-deps` and reading the one rendered index
//! page.

use std::path::Path;
use std::sync::{Arc, LazyLock};

use crate::journal::{JournalError, PartitionJournal};
use arc_swap::ArcSwapOption;
use arrow::array::Array as _;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use datafusion::execution::memory_pool::FairSpillPool;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
use datafusion::execution::{SessionState, SessionStateBuilder};
use polyc_crypto::session;
#[cfg(any(test, feature = "test-util"))]
use polyc_crypto::session::RevokedTokens;
#[cfg(any(test, feature = "test-util"))]
use polyc_crypto::signing_role::SessionRole;
use polyc_crypto::signing_role::{
    HandoffRole, RoleTrustSet, SigningRole as _, TurnReadRole, TurnReadSigner,
};
use polyc_persona::{PersonaHost, ScopeResolution};
use serde::{Deserialize, Serialize};

use crate::cache::{CacheConfig, DecodeCache, Lookup};
use crate::dashboard::DashboardCell;
use crate::engine::{
    PartitionEvents, PartitionTables, QueryEngine, QueryLimits, ReferenceData,
    decode_partition_tables,
};
use crate::output::{self, QueryResultJson};
use crate::routine_catalog::RoutineCatalog;
use crate::session::QueryScope;
use crate::statement_gate;

/// The routine scheduler's own dedicated, NON-conversation event-log
/// partition (`crates/control-plane/src/routine_scheduler.rs`'s
/// `ROUTINE_SCHEDULER_PARTITION`, kept in exact lockstep with that literal —
/// see [`ScopedQuery::replay_scoped_partitions`]'s Fleet arm for where this
/// is admitted).
const ROUTINE_SCHEDULER_PARTITION: &str = "routine-scheduler";

/// Ceiling on a persona's total participation count
/// [`QueryAuthority::resolve_search_scope`] will resolve before refusing
/// outright — see [`SearchScopeError::OverCap`].
///
/// Matches `PersonaHost::participation_scope`'s own `cap` parameter, which
/// checks this BEFORE any per-conversation tombstone read runs
/// (docs/proposals/participation-scoped-agent-search.md, "Work bounds":
/// resolving the scope is itself unbounded work on the approval path, and
/// the check has to bound it before that per-tie cost is paid, not after).
///
/// The design record deliberately does not pin an exact number here — W0
/// owns the operational SLOs — beyond stating that a real cap must sit well
/// below the persona-record capacity ceiling (~27,600 conversations, from the
/// participation index's own 1 MiB record cap; see
/// `PersonaStore::participation_scope`'s doc). 5,000 is a conservative,
/// documented placeholder that leaves that headroom; revisit once W0 lands an
/// operational number.
const SEARCH_SCOPE_CAP: usize = 5_000;

/// Domain separator for [`QueryAuthority::resolve_search_scope`]'s scope
/// hash — see [`SearchScope::hash`]'s own doc for the full five-step
/// algorithm this pins
/// (docs/proposals/participation-scoped-agent-search.md, "Binding the scope the
/// approver saw"). The version lives in this string (`v1`): changing any step
/// of the algorithm means minting a new domain string, which deliberately
/// invalidates every outstanding approval bound to the old hash rather than
/// silently reinterpreting it under a changed formula.
const SEARCH_SCOPE_HASH_DOMAIN: &[u8] = b"polychrome.search.scope.v1";

/// The single admission rule every Fleet-scope partition-discovery site in
/// this module shares: a partition is admitted iff it is a conversation
/// partition (`"conv-"`-prefixed) or exactly [`ROUTINE_SCHEDULER_PARTITION`]
/// — issue #1592's narrow, one-literal exception, never "any
/// non-conversation partition" (see this module's own "Admitting the
/// scheduler partition" doc section for the full rationale).
///
/// Before this helper existed, [`ScopedQuery::replay_scoped_partitions`]'s
/// Fleet arm open-coded this exact condition while
/// [`ScopedQuery::resolve_partitions_cached`]'s Fleet arm and
/// [`ScopedQuery::estimate_source_event_total`]'s Fleet arm each kept their
/// own, older `partition.starts_with("conv-")`-only filter — the two
/// decode-cache-path sites never picked up #1592's admission when it landed,
/// so a cache-enabled deployment (`crate::cache::CacheConfig::new`'s own
/// doc: any deployment with exactly one control-plane replica) silently
/// dropped every `routine-scheduler` row from `fires`/`routine_lifecycle`
/// fleet-wide, while every test passed because the fixture suite's shared
/// `Fixture::build` always leaves the cache disabled. All three call sites
/// admit the scheduler partition unconditionally — there is no production
/// call site that wants it excluded — so this helper takes no flag; each
/// site simply calls it.
fn is_admitted_partition(partition: &str) -> bool {
    partition.starts_with("conv-") || partition == ROUTINE_SCHEDULER_PARTITION
}

/// Compatibility handle to the reference persona host, re-read on every call.
///
/// Production composition uses [`PersonaSource`] through
/// [`QueryAuthority::new_state_backed`]. This public type remains only for the
/// legacy test-support constructor and retires in F0.
pub type PersonaCell = Arc<ArcSwapOption<PersonaHost>>;

/// Current persona reads the query authority needs, independent of where the
/// persona family is physically hosted.
#[async_trait::async_trait]
pub trait PersonaSource: Send + Sync {
    /// Resolves current liveness and privileged roles.
    async fn active_persona(
        &self,
        persona_id: String,
    ) -> Result<Option<polyc_persona::ActivePersona>, polyc_persona::PersonaError>;
    /// Reads current participation rows.
    async fn participations(
        &self,
        persona_id: String,
    ) -> Result<
        Vec<polyc_proto::proto::polychrome::persona::v1::Participation>,
        polyc_persona::PersonaError,
    >;
    /// Resolves bounded, visibility-filtered search scope.
    async fn participation_scope(
        &self,
        persona_id: String,
        cap: usize,
    ) -> Result<ScopeResolution, polyc_persona::PersonaError>;
    /// Enumerates personas with usage rollups.
    async fn usage_rollup_index(&self) -> Result<Vec<String>, polyc_persona::PersonaError>;
    /// Reads one complete query reference snapshot.
    async fn reference_snapshot(
        &self,
        persona_id: String,
    ) -> Result<Option<polyc_persona::PersonaReferenceSnapshot>, polyc_persona::PersonaError>;
}

#[async_trait::async_trait]
impl PersonaSource for PersonaHost {
    async fn active_persona(
        &self,
        persona_id: String,
    ) -> Result<Option<polyc_persona::ActivePersona>, polyc_persona::PersonaError> {
        Self::active_persona(self, persona_id).await
    }
    async fn participations(
        &self,
        persona_id: String,
    ) -> Result<
        Vec<polyc_proto::proto::polychrome::persona::v1::Participation>,
        polyc_persona::PersonaError,
    > {
        Self::participations(self, persona_id).await
    }
    async fn participation_scope(
        &self,
        persona_id: String,
        cap: usize,
    ) -> Result<ScopeResolution, polyc_persona::PersonaError> {
        Self::participation_scope(self, persona_id, cap).await
    }
    async fn usage_rollup_index(&self) -> Result<Vec<String>, polyc_persona::PersonaError> {
        Self::usage_rollup_index(self).await
    }
    async fn reference_snapshot(
        &self,
        persona_id: String,
    ) -> Result<Option<polyc_persona::PersonaReferenceSnapshot>, polyc_persona::PersonaError> {
        Self::reference_snapshot(self, persona_id).await
    }
}

#[derive(Clone)]
enum PersonaAccess {
    #[allow(
        dead_code,
        reason = "available only to the explicit test-support constructor"
    )]
    Legacy(PersonaCell),
    Current(Arc<dyn PersonaSource>),
}

impl PersonaAccess {
    fn load_full(&self) -> Option<Arc<dyn PersonaSource>> {
        match self {
            Self::Legacy(cell) => cell
                .load_full()
                .map(|host| -> Arc<dyn PersonaSource> { host }),
            Self::Current(source) => Some(Arc::clone(source)),
        }
    }
}

/// A verified admin principal — see [`QueryAuthority::verify_admin_session`].
///
/// No public constructor: the only way to obtain one is through that
/// verification method.
#[derive(Debug, Clone)]
pub struct AdminPrincipal {
    persona_id: String,
}

impl AdminPrincipal {
    /// The verified admin's durable principal id.
    #[must_use]
    pub fn persona_id(&self) -> &str {
        &self.persona_id
    }
}

/// What a conversation-grant token was minted for.
///
/// Every grant scopes queries to exactly one conversation (see
/// [`ConversationGrantPrincipal::conversation_id`]) — this enum names WHO the
/// grant additionally speaks for within that conversation, so a durable
/// [`crate::audit::ReadAuditRecord`] never has to fabricate a turn id for a
/// caller that has none. [`GrantSubject::Turn`] is a turn dispatch. Nothing
/// mints one today: a turn's own reads resolve their scope in process
/// (`crate::control-plane::query_nav`, `#1675`), with no token to sign or
/// verify. The variant stays because it is a value the wire format can carry
/// and every surface that verifies a grant must keep REFUSING one it is not
/// for — `crate::control-plane::forensics` pins exactly that.
/// [`GrantSubject::WebSession`] is an authenticated
/// explorer session viewing this conversation (#1576's own mint endpoint,
/// `crate::forensics::api_query_grant` in the control plane) — no turn backs
/// it, so it carries a session-scoped identifier and its own, separately
/// budgeted rolling window instead.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "id", rename_all = "snake_case")]
pub enum GrantSubject {
    /// A harness turn dispatch — the turn id the per-turn query budget keys
    /// on.
    Turn(String),
    /// An authenticated explorer session with no turn of its own — a
    /// session-scoped identifier (today, the session's own verified persona
    /// id), never a fabricated turn id.
    WebSession(String),
}

impl GrantSubject {
    /// The turn id, when this subject is [`GrantSubject::Turn`] — `None` for
    /// [`GrantSubject::WebSession`], never a fabricated value.
    ///
    /// Not `const`: `Some(id)` here relies on `String`'s non-const `Deref`
    /// coercion to `&str`, which `rustc` rejects in a `const fn` (E0015) —
    /// clippy's `missing_const_for_fn` suggestion is a false positive for
    /// this exact shape.
    #[allow(
        clippy::missing_const_for_fn,
        reason = "String's Deref to str isn't const (E0015)"
    )]
    #[must_use]
    pub fn turn_id(&self) -> Option<&str> {
        match self {
            Self::Turn(id) => Some(id),
            Self::WebSession(_) => None,
        }
    }

    /// The web-session id, when this subject is [`GrantSubject::WebSession`]
    /// — `None` for [`GrantSubject::Turn`].
    #[allow(clippy::missing_const_for_fn, reason = "see turn_id's own doc above")]
    #[must_use]
    pub fn web_session_id(&self) -> Option<&str> {
        match self {
            Self::WebSession(id) => Some(id),
            Self::Turn(_) => None,
        }
    }
}

/// A verified conversation-scoped grant principal — see
/// [`QueryAuthority::verify_conversation_grant`].
///
/// No public constructor: the only way to obtain one is through that
/// verification method.
#[derive(Debug, Clone)]
pub struct ConversationGrantPrincipal {
    conversation_id: String,
    subject: GrantSubject,
}

impl ConversationGrantPrincipal {
    /// The conversation this grant scopes queries to.
    #[must_use]
    pub fn conversation_id(&self) -> &str {
        &self.conversation_id
    }

    /// What this grant was minted for.
    #[must_use]
    pub const fn subject(&self) -> &GrantSubject {
        &self.subject
    }

    /// The turn this grant was minted for — the per-turn query budget's key.
    /// `None` when [`Self::subject`] is [`GrantSubject::WebSession`] (there is
    /// no turn to report; see that variant's doc for why this is `None`
    /// rather than a fabricated id).
    #[must_use]
    #[allow(
        clippy::missing_const_for_fn,
        reason = "GrantSubject::turn_id isn't const either — see its own doc"
    )]
    pub fn turn_id(&self) -> Option<&str> {
        self.subject.turn_id()
    }
}

/// A verified persona-scoped principal — a valid explorer session whose
/// persona does NOT carry the durable admin attribute.
///
/// `#1178`'s A3 item (docs/reference/datafusion-data-layer.md, "Rollout: a
/// phased plan" phase 6). Minted by [`QueryAuthority::verify_admin_session`]
/// for exactly this case (see that method's doc); no public constructor.
///
/// [`QueryAuthority::scope_for`] resolves this principal's own scope through
/// [`polyc_persona::PersonaHost::participations`], fresh per request — never
/// cached, never a full event replay: the persona's own participated
/// conversation partitions, registered through the SAME non-Fleet path (the
/// redacted committed-turns projection, no `events_raw`, no
/// `personas`/`participations` reference tables) a conversation grant
/// already gets. Three boundary conditions apply
/// (docs/reference/datafusion-data-layer.md, "Access control"):
///
/// - **Append-only participation.** A persona sees a conversation it EVER
///   participated in — participation carries no per-conversation revocation
///   claim, because the durable participation index the funnel reads is
///   itself append-only.
/// - **Raw/maintainer-only.** Same redaction as every other non-Fleet scope:
///   no raw partition access even for the persona's own conversations, and
///   no opt-in to widen it.
/// - **Incognito conversations.** Included under the same rule the existing
///   conversation-scoped surfaces already apply — no new behavior invented
///   for this principal kind.
///
/// A persona with zero participations still scopes successfully — an empty,
/// valid catalog that runs queries and returns no rows, never an error.
#[derive(Debug, Clone)]
pub struct PersonaPrincipal {
    persona_id: String,
}

impl PersonaPrincipal {
    /// The persona this principal's queries are scoped to.
    #[must_use]
    pub fn persona_id(&self) -> &str {
        &self.persona_id
    }
}

/// A verified caller identity, minted ONLY by [`QueryAuthority`]'s
/// verification methods — see this module's doc for why there is no public
/// constructor.
#[derive(Debug, Clone)]
pub enum Principal {
    /// A maintainer/operator session, verified fleet-wide admin.
    Admin(AdminPrincipal),
    /// A conversation's own agent turn, verified via a signed grant token.
    ConversationGrant(ConversationGrantPrincipal),
    /// An end-user session, scoped to exactly its own participated
    /// conversations — see [`PersonaPrincipal`]'s doc.
    Persona(PersonaPrincipal),
}

/// Failure verifying a [`Principal`] or scoping a query session to one.
#[derive(Debug, thiserror::Error)]
pub enum PrincipalError {
    /// No valid admin session: missing, malformed, expired, revoked, or
    /// signature-invalid token. Maps to 401 — see the module doc.
    #[error("no valid admin session")]
    InvalidSession,
    /// The persona store could not answer whether the caller is an admin —
    /// the cell is empty, or the store itself errored reading the profile.
    /// Transient infrastructure state, maps to 503 — never 401/403. See the
    /// module doc.
    #[error("persona store unavailable")]
    StoreUnavailable,
    /// A real, verified session, but the persona id it names no longer
    /// resolves to any profile at all — there is no persona left to mint ANY
    /// principal for, admin or persona-scoped. Maps to 403 — see the module
    /// doc. A verified session whose persona DOES resolve, but does not hold
    /// the fleet admin attribute, no longer lands here as of A3 — it mints a
    /// [`Principal::Persona`] instead.
    #[error("this account is not authorized for fleet-wide queries")]
    NotAuthorizedForFleet,
    /// The conversation grant token failed to verify: malformed, bad
    /// signature, or wrong `kind` tag — deliberately one undifferentiated
    /// outcome for every one of THOSE failure modes; see
    /// [`QueryAuthority::verify_conversation_grant`]. Distinct from
    /// [`PrincipalError::GrantExpired`] — an otherwise-valid grant past its
    /// TTL is never folded in here.
    #[error("conversation grant token invalid")]
    InvalidGrant,
    /// The conversation grant token verified in every other respect (real
    /// signature, correct `kind` tag) but its TTL has elapsed. Deliberately
    /// its OWN variant, not folded into [`PrincipalError::InvalidGrant`]: an
    /// ordinary expiry is not a forgery, so a caller (the explorer web
    /// client) can tell the two apart and silently re-mint a fresh grant and
    /// retry once, instead of treating routine expiry like a probing/forged
    /// token. Maps to 401, same as [`PrincipalError::InvalidGrant`] — the two
    /// differ only in whether the caller should retry.
    #[error("conversation grant token expired")]
    GrantExpired,
}

// ---------------------------------------------------------------------
// Trusted participation search-scope authority (W1 of
// docs/proposals/participation-scoped-agent-search.md, "Trusted search
// authority") — a narrow entry point for a TRUSTED caller (the control
// plane, holding a turn's own dispatch attribution) to resolve the
// participation-scoped search surface's authorized conversation set. See
// [`QueryAuthority::resolve_search_scope`].
// ---------------------------------------------------------------------

/// The result of resolving a persona's trusted participation search scope —
/// see [`QueryAuthority::resolve_search_scope`].
///
/// No public constructor: the only way to obtain one is through that method.
/// Carries the canonical (sorted, deduplicated) conversation set the
/// participation-scoped search surface may read, its count, and a stable hash
/// over it — never the [`Principal`] or raw `QueryScope` a caller could use
/// to mint its own session, and never the calling conversation, which
/// [`QueryAuthority::resolve_search_scope`] always removes before this value
/// is constructed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SearchScope {
    conversation_ids: Vec<String>,
    hash: String,
}

impl SearchScope {
    /// The canonical conversation ids this scope authorizes — visibility
    /// tombstones already applied, the calling conversation already removed,
    /// sorted and deduplicated in the same canonical order [`Self::hash`] is
    /// computed over (see that method's doc for the exact algorithm).
    #[must_use]
    pub fn conversation_ids(&self) -> &[String] {
        &self.conversation_ids
    }

    /// The number of conversations in this scope — `self.conversation_ids().len()`,
    /// exposed directly so a caller never has to reconstruct it (and can
    /// never observe it drift from the set the hash below actually covers).
    #[must_use]
    pub const fn count(&self) -> usize {
        self.conversation_ids.len()
    }

    /// The scope hash — a stable, participation-enumeration-order-independent
    /// digest over [`Self::conversation_ids`], for binding into an approval
    /// canonical (docs/proposals/participation-scoped-agent-search.md, "Binding
    /// the scope the approver saw").
    ///
    /// The pinned algorithm, exactly:
    ///
    /// 1. Apply visibility tombstones, then remove the calling conversation
    ///    (both already done by the time this value exists —
    ///    [`QueryAuthority::resolve_search_scope`] runs both before
    ///    constructing one).
    /// 2. Encode each remaining conversation id as UTF-8, prefixed by its
    ///    byte length as a `u32` big-endian.
    /// 3. Sort those encoded records bytewise, then deduplicate.
    /// 4. Prefix the concatenation with the domain string
    ///    `polychrome.search.scope.v1` followed by a `0x00` separator.
    /// 5. Hash with BLAKE3; the digest is the 32-byte output, lower-hex
    ///    encoded.
    ///
    /// Lower-hex encoded, 64 characters.
    #[must_use]
    pub fn hash(&self) -> &str {
        &self.hash
    }
}

/// Failure resolving a persona's trusted participation search scope — see
/// [`QueryAuthority::resolve_search_scope`].
#[derive(Debug, thiserror::Error)]
pub enum SearchScopeError {
    /// `principal_ref` failed to verify as an active persona —
    /// [`polyc_persona::PersonaHost::active_persona`] answered `None`: no
    /// profile at all, or one an admin has removed. There is no persona left
    /// to resolve a scope for, so this refuses unconditionally. Deliberately
    /// indistinguishable from "never existed", matching every other
    /// active-persona gate in this crate (see [`PrincipalError::NotAuthorizedForFleet`]'s
    /// own doc for why that collapse is intentional).
    #[error("persona is not active")]
    PersonaNotActive,
    /// The persona store could not answer — either the active-persona check
    /// or the participation-scope resolution itself errored, or the
    /// persona-store cell is empty. Transient INFRASTRUCTURE state, never
    /// folded into [`SearchScopeError::PersonaNotActive`] — mirrors
    /// [`PrincipalError::StoreUnavailable`]'s own posture: a store failure
    /// says nothing about whether the persona IS active, so collapsing it
    /// into a refusal would misrepresent an infrastructure fault as a
    /// deliberate access denial.
    #[error("persona store unavailable")]
    StoreUnavailable,
    /// The persona's total participation count exceeds the cap
    /// [`QueryAuthority::resolve_search_scope`] enforces, BEFORE any
    /// per-conversation visibility-tombstone read runs
    /// (docs/proposals/participation-scoped-agent-search.md, "Work bounds":
    /// resolving the scope is itself unbounded work on the approval path, and
    /// bounding it after the per-tie reads already ran bounds nothing).
    /// Carries the count so the caller can render a size-aware refusal — "the
    /// history is too large to search at once" — without probing further.
    #[error("participation scope of {count} conversations exceeds the search cap")]
    OverCap {
        /// The persona's total participation count, read from the
        /// enumeration index alone (never the per-tie visibility rows).
        count: usize,
    },
}

/// Canonicalize `conversation_ids` (encode, sort bytewise, dedupe) and hash
/// them per [`SearchScope::hash`]'s own pinned five-step algorithm.
///
/// Returns the canonical (sorted, deduplicated) conversation ids alongside
/// the lower-hex BLAKE3 digest computed over them. Sorting the length-
/// prefixed encoded records (not the raw strings) is what makes the result
/// independent of the caller's own enumeration order — the property the
/// approval binding this scope feeds into rests on
/// (docs/proposals/participation-scoped-agent-search.md, "the scope hash is
/// independent of participation enumeration order").
///
/// # Panics
///
/// Never panics in practice: a conversation id whose UTF-8 byte length
/// exceeds `u32::MAX` is not an id any part of this system mints (chat-edge
/// conversation ids are `UUIDv5` strings, a few dozen bytes).
fn canonical_search_scope(conversation_ids: Vec<String>) -> (Vec<String>, String) {
    let mut encoded: Vec<(Vec<u8>, String)> = conversation_ids
        .into_iter()
        .map(|id| {
            let bytes = id.as_bytes();
            let len = u32::try_from(bytes.len())
                .expect("conversation id byte length exceeds u32 — not an id this system mints");
            let mut record = Vec::with_capacity(4 + bytes.len());
            record.extend_from_slice(&len.to_be_bytes());
            record.extend_from_slice(bytes);
            (record, id)
        })
        .collect();
    encoded.sort_by(|(a, _), (b, _)| a.cmp(b));
    encoded.dedup_by(|(a, _), (b, _)| a == b);

    let total_len = SEARCH_SCOPE_HASH_DOMAIN.len()
        + 1
        + encoded
            .iter()
            .map(|(record, _)| record.len())
            .sum::<usize>();
    let mut buf = Vec::with_capacity(total_len);
    buf.extend_from_slice(SEARCH_SCOPE_HASH_DOMAIN);
    buf.push(0);
    for (record, _) in &encoded {
        buf.extend_from_slice(record);
    }
    let hash = blake3::hash(&buf).to_hex().to_string();

    let canonical_ids = encoded.into_iter().map(|(_, id)| id).collect();
    (canonical_ids, hash)
}

// ---------------------------------------------------------------------
// Conversation-grant token codec — moved verbatim out of the control plane
// (issue #1201), and now the explorer read path's alone.
// ---------------------------------------------------------------------

/// Domain-separator tag pinned as the FIRST field of [`GrantClaims`] (and so
/// the first bytes of the signed canonical, since `serde_json` serializes a
/// struct's fields in declaration order) — mirrors `polyc_crypto`'s
/// canonical-JSON "literal `kind` tag first" convention. The dedicated
/// turn-read signing role prevents cross-purpose use at the type/custody
/// boundary; this tag additionally binds the exact artifact shape.
const GRANT_KIND: &str = "query_conversation_grant.v2";

/// The signed claims a grant token carries, JSON-encoded then ed25519-signed
/// as a whole — never trust a conversation id or grant subject from anywhere
/// else on the conversation-scoped surface. `kind` is always [`GRANT_KIND`].
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct GrantClaims {
    kind: String,
    issuer: String,
    key_id: String,
    conversation_id: String,
    subject: GrantSubject,
    expires_at_ms: u64,
}

/// Mint a grant token binding `conversation_id` + `subject`, signed by
/// `signer` and valid until `expires_at_ms`.
///
/// The wire form is
/// `base64url(claims_json).base64url(signature)` — self-contained (no shared
/// mutable registry between the minting call site and
/// [`QueryAuthority::verify_conversation_grant`]).
///
/// `pub` for exactly one reason: minting needs the signer's PRIVATE key, which
/// only the control plane holds — today just the explorer's own mint endpoint
/// (`crate::forensics::api_query_grant`, minting a
/// [`GrantSubject::WebSession`], #1576). [`QueryAuthority`] never retains a
/// private key, only the public trust set used for verification.
///
/// # Panics
///
/// Never panics in practice: `GrantClaims` is a plain struct with no
/// fallible field types, so its own `serde_json` encoding step cannot fail.
#[must_use]
pub fn mint_conversation_grant(
    signer: &TurnReadSigner,
    conversation_id: &str,
    subject: GrantSubject,
    expires_at_ms: u64,
) -> String {
    let claims = GrantClaims {
        kind: GRANT_KIND.to_owned(),
        issuer: TurnReadRole::ISSUER.to_owned(),
        key_id: signer.identity().key_id().to_owned(),
        conversation_id: conversation_id.to_owned(),
        subject,
        expires_at_ms,
    };
    // `GrantClaims` is a plain, always-serializable struct (no maps, no
    // fallible field types) — encoding it can never fail.
    let canonical = serde_json::to_vec(&claims).expect("GrantClaims always serializes");
    let signature = signer.sign_turn_read_capability(&canonical);
    format!(
        "{}.{}",
        URL_SAFE_NO_PAD.encode(canonical),
        URL_SAFE_NO_PAD.encode(signature)
    )
}

/// Build the ONE process-wide shared base [`SessionState`]: the function/
/// analyzer/optimizer registry every [`ScopedQuery`] clones from (see
/// `crate::engine::QueryEngine::build`'s doc for why cloning it is cheap),
/// and the ONE [`FairSpillPool`]-backed `RuntimeEnv` every concurrent query
/// shares — `DataFusion`'s own prescription for multiple concurrent spillable
/// consumers, so one runaway query spills instead of starving its siblings
/// into `ResourcesExhausted` the way a shared *greedy* pool would
/// (docs/reference/datafusion-data-layer.md's 2026-07-21 decision).
///
/// Also pins the `RuntimeEnv`'s `DiskManager` to `spill_dir` at quota
/// `spill_quota_bytes` (both from [`QueryLimits::spill_dir`]/
/// [`QueryLimits::spill_quota_bytes`] — see each field's own doc for why
/// this crate's own portable default differs from a real deployment's
/// Kubernetes-specific one, and for that deployment default's ordering
/// dependency and three-tier quota reasoning) rather than `DataFusion`'s own
/// unconfigured default (an OS-chosen tmp directory with a 100 GiB quota,
/// `DiskManagerBuilder::default`) — a configured path makes the spill
/// location an observable, documented characteristic of this deployment
/// instead of whatever the OS happens to pick, and the configured quota
/// bounds worst-case disk usage from a spilling query.
/// `RuntimeEnvBuilder::with_temp_file_path`/`::with_max_temp_directory_size`
/// are the current (`DataFusion` 54.0.0) API for this — verified against
/// `datafusion-execution-54.0.0`'s own `runtime_env.rs`, since the
/// disk-manager builder API changed across recent `DataFusion` releases
/// (the older `DiskManagerConfig`-based `with_disk_manager` is deprecated as
/// of 48.0.0 in favor of this builder pair).
///
/// # `spill_dir` is created EAGERLY here, not lazily — a bad path PANICS
///
/// This function's own `.expect(..)` below means a `spill_dir` this process
/// cannot create or write to PANICS this call — i.e. control-plane STARTUP,
/// since [`QueryAuthority::new_state_backed`] calls this synchronously — NOT merely the
/// first spilling query. Verified against `datafusion-execution-54.0.0`'s
/// own `disk_manager.rs::create_local_dirs`: `RuntimeEnvBuilder::build`
/// creates `spill_dir` (and an initial working subdirectory inside it)
/// synchronously, during `DiskManager::try_new`. A deployment pointing this
/// at a Kubernetes-specific mount (as the control plane does via
/// `POLYCHROME_QUERY_SPILL_DIR`, e.g. `/var/query-spill` — the portable
/// `Config::query_spill_dir` default is overridden by the deployment env)
/// MUST ensure that mount already exists — via the PAIRED, SEPARATE manifest
/// PR that field's own doc names — before this code path runs, or the control
/// plane crash-loops at startup.
///
/// This crate deliberately keeps this a hard `.expect(..)` rather than
/// threading a `Result` through [`QueryAuthority::new_state_backed`]: `spill_dir` failing
/// to be creatable is exactly the fail-CLOSED behavior a misordered
/// deployment (this code deployed before its paired manifest PR) should
/// get — a loud startup crash, not a control plane that silently runs with
/// no working spill directory at all.
fn build_base_session_state(
    memory_bytes: usize,
    spill_dir: &Path,
    spill_quota_bytes: u64,
) -> SessionState {
    let runtime = RuntimeEnvBuilder::new()
        .with_memory_pool(Arc::new(FairSpillPool::new(memory_bytes)))
        .with_temp_file_path(spill_dir)
        .with_max_temp_directory_size(spill_quota_bytes)
        .build_arc()
        .unwrap_or_else(|err| {
            panic!(
                "building the process-wide DataFusion runtime failed — most likely `spill_dir` \
                 ({}) does not exist and could not be created; see \
                 `build_base_session_state`'s own doc for the ordering dependency on a paired \
                 manifest PR this may indicate was skipped: {err}",
                spill_dir.display()
            )
        });
    SessionStateBuilder::new()
        .with_runtime_env(runtime)
        .with_default_features()
        .build()
}

/// Constructed once at control-plane startup.
///
/// Holds every shared handle [`ScopedQuery::execute`] needs and exposes two
/// query APIs — [`QueryAuthority::scope_for`] for a verified [`Principal`]
/// and [`QueryAuthority::scope_for_turn`] for a trusted-side caller with
/// nothing to verify — both landing on one private session constructor. See
/// the module doc for the full sealed design.
pub struct QueryAuthority {
    base_state: SessionState,
    journal: Arc<dyn PartitionJournal>,
    persona: PersonaAccess,
    dashboard: DashboardCell,
    /// Retired process-local denylist, present only in legacy fixtures.
    #[cfg(any(test, feature = "test-util"))]
    legacy_revoked: Option<Arc<RevokedTokens>>,
    /// Current bearer authorization. Production always supplies this; the
    /// legacy denylist remains only for pre-cutover unit fixtures.
    bearer_authority: Option<Arc<dyn polyc_session_family::authority::BrowserSessionAuthority>>,
    turn_read_trust: RoleTrustSet<TurnReadRole>,
    /// Test-only trust for the retired stateless-session fixture path.
    #[cfg(any(test, feature = "test-util"))]
    legacy_session_trust: Option<RoleTrustSet<SessionRole>>,
    /// The trusted-signer allow-list every [`ScopedQuery`] this authority
    /// mints verifies lifecycle/receipt events against: the current approval
    /// signer plus every public key this
    /// deployment has ever retired from that role (`#1834`). The
    /// control-plane container resolves this list (from its own durable
    /// secret-store-backed record — a retired key can enter it only from
    /// there, never from an event payload) and passes it in already
    /// expanded; this type does no resolution of its own. This approval-family
    /// trust is independent from the turn-read and session role histories.
    trusted_signers: Vec<Vec<u8>>,
    /// The handoff role's own deployment trust set (`#1124`): the current
    /// handoff signer plus every public key this deployment has retired from
    /// that role. Independent of `trusted_signers` — the handoff role signs
    /// with its own custody key, so it verifies against its own root. Every
    /// [`ScopedQuery`] this authority mints decodes the `handoffs` table
    /// under it.
    handoff_trust: RoleTrustSet<HandoffRole>,
    limits: QueryLimits,
    /// The `routines` reference table's data source (issue #1592). `None`
    /// when this deployment wires no routine catalog (or in a test/tool
    /// embedding that never exercises it).
    routine_catalog: Option<Arc<dyn RoutineCatalog>>,
    /// The process-wide per-partition decode cache, shared (the same
    /// `Arc`) with every [`ScopedQuery`] this authority mints — the same
    /// sharing pattern as the durable bearer authority. See `crate::cache`'s
    /// module doc for the full lookup protocol, the runtime kill switch, and
    /// why this cache's own lifetime is exactly this authority's lifetime
    /// (load-bearing for the signer-soundness argument that module doc
    /// makes).
    cache: Arc<DecodeCache>,
    /// Test-only override for `SEARCH_SCOPE_CAP`, read by
    /// [`QueryAuthority::resolve_search_scope`] — lets its over-cap refusal
    /// be exercised without constructing thousands of real participation
    /// ties, mirroring `ScopedQuery`'s own `race_inject_after_count_read`
    /// test-injection pattern. `None` (the only value outside `#[cfg(test)]`
    /// code) means "use `SEARCH_SCOPE_CAP`".
    #[cfg(test)]
    test_search_scope_cap: std::sync::Mutex<Option<usize>>,
}

impl crate::feed::PartitionInvalidation for QueryAuthority {
    fn invalidate_partition(&self, partition: &str) {
        Self::invalidate_partition(self, partition);
    }
}

impl QueryAuthority {
    /// Build a `QueryAuthority` over `journal`/`persona`/`dashboard` (the
    /// SAME handles every other control-plane surface already shares — never
    /// a second one; `dashboard` is the query-owned cell #1584 built and
    /// #1585 registers as the Fleet-only `dashboard` reference table, see
    /// `crate::dashboard`'s module doc), `revoked` (see the module doc's
    /// "Fail-closed admin resolution" section for which instance to pass:
    /// the explorer gate's own, when configured, so logout revokes query
    /// access in the same stroke; a standalone instance otherwise),
    /// `test_signer_public_key` (test-only fixture material re-labeled for
    /// the retired stateless session/grant path), `trusted_signers` (`#1834`
    /// — the current approval signer plus every
    /// retired one; the allow-list every scoped session verifies
    /// lifecycle/receipt events against, resolved by the caller from its own
    /// durable record, not by this type), `limits` (whose `memory_bytes` sizes the ONE shared
    /// `FairSpillPool` — see [`QueryLimits::memory_bytes`]'s doc),
    /// `routine_catalog` — the `routines` reference table's data source
    /// (issue #1592). `None` when this deployment wires no routine catalog
    /// (or in a test/tool embedding that never exercises it): the
    /// `routines`/`fires` tables then simply build empty for a Fleet
    /// session, never a hard failure — see
    /// [`crate::routine_catalog::RoutineCatalog`]'s own doc — and
    /// `cache_config` (Phase A's decode cache — see `crate::cache`'s module
    /// doc, especially its "runtime kill switch" section for what the
    /// caller must positively establish before setting
    /// [`CacheConfig::enabled`]). Proactive eviction on a
    /// destroy/excise/repair/migrate is the caller's to drive, through
    /// [`Self::invalidate_partition`] — memory hygiene only, never the
    /// correctness mechanism (see `crate::cache`'s module doc).
    ///
    /// # Panics
    ///
    /// Panics when the test-only signer public key is not an encoded ed25519
    /// key. Production composition uses [`Self::new_state_backed`],
    /// whose typed trust sets have already passed that validation.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    // each arg is a distinct shared handle/config this authority holds for its whole lifetime — see the doc above for why each one is needed
    #[cfg(any(test, feature = "test-util"))]
    pub fn new(
        journal: Arc<dyn PartitionJournal>,
        persona: PersonaCell,
        dashboard: DashboardCell,
        revoked: Arc<RevokedTokens>,
        test_signer_public_key: Vec<u8>,
        trusted_signers: Vec<Vec<u8>>,
        limits: QueryLimits,
        routine_catalog: Option<Arc<dyn RoutineCatalog>>,
        cache_config: CacheConfig,
        handoff_trust: RoleTrustSet<HandoffRole>,
    ) -> Self {
        let cache = Arc::new(DecodeCache::new(cache_config));
        Self {
            base_state: build_base_session_state(
                limits.memory_bytes,
                &limits.spill_dir,
                limits.spill_quota_bytes,
            ),
            journal,
            persona: PersonaAccess::Legacy(persona),
            dashboard,
            legacy_revoked: Some(revoked),
            bearer_authority: None,
            turn_read_trust: RoleTrustSet::from_public_keys(vec![test_signer_public_key.clone()])
                .expect("test signer public key is encoded ed25519"),
            legacy_session_trust: Some(
                RoleTrustSet::from_public_keys(vec![test_signer_public_key])
                    .expect("test signer public key is encoded ed25519"),
            ),
            trusted_signers,
            handoff_trust,
            limits,
            routine_catalog,
            cache,
            #[cfg(test)]
            test_search_scope_cap: std::sync::Mutex::new(None),
        }
    }

    #[cfg(test)]
    fn with_turn_read_trust_for_test(mut self, trust: RoleTrustSet<TurnReadRole>) -> Self {
        self.turn_read_trust = trust;
        self
    }

    /// Build the production authority with durable State-backed bearer
    /// authorization. Unlike the legacy test constructor, this constructor has no
    /// process-local revocation fallback and cannot be composed partially.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub fn new_state_backed(
        journal: Arc<dyn PartitionJournal>,
        persona: Arc<dyn PersonaSource>,
        dashboard: DashboardCell,
        turn_read_trust: RoleTrustSet<TurnReadRole>,
        trusted_signers: Vec<Vec<u8>>,
        limits: QueryLimits,
        routine_catalog: Option<Arc<dyn RoutineCatalog>>,
        cache_config: CacheConfig,
        authority: Arc<dyn polyc_session_family::authority::BrowserSessionAuthority>,
        handoff_trust: RoleTrustSet<HandoffRole>,
    ) -> Self {
        let cache = Arc::new(DecodeCache::new(cache_config));
        Self {
            base_state: build_base_session_state(
                limits.memory_bytes,
                &limits.spill_dir,
                limits.spill_quota_bytes,
            ),
            journal,
            persona: PersonaAccess::Current(persona),
            dashboard,
            #[cfg(any(test, feature = "test-util"))]
            legacy_revoked: None,
            bearer_authority: Some(authority),
            turn_read_trust,
            #[cfg(any(test, feature = "test-util"))]
            legacy_session_trust: None,
            trusted_signers,
            handoff_trust,
            limits,
            routine_catalog,
            cache,
            #[cfg(test)]
            test_search_scope_cap: std::sync::Mutex::new(None),
        }
    }

    /// Override `SEARCH_SCOPE_CAP` for this authority instance — test-only,
    /// so [`QueryAuthority::resolve_search_scope`]'s over-cap refusal can be
    /// exercised without constructing thousands of real participation ties.
    #[cfg(test)]
    pub(crate) fn set_test_search_scope_cap(&self, cap: usize) {
        *self.test_search_scope_cap.lock().expect("poison") = Some(cap);
    }

    /// Drop whatever this authority cached for `partition`.
    ///
    /// Called by whoever issued a destroy, an excision, a repair, or a
    /// migration, once that command returned its receipt. Transport success is
    /// not authority (INV-22): a request that has not earned a receipt may
    /// never commit, so invalidating on the request would throw away a decode
    /// on behalf of a mutation that never happened.
    ///
    /// Memory hygiene, not correctness. Every lookup is keyed on the
    /// partition's freshly read event count AND mutation epoch, so a mutation
    /// nobody reported is caught on the very next lookup and misses there —
    /// see `crate::cache`'s module doc. What this buys is dropping the entry
    /// now rather than carrying it until something asks.
    pub fn invalidate_partition(&self, partition: &str) {
        self.cache.evict(partition);
    }

    /// 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 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 { 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 { 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 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 {
            conversation_id: claims.conversation_id,
            subject: 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 async fn scope_for(&self, principal: &Principal) -> Result<ScopedQuery, 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(self.session(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 async fn own_rows_scope(&self, persona_id: &str) -> Result<ScopedQuery, 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(self.session(scoping))
    }

    /// Scope a session to one conversation on behalf of one of its own turns,
    /// with no token to mint and none to verify.
    ///
    /// The trust contract: the caller supplies a `conversation_id` and
    /// `turn_id` taken trusted-side from the turn's own dispatch attribution —
    /// never from a wire field, a tool argument, or anything a model can
    /// choose. A surface that cannot say that about both values has no
    /// business calling this and must go through
    /// [`QueryAuthority::verify_conversation_grant`] +
    /// [`QueryAuthority::scope_for`] instead.
    ///
    /// The result is IDENTICAL to what [`QueryAuthority::scope_for`] returns
    /// for a [`Principal::ConversationGrant`] naming the same conversation and
    /// a [`GrantSubject::Turn`] subject — not by coincidence but by
    /// construction: both go through `Scoping::for_conversation` and then
    /// through the one private session constructor, so a later change to
    /// `allow_explain` or the redacted registration path cannot land on one
    /// path and miss the other. `polyc-query`'s own
    /// `scope_for_turn_matches_the_conversation_grant_path` pins that.
    ///
    /// Returns a [`ScopedQuery`] rather than a [`Principal`] on purpose: a
    /// [`Principal`] has no public constructor (see its own doc, and this
    /// module's "one scope path, sealed" section), and handing one out here —
    /// even a conversation-scoped one — would be the first crack in that
    /// seal. Nothing about this method widens what a caller can construct;
    /// it widens only what a TRUSTED caller can already prove.
    ///
    /// `caller_identity` stays `None`, matching the grant path exactly, even
    /// though a caller reaching this method usually has a persona id to hand.
    /// Deliberate follow-up rather than an oversight: today's audit records
    /// for a conversation-scoped query carry no caller identity, and matching
    /// that shape byte for byte is one fewer difference to review. Filling it
    /// in is a separate, reviewable change to the audit shape.
    #[must_use]
    pub fn scope_for_turn(&self, conversation_id: &str, turn_id: &str) -> ScopedQuery {
        self.session(Scoping::for_conversation(
            conversation_id,
            &GrantSubject::Turn(turn_id.to_owned()),
        ))
    }

    /// Resolve `principal_ref`'s trusted PARTICIPATION search scope for one
    /// turn — the set of PREVIOUS conversations the participation-scoped
    /// search surface may read
    /// (docs/proposals/participation-scoped-agent-search.md, "Trusted search
    /// authority"), 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 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 {
            conversation_ids,
            hash,
        })
    }

    /// Build the one [`ScopedQuery`] value this type ever hands out.
    ///
    /// The single owner of that struct literal, and therefore of every field
    /// a session's access policy lives in — the collaborators it replays and
    /// decodes through, the trust root it verifies receipts against, and the
    /// `allow_explain`/scope pair the redacted-view registration in
    /// [`ScopedQuery::execute_with_params`] keys off. Both entry points
    /// ([`QueryAuthority::scope_for`] and
    /// [`QueryAuthority::scope_for_turn`]) end here, so a change to what a
    /// scoped session is cannot reach one caller and miss the other.
    fn session(&self, scoping: Scoping) -> ScopedQuery {
        let Scoping {
            scope,
            allow_explain,
            caller_identity,
            conversation_id,
            turn_id,
            web_session_id,
        } = scoping;
        ScopedQuery {
            base_state: self.base_state.clone(),
            journal: self.journal.clone(),
            persona: self.persona.clone(),
            dashboard: self.dashboard.clone(),
            limits: self.limits.clone(),
            cache: self.cache.clone(),
            trusted_signers: self.trusted_signers.clone(),
            handoff_trust: self.handoff_trust.clone(),
            routine_catalog: self.routine_catalog.clone(),
            scope,
            allow_explain,
            caller_identity,
            conversation_id,
            turn_id,
            web_session_id,
            #[cfg(test)]
            race_inject_after_count_read: std::sync::Mutex::new(None),
        }
    }
}

/// What one [`ScopedQuery`] may read and what its audit record says about who
/// asked — the whole per-caller half of a session, separated from the
/// deployment-wide collaborators [`QueryAuthority::session`] fills in.
///
/// Private, and the only input that constructor takes, so every way of
/// obtaining a session states its scope, its `EXPLAIN` policy, and its
/// attribution in one place rather than as positional arguments a later
/// caller could pair up differently.
struct Scoping {
    /// The partitions this session may read.
    scope: QueryScope,
    /// Whether `EXPLAIN` is available — Fleet only.
    allow_explain: bool,
    /// The verified persona id an audit record attributes the query to.
    caller_identity: Option<String>,
    /// The conversation an audit record names, for a single-conversation
    /// session.
    conversation_id: Option<String>,
    /// The turn an audit record names, and the per-turn budget's key.
    turn_id: Option<String>,
    /// The web session an audit record names — mutually exclusive with
    /// `turn_id`.
    web_session_id: Option<String>,
}

impl Scoping {
    /// The ONE conversation-scoped shape, derived from a conversation and the
    /// [`GrantSubject`] acting within it.
    ///
    /// Both conversation-scoped entry points come through here — a verified
    /// [`Principal::ConversationGrant`] in [`QueryAuthority::scope_for`], and
    /// a turn the caller already holds trusted-side in
    /// [`QueryAuthority::scope_for_turn`] — so the two cannot drift. That
    /// matters most for `allow_explain`: a change that widened it (or the
    /// non-Fleet redacted registration it selects) on one path alone would be
    /// a redaction bypass reachable from exactly one surface, which is the
    /// hardest kind to notice in review.
    fn for_conversation(conversation_id: &str, subject: &GrantSubject) -> Self {
        Self {
            scope: QueryScope::Conversations(vec![conversation_id.to_owned()]),
            allow_explain: false,
            caller_identity: None,
            conversation_id: Some(conversation_id.to_owned()),
            turn_id: subject.turn_id().map(str::to_owned),
            web_session_id: subject.web_session_id().map(str::to_owned),
        }
    }
}

/// The opaque failure a [`ScopedQuery::execute`] caller sees.
///
/// [`ScopedQueryError::Rejected`] carries the statement gate's own safe,
/// user-actionable message (maps to 400); [`ScopedQueryError::SourceBudgetExceeded`]
/// (QRY-3) is the same kind of caller-actionable, pre-execution refusal — the
/// scope's own replayed data exceeded [`crate::engine::QueryLimits::max_source_events`]
/// — also mapped to 400; [`ScopedQueryError::UnknownTable`] (issue #2147) is
/// the third caller-actionable refusal — the submitted SQL named a table this
/// scope's catalog does not carry — and also maps to 400, carrying a FIXED,
/// scope-aware message that names what this scope CAN query rather than
/// echoing the planner's own text; [`ScopedQueryError::UnknownColumn`]
/// (issue #2206) is the fourth — the tables resolved, but the SQL used a
/// column no relation available at that point carries — and also maps to
/// 400, carrying a message BUILT from the columns the planner resolved
/// against, so it names what really was available there;
/// [`ScopedQueryError::Timeout`] is its own
/// variant (QRY-4) so a query-audit completion record can distinguish "ran out of
/// wall-clock time" from every other engine fault, even though a caller-
/// facing surface still maps it to the same generic 500 as
/// [`ScopedQueryError::Internal`]; every other failure — decode, replay,
/// output encoding, or a planning failure of any kind OTHER than the
/// unresolvable table and column names above — collapses into
/// [`ScopedQueryError::Internal`] (maps to 500). The real error is logged
/// server-side (`tracing::error!`)
/// inside [`ScopedQuery::execute`] before this type is ever constructed, so
/// no caller outside this crate needs, or gets, `DataFusionError`/`ArrowError`
/// `Display` text that could leak schema or file-path detail.
#[derive(Debug, thiserror::Error)]
pub enum ScopedQueryError {
    /// The submitted SQL was rejected before it ever reached the planner —
    /// carries `crate::statement_gate::StatementRejected`'s own rendered
    /// `Display` message (that type itself stays `pub(crate)` — sealed like
    /// every other engine internal — so this variant carries a pre-rendered
    /// `String`, not the type itself).
    #[error("query rejected: {0}")]
    Rejected(String),
    /// This scope's own replayed source data exceeded either
    /// `crate::engine::QueryLimits::max_source_bytes` (issue #1541 — checked
    /// DURING replay, in `ScopedQuery::replay_scoped_partitions`, aborting
    /// the read before the rest of the scope is materialized) or
    /// `crate::engine::QueryLimits::max_source_events` (checked TWICE: an
    /// O(1) pre-check summing `partition_event_count` across the scope
    /// BEFORE any replay or decode runs — see
    /// `ScopedQuery::estimate_source_event_total` — and a defense-in-depth
    /// re-check right after replay returns, before engine assembly — see
    /// `ScopedQuery::enforce_source_budget`). Carries a
    /// rendered, SAFE message — scope-aware (a Fleet
    /// message names the `query_max_source_events` config an administrator
    /// can raise; a `Conversations` message tells the caller their own
    /// accessible history is over budget and that an administrator must
    /// adjust retention or capacity — NOT that the caller can narrow their
    /// own SQL, since replay runs before planning and no `WHERE`/time-range
    /// clause can shrink data this check already rejected) but deliberately
    /// carrying NO partition count, conversation count, or event count: a
    /// caller must not be able to use a rejected query to probe this
    /// deployment's real data volume. Unlike [`ScopedQueryError::Rejected`],
    /// a `Conversations`-scope rejection here is not something the caller
    /// themselves can act on (a Fleet admin narrows their own query's scope
    /// or raises the config; a persona/conversation-grant session is bounded
    /// by its own conversation, hits this only in a genuinely pathological
    /// case, and needs a deployment-side fix either way). See
    /// `ScopedQuery::enforce_source_budget`'s doc for the exact message text
    /// and why it is safe.
    #[error("{0}")]
    SourceBudgetExceeded(String),
    /// The submitted SQL named a table this scope's catalog does not carry
    /// (issue #2147) — the planner could not resolve the name, so nothing
    /// ran.
    ///
    /// Split out of [`ScopedQueryError::Internal`] because a caller that
    /// guessed a table name is the one who can fix it, and the generic
    /// internal message gave it no way to tell a wrong name from a timeout:
    /// a model connected over `polychrome_query` would simply guess again.
    /// Carries a FIXED, scope-aware message — one of
    /// `FLEET_UNKNOWN_TABLE_MESSAGE`, `OWNER_UNKNOWN_TABLE_MESSAGE`, or
    /// `CONVERSATION_UNKNOWN_TABLE_MESSAGE`, chosen by
    /// `ScopedQuery::unknown_table_message` — never the planner's own
    /// text, which names the table the caller asked for and would let a
    /// probing caller confirm, one guess at a time, which names this
    /// deployment does and does not register. Naming this scope's OWN
    /// tables leaks nothing: they are the tables the caller may already
    /// select from.
    #[error("{0}")]
    UnknownTable(String),
    /// Every table the submitted SQL named resolved, but it then used a
    /// COLUMN no relation available at that point carries (issue #2206) —
    /// the planner refused during expression resolution, so nothing ran.
    ///
    /// Split out of [`ScopedQueryError::Internal`] for the same reason
    /// [`ScopedQueryError::UnknownTable`] was: production watched callers
    /// guess `message`, `event_type`, `type`, and `tool` inside one 90-minute
    /// window, and a generic internal failure gave each of them nothing to
    /// correct. Unlike the table case, the message is not fixed — it is
    /// BUILT by `unknown_column_message` from the `valid_fields`
    /// `DataFusion` resolved the expression against, so it names the column
    /// the caller asked for and the columns that were really available
    /// there, and cannot drift from the schema the engine registered.
    ///
    /// Naming those columns leaks nothing. `DataFusion` builds `valid_fields`
    /// from the `DFSchema`s of relations that ALREADY RESOLVED, so it can
    /// only ever hold columns of tables this scope registered — a name the
    /// catalog cannot resolve has no schema to contribute at all. That is a
    /// property of how the offer is built, not of planning order, so it
    /// survives the shapes where the planner does resolve a column and a
    /// table in the same pass (a `UNION ALL` arm, a scalar subquery). A
    /// redacted scope's views also carry only their redacted column set, so a
    /// conversation-scoped caller is never offered a Fleet-only column.
    #[error("{0}")]
    UnknownColumn(String),
    /// The whole pipeline exceeded [`crate::engine::QueryLimits::timeout`]
    /// before it could return a result. Split out from
    /// [`ScopedQueryError::Internal`] (QRY-4) purely so the query-audit
    /// completion record can carry a distinct `QUERY_AUDIT_OUTCOME_TIMEOUT`
    /// outcome — every caller-facing surface still renders it with the same
    /// generic message as `Internal`.
    #[error("query timed out")]
    Timeout,
    /// Everything else — logged in full detail server-side, not echoed here.
    #[error("query failed")]
    Internal,
}

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a Fleet-scoped
/// query (`crate::authority::ScopedQuery::enforce_source_budget`) —
/// deliberately carries NO partition count, conversation count, or event
/// count: a Fleet-wide caller could otherwise use a rejected query to probe
/// this deployment's total data volume. Names the actual config key an
/// administrator raises so the message stays actionable without a number.
const FLEET_BUDGET_EXCEEDED_MESSAGE: &str = "this fleet-wide query exceeds this deployment's \
     configured event budget; an administrator can raise the `query_max_source_events` setting \
     to allow a larger scope";

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a
/// [`QueryScope::Conversations`] query — a conversation grant or a persona
/// session (`crate::authority::ScopedQuery::enforce_source_budget`) — carries
/// no event count either. Deliberately does NOT tell the caller to narrow
/// the query: replay happens before planning, so no `WHERE` clause or time
/// range in the submitted SQL can reduce the source data this check already
/// rejected. The only actionable step is a deployment-side change, so the
/// message names that instead — an administrator adjusting retention or
/// capacity — even though, unlike the Fleet message, this caller has no
/// config key to point at themselves.
const CONVERSATION_BUDGET_EXCEEDED_MESSAGE: &str = "your accessible conversation history exceeds this deployment's configured budget; an \
     administrator needs to adjust retention or capacity before this query can run";

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a Fleet-scoped
/// query whose replayed source data tripped
/// [`crate::engine::QueryLimits::max_source_bytes`] (issue #1541) rather
/// than [`crate::engine::QueryLimits::max_source_events`] —
/// [`ScopedQuery::replay_scoped_partitions`] stopped READING from the
/// journal, before this deployment's full history was ever materialized,
/// the instant cumulative replayed payload bytes crossed the budget. Same
/// posture as [`FLEET_BUDGET_EXCEEDED_MESSAGE`] — no byte, partition, or
/// event count leaked — naming the byte-budget config key instead of the
/// event-count one.
const FLEET_BYTES_BUDGET_EXCEEDED_MESSAGE: &str = "this fleet-wide query exceeds this deployment's \
     configured source-data budget; an administrator can raise the `query_max_source_bytes` \
     setting to allow a larger scope";

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a
/// [`QueryScope::Conversations`] query whose replayed source data tripped
/// [`crate::engine::QueryLimits::max_source_bytes`] — the byte-budget
/// counterpart to [`CONVERSATION_BUDGET_EXCEEDED_MESSAGE`].
const CONVERSATION_BYTES_BUDGET_EXCEEDED_MESSAGE: &str = "your accessible conversation history \
     exceeds this deployment's configured source-data budget; an administrator needs to adjust \
     retention or capacity before this query can run";

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a Fleet-scoped
/// query whose EFFECTIVE volume (cached rows plus any freshly-replayed
/// tail, summed across every scoped partition) tripped
/// [`crate::cache::CacheConfig::max_cached_source_events`] — Phase A's
/// cached-scan volume bound (`crate::ScopedQuery::enforce_cached_volume_budget`).
/// Deliberately as count-free as [`FLEET_BUDGET_EXCEEDED_MESSAGE`], for the
/// identical reason: a caller must never use a rejected query to learn this
/// deployment's real data volume.
const FLEET_CACHED_VOLUME_BUDGET_EXCEEDED_MESSAGE: &str = "this fleet-wide query exceeds this \
     deployment's configured cached-scan volume budget; an administrator can raise the decode \
     cache's configured event-volume ceiling to allow a larger scope";

/// [`ScopedQueryError::SourceBudgetExceeded`]'s message for a
/// [`QueryScope::Conversations`] query whose effective cached-scan volume
/// tripped [`crate::cache::CacheConfig::max_cached_source_events`] — the
/// cached-volume counterpart to [`CONVERSATION_BUDGET_EXCEEDED_MESSAGE`].
const CONVERSATION_CACHED_VOLUME_BUDGET_EXCEEDED_MESSAGE: &str = "your accessible conversation \
     history exceeds this deployment's configured cached-scan volume budget; an administrator \
     needs to adjust the decode cache's configured event-volume ceiling before this query can \
     run";

/// [`ScopedQueryError::UnknownTable`]'s message for a [`QueryScope::Fleet`]
/// query (issue #2147).
///
/// A fleet-wide session is the one scope built with
/// `SessionConfig::with_information_schema(true)` (see `crate::engine`'s
/// module docs), so it can enumerate its own catalog — pointing at that is
/// both shorter and more durable than a hand-maintained list that would go
/// stale the next time a table lands. It also keeps the fixed-message
/// posture: the deployment's real table set never reaches a caller who
/// could not already read it.
const FLEET_UNKNOWN_TABLE_MESSAGE: &str = "that table is not in this deployment's catalog; run \
     `SELECT table_name FROM information_schema.tables` to see what you can query";

/// Every table a conversation-scoped catalog carries — a conversation grant,
/// or a persona session with no routine ownership context (issue #2147).
///
/// The ONE list every surface that names this catalog is held to, rather than
/// a sentence each of them spells by hand:
///
/// - Both of this module's `UnknownTable` messages — the conversation one and
///   the routine-owner one — are BUILT from it, through [`catalog_sentence`],
///   so a name added here reaches a refused caller with no second edit.
/// - The `polychrome_query` tool description is pinned to it from
///   `polyc-control-plane`, the one crate that can see both this crate and the
///   surface — a description literal cannot interpolate a constant, so a test
///   holds it instead. The agent-facing raw-SQL hatch that used to be pinned
///   the same way is deleted (`docs/reference/agent-read-surface.md`); its half of
///   that test went with it, this one did not.
/// - What the engine ACTUALLY registers is probed against it, in both
///   directions, by `conversation_grant_unknown_table_names_this_scopes_own_catalog`
///   in this module's tests: a name here that does not resolve fails, and a
///   thirteenth table registered without being listed here fails too. The
///   second direction is the one that matters — a table added to
///   `crate::engine::registration::create_scope_dependent_views` and left out
///   of this list would put the next caller right back to guessing, which is
///   the failure #2147 was filed for.
///
/// Listing these names to a caller who guessed leaks nothing: every one is a
/// table that caller may already select from.
pub const CONVERSATION_CATALOG: &[&str] = &[
    "events",
    "messages",
    "tool_calls",
    "approvals",
    "attribution",
    "payments",
    "handoffs",
    "grant_replays",
    "usage",
    "model_call",
    "turn_failed",
    "turn_dispatch",
    "refusals",
    "wallet_link_lifecycle",
];

/// The tables a routine-owning persona session adds to
/// [`CONVERSATION_CATALOG`].
///
/// `routines`/`fires` (issue #1882) plus the routine read surface
/// (`routine_grants`, `routine_active_grants`, `routine_setup`,
/// `routine_overview`).
///
/// `crate::engine::registration::RegistrationScope::Owner` registers every
/// one owner-filtered, and `RegistrationScope::Grant` registers none.
///
/// Held to the engine the same way [`CONVERSATION_CATALOG`] is; see that
/// constant's doc for the three surfaces this one also feeds.
pub const OWNER_ONLY_CATALOG: &[&str] = &[
    "routines",
    "fires",
    "routine_grants",
    "routine_active_grants",
    "routine_setup",
    "routine_overview",
];

/// Render a catalog as the sentence fragment every surface that names one
/// uses — `"a, b, and c"`, or `"a and b"` for a pair.
///
/// One renderer, so the two [`ScopedQueryError::UnknownTable`] messages below
/// and the tool descriptions pinned against them cannot spell the same list
/// three different ways. An empty `names` renders as the empty string.
#[must_use]
pub fn catalog_sentence(names: &[&str]) -> String {
    match names {
        [] => String::new(),
        [only] => (*only).to_owned(),
        [first, second] => format!("{first} and {second}"),
        [leading @ .., last] => format!("{}, and {last}", leading.join(", ")),
    }
}

/// The most column names a [`ScopedQueryError::UnknownColumn`] message spells
/// out before it stops naming them and says how many are left (issue #2206).
///
/// Set above the widest table this crate registers — `personas`, at
/// twenty-six Fleet columns — so a miss against a SINGLE table, which is
/// every miss production has seen, is never truncated and always leaves the
/// caller the whole set to correct against. A query joining several wide
/// relations is truncated on purpose: a refusal a reader has to scroll
/// through is a worse answer than one that names enough columns to make the
/// shape obvious and lets them ask about one table at a time.
/// `column_list_bound_covers_the_widest_registered_table` in this module's
/// tests reads the real registrations and fails if a new column pushes any
/// table past this bound, so the single-table guarantee is enforced rather
/// than asserted here. It measures a real `ORDER BY` miss as well as a
/// table's own width, because a miss in `GROUP BY`, `ORDER BY`, or `HAVING`
/// resolves against the projection schema followed by the input schema and
/// would otherwise arrive at twice the width — `crate::engine`'s
/// `unresolved_column` deduplicates for exactly that reason.
const MAX_ADVERTISED_COLUMNS: usize = 32;

/// Render `columns` as the sentence fragment a
/// [`ScopedQueryError::UnknownColumn`] message offers, bounded by
/// [`MAX_ADVERTISED_COLUMNS`].
///
/// Delegates to [`catalog_sentence`] whenever the whole list fits, so a
/// column list and a table list read identically; past the bound it names the
/// first [`MAX_ADVERTISED_COLUMNS`] and counts the rest, which still tells a
/// caller their name is not among a set this large without printing it.
fn column_list_sentence(columns: &[String]) -> String {
    if columns.len() <= MAX_ADVERTISED_COLUMNS {
        let borrowed: Vec<&str> = columns.iter().map(String::as_str).collect();
        return catalog_sentence(&borrowed);
    }
    let named = columns[..MAX_ADVERTISED_COLUMNS].join(", ");
    let remaining = columns.len() - MAX_ADVERTISED_COLUMNS;
    format!("{named}, and {remaining} more")
}

/// Build [`ScopedQueryError::UnknownColumn`]'s message from `unresolved`
/// (issue #2206) — the column the caller wrote, and the columns the planner
/// resolved it against.
///
/// Every word after the name comes from `valid_fields`, never from
/// `DataFusion`'s own `Display` text, so the offer cannot drift from the
/// schema the engine registered — the property
/// [`CONVERSATION_CATALOG`] gives [`ScopedQueryError::UnknownTable`], applied
/// one level down. Echoing the name the caller wrote is safe here in a way it
/// is not for a table: it is the caller's own input, and it reveals nothing
/// about which names this deployment does register.
///
/// `valid_fields` is the RESOLUTION SITE's schema, not the whole statement's,
/// so the message never claims otherwise. A name inside a correlated subquery
/// resolves against the subquery's own relations, and offering those while
/// saying "the tables this query selects from" would send a caller who wrote
/// `e.nope` off to rewrite it against `m` — a differently wrong query. Every
/// wording below is scoped to where the name was used instead.
///
/// Which columns to offer, in order:
///
/// 1. The table the caller qualified the name with (`events.message` offers
///    `events`' columns), when the planner did resolve against that table.
/// 2. The one relation available at the resolution site, when the caller
///    wrote a bare name. This is the case production actually hit — every
///    logged miss was an unqualified name against a single-table `SELECT`.
/// 3. A qualifier that matches NO relation available there — `SELECT
///    events.turn_id FROM events e`, the mistake a caller makes by aliasing
///    in `FROM` and forgetting the alias in the projection. The qualifier is
///    the real error, so the message echoes the whole name as written, names
///    the relations that ARE available, and offers their columns qualified,
///    which spells out the alias to use.
/// 4. Otherwise every column available there, qualified, because a join of
///    several relations has no single table to point at.
fn unknown_column_message(unresolved: &crate::engine::UnresolvedColumn) -> String {
    let requested = &unresolved.name;
    let written = unresolved.qualifier.as_ref().map_or_else(
        || requested.clone(),
        |qualifier| format!("{qualifier}.{requested}"),
    );
    let mut relations: Vec<&str> = Vec::new();
    for relation in unresolved
        .valid_fields
        .iter()
        .filter_map(|(relation, _)| relation.as_deref())
    {
        if !relations.contains(&relation) {
            relations.push(relation);
        }
    }
    let sole_relation = match relations.as_slice() {
        [only] => Some(*only),
        _ => None,
    };
    let anchor = unresolved
        .qualifier
        .as_deref()
        .map_or(sole_relation, |qualifier| {
            relations.contains(&qualifier).then_some(qualifier)
        });
    if let Some(anchor) = anchor {
        let columns: Vec<String> = unresolved
            .valid_fields
            .iter()
            .filter(|(relation, _)| relation.as_deref() == Some(anchor))
            .map(|(_, column)| column.clone())
            .collect();
        if !columns.is_empty() {
            return format!(
                "there is no `{requested}` column on `{anchor}`; you can select {}",
                column_list_sentence(&columns)
            );
        }
    }
    let columns: Vec<String> = unresolved
        .valid_fields
        .iter()
        .map(|(relation, column)| {
            relation
                .as_ref()
                .map_or_else(|| column.clone(), |relation| format!("{relation}.{column}"))
        })
        .collect();
    if columns.is_empty() {
        return format!(
            "there is no `{written}` column here; add a `FROM` clause naming the table to read it \
             from"
        );
    }
    if unresolved.qualifier.is_some() {
        let quoted: Vec<String> = relations.iter().map(|name| format!("`{name}`")).collect();
        let borrowed: Vec<&str> = quoted.iter().map(String::as_str).collect();
        return format!(
            "there is no `{written}` column available where you used it; that part of the query \
             reads from {}, so you can select {}",
            catalog_sentence(&borrowed),
            column_list_sentence(&columns)
        );
    }
    format!(
        "there is no `{requested}` column available where you used it; you can select {}",
        column_list_sentence(&columns)
    )
}

/// [`ScopedQueryError::UnknownTable`]'s message for a
/// [`QueryScope::Conversations`] query — a conversation grant, or a persona
/// session with no routine ownership context (issue #2147).
///
/// Names this scope's whole catalog outright, from
/// [`CONVERSATION_CATALOG`], because for this scope the set is fixed and
/// already documented: it is exactly what
/// `crate::engine::registration::create_scope_dependent_views` builds for
/// `RegistrationScope::Grant`, plus the always-present `events` view. Listing
/// it leaks nothing — every name here is a table the caller may already
/// select from — and it is the one thing that stops a caller guessing again.
static CONVERSATION_UNKNOWN_TABLE_MESSAGE: LazyLock<String> = LazyLock::new(|| {
    format!(
        "that table is not in this conversation's catalog; you can query {}",
        catalog_sentence(CONVERSATION_CATALOG)
    )
});

/// [`ScopedQueryError::UnknownTable`]'s message for a persona session that
/// owns routines (issue #2147) — [`CONVERSATION_CATALOG`] plus
/// [`OWNER_ONLY_CATALOG`], rendered by the same [`catalog_sentence`] the
/// message above uses.
static OWNER_UNKNOWN_TABLE_MESSAGE: LazyLock<String> = LazyLock::new(|| {
    let tables: Vec<&str> = CONVERSATION_CATALOG
        .iter()
        .chain(OWNER_ONLY_CATALOG.iter())
        .copied()
        .collect();
    format!(
        "that table is not in this account's catalog; you can query {}",
        catalog_sentence(&tables)
    )
});

/// [`ScopedQuery::replay_scoped_partitions`]'s own internal failure —
/// deliberately never exposed outside this module;
/// [`ScopedQuery::map_replay_error`] is the only place that translates one
/// into the sealed, public [`ScopedQueryError`] a caller actually sees.
#[derive(Debug, Clone, Copy)]
enum ReplayError {
    /// Listing partitions, or replaying one, failed for a reason OTHER than
    /// the byte budget (e.g. a corrupted conversation partition) — already
    /// logged in full detail at the call site inside
    /// [`ScopedQuery::replay_scoped_partitions`].
    Internal,
    /// [`crate::engine::QueryLimits::max_source_bytes`] tripped mid-replay
    /// (issue #1541): the partition read that crossed the budget is the
    /// last one [`ScopedQuery::replay_scoped_partitions`] ever performed —
    /// see that method's own doc for the exact early-abort mechanics.
    BytesBudgetExceeded {
        /// How many partitions had at least one event read before the
        /// budget tripped — including the one that tripped it. For
        /// [`QueryScope::Fleet`], strictly fewer than the deployment's real
        /// `conv-` partition count whenever a later partition existed and
        /// was never reached; this is the direct, test-asserted proof that
        /// replay stopped EARLY rather than materializing the whole scope
        /// first.
        partitions_replayed: usize,
        /// Cumulative payload bytes read across every partition replayed
        /// before (and including) the trip — at or just past
        /// [`crate::engine::QueryLimits::max_source_bytes`]. Recorded to the
        /// `polychrome_query_replayed_bytes` histogram so an operator sizing
        /// the budget sees the over-budget queries too, not just the ones
        /// that fit.
        bytes_read: u64,
    },
}

/// [`ScopedQuery::resolve_partitions`]'s own result: this scope's own
/// partitions, already decoded, plus the totals [`ScopedQuery::execute`]
/// needs for its metrics observation and its two pre-execution budget
/// checks.
struct ResolvedPartitions {
    /// Every scoped partition's decoded tables, ready for
    /// `crate::engine::QueryEngine::build_from_tables`.
    tables: Vec<PartitionTables>,
    /// How many [`QueryScope::Fleet`] partitions were skipped as unreadable
    /// (QRY-7) — always `0` for [`QueryScope::Conversations`], matching
    /// [`ScopedQuery::replay_scoped_partitions`]'s own posture.
    skipped_partitions: usize,
    /// Total events genuinely REPLAYED this call (a cache hit contributes
    /// zero; a tail or a miss contributes its own replayed length) — what
    /// [`ScopedQuery::enforce_source_budget`] checks, unchanged from before
    /// Phase A.
    replayed_events: usize,
    /// Total payload bytes genuinely REPLAYED this call — the same
    /// cache-aware accounting as `replayed_events`, for
    /// `crate::metrics::record_query_observed`.
    replayed_bytes: u64,
    /// Total EFFECTIVE event volume this scope hands to
    /// `crate::engine::QueryEngine::build_from_tables` — a cache hit's own
    /// full watermark count included, unlike `replayed_events`. What
    /// [`ScopedQuery::enforce_cached_volume_budget`] checks; see that
    /// method's own doc for why this differs from `replayed_events` and why
    /// it is only ever checked with the cache enabled.
    cached_volume_events: usize,
    /// The byte-sized counterpart to `cached_volume_events` — total Arrow
    /// array memory (`crate::engine::PartitionTables::memory_bytes`, summed
    /// across every resolved partition's tables, hit/tail/miss alike) this
    /// scope hands to `crate::engine::QueryEngine::build_from_tables`. Fed
    /// to `crate::metrics::record_cached_scan_volume` alongside
    /// `cached_volume_events` — its own labeled observability dimension, not
    /// folded into `replayed_bytes` (see that call site's own doc).
    cached_volume_bytes: u64,
}

/// A `DataFusion` session scoped to exactly one verified [`Principal`]'s own
/// authority, returned by [`QueryAuthority::scope_for`]. [`ScopedQuery::execute`]
/// is the only thing a caller can do with one.
pub struct ScopedQuery {
    base_state: SessionState,
    journal: Arc<dyn PartitionJournal>,
    persona: PersonaAccess,
    dashboard: DashboardCell,
    limits: QueryLimits,
    /// See [`QueryAuthority::cache`]'s doc — the SAME `Arc` every
    /// [`ScopedQuery`] this authority ever mints shares.
    cache: Arc<DecodeCache>,
    /// Current and retired approval-role public keys. Payment and approval
    /// projections use this same explicit trust root; session and turn-read
    /// grants use independent role trust sets and never enter this field. See
    /// `crate::decode::payments`'s module docs.
    trusted_signers: Vec<Vec<u8>>,
    /// See [`QueryAuthority::handoff_trust`]'s doc — the handoff role's own
    /// deployment trust set, the root the `handoffs` table's
    /// `signature_status` column is computed against (`#1124`).
    handoff_trust: RoleTrustSet<HandoffRole>,
    /// The `routines` reference table's data source (issue #1592) — `None`
    /// when this deployment wires no catalog; see
    /// [`QueryAuthority::new_state_backed`]'s doc.
    routine_catalog: Option<Arc<dyn RoutineCatalog>>,
    scope: QueryScope,
    allow_explain: bool,
    caller_identity: Option<String>,
    conversation_id: Option<String>,
    turn_id: Option<String>,
    /// This session's web-session id for an audit record — `Some` only for a
    /// [`Principal::ConversationGrant`] whose [`GrantSubject`] is
    /// [`GrantSubject::WebSession`] (#1576). Mutually exclusive with
    /// `turn_id`: a grant is minted for exactly one subject.
    web_session_id: Option<String>,
    /// Test-only stale-watermark-race injection point (the regression test
    /// for the fix documented on [`ScopedQuery::resolve_partitions_cached`]'s
    /// own "The stale-watermark race" doc section). `Some((partition,
    /// events))` makes [`ScopedQuery::resolve_partitions_cached`] append
    /// `events` to `partition` via [`PartitionJournal::append_batch`] exactly
    /// once — the instant it reaches that partition, right after its own
    /// `count`/`epoch` reads and before consulting the cache — simulating a
    /// concurrent writer's append landing in the window between this
    /// method's count snapshot and its own later replay call. A `Mutex`
    /// (not a plain field) so a test can arm it through a shared, non-`mut`
    /// `&ScopedQuery` (this crate's tests are a descendant module of
    /// `authority`, so they read/write this field directly); `.take()` on
    /// fire so it injects exactly once even if the same `ScopedQuery` runs
    /// more than one query. Always `None` outside `#[cfg(test)]` builds —
    /// this field does not exist at all in a production binary.
    #[cfg(test)]
    race_inject_after_count_read: std::sync::Mutex<Option<(String, Vec<polyc_eventlog::Event>)>>,
}

impl std::fmt::Debug for ScopedQuery {
    /// Manual, since `SessionState`'s the only field with a `Debug` impl to
    /// borrow — the partition journal and `PersonaHost` (behind `Arc`/`ArcSwapOption`)
    /// implement neither.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ScopedQuery")
            .field("caller_identity", &self.caller_identity)
            .field("conversation_id", &self.conversation_id)
            .field("turn_id", &self.turn_id)
            .field("web_session_id", &self.web_session_id)
            .finish_non_exhaustive()
    }
}

impl ScopedQuery {
    /// This session's caller identity for an audit record — the verified
    /// persona id for [`Principal::Admin`] and [`Principal::Persona`] alike,
    /// `None` for [`Principal::ConversationGrant`] (mirrors
    /// `polyc_query::audit::ReadAuditRecord`'s own "empty until it applies"
    /// contract for that field).
    #[must_use]
    pub fn caller_identity(&self) -> Option<&str> {
        self.caller_identity.as_deref()
    }

    /// This session's conversation id for an audit record — `Some` only for
    /// [`Principal::ConversationGrant`].
    #[must_use]
    pub fn conversation_id(&self) -> Option<&str> {
        self.conversation_id.as_deref()
    }

    /// This session's turn id for an audit record — `Some` only for a
    /// [`Principal::ConversationGrant`] whose subject is [`GrantSubject::Turn`].
    #[must_use]
    pub fn turn_id(&self) -> Option<&str> {
        self.turn_id.as_deref()
    }

    /// This session's web-session id for an audit record — `Some` only for a
    /// [`Principal::ConversationGrant`] whose subject is
    /// [`GrantSubject::WebSession`] (#1576).
    #[must_use]
    pub fn web_session_id(&self) -> Option<&str> {
        self.web_session_id.as_deref()
    }

    /// Run `sql` under this session's own scope, limits, and `EXPLAIN`
    /// policy (Fleet: allowed; conversation grant and persona: not — matching
    /// phase 1/2's existing policy split, extended identically to A3's
    /// persona scope).
    ///
    /// Pipeline, matching the pre-A2 control-plane call sites verbatim:
    /// statement gate first (before any replay/decode work — a rejected
    /// statement now costs no partition replay at all) → discover + replay
    /// this scope's own partitions (Fleet: every `conv-` partition
    /// [`PartitionJournal::list_partitions`] reports,
    /// unreadable ones skipped, logged, AND COUNTED (QRY-7 — see
    /// `Self::replay_scoped_partitions`'s doc), matching the pre-retrofit
    /// lenient posture; a conversation grant or persona: exactly its own
    /// participated partitions — zero for a persona with no participations,
    /// which is a valid empty scope, not an error — a replay failure on any
    /// one of them IS a real error, since `QueryScope::Conversations` never
    /// falls back to the Fleet-style skip-and-log posture) → record this
    /// query's observed size (`crate::metrics::record_query_observed` — see
    /// the "Observability, on every query" section below) → (QRY-3) enforce
    /// [`QueryLimits::max_source_events`] over the replayed total, BEFORE any
    /// of it reaches a decoder — see this method's "Why the source budget
    /// exists" section below → for Fleet only, resolve the
    /// `personas`/`participations` reference tables (empty, not a failure,
    /// when the persona store is unavailable — the SAME lenient posture as
    /// an unreadable partition, distinct from
    /// [`QueryAuthority::verify_admin_session`]'s stricter fail-closed 503
    /// during AUTHZ resolution) → build a scoped `QueryEngine` → execute →
    /// serialize to the shared [`crate::output::QueryResultJson`] envelope,
    /// stamped with the skipped-partition count. The whole pipeline runs under
    /// one wall-clock [`QueryLimits::timeout`].
    ///
    /// # Why the source budget exists
    ///
    /// `QueryEngine::build`'s decode loops (`crate::decode`, one pass per
    /// typed table over every replayed partition) are entirely SYNCHRONOUS —
    /// no `.await` anywhere inside a per-partition decode loop. That means
    /// the `tokio::time::timeout` wrapping this whole pipeline cannot preempt
    /// decode mid-pass the way it can a slow `.await` point elsewhere in the
    /// pipeline: once decode starts, the executor cannot yield back to poll
    /// the timeout future again until that synchronous pass returns control,
    /// however long it takes. A persona or conversation-grant scope is
    /// bounded by its own conversation's size either way, but a Fleet scope
    /// replays and decodes EVERY `conv-` partition in the deployment — with
    /// no cap of its own — before `memory_bytes`'s `FairSpillPool` is ever
    /// consulted (that pool only meters `DataFusion`'s OWN execution-time
    /// allocations, not this crate's own pre-execution replay/decode pass).
    /// [`QueryLimits::max_source_events`] is the actual bound on that
    /// unpreemptible work — not the timeout, and not the memory pool. It is
    /// enforced twice: an O(1) pre-check over `partition_event_count` sums
    /// runs BEFORE any replay, cache lookup, or decode
    /// (`ScopedQuery::estimate_source_event_total`), and the post-replay
    /// `ScopedQuery::enforce_source_budget` re-check remains as
    /// defense-in-depth against the pre-check's narrower TOCTOU window. See that field's own doc for why
    /// this is a decode-amplification backstop, NOT a memory/OOM bound.
    ///
    /// # Observability, on every query
    ///
    /// Every query records its replayed size, so the byte distribution an
    /// operator needs to size `max_source_bytes`/`max_source_events` is
    /// visible across ALL queries, not only the rare rejected one:
    /// - A query whose replay COMPLETES (success, or an event-count budget
    ///   rejection) records `crate::metrics::record_query_observed` right
    ///   after replay and before the count check — the total replayed event
    ///   count, the total replayed payload BYTES, and the scope-type label.
    /// - A query that trips the BYTE budget aborts mid-replay before its full
    ///   event count is knowable, so `map_replay_error` records
    ///   `crate::metrics::record_query_replayed_bytes` with the accumulated
    ///   over-budget byte count instead — the bytes histogram still sees it,
    ///   the event count is simply omitted (it stopped early).
    ///
    /// # Errors
    ///
    /// Returns [`ScopedQueryError::Rejected`] if `sql` fails the statement
    /// gate, [`ScopedQueryError::SourceBudgetExceeded`] if the scope's
    /// replayed source data exceeds [`QueryLimits::max_source_events`],
    /// [`ScopedQueryError::UnknownTable`] if `sql` names a table this scope's
    /// catalog does not carry (issue #2147 — the planner resolves names after
    /// the scope is replayed, so this one, unlike the two above, refuses a
    /// query whose data was already read),
    /// [`ScopedQueryError::UnknownColumn`] if every table resolved but `sql`
    /// then selected a column none of them carries (issue #2206 — the same
    /// already-read caveat applies; a statement that gets BOTH wrong returns
    /// the table failure, because the table check runs first over the whole
    /// error tree),
    /// [`ScopedQueryError::Timeout`] if the whole pipeline outruns
    /// [`QueryLimits::timeout`], and [`ScopedQueryError::Internal`] for every
    /// other failure (logged in full detail server-side).
    ///
    /// A caller running a fixed template with values to bind calls
    /// [`ScopedQuery::execute_with_params`] instead; this method is that one
    /// with no parameters, and everything documented above applies to both.
    pub async fn execute(&self, sql: &str) -> Result<QueryResultJson, ScopedQueryError> {
        self.execute_with_params(sql, &[]).await
    }

    /// Run `sql` under this session's own scope with `params` bound to its
    /// `$1`, `$2`, ... placeholders as text values, then behave exactly like
    /// [`ScopedQuery::execute`] (which is this method with an empty `params`)
    /// — same statement gate, same replay, same source budget, same redacted
    /// views, same timeout, same audit-visible envelope.
    ///
    /// A value in `params` is bound into the PLANNED query
    /// (the engine's own `execute_with_params`), never spliced
    /// into `sql` as text, so a caller may pass a string it does not trust —
    /// quotes, comment markers, semicolons, a whole `UNION SELECT` — and
    /// none of it can become SQL. That property is what lets a surface offer
    /// a fixed-template tool whose only variable parts are values, with no
    /// escaping step to get wrong — a tool a caller can steer the arguments
    /// of but not the statement.
    ///
    /// The values are bound as UTF-8; a template that needs a number binds
    /// it through a `CAST` in its own fixed text, so this signature stays
    /// free of `DataFusion`'s own scalar vocabulary.
    ///
    /// A template and its `params` are checked for correspondence in ONE
    /// direction: a placeholder no value was supplied for is refused, but a
    /// SURPLUS value — one no placeholder in `sql` claims — is silently
    /// ignored and the query runs. A surface building an optional predicate
    /// must derive the clause and its value from one decision, because the
    /// failure mode of splitting them is a statement whose predicate is gone
    /// while its parameter remains: it succeeds, and returns every row
    /// unfiltered. Returning both halves from one answer, rather than
    /// assembling them at a call site that holds each, is what makes the
    /// mismatch unrepresentable; this crate's own
    /// `a_surplus_parameter_is_ignored_while_a_missing_one_is_refused` pins
    /// the engine behavior that rule exists for.
    ///
    /// # Errors
    ///
    /// Same as [`ScopedQuery::execute`] — including
    /// [`ScopedQueryError::UnknownTable`] and
    /// [`ScopedQueryError::UnknownColumn`], which `params` can neither cause
    /// nor cure: a value binds into the planned query, never into the
    /// statement text, so `sql` alone decides which names the catalog and
    /// the resolved schemas are asked to resolve. A surface running a FIXED
    /// template therefore reaches `UnknownColumn` only through a bug in its
    /// own template text, which reads to its caller as a caller-actionable
    /// refusal it cannot act on — the cost of a shared entry point, and the
    /// reason this crate pins every fixed template against the real engine.
    /// A template with a placeholder `params` does not
    /// supply is a bug in the calling surface, not a caller-actionable
    /// failure, so it collapses into [`ScopedQueryError::Internal`] like
    /// every other engine fault.
    pub async fn execute_with_params(
        &self,
        sql: &str,
        params: &[&str],
    ) -> Result<QueryResultJson, ScopedQueryError> {
        // Gate first — ahead of every partition replay (see the doc above).
        if let Err(rejected) = statement_gate::check_statement_allowed(sql, self.allow_explain) {
            return Err(ScopedQueryError::Rejected(rejected.to_string()));
        }

        let body = async {
            let scope_label = self.scope_label();

            // QRY-3: a pre-check, from `PartitionJournal::partition_event_count`
            // sums, BEFORE any replay, cache lookup, or decode ever runs —
            // see the module doc's QRY-3 section for why this, and not the
            // post-replay check further below alone, is what keeps the
            // reject-before-decode invariant true now that `crate::cache`'s
            // decode cache resolves (and may decode) a partition INSIDE
            // `resolve_partitions` itself.
            let estimated_source_events = self
                .estimate_source_event_total()
                .await
                .map_err(|err| self.map_replay_error(err, scope_label))?;
            self.enforce_source_budget(estimated_source_events, scope_label)?;

            let resolved = self
                .resolve_partitions()
                .await
                .map_err(|err| self.map_replay_error(err, scope_label))?;

            // Unconditional — see this method's "Observability, on every
            // query" doc section above for why this runs before, and
            // independent of, the budget checks just below.
            crate::metrics::record_query_observed(
                scope_label,
                resolved.replayed_events,
                resolved.replayed_bytes,
            );
            // Its own labeled dimension — never folded into the REPLAYED
            // numbers just above (a cache hit's replayed events/bytes are
            // near-zero, which would make the cache invisible to this
            // metric) — see `crate::metrics::record_cached_scan_volume`'s own
            // doc.
            crate::metrics::record_cached_scan_volume(
                scope_label,
                resolved.cached_volume_events,
                resolved.cached_volume_bytes,
            );

            // Defense in depth against the pre-check's own TOCTOU window
            // (an append landing between the pre-check's count reads and the
            // actual replay just above) — see the module doc's QRY-3
            // section. With the cache disabled, every partition was a
            // genuine miss, so `resolved.replayed_events` already equals the
            // scope's whole effective volume.
            self.enforce_source_budget(resolved.replayed_events, scope_label)?;
            // Phase A: a cache hit reports near-zero REPLAYED events above
            // while `QueryEngine::build_from_tables` still hands `DataFusion`
            // the full cached table — this second, cache-aware budget closes
            // that gap. Only meaningful (and only checked) when the cache is
            // actually active; see `ScopedQuery::enforce_cached_volume_budget`'s
            // own doc.
            if self.cache.enabled() {
                self.enforce_cached_volume_budget(resolved.cached_volume_events, scope_label)?;
            }
            // #1882: a persona-scoped session (never Fleet, never a
            // conversation grant — see `caller_identity`'s own field doc and
            // `Scoping::for_conversation`, which always sets it `None`) owns
            // exactly its own `routines`/`fires` rows. `owner_persona_id`
            // carries that persona id through both `resolve_reference_data`
            // (the ROW-level filter on `routines`) and
            // `QueryEngine::build_from_tables` (whether `fires` registers at
            // all for this otherwise Fleet-only-reference-table build).
            let owner_persona_id = match &self.scope {
                QueryScope::Fleet => None,
                QueryScope::Conversations(_) => self.caller_identity.as_deref(),
            };
            let reference = self
                .resolve_reference_data(&resolved.tables, owner_persona_id)
                .await;

            let engine = QueryEngine::build_from_tables(
                &self.base_state,
                &self.scope,
                resolved.tables,
                reference,
                self.limits.clone(),
                owner_persona_id.is_some(),
            )
            .await
            .map_err(|err| {
                tracing::error!(error = %err, "failed to build the scoped query engine");
                ScopedQueryError::Internal
            })?;
            let output = engine
                .execute_with_params(sql, params, self.allow_explain)
                .await
                .map_err(|err| {
                    // Issue #2147: a name the catalog cannot resolve is the
                    // CALLER's to fix, and used to be indistinguishable from
                    // a timeout or a decode fault, so a model that guessed a
                    // table name simply guessed again. Logged at `warn`, not
                    // `error` — nothing in the deployment is wrong.
                    if crate::engine::is_unresolved_table_error(&err) {
                        tracing::warn!(
                            error = %err,
                            scope = scope_label,
                            "query named a table this scope's catalog does not carry"
                        );
                        return ScopedQueryError::UnknownTable(
                            self.unknown_table_message(owner_persona_id.is_some())
                                .to_owned(),
                        );
                    }
                    // Issue #2206: the same failure one level down — a
                    // column no relation available at that point carries.
                    // Checked AFTER the table case so a statement that gets
                    // both wrong reads as the table it named; the offer is
                    // safe either way, because `valid_fields` is built from
                    // relations that already resolved (see
                    // `crate::engine::unresolved_column`). Logged at `warn`
                    // for the same reason as above — nothing in the
                    // deployment is wrong.
                    //
                    // Not every planning failure the CALLER owns lands here:
                    // an ambiguous column across a join arrives as
                    // `SchemaError::AmbiguousReference`, falls through, and
                    // still records an internal failure. Widening the split
                    // to cover it is its own change.
                    if let Some(unresolved) = crate::engine::unresolved_column(&err) {
                        tracing::warn!(
                            error = %err,
                            scope = scope_label,
                            "query named a column the tables it selects from do not carry"
                        );
                        return ScopedQueryError::UnknownColumn(unknown_column_message(
                            &unresolved,
                        ));
                    }
                    tracing::error!(error = %err, "query execution failed");
                    ScopedQueryError::Internal
                })?;
            output::output_to_json(&output, resolved.skipped_partitions).map_err(|err| {
                tracing::error!(error = %err, "failed to encode query result as JSON");
                ScopedQueryError::Internal
            })
        };

        match tokio::time::timeout(self.limits.timeout, body).await {
            Ok(result) => result,
            Err(_elapsed) => {
                tracing::error!(timeout = ?self.limits.timeout, "query exceeded its timeout");
                Err(ScopedQueryError::Timeout)
            }
        }
    }

    /// This session's scope-TYPE label for `crate::metrics`'s low-cardinality
    /// `scope` label — `"fleet"` for [`Principal::Admin`], `"grant"` for
    /// [`Principal::ConversationGrant`], `"persona"` for
    /// [`Principal::Persona`] (A3). Derived from the fields
    /// [`QueryAuthority::scope_for`] already set per `Principal` arm, rather
    /// than a redundant new field: [`ScopedQuery::conversation_id`] is `Some`
    /// only for a conversation grant (both `Turn` and `WebSession` subjects
    /// alike — `self.turn_id` alone would miss a `WebSession` grant, which
    /// carries no turn id), and [`ScopedQuery::caller_identity`] is `Some`
    /// for both `Admin` and `Persona` but `Admin` is already distinguished by
    /// `self.scope` being [`QueryScope::Fleet`]. The fourth value,
    /// `"conversations"`, is never reached by any current `Principal`
    /// constructor (every [`QueryScope::Conversations`] session today is
    /// either a grant or a persona) — kept only so this match stays
    /// exhaustive over `QueryScope`'s own two-variant shape rather than
    /// panicking or silently mislabeling a future third way to build one.
    const fn scope_label(&self) -> &'static str {
        match &self.scope {
            QueryScope::Fleet => "fleet",
            QueryScope::Conversations(_) => {
                if self.conversation_id.is_some() {
                    "grant"
                } else if self.caller_identity.is_some() {
                    "persona"
                } else {
                    "conversations"
                }
            }
        }
    }

    /// (QRY-3) Reject a scope whose `total_events` — the SUM of every
    /// [`crate::engine::PartitionEvents::events`] length, across the whole
    /// scope, computed by [`ScopedQuery::execute`] alongside the
    /// unconditional metrics observation — exceeds
    /// [`QueryLimits::max_source_events`]. Called right after
    /// `replay_scoped_partitions` returns and strictly before
    /// [`crate::engine::QueryEngine::build`] decodes anything — see
    /// `execute`'s "Why the source budget exists" doc section for why this
    /// check, and not the wall-clock timeout or the memory pool, is what
    /// actually bounds a Fleet scope's pre-execution replay/decode pass.
    ///
    /// The returned message is deliberately scope-aware but carries NO
    /// event count, conversation count, or partition detail — a caller must
    /// never be able to use a rejected query to probe this deployment's real
    /// data volume (a Fleet-wide caller especially: fleet size is exactly
    /// the kind of aggregate a non-admin-adjacent caller has no other way to
    /// learn). [`FLEET_BUDGET_EXCEEDED_MESSAGE`] names the
    /// `query_max_source_events` config an administrator can raise;
    /// [`CONVERSATION_BUDGET_EXCEEDED_MESSAGE`] tells a conversation-grant or
    /// persona caller their OWN accessible history is over budget and that
    /// an administrator must adjust retention or capacity. Neither message
    /// tells the caller to narrow their query: replay happens BEFORE
    /// planning, so no `WHERE` clause or time range in the submitted SQL
    /// can reduce the source data this check already rejected — narrowing
    /// would not help, and the message must not imply it would.
    fn enforce_source_budget(
        &self,
        total_events: usize,
        scope_label: &'static str,
    ) -> Result<(), ScopedQueryError> {
        if total_events <= self.limits.max_source_events {
            return Ok(());
        }
        tracing::error!(
            total_events,
            max_source_events = self.limits.max_source_events,
            scope = scope_label,
            "scoped query's replayed source data exceeded the pre-execution decode-amplification \
             budget; refusing to decode"
        );
        crate::metrics::record_source_budget_exceeded(scope_label);
        let message = match &self.scope {
            QueryScope::Fleet => FLEET_BUDGET_EXCEEDED_MESSAGE,
            QueryScope::Conversations(_) => CONVERSATION_BUDGET_EXCEEDED_MESSAGE,
        };
        Err(ScopedQueryError::SourceBudgetExceeded(message.to_owned()))
    }

    /// Which fixed [`ScopedQueryError::UnknownTable`] message this session
    /// gets (issue #2147), selected by the same three postures
    /// `crate::engine::registration::RegistrationScope` registers under:
    /// [`FLEET_UNKNOWN_TABLE_MESSAGE`] for [`QueryScope::Fleet`],
    /// [`OWNER_UNKNOWN_TABLE_MESSAGE`] for a persona session that owns
    /// routines, and [`CONVERSATION_UNKNOWN_TABLE_MESSAGE`] otherwise.
    ///
    /// `owner_scoped_routines` is the SAME value
    /// [`ScopedQuery::execute`] hands
    /// `crate::engine::QueryEngine::build_from_tables`, so the message can
    /// never name a table that build did not register.
    fn unknown_table_message(&self, owner_scoped_routines: bool) -> &'static str {
        match &self.scope {
            QueryScope::Fleet => FLEET_UNKNOWN_TABLE_MESSAGE,
            QueryScope::Conversations(_) if owner_scoped_routines => {
                OWNER_UNKNOWN_TABLE_MESSAGE.as_str()
            }
            QueryScope::Conversations(_) => CONVERSATION_UNKNOWN_TABLE_MESSAGE.as_str(),
        }
    }

    /// The cached-scan volume bound: reject a scope whose
    /// EFFECTIVE volume — cached rows plus any freshly-replayed tail, summed
    /// across every scoped partition, computed by
    /// [`ScopedQuery::resolve_partitions_cached`] — exceeds
    /// [`crate::cache::CacheConfig::max_cached_source_events`]. Only ever
    /// called when [`DecodeCache::enabled`] is `true` (see
    /// [`ScopedQuery::execute`]): with the cache disabled, every partition
    /// is a genuine miss, so [`ScopedQuery::enforce_source_budget`]'s own
    /// replayed-event check already bounds the identical quantity.
    ///
    /// # Why this exists alongside [`ScopedQuery::enforce_source_budget`]
    ///
    /// A cache HIT reports near-zero REPLAYED events (replay was skipped
    /// entirely) to [`ScopedQuery::enforce_source_budget`], but
    /// `crate::engine::QueryEngine::build_from_tables` still hands
    /// `DataFusion` the FULL cached table to plan and scan. This check closes
    /// that gap: it is checked over the scope's TOTAL effective volume
    /// (hit + tail + miss partitions alike), strictly before
    /// `QueryEngine::build_from_tables` is ever called — see
    /// `ScopedQuery::execute`'s own doc.
    ///
    /// The returned message carries no count, mirroring
    /// [`ScopedQuery::enforce_source_budget`]'s own leak-freedom rationale.
    fn enforce_cached_volume_budget(
        &self,
        total_cached_volume_events: usize,
        scope_label: &'static str,
    ) -> Result<(), ScopedQueryError> {
        if total_cached_volume_events <= self.cache.max_cached_source_events() {
            return Ok(());
        }
        tracing::error!(
            total_cached_volume_events,
            max_cached_source_events = self.cache.max_cached_source_events(),
            scope = scope_label,
            "scoped query's effective cached-scan volume exceeded the pre-execution cached-scan \
             volume budget; refusing to hand it to the query engine"
        );
        crate::metrics::record_cached_volume_budget_exceeded(scope_label);
        let message = match &self.scope {
            QueryScope::Fleet => FLEET_CACHED_VOLUME_BUDGET_EXCEEDED_MESSAGE,
            QueryScope::Conversations(_) => CONVERSATION_CACHED_VOLUME_BUDGET_EXCEEDED_MESSAGE,
        };
        Err(ScopedQueryError::SourceBudgetExceeded(message.to_owned()))
    }

    /// Map a [`ReplayError`] from [`ScopedQuery::replay_scoped_partitions`]
    /// to the sealed [`ScopedQueryError`] a caller sees: a byte-budget trip
    /// gets the SAME kind of scope-aware, count-free, caller-actionable
    /// message [`ScopedQuery::enforce_source_budget`] renders for the
    /// event-count budget — only the config key it names differs — everything
    /// else collapses to [`ScopedQueryError::Internal`] (already logged in
    /// full server-side by [`ScopedQuery::replay_scoped_partitions`]).
    fn map_replay_error(&self, err: ReplayError, scope_label: &'static str) -> ScopedQueryError {
        match err {
            ReplayError::Internal => ScopedQueryError::Internal,
            ReplayError::BytesBudgetExceeded {
                partitions_replayed,
                bytes_read,
            } => {
                // `partitions_replayed` is already logged in full detail
                // (bytes read, the tripping partition, the configured
                // budget) at the point `replay_scoped_partitions` detected
                // the trip; this shorter line ties that event to the sealed
                // error a caller actually receives, and — in tests — is the
                // asserted proof that replay stopped before every partition
                // in scope was read.
                tracing::debug!(
                    partitions_replayed,
                    scope = scope_label,
                    "translating a byte-budget replay abort to the caller-facing rejection"
                );
                // Record the over-budget size so an operator sizing
                // `max_source_bytes` sees these queries in
                // `polychrome_query_replayed_bytes` too — the count-based
                // `record_query_observed` never runs on this early-abort path.
                crate::metrics::record_query_replayed_bytes(scope_label, bytes_read);
                crate::metrics::record_source_budget_exceeded(scope_label);
                let message = match &self.scope {
                    QueryScope::Fleet => FLEET_BYTES_BUDGET_EXCEEDED_MESSAGE,
                    QueryScope::Conversations(_) => CONVERSATION_BYTES_BUDGET_EXCEEDED_MESSAGE,
                };
                ScopedQueryError::SourceBudgetExceeded(message.to_owned())
            }
        }
    }

    /// (QRY-3) Sum [`PartitionJournal::partition_event_count`] across
    /// this scope's own partitions, discovering them the identical way
    /// [`ScopedQuery::resolve_partitions_cached`]/[`ScopedQuery::replay_scoped_partitions`]
    /// do: [`QueryScope::Fleet`] reads every `conv-` partition
    /// [`PartitionJournal::list_partitions`] reports, skipping (and logging, QRY-7's
    /// lenient posture — this probe never touches `skipped_partitions`
    /// itself, since it exists only to feed a budget check, not the caller's
    /// own completeness accounting) one it cannot read; `QueryScope::Conversations`
    /// reads exactly its own named partitions, and any one read failure is a
    /// real error for that scope.
    ///
    /// Called by [`ScopedQuery::execute`] BEFORE [`ScopedQuery::resolve_partitions`]
    /// — see the module doc's QRY-3 section for why this O(1) pre-check,
    /// not merely the post-replay [`ScopedQuery::enforce_source_budget`]
    /// call further down, is what keeps this crate's reject-before-decode
    /// invariant true now that a cache lookup (and therefore a possible
    /// decode) happens inside `resolve_partitions` itself.
    async fn estimate_source_event_total(&self) -> Result<usize, ReplayError> {
        let partition_names: Vec<String> = match &self.scope {
            QueryScope::Fleet => self
                .journal
                .list_partitions()
                .await
                .map_err(|err| {
                    tracing::error!(
                        error = %err,
                        "failed to list partitions while estimating a fleet query's pre-decode \
                         source-event total"
                    );
                    ReplayError::Internal
                })?
                .into_iter()
                .filter(|partition| is_admitted_partition(partition))
                .collect(),
            QueryScope::Conversations(conversation_ids) => conversation_ids
                .iter()
                .map(|conversation_id| format!("conv-{conversation_id}"))
                .collect(),
        };

        let mut skipped = 0_usize;
        let mut total: u64 = 0;
        for partition in partition_names {
            match self.journal.partition_event_count(partition.clone()).await {
                Ok(count) => total = total.saturating_add(count),
                Err(err) => {
                    self.handle_skippable_partition_error(
                        &err,
                        &partition,
                        "read the conversation's own partition event count while estimating its \
                         pre-decode source-event total",
                        &mut skipped,
                    )?;
                }
            }
        }
        Ok(usize::try_from(total).unwrap_or(usize::MAX))
    }

    /// Shared Fleet-skip/Conversations-fail dispatch for a per-partition read
    /// failure — repeated identically at [`ScopedQuery::estimate_source_event_total`]'s
    /// own count read and, before this extraction, three times inside
    /// [`ScopedQuery::resolve_partitions_cached`] (its own count read, tail
    /// replay, and full/bounded replay). [`QueryScope::Fleet`] logs a
    /// warning and counts the partition as skipped — QRY-7's lenient
    /// posture, mirroring [`ScopedQuery::replay_scoped_partitions`]'s own
    /// Fleet arm; any other scope is a real error for that scope, EXCEPT the
    /// routine scheduler's own partition under [`QueryScope::Conversations`]
    /// (issue #1882): that one partition is opportunistic added data this
    /// scope's own participation never promised (see
    /// `replay_scoped_partitions`'s own Conversations-arm admission comment),
    /// so it is logged-and-skipped WITHOUT incrementing `skipped_partitions`
    /// — that count tracks only this scope's own, legitimately-owned
    /// conversation partitions, matching the uncached arm's identical
    /// posture exactly. `context` names the specific read that failed, for
    /// the log line.
    ///
    /// Returns `Ok(())` for a lenient skip (the caller's own loop iteration
    /// is done — `continue`) and `Err(ReplayError::Internal)` otherwise (the
    /// caller propagates via `?`).
    fn handle_skippable_partition_error(
        &self,
        err: &JournalError,
        partition: &str,
        context: &str,
        skipped_partitions: &mut usize,
    ) -> Result<(), ReplayError> {
        if matches!(self.scope, QueryScope::Conversations(_))
            && partition == ROUTINE_SCHEDULER_PARTITION
        {
            tracing::warn!(
                error = %err,
                partition = %partition,
                "skipping the unreadable/not-yet-existing routine scheduler partition for a \
                 persona-scoped query"
            );
            return Ok(());
        }
        if matches!(self.scope, QueryScope::Fleet) {
            tracing::warn!(
                error = %err,
                partition = %partition,
                "skipping unreadable partition for a fleet query"
            );
            *skipped_partitions += 1;
            return Ok(());
        }
        tracing::error!(
            error = %err,
            partition = %partition,
            "failed to {}",
            context
        );
        Err(ReplayError::Internal)
    }

    /// Fire [`ScopedQuery::race_inject_after_count_read`] for `partition`, if
    /// armed and targeting this exact partition — test-only, see that
    /// field's own doc. Consumes the armed injection (`.take()`), so it
    /// fires at most once per [`ScopedQuery`] regardless of how many
    /// partitions or queries this session goes on to resolve.
    #[cfg(test)]
    async fn fire_race_test_hook(&self, partition: &str) -> Result<(), ReplayError> {
        let injected = {
            let mut slot = self.race_inject_after_count_read.lock().expect("poison");
            match slot.as_ref() {
                Some((target, _)) if target == partition => slot.take(),
                _ => None,
            }
        };
        let Some((target, events)) = injected else {
            return Ok(());
        };
        self.journal
            .append_batch(target, events)
            .await
            .map_err(|err| {
                tracing::error!(error = %err, partition = %partition, "test race injection append failed");
                ReplayError::Internal
            })?;
        Ok(())
    }

    /// Shared byte-budget-exceeded bookkeeping for
    /// [`ScopedQuery::resolve_partitions_cached`]'s tail and miss arms —
    /// both call a [`PartitionJournal`] replay method that can only report the
    /// trip AFTER the read already returned (unlike
    /// [`ScopedQuery::replay_scoped_partitions`]'s own STREAMING bounded
    /// replay), so both arms need this identical log-then-`Err` step once
    /// their own replay/accounting finished. `partitions_already_resolved`
    /// is `tables.len()` at the call site — the partition that just tripped
    /// the budget is not yet pushed onto `tables`, so the reported count is
    /// `+ 1`. `context` names which arm tripped it (`"mid-tail-replay"`/
    /// `"mid-replay"`) for the log line.
    fn cached_bytes_budget_exceeded(
        partitions_already_resolved: usize,
        partition: &str,
        replayed_bytes: u64,
        max_bytes: u64,
        context: &'static str,
    ) -> ReplayError {
        let partitions_replayed = partitions_already_resolved + 1;
        tracing::error!(
            bytes_read = replayed_bytes,
            max_source_bytes = max_bytes,
            partition = %partition,
            partitions_replayed,
            "cache-aware query's replayed source bytes exceeded the pre-execution byte budget \
             {}; aborting before the rest of this scope is read",
            context
        );
        ReplayError::BytesBudgetExceeded {
            partitions_replayed,
            bytes_read: replayed_bytes,
        }
    }

    /// Resolve every one of this scope's own partitions into already-decoded
    /// [`PartitionTables`], either via [`ScopedQuery::resolve_partitions_uncached`]
    /// (the cache disabled — every partition is a genuine miss, decoded via
    /// [`crate::engine::decode_partition_tables`]) or
    /// [`ScopedQuery::resolve_partitions_cached`] (Phase A's decode cache —
    /// see `crate::cache`'s module doc for the full lookup protocol). Either
    /// way, this is the ONE place [`ScopedQuery::execute`] calls to get from
    /// "a verified scope" to "decoded tables ready for
    /// `crate::engine::QueryEngine::build_from_tables`".
    async fn resolve_partitions(&self) -> Result<ResolvedPartitions, ReplayError> {
        if self.cache.enabled() {
            self.resolve_partitions_cached().await
        } else {
            self.resolve_partitions_uncached().await
        }
    }

    /// The cache-disabled path: exactly [`ScopedQuery`]'s pre-Phase-A
    /// behavior — [`ScopedQuery::replay_scoped_partitions`] unchanged, then
    /// [`crate::engine::decode_partition_tables`] over every replayed
    /// partition. `cached_volume_events` mirrors `replayed_events` here:
    /// with no cache, every partition genuinely was replayed, so there is no
    /// separate "effective volume" to track — see
    /// [`ScopedQuery::enforce_cached_volume_budget`]'s doc for why this
    /// field is only ever CHECKED when the cache is enabled regardless.
    async fn resolve_partitions_uncached(&self) -> Result<ResolvedPartitions, ReplayError> {
        let (partitions, skipped_partitions) = self.replay_scoped_partitions().await?;
        let replayed_events: usize = partitions.iter().map(|p| p.events.len()).sum();
        let replayed_bytes: u64 = partitions
            .iter()
            .flat_map(|p| &p.events)
            .map(|(_position, event)| event.payload.len() as u64)
            .sum();
        let tables = partitions
            .iter()
            .map(|partition| {
                decode_partition_tables(
                    &partition.partition,
                    &partition.events,
                    &self.trusted_signers,
                    &self.handoff_trust,
                )
                .map_err(|err| {
                    tracing::error!(
                        error = %err,
                        partition = %partition.partition,
                        "failed to decode a replayed partition"
                    );
                    ReplayError::Internal
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        Ok(ResolvedPartitions {
            tables,
            skipped_partitions,
            replayed_events,
            replayed_bytes,
            cached_volume_events: replayed_events,
            cached_volume_bytes: replayed_bytes,
        })
    }

    /// The cache-enabled path: for each of this scope's own partitions, read
    /// its current event count and mutation epoch (both O(1)), consult
    /// [`DecodeCache::lookup`], and resolve a [`Lookup::Hit`]/[`Lookup::Tail`]/
    /// [`Lookup::Miss`] into a decoded [`PartitionTables`] — see
    /// `crate::cache`'s module doc for the full protocol.
    ///
    /// Mirrors [`ScopedQuery::replay_scoped_partitions`]'s own Fleet/
    /// Conversations discovery split: [`QueryScope::Fleet`] skips (and
    /// counts) an unreadable partition rather than failing the whole query;
    /// [`QueryScope::Conversations`] never skips — a failure reading any one
    /// partition's own event count or replaying its tail/full content is a
    /// real error for that scope.
    ///
    /// # The tail's byte budget is enforced DURING replay, not after (closed gap)
    ///
    /// [`QueryLimits::max_source_bytes`]'s mid-stream early-abort (issue
    /// #1541) now applies identically to both arms. A [`Lookup::Miss`]
    /// partition's full replay uses
    /// [`PartitionJournal::replay_with_positions_bounded`]; a [`Lookup::Tail`]
    /// partition's tail replay uses
    /// [`PartitionJournal::replay_from_with_positions_bounded`], the combined
    /// resume-plus-budget primitive that closed a real gap: before it
    /// existed, a tail replay had no byte cap of its own and could
    /// materialize an entire oversized tail into memory — a single large
    /// ordinary append (e.g. a big tool result) landing on an
    /// already-cached partition was enough to trigger it, no adversarial
    /// input required — before this method's own post-replay
    /// `replayed_bytes > max_bytes` check ever ran. Both host calls now stop
    /// pulling from the journal's own replay stream the instant cumulative
    /// payload bytes cross what they were given, so a single partition's
    /// tail can trip the budget without ever finishing its own replay, and
    /// this method still guarantees a later partition is never reached once
    /// the budget trips — the same completeness property
    /// `replay_scoped_partitions`'s own doc establishes.
    ///
    /// # A tail that trips the budget is a hard error, not a silent partial merge — and why a "fall back to a full rebuild" degrade does not help
    ///
    /// When a [`Lookup::Tail`]'s bounded replay reports `budget_exceeded`,
    /// this method returns [`ReplayError::BytesBudgetExceeded`] — it does
    /// NOT silently merge the partial tail it did manage to read and serve
    /// an incomplete result as if it were complete. That mirrors the
    /// [`Lookup::Miss`] arm exactly (deliberately: a `COUNT(*)` or similar
    /// aggregate that silently dropped the newest events because a byte cap
    /// tripped would return a WRONG answer with no signal it was wrong,
    /// which is strictly worse than a caller-visible rejection).
    ///
    /// A tempting alternative is to degrade instead of failing outright: on
    /// `budget_exceeded`, discard the stale cached `base` and re-resolve the
    /// partition as a fresh [`Lookup::Miss`] (a full bounded rebuild from
    /// position `0`). This is NOT implemented, because it cannot reduce how
    /// often a real query gets rejected: a full rebuild from `0` always
    /// needs to replay AT LEAST as many bytes as the tail-only replay from
    /// `from` alone (everything the tail would have read, PLUS everything
    /// before `from` that the cache already held) — so any tail whose own
    /// bytes already exceed the budget makes a full-rebuild fallback
    /// exceed it too, never less. The only real lever for reducing
    /// rejections on legitimately large-but-normal conversations is the
    /// budget itself — see [`QueryLimits::max_source_bytes`]'s own doc for
    /// the current default and the reasoning behind it.
    ///
    /// # The stale-watermark race, and how this method avoids it
    ///
    /// Each iteration reads `count`/`epoch` in one round trip, then — for a
    /// [`Lookup::Tail`]/[`Lookup::Miss`] — replays in a SEPARATE, later round
    /// trip. A partition can grow in between: an append landing in that
    /// window makes the replay genuinely longer than `count` implied. This
    /// method NEVER stores `count` itself as the cache entry's watermark —
    /// [`DecodeCache::store_tail`]/[`DecodeCache::store_full`] are always
    /// called with the watermark DERIVED from what the replay actually
    /// returned (its last event's own journal position, plus one; an empty
    /// tail replay keeps the prior watermark, `from`, unchanged). Storing
    /// the stale `count` instead — this crate's behavior before this fix —
    /// left a later [`Lookup::Tail`] replay a range already inside the
    /// cached tables (the entry claimed a smaller watermark than what it
    /// actually held), duplicating rows on the next
    /// [`crate::engine::PartitionTables::concat`]. See `crate::cache`'s
    /// module doc for the cache side of this same invariant, and
    /// `crate::authority::tests::cache_tail_race_does_not_duplicate_rows_across_an_interleaved_append`
    /// for the regression test.
    #[allow(
        clippy::too_many_lines,
        reason = "one cohesive per-partition resolution loop; the Hit/Tail/Miss arms share \
                  watermark derivation and are clearer inline than split, mirroring \
                  replay_scoped_partitions's own allow"
    )]
    async fn resolve_partitions_cached(&self) -> Result<ResolvedPartitions, ReplayError> {
        let max_bytes = self.limits.max_source_bytes;
        let mut tables = Vec::new();
        let mut skipped_partitions = 0_usize;
        let mut replayed_events = 0_usize;
        let mut replayed_bytes: u64 = 0;
        let mut cached_volume_events: u64 = 0;

        let partition_names: Vec<String> = match &self.scope {
            QueryScope::Fleet => self
                .journal
                .list_partitions()
                .await
                .map_err(|err| {
                    tracing::error!(error = %err, "failed to list partitions for a fleet query");
                    ReplayError::Internal
                })?
                .into_iter()
                .filter(|partition| is_admitted_partition(partition))
                .collect(),
            QueryScope::Conversations(conversation_ids) => {
                let mut names: Vec<String> = conversation_ids
                    .iter()
                    .map(|conversation_id| format!("conv-{conversation_id}"))
                    .collect();
                // Mirrors `replay_scoped_partitions`'s own Conversations arm
                // (issue #1882): a verified persona-scoped session ADDITIONALLY
                // reads the routine scheduler's own dedicated partition, never
                // a conversation grant (`caller_identity` is `Some` only for
                // `Principal::Persona` — see that field's own doc). The
                // per-partition loop below treats a read failure on THIS one
                // partition leniently (skip-and-log), exactly as the uncached
                // arm does — see this method's own per-partition error
                // handling and `handle_skippable_partition_error`'s doc.
                if self.caller_identity.is_some() {
                    names.push(ROUTINE_SCHEDULER_PARTITION.to_owned());
                }
                names
            }
        };

        for partition in partition_names {
            let count = match self.journal.partition_event_count(partition.clone()).await {
                Ok(count) => count,
                Err(err) => {
                    self.handle_skippable_partition_error(
                        &err,
                        &partition,
                        "read the conversation's own partition event count",
                        &mut skipped_partitions,
                    )?;
                    continue;
                }
            };
            // Test-only race injection — see `Self::race_inject_after_count_read`'s
            // own doc. A no-op in production: the field never carries an
            // entry outside `#[cfg(test)]` code.
            #[cfg(test)]
            self.fire_race_test_hook(&partition).await?;

            let (resolved_tables, watermark) = match self.cache.lookup(&partition, count) {
                Lookup::Hit(cached) => (cached, count),
                Lookup::Tail { base, from } => {
                    let remaining = max_bytes.saturating_sub(replayed_bytes);
                    let bounded = match self
                        .journal
                        .replay_from_with_positions_bounded(partition.clone(), from, remaining)
                        .await
                    {
                        Ok(bounded) => bounded,
                        Err(err) => {
                            self.handle_skippable_partition_error(
                                &err,
                                &partition,
                                "replay the conversation's own partition tail",
                                &mut skipped_partitions,
                            )?;
                            continue;
                        }
                    };
                    replayed_events += bounded.events.len();
                    replayed_bytes = replayed_bytes.saturating_add(bounded.bytes_read);
                    let budget_exceeded = bounded.budget_exceeded;

                    // See this method's own "The stale-watermark race" doc
                    // section: derived from what this call actually
                    // replayed, never from `count`. An empty tail (nothing
                    // new since `from`) keeps the prior watermark.
                    let watermark = bounded
                        .events
                        .last()
                        .map_or(from, |(position, _)| position + 1);

                    let tail_tables = decode_partition_tables(
                        &partition,
                        &bounded.events,
                        &self.trusted_signers,
                        &self.handoff_trust,
                    )
                    .map_err(|err| {
                        tracing::error!(
                            error = %err,
                            partition = %partition,
                            "failed to decode a partition's replayed tail"
                        );
                        ReplayError::Internal
                    })?;
                    let merged = base.concat(&tail_tables).map_err(|err| {
                        tracing::error!(
                            error = %err,
                            partition = %partition,
                            "failed to merge a partition's cached tables with its replayed tail"
                        );
                        ReplayError::Internal
                    })?;
                    // Stored even when `budget_exceeded` below returns an
                    // error: `watermark` is derived from what THIS call
                    // actually replayed (never `count`), so the entry
                    // correctly reflects a partial-but-honest resync point,
                    // never a truncated result claimed as complete. The next
                    // query resumes the rest of this same tail from here,
                    // rather than re-paying for the prefix this call already
                    // read.
                    self.cache.store_tail(&partition, watermark, merged.clone());

                    if budget_exceeded {
                        return Err(Self::cached_bytes_budget_exceeded(
                            tables.len(),
                            &partition,
                            replayed_bytes,
                            max_bytes,
                            "mid-tail-replay",
                        ));
                    }
                    (merged, watermark)
                }
                Lookup::Miss => {
                    let remaining = max_bytes.saturating_sub(replayed_bytes);
                    let bounded = match self
                        .journal
                        .replay_with_positions_bounded(partition.clone(), remaining)
                        .await
                    {
                        Ok(bounded) => bounded,
                        Err(err) => {
                            self.handle_skippable_partition_error(
                                &err,
                                &partition,
                                "replay the conversation's own partition",
                                &mut skipped_partitions,
                            )?;
                            continue;
                        }
                    };
                    replayed_events += bounded.events.len();
                    replayed_bytes = replayed_bytes.saturating_add(bounded.bytes_read);
                    let budget_exceeded = bounded.budget_exceeded;

                    // Same derivation as the `Tail` arm above — the count of
                    // events THIS decode actually covers, never `count`
                    // (which, under the byte budget's own early-abort, may
                    // even exceed what was actually replayed).
                    let watermark = bounded
                        .events
                        .last()
                        .map_or(0, |(position, _)| position + 1);

                    let fresh = decode_partition_tables(
                        &partition,
                        &bounded.events,
                        &self.trusted_signers,
                        &self.handoff_trust,
                    )
                    .map_err(|err| {
                        tracing::error!(
                            error = %err,
                            partition = %partition,
                            "failed to decode a freshly-replayed partition"
                        );
                        ReplayError::Internal
                    })?;
                    self.cache.store_full(&partition, watermark, fresh.clone());

                    if budget_exceeded {
                        return Err(Self::cached_bytes_budget_exceeded(
                            tables.len(),
                            &partition,
                            replayed_bytes,
                            max_bytes,
                            "mid-replay",
                        ));
                    }
                    (fresh, watermark)
                }
            };

            cached_volume_events = cached_volume_events.saturating_add(watermark);
            tables.push(resolved_tables);
        }

        let cached_volume_bytes: u64 = tables.iter().map(|t| t.memory_bytes() as u64).sum();
        Ok(ResolvedPartitions {
            tables,
            skipped_partitions,
            replayed_events,
            replayed_bytes,
            cached_volume_events: usize::try_from(cached_volume_events).unwrap_or(usize::MAX),
            cached_volume_bytes,
        })
    }

    /// Discover and replay this scope's own partitions, alongside a count of
    /// how many were skipped as unreadable (QRY-7).
    ///
    /// # Byte budget enforced DURING replay, not after (issue #1541)
    ///
    /// Before this fix, this method replayed EVERY scoped partition — for
    /// [`QueryScope::Fleet`], every `conv-` partition in the whole
    /// deployment — fully into a `Vec<PartitionEvents>` before
    /// [`ScopedQuery::enforce_source_budget`] (an event-COUNT check) ever
    /// ran, so a single Fleet query could exhaust the pod's real memory
    /// before any budget bit. This method now calls
    /// [`PartitionJournal::replay_with_positions_bounded`]
    /// once per partition, passing it the REMAINING byte budget
    /// (`crate::engine::QueryLimits::max_source_bytes` minus what has
    /// already been read across earlier partitions this same call). That
    /// call stops pulling from the journal's own replay stream the instant
    /// cumulative payload bytes cross what it was given — see
    /// [`polyc_eventlog::EventLog::replay_with_positions_bounded`]'s doc for
    /// the exact mechanics — so a single oversized partition can trip the
    /// budget WITHOUT ever finishing its own replay, and a Fleet scope with
    /// many partitions can trip it without ever reaching a later partition
    /// at all. The moment any one call reports
    /// `BoundedReplay::budget_exceeded`, this method returns
    /// [`ReplayError::BytesBudgetExceeded`] immediately — the partition that
    /// tripped it is the LAST one this call ever reads.
    ///
    /// Only the [`QueryScope::Fleet`] arm ever skips a partition rather than
    /// failing the whole query — see this method's inline comment there —
    /// so the returned count is always `0` for [`QueryScope::Conversations`];
    /// that arm's own replay failure short-circuits the whole call with
    /// `Err(ReplayError::Internal)` instead, matching its existing "no
    /// skip-and-log fallback" posture. Before this count existed, a skipped
    /// Fleet partition was visible only as a server-side `tracing::warn!` —
    /// the caller received a result with no signal that it was computed
    /// over fewer partitions than actually exist, the exact completeness
    /// gap [`QueryLimits::row_cap`]'s `truncated` flag already closes for
    /// the ROW dimension.
    #[allow(
        clippy::too_many_lines,
        reason = "one cohesive replay routine; the Fleet and Conversations arms share the \
                  byte-budget/early-abort accounting and are clearer inline than split"
    )]
    async fn replay_scoped_partitions(&self) -> Result<(Vec<PartitionEvents>, usize), ReplayError> {
        let max_bytes = self.limits.max_source_bytes;
        let mut bytes_read: u64 = 0;
        match &self.scope {
            QueryScope::Fleet => {
                let partition_names = self.journal.list_partitions().await.map_err(|err| {
                    tracing::error!(error = %err, "failed to list partitions for a fleet query");
                    ReplayError::Internal
                })?;
                let mut partitions = Vec::new();
                let mut skipped = 0_usize;
                for partition in partition_names {
                    // Narrow, deliberate admission of the scheduler's own
                    // NON-conversation partition (issue #1592): every other
                    // partition this crate ever reads is a conversation
                    // partition (`"conv-{id}"`), and this hard filter used to
                    // reject anything else outright. `"routine-scheduler"` is
                    // the ONE exception — the routine scheduler's durable
                    // `routine_fired` markers (`fires`' one source table,
                    // `crate::decode::fires`) live there, never inside any
                    // conversation's own partition, because a fixed-content
                    // firing opens no turn
                    // (`crates/control-plane/src/routine_scheduler.rs`'s
                    // `ROUTINE_SCHEDULER_PARTITION` doc). This admits exactly
                    // that one literal name, Fleet-scope only — a
                    // `QueryScope::Conversations` replay (below) never lists
                    // partitions at all, so it can never reach this partition
                    // regardless; no OTHER non-`conv-` partition is ever
                    // admitted here.
                    if !is_admitted_partition(&partition) {
                        continue;
                    }
                    let remaining = max_bytes.saturating_sub(bytes_read);
                    match self
                        .journal
                        .replay_with_positions_bounded(partition.clone(), remaining)
                        .await
                    {
                        Ok(bounded) => {
                            bytes_read = bytes_read.saturating_add(bounded.bytes_read);
                            let budget_exceeded = bounded.budget_exceeded;
                            partitions.push(PartitionEvents {
                                partition: partition.clone(),
                                events: bounded.events,
                            });
                            if budget_exceeded {
                                let partitions_replayed = partitions.len();
                                tracing::error!(
                                    bytes_read,
                                    max_source_bytes = max_bytes,
                                    partition = %partition,
                                    partitions_replayed,
                                    "fleet query's replayed source bytes exceeded the \
                                     pre-execution byte budget mid-replay; aborting before the \
                                     rest of the deployment is read"
                                );
                                return Err(ReplayError::BytesBudgetExceeded {
                                    partitions_replayed,
                                    bytes_read,
                                });
                            }
                        }
                        Err(err) => {
                            tracing::warn!(
                                error = %err,
                                partition = %partition,
                                "skipping unreadable partition for a fleet query"
                            );
                            skipped += 1;
                        }
                    }
                }
                Ok((partitions, skipped))
            }
            QueryScope::Conversations(conversation_ids) => {
                let mut partitions = Vec::with_capacity(conversation_ids.len());
                for conversation_id in conversation_ids {
                    let partition = format!("conv-{conversation_id}");
                    let remaining = max_bytes.saturating_sub(bytes_read);
                    let bounded = self
                        .journal
                        .replay_with_positions_bounded(partition.clone(), remaining)
                        .await
                        .map_err(|err| {
                            tracing::error!(
                                error = %err,
                                conversation_id = %conversation_id,
                                "failed to replay the conversation's own partition"
                            );
                            ReplayError::Internal
                        })?;
                    bytes_read = bytes_read.saturating_add(bounded.bytes_read);
                    let budget_exceeded = bounded.budget_exceeded;
                    partitions.push(PartitionEvents {
                        partition: partition.clone(),
                        events: bounded.events,
                    });
                    if budget_exceeded {
                        let partitions_replayed = partitions.len();
                        tracing::error!(
                            bytes_read,
                            max_source_bytes = max_bytes,
                            partition = %partition,
                            partitions_replayed,
                            conversations_in_scope = conversation_ids.len(),
                            "conversation-scoped query's replayed source bytes exceeded the \
                             pre-execution byte budget mid-replay; aborting before the rest of \
                             this scope is read"
                        );
                        return Err(ReplayError::BytesBudgetExceeded {
                            partitions_replayed,
                            bytes_read,
                        });
                    }
                }
                // #1882: a persona-scoped session (`caller_identity` is
                // `Some` only for `Principal::Persona` — see
                // `Scoping::for_conversation`'s own doc for why a
                // conversation grant's `caller_identity` is always `None`)
                // additionally gets the routine scheduler's own dedicated
                // partition admitted here, narrowly and leniently: unlike
                // every conversation partition above (this scope's own
                // participation, a real failure reading any one of them is a
                // hard error), the scheduler partition is OPPORTUNISTIC
                // added data this scope's participation never promised —
                // its rows only ever surface through `fires`' own
                // owner-filtering join against this persona's own
                // `routines` (`crate::views::FIRES_OWNED_VIEW_SQL`), so an
                // UNREADABLE or not-yet-existing scheduler partition must
                // never fail a persona's otherwise-healthy conversation
                // query (the `Err(err)` arm below, skip-and-log). This
                // leniency is about READABILITY only: a `budget_exceeded`
                // reading a scheduler partition that DOES exist still
                // returns `Err(ReplayError::BytesBudgetExceeded)` below,
                // deliberately, the SAME hard-fail-the-whole-query posture
                // the Fleet arm above already gives that identical
                // partition — QRY-3/#1541's byte budget is a global
                // resource-protection backstop, not a per-partition
                // opportunism switch, so it stays load-bearing here exactly
                // as it is everywhere else in this method.
                if self.caller_identity.is_some() {
                    let remaining = max_bytes.saturating_sub(bytes_read);
                    match self
                        .journal
                        .replay_with_positions_bounded(
                            ROUTINE_SCHEDULER_PARTITION.to_owned(),
                            remaining,
                        )
                        .await
                    {
                        Ok(bounded) => {
                            bytes_read = bytes_read.saturating_add(bounded.bytes_read);
                            let budget_exceeded = bounded.budget_exceeded;
                            partitions.push(PartitionEvents {
                                partition: ROUTINE_SCHEDULER_PARTITION.to_owned(),
                                events: bounded.events,
                            });
                            if budget_exceeded {
                                let partitions_replayed = partitions.len();
                                tracing::error!(
                                    bytes_read,
                                    max_source_bytes = max_bytes,
                                    partition = ROUTINE_SCHEDULER_PARTITION,
                                    partitions_replayed,
                                    "persona-scoped query's replayed source bytes exceeded the \
                                     pre-execution byte budget reading the routine scheduler's \
                                     own partition; aborting"
                                );
                                return Err(ReplayError::BytesBudgetExceeded {
                                    partitions_replayed,
                                    bytes_read,
                                });
                            }
                        }
                        Err(err) => {
                            tracing::warn!(
                                error = %err,
                                partition = ROUTINE_SCHEDULER_PARTITION,
                                "skipping the unreadable/not-yet-existing routine scheduler \
                                 partition for a persona-scoped query"
                            );
                        }
                    }
                }
                // Never skips a CONVERSATION partition — a replay failure
                // above already returned `Err(ReplayError::Internal)`, so
                // reaching here means every one of this scope's own
                // conversation partitions succeeded (and none tripped the
                // byte budget). The scheduler-partition admission just above
                // is the one exception to "never skips": that partition was
                // never part of what this scope's own participation
                // promised, so its own failure is logged-and-skipped rather
                // than counted in `skipped_partitions` (which names only
                // conversation partitions this scope legitimately owns) or
                // treated as a hard error.
                Ok((partitions, 0))
            }
        }
    }

    /// Resolve every Fleet reference table
    /// (`personas`/`participations`/`persona_identities`/`persona_wallets`/
    /// `persona_spend_policies`/`persona_credentials`/`persona_usage`) from
    /// the union of two candidate sources, then ONE
    /// [`polyc_persona::PersonaReferenceSnapshot`] fetch per candidate
    /// (#1578, Phase D) — not five separate `PersonaHost` calls, the shape a
    /// naive per-table build would otherwise pay. `routines` and `dashboard`
    /// are resolved separately, outside this candidate union — see their own
    /// paragraphs below.
    ///
    /// **Candidate discovery — two sources, two DIFFERENT gaps, do not
    /// conflate them:**
    /// 1. Every distinct, non-empty `persona_id` recorded on a `caller`/
    ///    `participant` attribution event, read from `partition_tables`'
    ///    already-decoded `attribution_raw` batches (`PersonaHost` exposes no
    ///    full-store enumeration primitive — see the doc this carries forward
    ///    from `crates/control-plane/src/query_http.rs`'s pre-retrofit
    ///    `attributed_persona_ids`/`fetch_reference_data`). Reading the
    ///    already-decoded `attribution_raw` column here — rather than
    ///    re-parsing raw `caller`/`participant` events directly, as this
    ///    method did before Phase A's decode cache — is not merely a cache
    ///    accommodation: `crate::engine::decode_partition_tables` already
    ///    performs the identical kind-filter-and-decode
    ///    (`crate::decode::attribution::decode_attribution_events`) to build
    ///    that column, so scanning it here is one fewer independently
    ///    maintained copy of that same decode, not a second one. Narrowed by
    ///    QRY-7's budget-abort/skipped-partition posture: a partition this
    ///    scope's replay skipped or a query the source-event/bytes budget
    ///    aborted early never contributes its attribution events, so a
    ///    persona ONLY attributed inside one of those partitions is
    ///    invisible from this source alone.
    /// 2. [`polyc_persona::PersonaHost::usage_rollup_index`] — every persona
    ///    id currently holding a maintained usage rollup, fleet-wide,
    ///    independent of which partitions this query happened to replay.
    ///    Unioning this in closes gap 1's budget-abort/skipped-partition
    ///    narrowing (a persona with a rollup surfaces even if the turn that
    ///    built the rollup lives in a partition this query skipped).
    ///
    /// These two sources do NOT together make this a full-store scan: a
    /// persona that was provisioned but has never accrued a committed turn
    /// as caller (no rollup, no attribution event in ANY partition this
    /// process has ever replayed) is invisible to BOTH sources — a
    /// SEPARATE, still-open gap from the one the union closes. Do not
    /// conflate the two: the union closes the budget-abort narrowing, and it
    /// does NOT make a never-active persona visible. A true
    /// full-scan primitive exists as precedent if that gap ever needs
    /// closing: [`polyc_persona::PersonaHost::personas_with_unverified_passkey`]
    /// (`crates/persona/src/host.rs`) already walks every persona record.
    ///
    /// `routines` is resolved via [`Self::resolve_routines`] — see that
    /// method's own doc. `dashboard` is a full snapshot of
    /// [`crate::dashboard::DashboardProjection::rows`] — unlike every
    /// persona-side table above, it is NOT narrowed to any candidate set at
    /// all, since the dashboard/conversations list is a fleet-wide view by
    /// design (mirrors [`crate::dashboard::DashboardProjection`]'s own "one
    /// row per known conversation" contract, not a per-query filtered
    /// subset).
    ///
    /// Every table but `routines` stays empty for every non-Fleet scope
    /// (never even attempted — those tables are Fleet-only regardless, see
    /// `crate::engine`'s module docs); empty (not a failure) when the
    /// persona store is unavailable or an individual persona's snapshot
    /// fails to resolve — the same lenient, one-bad-row-must-not-fail-the-
    /// whole-query posture an unreadable partition already gets above.
    ///
    /// `owner_persona_id` (issue #1882) is `Some(id)` for a verified
    /// persona-scoped session (`ScopedQuery::execute` derives it from this
    /// session's own `caller_identity`, which is `Some` only for
    /// [`Principal::Persona`] — never [`Principal::ConversationGrant`], see
    /// that principal's own doc), `None` otherwise. When `Some`, this method
    /// returns EARLY with [`ReferenceData::empty_except_routines`]: every
    /// Fleet-only table above stays empty, but `routines` is resolved and
    /// filtered to exactly that persona's own rows — see
    /// [`Self::resolve_routines`]'s own doc for the filter.
    async fn resolve_reference_data(
        &self,
        partition_tables: &[PartitionTables],
        owner_persona_id: Option<&str>,
    ) -> ReferenceData {
        if !matches!(self.scope, QueryScope::Fleet) {
            return ReferenceData::empty_except_routines(
                self.resolve_routines(owner_persona_id).await,
            );
        }
        let dashboard_rows = self.dashboard.rows();
        let Some(persona) = self.persona.load_full() else {
            tracing::warn!(
                "persona host unavailable; the fleet query's persona reference tables build \
                 empty"
            );
            return ReferenceData::empty_except_dashboard(dashboard_rows);
        };

        let mut candidate_ids = std::collections::BTreeSet::new();
        for tables in partition_tables {
            let persona_id_column = tables
                .attribution_raw
                .column_by_name("persona_id")
                .and_then(|column| column.as_any().downcast_ref::<arrow::array::StringArray>());
            let Some(persona_id_column) = persona_id_column else {
                continue;
            };
            for persona_id in persona_id_column.iter().flatten() {
                if !persona_id.is_empty() {
                    candidate_ids.insert(persona_id.to_owned());
                }
            }
        }
        // Union in every persona currently holding a usage rollup — closes
        // the budget-abort/skipped-partition narrowing above; does NOT
        // surface a never-conversed persona (see this method's own doc).
        match persona.usage_rollup_index().await {
            Ok(ids) => candidate_ids.extend(ids),
            Err(err) => {
                tracing::warn!(
                    error = %err,
                    "usage-rollup-index union unavailable for a fleet query; candidate set \
                     falls back to attribution events only"
                );
            }
        }

        let mut seen_canonical = std::collections::BTreeSet::new();
        let mut personas = Vec::new();
        let mut participations = Vec::new();
        let mut wallets = Vec::new();
        let mut spend_policies = Vec::new();
        let mut credentials = Vec::new();
        let mut usage_rollups = Vec::new();
        for candidate in candidate_ids {
            let snapshot = match persona.reference_snapshot(candidate.clone()).await {
                Ok(Some(snapshot)) => snapshot,
                Ok(None) => continue,
                Err(err) => {
                    tracing::warn!(
                        error = %err,
                        persona_id = %candidate,
                        "skipping an unreadable persona's reference snapshot for a fleet query"
                    );
                    continue;
                }
            };
            let persona_id = snapshot.profile.persona_id.clone();
            if !seen_canonical.insert(persona_id.clone()) {
                // Already resolved via a different alias id — do not
                // duplicate the row.
                continue;
            }
            participations.extend(
                snapshot
                    .participations
                    .into_iter()
                    .map(|participation| (persona_id.clone(), participation)),
            );
            if let Some(wallet) = snapshot.wallet_link {
                wallets.push((persona_id.clone(), wallet));
            }
            if let Some(policy) = snapshot.spend_policy {
                spend_policies.push((persona_id.clone(), policy));
            }
            if let Some(credential) = snapshot.credential {
                credentials.push((persona_id.clone(), credential));
            }
            if let Some(usage_rollup) = snapshot.usage_rollup {
                usage_rollups.push((persona_id.clone(), usage_rollup));
            }
            personas.push(snapshot.profile);
        }

        // Fleet always resolves every routine, unfiltered — `owner_persona_id`
        // is `None` for every real Fleet caller (`ScopedQuery::execute`
        // never derives a Fleet session's own `owner_persona_id` as `Some`;
        // see that method's own doc), but this call still honors whatever
        // value the caller passed rather than hard-coding `None` here, so a
        // future caller cannot silently widen Fleet's own routines view by
        // mistake without this line also changing.
        let routines = self.resolve_routines(owner_persona_id).await;

        ReferenceData {
            personas,
            participations,
            wallets,
            spend_policies,
            credentials,
            usage_rollups,
            routines,
            dashboard: dashboard_rows,
        }
    }

    /// Resolve the `routines` reference table from this session's own
    /// [`RoutineCatalog`] handle, resolved FRESH on every call — never
    /// cached, the same posture [`Self::resolve_reference_data`] already
    /// gives `personas`/`participations`. Empty (not a failure) when no
    /// catalog is wired at all, or when the catalog read errors — a
    /// misbehaving or unconfigured routine catalog must never fail the
    /// whole query, the same lenient posture an unreadable partition or an
    /// unresolvable persona already get above.
    ///
    /// `owner_persona_id` (issue #1882) is `Some(id)` for a persona-scoped
    /// session — every row whose `creator_persona` is not exactly `id` is
    /// filtered out before this method ever returns, so the caller (this
    /// method's own caller, [`Self::resolve_reference_data`]) never even
    /// SEES another persona's routine, let alone hands it to
    /// `crate::engine`. `None` for [`Principal::Admin`] (fleet sees every
    /// routine, unfiltered) — the only two callers of this method
    /// ([`Self::resolve_reference_data`]'s Fleet and persona-scoped
    /// branches) are exactly the two scopes that ever reach it at all; a
    /// conversation-grant scope never calls this method (see that method's
    /// own doc).
    async fn resolve_routines(
        &self,
        owner_persona_id: Option<&str>,
    ) -> Vec<crate::routine_catalog::RoutineStatusRecord> {
        let Some(catalog) = &self.routine_catalog else {
            return Vec::new();
        };
        let routines = match catalog.list_routines().await {
            Ok(routines) => routines,
            Err(err) => {
                tracing::warn!(
                    error = %err,
                    "routine catalog unavailable; the query's routines table builds empty"
                );
                Vec::new()
            }
        };
        match owner_persona_id {
            Some(persona_id) => routines
                .into_iter()
                .filter(|routine| routine.creator_persona == persona_id)
                .collect(),
            None => routines,
        }
    }
}

#[cfg(test)]
mod tests;