polyc-query 2026.9.0

#![allow(clippy::pedantic, clippy::nursery, missing_docs)]

use std::sync::Arc;

use buffa::Message as _;
use polyc_crypto::approval::ApprovalSigner;
use polyc_crypto::session::{RevokedTokens, SessionScope, SessionSubject, mint_session};
use polyc_crypto::signing_role::{
    HandoffRole, RoleTrustSet, SessionSigner, TurnReadRole, TurnReadSigner,
};
use polyc_eventlog::Event;
use polyc_eventlog_host::{EventLogHost, RewriteDecision};
use polyc_persona::PersonaHost;
use polyc_proto::kinds;
use polyc_proto::proto::polychrome::events::v1::RoutineFiredEvent;
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
use tokio_util::sync::CancellationToken;

use crate::routine_catalog::{RoutineCatalog, RoutineCatalogError, RoutineStatusRecord};

use super::*;

const NOW: u64 = 1_700_000_000_000;
const TEST_TTL_MS: u64 = 8 * 60 * 60 * 1000;

/// The handoff trust used by each fixture authority. It matches
/// `crate::engine::fixture_handoff_signer`, so both entry points report
/// `verified`.
fn test_handoff_trust() -> RoleTrustSet<HandoffRole> {
    crate::engine::fixture_handoff_trust()
}

fn test_signer() -> ApprovalSigner {
    ApprovalSigner::from_seed(1)
}

/// One owner-scoped-routines fixture's `routines` catalog row: a ready,
/// unpaused, private routine named `name`, created by `creator_persona`. Its
/// own `uid` is derived deterministically from `name`
/// (`format!("{name}-uid")`), matching every other fixture in this module —
/// module-level (not per-test) so every test needing a plain, ordinary
/// routine row shares one definition rather than each hand-rolling its own
/// 18-field literal.
fn routine_record(name: &str, creator_persona: &str) -> RoutineStatusRecord {
    RoutineStatusRecord {
        name: name.to_owned(),
        uid: format!("{name}-uid"),
        fire_conversation_id: format!("{name}-fire-conv"),
        ready: true,
        phase: Some("Ready".to_owned()),
        message: None,
        last_fire_time_ms: None,
        next_fire_time_ms: None,
        conditions_json: "[]".to_owned(),
        creator_persona: creator_persona.to_owned(),
        provenance_conversation_id: "conv-1".to_owned(),
        schedule_json: r#"{"kind":"cron","expression":"0 9 * * *","timezone":null}"#.to_owned(),
        next_fires_json: "[]".to_owned(),
        suspended: false,
        paused_by: None,
        paused_at_ms: None,
        pause_reason: None,
        prompt: "post the morning standup".to_owned(),
        scope: "private".to_owned(),
        orphaned: false,
        display_name: String::new(),
        description: String::new(),
        schedule_timezone: "UTC".to_owned(),
    }
}

/// A fresh, empty dashboard cell over `eventlog` — none of this module's
/// tests exercise the `dashboard` reference table itself (that's
/// `crate::engine::tests`' and `crate::decode::dashboard::tests`' job), so an
/// unpopulated projection sharing the fixture's own eventlog handle is all a
/// `QueryAuthority` constructor here needs.
fn test_dashboard_cell(eventlog: &Arc<EventLogHost>) -> crate::dashboard::DashboardCell {
    crate::dashboard::DashboardProjection::new(
        Vec::new(),
        eventlog.clone(),
        polyc_payments::amount::DEFAULT_DECIMALS,
    )
}

struct Fixture {
    authority: QueryAuthority,
    eventlog: Arc<EventLogHost>,
    eventlog_shutdown: CancellationToken,
    eventlog_dir: std::path::PathBuf,
    persona: Arc<PersonaHost>,
    persona_shutdown: CancellationToken,
    persona_dir: std::path::PathBuf,
    signer: ApprovalSigner,
    revoked: Arc<RevokedTokens>,
}

impl Fixture {
    async fn build(test_name: &str) -> Self {
        let signer = test_signer();
        let eventlog_dir = std::env::temp_dir().join(format!(
            "polyc-query-authority-{test_name}-eventlog-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&eventlog_dir);
        let eventlog_shutdown = CancellationToken::new();
        let eventlog = Arc::new(
            EventLogHost::spawn(
                eventlog_dir.clone(),
                eventlog_shutdown.clone(),
                signer.relabel_for_test(),
            )
            .expect("spawn eventlog host"),
        );

        let persona_dir = std::env::temp_dir().join(format!(
            "polyc-query-authority-{test_name}-persona-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&persona_dir);
        let persona_shutdown = CancellationToken::new();
        let persona = Arc::new(
            PersonaHost::spawn(persona_dir.clone(), persona_shutdown.clone())
                .expect("spawn persona host"),
        );

        let revoked = Arc::new(RevokedTokens::new());
        let authority = QueryAuthority::new(
            eventlog.clone(),
            Arc::new(arc_swap::ArcSwapOption::new(Some(persona.clone()))),
            test_dashboard_cell(&eventlog),
            revoked.clone(),
            signer.public_key_bytes(),
            vec![signer.public_key_bytes()],
            QueryLimits::default(),
            None,
            CacheConfig::disabled(),
            test_handoff_trust(),
        );

        Self {
            authority,
            eventlog,
            eventlog_shutdown,
            eventlog_dir,
            persona,
            persona_shutdown,
            persona_dir,
            signer,
            revoked,
        }
    }

    /// Build a `QueryAuthority` whose persona cell is empty (the host never
    /// started) — the store-unavailable case.
    fn authority_with_empty_persona_cell(&self) -> QueryAuthority {
        QueryAuthority::new(
            self.eventlog.clone(),
            Arc::new(arc_swap::ArcSwapOption::empty()),
            test_dashboard_cell(&self.eventlog),
            self.revoked.clone(),
            self.signer.public_key_bytes(),
            vec![self.signer.public_key_bytes()],
            QueryLimits::default(),
            None,
            CacheConfig::disabled(),
            test_handoff_trust(),
        )
    }

    /// The one fully-parameterized `QueryAuthority` builder every
    /// `authority_with_*` helper below delegates to — each of those keeps its
    /// own name and signature (they're called by name across this file) but
    /// is now a one-line call through here, so the `eventlog`/`persona`/
    /// `dashboard`/`revoked`/`signer` wiring lives in exactly one place
    /// instead of drifting across near-identical copies. `catalog` and
    /// `limits` default to `None`/`QueryLimits::default()` respectively when
    /// a caller has nothing custom to supply for that field; `cache_config`
    /// has no default here since every call site already has an opinion
    /// (`CacheConfig::disabled()` or a caller-supplied one).
    fn authority_with(
        &self,
        catalog: Option<Arc<dyn RoutineCatalog>>,
        limits: Option<QueryLimits>,
        cache_config: CacheConfig,
    ) -> QueryAuthority {
        QueryAuthority::new(
            self.eventlog.clone(),
            Arc::new(arc_swap::ArcSwapOption::new(Some(self.persona.clone()))),
            test_dashboard_cell(&self.eventlog),
            self.revoked.clone(),
            self.signer.public_key_bytes(),
            vec![self.signer.public_key_bytes()],
            limits.unwrap_or_default(),
            catalog,
            cache_config,
            test_handoff_trust(),
        )
    }

    /// Build a `QueryAuthority` sharing this fixture's own eventlog/persona/
    /// revoked handles but with `catalog` as its `RoutineCatalog` — the
    /// `routines`/`fires` wiring tests use this to prove a Fleet session
    /// actually resolves through the caller-supplied catalog rather than
    /// silently building an empty table regardless.
    fn authority_with_routine_catalog(&self, catalog: Arc<dyn RoutineCatalog>) -> QueryAuthority {
        self.authority_with(Some(catalog), None, CacheConfig::disabled())
    }

    /// Build a `QueryAuthority` sharing this fixture's own eventlog/persona/
    /// revoked handles but under caller-supplied `limits` — QRY-3's source
    /// budget is exercised through a deliberately tiny
    /// `QueryLimits::max_source_events`, the same idiom
    /// `authority_with_empty_persona_cell` already uses for its own
    /// deliberately-broken persona cell.
    fn authority_with_limits(&self, limits: QueryLimits) -> QueryAuthority {
        self.authority_with(None, Some(limits), CacheConfig::disabled())
    }

    /// Build a `QueryAuthority` sharing this fixture's own eventlog/persona/
    /// revoked handles, with the decode cache turned ON under
    /// caller-supplied `cache_config` — the cache-lookup/erasure/eviction
    /// test suite's own fixture, mirroring `authority_with_limits`'s shape.
    fn authority_with_cache_config(&self, cache_config: CacheConfig) -> QueryAuthority {
        self.authority_with(None, None, cache_config)
    }

    /// Build a `QueryAuthority` sharing this fixture's own eventlog/persona/
    /// revoked handles, under BOTH caller-supplied `limits` and the decode
    /// cache's `cache_config` — needed together (unlike
    /// `authority_with_limits`/`authority_with_cache_config` alone) by the
    /// cache-tail byte-budget test, which needs the cache ON (to actually
    /// reach a `Lookup::Tail`) AND a deliberately tiny `max_source_bytes` (to
    /// trip the tail's own bounded replay).
    fn authority_with_limits_and_cache_config(
        &self,
        limits: QueryLimits,
        cache_config: CacheConfig,
    ) -> QueryAuthority {
        self.authority_with(None, Some(limits), cache_config)
    }

    /// Build a `QueryAuthority` sharing this fixture's own eventlog/persona/
    /// revoked handles, wired to caller-supplied `catalog` AND with the
    /// decode cache turned ON under caller-supplied `cache_config` — needed
    /// together by the cache-enabled `fires`/`routine_lifecycle` regression
    /// suite (issue #1592/#1882's decode-cache-path drift): the routine
    /// catalog is what lets `routines`/`fires`' owner-filtering join resolve
    /// at all (mirroring `authority_with_routine_catalog`), while the cache
    /// config is what actually routes discovery through
    /// `ScopedQuery::resolve_partitions_cached` instead of
    /// `ScopedQuery::resolve_partitions_uncached` (mirroring
    /// `authority_with_cache_config`) — production's exact shape per
    /// `crate::cache::CacheConfig::new`'s own doc (any single-replica
    /// deployment).
    fn authority_with_routine_catalog_and_cache_config(
        &self,
        catalog: Arc<dyn RoutineCatalog>,
        cache_config: CacheConfig,
    ) -> QueryAuthority {
        self.authority_with(Some(catalog), None, cache_config)
    }

    async fn make_admin(&self, label: &str) -> String {
        let identity = ExternalIdentity {
            provider: "test".to_owned(),
            scope: "s".to_owned(),
            external_id: label.to_owned(),
            display_name: label.to_owned(),
            ..Default::default()
        };
        self.persona
            .set_admin(identity, true, NOW)
            .await
            .expect("set_admin")
    }

    async fn make_non_admin(&self, label: &str) -> String {
        let identity = ExternalIdentity {
            provider: "test".to_owned(),
            scope: "s".to_owned(),
            external_id: label.to_owned(),
            display_name: label.to_owned(),
            ..Default::default()
        };
        // `attribute` resolves-or-provisions a fresh, non-admin persona for
        // an unseen identity (`resolve` alone is read-only and would return
        // `None` for one never seen before).
        self.persona
            .attribute(
                identity,
                format!("conv-{label}"),
                "initiator".to_owned(),
                NOW,
            )
            .await
            .expect("attribute")
            .persona_id
    }

    async fn teardown(self) {
        self.eventlog_shutdown.cancel();
        drop(self.eventlog);
        let _ = std::fs::remove_dir_all(&self.eventlog_dir);
        self.persona_shutdown.cancel();
        drop(self.persona);
        let _ = std::fs::remove_dir_all(&self.persona_dir);
    }
}

// ---- sealed-set pin: every item below is `pub(crate)` (or private), used
// here from within this crate to prove it compiles at that visibility. This
// test can only catch the seal being narrowed too far (an item this crate
// still needs demoted to something even this module can't reach) — it
// compiles identically whether an item is `pub(crate)` or widened back to
// `pub`, so it is NOT a backstop against a reviewer re-widening one. The real
// backstops against re-widening are `just arch`'s layer-boundary check
// (`crate::authority`'s own module doc, "Pinning the seal") and a
// cargo-doc/public-surface review at PR time.
#[test]
fn sealed_set_is_crate_only() {
    let _: crate::session::QueryScope = crate::session::QueryScope::Fleet;
    let _ = crate::engine::PartitionEvents {
        partition: "conv-x".to_string(),
        events: Vec::new(),
    };
    let _ = crate::engine::ReferenceData::empty();
    let _ = crate::provider::EventsTableProvider::new("conv-x");
    let _: &str = crate::views::COMMITTED_TURNS_VIEW_SQL;
    let _ = crate::statement_gate::check_statement_allowed("SELECT 1", false);
    // The documented bypass (`crate`'s own module doc): unredacted Arrow
    // data from a raw decode, reachable only from within this crate.
    let _ = crate::decode::events_batch("conv-x", &[]);
}

#[tokio::test]
async fn admin_session_valid_admin_scopes_fleet() {
    let fx = Fixture::build("admin-valid").await;
    let persona_id = fx.make_admin("alice").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );

    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("a valid admin session must verify");
    match &principal {
        Principal::Admin(admin) => assert_eq!(admin.persona_id(), persona_id),
        other => panic!("expected Principal::Admin, got {other:?}"),
    }

    let scoped = fx
        .authority
        .scope_for(&principal)
        .await
        .expect("Admin always scopes");
    assert!(matches!(scoped.scope, QueryScope::Fleet));
    assert_eq!(scoped.caller_identity(), Some(persona_id.as_str()));

    fx.teardown().await;
}

#[tokio::test]
async fn admin_session_no_token_is_invalid_session() {
    let fx = Fixture::build("admin-no-token").await;
    let err = fx
        .authority
        .verify_admin_session("not-a-real-token", NOW)
        .await
        .unwrap_err();
    assert!(matches!(err, PrincipalError::InvalidSession));
    fx.teardown().await;
}

#[tokio::test]
async fn admin_session_expired_is_invalid_session() {
    let fx = Fixture::build("admin-expired").await;
    let persona_id = fx.make_admin("bob").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        1_000,
    );

    let err = fx
        .authority
        .verify_admin_session(&token, NOW + 1_000)
        .await
        .unwrap_err();
    assert!(matches!(err, PrincipalError::InvalidSession));
    fx.teardown().await;
}

#[tokio::test]
async fn admin_session_revoked_is_invalid_session() {
    let fx = Fixture::build("admin-revoked").await;
    let persona_id = fx.make_admin("carol").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    fx.revoked.revoke(&token);

    let err = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .unwrap_err();
    assert!(matches!(err, PrincipalError::InvalidSession));
    fx.teardown().await;
}

/// A3: a valid session bound to a non-admin persona is no longer refused —
/// it mints a `Principal::Persona` instead.
#[tokio::test]
async fn admin_session_valid_non_admin_mints_a_persona_principal() {
    let fx = Fixture::build("admin-non-admin").await;
    let persona_id = fx.make_non_admin("dave").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );

    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("a valid non-admin session must verify as a persona principal");
    match &principal {
        Principal::Persona(persona) => assert_eq!(persona.persona_id(), persona_id),
        other => panic!("expected Principal::Persona, got {other:?}"),
    }
    fx.teardown().await;
}

/// A session whose persona id does not resolve to any profile at all (e.g.
/// the persona was never provisioned, or was removed after the session was
/// minted) is still refused — there is no persona left to mint ANY principal
/// for.
#[tokio::test]
async fn admin_session_unknown_persona_is_not_authorized_for_fleet() {
    let fx = Fixture::build("admin-unknown-persona").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: "persona-never-existed".to_owned(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );

    let err = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .unwrap_err();
    assert!(matches!(err, PrincipalError::NotAuthorizedForFleet));
    fx.teardown().await;
}

#[tokio::test]
async fn admin_session_store_down_is_unavailable_not_401_or_403() {
    let fx = Fixture::build("admin-store-down").await;
    let persona_id = fx.make_admin("erin").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let authority_no_store = fx.authority_with_empty_persona_cell();

    let err = authority_no_store
        .verify_admin_session(&token, NOW)
        .await
        .unwrap_err();
    assert!(
        matches!(err, PrincipalError::StoreUnavailable),
        "an unreadable persona store must surface as a distinct, transient error — never fold \
         into a 401/403 shape: {err:?}"
    );
    fx.teardown().await;
}

/// CORE isolation test (A3): a persona sees exactly its own participated
/// conversation and NEVER another persona's, even though both personas'
/// participation ties and committed turns live in the same event log store.
/// Proves cross-tenant leakage would fail this test if the funnel ever
/// regressed to registering more than the caller's own scope.
#[tokio::test]
async fn persona_execute_sees_only_its_own_participated_conversations() {
    let fx = Fixture::build("persona-isolation").await;

    let turn_a = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_a), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn_a), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");
    let turn_b = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-b".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_b), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn_b), Vec::new()),
            ],
        )
        .await
        .expect("append conv-b");

    // Persona A participates in conversation "a" only; persona B
    // participates in conversation "b" only — two distinct personas, two
    // distinct conversations, one shared event log.
    let persona_a = fx
        .persona
        .attribute(
            ExternalIdentity {
                provider: "test".to_owned(),
                scope: "s".to_owned(),
                external_id: "persona-a".to_owned(),
                display_name: "persona-a".to_owned(),
                ..Default::default()
            },
            "a".to_owned(),
            "initiator".to_owned(),
            NOW,
        )
        .await
        .expect("attribute persona A")
        .persona_id;
    let _persona_b = fx
        .persona
        .attribute(
            ExternalIdentity {
                provider: "test".to_owned(),
                scope: "s".to_owned(),
                external_id: "persona-b".to_owned(),
                display_name: "persona-b".to_owned(),
                ..Default::default()
            },
            "b".to_owned(),
            "initiator".to_owned(),
            NOW,
        )
        .await
        .expect("attribute persona B")
        .persona_id;

    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_a,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("a valid non-admin session must verify");
    assert!(
        matches!(principal, Principal::Persona(_)),
        "persona A's own session must mint a persona principal, not admin"
    );

    let scoped = fx
        .authority
        .scope_for(&principal)
        .await
        .expect("a persona principal with a live persona store always scopes");
    match &scoped.scope {
        QueryScope::Conversations(ids) => assert_eq!(ids, &vec!["a".to_owned()]),
        other => panic!("expected Conversations([\"a\"]), got {other:?}"),
    }

    let result = scoped
        .execute("SELECT DISTINCT partition FROM events")
        .await
        .expect("persona-scoped query over its own participated conversation");
    assert_eq!(
        result.rows,
        vec![vec![serde_json::json!("conv-a")]],
        "persona A must see exactly its own conversation's partition, never persona B's: {:?}",
        result.rows
    );

    // `EXPLAIN` is disallowed on the persona surface, matching the
    // conversation-grant posture.
    let err = scoped.execute("EXPLAIN SELECT 1").await.unwrap_err();
    assert!(matches!(err, ScopedQueryError::Rejected(_)));

    fx.teardown().await;
}

/// Boundary condition: a persona with zero participations still gets a
/// valid, empty scope — the query succeeds and returns no rows, never an
/// error.
#[tokio::test]
async fn persona_with_zero_participations_gets_an_empty_but_valid_scope() {
    let fx = Fixture::build("persona-zero-participations").await;
    // `PersonaStore::participations` returns empty for an unknown id (see
    // that method's doc) — a persona id that was never attributed to any
    // conversation exercises the exact same empty-scope path.
    let principal = Principal::Persona(PersonaPrincipal {
        persona_id: "persona-never-participated".to_owned(),
    });

    let scoped = fx
        .authority
        .scope_for(&principal)
        .await
        .expect("zero participations is a valid, empty scope, not an error");
    match &scoped.scope {
        QueryScope::Conversations(ids) => {
            assert!(
                ids.is_empty(),
                "expected zero participated conversations: {ids:?}"
            );
        }
        other => panic!("expected an empty Conversations scope, got {other:?}"),
    }

    let result = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("a zero-participation persona still runs queries successfully");
    assert_eq!(result.rows, vec![vec![serde_json::json!(0)]]);

    fx.teardown().await;
}

/// Boundary condition: the persona store being unavailable during
/// participation resolution is infrastructure, never an authorization
/// verdict — the same `StoreUnavailable`/503 path admin resolution already
/// gets.
#[tokio::test]
async fn persona_scope_store_down_is_unavailable() {
    let fx = Fixture::build("persona-store-down").await;
    let authority_no_store = fx.authority_with_empty_persona_cell();
    let principal = Principal::Persona(PersonaPrincipal {
        persona_id: "persona-x".to_owned(),
    });

    let err = authority_no_store.scope_for(&principal).await.unwrap_err();
    assert!(
        matches!(err, PrincipalError::StoreUnavailable),
        "a persona's participation resolution must surface store-unavailable as a distinct, \
         transient error — never fold into a hard refusal: {err:?}"
    );
    fx.teardown().await;
}

#[test]
fn conversation_grant_round_trips() {
    let signer = test_signer();
    let token = mint_conversation_grant(
        &signer.relabel_for_test(),
        "conv-1",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let authority_signer = signer.public_key_bytes();

    // Build a minimal authority just to call the (stateless) verify method —
    // eventlog/persona handles are unused by the grant path.
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-round-trip-{}",
        std::process::id()
    ));
    let eventlog = EventLogHost::spawn(dir, CancellationToken::new(), signer.relabel_for_test())
        .expect("spawn eventlog host");
    let eventlog = Arc::new(eventlog);
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        authority_signer.clone(),
        vec![authority_signer],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let principal = authority
        .verify_conversation_grant(&token, NOW)
        .expect("a freshly minted grant must verify");
    match principal {
        Principal::ConversationGrant(grant) => {
            assert_eq!(grant.conversation_id(), "conv-1");
            assert_eq!(grant.turn_id(), Some("turn-1"));
            assert_eq!(grant.subject(), &GrantSubject::Turn("turn-1".to_owned()));
        }
        other => panic!("expected Principal::ConversationGrant, got {other:?}"),
    }
}

#[test]
fn conversation_grant_minted_before_rotation_requires_historical_trust() {
    let fixture_signer = test_signer();
    let retired = TurnReadSigner::from_seed(91);
    let current = TurnReadSigner::from_seed(92);
    let token = mint_conversation_grant(
        &retired,
        "conv-before-rotation",
        GrantSubject::Turn("turn-before-rotation".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-history-{}",
        std::process::id()
    ));
    let eventlog = Arc::new(
        EventLogHost::spawn(
            dir,
            CancellationToken::new(),
            fixture_signer.relabel_for_test(),
        )
        .expect("spawn eventlog host"),
    );
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        current.public_key_bytes(),
        vec![fixture_signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );
    assert!(matches!(
        authority.verify_conversation_grant(&token, NOW),
        Err(PrincipalError::InvalidGrant)
    ));

    let historical =
        RoleTrustSet::<TurnReadRole>::checked(vec![current.identity(), retired.identity()])
            .expect("valid rotation history");
    let authority = authority.with_turn_read_trust_for_test(historical);
    assert!(authority.verify_conversation_grant(&token, NOW).is_ok());
}

/// A `GrantSubject::WebSession` grant round-trips the same way a `Turn` grant
/// does (#1576) — `turn_id()` reports `None` (never a fabricated turn id) and
/// `web_session_id()` reports the real session-scoped identifier.
#[test]
fn conversation_grant_web_session_subject_round_trips() {
    let signer = test_signer();
    let token = mint_conversation_grant(
        &signer.relabel_for_test(),
        "conv-1",
        GrantSubject::WebSession("persona-web-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-web-session-{}",
        std::process::id()
    ));
    let eventlog = Arc::new(
        EventLogHost::spawn(dir, CancellationToken::new(), signer.relabel_for_test())
            .expect("spawn eventlog host"),
    );
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        signer.public_key_bytes(),
        vec![signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let principal = authority
        .verify_conversation_grant(&token, NOW)
        .expect("a freshly minted web-session grant must verify");
    match principal {
        Principal::ConversationGrant(grant) => {
            assert_eq!(grant.conversation_id(), "conv-1");
            assert_eq!(
                grant.turn_id(),
                None,
                "a web-session grant must never report a fabricated turn id"
            );
            assert_eq!(
                grant.subject(),
                &GrantSubject::WebSession("persona-web-1".to_owned())
            );
        }
        other => panic!("expected Principal::ConversationGrant, got {other:?}"),
    }
}

/// An expired grant is a DISTINCT, still-opaque outcome from a malformed/
/// forged one — [`PrincipalError::GrantExpired`], not
/// [`PrincipalError::InvalidGrant`] — so a caller (the explorer web client)
/// can silently re-mint and retry once instead of treating routine expiry
/// like a probing/forged token.
#[test]
fn conversation_grant_expired_is_distinguishable_from_invalid() {
    let signer = test_signer();
    let token = mint_conversation_grant(
        &signer.relabel_for_test(),
        "conv-1",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW - 1,
    );
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-expired-{}",
        std::process::id()
    ));
    let eventlog = EventLogHost::spawn(dir, CancellationToken::new(), signer.relabel_for_test())
        .expect("spawn eventlog host");
    let eventlog = Arc::new(eventlog);
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        signer.public_key_bytes(),
        vec![signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let err = authority
        .verify_conversation_grant(&token, NOW)
        .unwrap_err();
    assert!(
        matches!(err, PrincipalError::GrantExpired),
        "an expired-but-otherwise-genuine grant must report GrantExpired, not InvalidGrant: \
         {err:?}"
    );
}

#[test]
fn conversation_grant_wrong_signer_is_invalid_grant() {
    let minting_signer = ApprovalSigner::from_seed(1);
    let verifying_signer = ApprovalSigner::from_seed(2);
    let token = mint_conversation_grant(
        &minting_signer.relabel_for_test(),
        "conv-1",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-wrong-signer-{}",
        std::process::id()
    ));
    let eventlog = EventLogHost::spawn(
        dir,
        CancellationToken::new(),
        verifying_signer.relabel_for_test(),
    )
    .expect("spawn eventlog host");
    let eventlog = Arc::new(eventlog);
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        verifying_signer.public_key_bytes(),
        vec![verifying_signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let err = authority
        .verify_conversation_grant(&token, NOW)
        .unwrap_err();
    assert!(matches!(err, PrincipalError::InvalidGrant));
}

/// Domain separation ([`GRANT_KIND`]) actually binds: a payload that is
/// otherwise a well-formed, genuinely-signed [`GrantClaims`] but carries the
/// WRONG `kind` tag must not verify — proving the tag is part of the signed
/// bytes [`QueryAuthority::verify_conversation_grant`] checks, not decoration
/// [`mint_conversation_grant`] adds and nothing reads back. Stands in for the
/// future-payload-type scenario the module doc names: bytes signed under the
/// same `approval_signer` key for a different purpose must never verify as a
/// query grant just because they happen to deserialize into `GrantClaims`'s
/// shape. Ported from the control plane's pre-move suite, dropped there when
/// the codec moved here.
#[test]
fn conversation_grant_wrong_kind_tag_is_invalid_grant() {
    let signer = test_signer();
    let turn_read_signer: TurnReadSigner = signer.relabel_for_test();
    let identity = turn_read_signer.identity();
    let claims = GrantClaims {
        kind: "some_other_signed_payload.v1".to_owned(),
        issuer: identity.issuer().to_owned(),
        key_id: identity.key_id().to_owned(),
        conversation_id: "conv-x".to_owned(),
        subject: GrantSubject::Turn("turn-y".to_owned()),
        expires_at_ms: NOW + TEST_TTL_MS,
    };
    let canonical = serde_json::to_vec(&claims).expect("GrantClaims always serializes");
    let signature = turn_read_signer.sign_turn_read_capability(&canonical);
    let token = format!(
        "{}.{}",
        URL_SAFE_NO_PAD.encode(canonical),
        URL_SAFE_NO_PAD.encode(signature)
    );
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-wrong-kind-{}",
        std::process::id()
    ));
    let eventlog = EventLogHost::spawn(dir, CancellationToken::new(), signer.relabel_for_test())
        .expect("spawn eventlog host");
    let eventlog = Arc::new(eventlog);
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        signer.public_key_bytes(),
        vec![signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let err = authority
        .verify_conversation_grant(&token, NOW)
        .unwrap_err();
    assert!(
        matches!(err, PrincipalError::InvalidGrant),
        "a genuinely-signed payload with the wrong kind tag must be refused, not silently \
         accepted"
    );
}

/// A token that is not shaped like `base64url(claims).base64url(signature)`
/// at all is rejected the same way a bad signature is — one undifferentiated
/// [`PrincipalError::InvalidGrant`], never a panic or a different error path
/// a probing caller could distinguish. Covers: no `.` separator at all, an
/// empty claims segment, an empty signature segment, and a signature segment
/// that itself contains a stray `.` (so `split_once` hands the base64
/// decoder a string with an out-of-alphabet character). Ported from the control
/// plane's pre-move suite, dropped there when the codec moved here.
#[test]
fn conversation_grant_malformed_shapes_are_invalid_grant() {
    let signer = test_signer();
    let dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-grant-malformed-{}",
        std::process::id()
    ));
    let eventlog = EventLogHost::spawn(dir, CancellationToken::new(), signer.relabel_for_test())
        .expect("spawn eventlog host");
    let eventlog = Arc::new(eventlog);
    let authority = QueryAuthority::new(
        eventlog.clone(),
        Arc::new(arc_swap::ArcSwapOption::empty()),
        test_dashboard_cell(&eventlog),
        Arc::new(RevokedTokens::new()),
        signer.public_key_bytes(),
        vec![signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    // No `.` separator at all.
    assert!(
        matches!(
            authority.verify_conversation_grant("not-a-grant-token", NOW),
            Err(PrincipalError::InvalidGrant)
        ),
        "a token with no `.` separator must be rejected"
    );

    // Empty claims segment.
    assert!(
        matches!(
            authority.verify_conversation_grant(".c2ln", NOW),
            Err(PrincipalError::InvalidGrant)
        ),
        "an empty claims segment must be rejected"
    );

    // Empty signature segment.
    assert!(
        matches!(
            authority.verify_conversation_grant("Y2xhaW1z.", NOW),
            Err(PrincipalError::InvalidGrant)
        ),
        "an empty signature segment must be rejected"
    );

    // The signature segment itself contains a stray `.` — `split_once` takes
    // only the FIRST `.` as the separator, so the extra dot rides into the
    // signature segment and is not valid base64url.
    assert!(
        matches!(
            authority.verify_conversation_grant("Y2xhaW1z.sig.with.dots", NOW),
            Err(PrincipalError::InvalidGrant)
        ),
        "a signature segment containing a stray `.` must be rejected"
    );
}

#[tokio::test]
async fn fleet_execute_runs_over_every_conversation_partition() {
    let fx = Fixture::build("fleet-execute").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-fleet-1".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append a committed turn");

    let persona_id = fx.make_admin("frank").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    let result = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("fleet query over the committed view");
    assert_eq!(result.columns, vec!["c"]);
    assert_eq!(result.rows, vec![vec![serde_json::json!(2)]]);

    fx.teardown().await;
}

/// Replace every regular file found under `dir` (recursively) with an empty
/// directory of the same name, returning how many were replaced. Opening any
/// of them as a file then fails with `EISDIR` — a user-id-independent,
/// self-heal-independent way to make a partition's on-disk journal unopenable.
/// See [`fleet_execute_reports_skipped_unreadable_partitions`] for why a
/// permission bit or byte corruption does not suffice.
fn block_every_blob_file(dir: &std::path::Path) -> usize {
    let entries: Vec<std::path::PathBuf> = std::fs::read_dir(dir)
        .expect("read the broken partition's data directory")
        .map(|entry| entry.expect("read a directory entry").path())
        .collect();
    let mut blocked = 0_usize;
    for path in entries {
        let file_type = std::fs::symlink_metadata(&path)
            .expect("stat a partition data entry")
            .file_type();
        if file_type.is_dir() {
            blocked += block_every_blob_file(&path);
        } else {
            std::fs::remove_file(&path).expect("remove a blob file");
            std::fs::create_dir(&path).expect("shadow the blob file with a directory");
            blocked += 1;
        }
    }
    blocked
}

/// QRY-7's own regression proof: a Fleet query tolerates one genuinely
/// unreadable partition (the pre-existing skip-and-log posture,
/// [`ScopedQuery::replay_scoped_partitions`]) but must now also REPORT it —
/// `result.skipped_partitions` surfaces the count, not just a server-side
/// `tracing::warn!` no caller ever sees.
///
/// The unreadable partition is simulated by replacing every blob FILE inside
/// its on-disk `{partition}_data` directory with a directory of the same name
/// AFTER writing real committed events into it through a first, short-lived
/// host, then reopening the SAME storage directory with a brand-new
/// `EventLogHost` (an empty open-logs cache, so this is a genuinely fresh
/// open attempt, not a cache hit). The `_data` directory itself stays a
/// directory, so `list_partitions` still surfaces the partition's NAME — but
/// opening its journal tries to open a blob path that is now a directory,
/// which fails with `EISDIR`, so replaying it fails exactly the way real disk
/// corruption or a permission drift would.
///
/// This is deliberately NOT a `chmod 000` on the directory: the CI image runs
/// as root (`cloudbuild/Dockerfile.ci` has no non-root `USER`), and root
/// bypasses directory permissions, so a permission-based simulation left this
/// test green locally but red in CI (the #1542 ci-main regression). Opening a
/// file-that-is-now-a-directory fails for root too, and the storage engine's
/// own tail self-heal (`#799`) recovers a torn tail, not a blob that is no
/// longer a file — so this shape is uid-independent AND self-heal-independent.
#[tokio::test]
async fn fleet_execute_reports_skipped_unreadable_partitions() {
    let signer = test_signer();
    let eventlog_dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-fleet-skip-eventlog-{}",
        std::process::id()
    ));
    let _ = std::fs::remove_dir_all(&eventlog_dir);
    let persona_dir = std::env::temp_dir().join(format!(
        "polyc-query-authority-fleet-skip-persona-{}",
        std::process::id()
    ));
    let _ = std::fs::remove_dir_all(&persona_dir);

    // First pass: a real host writes two committed conversations, one of
    // which is made unreadable afterward.
    {
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            eventlog_dir.clone(),
            shutdown.clone(),
            signer.relabel_for_test(),
        )
        .expect("spawn eventlog host");
        let turn_ok = uuid::Uuid::now_v7();
        host.append_batch(
            "conv-fleet-skip-ok".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_ok), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn_ok), Vec::new()),
            ],
        )
        .await
        .expect("append conv-fleet-skip-ok");

        let turn_broken = uuid::Uuid::now_v7();
        host.append_batch(
            "conv-fleet-skip-broken".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_broken), Vec::new()),
                Event::new(
                    kinds::tagged(kinds::TURN_COMPLETE, &turn_broken),
                    Vec::new(),
                ),
            ],
        )
        .await
        .expect("append conv-fleet-skip-broken");

        shutdown.cancel();
        drop(host);
    }

    let broken_dir = eventlog_dir.join("conv-fleet-skip-broken_data");
    let blocked = block_every_blob_file(&broken_dir);
    assert!(
        blocked > 0,
        "the broken partition must have written blob files to block: {broken_dir:?}"
    );

    // Second pass: a fresh host/authority over the same storage directory.
    let shutdown2 = CancellationToken::new();
    let host2 = Arc::new(
        EventLogHost::spawn(
            eventlog_dir.clone(),
            shutdown2.clone(),
            signer.relabel_for_test(),
        )
        .expect("reopen eventlog host"),
    );
    let persona_shutdown = CancellationToken::new();
    let persona = Arc::new(
        PersonaHost::spawn(persona_dir.clone(), persona_shutdown.clone())
            .expect("spawn persona host"),
    );
    let revoked = Arc::new(RevokedTokens::new());
    let authority = QueryAuthority::new(
        host2.clone(),
        Arc::new(arc_swap::ArcSwapOption::new(Some(persona.clone()))),
        test_dashboard_cell(&host2),
        revoked,
        signer.public_key_bytes(),
        vec![signer.public_key_bytes()],
        QueryLimits::default(),
        None,
        CacheConfig::disabled(),
        test_handoff_trust(),
    );

    let identity = ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: "fleet-skip-admin".to_owned(),
        display_name: "fleet-skip-admin".to_owned(),
        ..Default::default()
    };
    let persona_id = persona
        .set_admin(identity, true, NOW)
        .await
        .expect("set_admin");
    let token = mint_session(
        &signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = authority.scope_for(&principal).await.expect("scope_for");

    let result = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("a fleet query must tolerate one unreadable partition, not fail outright");
    assert_eq!(
        result.rows,
        vec![vec![serde_json::json!(2)]],
        "only the readable partition's two committed rows are visible"
    );
    assert_eq!(
        result.skipped_partitions, 1,
        "the unreadable partition must be reported to the caller, not just \
         tracing::warn!-logged server-side"
    );

    shutdown2.cancel();
    drop(host2);
    let _ = std::fs::remove_dir_all(&eventlog_dir);
    persona_shutdown.cancel();
    drop(persona);
    let _ = std::fs::remove_dir_all(&persona_dir);
}

#[tokio::test]
async fn conversation_grant_execute_sees_only_its_own_partition() {
    let fx = Fixture::build("grant-execute").await;
    let turn_a = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_a), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn_a), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");
    let turn_b = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-b".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn_b), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn_b), Vec::new()),
            ],
        )
        .await
        .expect("append conv-b");

    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    let result = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("conversation-scoped query");
    assert_eq!(result.rows, vec![vec![serde_json::json!(2)]]);

    // `EXPLAIN` is disallowed on the conversation-grant surface.
    let err = scoped.execute("EXPLAIN SELECT 1").await.unwrap_err();
    assert!(matches!(err, ScopedQueryError::Rejected(_)));

    fx.teardown().await;
}

/// Issue #2147: a conversation-scoped caller that names a table this scope
/// does not carry gets `ScopedQueryError::UnknownTable` — not the `Internal`
/// every planning failure would otherwise collapse into, indistinguishable
/// from a timeout and inviting a caller to simply retry with another guess.
///
/// Drives the two names production actually saw (`callers`, `turns`), plus
/// `personas` (a real Fleet-only table) and `events_raw` (registered, then
/// deregistered for this scope). All four must produce the SAME fixed
/// message: it names what this scope CAN query and never echoes what the
/// caller asked for, so no sequence of guesses tells a caller which names
/// exist elsewhere in the deployment.
///
/// Then holds [`CONVERSATION_CATALOG`] to the engine in BOTH directions.
/// Only one of the two is obvious: every advertised name must resolve, or the
/// message sends a refused caller straight back into the failure it exists to
/// end. The converse matters more and is easy to leave out — a table
/// registered for this scope and absent from the constant would go
/// unadvertised, with the whole suite green, and the next caller would be
/// guessing at it exactly the way #2147's callers guessed at `turns`. It is
/// checked by enumerating the catalog through a Fleet session (the one scope
/// built with `SessionConfig::with_information_schema(true)`, and a superset
/// of every scope's registrations — `create_scope_dependent_views` builds no
/// view for a grant that it does not also build for Fleet) and then probing
/// each name against THIS scope, so the expected set is read off the running
/// engine rather than hand-listed a second time.
#[tokio::test]
async fn conversation_grant_unknown_table_names_this_scopes_own_catalog() {
    let fx = Fixture::build("grant-unknown-table").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    for table in ["callers", "turns", "personas", "events_raw"] {
        let err = scoped
            .execute(&format!("SELECT * FROM {table}"))
            .await
            .unwrap_err();
        assert!(
            matches!(err, ScopedQueryError::UnknownTable(_)),
            "`{table}` must resolve to UnknownTable, not a generic internal failure: {err:?}"
        );
        let message = err.to_string();
        assert_eq!(
            message, *CONVERSATION_UNKNOWN_TABLE_MESSAGE,
            "every unresolvable name must return the one fixed message: {message:?}"
        );
        assert!(
            !message.contains(table),
            "the message must never echo the name the caller guessed: {message:?}"
        );
    }

    // Direction one: every table the message advertises must actually
    // resolve — a name in that copy that does not register would send the
    // caller straight back into the failure the message exists to end.
    for table in CONVERSATION_CATALOG {
        assert!(
            CONVERSATION_UNKNOWN_TABLE_MESSAGE.contains(table),
            "`{table}` is in this scope's catalog but the message does not name it"
        );
        scoped
            .execute(&format!("SELECT COUNT(*) AS c FROM {table}"))
            .await
            .unwrap_or_else(|err| panic!("advertised table `{table}` must resolve: {err:?}"));
    }

    // Direction two: nothing this scope can resolve may be missing from the
    // catalog. The candidate names come off a Fleet session's own
    // `information_schema` — a superset of what any narrower scope registers
    // — and each is then probed against the grant scope, so a thirteenth
    // table added to the engine shows up here without anyone remembering to
    // extend a list.
    let fleet_persona = fx.make_admin("catalog-probe").await;
    let fleet_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: fleet_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let fleet_principal = fx
        .authority
        .verify_admin_session(&fleet_token, NOW)
        .await
        .expect("valid admin session");
    let fleet = fx
        .authority
        .scope_for(&fleet_principal)
        .await
        .expect("scope_for");
    let listed = fleet
        .execute("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'")
        .await
        .expect("a Fleet session enumerates its own catalog");
    assert!(
        !listed.rows.is_empty(),
        "the Fleet catalog enumeration returned nothing, so this direction would pass vacuously"
    );

    let mut resolvable = Vec::new();
    for row in &listed.rows {
        let name = row[0]
            .as_str()
            .unwrap_or_else(|| panic!("information_schema.tables.table_name must be text: {row:?}"))
            .to_owned();
        match scoped
            .execute(&format!("SELECT * FROM {name} LIMIT 0"))
            .await
        {
            Ok(_) => resolvable.push(name),
            Err(ScopedQueryError::UnknownTable(_)) => {}
            Err(err) => {
                panic!("probing `{name}` against the grant scope failed unexpectedly: {err:?}")
            }
        }
    }
    resolvable.sort();
    let mut advertised: Vec<String> = CONVERSATION_CATALOG
        .iter()
        .map(|table| (*table).to_owned())
        .collect();
    advertised.sort();
    assert_eq!(
        resolvable, advertised,
        "CONVERSATION_CATALOG must name exactly what a grant scope can resolve — add the new \
         table to that constant (which is what the UnknownTable message and both SQL-hatch tool \
         descriptions are built from) rather than leaving callers to guess at it"
    );

    fx.teardown().await;
}

/// Issue #2147's Fleet half: a fleet-wide session gets its own message,
/// pointing at the `information_schema` catalog only this scope is built with
/// (`SessionConfig::with_information_schema(true)`) rather than a
/// hand-maintained list — and the query that message recommends must work.
#[tokio::test]
async fn fleet_unknown_table_points_at_the_catalog_this_scope_can_enumerate() {
    let fx = Fixture::build("fleet-unknown-table").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-fleet-unknown".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append a committed turn");

    let persona_id = fx.make_admin("grace").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    let err = scoped.execute("SELECT * FROM callers").await.unwrap_err();
    assert!(
        matches!(err, ScopedQueryError::UnknownTable(_)),
        "a fleet-wide caller must get UnknownTable too: {err:?}"
    );
    assert_eq!(err.to_string(), FLEET_UNKNOWN_TABLE_MESSAGE);

    scoped
        .execute("SELECT table_name FROM information_schema.tables")
        .await
        .expect("the query the fleet message recommends must actually run");

    fx.teardown().await;
}

/// Every column the engine really registered for `table`, read off the
/// running session rather than listed here (issue #2206).
///
/// `SELECT * FROM <table> LIMIT 0` returns the PLANNED schema even with no
/// rows (see `crate::output::output_to_json`), so this is the same registry
/// the planner resolves a caller's column against — which is the whole point:
/// a test that hardcoded the column list could pass while the message
/// advertised something the engine no longer builds.
async fn registered_columns(scoped: &ScopedQuery, table: &str) -> Vec<String> {
    let result = scoped
        .execute(&format!("SELECT * FROM {table} LIMIT 0"))
        .await
        .unwrap_or_else(|err| panic!("`{table}` must resolve for this scope: {err:?}"));
    assert!(
        !result.columns.is_empty(),
        "`{table}` returned no columns, so every assertion built from it would pass vacuously"
    );
    result.columns
}

/// The column list an `UnknownColumn` message offers, split back out of the
/// rendered sentence.
///
/// Lets a test assert on the SET and the COUNT a caller is actually handed.
/// A `contains` check passes just as happily on a list that names a column
/// twice or drops one for the "and N more" tail, and both of those are real
/// failures: `DataFusion` offers the projection schema followed by the input
/// schema at a `GROUP BY`/`ORDER BY`/`HAVING` miss, so duplication is the
/// default and truncation follows from it on a wide table.
fn offered_columns(message: &str) -> Vec<String> {
    let (_, offered) = message
        .split_once("you can select ")
        .unwrap_or_else(|| panic!("this message offers no columns at all: {message}"));
    offered
        .replace(", and ", ", ")
        .replace(" and ", ", ")
        .split(", ")
        .map(ToOwned::to_owned)
        .collect()
}

/// The exact message a miss against `table` must produce, built from
/// `columns` — the list read off the running engine.
fn expected_unknown_column_message(guessed: &str, table: &str, columns: &[String]) -> String {
    let borrowed: Vec<&str> = columns.iter().map(String::as_str).collect();
    format!(
        "there is no `{guessed}` column on `{table}`; you can select {}",
        catalog_sentence(&borrowed)
    )
}

/// Issue #2206: a conversation-scoped caller that names a column no table in
/// its query carries gets `ScopedQueryError::UnknownColumn` — not the
/// `Internal` every schema failure would otherwise collapse into, which
/// would tell the caller nothing, leave a model guessing again, and charge a
/// caller's typo to the deployment's own internal-failure rate.
///
/// Drives the four names production actually saw in one 90-minute window
/// (`message`, `event_type`, and `type` against `events`; `tool` against
/// `tool_calls`), plus the qualified form of the first, plus a Fleet-only
/// column a redacted scope must not be offered.
///
/// The message is asserted WHOLE against a list read off the running engine,
/// not checked for a substring. That is deliberate, and it is A7's lesson one
/// level down: a containment assertion would pass while the message quietly
/// dropped a column the engine registers, sending the corrected caller into a
/// second refusal. Equality fails the moment the two disagree in either
/// direction.
#[tokio::test]
async fn conversation_grant_unknown_column_names_the_columns_that_table_really_has() {
    let fx = Fixture::build("grant-unknown-column").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    let events_columns = registered_columns(&scoped, "events").await;
    // The three unqualified guesses, plus the qualified form: a bare name
    // against a single-relation query resolves to the same table, so both
    // spellings must produce the same offer.
    for (sql, guessed) in [
        ("SELECT message FROM events", "message"),
        ("SELECT event_type FROM events", "event_type"),
        ("SELECT type FROM events LIMIT 5", "type"),
        ("SELECT events.message FROM events", "message"),
        (
            "SELECT partition FROM events WHERE message = 'x'",
            "message",
        ),
    ] {
        let err = scoped.execute(sql).await.unwrap_err();
        assert!(
            matches!(err, ScopedQueryError::UnknownColumn(_)),
            "`{sql}` must resolve to UnknownColumn, not a generic internal failure: {err:?}"
        );
        assert_eq!(
            err.to_string(),
            expected_unknown_column_message(guessed, "events", &events_columns),
            "`{sql}` must name the column asked for and exactly the columns `events` registers"
        );
    }

    // The second table production's callers guessed at, so the message is
    // not accidentally pinned to one schema.
    let tool_calls_columns = registered_columns(&scoped, "tool_calls").await;
    let err = scoped
        .execute("SELECT tool FROM tool_calls")
        .await
        .unwrap_err();
    assert_eq!(
        err.to_string(),
        expected_unknown_column_message("tool", "tool_calls", &tool_calls_columns)
    );

    // Redaction: `attribution`'s four `identity_*` columns are built for
    // `QueryScope::Fleet` alone (see `crate::engine`'s "Identity redaction
    // invariant"). `valid_fields` is the REDACTED view's schema, so the
    // offer cannot name them — the property that makes listing columns safe
    // at all.
    let attribution_columns = registered_columns(&scoped, "attribution").await;
    assert!(
        !attribution_columns
            .iter()
            .any(|column| column.starts_with("identity_")),
        "a redacted scope must never be offered a Fleet-only column: {attribution_columns:?}"
    );
    let err = scoped
        .execute("SELECT identity_kind FROM attribution")
        .await
        .unwrap_err();
    assert_eq!(
        err.to_string(),
        expected_unknown_column_message("identity_kind", "attribution", &attribution_columns),
        "the offer is the redacted view's own schema, and the only `identity_` in the message is \
         the name the caller themselves wrote"
    );

    // A name with no relation to resolve against at all: the planner reports
    // an empty `valid_fields`, so the message has nothing to offer and says
    // the one thing that would help instead.
    let err = scoped.execute("SELECT nope").await.unwrap_err();
    assert!(matches!(err, ScopedQueryError::UnknownColumn(_)));
    assert_eq!(
        err.to_string(),
        "there is no `nope` column here; add a `FROM` clause naming the table to read it from"
    );

    // A bare name across a join has no single table to point at, so the
    // offer is qualified instead. Built from the same two schemas the
    // assertions above read off the engine. The message says "where you used
    // it", not "the tables this query selects from": `valid_fields` is the
    // schema at the RESOLUTION SITE, which the correlated-subquery case
    // below shows is not always the whole statement's.
    let messages_columns = registered_columns(&scoped, "messages").await;
    let joined: Vec<String> = events_columns
        .iter()
        .map(|column| format!("events.{column}"))
        .chain(
            messages_columns
                .iter()
                .map(|column| format!("messages.{column}")),
        )
        .collect();
    let err = scoped
        .execute("SELECT sender FROM events JOIN messages ON messages.turn_id = events.turn_id")
        .await
        .unwrap_err();
    assert!(matches!(err, ScopedQueryError::UnknownColumn(_)));
    assert_eq!(
        err.to_string(),
        format!(
            "there is no `sender` column available where you used it; you can select {}",
            column_list_sentence(&joined)
        )
    );

    // A miss in `ORDER BY`, `GROUP BY`, or `HAVING` resolves against the
    // projection schema FOLLOWED BY the input schema, so `DataFusion` offers
    // every projected column twice. The offer names each column once —
    // advertising `turn_id` twice reads as a bug, and on a wide table the
    // doubled list also blows `MAX_ADVERTISED_COLUMNS` and truncates a
    // single-table miss that should never truncate.
    for sql in [
        "SELECT turn_id FROM events ORDER BY nope",
        "SELECT turn_id, kind FROM events GROUP BY turn_id, kind HAVING nope > 1",
    ] {
        let err = scoped.execute(sql).await.unwrap_err();
        let rendered = err.to_string();
        assert!(
            matches!(err, ScopedQueryError::UnknownColumn(_)),
            "`{sql}` must resolve to UnknownColumn: {err:?}"
        );
        // Sorted set equality, not containment: the projection comes first
        // so the ORDER differs from a plain miss, but the offer must be the
        // same columns, each exactly once. A duplicate makes this list
        // longer than `events` is and fails here.
        let mut offered = offered_columns(&rendered);
        offered.sort();
        let mut expected = events_columns.clone();
        expected.sort();
        assert_eq!(
            offered, expected,
            "`{sql}` must offer every `events` column exactly once: {rendered}"
        );
    }

    // A qualifier that names no relation available where the name was used
    // — the mistake a caller makes by aliasing in `FROM` and forgetting the
    // alias in the projection. The qualifier IS the error, so the message
    // must not deny the column exists and then offer it back under the
    // alias; it names what the caller wrote, which relation is actually
    // there, and the alias to use.
    let err = scoped
        .execute("SELECT events.turn_id FROM events e")
        .await
        .unwrap_err();
    assert!(matches!(err, ScopedQueryError::UnknownColumn(_)));
    let aliased: Vec<String> = events_columns
        .iter()
        .map(|column| format!("e.{column}"))
        .collect();
    assert_eq!(
        err.to_string(),
        format!(
            "there is no `events.turn_id` column available where you used it; that part of the \
             query reads from `e`, so you can select {}",
            column_list_sentence(&aliased)
        )
    );

    // A bad OUTER reference inside a correlated subquery: `DataFusion` hands
    // us the INNER relation's schema alone, so the message can only be
    // honest about that. It must not claim those are the tables the whole
    // query selects from — a caller told that `e.nope` should have been an
    // `m.` column writes a differently wrong query.
    let aliased_messages: Vec<String> = messages_columns
        .iter()
        .map(|column| format!("m.{column}"))
        .collect();
    let err = scoped
        .execute(
            "SELECT e.turn_id FROM events e WHERE EXISTS (SELECT 1 FROM messages m WHERE m.role = \
             e.nope)",
        )
        .await
        .unwrap_err();
    assert!(matches!(err, ScopedQueryError::UnknownColumn(_)));
    let rendered = err.to_string();
    assert_eq!(
        rendered,
        format!(
            "there is no `e.nope` column available where you used it; that part of the query \
             reads from `m`, so you can select {}",
            column_list_sentence(&aliased_messages)
        )
    );
    assert!(
        !rendered.contains("this query selects from"),
        "the offer is the resolution site's schema, not the statement's: {rendered}"
    );

    fx.teardown().await;
}

/// The guard the assertion above needs to be worth anything: the
/// `identity_*` columns a grant scope is not offered DO exist for
/// `QueryScope::Fleet`. Without this, `attribution` could lose those columns
/// outright and the redaction assertion would keep passing.
#[tokio::test]
async fn fleet_attribution_still_carries_the_columns_a_grant_scope_is_not_offered() {
    let fx = Fixture::build("fleet-attribution-columns").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-fleet-columns".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append a committed turn");

    let persona_id = fx.make_admin("ida").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let fleet = fx.authority.scope_for(&principal).await.expect("scope_for");

    let columns = registered_columns(&fleet, "attribution").await;
    assert!(
        columns.iter().any(|column| column.starts_with("identity_")),
        "Fleet `attribution` must still carry the redacted columns: {columns:?}"
    );

    fx.teardown().await;
}

/// The security property this whole change rests on: a table this scope
/// cannot query decides the refusal, and no `UnknownColumn` message ever
/// names a column of a table the caller may not read.
///
/// Deliberately NOT "before any column is resolved" — the planner does
/// resolve a column and refuse a table in the same pass. The last two
/// statements below are the shapes where it does: a `UNION ALL` arm and a
/// scalar subquery each emit a `FieldNotFound` for the in-scope relation
/// ALONGSIDE the out-of-scope table's error, in one error `Collection`. What
/// holds regardless is the construction of `valid_fields` itself —
/// `DataFusion` builds it from relations that ALREADY RESOLVED, so an
/// unresolvable table contributes no schema to leak — plus
/// `is_unresolved_table_error` walking the whole tree and running first,
/// which is what makes the refusal a caller reads name the table.
///
/// `personas` is a real Fleet-only reference table and `events_raw` is
/// registered and then deregistered for this scope, so both genuinely exist
/// in the deployment — the interesting case, since a leak would need their
/// schemas in hand. `callers` never existed at all. Each is asked for with a
/// column name too, which is what makes the two arms race at all.
#[tokio::test]
async fn a_table_this_scope_cannot_query_decides_the_refusal_and_leaks_no_columns() {
    let fx = Fixture::build("grant-table-before-column").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    for sql in [
        "SELECT display_name FROM personas",
        "SELECT persona_id FROM personas WHERE anything = '1'",
        "SELECT payload FROM events_raw",
        "SELECT * FROM events_raw",
        "SELECT anything FROM callers",
        // A join whose FIRST relation is in scope and whose second is not:
        // the out-of-scope name must still decide the outcome.
        "SELECT e.turn_id FROM events e JOIN personas p ON p.persona_id = e.turn_id",
        // The two shapes where the planner really does resolve a column and
        // refuse a table in the same pass, and reports BOTH: a `UNION ALL`
        // arm, and a scalar subquery. The bad column belongs to the IN-SCOPE
        // arm, so the offer it would have produced is harmless — what
        // matters is that the out-of-scope table still decides the refusal.
        "SELECT bogus FROM events UNION ALL SELECT persona_id FROM personas",
        "SELECT bogus, (SELECT max(persona_id) FROM personas) FROM events",
    ] {
        let err = scoped.execute(sql).await.unwrap_err();
        assert!(
            matches!(err, ScopedQueryError::UnknownTable(_)),
            "`{sql}` must refuse the TABLE, never resolve columns against it: {err:?}"
        );
        assert_eq!(err.to_string(), *CONVERSATION_UNKNOWN_TABLE_MESSAGE);
    }

    // And no message this scope can produce names a column only an
    // out-of-scope table has. Asserted against `UnknownColumn`, which is the
    // thing at risk: the table refusal is a fixed constant that by
    // construction carries no column names at all, so asserting over it
    // proves nothing. Read `personas`' real columns off a Fleet session,
    // where it IS registered, drop the ones an in-scope table also carries,
    // and require the rest to be absent from every column refusal.
    let fleet_persona = fx.make_admin("column-leak-probe").await;
    let fleet_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: fleet_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let fleet_principal = fx
        .authority
        .verify_admin_session(&fleet_token, NOW)
        .await
        .expect("valid admin session");
    let fleet = fx
        .authority
        .scope_for(&fleet_principal)
        .await
        .expect("scope_for");
    let personas_columns = registered_columns(&fleet, "personas").await;

    let mut in_scope: std::collections::HashSet<String> = std::collections::HashSet::new();
    for table in CONVERSATION_CATALOG {
        in_scope.extend(registered_columns(&scoped, table).await);
    }
    let fleet_only: Vec<&String> = personas_columns
        .iter()
        .filter(|column| !in_scope.contains(*column))
        .collect();
    assert!(
        !fleet_only.is_empty(),
        "every `personas` column is also in scope, so this assertion would pass vacuously"
    );

    for sql in [
        "SELECT nope FROM events",
        "SELECT turn_id FROM events ORDER BY nope",
        "SELECT events.nope FROM events e",
        "SELECT nope FROM events JOIN messages ON messages.turn_id = events.turn_id",
    ] {
        let err = scoped.execute(sql).await.unwrap_err();
        let ScopedQueryError::UnknownColumn(message) = err else {
            panic!("`{sql}` must resolve to UnknownColumn: {err:?}");
        };
        for column in &fleet_only {
            assert!(
                !message.contains(column.as_str()),
                "`{sql}`'s refusal must not name the out-of-scope `personas`.`{column}`: {message}"
            );
        }
    }

    fx.teardown().await;
}

/// The other direction of the split: a planning or execution failure that is
/// NOT a name the caller can correct still collapses into
/// `ScopedQueryError::Internal`. Without this, a predicate that over-matched
/// would quietly report deployment faults as the caller's own — and,
/// through `query_audit`, stop recording them as internal failures at all.
#[tokio::test]
async fn a_genuine_engine_failure_still_collapses_into_internal() {
    let fx = Fixture::build("grant-internal-still-internal").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    for sql in [
        // An unregistered FUNCTION, not an unregistered name: a `Plan` error
        // the table predicate must not claim.
        "SELECT no_such_function(turn_id) FROM events",
        // A type failure inside an expression over columns that all exist.
        "SELECT turn_id + 1 FROM events",
    ] {
        let err = scoped.execute(sql).await.unwrap_err();
        assert!(
            matches!(err, ScopedQueryError::Internal),
            "`{sql}` is not a name the caller can correct and must stay Internal: {err:?}"
        );
    }

    fx.teardown().await;
}

/// `MAX_ADVERTISED_COLUMNS` is set above the widest table this crate
/// registers, so a miss against a single table — every miss production has
/// seen — always leaves the caller the WHOLE column set to correct against
/// rather than a truncated one.
///
/// Reads the widest table off a running Fleet engine (the one scope built
/// with `with_information_schema(true)`, and a superset of every narrower
/// scope's registrations), so a new column on any table pushes this over the
/// bound and fails here instead of silently truncating a refusal.
#[tokio::test]
async fn column_list_bound_covers_the_widest_registered_table() {
    let fx = Fixture::build("column-bound").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-column-bound".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append a committed turn");

    let persona_id = fx.make_admin("hana").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let fleet = fx.authority.scope_for(&principal).await.expect("scope_for");

    let listed = fleet
        .execute("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'")
        .await
        .expect("a Fleet session enumerates its own catalog");
    assert!(
        !listed.rows.is_empty(),
        "the Fleet catalog enumeration returned nothing, so this bound would pass vacuously"
    );

    let mut widest = (String::new(), 0usize);
    for row in &listed.rows {
        let name = row[0]
            .as_str()
            .unwrap_or_else(|| panic!("information_schema.tables.table_name must be text: {row:?}"))
            .to_owned();
        let count = registered_columns(&fleet, &name).await.len();
        if count > widest.1 {
            widest = (name, count);
        }
    }
    assert!(
        widest.1 <= MAX_ADVERTISED_COLUMNS,
        "`{}` registers {} columns, past the {MAX_ADVERTISED_COLUMNS} an UnknownColumn message \
         spells out — raise the bound or accept that a miss against this table is truncated",
        widest.0,
        widest.1
    );

    // A table's own width is not the width the bound has to cover: the
    // message is built at the RESOLUTION SITE, and a miss in `ORDER BY`,
    // `GROUP BY`, or `HAVING` resolves against the projection schema
    // followed by the input schema, so `SELECT * FROM <widest> ORDER BY
    // nope` reaches `unknown_column_message` at twice the width measured
    // above. The `SELECT *` measurement alone can never catch that, so
    // measure the refusal a caller actually reads.
    let widest_columns = registered_columns(&fleet, &widest.0).await;
    let err = fleet
        .execute(&format!("SELECT * FROM {} ORDER BY nope", widest.0))
        .await
        .unwrap_err();
    let ScopedQueryError::UnknownColumn(message) = err else {
        panic!(
            "an `ORDER BY` miss on `{}` must be UnknownColumn: {err:?}",
            widest.0
        );
    };
    let mut offered = offered_columns(&message);
    offered.sort();
    let mut expected = widest_columns;
    expected.sort();
    assert_eq!(
        offered, expected,
        "an `ORDER BY` miss against the widest table must offer its whole column set, each once \
         — a duplicated list is both wrong and past the {MAX_ADVERTISED_COLUMNS} bound: {message}"
    );
    assert!(
        !message.contains(" more"),
        "a miss against a SINGLE table must never truncate: {message}"
    );

    fx.teardown().await;
}

/// QRY-3: a Fleet-scoped query whose replayed source data exceeds
/// [`QueryLimits::max_source_events`] is rejected — BEFORE ANY DECODE, an
/// O(1) `EventLogHost::partition_event_count` sum pre-check now catches it —
/// never reaching `crate::engine::decode_partition_tables`. Proves the
/// pre-execution budget actually bites, independent of `memory_bytes` (the
/// `FairSpillPool` never sees this data at all) and independent of
/// `row_cap`/`timeout` (this query's real result is a single small `COUNT`
/// row, nowhere near either of those).
///
/// Item F (the QRY-3 hardening review), extended after the stale-decode
/// regression a later review pass found: this test does not just check the
/// error variant came back — it diffs
/// `crate::engine::partition_tables::DECODE_CALL_COUNT`, the genuine decode
/// fan-out counter, not `crate::engine::BUILD_CALL_COUNT` (which only
/// advances inside `QueryEngine::build_from_tables`'s ENGINE-ASSEMBLY step —
/// a step the decode cache moved decode out of, so it stopped
/// proving this on its own; see that item's own doc). If a future change
/// accidentally let a rejected query still reach
/// `crate::engine::decode_partition_tables` — for instance the O(1)
/// pre-check moved after `ScopedQuery::resolve_partitions`, or a refactor
/// dropped the early return — this assertion fails even though the error
/// variant returned would still look identical, which a variant-only
/// assertion could never catch.
#[tokio::test]
async fn fleet_query_over_source_budget_is_rejected_before_decode() {
    let fx = Fixture::build("fleet-source-budget").await;
    let persona_id = fx.make_admin("frank").await;

    // Six events in one partition — decode must never run for this test;
    // only `replay_scoped_partitions`'s own event count matters.
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-budget".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
                Event::new("k2".to_owned(), Vec::new()),
                Event::new("k3".to_owned(), Vec::new()),
                Event::new("k4".to_owned(), Vec::new()),
                Event::new("k5".to_owned(), Vec::new()),
            ],
        )
        .await
        .expect("append conv-budget");

    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_events: 4,
        ..QueryLimits::default()
    });
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = limited_authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes");

    let decode_calls_before =
        crate::engine::DECODE_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .unwrap_err();
    let decode_calls_after =
        crate::engine::DECODE_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "6 replayed events over a 4-event budget must be rejected before decode: {err:?}"
    );
    assert_eq!(
        decode_calls_after, decode_calls_before,
        "decode_partition_tables (the genuine decode fan-out entry point) must never be reached \
         when the source budget rejects a query — no raw or typed batch may be constructed"
    );

    // Item C: the message is scope-aware but must never leak the actual
    // replayed count (6) or the configured budget (4) — a caller must not be
    // able to use a rejected query to learn either number.
    let message = err.to_string();
    assert!(
        !message.contains('6') && !message.contains('4'),
        "the rejection message must not leak the actual or budgeted event count: {message:?}"
    );
    assert!(
        message.contains("query_max_source_events"),
        "a Fleet rejection must name the config an administrator can raise: {message:?}"
    );

    fx.teardown().await;
}

/// Item C, the `Conversations` half: a conversation-grant query over budget
/// gets [`CONVERSATION_BUDGET_EXCEEDED_MESSAGE`] (the caller's own accessible
/// history is over budget), never the Fleet message — and, like the Fleet
/// case above, the rejection must land before ANY decode
/// (`crate::engine::partition_tables::DECODE_CALL_COUNT`, not merely engine
/// assembly — see the Fleet test's own doc for why the two differ) and
/// must not leak the replayed/budgeted counts.
#[tokio::test]
async fn conversation_grant_query_over_source_budget_is_rejected_before_decode() {
    let fx = Fixture::build("grant-source-budget").await;

    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
                Event::new("k2".to_owned(), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_events: 1,
        ..QueryLimits::default()
    });
    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = limited_authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("grant always scopes");

    let decode_calls_before =
        crate::engine::DECODE_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .unwrap_err();
    let decode_calls_after =
        crate::engine::DECODE_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "3 replayed events over a 1-event budget must be rejected before decode: {err:?}"
    );
    assert_eq!(
        decode_calls_after, decode_calls_before,
        "decode_partition_tables must never be reached when a conversation-grant query trips the \
         source budget"
    );

    let message = err.to_string();
    assert_eq!(
        message, CONVERSATION_BUDGET_EXCEEDED_MESSAGE,
        "a Conversations-scope rejection must use the caller's-own-history message, not the \
         Fleet one, and must carry no count"
    );
    assert!(
        !message.contains("query_max_source_events"),
        "a Conversations-scope caller cannot raise a fleet-wide admin setting, so the message \
         must not point at one: {message:?}"
    );

    fx.teardown().await;
}

/// Issue #1541, the CRITICAL early-abort proof at the `ScopedQuery` level:
/// a Fleet scope with three partitions, replayed under a byte budget that
/// trips partway through the SECOND one, never even attempts the third.
/// Calls the private `replay_scoped_partitions` directly (this test module
/// is a descendant of `authority`, so it can) to inspect
/// [`ReplayError::BytesBudgetExceeded`]'s own `partitions_replayed` count —
/// deterministic here because `EventLogHost::list_partitions` returns
/// `conv-*` names sorted, so Fleet always visits `conv-a`, then `conv-b`,
/// then `conv-c` in that order.
///
/// This test is structured to FAIL if a future change reintroduces
/// full-materialize-then-check: `partitions_replayed` would then read `3`
/// (every partition read before any check ran), not `2`.
#[tokio::test]
async fn replay_scoped_partitions_aborts_before_reading_every_fleet_partition() {
    let fx = Fixture::build("fleet-bytes-early-abort").await;
    let persona_id = fx.make_admin("early-abort").await;

    // conv-a: tiny — comfortably fits under the budget on its own.
    fx.eventlog
        .append_batch("conv-a".to_owned(), vec![Event::new("k0", vec![0u8; 10])])
        .await
        .expect("append conv-a");
    // conv-b: 20 events of 1,000 bytes (20,000 bytes) — large enough that
    // reading it trips the budget before it finishes, and definitely before
    // conv-c is ever reached.
    fx.eventlog
        .append_batch(
            "conv-b".to_owned(),
            (0..20)
                .map(|i| Event::new(format!("k{i}"), vec![0u8; 1_000]))
                .collect(),
        )
        .await
        .expect("append conv-b");
    // conv-c: must never be touched.
    fx.eventlog
        .append_batch("conv-c".to_owned(), vec![Event::new("k0", vec![0u8; 10])])
        .await
        .expect("append conv-c");

    // conv-a costs 10 bytes, leaving 4,990 of the 5,000 budget for conv-b;
    // conv-b's own replay crosses that remainder after its 5th event
    // (5,000 cumulative bytes within that single call), so conv-b is the
    // partition that trips the budget — conv-c is never reached.
    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_bytes: 5_000,
        ..QueryLimits::default()
    });
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = limited_authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes");

    let err = scoped
        .replay_scoped_partitions()
        .await
        .expect_err("20,000 bytes in conv-b alone must exceed the 5,000 byte budget");
    match err {
        ReplayError::BytesBudgetExceeded {
            partitions_replayed,
            bytes_read,
        } => {
            assert_eq!(
                partitions_replayed, 2,
                "must abort right after conv-b (the partition that tripped the budget) — \
                 reading conv-c too (3) would mean the whole Fleet scope was materialized \
                 before the budget was ever checked"
            );
            assert!(
                partitions_replayed < 3,
                "3 partitions exist in this deployment; the replay must stop strictly before \
                 the last one once the budget trips"
            );
            assert!(
                bytes_read >= 5_000,
                "bytes_read is the accumulated size at the abort — at or just past the 5,000 \
                 byte budget, recorded to the replayed-bytes histogram so over-budget queries \
                 are visible: {bytes_read}"
            );
        }
        ReplayError::Internal => panic!("expected a byte-budget rejection, got Internal"),
    }

    fx.teardown().await;
}

/// Issue #1541, the public-API half: a Fleet-scoped query whose replayed
/// source BYTES exceed [`crate::engine::QueryLimits::max_source_bytes`] is
/// rejected before `QueryEngine::build` ever decodes anything — the byte-
/// budget counterpart to `fleet_query_over_source_budget_is_rejected_before_decode`.
#[tokio::test]
async fn fleet_query_over_bytes_budget_is_rejected_before_decode() {
    let fx = Fixture::build("fleet-bytes-budget").await;
    let persona_id = fx.make_admin("gina").await;

    // One partition, comfortably over a deliberately tiny byte budget.
    fx.eventlog
        .append_batch(
            "conv-bytes".to_owned(),
            (0..5)
                .map(|i| Event::new(format!("k{i}"), vec![0u8; 1_000]))
                .collect(),
        )
        .await
        .expect("append conv-bytes");

    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_bytes: 500,
        ..QueryLimits::default()
    });
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = limited_authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes");

    let build_calls_before =
        crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .unwrap_err();
    let build_calls_after =
        crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "5,000 replayed bytes over a 500-byte budget must be rejected before decode: {err:?}"
    );
    assert_eq!(
        build_calls_after, build_calls_before,
        "QueryEngine::build must never be reached when the byte budget rejects a query"
    );

    let message = err.to_string();
    assert_eq!(
        message, FLEET_BYTES_BUDGET_EXCEEDED_MESSAGE,
        "a Fleet byte-budget rejection must use the byte-budget message, not the event-count one"
    );
    assert!(
        message.contains("query_max_source_bytes"),
        "a Fleet rejection must name the config an administrator can raise: {message:?}"
    );
    assert!(
        !message.contains("500") && !message.contains("5000"),
        "the rejection message must not leak the actual or budgeted byte count: {message:?}"
    );

    fx.teardown().await;
}

/// The `Conversations` half of the byte-budget rejection: a conversation
/// grant whose OWN accessible history exceeds
/// [`crate::engine::QueryLimits::max_source_bytes`] gets
/// [`CONVERSATION_BYTES_BUDGET_EXCEEDED_MESSAGE`], never the Fleet message —
/// mirroring `conversation_grant_query_over_source_budget_is_rejected_before_decode`
/// for the byte dimension.
#[tokio::test]
async fn conversation_grant_query_over_bytes_budget_is_rejected_before_decode() {
    let fx = Fixture::build("grant-bytes-budget").await;

    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            (0..5)
                .map(|i| Event::new(format!("k{i}"), vec![0u8; 1_000]))
                .collect(),
        )
        .await
        .expect("append conv-a");

    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_bytes: 500,
        ..QueryLimits::default()
    });
    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = limited_authority
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("grant always scopes");

    let build_calls_before =
        crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .unwrap_err();
    let build_calls_after =
        crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "5,000 replayed bytes over a 500-byte budget must be rejected before decode: {err:?}"
    );
    assert_eq!(
        build_calls_after, build_calls_before,
        "QueryEngine::build must never be reached when a conversation-grant query trips the \
         byte budget"
    );

    let message = err.to_string();
    assert_eq!(
        message, CONVERSATION_BYTES_BUDGET_EXCEEDED_MESSAGE,
        "a Conversations-scope byte-budget rejection must use the caller's-own-history message"
    );
    assert!(
        !message.contains("query_max_source_bytes"),
        "a Conversations-scope caller cannot raise a fleet-wide admin setting, so the message \
         must not point at one: {message:?}"
    );

    fx.teardown().await;
}

/// A conversation-grant query whose replayed bytes stay UNDER the byte
/// budget runs normally — the budget must never false-positive on a scope
/// that is genuinely within it.
#[tokio::test]
async fn conversation_grant_query_under_bytes_budget_succeeds_normally() {
    let fx = Fixture::build("grant-bytes-under-budget").await;

    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-a");

    let authority_with_generous_bytes = fx.authority_with_limits(QueryLimits {
        max_source_bytes: 1_000_000,
        ..QueryLimits::default()
    });
    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "a",
        GrantSubject::Turn("turn-1".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = authority_with_generous_bytes
        .verify_conversation_grant(&token, NOW)
        .expect("valid grant");
    let scoped = authority_with_generous_bytes
        .scope_for(&principal)
        .await
        .expect("grant always scopes");

    let output = scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("a scope genuinely under the byte budget must run normally");
    assert!(
        !output.rows.is_empty(),
        "a normal query under budget must still return its real result: {output:?}"
    );

    fx.teardown().await;
}

/// Item D: `crate::authority::ScopedQuery::execute` records this query's
/// observed size (event count, replayed bytes) and scope label on EVERY
/// query, and the rejection counter when the budget trips — both wired all
/// the way from a real `execute()` call, not just `crate::metrics`'s own
/// unit tests in isolation. Diffs the shared default Prometheus registry
/// (process-global, so this asserts the delta, not an absolute count — the
/// same idiom `crate::control-plane`'s own metrics tests use).
#[tokio::test]
async fn execute_records_query_size_and_budget_rejection_metrics() {
    use prometheus::{Encoder as _, TextEncoder};

    fn scrape() -> String {
        let mut buf = Vec::new();
        TextEncoder::new()
            .encode(&prometheus::default_registry().gather(), &mut buf)
            .expect("encode");
        String::from_utf8(buf).expect("utf8")
    }
    fn counter_value(text: &str, metric: &str, scope: &str) -> f64 {
        let needle = format!("{metric}{{scope=\"{scope}\"}} ");
        text.lines()
            .find(|line| line.starts_with(&needle))
            .and_then(|line| line.rsplit(' ').next())
            .and_then(|v| v.parse::<f64>().ok())
            .unwrap_or(0.0)
    }

    let fx = Fixture::build("metrics-observed").await;
    let persona_id = fx.make_admin("mira").await;

    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-metrics".to_owned(),
            vec![
                Event::new(kinds::tagged(kinds::TURN_START, &turn), Vec::new()),
                Event::new(kinds::tagged(kinds::TURN_COMPLETE, &turn), Vec::new()),
            ],
        )
        .await
        .expect("append conv-metrics");

    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = fx
        .authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes");

    let before_rejections = counter_value(
        &scrape(),
        "polychrome_query_source_budget_exceeded_total",
        "fleet",
    );

    // A budget-passing query still records the source-events/replayed-bytes
    // observation (see `ScopedQuery::execute`'s "Observability, on every
    // query" doc section) — proven by the histogram's `_sum` line advancing,
    // since `_bucket` presence alone survives across the whole test binary.
    let sum_before = source_events_sum(&scrape());
    scoped
        .execute("SELECT COUNT(*) AS c FROM events")
        .await
        .expect("under-budget admin query");
    let sum_after = source_events_sum(&scrape());
    assert!(
        sum_after > sum_before,
        "a successful query must still advance polychrome_query_source_events's sum: \
         before={sum_before} after={sum_after}"
    );

    // Now trip the budget and confirm the rejection counter, not just the
    // observation, advances.
    let limited_authority = fx.authority_with_limits(QueryLimits {
        max_source_events: 1,
        ..QueryLimits::default()
    });
    let limited_scoped = limited_authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes");
    let err = limited_scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .unwrap_err();
    assert!(matches!(err, ScopedQueryError::SourceBudgetExceeded(_)));

    let after_rejections = counter_value(
        &scrape(),
        "polychrome_query_source_budget_exceeded_total",
        "fleet",
    );
    assert_eq!(
        after_rejections - before_rejections,
        1.0,
        "polychrome_query_source_budget_exceeded_total{{scope=\"fleet\"}} must increment by \
         exactly 1 per rejection"
    );

    fx.teardown().await;
}

/// Parse `polychrome_query_source_events`'s `_sum` line (summed across every
/// `scope` label) out of a scrape — used to prove an observation landed
/// without depending on an absolute bucket count in a registry shared
/// process-wide across every test in this binary.
#[cfg(test)]
fn source_events_sum(text: &str) -> f64 {
    text.lines()
        .filter(|line| line.starts_with("polychrome_query_source_events_sum"))
        .filter_map(|line| line.rsplit(' ').next())
        .filter_map(|v| v.parse::<f64>().ok())
        .sum()
}

/// QRY-3 fix 2: [`build_base_session_state`] pins a configured, bounded
/// spill directory rather than leaving `DataFusion`'s `DiskManager` at its
/// default (an OS-chosen tmp dir, 100 GiB quota) — spilling stays enabled,
/// but under a caller-supplied root and quota (`QueryLimits::spill_dir`/
/// `QueryLimits::spill_quota_bytes` in production; a test-only path/quota
/// here).
#[tokio::test]
async fn build_base_session_state_configures_a_bounded_spill_directory() {
    let spill_root = std::env::temp_dir().join("polyc-query-authority-tests-spill");
    let quota_bytes = 2 * 1024 * 1024 * 1024;
    let state = build_base_session_state(64 * 1024 * 1024, &spill_root, quota_bytes);
    let disk_manager = state.runtime_env().disk_manager.clone();

    assert!(
        disk_manager.tmp_files_enabled(),
        "spilling must stay enabled, just bounded — never silently disabled"
    );
    assert_eq!(
        disk_manager.max_temp_directory_size(),
        quota_bytes,
        "the configured quota must be the caller-supplied value, not DataFusion's own 100GiB \
         default"
    );

    let temp_dir_paths = disk_manager.temp_dir_paths();
    assert_eq!(
        temp_dir_paths.len(),
        1,
        "exactly one configured spill root is configured: {temp_dir_paths:?}"
    );
    assert!(
        temp_dir_paths[0].starts_with(&spill_root),
        "the spill dir must live under the caller-supplied root {spill_root:?}, got {:?}",
        temp_dir_paths[0]
    );
}

/// Issue #1592's narrow-admission proof: a Fleet replay reads exactly
/// `"conv-"`-prefixed partitions PLUS the routine scheduler's own dedicated
/// `"routine-scheduler"` partition — no OTHER non-conversation partition,
/// however it happens to be named, is ever admitted. Proven against the
/// real replay path (`ScopedQuery::execute` over `events_raw`, not a direct
/// `QueryEngine::build` fixture), so this exercises the actual
/// `replay_scoped_partitions` admission logic the module doc's "Admitting
/// the scheduler partition" section describes, not a hand-assembled
/// `PartitionEvents` list.
#[tokio::test]
async fn fleet_replay_admits_the_routine_scheduler_partition_but_no_other_non_conversation_partition()
 {
    let fx = Fixture::build("scheduler-partition-admission").await;
    let turn = uuid::Uuid::now_v7();
    fx.eventlog
        .append_batch(
            "conv-a".to_owned(),
            vec![Event::new(
                kinds::tagged(kinds::TURN_START, &turn),
                Vec::new(),
            )],
        )
        .await
        .expect("append conv-a");
    let fired = RoutineFiredEvent {
        routine: "daily-standup".to_owned(),
        occurrence: "daily-standup-1".to_owned(),
        scheduled_at_ms: 1,
        fired_at_ms: 2,
        ..Default::default()
    };
    fx.eventlog
        .append_batch(
            "routine-scheduler".to_owned(),
            vec![Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec())],
        )
        .await
        .expect("append routine-scheduler");
    // An arbitrary non-conversation, non-scheduler partition — must never
    // surface, proving the admission is exactly one literal name wider than
    // `"conv-"`, not "any non-conversation partition".
    fx.eventlog
        .append_batch(
            "some-other-partition".to_owned(),
            vec![Event::new(kinds::USAGE.to_owned(), Vec::new())],
        )
        .await
        .expect("append some-other-partition");

    let persona_id = fx.make_admin("grace").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = fx.authority.scope_for(&principal).await.expect("scope_for");

    let result = scoped
        .execute("SELECT DISTINCT partition FROM events_raw ORDER BY partition")
        .await
        .expect("fleet query over events_raw");
    assert_eq!(
        result.rows,
        vec![
            vec![serde_json::json!("conv-a")],
            vec![serde_json::json!("routine-scheduler")],
        ],
        "exactly conv-a and routine-scheduler must be admitted — some-other-partition must \
         never appear: {:?}",
        result.rows
    );

    fx.teardown().await;
}

/// A minimal, in-memory [`RoutineCatalog`] — [`Self::list_routines`] returns
/// a caller-supplied fixed list, proving [`ScopedQuery`]'s wiring actually
/// calls through this trait rather than silently building an empty
/// `routines` table regardless of what a real deployment's catalog would
/// answer.
struct FakeRoutineCatalog(Vec<RoutineStatusRecord>);

#[async_trait::async_trait]
impl RoutineCatalog for FakeRoutineCatalog {
    async fn list_routines(&self) -> Result<Vec<RoutineStatusRecord>, RoutineCatalogError> {
        Ok(self.0.clone())
    }
}

/// Issue #1592's end-to-end catalog-wiring proof: a Fleet session backed by
/// a [`FakeRoutineCatalog`] resolves `routines` from that EXACT catalog,
/// through [`ScopedQuery::execute`] — not a direct
/// `QueryEngine::build`/`ReferenceData` fixture, so this exercises
/// `ScopedQuery::resolve_routines`'s own real call path.
#[tokio::test]
async fn fleet_execute_resolves_routines_through_the_supplied_routine_catalog() {
    let fx = Fixture::build("routine-catalog-wiring").await;
    let catalog = Arc::new(FakeRoutineCatalog(vec![RoutineStatusRecord {
        name: "daily-standup".to_owned(),
        uid: "uid-daily-standup".to_owned(),
        fire_conversation_id: "fire-conv-daily-standup".to_owned(),
        ready: true,
        phase: Some("Ready".to_owned()),
        message: None,
        last_fire_time_ms: None,
        next_fire_time_ms: Some(1_784_883_600_000),
        conditions_json: "[]".to_owned(),
        creator_persona: "persona-1".to_owned(),
        provenance_conversation_id: "conv-1".to_owned(),
        schedule_json: r#"{"kind":"cron","expression":"0 9 * * *","timezone":null}"#.to_owned(),
        next_fires_json: r"[1784883600000]".to_owned(),
        suspended: false,
        paused_by: None,
        paused_at_ms: None,
        pause_reason: None,
        prompt: "post the morning standup".to_owned(),
        scope: "private".to_owned(),
        orphaned: false,
        display_name: String::new(),
        description: String::new(),
        schedule_timezone: "UTC".to_owned(),
    }]));
    let authority = fx.authority_with_routine_catalog(catalog);

    let persona_id = fx.make_admin("henry").await;
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    let scoped = authority.scope_for(&principal).await.expect("scope_for");

    let result = scoped
        .execute("SELECT name, ready, next_fire_time_ms FROM routines")
        .await
        .expect("fleet query over routines");
    assert_eq!(
        result.rows,
        vec![vec![
            serde_json::json!("daily-standup"),
            serde_json::json!(true),
            serde_json::json!(1_784_883_600_000_i64),
        ]]
    );

    fx.teardown().await;
}

/// Issue #1882's core acceptance proof: two personas each own exactly one
/// routine (and one fire against it); a persona-scoped session sees ONLY its
/// own routine and its own fire, never the other persona's, while Fleet sees
/// both. Exercised through `ScopedQuery::execute` end to end, the same real
/// call path `fleet_execute_resolves_routines_through_the_supplied_routine_catalog`
/// pins for Fleet alone.
#[tokio::test]
async fn persona_scoped_routines_and_fires_see_only_the_callers_own_routine() {
    let fx = Fixture::build("owner-scoped-routines").await;

    let persona_a = fx.make_non_admin("alice-owner").await;
    let persona_b = fx.make_non_admin("bob-owner").await;

    let catalog = Arc::new(FakeRoutineCatalog(vec![
        routine_record("routine-a", &persona_a),
        routine_record("routine-b", &persona_b),
    ]));
    let authority = fx.authority_with_routine_catalog(catalog);

    for (routine, occurrence) in [("routine-a", "routine-a-1"), ("routine-b", "routine-b-1")] {
        let fired = RoutineFiredEvent {
            routine: routine.to_owned(),
            occurrence: occurrence.to_owned(),
            scheduled_at_ms: 1,
            fired_at_ms: 2,
            routine_uid: format!("{routine}-uid"),
            ..Default::default()
        };
        fx.eventlog
            .append_batch(
                "routine-scheduler".to_owned(),
                vec![Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec())],
            )
            .await
            .expect("append routine-scheduler fire");
    }

    async fn scope_for_persona(
        authority: &QueryAuthority,
        signer: &SessionSigner,
        persona_id: &str,
    ) -> ScopedQuery {
        let token = mint_session(
            signer,
            &SessionSubject::Persona {
                persona_id: persona_id.to_owned(),
            },
            &[SessionScope::ExplorerRead],
            NOW,
            TEST_TTL_MS,
        );
        let principal = authority
            .verify_admin_session(&token, NOW)
            .await
            .expect("valid session");
        assert!(
            matches!(principal, Principal::Persona(_)),
            "a non-admin persona must mint Principal::Persona, got {principal:?}"
        );
        authority.scope_for(&principal).await.expect("scope_for")
    }

    let scoped_a = scope_for_persona(&authority, &fx.signer.relabel_for_test(), &persona_a).await;
    let routines_a = scoped_a
        .execute("SELECT name FROM routines ORDER BY name")
        .await
        .expect("persona a's own routines query");
    assert_eq!(
        routines_a.rows,
        vec![vec![serde_json::json!("routine-a")]],
        "persona a must see exactly their own routine, never bob's"
    );
    assert_eq!(
        routines_a.columns,
        vec!["name"],
        "columns must come from the planned schema, not merely be non-empty because rows exist"
    );
    let fires_a = scoped_a
        .execute("SELECT routine FROM fires ORDER BY routine")
        .await
        .expect("persona a's own fires query");
    assert_eq!(
        fires_a.rows,
        vec![vec![serde_json::json!("routine-a")]],
        "persona a must see exactly their own routine's fire, never bob's"
    );
    assert_eq!(fires_a.columns, vec!["routine"]);

    let scoped_b = scope_for_persona(&authority, &fx.signer.relabel_for_test(), &persona_b).await;
    let routines_b = scoped_b
        .execute("SELECT name FROM routines ORDER BY name")
        .await
        .expect("persona b's own routines query");
    assert_eq!(
        routines_b.rows,
        vec![vec![serde_json::json!("routine-b")]],
        "persona b must see exactly their own routine, never alice's"
    );
    assert_eq!(routines_b.columns, vec!["name"]);
    let fires_b = scoped_b
        .execute("SELECT routine FROM fires ORDER BY routine")
        .await
        .expect("persona b's own fires query");
    assert_eq!(
        fires_b.rows,
        vec![vec![serde_json::json!("routine-b")]],
        "persona b must see exactly their own routine's fire, never alice's"
    );
    assert_eq!(fires_b.columns, vec!["routine"]);

    let admin_persona = fx.make_admin("carol-admin").await;
    let admin_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: admin_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let admin_principal = authority
        .verify_admin_session(&admin_token, NOW)
        .await
        .expect("valid admin session");
    let scoped_fleet = authority
        .scope_for(&admin_principal)
        .await
        .expect("scope_for");
    let routines_fleet = scoped_fleet
        .execute("SELECT name FROM routines ORDER BY name")
        .await
        .expect("fleet routines query");
    assert_eq!(
        routines_fleet.rows,
        vec![
            vec![serde_json::json!("routine-a")],
            vec![serde_json::json!("routine-b")],
        ],
        "fleet must see every persona's routine, unfiltered"
    );
    let fires_fleet = scoped_fleet
        .execute("SELECT routine FROM fires ORDER BY routine")
        .await
        .expect("fleet fires query");
    assert_eq!(
        fires_fleet.rows,
        vec![
            vec![serde_json::json!("routine-a")],
            vec![serde_json::json!("routine-b")],
        ],
        "fleet must see every routine's fire, unfiltered"
    );

    fx.teardown().await;
}

/// POLY-160 review lower-priority finding: `routine_overview`'s
/// `last_fire_at_ms`/`last_fire_outcome` pick the single latest `fires_raw`
/// row via a `ROW_NUMBER()` window ordered `fired_at_ms DESC, position DESC`.
/// A scheduler catch-up burst can plausibly record two fires with the
/// IDENTICAL `fired_at_ms` — proven here directly: two fires for one
/// routine, same `fired_at_ms`, appended in one order: the view must name
/// the LAST-APPENDED fire's own outcome (the higher journal `position`),
/// deterministically, rather than leaving the tie's winner to query-plan
/// happenstance.
#[tokio::test]
async fn routine_overview_last_fire_breaks_a_fired_at_ms_tie_by_journal_position() {
    let fx = Fixture::build("routine-overview-fire-tie").await;

    let owner = fx.make_non_admin("dana-owner").await;
    let catalog = Arc::new(FakeRoutineCatalog(vec![routine_record(
        "routine-tie",
        &owner,
    )]));
    let authority = fx.authority_with_routine_catalog(catalog);

    for (occurrence, outcome) in [
        (
            "routine-tie-1",
            polyc_proto::proto::polychrome::events::v1::RoutineFireOutcome::Ok,
        ),
        (
            "routine-tie-2",
            polyc_proto::proto::polychrome::events::v1::RoutineFireOutcome::StoppedUngranted,
        ),
    ] {
        let fired = RoutineFiredEvent {
            routine: "routine-tie".to_owned(),
            occurrence: occurrence.to_owned(),
            scheduled_at_ms: 1,
            fired_at_ms: 2,
            routine_uid: "routine-tie-uid".to_owned(),
            ..Default::default()
        };
        let outcome_ev = polyc_proto::proto::polychrome::events::v1::RoutineFireOutcomeEvent {
            routine: fired.routine.clone(),
            occurrence: fired.occurrence.clone(),
            outcome: outcome.into(),
            fired_at_ms: fired.fired_at_ms,
            ..Default::default()
        };
        fx.eventlog
            .append_batch(
                "routine-scheduler".to_owned(),
                vec![
                    Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec()),
                    Event::trusted(kinds::ROUTINE_FIRE_OUTCOME, outcome_ev.encode_to_vec()),
                ],
            )
            .await
            .expect("append scheduler fire+outcome");
    }

    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: owner.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid session");
    let scoped = authority.scope_for(&principal).await.expect("scope_for");

    let overview = scoped
        .execute("SELECT last_fire_outcome FROM routine_overview WHERE name = 'routine-tie'")
        .await
        .expect("overview query");
    assert_eq!(
        overview.rows,
        vec![vec![serde_json::json!("stopped_ungranted")]],
        "the LATER-APPENDED fire (higher journal position) must win an equal-fired_at_ms tie"
    );

    fx.teardown().await;
}

/// A persona who owns no routine at all still gets a valid, empty `routines`/
/// `fires` catalog — never an error and never another persona's rows, the
/// same "empty scope is a valid scope" posture a persona with zero
/// participations already gets.
#[tokio::test]
async fn persona_scoped_routines_and_fires_are_empty_not_an_error_for_an_owner_with_none() {
    let fx = Fixture::build("owner-scoped-routines-empty").await;
    let persona_a = fx.make_non_admin("dana-no-routines").await;
    let persona_b = fx.make_non_admin("erin-owner").await;

    let catalog = Arc::new(FakeRoutineCatalog(vec![routine_record(
        "erins-routine",
        &persona_b,
    )]));
    let authority = fx.authority_with_routine_catalog(catalog);

    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_a.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid session");
    let scoped = authority.scope_for(&principal).await.expect("scope_for");

    let routines = scoped
        .execute("SELECT name FROM routines")
        .await
        .expect("a routine-owner-less persona's routines query must still succeed");
    assert!(
        routines.rows.is_empty(),
        "an owner with no routines must see zero rows, not erin's: {:?}",
        routines.rows
    );
    assert_eq!(
        routines.columns,
        vec!["name"],
        "an empty result must still carry its real column list (issue #1916), not an empty one"
    );
    let fires = scoped
        .execute("SELECT routine FROM fires")
        .await
        .expect("a routine-owner-less persona's fires query must still succeed");
    assert!(
        fires.rows.is_empty(),
        "an owner with no routines must see zero fires, not erin's: {:?}",
        fires.rows
    );
    assert_eq!(
        fires.columns,
        vec!["routine"],
        "an empty result must still carry its real column list (issue #1916), not an empty one"
    );

    // Issue #2147: a persona session carries `routines`/`fires` on top of the
    // conversation set, so it gets the owner message rather than the grant
    // one — and every table that message advertises must resolve, including
    // for an owner whose own `routines` happens to be empty.
    let err = scoped.execute("SELECT * FROM turns").await.unwrap_err();
    assert!(
        matches!(err, ScopedQueryError::UnknownTable(_)),
        "a persona session must get UnknownTable for a name it invented: {err:?}"
    );
    assert_eq!(err.to_string(), *OWNER_UNKNOWN_TABLE_MESSAGE);
    for table in CONVERSATION_CATALOG.iter().chain(OWNER_ONLY_CATALOG.iter()) {
        assert!(
            OWNER_UNKNOWN_TABLE_MESSAGE.contains(table),
            "`{table}` is registered for a persona session but the message does not name it"
        );
        scoped
            .execute(&format!("SELECT COUNT(*) AS c FROM {table}"))
            .await
            .unwrap_or_else(|err| panic!("advertised table `{table}` must resolve: {err:?}"));
    }

    fx.teardown().await;
}

/// Issue #1916: `FIRES_OWNED_VIEW_SQL`'s inner join (`fires_raw JOIN routines
/// ON fires_raw.routine_uid = routines.uid`) can legitimately eliminate every
/// row while BOTH join inputs are non-empty — the caller owns a routine (so
/// the persona-scoped `routines` side isn't trivially empty the way
/// `persona_scoped_routines_and_fires_are_empty_not_an_error_for_an_owner_with_none`
/// drives it), and fires exist in the event log (so `fires_raw` isn't empty
/// either), but none of those fires are against the caller's own routine.
/// The join output is zero rows either way, but this path — non-trivial
/// inputs producing an empty join, rather than an empty input trivially
/// producing an empty join — is the scenario report #1916 was filed against.
/// This test pins the end-to-end behavior through `ScopedQuery::execute`:
/// `columns` must still be populated from the planned schema even though
/// every row was eliminated. It does not, and cannot, distinguish DataFusion
/// returning zero batches from returning one empty batch — `ScopedQuery::execute`
/// only ever hands back the JSON envelope, never the intermediate
/// `QueryOutput`. The unit coverage that pins the zero-batches case
/// specifically — `output_to_json` deriving `columns` from `output.schema`
/// rather than `batches.first()` — lives in
/// `crate::output::tests::zero_batches_still_yields_columns_from_the_planned_schema`.
#[tokio::test]
async fn persona_scoped_fires_inner_join_eliminates_every_row_but_still_reports_columns() {
    let fx = Fixture::build("owner-scoped-fires-join-elimination").await;

    let persona_a = fx.make_non_admin("frank-owns-unfired-routine").await;
    let persona_b = fx.make_non_admin("grace-owns-fired-routine").await;

    let catalog = Arc::new(FakeRoutineCatalog(vec![
        routine_record("routine-a", &persona_a),
        routine_record("routine-b", &persona_b),
    ]));
    let authority = fx.authority_with_routine_catalog(catalog);

    // Only persona b's routine ever fires — persona a's own routine has zero
    // fires against it, but `fires_raw` itself is non-empty.
    let fired = RoutineFiredEvent {
        routine: "routine-b".to_owned(),
        occurrence: "routine-b-1".to_owned(),
        scheduled_at_ms: 1,
        fired_at_ms: 2,
        routine_uid: "routine-b-uid".to_owned(),
        ..Default::default()
    };
    fx.eventlog
        .append_batch(
            "routine-scheduler".to_owned(),
            vec![Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec())],
        )
        .await
        .expect("append routine-scheduler fire");

    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: persona_a.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid session");
    let scoped = authority.scope_for(&principal).await.expect("scope_for");

    // Sanity: persona a's own routine is a real, non-empty join input.
    let routines = scoped
        .execute("SELECT name FROM routines")
        .await
        .expect("persona a's own routines query");
    assert_eq!(routines.rows, vec![vec![serde_json::json!("routine-a")]]);

    let fires = scoped
        .execute("SELECT routine FROM fires")
        .await
        .expect("persona a's fires query, joined against a routine they own");
    assert!(
        fires.rows.is_empty(),
        "persona a's own routine never fired, so the inner join must eliminate every row: {:?}",
        fires.rows
    );
    assert_eq!(
        fires.columns,
        vec!["routine"],
        "the inner join eliminating every row must not erase the planned column list (issue #1916)"
    );

    fx.teardown().await;
}

// ---------------------------------------------------------------------
// The per-partition decode cache, end-to-end through
// `ScopedQuery::execute`. `crate::cache`'s own `#[cfg(test)] mod tests`
// covers the cache's internal lookup/LRU logic directly; these tests pin
// the cache wired all the way through a real `QueryAuthority`/`ScopedQuery`.
// ---------------------------------------------------------------------

/// Scrape the process-default Prometheus registry as text — the same
/// pattern `crate::metrics`'s own tests and [`source_events_sum`] use.
fn scrape_metrics() -> String {
    use prometheus::Encoder as _;
    let mut buf = Vec::new();
    prometheus::TextEncoder::new()
        .encode(&prometheus::default_registry().gather(), &mut buf)
        .expect("encode");
    String::from_utf8(buf).expect("utf8")
}

/// A plain (unlabeled) `IntCounter`'s current value out of a scrape — the
/// decode-cache hit/tail/full-rebuild/eviction series carry no label, unlike
/// [`source_events_sum`]'s `HistogramVec`. Diffed across a call, exactly
/// like every other counter helper in this suite, since the registry is
/// process-global and shared across every test in this binary.
fn counter_value(text: &str, metric: &str) -> f64 {
    let prefix = format!("{metric} ");
    text.lines()
        .find(|line| line.starts_with(&prefix))
        .and_then(|line| line.rsplit(' ').next())
        .and_then(|v| v.parse::<f64>().ok())
        .unwrap_or(0.0)
}

/// Mint a valid Fleet-admin [`ScopedQuery`] over `authority`, sharing
/// `fx`'s own signer/persona-store setup — the shape every test below
/// repeats.
async fn admin_scoped(fx: &Fixture, authority: &QueryAuthority, persona_id: String) -> ScopedQuery {
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona { persona_id },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid admin session");
    authority
        .scope_for(&principal)
        .await
        .expect("admin always scopes")
}

/// (a) A second, identical query over an unchanged partition is served
/// entirely from the cache: exactly one `polychrome_query_cache_hit_total`
/// increment, for the SECOND call only (the first is a genuine miss that
/// populates the cache).
#[tokio::test]
async fn second_identical_query_is_served_from_the_cache() {
    let fx = Fixture::build("cache-hit").await;
    let persona_id = fx.make_admin("cache-hit-admin").await;
    fx.eventlog
        .append_batch("conv-hit".to_owned(), vec![Event::new("k0", Vec::new())])
        .await
        .expect("append");

    let cached_authority = fx.authority_with_cache_config(CacheConfig::new(Some(1)));
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    let hits_before = counter_value(&scrape_metrics(), "polychrome_query_cache_hit_total");
    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("first query (miss)");
    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("second, identical query (hit)");
    let hits_after = counter_value(&scrape_metrics(), "polychrome_query_cache_hit_total");

    assert_eq!(
        hits_after - hits_before,
        1.0,
        "the second identical query over an unchanged partition must be a single cache hit"
    );

    fx.teardown().await;
}

/// (b) Appending to an already-cached partition, then querying again, only
/// replays and decodes the TAIL — exactly one `polychrome_query_cache_tail_total`
/// increment, and the query's own result reflects both the original and the
/// appended event.
#[tokio::test]
async fn append_then_query_replays_only_the_tail() {
    let fx = Fixture::build("cache-tail").await;
    let persona_id = fx.make_admin("cache-tail-admin").await;
    fx.eventlog
        .append_batch("conv-tail".to_owned(), vec![Event::new("k0", Vec::new())])
        .await
        .expect("append first event");

    let cached_authority = fx.authority_with_cache_config(CacheConfig::new(Some(1)));
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("first query (miss, populates the cache)");

    fx.eventlog
        .append_batch("conv-tail".to_owned(), vec![Event::new("k1", Vec::new())])
        .await
        .expect("append second event");

    let tails_before = counter_value(&scrape_metrics(), "polychrome_query_cache_tail_total");
    // Filtered to this test's own `k0`/`k1` kinds — `append_batch` also
    // durably appends its own MMR signed-root marker event per call (#799,
    // `EventLogHost::append_batch`'s doc), so a plain unfiltered `COUNT(*)`
    // would count those markers too, not just this test's own two events.
    let result = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw WHERE kind_base IN ('k0', 'k1')")
        .await
        .expect("second query (tail)");
    let tails_after = counter_value(&scrape_metrics(), "polychrome_query_cache_tail_total");

    assert_eq!(
        tails_after - tails_before,
        1.0,
        "an append onto an already-cached partition must resolve as exactly one tail, not a full \
         rebuild"
    );
    assert_eq!(
        result.rows,
        vec![vec![serde_json::json!(2)]],
        "the merged (cached base + replayed tail) result must count both events"
    );

    fx.teardown().await;
}

/// The stale-watermark race: `resolve_partitions_cached` must never key a
/// `Lookup::Tail`'s stored watermark to the O(1)
/// `EventLogHost::partition_event_count` read BEFORE actually replaying the
/// tail, in a separate round trip. An append landing in that window would
/// make the replay genuinely longer than the snapshot, but the cache entry
/// would still be stored under the smaller, stale count — so the NEXT
/// `Lookup::Tail` would re-replay a range already inside the cached tables,
/// duplicating rows on `PartitionTables::concat`.
///
/// This test interleaves a REAL append exactly into that window, via the
/// test-only `ScopedQuery::race_inject_after_count_read` hook (the race is a
/// timing accident this harness cannot otherwise force deterministically —
/// see that field's own doc), and asserts a subsequent query returns the
/// exact ground-truth row count for every event this test ever appended,
/// never a duplicated one.
#[tokio::test]
async fn cache_tail_race_does_not_duplicate_rows_across_an_interleaved_append() {
    let fx = Fixture::build("cache-race").await;
    let persona_id = fx.make_admin("race-admin").await;

    // Three events populate the cache's first entry — a genuine miss.
    fx.eventlog
        .append_batch(
            "conv-race".to_owned(),
            (0..3)
                .map(|i| Event::new(format!("k{i}"), Vec::new()))
                .collect(),
        )
        .await
        .expect("append initial 3 events");

    let cached_authority = fx.authority_with_cache_config(CacheConfig::new(Some(1)));
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("first query (miss, populates the cache)");

    // Two more REAL events land — an ordinary append — so the next query's
    // own count snapshot sees a higher count and enters the `Tail` branch.
    fx.eventlog
        .append_batch(
            "conv-race".to_owned(),
            (3..5)
                .map(|i| Event::new(format!("k{i}"), Vec::new()))
                .collect(),
        )
        .await
        .expect("append 2 more events");

    // Arm the race: the instant the second query's own
    // `resolve_partitions_cached` finishes reading `conv-race`'s count and
    // epoch (the count captured here is BEFORE this injection runs) but
    // before it replays the tail, two MORE events land — simulating a
    // concurrent writer landing in the exact window the fix closes. The
    // tail replay this triggers reads to the CURRENT end of the log, which
    // now includes these two events too, not just the ones the stale count
    // implied.
    *scoped.race_inject_after_count_read.lock().expect("poison") = Some((
        "conv-race".to_owned(),
        (5..7)
            .map(|i| Event::new(format!("k{i}"), Vec::new()))
            .collect(),
    ));

    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("second query (tail, racing an interleaved append)");

    // A third query: with the fix, the second query's own stored watermark
    // was derived from the tail's own last REPLAYED position — which
    // already covers the raced-in events — so this is a `Hit`, nothing to
    // duplicate. Under the pre-fix bug, the stored watermark was the STALE
    // count read before the race injection, so this third query re-entered
    // `Tail` from that stale point, re-replaying (and re-concatenating)
    // events the merged table already held.
    let result = scoped
        .execute(
            "SELECT COUNT(*) AS c FROM events_raw WHERE kind_base IN \
             ('k0','k1','k2','k3','k4','k5','k6')",
        )
        .await
        .expect("third query");

    assert_eq!(
        result.rows,
        vec![vec![serde_json::json!(7)]],
        "exactly 7 real events were ever appended to conv-race across this test — a cached-tail \
         race must never duplicate any of them: {result:?}"
    );

    fx.teardown().await;
}

/// (c) THE erasure test: an in-place rewrite replaces an event's payload
/// (unchanged position, unchanged event count — the #216/#860 erasure
/// primitive) and a subsequent query must NEVER serve the pre-erasure payload.
/// A `(partition, event-count)`-only watermark could not catch this, since the
/// count never changes.
///
/// What catches it is [`QueryAuthority::invalidate_partition`], which every
/// mutating journal command reports to on its receipt — in a deployment that is
/// `crate::projection_signal::ReportingRepair` around the whole capability
/// (#1565, chunk B6). This test issues the mutation and then that same report,
/// in the order the Container issues them, and asserts on what the next query
/// serves. Both halves are here on purpose: the pre-rewrite read establishes
/// that this partition really was cached and being served from cache, so the
/// post-rewrite read fails if the report ever stops being load-bearing rather
/// than passing for the wrong reason.
#[tokio::test]
async fn cache_never_serves_a_payload_erased_by_an_in_place_rewrite() {
    let fx = Fixture::build("cache-erasure").await;
    let persona_id = fx.make_admin("erasure-admin").await;
    fx.eventlog
        .append_batch(
            "conv-erasure".to_owned(),
            vec![Event::new(
                kinds::APPROVAL_DEFERRED,
                br#"{"marker":"before"}"#.to_vec(),
            )],
        )
        .await
        .expect("append");

    let cached_authority = fx.authority_with_cache_config(CacheConfig::new(Some(1)));
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    let before = scoped
        .execute("SELECT payload_json FROM events_raw WHERE kind_base = 'approval_deferred'")
        .await
        .expect("query before rewrite");
    assert_eq!(
        before.rows,
        vec![vec![serde_json::json!(r#"{"marker":"before"}"#)]],
        "sanity: the pre-rewrite payload must be visible first"
    );

    fx.eventlog
        .rewrite_partition(
            "conv-erasure".to_owned(),
            "test-authority-rewrite".to_owned(),
            Box::new(|event| {
                if event.kind == kinds::APPROVAL_DEFERRED {
                    RewriteDecision::Replace(br#"{"marker":"after"}"#.to_vec())
                } else {
                    RewriteDecision::Keep
                }
            }),
        )
        .await
        .expect("rewrite_partition");
    // The report the mutating command makes on its receipt. Nothing else in
    // this crate can know a rewrite happened: the record count is unchanged.
    cached_authority.invalidate_partition("conv-erasure");

    let after = scoped
        .execute("SELECT payload_json FROM events_raw WHERE kind_base = 'approval_deferred'")
        .await
        .expect("query after rewrite");
    assert_eq!(
        after.rows,
        vec![vec![serde_json::json!(r#"{"marker":"after"}"#)]],
        "the erased (pre-rewrite) payload must NEVER be served again — the reported mutation \
         must force a full rebuild even though the event count is unchanged"
    );

    fx.teardown().await;
}

/// (d) Destroying a partition evicts its cache entry (memory hygiene) and a
/// subsequent query over it returns empty, not the stale cached rows.
#[tokio::test]
async fn destroying_a_partition_evicts_its_cache_entry() {
    let fx = Fixture::build("cache-destroy").await;
    let persona_id = fx.make_admin("destroy-admin").await;
    fx.eventlog
        .append_batch(
            "conv-destroy".to_owned(),
            vec![Event::new("k0", Vec::new())],
        )
        .await
        .expect("append");

    let cached_authority = fx.authority_with_cache_config(CacheConfig::new(Some(1)));
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    // Filtered to this test's own `k0` kind — `append_batch` also durably
    // appends its own MMR signed-root marker event per call (#799,
    // `EventLogHost::append_batch`'s doc), which an unfiltered `COUNT(*)`
    // would count too.
    let before = scoped
        .execute(
            "SELECT COUNT(*) AS c FROM events_raw WHERE partition = 'conv-destroy' AND kind_base = 'k0'",
        )
        .await
        .expect("query before destroy");
    assert_eq!(before.rows, vec![vec![serde_json::json!(1)]]);

    fx.eventlog
        .destroy_partition("conv-destroy".to_owned())
        .await
        .expect("destroy_partition");

    let after = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw WHERE partition = 'conv-destroy'")
        .await
        .expect("query after destroy");
    assert_eq!(
        after.rows,
        vec![vec![serde_json::json!(0)]],
        "a destroyed partition must read back empty, not the evicted cache's stale rows"
    );

    fx.teardown().await;
}

/// (e) With the kill switch OFF (the default `Fixture` authority), every
/// query reaches `QueryEngine::build_from_tables` fresh — the pre-Phase-A
/// behavior, unchanged. Diffs `crate::engine::BUILD_CALL_COUNT` across two
/// IDENTICAL queries to prove neither is served from a cache that does not
/// exist for this authority.
#[tokio::test]
async fn cache_disabled_replays_and_rebuilds_every_query() {
    let fx = Fixture::build("cache-disabled").await;
    let persona_id = fx.make_admin("disabled-admin").await;
    fx.eventlog
        .append_batch(
            "conv-disabled".to_owned(),
            vec![Event::new("k0", Vec::new())],
        )
        .await
        .expect("append");

    // `fx.authority` was built via `Fixture::build`, which passes
    // `CacheConfig::disabled()`.
    let scoped = admin_scoped(&fx, &fx.authority, persona_id).await;

    let build_before = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("first query");
    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("second, identical query");
    let build_after = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert_eq!(
        build_after - build_before,
        2,
        "with the kill switch off, every query — even a repeat of the exact same SQL over an \
         unchanged partition — must reach engine assembly fresh"
    );

    fx.teardown().await;
}

/// (f) The decode cache's cached-scan volume bound: a scope whose EFFECTIVE volume
/// (here, a cache miss's own freshly-decoded tables) exceeds
/// `CacheConfig::max_cached_source_events` is rejected before
/// `QueryEngine::build_from_tables` ever runs — even though the SAME
/// scope's replayed-event count comfortably fits under
/// `QueryLimits::max_source_events`. Also pins leak-freedom (item h): the
/// rejection message names no count.
#[tokio::test]
async fn cached_volume_budget_trips_before_engine_assembly() {
    let fx = Fixture::build("cache-volume-budget").await;
    let persona_id = fx.make_admin("volume-admin").await;
    fx.eventlog
        .append_batch(
            "conv-volume".to_owned(),
            (0..6)
                .map(|i| Event::new(format!("k{i}"), Vec::new()))
                .collect(),
        )
        .await
        .expect("append");

    let cache_config = CacheConfig {
        enabled: true,
        max_bytes: 64 * 1024 * 1024,
        // Comfortably under the 6 events just appended, but the default
        // `QueryLimits::max_source_events` (500,000) would never trip on
        // its own — proving THIS check, not the pre-existing one, is what
        // rejects the query.
        max_cached_source_events: 4,
    };
    let cached_authority = fx.authority_with_cache_config(cache_config);
    let scoped = admin_scoped(&fx, &cached_authority, persona_id).await;

    let build_before = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .unwrap_err();
    let build_after = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "6 effective cached-scan events over a 4-event cached-volume budget must be rejected: \
         {err:?}"
    );
    assert_eq!(
        build_after, build_before,
        "QueryEngine::build_from_tables must never be reached when the cached-scan volume \
         budget rejects a query"
    );

    let message = err.to_string();
    assert!(
        !message.contains('6') && !message.contains('4'),
        "the cached-volume rejection message must not leak the actual or budgeted event count: \
         {message:?}"
    );
    assert!(
        message.contains("cached-scan volume"),
        "the message must name what budget tripped: {message:?}"
    );

    fx.teardown().await;
}

/// The defect this fix closes: a partition cached once, then receiving a
/// single large ORDINARY append (no adversarial input needed — a big tool
/// result is enough), must have its NEXT query's cache-tail replay bounded
/// by `QueryLimits::max_source_bytes` the same way a cache MISS already is —
/// never materializing the whole oversized tail into memory before the
/// budget is consulted.
///
/// This harness cannot observe peak process memory directly, so the proof is
/// indirect but still tight: (1) `polychrome_query_cache_tail_total`
/// increments, confirming this query actually took the `Lookup::Tail` arm
/// (not a fresh `Lookup::Miss`, which already had its own byte budget before
/// this fix) — so the assertion below is really exercising the tail's own
/// bound; (2) the query is rejected with the same
/// `ScopedQueryError::SourceBudgetExceeded` a miss-arm rejection gets; (3)
/// `QueryEngine::build_from_tables` is never reached, the same
/// rejected-before-decode proof the sibling byte-budget tests give for the
/// miss arm.
#[tokio::test]
async fn cache_tail_over_bytes_budget_is_rejected_before_engine_assembly() {
    let fx = Fixture::build("cache-tail-bytes-budget").await;
    let persona_id = fx.make_admin("tail-bytes-admin").await;

    // A small first event populates the cache — comfortably under the
    // 2,000-byte budget below, so the FIRST query is a normal miss, not
    // itself a rejection.
    fx.eventlog
        .append_batch(
            "conv-tail-bytes".to_owned(),
            vec![Event::new("k0", vec![0u8; 10])],
        )
        .await
        .expect("append first (small) event");

    let limited_cached_authority = fx.authority_with_limits_and_cache_config(
        QueryLimits {
            max_source_bytes: 2_000,
            ..QueryLimits::default()
        },
        CacheConfig::new(Some(1)),
    );
    let scoped = admin_scoped(&fx, &limited_cached_authority, persona_id).await;

    scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .expect("first query (miss, populates the cache; 10 bytes is comfortably under budget)");

    // A single large ordinary append onto the already-cached partition — the
    // exact scenario the bug report names: no attacker, just one big tool
    // result landing after the partition was last cached.
    fx.eventlog
        .append_batch(
            "conv-tail-bytes".to_owned(),
            vec![Event::new("k1", vec![0u8; 5_000])],
        )
        .await
        .expect("append the oversized tail event");

    let tails_before = counter_value(&scrape_metrics(), "polychrome_query_cache_tail_total");
    let build_before = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let err = scoped
        .execute("SELECT COUNT(*) AS c FROM events_raw")
        .await
        .unwrap_err();
    let build_after = crate::engine::BUILD_CALL_COUNT.load(std::sync::atomic::Ordering::SeqCst);
    let tails_after = counter_value(&scrape_metrics(), "polychrome_query_cache_tail_total");

    assert_eq!(
        tails_after - tails_before,
        1.0,
        "this query must actually take the Tail arm (not a fresh Miss) — otherwise this test \
         would only be re-proving the miss arm's own pre-existing budget, not the tail arm's"
    );
    assert!(
        matches!(err, ScopedQueryError::SourceBudgetExceeded(_)),
        "a 5,000-byte tail over a 2,000-byte budget must be rejected: {err:?}"
    );
    assert_eq!(
        build_after, build_before,
        "QueryEngine::build_from_tables must never be reached when the cache-tail arm's own \
         byte budget rejects a query — proving the tail was never fully materialized before \
         the rejection"
    );

    fx.teardown().await;
}

/// A rendering of a `ScopedQuery`'s whole per-caller half — what it may read,
/// whether it may `EXPLAIN`, and what its audit record will say — in a shape
/// two sessions can be compared on. `QueryScope` carries no `PartialEq`, so
/// the scope is rendered as its own conversation list rather than compared
/// structurally.
#[derive(Debug, PartialEq, Eq)]
struct RenderedScoping {
    conversations: Option<Vec<String>>,
    allow_explain: bool,
    caller_identity: Option<String>,
    conversation_id: Option<String>,
    turn_id: Option<String>,
    web_session_id: Option<String>,
}

fn scoping_of(scoped: &ScopedQuery) -> RenderedScoping {
    RenderedScoping {
        conversations: match &scoped.scope {
            QueryScope::Fleet => None,
            QueryScope::Conversations(ids) => Some(ids.clone()),
        },
        allow_explain: scoped.allow_explain,
        caller_identity: scoped.caller_identity.clone(),
        conversation_id: scoped.conversation_id.clone(),
        turn_id: scoped.turn_id.clone(),
        web_session_id: scoped.web_session_id.clone(),
    }
}

/// `scope_for_turn` and the verified-grant path produce the SAME session, not
/// two that happen to agree today.
///
/// The native builtin tools reach the engine through `scope_for_turn`, which
/// has no token to verify; the explorer and the pre-#1675 connector reach it
/// through a verified `Principal::ConversationGrant`. Both must land on one
/// conversation's redacted, `EXPLAIN`-less view of its own history. A change
/// that widened `allow_explain` — or the non-Fleet redacted registration
/// `ScopedQuery::execute_with_params` selects from the same two fields — on
/// one path alone would be a redaction bypass reachable from exactly one
/// surface. `Scoping::for_conversation` plus the one private session
/// constructor is what makes that unrepresentable; this holds the shape.
#[tokio::test]
async fn scope_for_turn_matches_the_conversation_grant_path() {
    let fx = Fixture::build("scope-for-turn-parity").await;
    let token = mint_conversation_grant(
        &fx.signer.relabel_for_test(),
        "conv-parity",
        GrantSubject::Turn("turn-parity".to_owned()),
        NOW + TEST_TTL_MS,
    );
    let principal = fx
        .authority
        .verify_conversation_grant(&token, NOW)
        .expect("a freshly minted grant must verify");
    let via_grant = fx
        .authority
        .scope_for(&principal)
        .await
        .expect("a conversation grant always scopes");
    let via_turn = fx.authority.scope_for_turn("conv-parity", "turn-parity");

    assert_eq!(
        scoping_of(&via_turn),
        scoping_of(&via_grant),
        "the trusted-side turn path and the verified-grant path must produce one session shape"
    );
    // Spelled out too, so a future reader of a failure sees WHICH property
    // the equality above is protecting rather than a tuple diff.
    assert!(
        !via_turn.allow_explain,
        "a conversation-scoped session never gets EXPLAIN, however it was obtained"
    );
    assert_eq!(via_turn.turn_id(), Some("turn-parity"));
    assert_eq!(via_turn.conversation_id(), Some("conv-parity"));
    assert_eq!(
        via_turn.caller_identity(),
        None,
        "matching the grant path's audit shape exactly is deliberate — see scope_for_turn's doc"
    );
    assert_eq!(via_turn.web_session_id(), None);
    match &via_turn.scope {
        QueryScope::Conversations(ids) => {
            assert_eq!(
                ids,
                &vec!["conv-parity".to_owned()],
                "one conversation, its own"
            );
        }
        QueryScope::Fleet => panic!("a turn-scoped session must never be Fleet"),
    }

    fx.teardown().await;
}

/// Unit-level proof of [`is_admitted_partition`] itself — the ONE admission
/// rule shared by [`ScopedQuery::replay_scoped_partitions`]'s Fleet arm,
/// [`ScopedQuery::resolve_partitions_cached`]'s Fleet arm, and
/// [`ScopedQuery::estimate_source_event_total`]. Every conversation partition
/// is admitted; the scheduler partition is admitted; no other
/// non-conversation partition is ever admitted — the exact narrowness issue
/// #1592 promises and
/// `fleet_replay_admits_the_routine_scheduler_partition_but_no_other_non_conversation_partition`
/// already pins end to end for one call site.
#[test]
fn is_admitted_partition_admits_exactly_conv_prefixed_and_the_scheduler_partition() {
    assert!(is_admitted_partition("conv-a"));
    assert!(
        is_admitted_partition(ROUTINE_SCHEDULER_PARTITION),
        "the scheduler partition must be admitted"
    );
    assert!(
        !is_admitted_partition("some-other-partition"),
        "no other non-conversation partition is ever admitted"
    );
}

/// Regression test for issue #1592/#1882's decode-cache-path drift:
/// `ScopedQuery::resolve_partitions_cached`'s Fleet arm must not hard-filter
/// every discovered partition to `"conv-"` — doing so would silently drop
/// the routine scheduler's own dedicated partition — and its Conversations
/// arm must admit that partition too, or a cache-ENABLED deployment
/// (`crate::cache::CacheConfig::new`'s own doc: any single-replica
/// deployment, production's exact shape) would return an empty `fires` table
/// fleet-wide and per-persona, even though the SAME query against a
/// cache-DISABLED authority (every other test in this module) returns rows.
/// This test builds a `QueryAuthority` with the cache turned ON via
/// `Fixture::authority_with_routine_catalog_and_cache_config`, appends a real
/// `RoutineFiredEvent` to `"routine-scheduler"` through the fixture's real
/// `EventLogHost`, then runs `SELECT routine FROM fires` under BOTH a Fleet
/// session and the routine's own owner-scoped persona session — each query
/// runs TWICE, so the second run exercises the cache-HIT path
/// (`Lookup::Hit`), not merely a cold `Lookup::Miss`.
#[tokio::test]
async fn cache_enabled_fleet_and_owner_scoped_fires_see_the_scheduler_partition() {
    let fx = Fixture::build("cache-enabled-fires-admission").await;

    let owner_persona = fx.make_non_admin("dana-owner").await;
    let catalog = Arc::new(FakeRoutineCatalog(vec![routine_record(
        "cache-path-standup",
        &owner_persona,
    )]));
    let authority =
        fx.authority_with_routine_catalog_and_cache_config(catalog, CacheConfig::new(Some(1)));

    let fired = RoutineFiredEvent {
        routine: "cache-path-standup".to_owned(),
        occurrence: "cache-path-standup-1".to_owned(),
        scheduled_at_ms: 1,
        fired_at_ms: 2,
        routine_uid: "cache-path-standup-uid".to_owned(),
        ..Default::default()
    };
    fx.eventlog
        .append_batch(
            "routine-scheduler".to_owned(),
            vec![Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec())],
        )
        .await
        .expect("append routine-scheduler fire");

    // Fleet session: run twice — the first resolves through
    // `Lookup::Miss`/`Lookup::Tail`, the second must hit `Lookup::Hit` for
    // every already-resolved partition (nothing was appended in between).
    let admin_persona = fx.make_admin("erin-admin").await;
    let admin_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: admin_persona.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let admin_principal = authority
        .verify_admin_session(&admin_token, NOW)
        .await
        .expect("valid admin session");
    let scoped_fleet = authority
        .scope_for(&admin_principal)
        .await
        .expect("scope_for fleet");
    for attempt in 0..2 {
        let fleet_result = scoped_fleet
            .execute("SELECT routine FROM fires ORDER BY routine")
            .await
            .unwrap_or_else(|err| panic!("fleet fires query attempt {attempt} failed: {err}"));
        assert_eq!(
            fleet_result.rows,
            vec![vec![serde_json::json!("cache-path-standup")]],
            "fleet-scoped fires must be non-empty with the cache enabled, attempt {attempt}: {:?}",
            fleet_result.rows
        );
        assert!(
            !fleet_result.columns.is_empty(),
            "fleet-scoped fires must report non-empty columns, attempt {attempt}"
        );
    }

    // Owner-scoped session: same two-attempt shape, proving the
    // `resolve_partitions_cached` Conversations arm's own scheduler-partition
    // admission (mirroring `replay_scoped_partitions`'s Conversations arm).
    let owner_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: owner_persona.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let owner_principal = authority
        .verify_admin_session(&owner_token, NOW)
        .await
        .expect("valid owner session");
    assert!(
        matches!(owner_principal, Principal::Persona(_)),
        "a non-admin persona must mint Principal::Persona, got {owner_principal:?}"
    );
    let scoped_owner = authority
        .scope_for(&owner_principal)
        .await
        .expect("scope_for owner");
    for attempt in 0..2 {
        let owner_result = scoped_owner
            .execute("SELECT routine FROM fires ORDER BY routine")
            .await
            .unwrap_or_else(|err| {
                panic!("owner-scoped fires query attempt {attempt} failed: {err}")
            });
        assert_eq!(
            owner_result.rows,
            vec![vec![serde_json::json!("cache-path-standup")]],
            "owner-scoped fires must be non-empty with the cache enabled, attempt {attempt}: {:?}",
            owner_result.rows
        );
        assert!(
            !owner_result.columns.is_empty(),
            "owner-scoped fires must report non-empty columns, attempt {attempt}"
        );
    }

    fx.teardown().await;
}

/// Mint and verify a persona-scoped session for `persona_id`, returning its
/// [`ScopedQuery`] — the module-level twin of the nested helper
/// `persona_scoped_routines_and_fires_see_only_the_callers_own_routine`
/// defines, shared by the grant/setup tests below.
async fn scope_for_persona(
    authority: &QueryAuthority,
    signer: &SessionSigner,
    persona_id: &str,
) -> ScopedQuery {
    let token = mint_session(
        signer,
        &SessionSubject::Persona {
            persona_id: persona_id.to_owned(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let principal = authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("valid session");
    assert!(
        matches!(principal, Principal::Persona(_)),
        "a non-admin persona must mint Principal::Persona, got {principal:?}"
    );
    authority.scope_for(&principal).await.expect("scope_for")
}

/// One signed routine-grant `approval_response` event — the exact
/// shape `append_signed_response`/`append_routine_grant` mint into a
/// routine's fire conversation, signed by the fixture's own trusted signer.
fn grant_event(
    tool_name: &str,
    approved: bool,
    conversation_id: &str,
    grant_scope: &str,
    signer: &ApprovalSigner,
) -> Event {
    let turn = uuid::Uuid::now_v7();
    let (payload, ..) = polyc_crypto::approval::routine_grant_payload(
        &format!("call-{tool_name}-{approved}"),
        tool_name,
        "{}",
        "",
        approved,
        "owner-persona",
        "",
        "default",
        "",
        &[],
        conversation_id,
        &uuid::Uuid::new_v4().to_string(),
        &turn.to_string(),
        "hash-fixture",
        grant_scope,
        signer,
    );
    Event::new(kinds::tagged(kinds::APPROVAL_RESPONSE, &turn), payload)
}

/// The core owner-scoping proof for the grants ledger: two personas
/// each own one routine whose fire conversation holds one signed per-tool
/// grant; a persona-scoped session sees ONLY its own routine's grant (in
/// `routine_grants`, `routine_active_grants`, and `routine_overview`), never
/// the other persona's, while Fleet sees both. The fire conversation reaches
/// a persona session through its own participations — a fire runs
/// owner-attributed (#1884), so the owner participates in it.
#[tokio::test]
async fn persona_scoped_routine_grants_see_only_the_callers_own_routine() {
    let fx = Fixture::build("owner-scoped-routine-grants").await;

    let persona_a = fx.make_non_admin("gina-owner").await;
    let persona_b = fx.make_non_admin("hana-owner").await;

    let catalog = Arc::new(FakeRoutineCatalog(vec![
        routine_record("routine-a", &persona_a),
        routine_record("routine-b", &persona_b),
    ]));
    let authority = fx.authority_with_routine_catalog(catalog);

    for (routine, persona, tool) in [
        ("routine-a", &persona_a, "tool-a"),
        ("routine-b", &persona_b, "tool-b"),
    ] {
        // `routine_record` derives `fire_conversation_id` as
        // `{name}-fire-conv`; the owner participates in that conversation
        // (the fire pipeline attributes the owner), so a persona-scoped
        // replay reads its partition.
        let fire_conv = format!("{routine}-fire-conv");
        let identity = ExternalIdentity {
            provider: "test".to_owned(),
            scope: "s".to_owned(),
            external_id: match routine {
                "routine-a" => "gina-owner".to_owned(),
                _ => "hana-owner".to_owned(),
            },
            display_name: persona.clone(),
            ..Default::default()
        };
        fx.persona
            .attribute(identity, fire_conv.clone(), "initiator".to_owned(), NOW)
            .await
            .expect("attribute owner to fire conversation");
        fx.eventlog
            .append_batch(
                format!("conv-{fire_conv}"),
                vec![grant_event(tool, true, &fire_conv, "tool", &fx.signer)],
            )
            .await
            .expect("append grant");
    }

    let scoped_a = scope_for_persona(&authority, &fx.signer.relabel_for_test(), &persona_a).await;
    let grants_a = scoped_a
        .execute("SELECT routine, tool_name, grant_scope, approved FROM routine_grants")
        .await
        .expect("persona a's routine_grants query");
    assert_eq!(
        grants_a.rows,
        vec![vec![
            serde_json::json!("routine-a"),
            serde_json::json!("tool-a"),
            serde_json::json!("tool"),
            serde_json::json!(true),
        ]],
        "persona a must see exactly their own routine's grant"
    );
    let active_a = scoped_a
        .execute("SELECT routine, tool_name FROM routine_active_grants")
        .await
        .expect("persona a's routine_active_grants query");
    assert_eq!(
        active_a.rows,
        vec![vec![
            serde_json::json!("routine-a"),
            serde_json::json!("tool-a")
        ]]
    );
    let overview_a = scoped_a
        .execute(
            "SELECT name, mode, active_tool_grants, setup_completed FROM routine_overview \
             ORDER BY name",
        )
        .await
        .expect("persona a's routine_overview query");
    assert_eq!(
        overview_a.rows,
        vec![vec![
            serde_json::json!("routine-a"),
            serde_json::json!("individual"),
            serde_json::json!(1),
            serde_json::json!(false),
        ]],
        "the aggregate answers from the tables alone, owner-scoped"
    );

    // POLY-160: the routine_overview columns that RIDE STRAIGHT off the
    // routines reference table (prompt, suspended/paused_by/paused_at_ms/
    // pause_reason) and the two that fold `fires_raw`/the fire
    // conversation's denials (last_fire_outcome, stopped_tool) are scoped
    // by the SAME `own_rows_scope`-filtered `routines` join every other
    // column above already proved — persona a sees exactly one row here
    // too, never persona b's, even though `routine_record`'s fixture data
    // gives both routines identical prompt/pause-field values (so the ROW
    // COUNT and the `name` column, not an incidental value difference, are
    // what this assertion actually pins).
    let overview_a_new_columns = scoped_a
        .execute(
            "SELECT name, prompt, suspended, paused_by, paused_at_ms, pause_reason, \
             last_fire_outcome, stopped_tool FROM routine_overview",
        )
        .await
        .expect("persona a's routine_overview POLY-160-columns query");
    assert_eq!(
        overview_a_new_columns.rows.len(),
        1,
        "persona a must see exactly one row on the POLY-160 columns too — never owner b's \
         routine"
    );
    let row = &overview_a_new_columns.rows[0];
    assert_eq!(row[0], serde_json::json!("routine-a"));
    assert_eq!(row[1], serde_json::json!("post the morning standup"));
    assert_eq!(row[2], serde_json::json!(false), "suspended");
    assert_eq!(row[3], serde_json::Value::Null, "paused_by while active");
    assert_eq!(row[4], serde_json::Value::Null, "paused_at_ms while active");
    assert_eq!(row[5], serde_json::Value::Null, "pause_reason while active");
    assert_eq!(row[6], serde_json::Value::Null, "no fire ever recorded");
    assert_eq!(
        row[7],
        serde_json::Value::Null,
        "no unattended denial ever recorded"
    );

    let admin_persona = fx.make_admin("ivy-admin").await;
    let admin_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: admin_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let admin_principal = authority
        .verify_admin_session(&admin_token, NOW)
        .await
        .expect("valid admin session");
    let scoped_fleet = authority
        .scope_for(&admin_principal)
        .await
        .expect("scope_for");
    let grants_fleet = scoped_fleet
        .execute("SELECT routine, tool_name FROM routine_grants ORDER BY routine")
        .await
        .expect("fleet routine_grants query");
    assert_eq!(
        grants_fleet.rows,
        vec![
            vec![serde_json::json!("routine-a"), serde_json::json!("tool-a")],
            vec![serde_json::json!("routine-b"), serde_json::json!("tool-b")],
        ],
        "fleet must see every routine's grants, unfiltered"
    );

    fx.teardown().await;
}

/// The aggregate-semantics proof, at Fleet scope: one routine whose
/// fire conversation carries a per-tool grant later revoked, a standing
/// per-tool grant, a blanket grant (the derived mode), an ordinary
/// unattended denial pair, and one unresolved setup `approval_request`;
/// plus a `routine_setup_completed` marker and one fire on the scheduler
/// partition. `routine_overview` composes all of it from the typed tables
/// alone.
#[tokio::test]
async fn routine_overview_composes_grants_mode_denials_pending_and_setup() {
    let fx = Fixture::build("routine-overview-composition").await;

    let owner = fx.make_non_admin("kira-owner").await;
    let catalog = Arc::new(FakeRoutineCatalog(vec![routine_record(
        "routine-x",
        &owner,
    )]));
    let authority = fx.authority_with_routine_catalog(catalog);
    let fire_conv = "routine-x-fire-conv";
    let partition = format!("conv-{fire_conv}");

    // Grant history: `revoked-tool` granted then revoked (latest
    // disapproving record removes the key), `kept-tool` granted and
    // standing, one blanket grant deriving mode `auto`.
    fx.eventlog
        .append_batch(
            partition.clone(),
            vec![
                grant_event("revoked-tool", true, fire_conv, "tool", &fx.signer),
                grant_event("kept-tool", true, fire_conv, "tool", &fx.signer),
                grant_event("revoked-tool", false, fire_conv, "tool", &fx.signer),
                grant_event("", true, fire_conv, "blanket_below_high", &fx.signer),
            ],
        )
        .await
        .expect("append grant history");

    // An ordinary unattended denial: signed request + denied (non-grant)
    // response for the same occurrence.
    let denial_turn = uuid::Uuid::now_v7();
    let request = polyc_crypto::approval::request_payload(
        "call-denied",
        "denied-tool",
        "{}",
        "default",
        "",
        &[],
        "",
        "",
        "",
        &[],
        false,
    );
    let (denied, ..) = polyc_crypto::approval::response_payload(
        "call-denied",
        "denied-tool",
        "{}",
        "",
        false,
        false,
        &[],
        &owner,
        "",
        "default",
        "not allowed on unattended runs",
        "",
        fire_conv,
        "nonce-denied",
        &denial_turn.to_string(),
        &fx.signer,
    );
    // A setup rehearsal's still-parked approval: a request with no response.
    let pending_turn = uuid::Uuid::now_v7();
    let pending_request = polyc_crypto::approval::request_payload(
        "call-pending",
        "pending-tool",
        "{}",
        "default",
        "",
        &[],
        "",
        "",
        "",
        &[],
        false,
    );
    fx.eventlog
        .append_batch(
            partition.clone(),
            vec![
                Event::new(
                    kinds::tagged(kinds::APPROVAL_REQUEST, &denial_turn),
                    request,
                ),
                Event::new(
                    kinds::tagged(kinds::APPROVAL_RESPONSE, &denial_turn),
                    denied,
                ),
                Event::new(
                    kinds::tagged(kinds::APPROVAL_REQUEST, &pending_turn),
                    pending_request,
                ),
            ],
        )
        .await
        .expect("append denial and pending request");

    // Scheduler partition: one fire (with its sibling outcome event, #1656)
    // and the setup-completed marker.
    let fired = RoutineFiredEvent {
        routine: "routine-x".to_owned(),
        occurrence: "routine-x-1".to_owned(),
        scheduled_at_ms: 10,
        fired_at_ms: 20,
        routine_uid: "routine-x-uid".to_owned(),
        ..Default::default()
    };
    let outcome_ev = polyc_proto::proto::polychrome::events::v1::RoutineFireOutcomeEvent {
        routine: fired.routine.clone(),
        occurrence: fired.occurrence.clone(),
        outcome: polyc_proto::proto::polychrome::events::v1::RoutineFireOutcome::StoppedUngranted
            .into(),
        fired_at_ms: fired.fired_at_ms,
        ..Default::default()
    };
    let setup = serde_json::json!({ "routine_uid": "routine-x-uid" }).to_string();
    fx.eventlog
        .append_batch(
            "routine-scheduler".to_owned(),
            vec![
                Event::trusted(kinds::ROUTINE_FIRED, fired.encode_to_vec()),
                Event::trusted(kinds::ROUTINE_FIRE_OUTCOME, outcome_ev.encode_to_vec()),
                Event::trusted(kinds::ROUTINE_SETUP_COMPLETED, setup.into_bytes()),
            ],
        )
        .await
        .expect("append scheduler markers");

    let admin_persona = fx.make_admin("liam-admin").await;
    let admin_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: admin_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let admin_principal = authority
        .verify_admin_session(&admin_token, NOW)
        .await
        .expect("valid admin session");
    let scoped = authority
        .scope_for(&admin_principal)
        .await
        .expect("scope_for");

    let active = scoped
        .execute("SELECT tool_name FROM routine_active_grants WHERE grant_scope = 'tool'")
        .await
        .expect("active grants query");
    assert_eq!(
        active.rows,
        vec![vec![serde_json::json!("kept-tool")]],
        "a latest disapproving record removes its key; the standing grant survives"
    );

    let overview = scoped
        .execute(
            "SELECT name, mode, setup_completed, fire_count, active_tool_grants, \
             denial_count, pending_setup_approvals FROM routine_overview",
        )
        .await
        .expect("overview query");
    assert_eq!(
        overview.rows,
        vec![vec![
            serde_json::json!("routine-x"),
            serde_json::json!("auto"),
            serde_json::json!(true),
            serde_json::json!(1),
            serde_json::json!(1),
            serde_json::json!(1),
            serde_json::json!(1),
        ]],
        "the per-routine aggregate composes fires, grants, mode, denials, pending setup \
         approvals, and setup state from the typed tables alone"
    );

    // POLY-160: the routine's own spec fields (carried straight from
    // `routines`), the latest-fire outcome (windowed, not aggregated), and
    // `stopped_tool` (the routine's latest unattended-fire denial).
    let extended = scoped
        .execute(
            "SELECT prompt, schedule_json, next_fires_json, suspended, paused_by, \
             paused_at_ms, pause_reason, last_fire_at_ms, last_fire_outcome, stopped_tool \
             FROM routine_overview",
        )
        .await
        .expect("extended overview query");
    let record = routine_record("routine-x", &owner);
    assert_eq!(
        extended.rows,
        vec![vec![
            serde_json::json!(record.prompt),
            serde_json::json!(record.schedule_json),
            serde_json::json!(record.next_fires_json),
            serde_json::json!(record.suspended),
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::Value::Null,
            serde_json::json!(20),
            serde_json::json!("stopped_ungranted"),
            serde_json::json!("denied-tool"),
        ]],
        "the routine's own spec fields pass through from `routines`, the latest fire's own \
         outcome is picked by window rather than aggregated, and stopped_tool names the \
         routine's latest unattended-fire denial"
    );

    fx.teardown().await;
}

/// The routine-read-surface mirror of
/// [`cache_enabled_fleet_and_owner_scoped_fires_see_the_scheduler_partition`]
/// — the pinned regression class from issue #1918: the decode cache path
/// (production's exact shape on any single-replica deployment) must decode
/// and admit the `routine_setup` table's scheduler-partition rows and the
/// `routine_grants` fire-conversation rows for BOTH a Fleet and an
/// owner-scoped session, on the cache-hit path specifically (each query runs
/// twice; the second run is `Lookup::Hit`).
#[tokio::test]
async fn cache_enabled_routine_setup_and_grants_see_their_partitions() {
    let fx = Fixture::build("cache-enabled-routine-setup").await;

    let owner = fx.make_non_admin("mona-owner").await;
    let catalog = Arc::new(FakeRoutineCatalog(vec![routine_record(
        "cache-routine",
        &owner,
    )]));
    let authority =
        fx.authority_with_routine_catalog_and_cache_config(catalog, CacheConfig::new(Some(1)));

    let fire_conv = "cache-routine-fire-conv";
    let owner_identity = ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: "mona-owner".to_owned(),
        display_name: "mona-owner".to_owned(),
        ..Default::default()
    };
    fx.persona
        .attribute(
            owner_identity,
            fire_conv.to_owned(),
            "initiator".to_owned(),
            NOW,
        )
        .await
        .expect("attribute owner to fire conversation");
    fx.eventlog
        .append_batch(
            format!("conv-{fire_conv}"),
            vec![grant_event(
                "cache-tool",
                true,
                fire_conv,
                "tool",
                &fx.signer,
            )],
        )
        .await
        .expect("append grant");
    let setup = serde_json::json!({ "routine_uid": "cache-routine-uid" }).to_string();
    fx.eventlog
        .append_batch(
            "routine-scheduler".to_owned(),
            vec![Event::trusted(
                kinds::ROUTINE_SETUP_COMPLETED,
                setup.into_bytes(),
            )],
        )
        .await
        .expect("append setup marker");

    let admin_persona = fx.make_admin("nova-admin").await;
    let admin_token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: admin_persona,
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    let admin_principal = authority
        .verify_admin_session(&admin_token, NOW)
        .await
        .expect("valid admin session");
    let scoped_fleet = authority
        .scope_for(&admin_principal)
        .await
        .expect("scope_for fleet");
    for attempt in 0..2 {
        let setup_rows = scoped_fleet
            .execute("SELECT routine_uid FROM routine_setup")
            .await
            .unwrap_or_else(|err| panic!("fleet routine_setup attempt {attempt} failed: {err}"));
        assert_eq!(
            setup_rows.rows,
            vec![vec![serde_json::json!("cache-routine-uid")]],
            "fleet routine_setup must be non-empty with the cache enabled, attempt {attempt}"
        );
        let grant_rows = scoped_fleet
            .execute("SELECT tool_name FROM routine_grants")
            .await
            .unwrap_or_else(|err| panic!("fleet routine_grants attempt {attempt} failed: {err}"));
        assert_eq!(
            grant_rows.rows,
            vec![vec![serde_json::json!("cache-tool")]],
            "fleet routine_grants must be non-empty with the cache enabled, attempt {attempt}"
        );
    }

    let scoped_owner = scope_for_persona(&authority, &fx.signer.relabel_for_test(), &owner).await;
    for attempt in 0..2 {
        let setup_rows = scoped_owner
            .execute("SELECT routine_uid FROM routine_setup")
            .await
            .unwrap_or_else(|err| panic!("owner routine_setup attempt {attempt} failed: {err}"));
        assert_eq!(
            setup_rows.rows,
            vec![vec![serde_json::json!("cache-routine-uid")]],
            "owner routine_setup must be non-empty with the cache enabled, attempt {attempt}"
        );
        let overview = scoped_owner
            .execute("SELECT name, setup_completed, active_tool_grants FROM routine_overview")
            .await
            .unwrap_or_else(|err| panic!("owner routine_overview attempt {attempt} failed: {err}"));
        assert_eq!(
            overview.rows,
            vec![vec![
                serde_json::json!("cache-routine"),
                serde_json::json!(true),
                serde_json::json!(1),
            ]],
            "the owner-scoped aggregate must compose on the cache-enabled path, attempt {attempt}"
        );
    }

    fx.teardown().await;
}

/// A session outlives the de-admission of the persona it names, and is
/// refused on the very next request.
///
/// The window this closes is real, not theoretical. `remove_access` KEEPS the
/// persona's record as a forensics-readable tombstone, so the `profile` read
/// alone would still answer `Some` for a removed persona, letting this
/// funnel mint a participation-scoped principal for one. Removal cannot revoke the
/// session either: sessions are stateless, and `RevokedTokens` denies only an
/// exact token string, which de-admission never sees. So a de-admitted caller
/// arrives here with a signature-valid, unexpired token, and this check is
/// the only thing between it and that persona's whole participation scope.
#[tokio::test]
async fn admin_session_removed_persona_is_not_authorized_for_fleet() {
    let fx = Fixture::build("admin-removed-persona").await;
    let admin_id = fx.make_admin("remover").await;
    let target_id = fx.make_non_admin("removed-later").await;
    let target_identity = ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: "removed-later".to_owned(),
        display_name: "removed-later".to_owned(),
        ..Default::default()
    };
    // Minted while the persona was still live — the whole point.
    let token = mint_session(
        &fx.signer.relabel_for_test(),
        &SessionSubject::Persona {
            persona_id: target_id.clone(),
        },
        &[SessionScope::ExplorerRead],
        NOW,
        TEST_TTL_MS,
    );
    fx.authority
        .verify_admin_session(&token, NOW)
        .await
        .expect("precondition: the session verifies while the persona is live");

    fx.persona
        .remove_access(admin_id, target_identity, NOW + 1)
        .await
        .expect("remove_access");

    let err = fx
        .authority
        .verify_admin_session(&token, NOW + 2)
        .await
        .unwrap_err();
    assert!(
        matches!(err, PrincipalError::NotAuthorizedForFleet),
        "a de-admitted persona's still-valid session must mint no principal at all, got {err:?}"
    );
    fx.teardown().await;
}

// ---------------------------------------------------------------------
// `resolve_search_scope` — the trusted participation search-scope authority.
// ---------------------------------------------------------------------

/// Tie a fresh persona (a distinct `test`-provider identity keyed on `label`)
/// to every conversation in `conversation_ids`, in the given order, and
/// return its persona id. One `attribute` call per conversation — mirrors
/// how a real caller accrues participation one turn at a time.
async fn attribute_persona_to(fx: &Fixture, label: &str, conversation_ids: &[&str]) -> String {
    let identity = ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: label.to_owned(),
        display_name: label.to_owned(),
        ..Default::default()
    };
    let mut persona_id = None;
    for conversation_id in conversation_ids {
        let resolved = fx
            .persona
            .attribute(
                identity.clone(),
                (*conversation_id).to_owned(),
                "initiator".to_owned(),
                NOW,
            )
            .await
            .expect("attribute")
            .persona_id;
        persona_id = Some(resolved);
    }
    persona_id.expect("at least one conversation id")
}

/// Independently reimplements [`SearchScope::hash`]'s pinned five-step
/// algorithm — deliberately NOT by calling `canonical_search_scope`, so a bug
/// in that function's own sort/dedupe/hash logic doesn't also corrupt this
/// check. Sorting by `(byte length, then bytes)` is mathematically identical
/// to sorting the length-prefixed encoded records bytewise: the `u32`
/// big-endian length prefix always differs before the content can, so the
/// length compares first either way.
fn expected_scope_hash(conversation_ids: &[&str]) -> String {
    let mut ids: Vec<&str> = conversation_ids.to_vec();
    ids.sort_by_key(|id| (id.len(), *id));
    ids.dedup();

    let mut buf = Vec::new();
    buf.extend_from_slice(b"polychrome.search.scope.v1");
    buf.push(0);
    for id in ids {
        let bytes = id.as_bytes();
        buf.extend_from_slice(&u32::try_from(bytes.len()).unwrap().to_be_bytes());
        buf.extend_from_slice(bytes);
    }
    blake3::hash(&buf).to_hex().to_string()
}

/// A removed persona refuses search-scope resolution, distinctly from any
/// other failure — mirrors `admin_session_removed_persona_is_not_authorized_for_fleet`'s
/// own de-admission window: the record is kept (a tombstone, not a
/// deletion), so this must read through `active_persona` fresh rather than a
/// cached or point-in-time verdict.
#[tokio::test]
async fn search_scope_removed_persona_refuses() {
    let fx = Fixture::build("search-scope-removed").await;
    let admin_id = fx.make_admin("remover").await;
    let target_id = attribute_persona_to(&fx, "removed-later", &["conv-a"]).await;
    let target_identity = ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: "removed-later".to_owned(),
        display_name: "removed-later".to_owned(),
        ..Default::default()
    };

    fx.persona
        .remove_access(admin_id, target_identity, NOW + 1)
        .await
        .expect("remove_access");

    let err = fx
        .authority
        .resolve_search_scope(&target_id, "conv-caller", "turn-1")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SearchScopeError::PersonaNotActive),
        "a de-admitted persona must refuse search-scope resolution: {err:?}"
    );
    fx.teardown().await;
}

/// An unknown persona id — one that never resolved to any profile — refuses
/// exactly like a removed one; the two are deliberately indistinguishable to
/// the caller (see `SearchScopeError::PersonaNotActive`'s own doc).
#[tokio::test]
async fn search_scope_unknown_persona_refuses() {
    let fx = Fixture::build("search-scope-unknown").await;
    let err = fx
        .authority
        .resolve_search_scope("persona-never-existed", "conv-caller", "turn-1")
        .await
        .unwrap_err();
    assert!(matches!(err, SearchScopeError::PersonaNotActive));
    fx.teardown().await;
}

/// The persona store being unreachable is infrastructure, never an
/// authorization verdict — the same posture every other funnel in this
/// module already takes for its own store read.
#[tokio::test]
async fn search_scope_store_down_is_unavailable() {
    let fx = Fixture::build("search-scope-store-down").await;
    let authority_no_store = fx.authority_with_empty_persona_cell();

    let err = authority_no_store
        .resolve_search_scope("persona-x", "conv-caller", "turn-1")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SearchScopeError::StoreUnavailable),
        "an unreadable persona store must surface as a distinct, transient error — never fold \
         into a hard refusal: {err:?}"
    );
    fx.teardown().await;
}

/// The calling conversation is excluded from the resolved scope —
/// participation-scoped search covers PREVIOUS conversations only.
#[tokio::test]
async fn search_scope_excludes_the_calling_conversation() {
    let fx = Fixture::build("search-scope-excludes-caller").await;
    let persona_id = attribute_persona_to(&fx, "caller-exclusion", &["conv-a", "conv-b"]).await;

    let scope = fx
        .authority
        .resolve_search_scope(&persona_id, "conv-a", "turn-1")
        .await
        .expect("an active persona with a bounded participation set resolves");

    assert_eq!(
        scope.conversation_ids(),
        &["conv-b".to_owned()],
        "the calling conversation must never appear in its own search scope"
    );
    assert_eq!(scope.count(), 1);
    assert_eq!(scope.hash(), expected_scope_hash(&["conv-b"]));
    fx.teardown().await;
}

/// A tombstoned conversation is excluded from the resolved scope — the sole
/// conversation-level revocation mechanism, applied before the scope hash is
/// computed. Search-visibility tombstones are required for V1.
#[tokio::test]
async fn search_scope_excludes_a_tombstoned_conversation() {
    let fx = Fixture::build("search-scope-excludes-tombstoned").await;
    let persona_id =
        attribute_persona_to(&fx, "tombstone-exclusion", &["conv-a", "conv-b", "conv-c"]).await;

    fx.persona
        .set_search_visibility(
            persona_id.clone(),
            "conv-b".to_owned(),
            true,
            persona_id.clone(),
            NOW + 1,
        )
        .await
        .expect("set_search_visibility");

    let scope = fx
        .authority
        .resolve_search_scope(&persona_id, "conv-a", "turn-1")
        .await
        .expect("an active persona with a bounded participation set resolves");

    assert_eq!(
        scope.conversation_ids(),
        &["conv-c".to_owned()],
        "conv-a is excluded as the caller, conv-b as tombstoned — only conv-c remains: {:?}",
        scope.conversation_ids()
    );
    fx.teardown().await;
}

/// An over-cap participation set refuses, and the refusal carries the
/// persona's total participation count — read from the enumeration index
/// alone, BEFORE any per-conversation tombstone read runs
/// (`PersonaStore::participation_scope`'s own refuse-before-per-tie-reads
/// contract).
#[tokio::test]
async fn search_scope_over_cap_refuses_with_count() {
    let fx = Fixture::build("search-scope-over-cap").await;
    let persona_id = attribute_persona_to(&fx, "over-cap", &["conv-a", "conv-b", "conv-c"]).await;
    fx.authority.set_test_search_scope_cap(2);

    let err = fx
        .authority
        .resolve_search_scope(&persona_id, "conv-a", "turn-1")
        .await
        .unwrap_err();
    assert!(
        matches!(err, SearchScopeError::OverCap { count: 3 }),
        "expected OverCap{{count: 3}}, got {err:?}"
    );
    fx.teardown().await;
}

/// The scope hash is independent of participation enumeration order — the
/// property the whole approval binding this scope feeds into rests on.
#[tokio::test]
async fn search_scope_hash_is_independent_of_enumeration_order() {
    let fx = Fixture::build("search-scope-order-independent").await;
    let ascending =
        attribute_persona_to(&fx, "order-ascending", &["conv-a", "conv-b", "conv-c"]).await;
    let descending =
        attribute_persona_to(&fx, "order-descending", &["conv-c", "conv-b", "conv-a"]).await;

    // "conv-x" names neither participant's own conversation, so it excludes
    // nothing from either scope — the two resolved sets stay identical, only
    // the ORDER each persona accrued them in differs.
    let via_ascending = fx
        .authority
        .resolve_search_scope(&ascending, "conv-x", "turn-1")
        .await
        .expect("ascending order resolves");
    let via_descending = fx
        .authority
        .resolve_search_scope(&descending, "conv-x", "turn-1")
        .await
        .expect("descending order resolves");

    assert_eq!(
        via_ascending.hash(),
        via_descending.hash(),
        "the same conversation set, accrued in a different order, must hash identically"
    );
    assert_eq!(
        via_ascending.conversation_ids(),
        via_descending.conversation_ids()
    );
    fx.teardown().await;
}

/// The scope hash changes when the underlying set changes — a hash that
/// cannot detect drift would defeat the whole reason it exists: binding
/// the scope the approver saw.
#[tokio::test]
async fn search_scope_hash_changes_when_the_set_changes() {
    let fx = Fixture::build("search-scope-hash-changes").await;
    let smaller = attribute_persona_to(&fx, "hash-smaller", &["conv-a", "conv-b"]).await;
    let larger = attribute_persona_to(&fx, "hash-larger", &["conv-a", "conv-b", "conv-c"]).await;

    let via_smaller = fx
        .authority
        .resolve_search_scope(&smaller, "conv-x", "turn-1")
        .await
        .expect("smaller set resolves");
    let via_larger = fx
        .authority
        .resolve_search_scope(&larger, "conv-x", "turn-1")
        .await
        .expect("larger set resolves");

    assert_ne!(
        via_smaller.hash(),
        via_larger.hash(),
        "a different conversation set must hash differently"
    );
    fx.teardown().await;
}

#[test]
fn core_requester_encoding_has_no_delimiter_collision() {
    let first = core_requester_id("turn", &["a:b", "c"]);
    let second = core_requester_id("turn", &["a", "b:c"]);
    let another_kind = core_requester_id("web-session", &["a:b", "c"]);

    assert_ne!(first, second);
    assert_ne!(first, another_kind);
    assert!(first.as_str().starts_with("turn-"));
}

// ---------------------------------------------------------------------------
// Entry-point parity (Step 3): both surfaces, one mechanism
// ---------------------------------------------------------------------------

/// Proves the two credential entry points cannot drift.
///
/// The embedded entry point is [`QueryAuthority`]'s own `verify_*` +
/// `scope_for` pair, which the control plane calls in process. The service
/// entry point is [`CredentialWitness::admit`], which the Query service calls
/// over the wire. Both funnel through the one shared
/// [`crate::credential::CredentialAuthority`], and each test below asserts
/// that they answer identically — the same grant, the same refusal, the same
/// scope — for one credential state.
///
/// A divergence here would be an authorization difference reachable from
/// exactly one surface, which is the hardest kind to notice in review.
mod credential_parity {
    use super::*;

    use crate::credential::{CredentialWitness, PresentedCredential, UnixClock};

    /// A clock the test sets, so expiry is exercised without waiting.
    struct FixedClock(u64);

    impl UnixClock for FixedClock {
        fn now_unix_ms(&self) -> u64 {
            self.0
        }
    }

    /// Which credential a case presents. Kept separate from the token so one
    /// case can build the same credential twice — `PresentedCredential` is
    /// deliberately not `Clone`.
    #[derive(Clone, Copy)]
    enum Kind {
        Bearer,
        Grant,
    }

    fn present(kind: Kind, token: &str) -> PresentedCredential {
        match kind {
            Kind::Bearer => PresentedCredential::Bearer(token.to_owned()),
            Kind::Grant => PresentedCredential::ConversationGrant(token.to_owned()),
        }
    }

    /// What an entry point decided, in a form the two can be compared by.
    ///
    /// The granted arm renders the whole authorization — scope shape, scope
    /// members, `EXPLAIN` policy, and every attribution field — so a case
    /// where one entry point granted a wider scope, or attributed the audit
    /// record to a different persona, fails rather than passing on a shared
    /// "granted".
    #[derive(Debug, PartialEq, Eq)]
    enum Decision {
        Granted(String),
        Refused(&'static str),
    }

    /// The refusal's identity, by variant. Never its message: two entry
    /// points reporting the same words for different variants is drift.
    const fn refusal(error: &PrincipalError) -> &'static str {
        match error {
            PrincipalError::InvalidSession => "InvalidSession",
            PrincipalError::StoreUnavailable => "StoreUnavailable",
            PrincipalError::NotAuthorizedForFleet => "NotAuthorizedForFleet",
            PrincipalError::InvalidGrant => "InvalidGrant",
            PrincipalError::GrantExpired => "GrantExpired",
        }
    }

    /// One renderer, fed by whichever entry point produced the fields — the
    /// embedded one mints a whole [`ScopedQuery`], the service one stops at
    /// the [`Scoping`] it will build a session from later. Comparing them
    /// through this single function is the point: a field the two populate
    /// differently shows up as unequal text.
    fn render(
        scope: &QueryScope,
        allow_explain: bool,
        caller_identity: Option<&str>,
        conversation_id: Option<&str>,
        turn_id: Option<&str>,
        web_session_id: Option<&str>,
    ) -> Decision {
        let scope = match scope {
            QueryScope::Fleet => "fleet".to_owned(),
            QueryScope::Conversations(conversations) => {
                let mut ids = conversations.clone();
                ids.sort_unstable();
                format!("conversations[{}]", ids.join(","))
            }
        };
        Decision::Granted(format!(
            "{scope} explain={allow_explain} caller={caller_identity:?} \
             conversation={conversation_id:?} turn={turn_id:?} \
             web_session={web_session_id:?}",
        ))
    }

    fn granted(scoping: &Scoping) -> Decision {
        render(
            &scoping.scope,
            scoping.allow_explain,
            scoping.caller_identity.as_deref(),
            scoping.conversation_id.as_deref(),
            scoping.turn_id.as_deref(),
            scoping.web_session_id.as_deref(),
        )
    }

    fn granted_session(session: &ScopedQuery) -> Decision {
        render(
            &session.scope,
            session.allow_explain,
            session.caller_identity.as_deref(),
            session.conversation_id.as_deref(),
            session.turn_id.as_deref(),
            session.web_session_id.as_deref(),
        )
    }

    /// The embedded entry point: verify, then scope.
    async fn embedded(authority: &QueryAuthority, kind: Kind, token: &str, now: u64) -> Decision {
        let verified = match kind {
            Kind::Bearer => authority.verify_admin_session(token, now).await,
            Kind::Grant => authority.verify_conversation_grant(token, now),
        };
        match verified {
            Err(error) => Decision::Refused(refusal(&error)),
            Ok(principal) => match authority.scope_for(&principal).await {
                Err(error) => Decision::Refused(refusal(&error)),
                Ok(session) => granted_session(&session),
            },
        }
    }

    /// The service entry point: admit a witness.
    async fn service(authority: &QueryAuthority, kind: Kind, token: &str, now: u64) -> Decision {
        let admitted = CredentialWitness::admit(
            present(kind, token),
            authority.credential_authority(),
            Arc::new(FixedClock(now)),
        )
        .await;
        match admitted {
            Err(error) => Decision::Refused(refusal(&error)),
            Ok((_witness, scoping)) => granted(&scoping),
        }
    }

    /// Runs both entry points over one credential state and asserts they
    /// agree. Returns the shared decision so a caller can assert what it is.
    async fn parity(authority: &QueryAuthority, kind: Kind, token: &str, now: u64) -> Decision {
        let embedded = embedded(authority, kind, token, now).await;
        let service = service(authority, kind, token, now).await;
        assert_eq!(
            embedded, service,
            "the embedded and service entry points must decide identically"
        );
        embedded
    }

    fn admin_token(fx: &Fixture, persona_id: &str, expires_in_ms: u64) -> String {
        mint_session(
            &fx.signer.relabel_for_test(),
            &SessionSubject::Persona {
                persona_id: persona_id.to_owned(),
            },
            &[SessionScope::ExplorerRead],
            NOW,
            expires_in_ms,
        )
    }

    #[tokio::test]
    async fn a_valid_admin_session_grants_the_same_fleet_scope_on_both_entry_points() {
        let fx = Fixture::build("parity-valid").await;
        let persona_id = fx.make_admin("alice").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        let decision = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(
            decision,
            Decision::Granted(format!(
                "fleet explain=true caller=Some({persona_id:?}) conversation=None turn=None \
                 web_session=None"
            ))
        );

        fx.teardown().await;
    }

    #[tokio::test]
    async fn an_expired_session_refuses_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-expired").await;
        let persona_id = fx.make_admin("alice").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        // The same token that just granted fleet scope, read one millisecond
        // past its expiry.
        let decision = parity(&fx.authority, Kind::Bearer, &token, NOW + TEST_TTL_MS + 1).await;
        assert_eq!(decision, Decision::Refused("InvalidSession"));

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_revoked_session_refuses_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-revoked").await;
        let persona_id = fx.make_admin("alice").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        // Granted while live, so the refusal below is the revocation and not
        // some other property of the token.
        assert!(matches!(
            parity(&fx.authority, Kind::Bearer, &token, NOW).await,
            Decision::Granted(_)
        ));

        fx.revoked.revoke(&token);
        let decision = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(decision, Decision::Refused("InvalidSession"));

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_malformed_session_refuses_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-malformed").await;

        for token in ["", "not-a-token", "aaaa.bbbb.cccc"] {
            let decision = parity(&fx.authority, Kind::Bearer, token, NOW).await;
            assert_eq!(
                decision,
                Decision::Refused("InvalidSession"),
                "token {token:?}"
            );
        }

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_non_admin_session_scopes_to_its_own_participations_on_both_entry_points() {
        let fx = Fixture::build("parity-non-admin").await;
        let persona_id = fx.make_non_admin("bob").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        // A verified non-admin is not refused: it mints a persona principal
        // scoped to what it participated in. Both entry points must agree on
        // that set, not merely on "granted".
        let decision = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(
            decision,
            Decision::Granted(format!(
                "conversations[conv-bob] explain=false caller=Some({persona_id:?}) \
                 conversation=None turn=None web_session=None"
            ))
        );

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_participation_change_moves_both_entry_points_together() {
        let fx = Fixture::build("parity-participation").await;
        let persona_id = fx.make_non_admin("carol").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        let before = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(
            before,
            Decision::Granted(format!(
                "conversations[conv-carol] explain=false caller=Some({persona_id:?}) \
                 conversation=None turn=None web_session=None"
            ))
        );

        // The same persona joins a second conversation. The token did not
        // change; the authorization did.
        fx.persona
            .attribute_persona(
                persona_id.clone(),
                "conv-carol-2".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("attribute the persona to a second conversation");

        let after = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(
            after,
            Decision::Granted(format!(
                "conversations[conv-carol,conv-carol-2] explain=false caller=Some({persona_id:?}) \
                 conversation=None turn=None web_session=None"
            )),
            "both entry points must observe the widened participation set"
        );

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_merged_persona_resolves_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-merged").await;

        let absorbed_identity = ExternalIdentity {
            provider: "test".to_owned(),
            scope: "s".to_owned(),
            external_id: "dana-old".to_owned(),
            display_name: "dana-old".to_owned(),
            ..Default::default()
        };
        let survivor_identity = ExternalIdentity {
            provider: "test".to_owned(),
            scope: "s".to_owned(),
            external_id: "dana-new".to_owned(),
            display_name: "dana-new".to_owned(),
            ..Default::default()
        };

        let first = fx
            .persona
            .attribute(
                absorbed_identity.clone(),
                "conv-dana-old".to_owned(),
                "initiator".to_owned(),
                NOW,
            )
            .await
            .expect("provision the first persona")
            .persona_id;
        let second = fx
            .persona
            .attribute(
                survivor_identity.clone(),
                "conv-dana-new".to_owned(),
                "initiator".to_owned(),
                NOW,
            )
            .await
            .expect("provision the second persona")
            .persona_id;
        assert_ne!(first, second, "the fixture needs two distinct personas");

        // A link ceremony merges the two. One id survives; the other becomes
        // a tombstone that resolves to no live profile.
        fx.persona
            .start_link(absorbed_identity, "link-code".to_owned(), TEST_TTL_MS, NOW)
            .await
            .expect("start the link");
        fx.persona
            .complete_link("link-code".to_owned(), survivor_identity, NOW)
            .await
            .expect("complete the link");

        // `active_persona` follows a merge alias, so the id that still names
        // itself is the survivor and the other one is the tombstone.
        let first_is_alive = fx
            .persona
            .active_persona(first.clone())
            .await
            .expect("read the first persona")
            .is_some_and(|active| active.persona_id == first);
        let (absorbed, survivor) = if first_is_alive {
            (second, first)
        } else {
            (first, second)
        };

        // A session minted before the merge still names the absorbed id. The
        // merge alias is followed, so the session keeps working — that is the
        // point of a merge. What matters is that both entry points follow it
        // to the SAME place: the audit record must name the survivor and not
        // the tombstone, and the scope must be the survivor's own
        // participation set, which now includes what it absorbed.
        let token = admin_token(&fx, &absorbed, TEST_TTL_MS);
        let decision = parity(&fx.authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(
            decision,
            Decision::Granted(format!(
                "conversations[conv-dana-new,conv-dana-old] explain=false \
                 caller=Some({survivor:?}) conversation=None turn=None web_session=None"
            )),
            "both entry points must follow the merge alias to the same survivor and scope"
        );
        assert_ne!(
            absorbed, survivor,
            "the attributed identity must be the survivor, never the tombstone"
        );

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_store_outage_refuses_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-store-outage").await;
        let persona_id = fx.make_admin("erin").await;
        let token = admin_token(&fx, &persona_id, TEST_TTL_MS);

        // The token verifies; the store cannot answer whether it is an admin.
        // That is infrastructure, and both entry points must call it that —
        // never a 401 or a silently narrowed scope.
        let authority = fx.authority_with_empty_persona_cell();
        let decision = parity(&authority, Kind::Bearer, &token, NOW).await;
        assert_eq!(decision, Decision::Refused("StoreUnavailable"));

        fx.teardown().await;
    }

    #[tokio::test]
    async fn a_conversation_grant_decides_identically_on_both_entry_points() {
        let fx = Fixture::build("parity-grant").await;
        let token = mint_conversation_grant(
            &fx.signer.relabel_for_test(),
            "conv-grant",
            GrantSubject::Turn("turn-7".to_owned()),
            NOW + TEST_TTL_MS,
        );

        let decision = parity(&fx.authority, Kind::Grant, &token, NOW).await;
        assert_eq!(
            decision,
            Decision::Granted(
                "conversations[conv-grant] explain=false caller=None \
                 conversation=Some(\"conv-grant\") turn=Some(\"turn-7\") web_session=None"
                    .to_owned()
            )
        );

        // Expiry and forgery keep their own distinct refusals on both.
        assert_eq!(
            parity(&fx.authority, Kind::Grant, &token, NOW + TEST_TTL_MS + 1).await,
            Decision::Refused("GrantExpired")
        );
        assert_eq!(
            parity(&fx.authority, Kind::Grant, "not-a-grant", NOW).await,
            Decision::Refused("InvalidGrant")
        );

        fx.teardown().await;
    }
}