polyc-query-credential 2026.10.2

#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
//! Regression suite for [`polyc_query_credential::credential::CredentialAuthority`]
//! and [`polyc_query_credential::credential::CredentialWitness`] — the one
//! credential verification and scoping mechanism (Epic 1565, chunk 10C-1's
//! precursor).
//!
//! Every case here drives `CredentialAuthority`/`CredentialWitness` directly,
//! through their public API plus this crate's own `test-util` constructors —
//! no `DataFusion`, no journal, no `QueryAuthority`. Most of these cases were
//! moved, verbatim in behaviour, from `polyc-query`'s own
//! `authority::tests` module, where they had reached this mechanism only
//! through the embedded engine's own funnel; `authority::tests` kept the
//! cases whose subject was that embedded engine until the engine and the
//! module were deleted together (#3109). See
//! `docs/decisions/0023-credential-crate-split.md` and this PR's own body for
//! the full classification.

use std::sync::Arc;

use polyc_crypto::session::{RevokedTokens, SessionScope, SessionSubject, mint_session};
use polyc_crypto::signing_role::{
    ApprovalSigner, RoleTrustSet, SessionRole, TrustedKey, TurnReadRole, TurnReadSigner,
};
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
use polyc_query_credential::credential::{
    CredentialAuthority, CredentialWitness, PresentedCredential, SystemUnixClock, UnixClock,
};
use polyc_query_credential::principal::{
    PersonaSource, PersonaSourceHandle, Principal, PrincipalError, Scoping, SearchScopeError,
};
use polyc_query_credential::session::QueryScope;
use polyc_query_credential::test_support::TestPersonas;
use polyc_query_model::GrantSubject;
use polyc_query_model::{
    mint_admin_fleet_grant, mint_control_fleet_grant, mint_conversation_grant,
    mint_conversation_grant_for_memory, mint_persona_wide_grant,
};

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

/// The one signer every fixture in this file trusts, unless a case is
/// specifically about a DIFFERENT signer (rotation, wrong-signer).
fn test_signer() -> ApprovalSigner {
    ApprovalSigner::from_seed(1)
}

/// Builds a `CredentialAuthority` over `personas`, trusting `test_signer()`'s
/// public key for both the turn-read grant role and the legacy session role
/// — mirrors `polyc-query`'s own former fixture composition
/// (`QueryAuthority::new`), minus the eventlog/`DataFusion` wiring this crate
/// never links.
fn authority_over(personas: Arc<TestPersonas>) -> CredentialAuthority {
    let signer = test_signer();
    CredentialAuthority::legacy(
        PersonaSourceHandle::current(personas as Arc<dyn PersonaSource>),
        Arc::new(RevokedTokens::new()),
        RoleTrustSet::<TurnReadRole>::from_public_keys(vec![signer.public_key_bytes()])
            .expect("test signer public key is encoded ed25519"),
        RoleTrustSet::<SessionRole>::from_public_keys(vec![signer.public_key_bytes()])
            .expect("test signer public key is encoded ed25519"),
    )
}

/// The source-unavailable composition — no persona store bound at all.
fn authority_with_no_persona_source() -> CredentialAuthority {
    let signer = test_signer();
    CredentialAuthority::legacy(
        PersonaSourceHandle::unavailable(),
        Arc::new(RevokedTokens::new()),
        RoleTrustSet::<TurnReadRole>::from_public_keys(vec![signer.public_key_bytes()])
            .expect("test signer public key is encoded ed25519"),
        RoleTrustSet::<SessionRole>::from_public_keys(vec![signer.public_key_bytes()])
            .expect("test signer public key is encoded ed25519"),
    )
}

fn identity(label: &str) -> ExternalIdentity {
    ExternalIdentity {
        provider: "test".to_owned(),
        scope: "s".to_owned(),
        external_id: label.to_owned(),
        display_name: label.to_owned(),
        ..Default::default()
    }
}

async fn make_admin(personas: &TestPersonas, label: &str) -> String {
    let persona_id = identity(label).external_id;
    personas.grant(&persona_id, true);
    persona_id
}

/// Provisions a non-admin persona and ties it to `conv-{label}` — mirrors
/// `polyc-query`'s own former `Fixture::make_non_admin`.
async fn make_non_admin(personas: &TestPersonas, label: &str) -> String {
    personas
        .attribute(
            identity(label),
            format!("conv-{label}"),
            "initiator".to_owned(),
            NOW,
        )
        .await
        .expect("attribute")
        .persona_id
}

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

// ---------------------------------------------------------------------------
// `verify_admin_session` — bearer sessions
// ---------------------------------------------------------------------------

mod admin_session {
    use super::*;

    #[tokio::test]
    async fn admin_session_valid_admin_scopes_fleet() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "alice").await;
        let authority = authority_over(personas);
        let token = admin_token(&persona_id, TEST_TTL_MS);

        let principal = 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 scoping = authority
            .scoping_for(&principal)
            .await
            .expect("Admin always scopes");
        assert!(matches!(scoping.scope(), QueryScope::Fleet));
        assert!(scoping.allow_explain());
        assert_eq!(scoping.caller_identity(), Some(persona_id.as_str()));
    }

    /// Proof (PR 10A): the edge listener's own admission path
    /// (`CredentialWitness::admit_grant_only`) never reaches session
    /// verification at all — a REAL, validly-signed session bearer, the
    /// exact shape `admit_bearer`'s fallback accepts, is refused with
    /// [`PrincipalError::InvalidGrant`] (the grant-verification path's own
    /// refusal), never [`PrincipalError::InvalidSession`] (which only
    /// `verify_admin_session` can produce). The two are structurally
    /// distinct variants, so getting the first is proof the second was never
    /// reached — not merely unreached by this one call.
    #[tokio::test]
    async fn edge_admission_never_reaches_session_verification() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "dave").await;
        let authority = Arc::new(authority_over(personas));
        let session_token = admin_token(&persona_id, TEST_TTL_MS);

        let result = CredentialWitness::admit_grant_only(
            session_token,
            Arc::clone(&authority),
            Arc::new(SystemUnixClock),
        )
        .await;
        let err = match result {
            Ok(_) => panic!("a session bearer must not verify as a signed conversation grant"),
            Err(err) => err,
        };
        assert!(
            matches!(err, PrincipalError::InvalidGrant),
            "expected InvalidGrant (the grant path's own refusal), got {err:?}"
        );
    }

    #[tokio::test]
    async fn admin_session_no_token_is_invalid_session() {
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let err = authority
            .verify_admin_session("not-a-real-token", NOW)
            .await
            .unwrap_err();
        assert!(matches!(err, PrincipalError::InvalidSession));
    }

    #[tokio::test]
    async fn admin_session_expired_is_invalid_session() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "bob").await;
        let authority = authority_over(personas);
        let token = admin_token(&persona_id, 1_000);

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

    #[tokio::test]
    async fn admin_session_revoked_is_invalid_session() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "carol").await;
        let revoked = Arc::new(RevokedTokens::new());
        let signer = test_signer();
        let authority = CredentialAuthority::legacy(
            PersonaSourceHandle::current(personas as Arc<dyn PersonaSource>),
            Arc::clone(&revoked),
            RoleTrustSet::<TurnReadRole>::from_public_keys(vec![signer.public_key_bytes()])
                .expect("test signer public key is encoded ed25519"),
            RoleTrustSet::<SessionRole>::from_public_keys(vec![signer.public_key_bytes()])
                .expect("test signer public key is encoded ed25519"),
        );
        let token = admin_token(&persona_id, TEST_TTL_MS);
        revoked.revoke(&token);

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

    /// A session that never carried `ExplorerRead` at all — a wallet-rooted
    /// session, or one that only ever carries `WalletManage` — is treated
    /// exactly like no session at all, never as a fleet-wide admin (ADR
    /// 0007's own gate).
    #[tokio::test]
    async fn admin_session_missing_explorer_read_scope_is_invalid_session() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "wallet-only").await;
        let authority = authority_over(personas);
        let token = mint_session(
            &test_signer().relabel_for_test(),
            &SessionSubject::Persona { persona_id },
            &[SessionScope::WalletManage],
            NOW,
            TEST_TTL_MS,
        );

        let err = authority
            .verify_admin_session(&token, NOW)
            .await
            .unwrap_err();
        assert!(
            matches!(err, PrincipalError::InvalidSession),
            "a session with no ExplorerRead scope must never mint any principal: {err:?}"
        );
    }

    /// A3: a valid session bound to a non-admin persona mints a
    /// `Principal::Persona` instead of a refusal.
    #[tokio::test]
    async fn admin_session_valid_non_admin_mints_a_persona_principal() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_non_admin(&personas, "dave").await;
        let authority = authority_over(personas);
        let token = admin_token(&persona_id, TEST_TTL_MS);

        let principal = 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:?}"),
        }
    }

    /// A session whose persona id does not resolve to any profile at all —
    /// there is no persona left to mint ANY principal for.
    #[tokio::test]
    async fn admin_session_unknown_persona_is_not_authorized_for_fleet() {
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let token = admin_token("persona-never-existed", TEST_TTL_MS);

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

    #[tokio::test]
    async fn admin_session_store_down_is_unavailable_not_401_or_403() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = make_admin(&personas, "erin").await;
        let token = admin_token(&persona_id, TEST_TTL_MS);
        let authority_no_store = authority_with_no_persona_source();

        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:?}"
        );
    }

    /// The record is a tombstone, not a deletion — de-admission must be read
    /// FRESH, never from a cached or point-in-time verdict.
    #[tokio::test]
    async fn admin_session_removed_persona_is_not_authorized_for_fleet() {
        let personas = Arc::new(TestPersonas::default());
        let admin_id = make_admin(&personas, "remover").await;
        let target_id = make_non_admin(&personas, "removed-later").await;
        let authority = authority_over(Arc::clone(&personas));

        let token = admin_token(&target_id, TEST_TTL_MS);
        authority
            .verify_admin_session(&token, NOW)
            .await
            .expect("precondition: the session verifies while the persona is live");

        personas
            .remove_access(admin_id, identity("removed-later"), NOW + 1)
            .await
            .expect("remove_access");

        let err = 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:?}"
        );
    }
}

// ---------------------------------------------------------------------------
// `verify_conversation_grant` — signed conversation-grant tokens
// ---------------------------------------------------------------------------

mod conversation_grant {
    use super::*;

    #[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 = authority_over(Arc::new(TestPersonas::default()));

        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:?}"),
        }
    }

    /// `scoping_for` on a `Turn`-subject grant scopes to exactly that
    /// conversation, with no `EXPLAIN` and the turn attributed for the
    /// per-turn query budget.
    #[tokio::test]
    async fn conversation_grant_turn_subject_scopes_to_its_conversation() {
        let signer = test_signer();
        let token = mint_conversation_grant(
            &signer.relabel_for_test(),
            "conv-parity",
            GrantSubject::Turn("turn-parity".to_owned()),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted grant must verify");

        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a conversation grant always scopes");
        assert!(!scoping.allow_explain());
        assert_eq!(scoping.turn_id(), Some("turn-parity"));
        assert_eq!(scoping.conversation_id(), Some("conv-parity"));
        assert_eq!(
            scoping.caller_identity(),
            None,
            "a turn subject has no owner identity"
        );
        assert_eq!(scoping.web_session_id(), None);
        match scoping.scope() {
            QueryScope::Conversations { conversations, .. } => {
                assert_eq!(conversations, &vec!["conv-parity".to_owned()]);
            }
            QueryScope::Fleet | QueryScope::FleetConversation { .. } => {
                panic!("a turn-scoped session must never be Fleet")
            }
        }
    }

    #[test]
    fn conversation_grant_minted_before_rotation_requires_historical_trust() {
        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 session_signer = test_signer();
        let session_trust =
            RoleTrustSet::<SessionRole>::from_public_keys(vec![session_signer.public_key_bytes()])
                .expect("test signer public key is encoded ed25519");

        let current_only =
            RoleTrustSet::<TurnReadRole>::checked(vec![TrustedKey::current(current.identity())])
                .expect("valid trust");
        let mut authority = CredentialAuthority::legacy(
            PersonaSourceHandle::unavailable(),
            Arc::new(RevokedTokens::new()),
            current_only,
            session_trust,
        );
        assert!(matches!(
            authority.verify_conversation_grant(&token, NOW),
            Err(PrincipalError::InvalidGrant)
        ));

        let historical = RoleTrustSet::<TurnReadRole>::checked(vec![
            TrustedKey::current(current.identity()),
            TrustedKey::retired(retired.identity()),
        ])
        .expect("valid rotation history");
        authority.replace_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` 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 authority = authority_over(Arc::new(TestPersonas::default()));

        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:?}"),
        }
    }

    /// `polychrome query verify` mints its grant by calling
    /// `polyc_query_model::mint_conversation_grant` directly — this test pins
    /// that what a CLI-side caller can reach is the SAME function
    /// `CredentialAuthority::verify_conversation_grant` verifies against,
    /// never a second, independently-maintained copy of the wire shape.
    #[test]
    fn conversation_grant_minted_via_the_shared_query_model_function_verifies() {
        let signer = test_signer();
        let token = mint_conversation_grant(
            &signer.relabel_for_test(),
            "conv-1",
            GrantSubject::WebSession("polychrome-query-verify".to_owned()),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));

        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a grant minted via polyc_query_model::mint_conversation_grant must verify");
        match principal {
            Principal::ConversationGrant(grant) => {
                assert_eq!(grant.conversation_id(), "conv-1");
                assert_eq!(
                    grant.subject(),
                    &GrantSubject::WebSession("polychrome-query-verify".to_owned())
                );
            }
            other => panic!("expected Principal::ConversationGrant, got {other:?}"),
        }
    }

    /// An expired grant is a DISTINCT, still-opaque outcome from a
    /// malformed/forged one, so a caller 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 authority = authority_over(Arc::new(TestPersonas::default()));

        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 authority = CredentialAuthority::legacy(
            PersonaSourceHandle::unavailable(),
            Arc::new(RevokedTokens::new()),
            RoleTrustSet::<TurnReadRole>::from_public_keys(vec![
                verifying_signer.public_key_bytes(),
            ])
            .expect("test signer public key is encoded ed25519"),
            RoleTrustSet::<SessionRole>::from_public_keys(vec![
                verifying_signer.public_key_bytes(),
            ])
            .expect("test signer public key is encoded ed25519"),
        );

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

    /// Domain separation (`GRANT_KIND`) actually binds: a genuinely-signed
    /// payload with the WRONG `kind` tag must not verify.
    #[test]
    fn conversation_grant_wrong_kind_tag_is_invalid_grant() {
        use base64::Engine as _;
        use base64::engine::general_purpose::URL_SAFE_NO_PAD;

        let signer = test_signer();
        let turn_read_signer: TurnReadSigner = signer.relabel_for_test();
        let signer_identity = turn_read_signer.identity();
        let claims = serde_json::json!({
            "kind": "some_other_signed_payload.v1",
            "issuer": signer_identity.issuer(),
            "key_id": signer_identity.key_id(),
            "conversation_id": "conv-x",
            "subject": GrantSubject::Turn("turn-y".to_owned()),
            "expires_at_ms": NOW + TEST_TTL_MS,
        });
        let canonical = serde_json::to_vec(&claims).expect("a JSON value 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 authority = authority_over(Arc::new(TestPersonas::default()));

        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 `InvalidGrant`.
    #[test]
    fn conversation_grant_malformed_shapes_are_invalid_grant() {
        let authority = authority_over(Arc::new(TestPersonas::default()));

        assert!(
            matches!(
                authority.verify_conversation_grant("not-a-grant-token", NOW),
                Err(PrincipalError::InvalidGrant)
            ),
            "a token with no `.` separator must be rejected"
        );
        assert!(
            matches!(
                authority.verify_conversation_grant(".c2ln", NOW),
                Err(PrincipalError::InvalidGrant)
            ),
            "an empty claims segment must be rejected"
        );
        assert!(
            matches!(
                authority.verify_conversation_grant("Y2xhaW1z.", NOW),
                Err(PrincipalError::InvalidGrant)
            ),
            "an empty signature segment must be rejected"
        );
        assert!(
            matches!(
                authority.verify_conversation_grant("Y2xhaW1z.sig.with.dots", NOW),
                Err(PrincipalError::InvalidGrant)
            ),
            "a signature segment containing a stray `.` must be rejected"
        );
    }

    /// A non-turn persona grant keeps the verified actor in the audit
    /// attribution without pretending the call came from a browser session
    /// or a turn — the routine-fire shape of `GrantSubject::Persona`
    /// (a real conversation id, unlike the persona-wide sentinel).
    #[tokio::test]
    async fn persona_subject_grant_preserves_the_real_actor_shape() {
        let signer = test_signer();
        let token = mint_conversation_grant(
            &signer.relabel_for_test(),
            "conv-routine-fire",
            GrantSubject::Persona("persona-owner".to_owned()),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona grant must verify");
        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a persona-subject conversation grant always scopes");

        assert_eq!(scoping.caller_identity(), Some("persona-owner"));
        assert_eq!(scoping.conversation_id(), Some("conv-routine-fire"));
        assert_eq!(scoping.turn_id(), None);
        assert_eq!(scoping.web_session_id(), None);
        assert!(matches!(scoping.scope(), QueryScope::Conversations { .. }));
        assert!(!scoping.allow_explain());
    }

    /// Q10/Q11 (`GetMyPayments`, 8C-2): a persona-wide grant — the
    /// empty-conversation-id sentinel — scopes to exactly the caller's own
    /// participated conversations, never a conversation they never joined
    /// and never the fleet.
    #[tokio::test]
    async fn persona_wide_grant_scopes_to_only_the_callers_own_participations() {
        let personas = Arc::new(TestPersonas::default());
        let caller = make_non_admin(&personas, "caller-persona").await;
        personas
            .attribute_persona(
                caller.clone(),
                "conv-mine".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("tie the caller to their own conversation");
        let authority = authority_over(personas);

        let token = mint_persona_wide_grant(
            &test_signer().relabel_for_test(),
            &caller,
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona-wide grant must verify");
        assert!(
            matches!(principal, Principal::Persona(_)),
            "the empty-conversation-id sentinel must mint Principal::Persona, not ConversationGrant"
        );
        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a persona-wide grant always scopes, even to zero conversations");

        match scoping.scope() {
            QueryScope::Conversations { conversations, .. } => {
                let mut sorted = conversations.clone();
                sorted.sort_unstable();
                assert_eq!(
                    sorted,
                    vec!["conv-caller-persona".to_owned(), "conv-mine".to_owned()],
                    "must see exactly the caller's own participations"
                );
            }
            QueryScope::Fleet | QueryScope::FleetConversation { .. } => {
                panic!("a persona-wide grant must never scope to the fleet")
            }
        }
        assert!(!scoping.allow_explain());
    }

    /// Boundary condition, and the credential half of what was previously
    /// only proven end to end through the embedded engine: zero
    /// participations still scopes VALIDLY, to an empty conversation set —
    /// never an error.
    #[tokio::test]
    async fn persona_scoping_for_widens_as_participation_changes() {
        let personas = Arc::new(TestPersonas::default());
        let authority = authority_over(Arc::clone(&personas));
        let token = mint_persona_wide_grant(
            &test_signer().relabel_for_test(),
            "persona-never-participated",
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona-wide grant must verify");

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

        // The same persona joins a conversation. The token did not change;
        // scoping_for re-reads participations fresh every call, so the
        // authorization widens without a new grant.
        personas
            .attribute_persona(
                "persona-never-participated".to_owned(),
                "conv-new".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("attribute");
        let widened = authority
            .scoping_for(&principal)
            .await
            .expect("a persona principal always scopes");
        match widened.scope() {
            QueryScope::Conversations { conversations, .. } => {
                assert_eq!(conversations, &vec!["conv-new".to_owned()]);
            }
            other => panic!("expected Conversations([conv-new]), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn persona_scope_store_down_is_unavailable() {
        let authority_no_store = authority_with_no_persona_source();
        let token = mint_persona_wide_grant(
            &test_signer().relabel_for_test(),
            "persona-x",
            NOW + TEST_TTL_MS,
        );
        let principal = authority_no_store
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona-wide grant must verify");

        // `Scoping` derives no `Debug` (the seal's own doc), so the `Ok` half
        // cannot flow through `.unwrap_err()`/`.expect_err()` — matched by
        // hand instead, mirroring `edge_admission_never_reaches_session_verification`.
        let err = match authority_no_store.scoping_for(&principal).await {
            Ok(_) => panic!("a persona's participation resolution must fail when unavailable"),
            Err(err) => err,
        };
        assert!(
            matches!(err, PrincipalError::StoreUnavailable),
            "a persona's participation resolution must surface store-unavailable as a distinct, \
             transient error: {err:?}"
        );
    }

    /// Mirrors `admin_fleet_scoping_admits_a_currently_admin_persona`'s Fleet
    /// counterpart (PR 10A review round 3): the standalone Query plane's own
    /// admission checks the realm and the statement digest, but the
    /// scope-derivation shape itself — Fleet, no `EXPLAIN`, no caller
    /// identity of its own — is this crate's to prove.
    #[tokio::test]
    async fn control_fleet_scoping_admits_fleet_without_explain() {
        let token = mint_control_fleet_grant(
            &test_signer().relabel_for_test(),
            "dashboard.conversations",
            &"ab".repeat(32),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted control-fleet grant must verify");
        assert!(matches!(principal, Principal::ControlFleet(_)));

        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a control-fleet principal always scopes, at the credential layer");
        assert!(matches!(scoping.scope(), QueryScope::Fleet));
        assert!(!scoping.allow_explain());
        assert_eq!(scoping.caller_identity(), None);
        let control_fleet = scoping
            .control_fleet()
            .expect("a control-fleet scoping must carry its own capability");
        assert_eq!(control_fleet.purpose(), "dashboard.conversations");
        assert_eq!(control_fleet.statement_digest(), &[0xab; 32]);
    }

    /// PR 10A: the AdminFleet re-check reads CURRENT admin status fresh at
    /// scoping time, never trusted from mint time alone.
    #[tokio::test]
    async fn admin_fleet_scoping_admits_a_currently_admin_persona() {
        let personas = Arc::new(TestPersonas::default());
        let admin_persona = make_admin(&personas, "fleet-admin").await;
        let authority = authority_over(personas);
        let token = mint_admin_fleet_grant(
            &test_signer().relabel_for_test(),
            &admin_persona,
            "session-a",
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted admin-fleet grant must verify");
        assert!(matches!(principal, Principal::AdminFleet(_)));

        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a currently-admin AdminFleet principal scopes to Fleet");
        assert!(matches!(scoping.scope(), QueryScope::Fleet));
        assert!(scoping.allow_explain());
        assert_eq!(scoping.caller_identity(), Some(admin_persona.as_str()));
        let admin_fleet = scoping
            .admin_fleet()
            .expect("an admin-fleet scoping must carry its own capability");
        assert_eq!(admin_fleet.admin_persona(), admin_persona);
        assert_eq!(admin_fleet.session(), "session-a");
    }

    /// The other half: an admin demoted since the grant was minted is
    /// refused — the grant's own TTL does not outlive the admin attribute.
    #[tokio::test]
    async fn admin_fleet_scoping_refuses_a_demoted_admin() {
        let personas = Arc::new(TestPersonas::default());
        let admin_persona = make_admin(&personas, "soon-demoted").await;
        let authority = authority_over(Arc::clone(&personas));
        let token = mint_admin_fleet_grant(
            &test_signer().relabel_for_test(),
            &admin_persona,
            "session-a",
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted admin-fleet grant must still verify cryptographically");

        personas.grant(&admin_persona, false);

        let err = match authority.scoping_for(&principal).await {
            Ok(_) => panic!(
                "a demoted admin must be refused, not silently honored for this grant's \
                 remaining TTL"
            ),
            Err(err) => err,
        };
        assert!(matches!(err, PrincipalError::NotAuthorizedForFleet));
    }

    /// PR 10A review round 3: `Scoping::for_conversation` REFUSES a
    /// `ControlFleet`/`AdminFleet`-shaped subject rather than panicking —
    /// provably unreachable through `verify_conversation_grant` in
    /// production, but this test constructs the shape directly, bypassing
    /// that upstream gate entirely, to prove a broken invariant refuses the
    /// one call that exposed it instead of taking down the worker running
    /// it.
    #[test]
    fn for_conversation_refuses_a_control_fleet_subject_instead_of_panicking() {
        let control_fleet = GrantSubject::ControlFleet {
            purpose: "test".to_owned(),
            statement_digest: "deadbeef".to_owned(),
        };
        assert!(matches!(
            Scoping::for_conversation_for_test("conv-1", &control_fleet),
            Err(PrincipalError::MalformedConversationGrantSubject)
        ));

        let admin_fleet = GrantSubject::AdminFleet {
            admin_persona: "admin-1".to_owned(),
            session: "session-1".to_owned(),
        };
        assert!(matches!(
            Scoping::for_conversation_for_test("conv-1", &admin_fleet),
            Err(PrincipalError::MalformedConversationGrantSubject)
        ));
    }
}

// ---------------------------------------------------------------------------
// Composite-trace grants
// ---------------------------------------------------------------------------

mod composite_trace {
    use super::*;

    /// A composite-trace grant scopes to its own conversation, exactly like
    /// any other conversation-scoped grant, AND carries its own fixed-
    /// statement capability — the digests the standalone Query plane's own
    /// admission checks the statement being run against.
    #[tokio::test]
    async fn composite_trace_grant_scopes_to_its_conversation_and_capability() {
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-trace",
            GrantSubject::CompositeTrace {
                persona_id: "persona-owner".to_owned(),
                trace_statement_digest: "aa".repeat(32),
                memory_statement_digest: "bb".repeat(32),
                routine_statement_digest: "cc".repeat(32),
            },
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted composite-trace grant must verify");
        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a composite-trace grant always scopes");

        assert_eq!(scoping.conversation_id(), Some("conv-trace"));
        assert_eq!(scoping.caller_identity(), Some("persona-owner"));
        assert!(!scoping.allow_explain());
        let composite = scoping
            .composite_trace()
            .expect("a composite-trace scoping must carry its own capability");
        assert_eq!(composite.trace_statement_digest(), "aa".repeat(32));
        assert_eq!(
            composite.memory_statement_digest(),
            Some("bb".repeat(32).as_str())
        );
        assert_eq!(composite.routine_statement_digest(), "cc".repeat(32));
    }
}

mod admin_composite_trace {
    use super::*;

    /// A clock fixed at [`NOW`].
    struct AtNow;

    impl UnixClock for AtNow {
        fn now_unix_ms(&self) -> u64 {
            NOW
        }
    }

    fn admin_subject() -> GrantSubject {
        GrantSubject::AdminCompositeTrace {
            trace_statement_digest: "aa".repeat(32),
            routine_statement_digest: "cc".repeat(32),
            address_statement_digest: "dd".repeat(32),
        }
    }

    /// POLY-394: an admin composite-trace grant scopes to its conversation,
    /// carries no caller identity and no owner memory partition, and its
    /// capability carries no memory digest.
    #[tokio::test]
    async fn admin_composite_trace_grant_scopes_without_persona_or_memory() {
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-trace",
            admin_subject(),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted admin composite-trace grant must verify");
        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("an admin composite-trace grant always scopes");

        assert_eq!(scoping.conversation_id(), Some("conv-trace"));
        assert_eq!(scoping.caller_identity(), None);
        assert_eq!(
            scoping.principal_kind(),
            polyc_query_credential::principal::PrincipalKind::ConversationGrantAdminCompositeTrace
        );
        match scoping.scope() {
            polyc_query_credential::session::QueryScope::Conversations { memory, .. } => {
                assert_eq!(memory.owner, None);
                assert!(memory.participants.is_empty());
            }
            polyc_query_credential::session::QueryScope::Fleet
            | polyc_query_credential::session::QueryScope::FleetConversation { .. } => {
                panic!("an admin composite trace must scope to its conversation")
            }
        }
        let composite = scoping
            .composite_trace()
            .expect("the scoping carries its fixed-statement capability");
        assert_eq!(composite.trace_statement_digest(), "aa".repeat(32));
        assert_eq!(composite.memory_statement_digest(), None);
        assert_eq!(composite.routine_statement_digest(), "cc".repeat(32));
        assert!(matches!(
            authority
                .composite_trace_memory_scope(&scoping, &["persona-x".to_owned()])
                .await,
            Err(PrincipalError::SourceOutsideScope)
        ));
    }

    /// POLY-464: a witness that serves the participant-address statement
    /// re-derives the one-conversation Fleet scope for its own grant's
    /// conversation on every revalidation. It is never the Fleet-wide
    /// scope: a running stream treats `Fleet` as containing every scope, so
    /// a regression to it would pass the containment check silently.
    #[tokio::test]
    async fn an_address_witness_revalidates_to_its_one_conversation() {
        let authority = Arc::new(authority_over(Arc::new(TestPersonas::default())));
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-trace",
            admin_subject(),
            NOW + TEST_TTL_MS,
        );
        let (mut witness, scoping) = CredentialWitness::admit(
            PresentedCredential::ConversationGrant(token),
            Arc::clone(&authority),
            Arc::new(AtNow),
        )
        .await
        .expect("a freshly minted admin composite-trace grant must admit");
        assert_eq!(scoping.conversation_id(), Some("conv-trace"));

        witness.retain_composite_trace_addresses();
        let scope = witness
            .current_scope()
            .await
            .expect("the admin grant still carries the address digest");
        assert_ne!(scope, QueryScope::Fleet);
        assert_eq!(
            scope,
            QueryScope::FleetConversation {
                conversation: "conv-trace".to_owned(),
            }
        );
    }

    /// POLY-464: a persona composite-trace grant carries no address digest,
    /// so an address witness over it refuses on revalidation.
    #[tokio::test]
    async fn an_address_witness_over_a_persona_grant_refuses() {
        let authority = Arc::new(authority_over(Arc::new(TestPersonas::default())));
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-trace",
            GrantSubject::CompositeTrace {
                persona_id: "persona-x".to_owned(),
                trace_statement_digest: "aa".repeat(32),
                memory_statement_digest: "bb".repeat(32),
                routine_statement_digest: "cc".repeat(32),
            },
            NOW + TEST_TTL_MS,
        );
        let (mut witness, _scoping) = CredentialWitness::admit(
            PresentedCredential::ConversationGrant(token),
            Arc::clone(&authority),
            Arc::new(AtNow),
        )
        .await
        .expect("a persona composite-trace grant admits");
        witness.retain_composite_trace_addresses();
        assert!(matches!(
            witness.current_scope().await,
            Err(PrincipalError::SourceOutsideScope)
        ));
    }

    /// An admin composite-trace grant that proposes memory sources has no
    /// legitimate mint site and is refused as a whole.
    #[test]
    fn admin_composite_trace_grant_with_memory_sources_is_refused() {
        let token = mint_conversation_grant_for_memory(
            &test_signer().relabel_for_test(),
            "conv-trace",
            admin_subject(),
            NOW + TEST_TTL_MS,
            vec!["persona-x".to_owned()],
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        assert!(matches!(
            authority.verify_conversation_grant(&token, NOW),
            Err(PrincipalError::InvalidGrant)
        ));
    }

    /// A persona composite-trace grant with an empty persona is refused: an
    /// empty id is never a stand-in for "no persona".
    #[test]
    fn persona_composite_trace_grant_with_an_empty_persona_is_refused() {
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-trace",
            GrantSubject::CompositeTrace {
                persona_id: String::new(),
                trace_statement_digest: "aa".repeat(32),
                memory_statement_digest: "bb".repeat(32),
                routine_statement_digest: "cc".repeat(32),
            },
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        assert!(matches!(
            authority.verify_conversation_grant(&token, NOW),
            Err(PrincipalError::InvalidGrant)
        ));
    }

    /// The persona subject keeps its exact wire shape; the admin subject is a
    /// separate tag, so neither decodes as the other.
    #[test]
    fn the_two_composite_subjects_have_distinct_wire_tags() {
        let persona = serde_json::to_value(GrantSubject::CompositeTrace {
            persona_id: "p".to_owned(),
            trace_statement_digest: "t".to_owned(),
            memory_statement_digest: "m".to_owned(),
            routine_statement_digest: "r".to_owned(),
        })
        .expect("serializes");
        assert_eq!(
            persona,
            serde_json::json!({"kind": "composite_trace", "id": {
                "persona_id": "p",
                "trace_statement_digest": "t",
                "memory_statement_digest": "m",
                "routine_statement_digest": "r",
            }})
        );
        let admin = serde_json::to_value(GrantSubject::AdminCompositeTrace {
            trace_statement_digest: "t".to_owned(),
            routine_statement_digest: "r".to_owned(),
            address_statement_digest: "a".to_owned(),
        })
        .expect("serializes");
        assert_eq!(
            admin,
            serde_json::json!({"kind": "admin_composite_trace", "id": {
                "trace_statement_digest": "t",
                "routine_statement_digest": "r",
                "address_statement_digest": "a",
            }})
        );
    }
}

// ---------------------------------------------------------------------------
// `authorize_conversation_read` — an authorization verdict, no session built
// ---------------------------------------------------------------------------

mod authorize_conversation_read {
    use super::*;

    #[tokio::test]
    async fn authorize_conversation_read_refuses_control_fleet() {
        let token = mint_control_fleet_grant(
            &test_signer().relabel_for_test(),
            "dashboard.conversations",
            &"0".repeat(64),
            NOW + TEST_TTL_MS,
        );
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted control-fleet grant must verify cryptographically");

        let err = authority
            .authorize_conversation_read(&principal)
            .await
            .expect_err("the forensics conversation-read funnel must refuse ControlFleet");
        assert!(matches!(err, PrincipalError::ControlFleetNotUsableEmbedded));
    }

    #[tokio::test]
    async fn authorize_conversation_read_refuses_admin_fleet() {
        let personas = Arc::new(TestPersonas::default());
        let admin_persona = make_admin(&personas, "fleet-reader").await;
        let authority = authority_over(personas);
        let token = mint_admin_fleet_grant(
            &test_signer().relabel_for_test(),
            &admin_persona,
            "session-a",
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted admin-fleet grant must verify cryptographically");

        let err = authority
            .authorize_conversation_read(&principal)
            .await
            .expect_err("the forensics conversation-read funnel must refuse AdminFleet");
        assert!(matches!(err, PrincipalError::AdminFleetNotUsableEmbedded));
    }

    /// The positive case: an ordinary conversation-scoped principal is
    /// admitted with `Ok(())` — an authorization verdict, never a runnable
    /// session (there is none to build here — this crate has no engine).
    #[tokio::test]
    async fn authorize_conversation_read_admits_a_persona_principal_without_building_a_session() {
        let personas = Arc::new(TestPersonas::default());
        let caller = make_non_admin(&personas, "reader").await;
        let authority = authority_over(personas);
        let token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-reader",
            GrantSubject::Persona(caller),
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona grant must verify");

        authority
            .authorize_conversation_read(&principal)
            .await
            .expect("an ordinary conversation-scoped principal must be admitted");
    }
}

// ---------------------------------------------------------------------------
// `own_rows_scoping` — "my own rows," independent of admin status
// ---------------------------------------------------------------------------

mod own_rows {
    use super::*;

    /// Admin status must never widen this scope, by construction: this
    /// method never reads it at all.
    #[tokio::test]
    async fn own_rows_scoping_ignores_admin_status() {
        let personas = Arc::new(TestPersonas::default());
        let admin_persona = make_admin(&personas, "admin-with-own-rows").await;
        personas
            .attribute_persona(
                admin_persona.clone(),
                "conv-admins-own".to_owned(),
                "initiator".to_owned(),
                NOW,
            )
            .await
            .expect("tie the admin to their own conversation");
        let authority = authority_over(personas);

        let scoping = authority
            .own_rows_scoping(&admin_persona)
            .await
            .expect("own_rows_scoping always resolves for a known persona");
        assert!(
            !scoping.allow_explain(),
            "own_rows_scoping must never admit EXPLAIN, admin or not"
        );
        assert_eq!(scoping.caller_identity(), Some(admin_persona.as_str()));
        match scoping.scope() {
            QueryScope::Conversations { conversations, .. } => {
                assert_eq!(conversations, &vec!["conv-admins-own".to_owned()]);
            }
            QueryScope::Fleet | QueryScope::FleetConversation { .. } => {
                panic!("own_rows_scoping must never scope to the fleet, even for an admin persona")
            }
        }
    }
}

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

mod search_scope {
    use super::*;

    async fn attribute_persona_to(
        personas: &TestPersonas,
        label: &str,
        conversation_ids: &[&str],
    ) -> String {
        let mut persona_id = None;
        for conversation_id in conversation_ids {
            persona_id = Some(
                personas
                    .attribute(
                        identity(label),
                        (*conversation_id).to_owned(),
                        "initiator".to_owned(),
                        NOW,
                    )
                    .await
                    .expect("attribute")
                    .persona_id,
            );
        }
        persona_id.expect("at least one conversation id")
    }

    /// Independently reimplements `SearchScope::hash`'s pinned five-step
    /// algorithm — deliberately NOT by calling the crate's own
    /// `canonical_search_scope`, so a bug in that function's own logic
    /// doesn't also corrupt this check.
    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())
                    .expect("a conversation id's byte length fits in u32")
                    .to_be_bytes(),
            );
            buf.extend_from_slice(bytes);
        }
        blake3::hash(&buf).to_hex().to_string()
    }

    #[tokio::test]
    async fn search_scope_removed_persona_refuses() {
        let personas = Arc::new(TestPersonas::default());
        let admin_id = make_admin(&personas, "remover").await;
        let target_id = attribute_persona_to(&personas, "removed-later", &["conv-a"]).await;
        let authority = authority_over(Arc::clone(&personas));

        personas
            .remove_access(admin_id, identity("removed-later"), NOW + 1)
            .await
            .expect("remove_access");

        let err = authority
            .resolve_search_scope(&target_id, "conv-caller", "turn-1")
            .await
            .unwrap_err();
        assert!(matches!(err, SearchScopeError::PersonaNotActive));
    }

    #[tokio::test]
    async fn search_scope_unknown_persona_refuses() {
        let authority = authority_over(Arc::new(TestPersonas::default()));
        let err = authority
            .resolve_search_scope("persona-never-existed", "conv-caller", "turn-1")
            .await
            .unwrap_err();
        assert!(matches!(err, SearchScopeError::PersonaNotActive));
    }

    #[tokio::test]
    async fn search_scope_store_down_is_unavailable() {
        let authority_no_store = authority_with_no_persona_source();
        let err = authority_no_store
            .resolve_search_scope("persona-x", "conv-caller", "turn-1")
            .await
            .unwrap_err();
        assert!(matches!(err, SearchScopeError::StoreUnavailable));
    }

    #[tokio::test]
    async fn search_scope_excludes_the_calling_conversation() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id =
            attribute_persona_to(&personas, "caller-exclusion", &["conv-a", "conv-b"]).await;
        let authority = authority_over(personas);

        let scope = 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()]);
        assert_eq!(scope.count(), 1);
        assert_eq!(scope.hash(), expected_scope_hash(&["conv-b"]));
    }

    #[tokio::test]
    async fn search_scope_excludes_a_tombstoned_conversation() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id = attribute_persona_to(
            &personas,
            "tombstone-exclusion",
            &["conv-a", "conv-b", "conv-c"],
        )
        .await;
        personas
            .set_search_visibility(
                persona_id.clone(),
                "conv-b".to_owned(),
                true,
                persona_id.clone(),
                NOW + 1,
            )
            .await
            .expect("set_search_visibility");
        let authority = authority_over(personas);

        let scope = 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()]);
    }

    #[tokio::test]
    async fn search_scope_over_cap_refuses_with_count() {
        let personas = Arc::new(TestPersonas::default());
        let persona_id =
            attribute_persona_to(&personas, "over-cap", &["conv-a", "conv-b", "conv-c"]).await;
        let authority = authority_over(personas);
        authority.set_test_search_scope_cap(2);

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

    #[tokio::test]
    async fn search_scope_hash_is_independent_of_enumeration_order() {
        let personas = Arc::new(TestPersonas::default());
        let ascending = attribute_persona_to(
            &personas,
            "order-ascending",
            &["conv-a", "conv-b", "conv-c"],
        )
        .await;
        let descending = attribute_persona_to(
            &personas,
            "order-descending",
            &["conv-c", "conv-b", "conv-a"],
        )
        .await;
        let authority = authority_over(personas);

        let via_ascending = authority
            .resolve_search_scope(&ascending, "conv-x", "turn-1")
            .await
            .expect("ascending order resolves");
        let via_descending = authority
            .resolve_search_scope(&descending, "conv-x", "turn-1")
            .await
            .expect("descending order resolves");

        assert_eq!(via_ascending.hash(), via_descending.hash());
        assert_eq!(
            via_ascending.conversation_ids(),
            via_descending.conversation_ids()
        );
    }

    #[tokio::test]
    async fn search_scope_hash_changes_when_the_set_changes() {
        let personas = Arc::new(TestPersonas::default());
        let smaller = attribute_persona_to(&personas, "hash-smaller", &["conv-a", "conv-b"]).await;
        let larger =
            attribute_persona_to(&personas, "hash-larger", &["conv-a", "conv-b", "conv-c"]).await;
        let authority = authority_over(personas);

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

        assert_ne!(via_smaller.hash(), via_larger.hash());
    }
}

// ---------------------------------------------------------------------------
// `scoping_for`'s persona-memory participant verification (N-1, N-3)
// ---------------------------------------------------------------------------

/// `CredentialAuthority::scoping_for`'s own proofs for 02-DESIGN.md §5's N-1
/// and N-3: the proposed co-participant check runs before any catalog read,
/// and a persona session has no channel to propose one at all.
mod persona_memory_scoping {
    use super::*;

    /// N-1: a conversation grant for `X` proposes co-participant `Q`, but
    /// `Q`'s own participations do not include `X` — refused with
    /// `SourceOutsideScope` before any catalog read, and the whole mint
    /// refuses rather than silently dropping just `Q`.
    #[tokio::test]
    async fn a_proposed_source_outside_the_grants_conversation_is_refused() {
        let personas = Arc::new(TestPersonas::default());
        personas.grant("persona-q", false);
        personas
            .attribute_persona(
                "persona-q".to_owned(),
                "conv-other".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("tie persona-q to a conversation that is not the grant's own");
        let authority = authority_over(personas);
        let token = mint_conversation_grant_for_memory(
            &test_signer().relabel_for_test(),
            "conv-x",
            GrantSubject::WebSession("persona-owner".to_owned()),
            NOW + TEST_TTL_MS,
            vec!["persona-q".to_owned()],
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted conversation grant must verify");

        let result = authority.scoping_for(&principal).await;
        assert!(
            matches!(result, Err(PrincipalError::SourceOutsideScope)),
            "persona-q never participated in conv-x"
        );
    }

    /// The positive counterpart: `Q` did participate in `X`, so the mint
    /// succeeds and `Q` joins the scope's verified participant set.
    #[tokio::test]
    async fn a_proposed_source_inside_the_grants_conversation_is_admitted() {
        let personas = Arc::new(TestPersonas::default());
        personas.grant("persona-q", false);
        personas
            .attribute_persona(
                "persona-q".to_owned(),
                "conv-x".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("tie persona-q to the grant's own conversation");
        let authority = authority_over(personas);
        let token = mint_conversation_grant_for_memory(
            &test_signer().relabel_for_test(),
            "conv-x",
            GrantSubject::WebSession("persona-owner".to_owned()),
            NOW + TEST_TTL_MS,
            vec!["persona-q".to_owned()],
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted conversation grant must verify");

        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("persona-q did participate in conv-x");
        let QueryScope::Conversations { memory, .. } = scoping.scope() else {
            panic!("a conversation grant must scope to Conversations, never Fleet");
        };
        assert_eq!(memory.owner.as_deref(), Some("persona-owner"));
        assert_eq!(memory.participants, vec!["persona-q".to_owned()]);
    }

    /// N-3: a persona session's principal type carries no field a caller
    /// could use to propose a co-participant at all.
    #[tokio::test]
    async fn a_persona_sessions_memory_scope_admits_no_participants() {
        let personas = Arc::new(TestPersonas::default());
        personas.grant("persona-p", false);
        let authority = authority_over(personas);

        let token = mint_persona_wide_grant(
            &test_signer().relabel_for_test(),
            "persona-p",
            NOW + TEST_TTL_MS,
        );
        let principal = authority
            .verify_conversation_grant(&token, NOW)
            .expect("a freshly minted persona-wide grant must verify");
        let scoping = authority
            .scoping_for(&principal)
            .await
            .expect("a persona session always scopes");
        let QueryScope::Conversations { memory, .. } = scoping.scope() else {
            panic!("a persona principal must never scope to Fleet");
        };
        assert_eq!(memory.owner.as_deref(), Some("persona-p"));
        assert!(
            memory.participants.is_empty(),
            "a persona session must never carry a co-participant"
        );
    }
}

// ---------------------------------------------------------------------------
// `CredentialWitness` — admission and re-validation
// ---------------------------------------------------------------------------

mod credential_witness {
    use super::*;

    /// A clock the test sets, so expiry and revalidation are exercised
    /// without waiting.
    struct FixedClock(std::sync::atomic::AtomicU64);

    impl FixedClock {
        fn new(now: u64) -> Self {
            Self(std::sync::atomic::AtomicU64::new(now))
        }
        fn set(&self, now: u64) {
            self.0.store(now, std::sync::atomic::Ordering::SeqCst);
        }
    }

    impl UnixClock for FixedClock {
        fn now_unix_ms(&self) -> u64 {
            self.0.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    /// `admit_bearer` decides which credential kind a token is ONCE: the
    /// grant path runs first (purely cryptographic), and a genuinely expired
    /// grant keeps its own `GrantExpired` refusal rather than being retried
    /// as a session — it is a real grant, not "not a grant."
    #[tokio::test]
    async fn witness_admit_bearer_prefers_grant_verification_and_keeps_expiry_distinct() {
        let personas = Arc::new(TestPersonas::default());
        let admin_persona = make_admin(&personas, "bearer-dispatch-admin").await;
        let authority = Arc::new(authority_over(personas));

        let grant_token = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-bearer-dispatch",
            GrantSubject::Turn("turn-1".to_owned()),
            NOW + TEST_TTL_MS,
        );
        let (_witness, scoping) = CredentialWitness::admit_bearer(
            grant_token,
            Arc::clone(&authority),
            Arc::new(FixedClock::new(NOW)),
        )
        .await
        .expect("a genuine grant token must admit via the grant path");
        assert_eq!(scoping.conversation_id(), Some("conv-bearer-dispatch"));

        let session_token = admin_token(&admin_persona, TEST_TTL_MS);
        let (_witness, scoping) = CredentialWitness::admit_bearer(
            session_token,
            Arc::clone(&authority),
            Arc::new(FixedClock::new(NOW)),
        )
        .await
        .expect("a session token that is not grant-shaped must fall back to session verification");
        assert!(matches!(scoping.scope(), QueryScope::Fleet));

        let expired_grant = mint_conversation_grant(
            &test_signer().relabel_for_test(),
            "conv-bearer-dispatch",
            GrantSubject::Turn("turn-1".to_owned()),
            NOW - 1,
        );
        let err = match CredentialWitness::admit_bearer(
            expired_grant,
            authority,
            Arc::new(FixedClock::new(NOW)),
        )
        .await
        {
            Ok(_) => panic!("an expired grant must not be retried as a session"),
            Err(err) => err,
        };
        assert!(
            matches!(err, PrincipalError::GrantExpired),
            "expected GrantExpired, not a session-path refusal: {err:?}"
        );
    }

    /// `admit` builds a witness and returns its initial authorized scope
    /// together — the pair every admission path (`admit_bearer`,
    /// `admit_grant_only`) delegates to.
    #[tokio::test]
    async fn witness_admit_builds_a_witness_and_its_initial_scope_together() {
        let personas = Arc::new(TestPersonas::default());
        let caller = make_non_admin(&personas, "witness-admit").await;
        let authority = Arc::new(authority_over(personas));
        let token = mint_persona_wide_grant(
            &test_signer().relabel_for_test(),
            &caller,
            NOW + TEST_TTL_MS,
        );

        let (witness, scoping) = CredentialWitness::admit(
            PresentedCredential::ConversationGrant(token),
            Arc::clone(&authority),
            Arc::new(FixedClock::new(NOW)),
        )
        .await
        .expect("a freshly minted persona-wide grant must admit");
        assert_eq!(scoping.caller_identity(), Some(caller.as_str()));

        // The witness independently re-proves the same authorization when
        // asked for its current scope — proving `admit` retained enough to
        // revalidate, not just to answer once.
        let revalidated = witness
            .current_scope()
            .await
            .expect("a witness admitted moments ago must still revalidate");
        match revalidated {
            QueryScope::Conversations { conversations, .. } => {
                assert_eq!(conversations, vec!["conv-witness-admit".to_owned()]);
            }
            QueryScope::Fleet | QueryScope::FleetConversation { .. } => {
                panic!("a persona-wide grant must never scope to the fleet")
            }
        }
    }

    /// `current_scope` re-proves the credential and REFLECTS whatever
    /// authorization now holds — narrower after a de-admission, wider after
    /// a new participation — never the scope frozen at admission time.
    #[tokio::test]
    async fn witness_current_scope_revalidates_and_reflects_changed_authorization() {
        let personas = Arc::new(TestPersonas::default());
        let admin_id = make_admin(&personas, "revoker").await;
        let target_id = make_non_admin(&personas, "witness-target").await;
        let authority = Arc::new(authority_over(Arc::clone(&personas)));
        let clock = Arc::new(FixedClock::new(NOW));
        let token = admin_token(&target_id, TEST_TTL_MS);

        let (witness, _scoping) = CredentialWitness::admit(
            PresentedCredential::Bearer(token),
            Arc::clone(&authority),
            Arc::clone(&clock) as Arc<dyn UnixClock>,
        )
        .await
        .expect("a live persona's session must admit");

        // Widens: a second participation joins the scope without a new
        // credential.
        personas
            .attribute_persona(
                target_id.clone(),
                "conv-witness-target-2".to_owned(),
                "participant".to_owned(),
                NOW,
            )
            .await
            .expect("attribute a second conversation");
        let widened = witness
            .current_scope()
            .await
            .expect("a live persona's scope must revalidate");
        match widened {
            QueryScope::Conversations { conversations, .. } => {
                let mut sorted = conversations.clone();
                sorted.sort_unstable();
                assert_eq!(
                    sorted,
                    vec![
                        "conv-witness-target".to_owned(),
                        "conv-witness-target-2".to_owned()
                    ]
                );
            }
            QueryScope::Fleet | QueryScope::FleetConversation { .. } => {
                panic!("a persona session must never scope to the fleet")
            }
        }

        // Narrows to a hard refusal: the persona is de-admitted mid-stream.
        personas
            .remove_access(admin_id, identity("witness-target"), NOW + 1)
            .await
            .expect("remove_access");
        clock.set(NOW + 2);
        let err = witness
            .current_scope()
            .await
            .expect_err("a de-admitted persona's still-unexpired token must stop releasing rows");
        assert!(matches!(err, PrincipalError::NotAuthorizedForFleet));
    }
}